mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: keep workbench projection writes durable without reads

Browse Source
This commit is contained in:
lyon
2026-06-25 10:23:17 +08:00
parent 097d991aea
commit a94113a4b9
2 changed files with 218 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/workbench-projection-writer.test.ts
+118
View File
@@ -90,6 +90,124 @@ test("workbench projection writer commits terminal owner evidence as sealed dura
assert.equal(facts.checkpoints[0].timing.finishedAt, "2026-06-20T11:00:00.000Z");
});
test("workbench projection writer keeps event writes durable when checkpoint reads are blocked", async () => {
const factWrites = [];
const baseStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:30:00.000Z" });
const blocked = new Error("Postgres durable runtime adapter is blocked for runtime evidence queries");
blocked.code = "runtime_evidence_queries_blocked";
const runtimeStore = {
allocateWorkbenchProjectedSeq(params, requestMeta) {
return baseStore.allocateWorkbenchProjectedSeq(params, requestMeta);
},
async queryWorkbenchFacts() {
throw blocked;
},
async writeWorkbenchFacts(params, requestMeta) {
const result = baseStore.writeWorkbenchFacts(params, requestMeta);
factWrites.push({ params, requestMeta });
return result;
}
};
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: "trc_writer_blocked_checkpoint_event",
sessionId: "ses_writer_blocked_checkpoint_event",
sourceSeq: 1,
type: "backend",
eventType: "backend",
status: "running",
label: "agentrun:backend:running",
runId: "run_writer_blocked_checkpoint_event",
commandId: "cmd_writer_blocked_checkpoint_event",
startedAt: "2026-06-20T11:29:50.000Z",
createdAt: "2026-06-20T11:29:55.000Z"
}
});
assert.equal(factWrites.length, 1);
assert.equal(factWrites[0].params.facts.traceEvents[0].traceId, "trc_writer_blocked_checkpoint_event");
assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 1);
assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting");
});
test("workbench projection writer keeps terminal session and trace facts durable when reads are blocked", async () => {
const factWrites = [];
let allocationSeq = 0;
let queryCount = 0;
const blocked = new Error("Postgres durable runtime adapter is blocked for runtime evidence queries");
blocked.code = "runtime_evidence_queries_blocked";
const runtimeStore = {
async writeWorkbenchFacts(params, requestMeta) {
factWrites.push({ params, requestMeta });
return { written: true, facts: params.facts };
},
async queryWorkbenchFacts() {
queryCount += 1;
throw blocked;
},
async allocateWorkbenchProjectedSeq() {
allocationSeq += 1;
return { projectedSeq: allocationSeq };
}
};
const accessController = {
async recordAgentSessionOwner(input) {
return {
id: input.sessionId,
projectId: input.projectId,
ownerUserId: input.ownerUserId,
conversationId: input.conversationId,
threadId: input.threadId,
lastTraceId: input.traceId,
status: input.status,
session: input.session,
updatedAt: "2026-06-20T11:40:00.000Z"
};
}
};
const owner = await writeWorkbenchProjectionSession({
accessController,
runtimeStore,
traceId: "trc_writer_blocked_backfill",
ownerUserId: "usr_writer",
ownerRole: "user",
sessionId: "ses_writer_blocked_backfill",
projectId: "prj_writer",
conversationId: "cnv_writer_blocked_backfill",
threadId: "thread-writer-blocked-backfill",
status: "completed",
payload: {
traceId: "trc_writer_blocked_backfill",
status: "completed",
finalResponse: { text: "BLOCKED_BACKFILL_DONE", status: "completed", traceId: "trc_writer_blocked_backfill" },
agentRun: { runId: "run_writer_blocked_backfill", commandId: "cmd_writer_blocked_backfill", status: "completed", terminalStatus: "completed", lastSeq: 3 },
updatedAt: "2026-06-20T11:40:00.000Z"
},
session: {
sessionStatus: "completed",
messages: [
{ messageId: "msg_writer_blocked_backfill_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_blocked_backfill", traceId: "trc_writer_blocked_backfill" },
{ messageId: "msg_writer_blocked_backfill_agent", role: "agent", text: "BLOCKED_BACKFILL_DONE", status: "completed", turnId: "trc_writer_blocked_backfill", traceId: "trc_writer_blocked_backfill" }
],
finalResponse: { text: "BLOCKED_BACKFILL_DONE", status: "completed", traceId: "trc_writer_blocked_backfill" }
}
});
assert.equal(owner.id, "ses_writer_blocked_backfill");
assert.equal(queryCount, 1);
assert.equal(factWrites.length, 3);
assert.equal(factWrites[0].params.facts.checkpoints[0].traceId, "trc_writer_blocked_backfill");
assert.equal(factWrites[0].params.facts.turns[0].finalResponse.text, "BLOCKED_BACKFILL_DONE");
assert.equal(factWrites[1].params.facts.traceEvents[0].label, "agentrun:assistant:message");
assert.equal(factWrites[1].params.facts.traceEvents[0].text, "BLOCKED_BACKFILL_DONE");
assert.equal(factWrites[2].params.facts.traceEvents[0].label, "agentrun:result:completed");
assert.equal(factWrites[2].params.facts.traceEvents[0].terminal, true);
assert.equal(allocationSeq, 2);
});
test("workbench projection writer backfills terminal final and completion trace events from sealed facts", async () => {
const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:20:00.000Z" });
const accessController = {
internal/cloud/workbench-projection-writer.ts
+100 -110
View File
@@ -103,7 +103,7 @@ export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null,
});
}
export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event = {}, requestMeta = {} } = {}) {
export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event = {}, requestMeta = {}, previousCheckpoint = null } = {}) {
if (typeof runtimeStore?.writeWorkbenchFacts !== "function") return null;
if (typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") throw new Error("Workbench projection event writer requires a durable projectedSeq allocator.");
const traceId = safeTraceId(event.traceId ?? requestMeta.traceId);
@@ -114,11 +114,12 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
const turnId = textValue(event.turnId) || traceId;
const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status));
const status = terminal ? normalizeWorkbenchStatus(event.status) : "running";
const previousCheckpoint = await latestWorkbenchCheckpoint(runtimeStore, traceId);
const previousTiming = normalizeTimingProjection(previousCheckpoint?.timing) ?? normalizeTimingProjection(previousCheckpoint);
const checkpointHint = previousCheckpoint && typeof previousCheckpoint === "object" ? previousCheckpoint : null;
const resolvedPreviousCheckpoint = checkpointHint ?? await latestWorkbenchCheckpoint(runtimeStore, traceId, { traceStore: defaultCodeAgentTraceStore, phase: "event-checkpoint" });
const previousTiming = normalizeTimingProjection(resolvedPreviousCheckpoint?.timing) ?? normalizeTimingProjection(resolvedPreviousCheckpoint);
const sourceOccurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt);
const occurredAt = latestTimestamp(previousTiming?.lastEventAt, sourceOccurredAt) ?? sourceOccurredAt;
const previousTerminal = checkpointIsTerminal(previousCheckpoint);
const previousTerminal = checkpointIsTerminal(resolvedPreviousCheckpoint);
const suppressedAfterSeal = previousTerminal && !terminal;
const explicitEventTiming = normalizeTimingProjection(event.timing) ?? normalizeTimingProjection(event);
const explicitDurationMs = durationValue(event.durationMs ?? explicitEventTiming?.durationMs);
@@ -162,9 +163,9 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event
const projectedSeq = nonNegativeInteger(allocation?.projectedSeq);
if (projectedSeq <= 0) throw new Error("Workbench projection allocator returned an invalid projectedSeq.");
const eventId = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId });
const previousFinalText = finalResponseTextValue(previousCheckpoint?.finalResponse, previousCheckpoint?.assistantText, previousCheckpoint?.finalText);
const previousFinalText = finalResponseTextValue(resolvedPreviousCheckpoint?.finalResponse, resolvedPreviousCheckpoint?.assistantText, resolvedPreviousCheckpoint?.finalText);
const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, event.assistantText, event.reply, previousFinalText) : null;
const userMessageFact = sessionId && !suppressedAfterSeal && previousCheckpoint?.userMessage ? normalizeMessageFact(previousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null;
const userMessageFact = sessionId && !suppressedAfterSeal && resolvedPreviousCheckpoint?.userMessage ? normalizeMessageFact(resolvedPreviousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null;
const messageFact = sessionId && !suppressedAfterSeal ? normalizeMessageFact({
messageId: eventProjectionAssistantMessageId(traceId, event),
role: "agent",
@@ -307,7 +308,7 @@ function eventProjectionAssistantMessageId(traceId, event = {}) {
async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {} } = {}) {
if (typeof runtimeStore?.writeWorkbenchFacts !== "function") return null;
const safeId = safeTraceId(traceId ?? session?.traceId ?? payload?.traceId ?? params?.traceId);
const previousCheckpoint = safeId ? await latestWorkbenchCheckpoint(runtimeStore, safeId) : null;
const previousCheckpoint = safeId ? await latestWorkbenchCheckpoint(runtimeStore, safeId, { traceStore, phase: "facts-checkpoint" }) : null;
const facts = buildWorkbenchProjectionFacts({ traceId, ownerUserId, ownerRole, sessionId, projectId, conversationId, threadId, status, session, payload, params, previousCheckpoint });
if (isEmptyFacts(facts)) return null;
const timingAuthorityIssue = facts.checkpoints?.[0]?.diagnostic?.timingAuthorityIssue ?? null;
@@ -356,7 +357,7 @@ async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore =
}
async function backfillTerminalTraceEvents({ runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId = null, facts = {}, payload = null, status = null } = {}) {
if (typeof runtimeStore?.queryWorkbenchFacts !== "function" || typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") return null;
if (typeof runtimeStore?.writeWorkbenchFacts !== "function" || typeof runtimeStore?.allocateWorkbenchProjectedSeq !== "function") return null;
const safeId = safeTraceId(traceId ?? facts?.checkpoints?.[0]?.traceId ?? facts?.turns?.[0]?.traceId);
const checkpoint = facts?.checkpoints?.[0] ?? null;
const turn = facts?.turns?.[0] ?? null;
@@ -371,87 +372,80 @@ async function backfillTerminalTraceEvents({ runtimeStore = null, traceStore = d
const commandId = textValue(checkpoint?.commandId ?? payload?.agentRun?.commandId) || null;
const assistantMessageId = textValue(turn?.messageId ?? facts?.messages?.find((message) => message.role !== "user" && message.traceId === safeId)?.messageId) || null;
try {
const existing = await runtimeStore.queryWorkbenchFacts({ traceId: safeId, families: ["traceEvents"], limit: 500 });
const events = Array.isArray(existing?.facts?.traceEvents) ? existing.facts.traceEvents : [];
const missingAssistantFinal = !events.some((event) => terminalAssistantTraceEventMatches(event, finalText));
const missingCompletion = !events.some((event) => completedTerminalTraceEvent(event));
if (!missingAssistantFinal && !missingCompletion) return { written: 0, missingAssistantFinal: false, missingCompletion: false, valuesPrinted: false };
const timestamp = latestTimestamp(turn?.finishedAt, checkpoint?.finishedAt, turn?.lastEventAt, checkpoint?.lastEventAt, turn?.updatedAt, checkpoint?.updatedAt, payload?.updatedAt) ?? new Date().toISOString();
const baseSourceSeq = Math.max(nonNegativeInteger(checkpoint?.sourceSeq), nonNegativeInteger(turn?.sourceSeq), ...events.map((event) => nonNegativeInteger(event?.sourceSeq)));
const baseSourceSeq = Math.max(nonNegativeInteger(checkpoint?.sourceSeq), nonNegativeInteger(turn?.sourceSeq));
let offset = 1;
let written = 0;
if (missingAssistantFinal) {
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:assistant-final`,
type: "assistant",
eventType: "assistant",
status: "completed",
label: "agentrun:assistant:message",
message: finalText,
text: finalText,
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
final: true,
replyAuthority: true,
terminal: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
offset += 1;
}
if (missingCompletion || missingAssistantFinal) {
await writeWorkbenchProjectionEvent({
runtimeStore,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:completion`,
type: "result",
eventType: "terminal",
status: "completed",
label: "agentrun:result:completed",
message: "AgentRun result is ready for HWLAB short-connection polling.",
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
terminal: true,
sealed: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
}
return { written, missingAssistantFinal, missingCompletion, valuesPrinted: false };
await writeWorkbenchProjectionEvent({
runtimeStore,
previousCheckpoint: checkpoint,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:assistant-final`,
type: "assistant",
eventType: "assistant",
status: "completed",
label: "agentrun:assistant:message",
message: finalText,
text: finalText,
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
final: true,
replyAuthority: true,
terminal: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
offset += 1;
await writeWorkbenchProjectionEvent({
runtimeStore,
previousCheckpoint: checkpoint,
event: {
traceId: safeId,
sessionId,
turnId,
messageId: assistantMessageId,
source: "workbench-terminal-projection",
sourceSeq: baseSourceSeq + offset,
sourceEventId: `${safeId}:workbench-terminal-projection:completion`,
type: "result",
eventType: "terminal",
status: "completed",
label: "agentrun:result:completed",
message: "AgentRun result is ready for HWLAB short-connection polling.",
finalResponse: { text: finalText, status: "completed", traceId: safeId, valuesPrinted: false },
terminal: true,
sealed: true,
runId,
commandId,
timing: checkpoint?.timing,
startedAt: checkpoint?.startedAt,
lastEventAt: checkpoint?.lastEventAt,
finishedAt: checkpoint?.finishedAt,
durationMs: checkpoint?.durationMs,
createdAt: timestamp,
occurredAt: timestamp,
updatedAt: timestamp,
valuesPrinted: false
}
});
written += 1;
return { written, idempotent: true, valuesPrinted: false };
} catch (error) {
emitWorkbenchProjectionOtel("workbench.projection.terminal_trace_backfill", safeId, "error", {
"workbench.projection.phase": "terminal-trace-backfill",
@@ -471,27 +465,6 @@ async function backfillTerminalTraceEvents({ runtimeStore = null, traceStore = d
}
}
function terminalAssistantTraceEventMatches(event = {}, finalText = null) {
if (!isAssistantTraceEvent(event)) return false;
const text = finalResponseTextValue(event?.finalResponse, event?.text, event?.message, event?.content, event?.payload?.text, event?.payload?.message);
if (!text || text !== finalText) return false;
return event?.terminal === true || event?.final === true || event?.replyAuthority === true || normalizeWorkbenchStatus(event?.status) === "completed";
}
function completedTerminalTraceEvent(event = {}) {
const status = normalizeWorkbenchStatus(event?.status ?? event?.terminalStatus ?? event?.payload?.terminalStatus);
if (status !== "completed") return false;
if (event?.terminal === true) return true;
const label = textValue(event?.label ?? event?.type ?? event?.eventType).toLowerCase();
return /result:completed|terminal:completed|turn:completed/u.test(label);
}
function isAssistantTraceEvent(event = {}) {
const type = textValue(event?.type ?? event?.eventType).toLowerCase();
if (type === "assistant" || type === "assistant_message") return true;
return /assistant:message|assistant_message/u.test(textValue(event?.label).toLowerCase());
}
function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {}, previousCheckpoint = null } = {}) {
const safeId = safeTraceId(traceId ?? session?.traceId ?? payload?.traceId ?? params?.traceId);
const resolvedSessionId = textValue(sessionId ?? session?.sessionId ?? payload?.sessionId ?? params?.sessionId) || null;
@@ -770,10 +743,27 @@ function finalResponseTextValue(...values) {
return null;
}
async function latestWorkbenchCheckpoint(runtimeStore, traceId) {
async function latestWorkbenchCheckpoint(runtimeStore, traceId, { traceStore = defaultCodeAgentTraceStore, phase = "checkpoint-read" } = {}) {
if (typeof runtimeStore?.queryWorkbenchFacts !== "function") return null;
const result = await runtimeStore.queryWorkbenchFacts({ traceId, families: ["checkpoints"], limit: 1 });
return Array.isArray(result?.facts?.checkpoints) ? result.facts.checkpoints[0] ?? null : null;
const safeId = safeTraceId(traceId);
try {
const result = await runtimeStore.queryWorkbenchFacts({ traceId: safeId ?? traceId, families: ["checkpoints"], limit: 1 });
return Array.isArray(result?.facts?.checkpoints) ? result.facts.checkpoints[0] ?? null : null;
} catch (error) {
emitWorkbenchProjectionOtel("workbench.projection.checkpoint.read", safeId, "error", {
"workbench.projection.phase": phase
}, error);
if (safeId) appendProjectionDiagnostic(traceStore, safeId, {
type: "facts",
status: "degraded",
label: "projection-writer:checkpoint-read-failed",
errorCode: error?.code ?? "workbench_checkpoint_read_failed",
message: error?.message ?? "Workbench checkpoint read failed.",
terminal: false,
valuesPrinted: false
});
return null;
}
}
function projectionTimingForStatus(value, terminal) {