From e79ea2cb42ef98e29587a24793c8cf0a94804244 Mon Sep 17 00:00:00 2001 From: lyon Date: Sat, 20 Jun 2026 08:57:21 +0800 Subject: [PATCH] feat: persist workbench projection facts --- internal/cloud/code-agent-trace-store.test.ts | 8 + internal/cloud/code-agent-trace-store.ts | 9 +- internal/cloud/server-agent-chat.test.ts | 42 +-- internal/cloud/server-code-agent-http.ts | 5 +- internal/cloud/server.ts | 4 +- .../cloud/workbench-projection-writer.test.ts | 125 ++++++++ internal/cloud/workbench-projection-writer.ts | 270 +++++++++++++++++- 7 files changed, 437 insertions(+), 26 deletions(-) create mode 100644 internal/cloud/workbench-projection-writer.test.ts diff --git a/internal/cloud/code-agent-trace-store.test.ts b/internal/cloud/code-agent-trace-store.test.ts index fc7e944e..a7b5a449 100644 --- a/internal/cloud/code-agent-trace-store.test.ts +++ b/internal/cloud/code-agent-trace-store.test.ts @@ -47,6 +47,7 @@ test("code agent trace store dedupes repeated AgentRun source events", () => { test("durable code agent trace store projects appended events without counting assistant deltas", () => { const writes = []; + const workbenchWrites = []; const baseStore = createCodeAgentTraceStore({ maxEvents: 20 }); const traceStore = createDurableCodeAgentTraceStore({ traceStore: baseStore, @@ -55,6 +56,10 @@ test("durable code agent trace store projects appended events without counting a writes.push({ params, requestMeta }); return { written: true }; } + }, + workbenchEventWriter(event, requestMeta) { + workbenchWrites.push({ event, requestMeta }); + return { written: true }; } }); const traceId = "trc_trace-store-durable-projection"; @@ -69,6 +74,9 @@ test("durable code agent trace store projects appended events without counting a assert.equal(writes[0].params.event.traceId, traceId); assert.equal(writes[0].params.event.seq, 1); assert.equal(writes[0].requestMeta.agentSessionId, "ses_trace_store_durable"); + assert.equal(workbenchWrites.length, 1); + assert.equal(workbenchWrites[0].event.traceId, traceId); + assert.equal(workbenchWrites[0].requestMeta.sessionId, undefined); }); test("code agent trace store retains six thousand regular events", () => { diff --git a/internal/cloud/code-agent-trace-store.ts b/internal/cloud/code-agent-trace-store.ts index 713d757a..2f792cf4 100644 --- a/internal/cloud/code-agent-trace-store.ts +++ b/internal/cloud/code-agent-trace-store.ts @@ -228,12 +228,12 @@ export function createCodeAgentTraceStore(options = {}) { export const defaultCodeAgentTraceStore = createCodeAgentTraceStore(); -export function createDurableCodeAgentTraceStore({ traceStore = defaultCodeAgentTraceStore, traceEventStore = null } = {}) { +export function createDurableCodeAgentTraceStore({ traceStore = defaultCodeAgentTraceStore, traceEventStore = null, workbenchEventWriter = null } = {}) { if (!traceEventStore || typeof traceEventStore.writeAgentTraceEvent !== "function") return traceStore; function append(traceId, event = {}, meta = {}) { const normalized = traceStore.append(traceId, event, meta); - if (normalized?.traceId) persistTraceEvent(traceEventStore, normalized, meta); + if (normalized?.traceId) persistTraceEvent(traceEventStore, normalized, meta, workbenchEventWriter); return normalized; } @@ -247,13 +247,16 @@ export function createDurableCodeAgentTraceStore({ traceStore = defaultCodeAgent }; } -function persistTraceEvent(traceEventStore, event, meta = {}) { +function persistTraceEvent(traceEventStore, event, meta = {}, workbenchEventWriter = null) { try { Promise.resolve(traceEventStore.writeAgentTraceEvent({ event }, { traceId: event.traceId, agentSessionId: event.agentSessionId ?? event.sessionId ?? meta.agentSessionId ?? meta.sessionId, sessionId: event.sessionId ?? meta.sessionId })).catch(() => {}); + if (typeof workbenchEventWriter === "function") { + Promise.resolve(workbenchEventWriter(event, meta)).catch(() => {}); + } } catch { // Trace capture must not fail the user turn when the durable projection is temporarily unavailable. } diff --git a/internal/cloud/server-agent-chat.test.ts b/internal/cloud/server-agent-chat.test.ts index 9776f545..bacbc7d0 100644 --- a/internal/cloud/server-agent-chat.test.ts +++ b/internal/cloud/server-agent-chat.test.ts @@ -1972,10 +1972,12 @@ test("cloud api AgentRun adapter rejects non-internal manager URLs by default", headers: { "content-type": "application/json", "x-trace-id": "trc_server-test-agentrun-public-url", cookie: manualSession.cookie }, body: JSON.stringify({ conversationId: manualSession.conversationId, sessionId: manualSession.sessionId, message: "must reject public manager url" }) }); - assert.equal(response.status, 500); + assert.equal(response.status, 202); const body = await response.json(); - assert.equal(body.error.code, "internal_error"); - assert.match(body.error.reason, /internal k3s Service DNS/u); + assert.equal(body.accepted, true); + const result = await pollAgentResult(port, "trc_server-test-agentrun-public-url"); + assert.equal(result.status, "failed"); + assert.match(JSON.stringify(result), /internal k3s Service DNS/u); } finally { await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); } @@ -2041,22 +2043,24 @@ test("cloud api /v1/agent/chat records authenticated owner on agent session", as }); validateCodeAgentChatSchema(payload); assert.equal(payload.ownerUserId, "usr_agent_owner"); - assert.equal(ownerRecords.length, 1); - assert.equal(ownerRecords[0].ownerUserId, "usr_agent_owner"); - assert.equal(ownerRecords[0].sessionId, "ses_server_stdio_pwd"); - assert.equal(ownerRecords[0].conversationId, "cnv_server-test-owner"); - assert.equal(ownerRecords[0].traceId, "trc_server-test-owner"); - assert.equal(ownerRecords[0].status, "completed"); - assert.equal(ownerRecords[0].session.messageCount, 2); - assert.equal(ownerRecords[0].session.messages.length, 2); - assert.equal(ownerRecords[0].session.messages[0].role, "user"); - assert.equal(ownerRecords[0].session.messages[0].text, "owner binding smoke"); - assert.equal(ownerRecords[0].session.messages[0].status, "sent"); - assert.equal(ownerRecords[0].session.messages[1].role, "agent"); - assert.equal(ownerRecords[0].session.messages[1].status, "completed"); - assert.equal(ownerRecords[0].session.messages[1].traceId, "trc_server-test-owner"); - assert.equal(ownerRecords[0].session.firstUserMessagePreview, "owner binding smoke"); - assert.equal(ownerRecords[0].session.valuesRedacted, true); + assert.equal(ownerRecords.length, 2); + assert.equal(ownerRecords[0].status, "running"); + const terminalOwnerRecord = ownerRecords.at(-1); + assert.equal(terminalOwnerRecord.ownerUserId, "usr_agent_owner"); + assert.equal(terminalOwnerRecord.sessionId, "ses_server_stdio_pwd"); + assert.equal(terminalOwnerRecord.conversationId, "cnv_server-test-owner"); + assert.equal(terminalOwnerRecord.traceId, "trc_server-test-owner"); + assert.equal(terminalOwnerRecord.status, "completed"); + assert.equal(terminalOwnerRecord.session.messageCount, 2); + assert.equal(terminalOwnerRecord.session.messages.length, 2); + assert.equal(terminalOwnerRecord.session.messages[0].role, "user"); + assert.equal(terminalOwnerRecord.session.messages[0].text, "owner binding smoke"); + assert.equal(terminalOwnerRecord.session.messages[0].status, "sent"); + assert.equal(terminalOwnerRecord.session.messages[1].role, "agent"); + assert.equal(terminalOwnerRecord.session.messages[1].status, "completed"); + assert.equal(terminalOwnerRecord.session.messages[1].traceId, "trc_server-test-owner"); + assert.equal(terminalOwnerRecord.session.firstUserMessagePreview, "owner binding smoke"); + assert.equal(terminalOwnerRecord.session.valuesRedacted, true); assert.equal(JSON.stringify(ownerRecords).includes("test-openai-key-material"), false); } finally { await new Promise((resolve, reject) => { diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index 5dffb8e9..0db441c7 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -2649,6 +2649,7 @@ async function recordCodeAgentSessionOwner({ payload = {}, params = {}, options const threadId = safeOpaqueId(session?.threadId ?? sessionReuse?.threadId ?? payload.threadId ?? params.threadId) || null; return writeWorkbenchProjectionSession({ accessController: options.accessController, + runtimeStore: options.runtimeStore, traceStore: options.traceStore ?? defaultCodeAgentTraceStore, traceId, ownerUserId, @@ -2659,7 +2660,9 @@ async function recordCodeAgentSessionOwner({ payload = {}, params = {}, options threadId, status, preserveLastTraceId, - session: codeAgentSessionOwnerEvidence(payload, params) + session: codeAgentSessionOwnerEvidence(payload, params), + payload, + params }); } diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index c4a7f3dc..c6227f50 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -28,6 +28,7 @@ import { handleCodeAgentChat } from "./code-agent-chat.ts"; import { createDurableCodeAgentTraceStore, defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { writeWorkbenchProjectionEvent } from "./workbench-projection-writer.ts"; import { createCodeAgentSessionRegistry } from "./code-agent-session-registry.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { createCodexStdioSessionManager } from "./codex-stdio-session.ts"; @@ -100,7 +101,8 @@ export function createCloudApiServer(options = {}) { const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env }); const traceStore = createDurableCodeAgentTraceStore({ traceStore: options.traceStore || defaultCodeAgentTraceStore, - traceEventStore: runtimeStore + traceEventStore: runtimeStore, + workbenchEventWriter: (event, meta) => writeWorkbenchProjectionEvent({ runtimeStore, event, requestMeta: meta }) }); const codexStdioManager = options.codexStdioManager || createCodexStdioSessionManager({ traceStore }); const codeAgentChatResults = options.codeAgentChatResults || createCodeAgentChatResultStore({ diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts new file mode 100644 index 00000000..0cc9b73f --- /dev/null +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -0,0 +1,125 @@ +// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor. +// Responsibility: Workbench projection writer/finalizer durable facts regression tests. + +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import { writeWorkbenchProjectionEvent, writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts"; + +test("workbench projection writer commits terminal owner evidence as sealed durable facts", async () => { + const factWrites = []; + const accessWrites = []; + const runtimeStore = { + async writeWorkbenchFacts(params, requestMeta) { + factWrites.push({ params, requestMeta }); + return { written: true, facts: params.facts }; + } + }; + const accessController = { + async recordAgentSessionOwner(input) { + accessWrites.push(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:00:00.000Z" + }; + } + }; + + await writeWorkbenchProjectionSession({ + accessController, + runtimeStore, + traceId: "trc_writer_terminal", + ownerUserId: "usr_writer", + ownerRole: "user", + sessionId: "ses_writer_terminal", + projectId: "prj_writer", + conversationId: "cnv_writer_terminal", + threadId: "thread-writer-terminal", + status: "completed", + payload: { + traceId: "trc_writer_terminal", + status: "completed", + assistantText: "final answer", + agentRun: { runId: "run_writer_terminal", commandId: "cmd_writer_terminal", lastSeq: 42 }, + updatedAt: "2026-06-20T11:00:00.000Z" + }, + session: { + sessionStatus: "completed", + messages: [ + { messageId: "msg_writer_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal", traceId: "trc_writer_terminal" }, + { messageId: "msg_writer_agent", role: "agent", text: "final answer", status: "completed", turnId: "trc_writer_terminal", traceId: "trc_writer_terminal" } + ], + finalResponse: { text: "final answer", status: "completed", traceId: "trc_writer_terminal" } + } + }); + + assert.equal(accessWrites.length, 1); + assert.equal(factWrites.length, 1); + const facts = factWrites[0].params.facts; + assert.equal(facts.sessions[0].status, "completed"); + assert.equal(facts.sessions[0].sealed, true); + assert.equal(facts.messages.length, 2); + assert.equal(facts.messages.find((message) => message.messageId === "msg_writer_agent").sealed, true); + assert.equal(facts.parts.some((part) => part.messageId === "msg_writer_agent" && part.text === "final answer" && part.sealed === true), true); + assert.equal(facts.turns[0].terminal, true); + assert.equal(facts.turns[0].sealed, true); + assert.equal(facts.turns[0].finalResponse.text, "final answer"); + assert.equal(facts.checkpoints[0].projectionStatus, "caught_up"); + assert.equal(facts.checkpoints[0].sealed, true); +}); + +test("workbench projection event commit writes trace event and checkpoint facts", async () => { + const factWrites = []; + const runtimeStore = { + async writeWorkbenchFacts(params, requestMeta) { + factWrites.push({ params, requestMeta }); + return { written: true, facts: params.facts }; + } + }; + + await writeWorkbenchProjectionEvent({ + runtimeStore, + event: { + traceId: "trc_writer_event", + sessionId: "ses_writer_event", + seq: 7, + sourceSeq: 70, + type: "backend", + status: "running", + label: "agentrun:backend:running", + runId: "run_writer_event", + commandId: "cmd_writer_event", + createdAt: "2026-06-20T11:01:00.000Z" + } + }); + await writeWorkbenchProjectionEvent({ + runtimeStore, + event: { + traceId: "trc_writer_event", + sessionId: "ses_writer_event", + seq: 8, + sourceSeq: 71, + type: "result", + status: "completed", + label: "agentrun:result:completed", + terminal: true, + runId: "run_writer_event", + commandId: "cmd_writer_event", + createdAt: "2026-06-20T11:02:00.000Z" + } + }); + + assert.equal(factWrites.length, 2); + assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 7); + assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting"); + assert.equal(factWrites[1].params.facts.sessions[0].status, "completed"); + assert.equal(factWrites[1].params.facts.traceEvents[0].sealed, true); + assert.equal(factWrites[1].params.facts.checkpoints[0].projectionStatus, "caught_up"); +}); diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index b7f269ef..d270caf5 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -2,14 +2,17 @@ * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0. * 职责: WorkbenchProjectionWriter 组件入口。唯一封装 Code Agent/AgentRun facts 到 Workbench session projection facts 的持久化写入。 */ +import { createHash } from "node:crypto"; + import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { safeTraceId } from "./server-http-utils.ts"; +import { createWorkbenchTurnProjection, normalizeWorkbenchStatus, projectionDiagnostics, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; -export async function writeWorkbenchProjectionSession({ accessController, traceStore = defaultCodeAgentTraceStore, traceId, ownerUserId, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, preserveLastTraceId = false } = {}) { +export async function writeWorkbenchProjectionSession({ accessController, runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId, ownerUserId, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {}, preserveLastTraceId = false } = {}) { if (!ownerUserId || typeof accessController?.recordAgentSessionOwner !== "function") return null; const safeId = safeTraceId(traceId); try { - return await accessController.recordAgentSessionOwner({ + const ownerRecord = await accessController.recordAgentSessionOwner({ ownerUserId, ownerRole, sessionId, @@ -20,6 +23,22 @@ export async function writeWorkbenchProjectionSession({ accessController, traceS status, session }); + await writeWorkbenchProjectionFacts({ + runtimeStore, + traceStore, + traceId: safeId, + ownerUserId, + ownerRole, + sessionId: ownerRecord?.id ?? sessionId, + projectId: ownerRecord?.projectId ?? projectId, + conversationId: ownerRecord?.conversationId ?? conversationId, + threadId: ownerRecord?.threadId ?? threadId, + status: ownerRecord?.status ?? status, + session: ownerRecord?.session ?? session, + payload, + params + }); + return ownerRecord; } catch (error) { if (safeId) appendProjectionDiagnostic(traceStore, safeId, { type: "session-owner", @@ -33,6 +52,253 @@ export async function writeWorkbenchProjectionSession({ accessController, traceS } } +export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event = {}, requestMeta = {} } = {}) { + if (typeof runtimeStore?.writeWorkbenchFacts !== "function") return null; + const traceId = safeTraceId(event.traceId ?? requestMeta.traceId); + if (!traceId) return null; + const sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq); + const projectedSeq = nonNegativeInteger(event.seq ?? event.projectedSeq ?? sourceSeq); + const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt); + const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null; + 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 sourceEventId = textValue(event.sourceEventId ?? event.id) || `${traceId}:${sourceSeq || projectedSeq}`; + const eventId = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId, sourceSeq, projectedSeq, label: event.label, occurredAt }); + const facts = { + sessions: sessionId ? [{ + sessionId, + status, + lastTraceId: traceId, + projectedSeq, + sourceSeq, + sourceEventId, + terminal, + sealed: terminal, + updatedAt: occurredAt + }] : [], + traceEvents: [{ + ...event, + id: eventId, + traceId, + sessionId, + turnId, + messageId: textValue(event.messageId) || null, + sourceSeq, + sourceEventId, + projectedSeq, + eventType: textValue(event.eventType ?? event.type ?? event.label) || "event", + terminal, + sealed: terminal, + occurredAt, + updatedAt: timestampValue(event.updatedAt ?? occurredAt) + }], + checkpoints: [{ + traceId, + sessionId, + turnId, + runId: textValue(event.runId) || null, + commandId: textValue(event.commandId) || null, + projectedSeq, + sourceSeq, + sourceEventId, + projectionStatus: terminal ? "caught_up" : "projecting", + projectionHealth: "healthy", + terminal, + sealed: terminal, + diagnostic: { + lastEventLabel: textValue(event.label) || null, + lastEventStatus: textValue(event.status) || null, + valuesRedacted: true + }, + updatedAt: timestampValue(event.updatedAt ?? occurredAt) + }] + }; + return runtimeStore.writeWorkbenchFacts({ facts }, { + traceId, + sessionId, + agentSessionId: requestMeta.agentSessionId ?? sessionId, + valuesPrinted: false + }); +} + +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 facts = buildWorkbenchProjectionFacts({ traceId, ownerUserId, ownerRole, sessionId, projectId, conversationId, threadId, status, session, payload, params }); + if (isEmptyFacts(facts)) return null; + try { + return await runtimeStore.writeWorkbenchFacts({ facts }, { + traceId: facts.checkpoints[0]?.traceId ?? traceId, + sessionId: facts.sessions[0]?.sessionId ?? sessionId, + ownerUserId, + projectId, + valuesPrinted: false + }); + } catch (error) { + const safeId = safeTraceId(traceId); + if (safeId) appendProjectionDiagnostic(traceStore, safeId, { + type: "facts", + status: "degraded", + label: "projection-writer:facts:persist-failed", + errorCode: error?.code ?? "workbench_facts_persist_failed", + message: error?.message ?? "Workbench facts persistence failed.", + terminal: false + }); + return null; + } +} + +function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, payload = null, params = {} } = {}) { + const safeId = safeTraceId(traceId ?? session?.traceId ?? payload?.traceId ?? params?.traceId); + const resolvedSessionId = textValue(sessionId ?? session?.sessionId ?? payload?.sessionId ?? params?.sessionId) || null; + const resolvedConversationId = textValue(conversationId ?? session?.conversationId ?? payload?.conversationId ?? params?.conversationId) || null; + const resolvedThreadId = textValue(threadId ?? session?.threadId ?? payload?.threadId ?? params?.threadId) || null; + const agentRun = payload?.agentRun ?? session?.agentRun ?? null; + const timestamp = timestampValue(payload?.updatedAt ?? session?.updatedAt ?? payload?.createdAt ?? session?.createdAt); + const normalizedStatus = normalizeWorkbenchStatus(payload?.status ?? session?.sessionStatus ?? status); + const terminal = TERMINAL_STATUSES.has(normalizedStatus); + const projection = createWorkbenchTurnProjection({ traceId: safeId, result: payload, session: { id: resolvedSessionId, status: normalizedStatus, session }, trace: payload?.runnerTrace ?? null }); + const diagnostic = projectionDiagnostics({ traceId: safeId, result: payload, trace: payload?.runnerTrace ?? null, projection }); + const messages = normalizeMessages(session?.messages, { + traceId: safeId, + sessionId: resolvedSessionId, + conversationId: resolvedConversationId, + threadId: resolvedThreadId, + terminal, + timestamp + }); + return { + sessions: resolvedSessionId ? [{ + sessionId: resolvedSessionId, + ownerUserId, + ownerRole, + projectId: projectId ?? session?.projectId ?? payload?.projectId ?? params?.projectId ?? null, + conversationId: resolvedConversationId, + threadId: resolvedThreadId, + status: terminal ? normalizedStatus : "running", + lastTraceId: safeId, + projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), + sourceSeq: nonNegativeInteger(projection.lastProjectedSeq), + sourceEventId: safeId, + terminal, + sealed: terminal, + sessionJson: session, + createdAt: timestampValue(session?.createdAt ?? payload?.createdAt ?? timestamp), + updatedAt: timestamp + }] : [], + messages, + parts: messages.flatMap((message) => message.text ? [{ + partId: stableFactId("wpt", { messageId: message.messageId, index: 0 }), + messageId: message.messageId, + sessionId: message.sessionId, + turnId: message.turnId, + traceId: message.traceId, + partIndex: 0, + partType: "text", + status: message.status, + projectedSeq: message.projectedSeq, + sourceSeq: message.sourceSeq, + sourceEventId: message.sourceEventId, + terminal: message.terminal, + sealed: message.sealed, + text: message.text, + updatedAt: message.updatedAt + }] : []), + turns: safeId && resolvedSessionId ? [{ + turnId: projection.turnId ?? safeId, + sessionId: resolvedSessionId, + traceId: safeId, + messageId: messages.find((message) => message.role !== "user")?.messageId ?? null, + status: projection.status ?? normalizedStatus, + projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), + sourceSeq: nonNegativeInteger(projection.lastProjectedSeq), + sourceEventId: safeId, + terminal: projection.terminal, + sealed: projection.terminal, + finalResponse: projection.finalResponse, + diagnostic, + updatedAt: timestamp + }] : [], + checkpoints: safeId ? [{ + traceId: safeId, + sessionId: resolvedSessionId, + turnId: projection.turnId ?? safeId, + runId: textValue(agentRun?.runId ?? projection.sourceRunId) || null, + commandId: textValue(agentRun?.commandId ?? projection.sourceCommandId) || null, + projectedSeq: nonNegativeInteger(projection.lastProjectedSeq), + sourceSeq: nonNegativeInteger(agentRun?.lastSeq ?? projection.lastProjectedSeq), + sourceEventId: safeId, + projectionStatus: projection.terminal ? "caught_up" : "projecting", + projectionHealth: diagnostic.projectionHealth === "unavailable" ? "unavailable" : diagnostic.projectionHealth === "stalled" ? "stalled" : "healthy", + terminal: projection.terminal, + sealed: projection.terminal, + diagnostic, + updatedAt: timestamp + }] : [] + }; +} + +function normalizeMessages(messages, context) { + if (!Array.isArray(messages)) return []; + return messages.map((message, index) => normalizeMessageFact(message, index, context)).filter(Boolean); +} + +function normalizeMessageFact(message = {}, index, { traceId, sessionId, conversationId, threadId, terminal, timestamp }) { + const messageId = textValue(message.messageId ?? message.id) || stableFactId("msg", { traceId, index, role: message.role }); + if (!sessionId || !messageId) return null; + const role = textValue(message.role) || "agent"; + const status = normalizeWorkbenchStatus(message.status ?? (terminal && role !== "user" ? "completed" : role === "user" ? "sent" : "running")); + const messageTerminal = role !== "user" && TERMINAL_STATUSES.has(status); + return { + ...message, + messageId, + sessionId, + conversationId, + threadId, + turnId: textValue(message.turnId) || traceId, + traceId: safeTraceId(message.traceId ?? traceId) ?? traceId, + role, + status, + projectedSeq: nonNegativeInteger(message.projectedSeq ?? message.seq ?? index + 1), + sourceSeq: nonNegativeInteger(message.sourceSeq ?? message.seq ?? index + 1), + sourceEventId: textValue(message.sourceEventId) || `${messageId}:0`, + terminal: messageTerminal, + sealed: messageTerminal, + text: textValue(message.text ?? message.content ?? message.message), + createdAt: timestampValue(message.createdAt ?? timestamp), + updatedAt: timestampValue(message.updatedAt ?? timestamp), + valuesPrinted: false + }; +} + +function isEmptyFacts(facts) { + return Object.values(facts).every((items) => !Array.isArray(items) || items.length === 0); +} + +function stableFactId(prefix, value) { + return `${prefix}_${createHash("sha256").update(JSON.stringify(sortJson(value))).digest("hex").slice(0, 32)}`; +} + +function sortJson(value) { + if (Array.isArray(value)) return value.map(sortJson); + if (value && typeof value === "object") return Object.fromEntries(Object.keys(value).sort().map((key) => [key, sortJson(value[key])])); + return value; +} + +function nonNegativeInteger(value) { + const parsed = Number.parseInt(value ?? "", 10); + return Number.isInteger(parsed) && parsed >= 0 ? parsed : 0; +} + +function timestampValue(value) { + const ms = Date.parse(String(value ?? "")); + return Number.isFinite(ms) ? new Date(ms).toISOString() : new Date().toISOString(); +} + +function textValue(value) { + return String(value ?? "").trim(); +} + export function appendProjectionDiagnostic(traceStore = defaultCodeAgentTraceStore, traceId, event = {}) { const safeId = safeTraceId(traceId); if (!safeId || typeof traceStore?.append !== "function") return null;