Merge pull request #1670 from pikasTech/codex/hwlab-1656-workbench-projection-writer

feat: persist workbench projection facts

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", () => {

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.
   }

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) => {

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
   });
 }
 

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({

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");
+});

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;