diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index f6d0aade..420332ce 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -90,6 +90,95 @@ 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 backfills terminal final and completion trace events from sealed facts", async () => { + const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:20:00.000Z" }); + 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:20:00.000Z" + }; + } + }; + + for (const [sourceSeq, label] of [ + [1, "agentrun:backend:admitted"], + [2, "agentrun:run:created"], + [3, "agentrun:runner-job:queued"] + ]) { + await writeWorkbenchProjectionEvent({ + runtimeStore, + event: { + traceId: "trc_writer_terminal_backfill", + sessionId: "ses_writer_terminal_backfill", + source: "agentrun", + sourceSeq, + type: "backend", + eventType: "backend", + status: "running", + label, + runId: "run_writer_terminal_backfill", + commandId: "cmd_writer_terminal_backfill", + createdAt: `2026-06-20T11:19:0${sourceSeq}.000Z` + } + }); + } + + await writeWorkbenchProjectionSession({ + accessController, + runtimeStore, + traceId: "trc_writer_terminal_backfill", + ownerUserId: "usr_writer", + ownerRole: "user", + sessionId: "ses_writer_terminal_backfill", + projectId: "prj_writer", + conversationId: "cnv_writer_terminal_backfill", + threadId: "thread-writer-terminal-backfill", + status: "completed", + payload: { + traceId: "trc_writer_terminal_backfill", + status: "completed", + finalResponse: { text: "BACKFILL_DONE", status: "completed", traceId: "trc_writer_terminal_backfill" }, + runnerTrace: { + traceId: "trc_writer_terminal_backfill", + status: "completed", + startedAt: "2026-06-20T11:19:01.000Z", + lastEventAt: "2026-06-20T11:19:13.000Z", + finishedAt: "2026-06-20T11:19:13.000Z", + eventCount: 3, + events: [] + }, + agentRun: { runId: "run_writer_terminal_backfill", commandId: "cmd_writer_terminal_backfill", status: "completed", terminalStatus: "completed", lastSeq: 3 }, + updatedAt: "2026-06-20T11:19:13.000Z" + }, + session: { + sessionStatus: "completed", + messages: [ + { messageId: "msg_writer_backfill_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal_backfill", traceId: "trc_writer_terminal_backfill" }, + { messageId: "msg_writer_backfill_agent", role: "agent", text: "BACKFILL_DONE", status: "completed", turnId: "trc_writer_terminal_backfill", traceId: "trc_writer_terminal_backfill" } + ], + finalResponse: { text: "BACKFILL_DONE", status: "completed", traceId: "trc_writer_terminal_backfill" } + } + }); + + const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_terminal_backfill", families: ["traceEvents", "checkpoints"], limit: 20 }); + const events = loaded.facts.traceEvents; + assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2, 3, 4, 5]); + assert.deepEqual(events.slice(-2).map((event) => event.label), ["agentrun:assistant:message", "agentrun:result:completed"]); + assert.equal(events.at(-2).text, "BACKFILL_DONE"); + assert.equal(events.at(-1).terminal, true); + assert.equal(events.at(-1).sealed, true); + assert.equal(loaded.facts.checkpoints[0].projectedSeq, 5); + assert.equal(loaded.facts.checkpoints[0].projectionStatus, "caught_up"); +}); + test("workbench projection writer does not rewrite prior-turn messages from merged owner session", async () => { const factWrites = []; const runtimeStore = { diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index 0e8e4b6e..204d1738 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -324,14 +324,16 @@ async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore = projectId, valuesPrinted: false }); + const terminalTraceBackfill = await backfillTerminalTraceEvents({ runtimeStore, traceStore, traceId: safeId, facts, payload, status }); emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "ok", { ...workbenchProjectionFactCountAttributes(facts), + "workbench.projection.terminal_trace_backfill_written": terminalTraceBackfill?.written ?? 0, "workbench.projection.phase": "session-facts", "workbench.projection.terminal": facts.turns.some((turn) => turn.terminal === true), "workbench.projection.session_id": facts.sessions[0]?.sessionId ?? sessionId ?? null, "workbench.projection.turn_id": facts.turns[0]?.turnId ?? null }); - return result; + return terminalTraceBackfill ? { ...result, terminalTraceBackfill } : result; } catch (error) { const safeId = safeTraceId(traceId); emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", { @@ -353,6 +355,143 @@ 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; + const safeId = safeTraceId(traceId ?? facts?.checkpoints?.[0]?.traceId ?? facts?.turns?.[0]?.traceId); + const checkpoint = facts?.checkpoints?.[0] ?? null; + const turn = facts?.turns?.[0] ?? null; + if (!safeId || checkpoint?.terminal !== true || turn?.terminal !== true) return null; + const terminalStatus = normalizeWorkbenchStatus(turn?.status ?? checkpoint?.status ?? status); + if (terminalStatus !== "completed") return null; + const finalText = finalResponseTextValue(turn?.finalResponse, checkpoint?.finalResponse, turn?.assistantText, checkpoint?.assistantText, payload?.finalResponse, payload?.assistantText); + if (!finalText) return null; + const sessionId = textValue(checkpoint?.sessionId ?? facts?.sessions?.[0]?.sessionId) || null; + const turnId = textValue(turn?.turnId ?? checkpoint?.turnId) || safeId; + const runId = textValue(checkpoint?.runId ?? payload?.agentRun?.runId) || null; + 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))); + 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 }; + } catch (error) { + emitWorkbenchProjectionOtel("workbench.projection.terminal_trace_backfill", safeId, "error", { + "workbench.projection.phase": "terminal-trace-backfill", + "workbench.projection.session_id": sessionId, + "workbench.projection.turn_id": turnId + }, error); + appendProjectionDiagnostic(traceStore, safeId, { + type: "facts", + status: "degraded", + label: "projection-writer:terminal-trace-backfill-failed", + errorCode: error?.code ?? "workbench_terminal_trace_backfill_failed", + message: error?.message ?? "Workbench terminal trace event backfill failed.", + terminal: false, + valuesPrinted: false + }); + return { written: 0, errorCode: error?.code ?? "workbench_terminal_trace_backfill_failed", valuesPrinted: false }; + } +} + +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;