From a94113a4b92e08fbd41c2f80361a1112700a6ca0 Mon Sep 17 00:00:00 2001 From: lyon Date: Thu, 25 Jun 2026 10:23:17 +0800 Subject: [PATCH] fix: keep workbench projection writes durable without reads --- .../cloud/workbench-projection-writer.test.ts | 118 ++++++++++ internal/cloud/workbench-projection-writer.ts | 210 +++++++++--------- 2 files changed, 218 insertions(+), 110 deletions(-) diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index 420332ce..a1598d3c 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -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 = { diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index 204d1738..d5489126 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -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) {