From efe4b62a0b8da7d3c2854d8e7fbd0e12da22d270 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 06:51:09 +0200 Subject: [PATCH] fix: make Kafka projection the sole Workbench authority --- .../cloud/server-code-agent-admission-http.ts | 12 +- .../cloud/server-code-agent-http-support.ts | 10 +- .../cloud/server-code-agent-trace-http.ts | 1 - internal/cloud/server-code-agent-turn-http.ts | 1 - .../server-workbench-launch-http.test.ts | 6 +- .../cloud/workbench-projection-writer.test.ts | 1582 ++--------------- internal/cloud/workbench-projection-writer.ts | 891 +--------- internal/db/runtime-store-memory.ts | 20 + ...e-store-postgres-kafka.integration.test.ts | 307 ++++ .../db/runtime-store-postgres-kafka.test.ts | 65 +- internal/db/runtime-store-postgres-kafka.ts | 17 +- internal/db/runtime-store-postgres.ts | 20 + 12 files changed, 660 insertions(+), 2272 deletions(-) create mode 100644 internal/db/runtime-store-postgres-kafka.integration.test.ts diff --git a/internal/cloud/server-code-agent-admission-http.ts b/internal/cloud/server-code-agent-admission-http.ts index 92addb2d..293c9a70 100644 --- a/internal/cloud/server-code-agent-admission-http.ts +++ b/internal/cloud/server-code-agent-admission-http.ts @@ -17,7 +17,7 @@ import { createCodeAgentErrorPayload, handleCodeAgentChat } from "./code-agent-c import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { messageAuthorityTextValue } from "./code-agent-agentrun-prompt.ts"; -import { writeWorkbenchProjectionSession, writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts"; +import { writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { @@ -1194,6 +1194,16 @@ async function recordCodeAgentTurnAdmission({ payload = {}, params = {}, options error.valuesRedacted = true; throw error; } + await writeWorkbenchSessionAdmissionFact({ + runtimeStore: options.runtimeStore, + session: ownerRecord, + ownerUserId: options.actor?.id ?? params.ownerUserId, + ownerRole: options.actor?.role ?? params.ownerRole, + projectId: params.projectId ?? ownerRecord.projectId ?? null, + conversationId: params.conversationId ?? ownerRecord.conversationId ?? null, + threadId: params.threadId ?? ownerRecord.threadId ?? null, + status: "running" + }); await recordCodeAgentSessionInputFact({ params, options, diff --git a/internal/cloud/server-code-agent-http-support.ts b/internal/cloud/server-code-agent-http-support.ts index f37d50af..07629b68 100644 --- a/internal/cloud/server-code-agent-http-support.ts +++ b/internal/cloud/server-code-agent-http-support.ts @@ -17,7 +17,7 @@ import { createCodeAgentErrorPayload, handleCodeAgentChat } from "./code-agent-c import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { messageAuthorityTextValue } from "./code-agent-agentrun-prompt.ts"; -import { writeWorkbenchProjectionSession, writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts"; +import { recordWorkbenchSessionOwner } from "./workbench-projection-writer.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { @@ -450,10 +450,8 @@ async function recordCodeAgentSessionOwner({ payload = {}, params = {}, options const sessionId = safeSessionId(payload.sessionId ?? session?.sessionId ?? sessionReuse?.sessionId ?? params.sessionId) || null; const conversationId = safeConversationId(payload.conversationId ?? session?.conversationId ?? sessionReuse?.conversationId ?? params.conversationId) || null; const threadId = safeOpaqueId(session?.threadId ?? sessionReuse?.threadId ?? payload.threadId ?? params.threadId) || null; - return writeWorkbenchProjectionSession({ + return recordWorkbenchSessionOwner({ accessController: options.accessController, - runtimeStore: options.runtimeStore, - traceStore: options.traceStore ?? defaultCodeAgentTraceStore, traceId, ownerUserId, ownerRole: options.actor?.role ?? params.ownerRole ?? null, @@ -463,9 +461,7 @@ async function recordCodeAgentSessionOwner({ payload = {}, params = {}, options threadId, status, preserveLastTraceId, - session: codeAgentSessionOwnerEvidence(payload, params), - payload, - params + session: codeAgentSessionOwnerEvidence(payload, params) }); } diff --git a/internal/cloud/server-code-agent-trace-http.ts b/internal/cloud/server-code-agent-trace-http.ts index 1d472bd9..84f6e35b 100644 --- a/internal/cloud/server-code-agent-trace-http.ts +++ b/internal/cloud/server-code-agent-trace-http.ts @@ -17,7 +17,6 @@ import { createCodeAgentErrorPayload, handleCodeAgentChat } from "./code-agent-c import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { messageAuthorityTextValue } from "./code-agent-agentrun-prompt.ts"; -import { writeWorkbenchProjectionSession, writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { diff --git a/internal/cloud/server-code-agent-turn-http.ts b/internal/cloud/server-code-agent-turn-http.ts index 797cedeb..ae9c5244 100644 --- a/internal/cloud/server-code-agent-turn-http.ts +++ b/internal/cloud/server-code-agent-turn-http.ts @@ -17,7 +17,6 @@ import { createCodeAgentErrorPayload, handleCodeAgentChat } from "./code-agent-c import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { codeAgentSessionLifecycleSummary } from "./code-agent-session-lifecycle.ts"; import { messageAuthorityTextValue } from "./code-agent-agentrun-prompt.ts"; -import { writeWorkbenchProjectionSession, writeWorkbenchSessionAdmissionFact } from "./workbench-projection-writer.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { codeAgentOtelTraceFields, emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { diff --git a/internal/cloud/server-workbench-launch-http.test.ts b/internal/cloud/server-workbench-launch-http.test.ts index 9d4532b0..2bb9e284 100644 --- a/internal/cloud/server-workbench-launch-http.test.ts +++ b/internal/cloud/server-workbench-launch-http.test.ts @@ -52,9 +52,9 @@ test("workbench launch records OTel stage for Project Management link write 404" } }, runtimeStore: { - async writeWorkbenchFacts(params, requestMeta) { + async writeWorkbenchSessionAdmissionFact(params, requestMeta) { factWrites.push({ params, requestMeta }); - return { ok: true, facts: params.facts }; + return { ok: true, fact: params.fact, admissionOnly: true }; } } }); @@ -86,7 +86,7 @@ test("workbench launch records OTel stage for Project Management link write 404" assert.match(context.otelAttributes["workbench.launch.task_ref_hash"], /^[0-9a-f]{24}$/u); assert.equal(context.otelAttributes["workbench.launch.values_redacted"], true); assert.equal(factWrites.length, 1); - const sessionJson = factWrites[0].params.facts.sessions[0].sessionJson; + const sessionJson = factWrites[0].params.fact.sessionJson; assert.equal(sessionJson.launchContext.sourceId, "hwlab-v03-mdtodo"); assert.equal(sessionJson.launchContext.hwpodId, "constart-71freq-c"); assert.equal(sessionJson.launchContext.contextFingerprint, "ctx_launch_otel"); diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index a3ba0bbd..5d95effc 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -1,1436 +1,220 @@ -// SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-19-p1-agentrun-incremental-cursor. -// Responsibility: Workbench Kafka projection writer durable facts regression tests. - import assert from "node:assert/strict"; import { test } from "bun:test"; -import { createCloudRuntimeStore } from "../db/runtime-store.ts"; -import { buildWorkbenchProjectionEventFacts, writeWorkbenchProjectionEvent, writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts"; +import { + buildWorkbenchProjectionEventFacts, + recordWorkbenchSessionOwner, + writeWorkbenchSessionAdmissionFact +} from "./workbench-projection-writer.ts"; -test("Kafka projection keeps assistant text visible through progress, final, and terminal status", () => { - const traceId = "trc_kafka_assistant_lifecycle"; - const sessionId = "ses_kafka_assistant_lifecycle"; - const progress = buildWorkbenchProjectionEventFacts({ - projectedSeq: 1, - projectedAt: "2026-07-10T10:00:01.000Z", - event: { traceId, sessionId, sourceSeq: 1, sourceEventId: "evt-progress", type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", assistantText: "first progress", text: "first progress", createdAt: "2026-07-10T10:00:01.000Z" } +test("HTTP lifecycle records ownership without writing projection facts", async () => { + const calls = []; + const accessController = { + async recordAgentSessionOwner(input) { + calls.push(input); + return { id: input.sessionId, ownerUserId: input.ownerUserId, status: input.status }; + } + }; + let projectionWrites = 0; + const owner = await recordWorkbenchSessionOwner({ + accessController, + runtimeStore: { writeWorkbenchFacts() { projectionWrites += 1; } }, + traceId: "trc_owner_only", + ownerUserId: "usr_owner", + ownerRole: "user", + sessionId: "ses_owner_only", + projectId: "prj_owner", + status: "completed", + session: { source: "http-result" } }); - const toolProgress = buildWorkbenchProjectionEventFacts({ - projectedSeq: 2, - projectedAt: "2026-07-10T10:00:02.000Z", - previousCheckpoint: progress.facts.checkpoints[0], - event: { traceId, sessionId, sourceSeq: 2, sourceEventId: "evt-tool-progress", type: "tool", eventType: "tool", status: "running", label: "agentrun:tool:completed", createdAt: "2026-07-10T10:00:02.000Z" } + + assert.equal(owner.id, "ses_owner_only"); + assert.equal(calls.length, 1); + assert.equal(calls[0].traceId, "trc_owner_only"); + assert.equal(projectionWrites, 0); +}); + +test("admission persists only the dedicated session scope fact", async () => { + const calls = []; + const runtimeStore = { + async writeWorkbenchSessionAdmissionFact(params, meta) { + calls.push({ params, meta }); + return { written: true, admissionOnly: true }; + }, + writeWorkbenchFacts() { + throw new Error("generic projection writer must not be used by admission"); + } + }; + const result = await writeWorkbenchSessionAdmissionFact({ + runtimeStore, + session: { + id: "ses_admission_only", + ownerUserId: "usr_owner", + ownerRole: "user", + projectId: "prj_owner", + conversationId: "cnv_owner", + lastTraceId: "trc_admission_only", + status: "running", + session: { source: "manual-session-create", launchContext: { source: "workbench" } }, + createdAt: "2026-07-10T10:00:00.000Z", + updatedAt: "2026-07-10T10:00:01.000Z" + } }); - const final = buildWorkbenchProjectionEventFacts({ - projectedSeq: 3, - projectedAt: "2026-07-10T10:00:03.000Z", - previousCheckpoint: toolProgress.facts.checkpoints[0], - event: { traceId, sessionId, sourceSeq: 3, sourceEventId: "evt-final", type: "assistant", eventType: "assistant", status: "completed", label: "agentrun:assistant:message", assistantText: "final answer", text: "final answer", finalResponse: { text: "final answer", status: "completed", traceId }, terminal: true, createdAt: "2026-07-10T10:00:03.000Z" } - }); - const terminal = buildWorkbenchProjectionEventFacts({ + + assert.equal(result.admissionOnly, true); + assert.equal(calls.length, 1); + const fact = calls[0].params.fact; + assert.equal(fact.sessionId, "ses_admission_only"); + assert.equal(fact.ownerUserId, "usr_owner"); + assert.equal(fact.lastTraceId, "trc_admission_only"); + assert.equal(fact.projectedSeq, 0); + assert.equal(fact.sourceEventId, "ses_admission_only:session-admission"); + assert.deepEqual(fact.sessionJson.launchContext, { source: "workbench", valuesRedacted: true }); +}); + +test("admission fails closed when the dedicated durable entry is absent", async () => { + await assert.rejects( + writeWorkbenchSessionAdmissionFact({ runtimeStore: { writeWorkbenchFacts() {} }, session: { id: "ses_missing_entry" } }), + (error) => error?.code === "workbench_admission_store_unconfigured" + ); +}); + +test("Kafka projector builds running facts from one canonical source event", () => { + const built = buildWorkbenchProjectionEventFacts({ projectedSeq: 4, + projectedAt: "2026-07-10T10:00:02.000Z", + event: { + traceId: "trc_running", + sessionId: "ses_running", + runId: "run_running", + commandId: "cmd_running", + sourceEventId: "evt_running", + sourceSeq: 7, + type: "assistant", + eventType: "assistant", + status: "running", + text: "committed delta", + createdAt: "2026-07-10T10:00:01.000Z" + } + }); + + assert.equal(built.written, true); + assert.equal(built.facts.sessions[0].lastTraceId, "trc_running"); + assert.equal(built.facts.messages[0].text, "committed delta"); + assert.equal(built.facts.turns[0].terminal, false); + assert.equal(built.facts.checkpoints[0].projectedSeq, 4); + assert.equal(built.facts.traceEvents[0].sourceEventId, "evt_running"); +}); + +test("authoritative assistant final seals terminal with a final response", () => { + const built = buildWorkbenchProjectionEventFacts({ + projectedSeq: 5, projectedAt: "2026-07-10T10:00:04.000Z", - previousCheckpoint: final.facts.checkpoints[0], - event: { traceId, sessionId, sourceSeq: 4, sourceEventId: "evt-terminal", type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", message: "AgentRun terminal status completed", terminal: true, createdAt: "2026-07-10T10:00:04.000Z" } - }); - - assert.equal(progress.facts.messages[0].text, "first progress"); - assert.equal(progress.facts.parts[0].text, "first progress"); - assert.equal(progress.facts.checkpoints[0].assistantText, "first progress"); - assert.equal(progress.facts.turns[0].status, "running"); - assert.equal(progress.facts.turns[0].terminal, false); - assert.equal(toolProgress.facts.messages[0].text, "first progress"); - assert.equal(toolProgress.facts.parts[0].text, "first progress"); - assert.equal(toolProgress.facts.checkpoints[0].assistantText, "first progress"); - assert.equal(final.facts.messages[0].text, "final answer"); - assert.equal(final.facts.parts[0].partType, "final_response"); - assert.equal(terminal.facts.messages[0].text, "final answer"); - assert.equal(terminal.facts.parts[0].text, "final answer"); - assert.equal(terminal.facts.checkpoints[0].finalResponse.text, "final answer"); - assert.equal(terminal.facts.turns[0].terminal, true); - assert.equal(terminal.facts.turns[0].sealed, true); - assert.equal(terminal.facts.turns[0].finalResponse.text, "final answer"); - assert.equal(terminal.facts.turns[0].assistantText, "final answer"); -}); - -test("Kafka projection seals failed terminal turn with a failure response", () => { - const result = buildWorkbenchProjectionEventFacts({ - projectedSeq: 1, - projectedAt: "2026-07-10T10:01:00.000Z", - event: { traceId: "trc_kafka_failed", sessionId: "ses_kafka_failed", sourceSeq: 1, sourceEventId: "evt-kafka-failed", type: "result", eventType: "terminal", status: "failed", label: "agentrun:terminal:failed", failureKind: "provider-stream-disconnected", terminal: true, createdAt: "2026-07-10T10:01:00.000Z" } - }); - - assert.equal(result.facts.turns[0].terminal, true); - assert.equal(result.facts.turns[0].sealed, true); - assert.equal(result.facts.turns[0].failureKind, "provider-stream-disconnected"); - assert.match(result.facts.turns[0].finalResponse.text, /provider-stream-disconnected/u); -}); - -test("workbench projection writer preserves Workbench launch context on Code Agent session facts", async () => { - const factWrites = []; - const accessWrites = []; - const launchContext = { - source: "project-management", - projectId: "project_constart_71freq", - taskRef: "mdtodo:constart-71freq-mdtodo:file_c5abae4e69c60370:R1", - sourceId: "constart-71freq-mdtodo", - sourceKind: "hwpod-workspace", - fileRef: "file_c5abae4e69c60370", - taskId: "R1", - hwpodId: "constart-71freq-c", - nodeId: "node-d601-f103-v2", - mdtodoRootRef: "docs/MDTODO", - workspaceRootHash: "workspace-hash", - hwpodWorkspaceArgs: "--hwpod-id constart-71freq-c --workspace-path 'F:\\Work\\ConStart'", - contextFingerprint: "ctx_projection_launch", - executionContext: { - sourceId: "constart-71freq-mdtodo", - sourceKind: "hwpod-workspace", - projectId: "project_constart_71freq", - taskRef: "mdtodo:constart-71freq-mdtodo:file_c5abae4e69c60370:R1", - fileRef: "file_c5abae4e69c60370", - taskId: "R1", - hwpodId: "constart-71freq-c", - nodeId: "node-d601-f103-v2", - mdtodoRootRef: "docs/MDTODO", - workspaceRootHash: "workspace-hash", - hwpodWorkspaceArgs: "--hwpod-id constart-71freq-c --workspace-path 'F:\\Work\\ConStart'", - contextFingerprint: "ctx_projection_launch", - valuesRedacted: true + previousCheckpoint: { + traceId: "trc_terminal", + sessionId: "ses_terminal", + projectedSeq: 4, + sourceSeq: 4, + startedAt: "2026-07-10T10:00:00.000Z", + lastEventAt: "2026-07-10T10:00:03.000Z", + timing: { startedAt: "2026-07-10T10:00:00.000Z", lastEventAt: "2026-07-10T10:00:03.000Z", finishedAt: null, durationMs: null } }, - valuesRedacted: true - }; - 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-28T11:15:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_projection_launch_context", - ownerUserId: "usr_projection_launch", - ownerRole: "admin", - sessionId: "ses_projection_launch_context", - projectId: "project_constart_71freq", - conversationId: "cnv_projection_launch_context", - threadId: "thread-projection-launch-context", - status: "running", - payload: { - traceId: "trc_projection_launch_context", - status: "running", - updatedAt: "2026-06-28T11:15:00.000Z" - }, - params: { - projectId: "project_constart_71freq", - taskRef: "mdtodo:constart-71freq-mdtodo:file_c5abae4e69c60370:R1", - message: "run this MDTODO task", - launchContext - }, - session: { - sessionStatus: "running", - messages: [ - { messageId: "msg_projection_launch_user", role: "user", text: "run this MDTODO task", status: "sent", turnId: "trc_projection_launch_context", traceId: "trc_projection_launch_context" } - ], - valuesRedacted: true, - secretMaterialStored: false - } - }); - - assert.equal(accessWrites.length, 1); - assert.equal(factWrites.length, 1); - const sessionJson = factWrites[0].params.facts.sessions[0].sessionJson; - assert.equal(sessionJson.launchContext.sourceId, "constart-71freq-mdtodo"); - assert.equal(sessionJson.launchContext.hwpodId, "constart-71freq-c"); - assert.equal(sessionJson.launchContext.contextFingerprint, "ctx_projection_launch"); - assert.equal(sessionJson.launchContext.executionContext.hwpodWorkspaceArgs, "--hwpod-id constart-71freq-c --workspace-path 'F:\\Work\\ConStart'"); - assert.equal(sessionJson.launchContext.valuesRedacted, true); - assert.equal(sessionJson.secretMaterialStored, false); -}); - -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", - runnerTrace: { - traceId: "trc_writer_terminal", - status: "completed", - startedAt: "2026-06-20T10:59:30.000Z", - lastEventAt: "2026-06-20T11:00:00.000Z", - finishedAt: "2026-06-20T11:00:00.000Z", - elapsedMs: 30000 - }, - agentRun: { runId: "run_writer_terminal", commandId: "cmd_writer_terminal", status: "completed", terminalStatus: "completed", 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.parts.some((part) => part.messageId === "msg_writer_agent" && part.partType === "final_response" && 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.messages.find((message) => message.messageId === "msg_writer_agent").lastEventAt, "2026-06-20T11:00:00.000Z"); - assert.equal(facts.messages.find((message) => message.messageId === "msg_writer_agent").durationMs, 30000); - assert.equal(facts.turns[0].startedAt, "2026-06-20T10:59:30.000Z"); - assert.equal(facts.turns[0].durationMs, 30000); - assert.equal(facts.checkpoints[0].projectionStatus, "caught_up"); - assert.equal(facts.checkpoints[0].sealed, true); - assert.equal(facts.checkpoints[0].timing.finishedAt, "2026-06-20T11:00:00.000Z"); -}); - -test("workbench projection writer keeps completed AgentRun without final response unsealed", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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:02:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_terminal_missing_final", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_terminal_missing_final", - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_missing_final", - threadId: "thread-writer-terminal-missing-final", - status: "completed", - payload: { - traceId: "trc_writer_terminal_missing_final", - status: "completed", - runnerTrace: { - traceId: "trc_writer_terminal_missing_final", - status: "completed", - startedAt: "2026-06-20T11:01:30.000Z", - lastEventAt: "2026-06-20T11:02:00.000Z", - finishedAt: "2026-06-20T11:02:00.000Z", - events: [ - { seq: 1, type: "assistant_message", status: "running", message: "progress only" }, - { seq: 2, type: "result", status: "completed", terminal: true, label: "agentrun:terminal:completed" } - ], - eventCount: 2 - }, - agentRun: { runId: "run_writer_terminal_missing_final", commandId: "cmd_writer_terminal_missing_final", status: "completed", terminalStatus: "completed", lastSeq: 2 }, - updatedAt: "2026-06-20T11:02:00.000Z" - }, - session: { - sessionStatus: "completed", - messages: [ - { messageId: "msg_writer_missing_final_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal_missing_final", traceId: "trc_writer_terminal_missing_final" }, - { messageId: "msg_writer_missing_final_agent", role: "agent", text: "", status: "completed", turnId: "trc_writer_terminal_missing_final", traceId: "trc_writer_terminal_missing_final" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_missing_final_agent"); - assert.equal(facts.sessions[0].status, "running"); - assert.equal(facts.sessions[0].terminal, false); - assert.equal(facts.sessions[0].sealed, false); - assert.equal(agentMessage.status, "running"); - assert.equal(agentMessage.terminal, false); - assert.equal(agentMessage.sealed, false); - assert.equal(facts.parts.some((part) => part.partType === "final_response"), false); - assert.equal(facts.turns[0].status, "running"); - assert.equal(facts.turns[0].terminal, false); - assert.equal(facts.turns[0].sealed, false); - assert.equal(facts.turns[0].finalResponse, null); - assert.equal(facts.turns[0].diagnostic.projectionStatus, "projecting"); - assert.equal(facts.turns[0].diagnostic.waitingFor, "final_response"); - assert.equal(facts.checkpoints[0].projectionStatus, "projecting"); - assert.equal(facts.checkpoints[0].terminal, false); - assert.equal(facts.checkpoints[0].sealed, false); -}); - -test("workbench projection writer seals failed AgentRun turns with failure final response", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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:03:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_terminal_failed", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_terminal_failed", - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_failed", - threadId: "thread-writer-terminal-failed", - status: "failed", - payload: { - traceId: "trc_writer_terminal_failed", - status: "failed", - error: { code: "provider-stream-disconnected", message: "provider stream disconnected" }, - runnerTrace: { - traceId: "trc_writer_terminal_failed", - status: "failed", - startedAt: "2026-06-20T11:02:30.000Z", - lastEventAt: "2026-06-20T11:03:00.000Z", - finishedAt: "2026-06-20T11:03:00.000Z", - events: [ - { seq: 1, type: "assistant_message", status: "running", message: "progress only" }, - { seq: 2, type: "result", status: "failed", terminal: true, label: "agentrun:terminal:failed", errorCode: "provider-stream-disconnected" } - ], - eventCount: 2 - }, - agentRun: { runId: "run_writer_terminal_failed", commandId: "cmd_writer_terminal_failed", status: "failed", terminalStatus: "failed", failureKind: "provider-stream-disconnected", lastSeq: 2 }, - updatedAt: "2026-06-20T11:03:00.000Z" - }, - session: { - sessionStatus: "failed", - messages: [ - { messageId: "msg_writer_failed_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_terminal_failed", traceId: "trc_writer_terminal_failed" }, - { messageId: "msg_writer_failed_agent", role: "agent", text: "", status: "failed", turnId: "trc_writer_terminal_failed", traceId: "trc_writer_terminal_failed" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_failed_agent"); - const finalPart = facts.parts.find((part) => part.messageId === "msg_writer_failed_agent" && part.partType === "final_response"); - assert.equal(facts.sessions[0].status, "failed"); - assert.equal(facts.sessions[0].terminal, true); - assert.equal(facts.sessions[0].sealed, true); - assert.equal(agentMessage.status, "failed"); - assert.equal(agentMessage.terminal, true); - assert.equal(agentMessage.sealed, true); - assert.match(agentMessage.text, /provider-stream-disconnected/u); - assert.equal(finalPart?.status, "failed"); - assert.match(finalPart?.text ?? "", /provider-stream-disconnected/u); - assert.equal(facts.turns[0].status, "failed"); - assert.equal(facts.turns[0].terminal, true); - assert.equal(facts.turns[0].sealed, true); - assert.match(facts.turns[0].finalResponse.text, /provider-stream-disconnected/u); - assert.equal(facts.turns[0].diagnostic.projectionStatus, "caught-up"); - assert.equal(facts.turns[0].diagnostic.projectionHealth, "healthy"); - assert.equal(facts.turns[0].diagnostic.blocker, null); - assert.equal(facts.checkpoints[0].projectionStatus, "caught_up"); - assert.equal(facts.checkpoints[0].terminal, true); - assert.equal(facts.checkpoints[0].sealed, true); - assert.match(facts.checkpoints[0].finalResponse.text, /provider-stream-disconnected/u); -}); - -test("workbench projection writer does not synthesize terminal duration from updatedAt", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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:05:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_terminal_no_finished_at", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_terminal_no_finished_at", - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_no_finished_at", - threadId: "thread-writer-terminal-no-finished-at", - status: "canceled", - payload: { - traceId: "trc_writer_terminal_no_finished_at", - status: "canceled", - startedAt: "2026-06-20T11:00:00.000Z", - updatedAt: "2026-06-20T11:05:00.000Z", - agentRun: { runId: "run_writer_terminal_no_finished_at", commandId: "cmd_writer_terminal_no_finished_at", terminalStatus: "canceled" } - }, - session: { - sessionStatus: "canceled", - messages: [ - { messageId: "msg_writer_terminal_no_finished_user", role: "user", text: "cancel me", status: "sent", turnId: "trc_writer_terminal_no_finished_at", traceId: "trc_writer_terminal_no_finished_at" }, - { messageId: "msg_writer_terminal_no_finished_agent", role: "agent", text: "hwlab-user-cancel", status: "canceled", turnId: "trc_writer_terminal_no_finished_at", traceId: "trc_writer_terminal_no_finished_at" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_terminal_no_finished_agent"); - assert.equal(facts.turns[0].status, "canceled"); - assert.equal(facts.turns[0].finishedAt, null); - assert.equal(facts.turns[0].durationMs, null); - assert.equal(agentMessage.finishedAt, null); - assert.equal(agentMessage.durationMs, null); - assert.equal(facts.checkpoints[0].timing.finishedAt, null); - assert.equal(facts.checkpoints[0].timing.durationMs, null); - assert.equal(facts.checkpoints[0].diagnostic.blocker.code, "workbench_terminal_timing_authority_missing"); -}); - -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 facts durable without trace backfill queries", 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, 1); - 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[0].params.facts.traceEvents?.length ?? 0, 0); - assert.equal(allocationSeq, 0); -}); - -test("workbench projection writer does not synthesize terminal 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", "turns", "messages"], limit: 20 }); - const events = loaded.facts.traceEvents; - assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2, 3]); - assert.deepEqual(events.map((event) => event.label), ["agentrun:backend:admitted", "agentrun:run:created", "agentrun:runner-job:queued"]); - assert.equal(events.some((event) => event.source === "workbench-terminal-projection"), false); - const terminalTurn = loaded.facts.turns.find((turn) => turn.terminal === true); - assert.equal(terminalTurn.finalResponse.text, "BACKFILL_DONE"); - assert.equal(loaded.facts.messages.some((message) => message.role !== "user" && message.text === "BACKFILL_DONE"), true); -}); - -test("workbench projection terminal facts do not advance checkpoint past durable trace events", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:45:00.000Z" }); - const traceId = "trc_writer_terminal_checkpoint_gap"; - const sessionId = "ses_writer_terminal_checkpoint_gap"; - const finalText = "CHECKPOINT_GAP_DONE"; - 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:45:00.000Z" - }; - } - }; - - for (let sourceSeq = 1; sourceSeq <= 33; sourceSeq += 1) { - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId, - source: "agentrun", - sourceSeq, - type: sourceSeq === 33 ? "terminal" : "backend", - eventType: sourceSeq === 33 ? "terminal" : "backend", - status: sourceSeq === 33 ? "completed" : "running", - label: sourceSeq === 33 ? "agentrun:result:completed" : `agentrun:backend:${sourceSeq}`, - terminal: sourceSeq === 33, - runId: "run_writer_terminal_checkpoint_gap", - commandId: "cmd_writer_terminal_checkpoint_gap", - createdAt: `2026-06-20T11:44:${String(sourceSeq).padStart(2, "0")}.000Z` - } - }); - } - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId, - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId, - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_checkpoint_gap", - threadId: "thread-writer-terminal-checkpoint-gap", - status: "completed", - payload: { - traceId, - status: "completed", - finalResponse: { text: finalText, status: "completed", traceId }, - runnerTrace: { - traceId, - status: "completed", - eventCount: 34, - events: Array.from({ length: 34 }, (_, index) => ({ seq: index + 1, type: "backend", status: index === 33 ? "completed" : "running", terminal: index === 33 })) - }, - agentRun: { runId: "run_writer_terminal_checkpoint_gap", commandId: "cmd_writer_terminal_checkpoint_gap", status: "completed", terminalStatus: "completed", lastSeq: 30 }, - updatedAt: "2026-06-20T11:45:00.000Z" - }, - session: { - sessionStatus: "completed", - messages: [ - { messageId: "msg_writer_checkpoint_gap_user", role: "user", text: "question", status: "sent", turnId: traceId, traceId }, - { messageId: "msg_writer_checkpoint_gap_agent", role: "agent", text: finalText, status: "completed", turnId: traceId, traceId } - ], - finalResponse: { text: finalText, status: "completed", traceId } - } - }); - - const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints", "turns"], limit: 80 }); - assert.equal(loaded.facts.traceEvents.length, 33); - assert.equal(Math.max(...loaded.facts.traceEvents.map((event) => event.projectedSeq)), 33); - assert.equal(loaded.facts.checkpoints[0].projectedSeq, 33); - assert.equal(loaded.facts.checkpoints[0].sourceSeq, 30); - assert.equal(loaded.facts.turns[0].terminal, true); -}); - -test("workbench projection writer does not duplicate terminal trace events that already exist", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:25:00.000Z" }); - const traceId = "trc_writer_terminal_backfill_existing"; - 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:25:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId: "ses_writer_terminal_backfill_existing", - sourceSeq: 1, - type: "backend", - status: "running", - label: "agentrun:backend:admitted", - createdAt: "2026-06-20T11:24:01.000Z" - } - }); - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId: "ses_writer_terminal_backfill_existing", - sourceSeq: 2, + traceId: "trc_terminal", + sessionId: "ses_terminal", + sourceEventId: "evt_terminal", + sourceSeq: 5, type: "assistant", eventType: "assistant", status: "completed", - label: "agentrun:assistant:message", - text: "ALREADY_DONE", terminal: true, - createdAt: "2026-06-20T11:24:12.000Z" + replyAuthority: true, + text: "durable final answer", + createdAt: "2026-07-10T10:00:04.000Z" } }); - await writeWorkbenchProjectionEvent({ - runtimeStore, + + const turn = built.facts.turns[0]; + assert.equal(turn.terminal, true); + assert.equal(turn.sealed, true); + assert.equal(turn.finalResponse.text, "durable final answer"); + assert.equal(built.facts.messages[0].terminal, true); + assert.equal(built.facts.parts[0].partType, "final_response"); + assert.equal(built.facts.checkpoints[0].projectionStatus, "caught_up"); +}); + +test("completed terminal without final response remains visibly unsealed", () => { + const built = buildWorkbenchProjectionEventFacts({ + projectedSeq: 1, event: { - traceId, - sessionId: "ses_writer_terminal_backfill_existing", - sourceSeq: 3, + traceId: "trc_missing_final", + sessionId: "ses_missing_final", + sourceEventId: "evt_missing_final", + sourceSeq: 1, type: "result", eventType: "terminal", status: "completed", - label: "agentrun:result:completed", terminal: true, - createdAt: "2026-06-20T11:24:13.000Z" + createdAt: "2026-07-10T10:00:00.000Z" } }); - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId, - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_terminal_backfill_existing", - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_backfill_existing", - threadId: "thread-writer-terminal-backfill-existing", - status: "completed", - payload: { - traceId, - status: "completed", - finalResponse: { text: "ALREADY_DONE", status: "completed", traceId }, - runnerTrace: { - traceId, - status: "completed", - startedAt: "2026-06-20T11:24:01.000Z", - lastEventAt: "2026-06-20T11:24:13.000Z", - finishedAt: "2026-06-20T11:24:13.000Z", - eventCount: 3, - events: [] - }, - agentRun: { runId: "run_writer_terminal_backfill_existing", commandId: "cmd_writer_terminal_backfill_existing", status: "completed", terminalStatus: "completed", lastSeq: 3 }, - updatedAt: "2026-06-20T11:24:13.000Z" - }, - session: { - sessionStatus: "completed", - messages: [ - { messageId: "msg_writer_backfill_existing_user", role: "user", text: "question", status: "sent", turnId: traceId, traceId }, - { messageId: "msg_writer_backfill_existing_agent", role: "agent", text: "ALREADY_DONE", status: "completed", turnId: traceId, traceId } - ], - finalResponse: { text: "ALREADY_DONE", status: "completed", traceId } - } - }); - - const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints"], limit: 20 }); - assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2, 3]); - assert.deepEqual(loaded.facts.traceEvents.map((event) => event.label), ["agentrun:backend:admitted", "agentrun:assistant:message", "agentrun:result:completed"]); - assert.equal(loaded.facts.checkpoints[0].projectedSeq, 3); + assert.equal(built.facts.turns[0].terminal, false); + assert.equal(built.facts.turns[0].finalResponse, null); + assert.equal(built.facts.turns[0].diagnostic.terminalSealBlocked, true); + assert.equal(built.facts.checkpoints[0].projectionHealth, "degraded"); + assert.equal(built.facts.checkpoints[0].sealed, false); }); -test("workbench projection event writer aggregates assistant updates and skips late non-terminal rows after seal", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:32:00.000Z" }); - const traceId = "trc_writer_assistant_aggregate"; - const sessionId = "ses_writer_assistant_aggregate"; - - await writeWorkbenchProjectionEvent({ - runtimeStore, +test("late running event cannot overwrite a sealed terminal", () => { + const built = buildWorkbenchProjectionEventFacts({ + projectedSeq: 10, + previousCheckpoint: { + traceId: "trc_sealed", + sessionId: "ses_sealed", + projectedSeq: 9, + sourceSeq: 9, + terminal: true, + sealed: true, + finalResponse: { text: "sealed answer", status: "completed" } + }, event: { - traceId, - sessionId, - source: "agentrun", - sourceSeq: 1, + traceId: "trc_sealed", + sessionId: "ses_sealed", + sourceEventId: "evt_late", + sourceSeq: 8, type: "backend", - eventType: "backend", - status: "running", - label: "agentrun:backend:running", - createdAt: "2026-06-20T11:31:00.000Z" + status: "running" } }); - await writeWorkbenchProjectionEvent({ - runtimeStore, + + assert.equal(built.written, false); + assert.equal(built.suppressedAfterSeal, true); + assert.equal(built.projectedSeq, 9); + assert.equal(built.sourceEventId, "evt_late"); + assert.equal("facts" in built, false); +}); + +test("same canonical source event derives deterministic fact identities", () => { + const input = { + projectedSeq: 3, + projectedAt: "2026-07-10T10:00:03.000Z", event: { - traceId, - sessionId, - source: "agentrun", - sourceSeq: 25, + traceId: "trc_deterministic", + sessionId: "ses_deterministic", + sourceEventId: "evt_deterministic", + sourceSeq: 3, type: "assistant", eventType: "assistant", status: "running", - label: "agentrun:assistant:message", - text: "AGGREGATED_FINAL", - createdAt: "2026-06-20T11:31:10.000Z" + text: "same", + createdAt: "2026-07-10T10:00:02.000Z" } - }); - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId, - source: "agentrun", - sourceSeq: 26, - type: "assistant", - eventType: "assistant", - status: "completed", - label: "agentrun:assistant:message", - text: "AGGREGATED_FINAL", - finalResponse: { text: "AGGREGATED_FINAL", status: "completed", traceId }, - terminal: true, - createdAt: "2026-06-20T11:31:12.000Z" - } - }); - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId, - source: "agentrun", - sourceSeq: 24, - type: "assistant", - eventType: "assistant", - status: "running", - label: "agentrun:assistant:message", - text: "AGGREGATED_FINAL", - createdAt: "2026-06-20T11:31:09.000Z" - } - }); + }; + const first = buildWorkbenchProjectionEventFacts(input); + const second = buildWorkbenchProjectionEventFacts(input); - const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["traceEvents", "messages", "checkpoints"], limit: 20 }); - const events = loaded.facts.traceEvents; - assert.equal(events.length, 2); - assert.deepEqual(events.map((event) => event.projectedSeq), [1, 2]); - assert.equal(events[1].sourceEventId, `${traceId}:assistant-message:msg_writer_assistant_aggregate_agent`); - assert.equal(events[1].status, "completed"); - assert.equal(events[1].text, "AGGREGATED_FINAL"); - assert.equal(events.some((event) => event.sourceSeq === 24), false); - assert.equal(loaded.facts.messages.filter((message) => message.role !== "user").length, 1); - assert.equal(loaded.facts.checkpoints[0].terminal, true); -}); - -test("workbench projection writer does not rewrite prior-turn messages from merged owner session", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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, - updatedAt: "2026-06-20T11:10:00.000Z", - session: { - sessionStatus: "completed", - lastTraceId: input.traceId, - messages: [ - { messageId: "msg_writer_r1_user", role: "user", text: "round1 prompt", status: "sent", turnId: "trc_writer_r1", traceId: "trc_writer_r1" }, - { messageId: "msg_writer_r1_agent", role: "agent", text: "WRONG_ROUND2_TEXT", status: "completed", turnId: "trc_writer_r1", traceId: "trc_writer_r1" }, - ...input.session.messages - ] - } - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_r2", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_multi_turn", - projectId: "prj_writer", - conversationId: "cnv_writer_multi_turn", - threadId: "thread-writer-multi-turn", - status: "completed", - payload: { - traceId: "trc_writer_r2", - status: "completed", - finalResponse: { text: "ROUND2_DONE", status: "completed", traceId: "trc_writer_r2" }, - agentRun: { runId: "run_writer_r2", commandId: "cmd_writer_r2", status: "completed", terminalStatus: "completed", lastSeq: 9 }, - updatedAt: "2026-06-20T11:10:00.000Z" - }, - session: { - sessionStatus: "completed", - messages: [ - { messageId: "msg_writer_r2_user", role: "user", text: "round2 prompt", status: "sent", turnId: "trc_writer_r2", traceId: "trc_writer_r2" }, - { messageId: "msg_writer_r2_agent", role: "agent", text: "ROUND2_DONE", status: "completed", turnId: "trc_writer_r2", traceId: "trc_writer_r2" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - assert.deepEqual(facts.messages.map((message) => message.messageId), ["msg_writer_r2_user", "msg_writer_r2_agent"]); - assert.equal(facts.messages.some((message) => message.traceId === "trc_writer_r1"), false); - assert.equal(facts.messages.find((message) => message.messageId === "msg_writer_r2_agent").text, "ROUND2_DONE"); - assert.equal(facts.turns[0].traceId, "trc_writer_r2"); - assert.equal(facts.turns[0].finalResponse.text, "ROUND2_DONE"); -}); - -test("workbench projection writer seals canceled AgentRun turns with canonical cancel final response", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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:02:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_canceled", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_canceled", - projectId: "prj_writer", - conversationId: "cnv_writer_canceled", - threadId: "thread-writer-canceled", - status: "canceled", - payload: { - traceId: "trc_writer_canceled", - status: "canceled", - finalResponse: { text: "command cancelled before backend start", status: "canceled", traceId: "trc_writer_canceled" }, - userMessage: "当前 AgentRun 请求已取消；traceId/runId/commandId 已保留，可重试上一条消息。", - runnerTrace: { - traceId: "trc_writer_canceled", - status: "canceled", - events: [{ type: "cancel", status: "canceled", terminal: true, createdAt: "2026-06-20T11:02:00.000Z" }], - updatedAt: "2026-06-20T11:02:00.000Z" - }, - agentRun: { runId: "run_writer_canceled", commandId: "cmd_writer_canceled", status: "cancelled", commandState: "cancelled", terminalStatus: "cancelled", lastSeq: 12 }, - updatedAt: "2026-06-20T11:02:00.000Z" - }, - session: { - sessionStatus: "canceled", - messages: [ - { messageId: "msg_writer_canceled_user", role: "user", text: "cancel me", status: "sent", turnId: "trc_writer_canceled", traceId: "trc_writer_canceled" }, - { messageId: "msg_writer_canceled_agent", role: "agent", text: "", status: "canceled", turnId: "trc_writer_canceled", traceId: "trc_writer_canceled" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_canceled_agent"); - assert.equal(facts.sessions[0].status, "canceled"); - assert.equal(facts.turns[0].terminal, true); - assert.equal(facts.turns[0].finalResponse.text, "hwlab-user-cancel"); - assert.equal(facts.turns[0].assistantText, "hwlab-user-cancel"); - assert.equal(agentMessage.text, "hwlab-user-cancel"); - assert.equal(facts.parts.some((part) => part.messageId === "msg_writer_canceled_agent" && part.partType === "final_response" && part.text === "hwlab-user-cancel" && part.sealed === true), true); -}); - -test("workbench projection writer does not replace prior user prompt with cancel notice", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:04:00.000Z" }); - const traceId = "trc_writer_canceled_prompt_preserved"; - 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: input.status === "canceled" ? "2026-06-20T11:04:30.000Z" : "2026-06-20T11:04:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId, - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_canceled_prompt_preserved", - projectId: "prj_writer", - conversationId: "cnv_writer_canceled_prompt_preserved", - threadId: "thread-writer-canceled-prompt-preserved", - status: "running", - payload: { - traceId, - status: "running", - prompt: "ORIGINAL CANCEL PROMPT", - updatedAt: "2026-06-20T11:04:00.000Z" - }, - session: { sessionStatus: "running", lastTraceId: traceId } - }); - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId: "ses_writer_canceled_prompt_preserved", - source: "agentrun", - sourceSeq: 1, - type: "backend", - eventType: "backend", - status: "running", - label: "agentrun:request:accepted", - createdAt: "2026-06-20T11:04:10.000Z" - } - }); - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId, - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_canceled_prompt_preserved", - projectId: "prj_writer", - conversationId: "cnv_writer_canceled_prompt_preserved", - threadId: "thread-writer-canceled-prompt-preserved", - status: "canceled", - payload: { - traceId, - status: "canceled", - message: "当前 AgentRun 请求已取消；traceId/runId/commandId 已保留，可重试上一条消息。", - finalResponse: { text: "command cancelled before backend start", status: "canceled", traceId }, - runnerTrace: { - traceId, - status: "canceled", - events: [{ type: "cancel", status: "canceled", terminal: true, createdAt: "2026-06-20T11:04:30.000Z" }], - updatedAt: "2026-06-20T11:04:30.000Z" - }, - agentRun: { runId: "run_writer_canceled_prompt_preserved", commandId: "cmd_writer_canceled_prompt_preserved", status: "cancelled", commandState: "cancelled", terminalStatus: "cancelled", lastSeq: 12 }, - updatedAt: "2026-06-20T11:04:30.000Z" - }, - session: { - sessionStatus: "canceled", - lastTraceId: traceId, - messages: [ - { messageId: "msg_writer_canceled_prompt_preserved_user", role: "user", text: "当前 AgentRun 请求已取消；traceId/runId/commandId 已保留，可重试上一条消息。", status: "sent", turnId: traceId, traceId }, - { messageId: "msg_writer_canceled_prompt_preserved_agent", role: "agent", text: "", status: "canceled", turnId: traceId, traceId } - ] - } - }); - - const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["messages", "parts", "turns"], limit: 20 }); - const userMessages = loaded.facts.messages.filter((message) => message.role === "user"); - const agentMessage = loaded.facts.messages.find((message) => message.role !== "user"); - assert.deepEqual(userMessages.map((message) => message.text), ["ORIGINAL CANCEL PROMPT"]); - assert.equal(agentMessage.text, "hwlab-user-cancel"); - assert.equal(loaded.facts.parts.some((part) => part.messageId === "msg_writer_canceled_prompt_preserved_user" && part.text.includes("请求已取消")), false); -}); - -test("workbench projection writer does not seal running assistant text as final response", async () => { - const factWrites = []; - const runtimeStore = { - async writeWorkbenchFacts(params, requestMeta) { - factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; - } - }; - 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:05:00.000Z" - }; - } - }; - - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId: "trc_writer_running", - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_running", - projectId: "prj_writer", - conversationId: "cnv_writer_running", - threadId: "thread-writer-running", - status: "running", - payload: { - traceId: "trc_writer_running", - status: "running", - assistantText: "backend diagnostic text should not become final", - updatedAt: "2026-06-20T11:05:00.000Z" - }, - session: { - sessionStatus: "running", - messages: [ - { messageId: "msg_writer_running_user", role: "user", text: "question", status: "sent", turnId: "trc_writer_running", traceId: "trc_writer_running" }, - { messageId: "msg_writer_running_agent", role: "agent", text: "", status: "running", turnId: "trc_writer_running", traceId: "trc_writer_running" } - ] - } - }); - - assert.equal(factWrites.length, 1); - const facts = factWrites[0].params.facts; - const agentMessage = facts.messages.find((message) => message.messageId === "msg_writer_running_agent"); - assert.equal(agentMessage.text, ""); - assert.equal(facts.parts.some((part) => part.messageId === "msg_writer_running_agent"), false); - assert.equal(facts.turns[0].terminal, false); - assert.equal(facts.turns[0].finalResponse, null); - assert.equal(facts.turns[0].assistantText, null); -}); - -test("workbench projection event commit writes trace event and checkpoint facts", async () => { - const factWrites = []; - let now = "2026-06-20T11:01:05.000Z"; - const baseStore = createCloudRuntimeStore({ now: () => now }); - const runtimeStore = { - allocateWorkbenchProjectedSeq(params, requestMeta) { - return baseStore.allocateWorkbenchProjectedSeq(params, requestMeta); - }, - queryWorkbenchFacts(params) { - return baseStore.queryWorkbenchFacts(params); - }, - async writeWorkbenchFacts(params, requestMeta) { - const result = baseStore.writeWorkbenchFacts(params, requestMeta); - factWrites.push({ params, requestMeta }); - return result; - } - }; - - now = "2026-06-20T11:03:30.000Z"; - 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", - startedAt: "2026-06-20T11:00:30.000Z", - durationMs: 3000, - 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, 1); - assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting"); - assert.equal(factWrites[0].params.facts.checkpoints[0].startedAt, "2026-06-20T11:00:30.000Z"); - assert.equal(factWrites[0].params.facts.checkpoints[0].lastEventAt, "2026-06-20T11:01:00.000Z"); - assert.equal(factWrites[0].params.facts.checkpoints[0].durationMs, null); - assert.equal(factWrites[0].params.facts.checkpoints[0].timing.durationMs, null); - 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.traceEvents[0].projectedSeq, 2); - assert.equal(factWrites[1].params.facts.checkpoints[0].projectionStatus, "caught_up"); - assert.equal(factWrites[1].params.facts.checkpoints[0].durationMs, 90000); - assert.equal(factWrites[1].params.facts.checkpoints[0].finishedAt, "2026-06-20T11:02:00.000Z"); - assert.equal(factWrites[1].params.facts.checkpoints[0].lastEventAt, "2026-06-20T11:02:00.000Z"); -}); - -test("workbench projection writer reuses lifecycle assistant message id for terminal session projection", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:10:00.000Z" }); - const traceId = "trc_writer_terminal_same_message_id"; - const expectedAssistantMessageId = "msg_writer_terminal_same_message_id_agent"; - const accessController = { - async recordAgentSessionOwner(input) { - return { - id: input.sessionId, - projectId: input.projectId, - ownerUserId: input.ownerUserId, - ownerRole: input.ownerRole, - conversationId: input.conversationId, - threadId: input.threadId, - lastTraceId: input.traceId, - status: input.status, - session: input.session, - updatedAt: "2026-06-20T11:10:05.000Z" - }; - } - }; - - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - traceId, - sessionId: "ses_writer_terminal_same_message_id", - sourceSeq: 10, - type: "backend", - status: "running", - label: "agentrun:backend:running", - startedAt: "2026-06-20T11:09:55.000Z", - createdAt: "2026-06-20T11:09:55.000Z" - } - }); - await writeWorkbenchProjectionSession({ - accessController, - runtimeStore, - traceId, - ownerUserId: "usr_writer", - ownerRole: "user", - sessionId: "ses_writer_terminal_same_message_id", - projectId: "prj_writer", - conversationId: "cnv_writer_terminal_same_message_id", - threadId: "thread-writer-terminal-same-message-id", - status: "completed", - payload: { - traceId, - status: "completed", - assistantText: "same lifecycle final", - finalResponse: { text: "same lifecycle final", status: "completed", traceId }, - runnerTrace: { - traceId, - status: "completed", - startedAt: "2026-06-20T11:09:55.000Z", - lastEventAt: "2026-06-20T11:10:05.000Z", - finishedAt: "2026-06-20T11:10:05.000Z", - elapsedMs: 10000 - }, - agentRun: { runId: "run_writer_terminal_same_message_id", commandId: "cmd_writer_terminal_same_message_id", status: "completed", terminalStatus: "completed", lastSeq: 11 }, - updatedAt: "2026-06-20T11:10:05.000Z" - }, - session: { - sessionStatus: "completed", - lastTraceId: traceId, - messages: [ - { messageId: "msg_writer_terminal_same_message_id_user", role: "user", text: "same lifecycle", traceId, status: "sent" } - ], - finalResponse: { text: "same lifecycle final", status: "completed", traceId } - } - }); - - const loaded = runtimeStore.queryWorkbenchFacts({ traceId, families: ["messages", "parts", "turns"], limit: 20 }); - const assistantMessages = loaded.facts.messages.filter((message) => message.role !== "user"); - const terminalTurn = loaded.facts.turns.find((turn) => turn.terminal === true); - assert.deepEqual(assistantMessages.map((message) => message.messageId), [expectedAssistantMessageId]); - assert.equal(terminalTurn.messageId, expectedAssistantMessageId); - assert.equal(loaded.facts.parts.some((part) => part.messageId === expectedAssistantMessageId && part.partType === "final_response" && part.text === "same lifecycle final"), true); -}); - -test("workbench projection event writer allocates projectedSeq idempotently by source event identity", async () => { - const runtimeStore = createCloudRuntimeStore({ now: () => "2026-06-20T12:00:00.000Z" }); - - const repeatedSourceEvent = { - traceId: "trc_writer_idempotent", - sessionId: "ses_writer_idempotent", - seq: 1, - source: "agentrun", - sourceSeq: 70, - type: "backend", - status: "running", - label: "agentrun:backend:running", - runId: "run_writer_idempotent", - commandId: "cmd_writer_idempotent", - startedAt: "2026-06-20T12:00:01.000Z", - createdAt: "2026-06-20T12:00:01.000Z" - }; - await writeWorkbenchProjectionEvent({ runtimeStore, event: repeatedSourceEvent }); - await writeWorkbenchProjectionEvent({ - runtimeStore, - event: { - ...repeatedSourceEvent, - seq: 1, - sourceSeq: 71, - label: "agentrun:backend:runner-ready", - startedAt: null, - durationMs: 999, - createdAt: "2026-06-20T12:00:02.000Z" - } - }); - await writeWorkbenchProjectionEvent({ runtimeStore, event: repeatedSourceEvent }); - - const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_idempotent", families: ["traceEvents", "checkpoints"], limit: 10 }); - assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2]); - assert.deepEqual(loaded.facts.traceEvents.map((event) => event.sourceSeq), [70, 71]); - assert.equal(loaded.facts.checkpoints[0].projectedSeq, 2); - assert.equal(loaded.facts.checkpoints[0].startedAt, "2026-06-20T12:00:01.000Z"); - assert.equal(loaded.facts.checkpoints[0].lastEventAt, "2026-06-20T12:00:02.000Z"); - assert.equal(loaded.facts.checkpoints[0].durationMs, null); + assert.equal(first.facts.traceEvents[0].id, second.facts.traceEvents[0].id); + assert.equal(first.facts.messages[0].messageId, second.facts.messages[0].messageId); + assert.equal(first.facts.parts[0].partId, second.facts.parts[0].partId); }); diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index 5fed372c..4af8ee7b 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -1,70 +1,39 @@ /* - * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; draft-2026-06-20-p1-zero-split-durable-realtime; PJ2026-010205 HWLAB接入 draft-2026-06-17-r0. - * 职责: WorkbenchProjectionWriter 组件入口。唯一封装 Code Agent/AgentRun facts 到 Workbench session projection facts 的持久化写入。 + * SPEC: PJ2026-0104010803 Workbench唯一投影; HWLAB#2464 transactional Kafka projector. + * Responsibility: persist admission-only session ownership and build pure Kafka projection facts. + * Message/turn/checkpoint facts are committed only by commitAgentRunKafkaProjection. */ import { createHash } from "node:crypto"; -import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; -import { emitCodeAgentOtelSpan } from "./otel-trace.ts"; import { safeTraceId } from "./server-http-utils.ts"; -import { createWorkbenchTurnProjection, normalizeWorkbenchStatus, projectionDiagnostics, terminalFinalResponse, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; +import { normalizeWorkbenchStatus, terminalFinalResponse, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; -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 } = {}) { +export async function recordWorkbenchSessionOwner({ accessController, traceId, ownerUserId, ownerRole = null, sessionId = null, projectId = null, conversationId = null, threadId = null, status = "active", session = {}, preserveLastTraceId = false } = {}) { if (!ownerUserId || typeof accessController?.recordAgentSessionOwner !== "function") return null; - const safeId = safeTraceId(traceId); - try { - const ownerRecord = await accessController.recordAgentSessionOwner({ - ownerUserId, - ownerRole, - sessionId, - projectId, - conversationId, - threadId, - traceId: preserveLastTraceId ? undefined : safeId, - status, - session - }); - const factsWrite = 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, - payload, - params - }); - if (!factsWrite) { - throw workbenchProjectionError("workbench_projection_facts_not_written", "Workbench projection facts were not durably written."); - } - return ownerRecord; - } catch (error) { - emitWorkbenchProjectionOtel("workbench.projection.session.persist", safeId, "error", { - "workbench.projection.phase": "session-owner", - "workbench.projection.has_owner_user_id": Boolean(ownerUserId), - "workbench.projection.has_session_id": Boolean(sessionId) - }, error); - if (safeId) appendProjectionDiagnostic(traceStore, safeId, { - type: "session-owner", - status: "degraded", - label: "projection-writer:session-owner:persist-failed", - errorCode: "workbench_projection_session_persist_failed", - message: error?.message ?? "Workbench projection session persistence failed.", - valuesPrinted: false - }); - throw error; - } + return accessController.recordAgentSessionOwner({ + ownerUserId, + ownerRole, + sessionId, + projectId, + conversationId, + threadId, + traceId: preserveLastTraceId ? undefined : safeTraceId(traceId), + status, + session + }); } export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null, session = null, ownerUserId = null, ownerRole = null, projectId = null, conversationId = null, threadId = null, status = "idle", now = new Date().toISOString() } = {}) { - if (typeof runtimeStore?.writeWorkbenchFacts !== "function" || !session || typeof session !== "object") return null; + if (typeof runtimeStore?.writeWorkbenchSessionAdmissionFact !== "function") { + throw workbenchProjectionError("workbench_admission_store_unconfigured", "Workbench admission requires the durable session admission store."); + } + if (!session || typeof session !== "object") { + throw workbenchProjectionError("workbench_admission_session_required", "Workbench admission requires a session record."); + } const sessionId = textValue(session.id ?? session.sessionId); - if (!sessionId) return null; + if (!sessionId) { + throw workbenchProjectionError("workbench_admission_session_required", "Workbench admission requires sessionId."); + } const createdAt = timestampValue(session.startedAt ?? session.createdAt ?? session.session?.createdAt ?? now); const updatedAt = timestampValue(session.updatedAt ?? createdAt); const normalizedStatus = normalizeWorkbenchStatus(session.status ?? session.session?.sessionStatus ?? status) || "idle"; @@ -87,7 +56,7 @@ export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null, sessionJson: { ...sessionJson, sessionStatus: normalizedStatus, - source: textValue(sessionJson.source ?? session.session?.source) || "manual-session-create", + source: textValue(sessionJson.source ?? session.session?.source) || "session-admission", providerProfile: textValue(sessionJson.providerProfile ?? session.session?.providerProfile) || null, valuesRedacted: true, secretMaterialStored: false @@ -96,7 +65,7 @@ export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null, updatedAt, valuesRedacted: true }; - return runtimeStore.writeWorkbenchFacts({ facts: { sessions: [fact] } }, { + return runtimeStore.writeWorkbenchSessionAdmissionFact({ fact }, { sessionId, ownerUserId: fact.ownerUserId, projectId: fact.projectId, @@ -104,257 +73,6 @@ export async function writeWorkbenchSessionAdmissionFact({ runtimeStore = null, }); } -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); - if (!traceId) return null; - const projectedAt = new Date().toISOString(); - const sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq); - const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null; - const turnId = textValue(event.turnId) || traceId; - const eventType = textValue(event.eventType ?? event.type ?? event.label) || "event"; - const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status)); - const status = terminal ? normalizeWorkbenchStatus(event.status) : "running"; - 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(resolvedPreviousCheckpoint); - const suppressedAfterSeal = previousTerminal && !terminal; - if (suppressedAfterSeal) { - emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "ok", { - "workbench.projection.phase": "event-suppressed-after-seal", - "workbench.projection.terminal": false, - "workbench.projection.suppressed_after_seal": true, - "workbench.projection.session_id": sessionId, - "workbench.projection.turn_id": turnId, - "workbench.projection.source_seq": sourceSeq, - "workbench.projection.event_type": eventType - }); - return { written: false, suppressedAfterSeal: true, valuesPrinted: false }; - } - const explicitEventTiming = normalizeTimingProjection(event.timing) ?? normalizeTimingProjection(event); - const explicitDurationMs = durationValue(event.durationMs ?? explicitEventTiming?.durationMs); - const explicitStartedAt = optionalTimestampValue(event.startedAt ?? event.traceStartedAt ?? event.runnerStartedAt ?? explicitEventTiming?.startedAt); - // Once a trace has a durable checkpoint, startedAt is sealed as the card - // elapsed-time source. Later trace/result hydration can carry older or newer - // startedAt values, but it must not make the visible Code Agent card jump by - // switching to a different clock origin. - const provisionalLastEventAt = latestTimestamp(previousTiming?.lastEventAt, explicitEventTiming?.lastEventAt, occurredAt); - const provisionalFinishedAt = terminal ? latestTimestamp(explicitEventTiming?.finishedAt, optionalTimestampValue(event.finishedAt), provisionalLastEventAt, occurredAt) : null; - const startedAt = previousTiming?.startedAt ?? explicitStartedAt ?? startedAtFromDuration(provisionalFinishedAt, explicitDurationMs) ?? (sourceSeq <= 1 ? occurredAt : null); - const lastEventAt = provisionalLastEventAt; - const finishedAt = terminal ? latestTimestamp(provisionalFinishedAt, lastEventAt) : null; - const timing = eventTimingProjection({ startedAt, lastEventAt, finishedAt, terminal, durationMs: explicitDurationMs ?? previousTiming?.durationMs }); - const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal, traceId, source: "event", status, label: event.label, sourceSeq }); - if (timingAuthorityIssue) emitTerminalTimingAuthorityDiagnostic(defaultCodeAgentTraceStore, traceId, timingAuthorityIssue, { - "workbench.projection.phase": "event-timing", - "workbench.projection.session_id": sessionId, - "workbench.projection.turn_id": turnId, - "workbench.projection.source_seq": sourceSeq - }); - const eventTiming = suppressedAfterSeal ? previousTiming ?? timing : timing; - const sourceEventId = workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt }); - const allocation = await runtimeStore.allocateWorkbenchProjectedSeq({ - traceId, - sessionId, - turnId, - runId: textValue(event.runId) || null, - commandId: textValue(event.commandId) || null, - sourceSeq, - sourceEventId, - eventType, - occurredAt, - updatedAt: projectedAt - }, { - traceId, - sessionId, - agentSessionId: requestMeta.agentSessionId ?? sessionId, - valuesPrinted: false - }); - const projectedSeq = nonNegativeInteger(allocation?.projectedSeq); - if (projectedSeq <= 0) throw new Error("Workbench projection allocator returned an invalid projectedSeq."); - const eventId = isAssistantMessageTraceEvent(event) ? stableFactId("wte", { traceId, sourceEventId }) : textValue(event.id) || stableFactId("wte", { traceId, sourceEventId }); - const previousFinalText = finalResponseTextValue(resolvedPreviousCheckpoint?.finalResponse, resolvedPreviousCheckpoint?.assistantText, resolvedPreviousCheckpoint?.finalText); - const liveAssistantText = isAssistantMessageTraceEvent(event) ? finalResponseTextValue(event.assistantText, event.text, event.reply, event.message) : finalResponseTextValue(event.assistantText, event.reply); - const eventTerminalFinalResponse = terminal ? terminalFinalResponse(status, event, { evidence: event, finalResponse: event.finalResponse }) : null; - const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, liveAssistantText, previousFinalText, eventTerminalFinalResponse) : null; - const checkpointAssistantText = messageFinalText ?? liveAssistantText ?? previousFinalText ?? null; - const checkpointFinalResponse = terminal && messageFinalText - ? { ...(eventTerminalFinalResponse ?? {}), text: messageFinalText, status, traceId, source: "workbench-event-projection", valuesPrinted: false } - : resolvedPreviousCheckpoint?.finalResponse ?? null; - const terminalSealBlocked = terminal && status === "completed" && !messageFinalText; - const factTerminal = terminal && !terminalSealBlocked; - 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", - status, - traceId, - turnId, - text: messageFinalText ?? liveAssistantText ?? previousFinalText ?? "", - projectedSeq, - sourceSeq, - sourceEventId, - terminal: factTerminal, - sealed: factTerminal, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - createdAt: occurredAt, - updatedAt: projectedAt, - source: "workbench-event-projection" - }, userMessageFact ? 1 : 0, { traceId, sessionId, turnId, terminal, terminalStatus: status, finalText: messageFinalText, terminalSealBlocked, timestamp: projectedAt, timing }) : null; - const facts = { - sessions: sessionId && !suppressedAfterSeal ? [{ - sessionId, - status, - lastTraceId: traceId, - projectedSeq, - sourceSeq, - sourceEventId, - terminal, - sealed: terminal, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - updatedAt: projectedAt - }] : [], - traceEvents: [{ - ...event, - id: eventId, - traceId, - sessionId, - turnId, - messageId: textValue(event.messageId) || null, - sourceSeq, - sourceEventId, - projectedSeq, - eventType, - terminal, - sealed: terminal, - timing: eventTiming, - startedAt: eventTiming?.startedAt ?? null, - lastEventAt: eventTiming?.lastEventAt ?? null, - finishedAt: eventTiming?.finishedAt ?? null, - durationMs: eventTiming?.durationMs ?? null, - suppressedAfterSeal, - sourceOccurredAt: sourceOccurredAt !== occurredAt ? sourceOccurredAt : null, - createdAt: occurredAt, - occurredAt, - updatedAt: projectedAt - }], - messages: [userMessageFact, messageFact].filter(Boolean), - parts: [userMessageFact, messageFact].filter(Boolean).flatMap((message) => messagePartFacts(message, { finalText: message === messageFact ? messageFinalText : null })), - turns: sessionId && !suppressedAfterSeal ? [{ - turnId, - sessionId, - traceId, - messageId: messageFact?.messageId ?? null, - status, - projectedSeq, - sourceSeq, - sourceEventId, - terminal: factTerminal, - sealed: factTerminal, - finalResponse: checkpointFinalResponse, - assistantText: checkpointAssistantText, - failureKind: textValue(event.failureKind ?? event.errorCode) || null, - diagnostic: { timingAuthorityIssue, terminalSealBlocked, authority: "workbench-event-projection", valuesRedacted: true }, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - createdAt: occurredAt, - updatedAt: projectedAt - }] : [], - checkpoints: !suppressedAfterSeal ? [{ - traceId, - sessionId, - turnId, - runId: textValue(event.runId) || null, - commandId: textValue(event.commandId) || null, - projectedSeq, - sourceSeq, - sourceEventId, - projectionStatus: terminal ? "caught_up" : "projecting", - projectionHealth: terminalSealBlocked ? "degraded" : "healthy", - terminal: factTerminal, - sealed: factTerminal, - assistantText: checkpointAssistantText, - finalResponse: checkpointFinalResponse, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - diagnostic: { - lastEventLabel: textValue(event.label) || null, - lastEventStatus: textValue(event.status) || null, - timingAuthorityIssue, - terminalSealBlocked, - valuesRedacted: true - }, - updatedAt: projectedAt - }] : [] - }; - try { - const result = await runtimeStore.writeWorkbenchFacts({ facts }, { - traceId, - sessionId, - agentSessionId: requestMeta.agentSessionId ?? sessionId, - valuesPrinted: false - }); - emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "ok", { - ...workbenchProjectionFactCountAttributes(facts), - "workbench.projection.phase": terminal ? "event-terminal" : "event-running", - "workbench.projection.terminal": terminal, - "workbench.projection.suppressed_after_seal": suppressedAfterSeal, - "workbench.projection.session_id": sessionId, - "workbench.projection.turn_id": turnId, - "workbench.projection.projected_seq": projectedSeq, - "workbench.projection.source_seq": sourceSeq, - "workbench.projection.source_event_id": sourceEventId, - "workbench.projection.event_type": eventType, - "workbench.projection.event_id": eventId, - "workbench.projection.message_id": messageFact?.messageId ?? null - }); - return result; - } catch (error) { - emitWorkbenchProjectionOtel("workbench.projection.event.write", traceId, "error", { - ...workbenchProjectionFactCountAttributes(facts), - "workbench.projection.phase": terminal ? "event-terminal" : "event-running", - "workbench.projection.terminal": terminal, - "workbench.projection.suppressed_after_seal": suppressedAfterSeal, - "workbench.projection.session_id": sessionId, - "workbench.projection.turn_id": turnId, - "workbench.projection.projected_seq": projectedSeq, - "workbench.projection.source_seq": sourceSeq, - "workbench.projection.source_event_id": sourceEventId, - "workbench.projection.event_type": eventType, - "workbench.projection.event_id": eventId, - "workbench.projection.message_id": messageFact?.messageId ?? null - }, error); - appendProjectionDiagnostic(defaultCodeAgentTraceStore, traceId, { - type: "facts", - status: "degraded", - label: "projection-writer:event-facts:persist-failed", - errorCode: error?.code ?? "workbench_event_facts_persist_failed", - message: error?.message ?? "Workbench event facts persistence failed.", - terminal: false, - valuesPrinted: false - }); - throw error; - } -} - export function buildWorkbenchProjectionEventFacts({ event = {}, requestMeta = {}, previousCheckpoint = null, projectedSeq, projectedAt = new Date().toISOString() } = {}) { const traceId = safeTraceId(event.traceId ?? requestMeta.traceId); if (!traceId) throw workbenchProjectionError("workbench_projection_trace_required", "Kafka projection event requires traceId."); @@ -370,7 +88,16 @@ export function buildWorkbenchProjectionEventFacts({ event = {}, requestMeta = { const sourceOccurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt ?? projectedAt); const occurredAt = latestTimestamp(previousTiming?.lastEventAt, sourceOccurredAt) ?? sourceOccurredAt; if (checkpointIsTerminal(previousCheckpoint) && !terminal) { - return { written: false, suppressedAfterSeal: true, traceId, sessionId, sourceSeq, valuesPrinted: false }; + return { + written: false, + suppressedAfterSeal: true, + traceId, + sessionId, + sourceSeq, + sourceEventId: workbenchSourceEventId(event, { traceId, sourceSeq, occurredAt }), + projectedSeq: nonNegativeInteger(previousCheckpoint?.projectedSeq) || durableProjectedSeq, + valuesPrinted: false + }; } const explicitEventTiming = normalizeTimingProjection(event.timing) ?? normalizeTimingProjection(event); const explicitDurationMs = durationValue(event.durationMs ?? explicitEventTiming?.durationMs); @@ -491,293 +218,6 @@ function eventProjectionMessageId(traceId, role, event = {}) { return `msg_${traceSuffix}_${role === "user" ? "user" : "agent"}`; } -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, { 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; - if (timingAuthorityIssue) emitTerminalTimingAuthorityDiagnostic(traceStore, safeId, timingAuthorityIssue, { - "workbench.projection.phase": "facts-timing", - "workbench.projection.session_id": facts.sessions[0]?.sessionId ?? sessionId ?? null, - "workbench.projection.turn_id": facts.turns[0]?.turnId ?? null - }); - try { - const result = await runtimeStore.writeWorkbenchFacts({ facts }, { - traceId: facts.checkpoints[0]?.traceId ?? traceId, - sessionId: facts.sessions[0]?.sessionId ?? sessionId, - ownerUserId, - projectId, - valuesPrinted: false - }); - emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "ok", { - ...workbenchProjectionFactCountAttributes(facts), - "workbench.projection.terminal_trace_backfill_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, - "workbench.projection.projected_seq": facts.checkpoints[0]?.projectedSeq ?? facts.turns[0]?.projectedSeq ?? facts.messages[0]?.projectedSeq ?? null, - "workbench.projection.source_seq": facts.checkpoints[0]?.sourceSeq ?? facts.turns[0]?.sourceSeq ?? facts.messages[0]?.sourceSeq ?? null, - "workbench.projection.source_event_id": facts.checkpoints[0]?.sourceEventId ?? facts.turns[0]?.sourceEventId ?? facts.messages[0]?.sourceEventId ?? null, - "workbench.projection.message_id": facts.messages[0]?.messageId ?? null - }); - return result; - } catch (error) { - const safeId = safeTraceId(traceId); - emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", { - ...workbenchProjectionFactCountAttributes(facts), - "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, - "workbench.projection.projected_seq": facts.checkpoints[0]?.projectedSeq ?? facts.turns[0]?.projectedSeq ?? facts.messages[0]?.projectedSeq ?? null, - "workbench.projection.source_seq": facts.checkpoints[0]?.sourceSeq ?? facts.turns[0]?.sourceSeq ?? facts.messages[0]?.sourceSeq ?? null, - "workbench.projection.source_event_id": facts.checkpoints[0]?.sourceEventId ?? facts.turns[0]?.sourceEventId ?? facts.messages[0]?.sourceEventId ?? null, - "workbench.projection.message_id": facts.messages[0]?.messageId ?? null - }, error); - 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 - }); - throw error; - } -} - -async function backfillTerminalTraceEvents({ runtimeStore = null, traceStore = defaultCodeAgentTraceStore, traceId = null, facts = {}, payload = null, status = 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; - 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 existingEvents = await existingWorkbenchTraceEvents(runtimeStore, safeId, { traceStore }); - if (existingEvents === null) return { written: 0, skipped: true, errorCode: "workbench_terminal_trace_backfill_probe_failed", valuesPrinted: false }; - const writeAssistantFinal = !hasTerminalAssistantFinalTraceEvent(existingEvents, finalText); - const writeCompletion = !hasTerminalResultTraceEvent(existingEvents); - if (!writeAssistantFinal && !writeCompletion) return { written: 0, idempotent: true, skippedExisting: true, 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)); - let offset = 1; - let written = 0; - if (writeAssistantFinal) { - 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; - } - if (writeCompletion) { - 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 committed to the Workbench projection event stream.", - 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", - "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 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; - 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 projection = createWorkbenchTurnProjection({ traceId: safeId, result: payload, session: { id: resolvedSessionId, status: normalizedStatus, session }, trace: payload?.runnerTrace ?? null }); - const projectedStatus = normalizeWorkbenchStatus(projection.status ?? normalizedStatus); - const terminal = projection.terminal === true; - const terminalSealBlocked = projection.terminalSealBlocked === true; - const terminalStatus = terminal ? (TERMINAL_STATUSES.has(projectedStatus) ? projectedStatus : normalizedStatus) : projectedStatus; - const timing = terminalTimingAtLeastProjectedAt(projectionTimingForStatus(projection.timing, terminal), terminal); - const timingAuthorityIssue = terminalTimingAuthorityIssue(timing, { terminal, traceId: safeId, source: "facts", status: terminalStatus, label: payload?.lastEventLabel, sourceSeq: projection.lastProjectedSeq }); - const baseDiagnostic = projectionDiagnostics({ traceId: safeId, result: payload, trace: payload?.runnerTrace ?? null, projection }); - const diagnostic = timingAuthorityIssue - ? { ...baseDiagnostic, projectionHealth: baseDiagnostic.projectionHealth === "unavailable" ? "unavailable" : "degraded", blocker: baseDiagnostic.blocker ?? timingAuthorityIssue, timingAuthorityIssue, valuesRedacted: true } - : baseDiagnostic; - const factSeq = workbenchProjectionFactSequence({ projection, previousCheckpoint, agentRun }); - const previousFinalText = finalResponseTextValue(previousCheckpoint?.finalResponse, previousCheckpoint?.assistantText, previousCheckpoint?.finalText); - const finalText = terminal ? finalResponseTextValue(projection.finalResponse, payload?.finalResponse, session?.finalResponse, payload?.assistantText, previousFinalText) : null; - const projectedFinalResponse = terminal && finalText ? { text: finalText, status: terminalStatus, traceId: safeId, valuesPrinted: false } : null; - const messages = normalizeMessages(workbenchProjectionInputMessages({ session, payload, params, traceId: safeId, timestamp, terminal, terminalStatus, finalText, previousCheckpoint }), { - traceId: safeId, - sessionId: resolvedSessionId, - conversationId: resolvedConversationId, - threadId: resolvedThreadId, - terminal, - terminalStatus, - finalText, - terminalSealBlocked, - timestamp, - timing - }); - const sessionJson = workbenchSessionJsonWithLaunchContext(session, { payload, params }); - return { - sessions: resolvedSessionId ? [{ - sessionId: resolvedSessionId, - ownerUserId, - ownerRole, - projectId: projectId ?? session?.projectId ?? payload?.projectId ?? params?.projectId ?? null, - conversationId: resolvedConversationId, - threadId: resolvedThreadId, - status: terminal ? terminalStatus : "running", - lastTraceId: safeId, - projectedSeq: factSeq.projectedSeq, - sourceSeq: factSeq.sourceSeq, - sourceEventId: safeId, - terminal, - sealed: terminal, - sessionJson, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - createdAt: timestampValue(session?.createdAt ?? payload?.createdAt ?? timestamp), - updatedAt: timestamp - }] : [], - messages, - parts: messages.flatMap((message) => messagePartFacts(message, { finalText: message.role !== "user" ? finalText : null })), - 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: factSeq.projectedSeq, - sourceSeq: factSeq.sourceSeq, - sourceEventId: safeId, - terminal: projection.terminal, - sealed: projection.terminal, - finalResponse: projectedFinalResponse, - diagnostic, - assistantText: finalText, - userMessage: previousCheckpoint?.userMessage ?? null, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - 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: factSeq.projectedSeq, - sourceSeq: factSeq.sourceSeq, - 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, - finalResponse: projectedFinalResponse, - assistantText: finalText, - userMessage: messages.find((message) => message.role === "user") ?? null, - timing, - startedAt: timing.startedAt, - lastEventAt: timing.lastEventAt, - finishedAt: timing.finishedAt, - durationMs: timing.durationMs, - updatedAt: timestamp - }] : [] - }; -} - function workbenchSessionJsonWithLaunchContext(session = {}, { payload = null, params = null } = {}) { const base = plainObjectValue(session) ?? {}; const launchContext = firstPlainObjectValue( @@ -792,14 +232,7 @@ function workbenchSessionJsonWithLaunchContext(session = {}, { payload = null, p params?.session?.launchContext ); if (!launchContext) return base; - const next = { - ...base, - launchContext: { - ...launchContext, - valuesRedacted: true - }, - valuesRedacted: true - }; + const next = { ...base, launchContext: { ...launchContext, valuesRedacted: true }, valuesRedacted: true }; if (!Object.prototype.hasOwnProperty.call(next, "secretMaterialStored")) next.secretMaterialStored = false; return next; } @@ -816,16 +249,6 @@ function plainObjectValue(value) { return value && typeof value === "object" && !Array.isArray(value) ? value : null; } -function workbenchProjectionFactSequence({ projection = {}, previousCheckpoint = null, agentRun = null } = {}) { - const projectionSeq = nonNegativeInteger(projection?.lastProjectedSeq); - const previousSeq = nonNegativeInteger(previousCheckpoint?.projectedSeq ?? previousCheckpoint?.lastProjectedSeq); - const sourceSeq = nonNegativeInteger(agentRun?.lastSeq ?? projection?.sourceSeq ?? projectionSeq); - return { - projectedSeq: previousSeq > 0 ? Math.min(projectionSeq || previousSeq, previousSeq) : projectionSeq, - sourceSeq - }; -} - function workbenchProjectionError(code, message) { const error = new Error(message); error.code = code; @@ -833,121 +256,6 @@ function workbenchProjectionError(code, message) { return error; } -function emitWorkbenchProjectionOtel(name, traceId, status = "ok", attributes = {}, error = null) { - void emitCodeAgentOtelSpan(name, safeTraceId(traceId) ?? "trc_unassigned", process.env, { - status, - error, - attributes: { - ...attributes, - "workbench.projection.zero_implicit_fallback": true, - "workbench.projection.error_visible": status === "error" - } - }); -} - -function workbenchProjectionFactCountAttributes(facts = {}) { - return { - "workbench.facts.sessions": Array.isArray(facts.sessions) ? facts.sessions.length : 0, - "workbench.facts.messages": Array.isArray(facts.messages) ? facts.messages.length : 0, - "workbench.facts.parts": Array.isArray(facts.parts) ? facts.parts.length : 0, - "workbench.facts.turns": Array.isArray(facts.turns) ? facts.turns.length : 0, - "workbench.facts.trace_events": Array.isArray(facts.traceEvents) ? facts.traceEvents.length : 0, - "workbench.facts.checkpoints": Array.isArray(facts.checkpoints) ? facts.checkpoints.length : 0 - }; -} - -function workbenchProjectionInputMessages({ session = {}, payload = {}, params = {}, traceId = null, timestamp = null, terminal = false, terminalStatus = "completed", finalText = null, previousCheckpoint = null } = {}) { - if (Array.isArray(session?.messages) && session.messages.length > 0) return preserveTerminalUserMessage(session.messages, { terminal, traceId, previousCheckpoint }); - if (Array.isArray(payload?.messages) && payload.messages.length > 0) return preserveTerminalUserMessage(payload.messages, { terminal, traceId, previousCheckpoint }); - if (!traceId) return []; - const turnId = textValue(payload?.turnId) || traceId; - const assistantMessageId = eventProjectionAssistantMessageId(traceId, payload); - const createdAt = timestampValue(timestamp ?? payload?.createdAt ?? payload?.updatedAt); - const messages = []; - const userMessage = workbenchProjectionUserMessage({ payload, params, traceId, turnId, timestamp: createdAt, terminal }); - if (userMessage) messages.push(userMessage); - messages.push({ - messageId: assistantMessageId, - role: "agent", - status: terminal ? terminalStatus : "running", - text: terminal && finalText ? finalText : "", - traceId, - turnId, - timing: payload?.timing, - startedAt: payload?.startedAt, - lastEventAt: payload?.lastEventAt, - finishedAt: payload?.finishedAt, - durationMs: payload?.durationMs, - createdAt, - updatedAt: createdAt, - valuesPrinted: false - }); - return messages; -} - -function workbenchProjectionUserMessage({ payload = {}, params = {}, traceId = null, turnId = null, timestamp = null, terminal = false } = {}) { - const userText = textValue(params?.message ?? params?.prompt ?? payload?.prompt ?? payload?.userText ?? payload?.userMessage?.text ?? payload?.userMessage?.content ?? (!terminal ? payload?.message : null)); - if (!traceId || !userText) return null; - const createdAt = timestampValue(timestamp ?? payload?.createdAt ?? payload?.updatedAt); - return { - messageId: eventProjectionUserMessageId(traceId, payload), - role: "user", - status: "sent", - text: userText, - traceId, - turnId: textValue(turnId) || traceId, - createdAt, - updatedAt: createdAt, - valuesPrinted: false - }; -} - -function preserveTerminalUserMessage(messages = [], { terminal = false, traceId = null, previousCheckpoint = null } = {}) { - if (!terminal || !traceId) return messages; - const previousUserMessage = previousCheckpoint?.userMessage ?? null; - const previousText = textValue(previousUserMessage?.text ?? previousUserMessage?.content ?? previousUserMessage?.message); - let handledUser = false; - const normalizedTraceId = safeTraceId(traceId) ?? traceId; - const preserved = messages.map((message) => { - const role = textValue(message?.role) || "agent"; - const messageTraceId = safeTraceId(message?.traceId ?? message?.turnId ?? normalizedTraceId) ?? normalizedTraceId; - if (role !== "user" || messageTraceId !== normalizedTraceId) return message; - handledUser = true; - const messageText = textValue(message?.text ?? message?.content ?? message?.message); - if (!previousText && isSyntheticTerminalUserNotice(messageText)) return null; - if (!previousText) return message; - return { - ...message, - ...previousUserMessage, - messageId: textValue(previousUserMessage.messageId ?? previousUserMessage.id ?? message?.messageId ?? message?.id) || eventProjectionUserMessageId(normalizedTraceId, message), - role: "user", - status: "sent", - text: previousText, - traceId: normalizedTraceId, - turnId: textValue(previousUserMessage.turnId ?? message?.turnId) || normalizedTraceId, - terminal: false, - sealed: false - }; - }).filter(Boolean); - if (handledUser) return preserved; - if (!previousText) return preserved; - return [previousUserMessage, ...messages]; -} - -function isSyntheticTerminalUserNotice(text) { - const value = textValue(text); - return /当前\s*AgentRun\s*请求已取消|请求已取消|hwlab-user-cancel/u.test(value); -} - -function normalizeMessages(messages, context) { - const normalized = Array.isArray(messages) ? messages.map((message, index) => normalizeMessageFact(message, index, context)).filter(Boolean) : []; - if (context?.terminal && context?.finalText && context?.traceId && !normalized.some((message) => message.role !== "user" && message.traceId === context.traceId)) { - const projected = normalizeMessageFact({ messageId: eventProjectionAssistantMessageId(context.traceId), role: "agent", status: context.terminalStatus ?? "completed", text: context.finalText, traceId: context.traceId, source: "workbench-terminal-projection" }, normalized.length, context); - if (projected) normalized.push(projected); - } - return normalized; -} - function messagePartFacts(message = {}, { finalText = null } = {}) { const messageId = textValue(message.messageId); if (!messageId || !message.sessionId) return []; @@ -978,50 +286,6 @@ function messagePartFacts(message = {}, { finalText = null } = {}) { }]; } -async function existingWorkbenchTraceEvents(runtimeStore, traceId, { traceStore = defaultCodeAgentTraceStore } = {}) { - if (typeof runtimeStore?.queryWorkbenchFacts !== "function") return []; - const safeId = safeTraceId(traceId); - try { - const result = await runtimeStore.queryWorkbenchFacts({ traceId: safeId ?? traceId, families: ["traceEvents"], limit: 500 }); - return Array.isArray(result?.facts?.traceEvents) ? result.facts.traceEvents : []; - } catch (error) { - emitWorkbenchProjectionOtel("workbench.projection.terminal_trace_backfill.probe", safeId, "error", { - "workbench.projection.phase": "terminal-trace-backfill-probe" - }, error); - if (safeId) appendProjectionDiagnostic(traceStore, safeId, { - type: "facts", - status: "degraded", - label: "projection-writer:terminal-trace-backfill-probe-failed", - errorCode: error?.code ?? "workbench_terminal_trace_backfill_probe_failed", - message: error?.message ?? "Workbench terminal trace backfill probe failed.", - terminal: false, - valuesPrinted: false - }); - return null; - } -} - -function hasTerminalAssistantFinalTraceEvent(events = [], finalText = null) { - const expectedText = finalResponseTextValue(finalText); - return events.some((event) => { - const type = textValue(event?.eventType ?? event?.type).toLowerCase(); - const label = textValue(event?.label).toLowerCase(); - if (type !== "assistant" && !/assistant:message|assistant_message/u.test(label)) return false; - if (event?.terminal !== true && normalizeWorkbenchStatus(event?.status) !== "completed") return false; - const text = finalResponseTextValue(event?.finalResponse, event?.text, event?.message, event?.content); - return !expectedText || text === expectedText; - }); -} - -function hasTerminalResultTraceEvent(events = []) { - return events.some((event) => { - const type = textValue(event?.eventType ?? event?.type).toLowerCase(); - const label = textValue(event?.label).toLowerCase(); - const status = normalizeWorkbenchStatus(event?.status); - return (type === "terminal" || type === "result" || /result:completed|result_completed/u.test(label)) && event?.terminal === true && status === "completed"; - }); -} - function normalizeMessageFact(message = {}, index, { traceId, sessionId, conversationId, threadId, terminal, terminalStatus = "completed", finalText = null, terminalSealBlocked = false, timestamp, timing: contextTiming }) { const role = textValue(message.role) || "agent"; const messageId = textValue(message.messageId ?? message.id) || (role === "user" ? eventProjectionUserMessageId(traceId, message) : eventProjectionAssistantMessageId(traceId, message)); @@ -1077,48 +341,6 @@ function finalResponseTextValue(...values) { return null; } -async function latestWorkbenchCheckpoint(runtimeStore, traceId, { traceStore = defaultCodeAgentTraceStore, phase = "checkpoint-read" } = {}) { - if (typeof runtimeStore?.queryWorkbenchFacts !== "function") return 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) { - const timing = normalizeTimingProjection(value) ?? emptyTimingProjection(); - if (terminal) return timing; - return { ...timing, finishedAt: null, durationMs: null, valuesRedacted: timing.valuesRedacted !== false }; -} - -function terminalTimingAtLeastProjectedAt(value, terminal) { - const timing = normalizeTimingProjection(value) ?? emptyTimingProjection(); - if (!terminal) return timing; - const finishedAt = timing.finishedAt ?? null; - const durationMs = durationValue(timing.durationMs) ?? elapsedBetween(timing.startedAt, finishedAt); - return { - ...timing, - finishedAt, - durationMs, - valuesRedacted: timing.valuesRedacted !== false - }; -} - function checkpointIsTerminal(value) { const source = value && typeof value === "object" ? value : null; if (!source) return false; @@ -1200,27 +422,6 @@ function terminalTimingAuthorityIssue(timing, { terminal = false, traceId = null }; } -function emitTerminalTimingAuthorityDiagnostic(traceStore, traceId, issue, attributes = {}) { - if (!issue) return null; - emitWorkbenchProjectionOtel("workbench.projection.terminal_timing_authority", traceId, "error", { - ...attributes, - "workbench.projection.timing_authority_missing": true, - "workbench.projection.missing_started_at": issue.missingStartedAt === true, - "workbench.projection.missing_finished_at": issue.missingFinishedAt === true, - "workbench.projection.missing_duration_ms": issue.missingDurationMs === true - }, workbenchProjectionError(issue.code, issue.message)); - return appendProjectionDiagnostic(traceStore, traceId, { - type: "projection-timing", - status: "degraded", - label: "projection-writer:terminal-timing:missing-authority", - errorCode: issue.code, - message: issue.message, - terminal: false, - timingAuthorityIssue: issue, - valuesPrinted: false - }); -} - function latestTimestamp(...values) { let latest = null; let latestMs = Number.NEGATIVE_INFINITY; @@ -1234,10 +435,6 @@ function latestTimestamp(...values) { return latest; } -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)}`; } @@ -1296,15 +493,3 @@ function optionalTimestampValue(value) { 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; - return traceStore.append(safeId, { - type: "projection-diagnostic", - status: "degraded", - label: "projection:diagnostic", - ...event, - valuesPrinted: false - }); -} diff --git a/internal/db/runtime-store-memory.ts b/internal/db/runtime-store-memory.ts index 761626bf..40932361 100644 --- a/internal/db/runtime-store-memory.ts +++ b/internal/db/runtime-store-memory.ts @@ -681,6 +681,26 @@ export class CloudRuntimeStore { }; } + writeWorkbenchSessionAdmissionFact(params = {}, requestMeta = {}) { + const facts = normalizeWorkbenchFacts({ sessions: [params.fact ?? params.session ?? params] }, requestMeta, this.now()); + const fact = facts.sessions[0]; + if (!fact?.sessionId) throw new Error("Workbench session admission fact requires sessionId."); + const previous = this.workbenchSessions.get(fact.sessionId) ?? null; + const projected = previous && (nonNegativeInteger(previous.projectedSeq) > 0 || previous.sealed === true); + const ownership = { + ownerUserId: previous?.ownerUserId ?? fact.ownerUserId ?? null, + ownerRole: previous?.ownerRole ?? fact.ownerRole ?? null, + projectId: previous?.projectId ?? fact.projectId ?? null, + conversationId: previous?.conversationId ?? fact.conversationId ?? null, + threadId: previous?.threadId ?? fact.threadId ?? null + }; + const stored = projected + ? { ...previous, ...ownership } + : { ...previous, ...fact, ...ownership, createdAt: previous?.createdAt ?? fact.createdAt }; + this.workbenchSessions.set(fact.sessionId, stored); + return { written: true, admissionOnly: true, fact: stored, persistence: this.summary(), valuesPrinted: false }; + } + queryWorkbenchFacts(params = {}) { const families = workbenchFactFamilySet(params); const sessions = [...this.workbenchSessions.values()].filter((record) => matchesWorkbenchSessionFactQuery(record, params)); diff --git a/internal/db/runtime-store-postgres-kafka.integration.test.ts b/internal/db/runtime-store-postgres-kafka.integration.test.ts new file mode 100644 index 00000000..a3bc7748 --- /dev/null +++ b/internal/db/runtime-store-postgres-kafka.integration.test.ts @@ -0,0 +1,307 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { test } from "bun:test"; +import pg from "pg"; + +import { buildWorkbenchProjectionEventFacts } from "../cloud/workbench-projection-writer.ts"; +import { PostgresCloudRuntimeStore } from "./runtime-store-postgres.ts"; +import { commitAgentRunKafkaProjection as commitProjection } from "./runtime-store-postgres-kafka.ts"; + +const dbUrl = String(process.env.HWLAB_KAFKA_PROJECTOR_INTEGRATION_DB_URL ?? "").trim(); +const enabled = Boolean(dbUrl) && process.env.HWLAB_KAFKA_PROJECTOR_INTEGRATION_CONFIRM_NON_PRODUCTION === "1"; + +if (!enabled) { + test.skip("Kafka projector real PostgreSQL transaction and fault-injection contract", () => {}); +} else { + test("Kafka projector real PostgreSQL transaction and fault-injection contract", async () => { + const { Pool } = pg; + const pool = new Pool({ connectionString: dbUrl, ssl: false, max: 4 }); + const suffix = randomUUID().replaceAll("-", ""); + const traceId = `trc_it_${suffix}`; + const sessionId = `ses_it_${suffix}`; + const ownerUserId = `usr_it_${suffix}`; + const now = "2026-07-10T12:00:00.000Z"; + const store = new PostgresCloudRuntimeStore({ + dbUrl, + sslMode: "disable", + env: { ...process.env, HWLAB_CLOUD_DB_SSL_MODE: "disable" }, + queryClient: pool, + now: () => now, + logger: { error() {}, warn() {}, info() {} } + }); + + try { + const readiness = await store.readiness(); + assert.equal(readiness.ready, true); + assert.equal(readiness.durable, true); + assert.equal(readiness.migration?.ready, true); + + await pool.query( + "INSERT INTO users (id, username, display_name, role, status, created_at, updated_at) VALUES ($1,$2,$2,'user','active',$3,$3)", + [ownerUserId, `integration-${suffix}`, now] + ); + + await store.writeWorkbenchSessionAdmissionFact({ + fact: { + sessionId, + ownerUserId, + projectId: `prj_it_${suffix}`, + conversationId: `cnv_it_${suffix}`, + threadId: `thread_it_${suffix}`, + status: "queued", + lastTraceId: traceId, + projectedSeq: 0, + sourceSeq: 0, + terminal: false, + sealed: false, + sessionJson: { sessionId, launchContext: { source: "integration-test" }, valuesRedacted: true }, + createdAt: now, + updatedAt: now + } + }); + + const running = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_running_${suffix}`, + sourceSeq: 1, + sourceOffset: "100", + inputSha256: "1".repeat(64), + event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now } + }); + assert.equal(running.duplicate, false); + assert.equal(running.projectedSeq, 1); + assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [traceId]), 1); + assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [traceId]), 1); + assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 2); + assert.equal(await count(pool, "hwlab_kafka_outbox", "partition_key = $1", [traceId]), 1); + + const sameTransport = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_running_${suffix}`, + sourceSeq: 1, + sourceOffset: "100", + inputSha256: "1".repeat(64), + event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now } + }); + const newOffsetSameEvent = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_running_${suffix}`, + sourceSeq: 1, + sourceOffset: "101", + inputSha256: "1".repeat(64), + event: { eventType: "assistant_message", status: "running", assistantText: "partial", createdAt: now } + }); + assert.equal(sameTransport.duplicate, true); + assert.equal(newOffsetSameEvent.duplicate, true); + assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [traceId]), 1); + assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 2); + + const conflict = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_running_${suffix}`, + sourceSeq: 1, + sourceOffset: "102", + inputSha256: "2".repeat(64), + event: { eventType: "assistant_message", status: "running", assistantText: "mutated", createdAt: now } + }); + assert.equal(conflict.conflict, true); + assert.equal(await count(pool, "workbench_kafka_dlq", "source_event_id = $1", [`evt_running_${suffix}`]), 1); + const projectedInbox = await pool.query( + "SELECT status, projected_seq FROM workbench_kafka_inbox WHERE source_topic = $1 AND source_event_id = $2", + ["agentrun.event.v1", `evt_running_${suffix}`] + ); + assert.deepEqual(projectedInbox.rows[0], { status: "projected", projected_seq: 1 }); + + await pool.query(` + CREATE OR REPLACE FUNCTION issue_2464_delay_inbox_insert() RETURNS trigger AS $$ + BEGIN + IF NEW.source_topic = 'agentrun.event.v1' AND NEW.source_partition = 0 AND NEW.source_offset = 300 THEN + PERFORM pg_sleep(0.25); + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql; + DROP TRIGGER IF EXISTS issue_2464_delay_inbox_insert ON workbench_kafka_inbox; + CREATE TRIGGER issue_2464_delay_inbox_insert BEFORE INSERT ON workbench_kafka_inbox + FOR EACH ROW EXECUTE FUNCTION issue_2464_delay_inbox_insert(); + `); + try { + const concurrent = await Promise.allSettled([ + project(store, { + traceId: `trc_transport_a_${suffix}`, + sessionId: `ses_transport_a_${suffix}`, + sourceEventId: `evt_transport_a_${suffix}`, + sourceSeq: 1, + sourceOffset: "300", + inputSha256: "6".repeat(64), + event: { eventType: "backend_status", status: "running", createdAt: now } + }), + project(store, { + traceId: `trc_transport_b_${suffix}`, + sessionId: `ses_transport_b_${suffix}`, + sourceEventId: `evt_transport_b_${suffix}`, + sourceSeq: 1, + sourceOffset: "300", + inputSha256: "7".repeat(64), + event: { eventType: "backend_status", status: "running", createdAt: now } + }) + ]); + assert.deepEqual(concurrent.map((result) => result.status), ["fulfilled", "fulfilled"]); + const values = concurrent.map((result) => { + if (result.status !== "fulfilled") throw result.reason; + return result.value; + }); + assert.equal(values.filter((result) => result.conflict === true).length, 1); + assert.equal(values.filter((result) => result.conflict !== true && result.duplicate !== true).length, 1); + assert.equal(await count(pool, "workbench_kafka_inbox", "source_topic = $1 AND source_partition = 0 AND source_offset = 300", ["agentrun.event.v1"]), 1); + assert.equal(await count(pool, "workbench_kafka_dlq", "source_topic = $1 AND source_partition = 0 AND source_offset = 300", ["agentrun.event.v1"]), 1); + } finally { + await pool.query("DROP TRIGGER IF EXISTS issue_2464_delay_inbox_insert ON workbench_kafka_inbox; DROP FUNCTION IF EXISTS issue_2464_delay_inbox_insert()"); + } + + const faultTraceId = `trc_fault_${suffix}`; + const faultSessionId = `ses_fault_${suffix}`; + const faultStore = new PostgresCloudRuntimeStore({ + dbUrl, + sslMode: "disable", + env: { ...process.env, HWLAB_CLOUD_DB_SSL_MODE: "disable" }, + queryClient: faultInjectingPool(pool, /^INSERT INTO workbench_projection_outbox/u), + now: () => now, + logger: { error() {}, warn() {}, info() {} } + }); + await assert.rejects(project(faultStore, { + traceId: faultTraceId, + sessionId: faultSessionId, + sourceEventId: `evt_fault_${suffix}`, + sourceSeq: 1, + sourceOffset: "200", + inputSha256: "3".repeat(64), + event: { eventType: "assistant_message", status: "running", assistantText: "must roll back", createdAt: now } + }), /injected projection outbox failure/u); + assert.equal(await count(pool, "workbench_kafka_inbox", "trace_id = $1", [faultTraceId]), 0); + assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [faultTraceId]), 0); + assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [faultTraceId]), 0); + assert.equal(await count(pool, "hwlab_kafka_outbox", "partition_key = $1", [faultTraceId]), 0); + + const terminal = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_terminal_${suffix}`, + sourceSeq: 2, + sourceOffset: "103", + inputSha256: "4".repeat(64), + event: { + eventType: "terminal_status", + status: "completed", + terminal: true, + assistantText: "authoritative final", + finalResponse: { text: "authoritative final", status: "completed" }, + startedAt: now, + finishedAt: "2026-07-10T12:00:01.000Z", + durationMs: 1000, + createdAt: "2026-07-10T12:00:01.000Z" + } + }); + assert.equal(terminal.projectedSeq, 2); + const sealed = await pool.query( + "SELECT projected_seq, terminal, sealed, checkpoint_json FROM workbench_projection_checkpoints WHERE trace_id = $1", + [traceId] + ); + assert.equal(sealed.rows[0].projected_seq, 2); + assert.equal(sealed.rows[0].terminal, true); + assert.equal(sealed.rows[0].sealed, true); + assert.equal(JSON.parse(sealed.rows[0].checkpoint_json).finalResponse.text, "authoritative final"); + assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 5); + + const late = await project(store, { + traceId, + sessionId, + sourceEventId: `evt_late_${suffix}`, + sourceSeq: 3, + sourceOffset: "104", + inputSha256: "5".repeat(64), + event: { eventType: "backend_status", status: "running", terminal: false, createdAt: "2026-07-10T12:00:02.000Z" } + }); + assert.equal(late.suppressedAfterSeal, true); + assert.equal(late.projectedSeq, 2); + assert.equal(await count(pool, "workbench_trace_events", "trace_id = $1", [traceId]), 2); + assert.equal(await count(pool, "workbench_projection_outbox", "trace_id = $1", [traceId]), 5); + const lateInbox = await pool.query( + "SELECT status, projected_seq FROM workbench_kafka_inbox WHERE source_topic = $1 AND source_event_id = $2", + ["agentrun.event.v1", `evt_late_${suffix}`] + ); + assert.deepEqual(lateInbox.rows[0], { status: "projected", projected_seq: 2 }); + + const firstSync = await store.readAtomicWorkbenchProjectionSync({ traceId, afterOutboxSeq: 0, limit: 100 }); + assert.equal(firstSync.events.length, 5); + assert.equal(firstSync.hasMore, false); + assert.equal(firstSync.cursorOutboxSeq, firstSync.cutoffOutboxSeq); + assert.equal(new Set(firstSync.events.map((event) => event.outboxSeq)).size, firstSync.events.length); + assert.equal(firstSync.facts.checkpoints[0].projectedSeq, 2); + assert.equal(firstSync.facts.turns[0].finalResponse.text, "authoritative final"); + const caughtUp = await store.readAtomicWorkbenchProjectionSync({ traceId, afterOutboxSeq: firstSync.cursorOutboxSeq, limit: 100, deltaOnly: true }); + assert.equal(caughtUp.events.length, 0); + assert.equal(caughtUp.cursorOutboxSeq, firstSync.cursorOutboxSeq); + } finally { + await pool.end(); + } + }, 30_000); +} + +async function project(store, { traceId, sessionId, sourceEventId, sourceSeq, sourceOffset, inputSha256, event }) { + const canonicalEvent = { + schema: "agentrun.event.v1", + eventId: sourceEventId, + sourceSeq, + traceId, + hwlabSessionId: sessionId, + runId: `run_${traceId}`, + commandId: `cmd_${traceId}`, + event + }; + const projectedEvent = { ...event, traceId, sessionId, sourceEventId, sourceSeq, runId: canonicalEvent.runId, commandId: canonicalEvent.commandId }; + return commitProjection(store, { + transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset, sourceKey: canonicalEvent.runId, inputSha256 }, + canonicalEvent, + projectedEvent, + requestMeta: { traceId, sessionId, valuesPrinted: false }, + factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({ projectedSeq, previousCheckpoint, projectedAt, event: projectedEvent }), + hwlabEvent: { schema: "hwlab.event.v1", eventId: `hwlab:${sourceEventId}`, traceId, sessionId, sourceEventId, sourceSeq, valuesPrinted: false }, + hwlabTopic: "hwlab.event.v1", + partitionKey: traceId, + headers: { sourceEventId } + }); +} + +function faultInjectingPool(pool, pattern) { + return { + async connect() { + const client = await pool.connect(); + let injected = false; + return { + async query(sql, params) { + if (!injected && pattern.test(String(sql))) { + injected = true; + const error = new Error("injected projection outbox failure"); + error.code = "XX2464"; + throw error; + } + return client.query(sql, params); + }, + release() { + client.release(); + } + }; + } + }; +} + +async function count(pool, table, where, params) { + if (!/^[a-z_]+$/u.test(table)) throw new Error("unsafe test table"); + const result = await pool.query(`SELECT COUNT(*)::int AS count FROM ${table} WHERE ${where}`, params); + return Number(result.rows[0].count); +} diff --git a/internal/db/runtime-store-postgres-kafka.test.ts b/internal/db/runtime-store-postgres-kafka.test.ts index dc56a647..d5f64614 100644 --- a/internal/db/runtime-store-postgres-kafka.test.ts +++ b/internal/db/runtime-store-postgres-kafka.test.ts @@ -9,6 +9,7 @@ import { completeHwlabKafkaOutbox, projectionOutboxEventId, recordFailedAgentRunKafkaMessage, + recordIgnoredAgentRunKafkaMessage, retryHwlabKafkaOutbox } from "./runtime-store-postgres-kafka.ts"; @@ -34,11 +35,35 @@ test("malformed duplicate source event at a new offset is durably DLQed without assert.equal(result.recorded, true); assert.equal(result.conflict, true); assert.equal(result.priorStatus, "projected"); - assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-duplicate"); + assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:0:11"); + assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-duplicate"); assert.equal(queries.some(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox")), false); assert.equal(queries.some(({ sql }) => sql.startsWith("INSERT INTO workbench_kafka_dlq")), true); }); +test("ignored Kafka messages acquire transport then event identity locks", async () => { + const queries = []; + const client = { + async query(sql, params) { + queries.push({ sql, params }); + if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] }; + if (sql.startsWith("SELECT input_sha256")) return { rows: [] }; + if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] }; + throw new Error(`unexpected SQL: ${sql}`); + } + }; + const result = await recordIgnoredAgentRunKafkaMessage(transactionStore(client), { + transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 2, sourceOffset: "31", sourceKey: "run-ignored", inputSha256: "hash-ignored" }, + envelope: { eventId: "evt-ignored", sourceSeq: 1, runId: "run-ignored" }, + reason: "hwlab-correlation-absent" + }); + + assert.equal(result.ignored, true); + assert.equal(result.duplicate, false); + assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:2:31"); + assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-ignored"); +}); + test("projection outbox derives a stable non-null identity when a fact has no source event", () => { const input = { fact: { traceId: "trc-derived", projectedSeq: 3, sourceSeq: 2 }, @@ -147,6 +172,44 @@ test("terminal Kafka transaction persists sealed turn and immutable SSE snapshot assert.equal(sse.payload.turn.finalResponse.text, "final body"); }); +test("late nonterminal Kafka event is acknowledged without advancing sealed projectedSeq", async () => { + const queries = []; + const client = { + async query(sql, params) { + queries.push({ sql, params }); + if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] }; + if (sql.startsWith("SELECT source_topic")) return { rows: [] }; + if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] }; + if (sql.startsWith("SELECT checkpoint_json")) return { rows: [{ checkpoint_json: { traceId: "trc_sealed", projectedSeq: 9, terminal: true, sealed: true } }] }; + if (sql.startsWith("SELECT owner_user_id")) return { rows: [] }; + if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 9 }] }; + if (sql.startsWith("INSERT INTO hwlab_kafka_outbox") || sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] }; + throw new Error(`unexpected SQL: ${sql}`); + } + }; + const store = { ...transactionStore(client), memory: { writeWorkbenchFacts() { throw new Error("suppressed facts must not reach memory"); } } }; + const result = await commitAgentRunKafkaProjection(store, { + transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "21", inputSha256: "hash-late" }, + canonicalEvent: { eventId: "evt-late", sourceSeq: 10, traceId: "trc_sealed", hwlabSessionId: "ses_sealed" }, + projectedEvent: { traceId: "trc_sealed", sessionId: "ses_sealed" }, + factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({ + projectedSeq, + previousCheckpoint, + projectedAt, + event: { traceId: "trc_sealed", sessionId: "ses_sealed", sourceEventId: "evt-late", sourceSeq: 10, status: "running", terminal: false } + }), + hwlabEvent: { eventId: "hwlab:evt-late" }, + hwlabTopic: "hwlab.event.v1", + partitionKey: "trc_sealed", + headers: {} + }); + + assert.equal(result.suppressedAfterSeal, true); + assert.equal(result.projectedSeq, 9); + const inboxUpdate = queries.find(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox")); + assert.equal(inboxUpdate.params[0], 9); +}); + test("HWLAB outbox claim and completion use attempt and lease fencing", async () => { const calls = []; const row = { outbox_seq: 3, event_id: "evt-3", topic: "hwlab.event.v1", partition_key: "trc-3", payload_json: {}, headers_json: {}, attempt_count: 4, lease_owner: "relay-a", lease_expires_at: "2026-07-10T10:05:00.000Z", next_attempt_at: "2026-07-10T10:00:00.000Z", created_at: "2026-07-10T10:00:00.000Z" }; diff --git a/internal/db/runtime-store-postgres-kafka.ts b/internal/db/runtime-store-postgres-kafka.ts index 72ae184c..690ec4a8 100644 --- a/internal/db/runtime-store-postgres-kafka.ts +++ b/internal/db/runtime-store-postgres-kafka.ts @@ -71,6 +71,9 @@ export async function commitAgentRunKafkaProjection(store, params = {}) { const persisted = await persistFacts(store, client, facts, params.requestMeta ?? {}); persistedEvents = persisted.events; } + const committedProjectedSeq = built?.written === false + ? nonNegativeInteger(built?.projectedSeq) || nonNegativeInteger(previousCheckpoint?.projectedSeq) || projectedSeq + : projectedSeq; const hwlabEvent = objectValue(params.hwlabEvent); const hwlabEventId = requiredText(hwlabEvent.eventId, "HWLAB Kafka eventId"); await client.query( @@ -79,9 +82,9 @@ export async function commitAgentRunKafkaProjection(store, params = {}) { ); await client.query( "UPDATE workbench_kafka_inbox SET status = 'projected', projected_seq = $1, updated_at = $2, projected_at = $2 WHERE source_topic = $3 AND source_partition = $4 AND source_offset = $5", - [projectedSeq, now, transport.sourceTopic, transport.sourcePartition, transport.sourceOffset] + [committedProjectedSeq, now, transport.sourceTopic, transport.sourcePartition, transport.sourceOffset] ); - return { duplicate: false, conflict: false, projectedSeq, facts, events: persistedEvents, suppressedAfterSeal: built?.suppressedAfterSeal === true }; + return { duplicate: false, conflict: false, projectedSeq: committedProjectedSeq, facts, events: persistedEvents, suppressedAfterSeal: built?.suppressedAfterSeal === true }; }); if (result.facts) store.memory.writeWorkbenchFacts({ facts: result.facts }, params.requestMeta ?? {}); return { ...result, transport, sourceEventId, traceId, sessionId, sourceSeq, valuesPrinted: false }; @@ -350,10 +353,12 @@ async function insertDlq(client, record) { } async function lockKafkaInboxIdentity(client, transport, sourceEventId) { - const identity = sourceEventId - ? `${transport.sourceTopic}:event:${sourceEventId}` - : `${transport.sourceTopic}:transport:${transport.sourcePartition}:${transport.sourceOffset}`; - await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${identity}`]); + const transportIdentity = `${transport.sourceTopic}:transport:${transport.sourcePartition}:${transport.sourceOffset}`; + await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${transportIdentity}`]); + if (sourceEventId) { + const eventIdentity = `${transport.sourceTopic}:event:${sourceEventId}`; + await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-kafka-inbox:${eventIdentity}`]); + } } function normalizeTransport(value = {}) { diff --git a/internal/db/runtime-store-postgres.ts b/internal/db/runtime-store-postgres.ts index 3f5b77c2..4a4518ed 100644 --- a/internal/db/runtime-store-postgres.ts +++ b/internal/db/runtime-store-postgres.ts @@ -763,6 +763,19 @@ export class PostgresCloudRuntimeStore { return withPersistence({ written: true, facts: persistedFacts, events: txResult.events }, this.summary()); } + async writeWorkbenchSessionAdmissionFact(params = {}, requestMeta = {}) { + const readiness = this.summary(); + if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) { + await this.assertReadyForWrites(); + } + const facts = normalizeWorkbenchFacts({ sessions: [params.fact ?? params.session ?? params] }, requestMeta, this.now()); + const fact = facts.sessions[0]; + if (!fact?.sessionId) throw new Error("Workbench session admission fact requires sessionId."); + await this.withDurableTransaction("workbench.session-admission.write", (client) => this.persistWorkbenchSessionAdmissionFact(fact, client)); + const memoryResult = this.memory.writeWorkbenchSessionAdmissionFact({ fact }, requestMeta); + return withPersistence({ written: true, admissionOnly: true, fact: memoryResult.fact, valuesPrinted: false }, this.summary()); + } + async commitAgentRunKafkaProjection(params = {}) { await this.assertReadyForWrites(); return commitAgentRunKafkaProjectionOperation(this, params); @@ -1181,6 +1194,13 @@ export class PostgresCloudRuntimeStore { ); } + async persistWorkbenchSessionAdmissionFact(record, client = this) { + await client.query( + "INSERT INTO workbench_sessions (session_id, owner_user_id, project_id, conversation_id, thread_id, status, last_trace_id, projected_seq, source_seq, source_event_id, terminal, sealed, session_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (session_id) DO UPDATE SET owner_user_id = COALESCE(workbench_sessions.owner_user_id, EXCLUDED.owner_user_id), project_id = COALESCE(workbench_sessions.project_id, EXCLUDED.project_id), conversation_id = COALESCE(workbench_sessions.conversation_id, EXCLUDED.conversation_id), thread_id = COALESCE(workbench_sessions.thread_id, EXCLUDED.thread_id), status = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.status ELSE workbench_sessions.status END, last_trace_id = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN COALESCE(EXCLUDED.last_trace_id, workbench_sessions.last_trace_id) ELSE workbench_sessions.last_trace_id END, session_json = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.session_json ELSE workbench_sessions.session_json END, updated_at = CASE WHEN workbench_sessions.projected_seq <= 0 AND NOT workbench_sessions.sealed THEN EXCLUDED.updated_at ELSE workbench_sessions.updated_at END", + [record.sessionId, record.ownerUserId, record.projectId, record.conversationId, record.threadId, record.status, record.lastTraceId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.terminal, record.sealed, stableJson(record), record.createdAt, record.updatedAt] + ); + } + async persistWorkbenchSessionInputFact(record, client = this) { await client.query( "INSERT INTO workbench_session_inputs (input_id, session_id, turn_id, trace_id, message_id, command_id, delivery, admitted_seq, promoted_seq, status, error_code, source_event_id, input_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT (input_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, trace_id = EXCLUDED.trace_id, message_id = EXCLUDED.message_id, command_id = COALESCE(EXCLUDED.command_id, workbench_session_inputs.command_id), delivery = EXCLUDED.delivery, admitted_seq = CASE WHEN workbench_session_inputs.admitted_seq > 0 THEN workbench_session_inputs.admitted_seq ELSE EXCLUDED.admitted_seq END, promoted_seq = COALESCE(EXCLUDED.promoted_seq, workbench_session_inputs.promoted_seq), status = EXCLUDED.status, error_code = EXCLUDED.error_code, source_event_id = COALESCE(EXCLUDED.source_event_id, workbench_session_inputs.source_event_id), input_json = EXCLUDED.input_json, updated_at = EXCLUDED.updated_at",