diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index d456cb6a..655e02c6 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1222,6 +1222,58 @@ test("workbench read model keeps non-terminal durable tool completed events out } }); +test("workbench trace events API blocks historical projectedSeq collisions instead of handing them to renderer", async () => { + const traceStore = createCodeAgentTraceStore(); + const traceId = "trc_workbench_projected_seq_collision"; + const session = { + id: "ses_workbench_projected_seq_collision", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_projected_seq_collision", + threadId: "thread-workbench-projected-seq-collision", + lastTraceId: traceId, + updatedAt: "2026-06-20T12:20:00.000Z", + session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "collision", traceId, status: "sent" }] } + }; + const accessController = { + store: { + async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const runtimeStore = createDurableFactsRuntimeStore({ + sessions: [{ + session, + events: [ + { id: "wte_collision_a", projectedSeq: 1, sourceSeq: 1, sourceEventId: "source-a", type: "backend", status: "running", label: "projected:1:a", createdAt: "2026-06-20T12:19:58.000Z" }, + { id: "wte_collision_b", projectedSeq: 1, sourceSeq: 2, sourceEventId: "source-b", type: "assistant", status: "running", label: "projected:1:b", message: "later", createdAt: "2026-06-20T12:19:59.000Z" } + ], + status: "running", + projectionStatus: "projecting", + lastProjectedSeq: 1 + }] + }); + const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`); + assert.equal(trace.status, 200); + assert.equal(trace.body.status, "blocked"); + assert.equal(trace.body.projectionStatus, "blocked"); + assert.equal(trace.body.projectionHealth, "degraded"); + assert.equal(trace.body.blocker.code, "projected_seq_collision"); + assert.deepEqual(trace.body.events, []); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench realtime stream accepts session authority without project or workspace", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 21f6f002..5da046e9 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -933,22 +933,23 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 }); if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error)); const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status }); + const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection; sendJson(response, 200, { ok: true, - status: "succeeded", + status: page.blocker ? "blocked" : "succeeded", contractVersion: "workbench-trace-events-v1", traceId, sessionId: factSessionId(session), threadId: safeOpaqueId(session?.threadId) ?? (textValue(session?.threadId) || null), ...page, - projection, - projectionStatus: projection.projectionStatus, - projectionHealth: projection.projectionHealth, - lastProjectedSeq: projection.lastProjectedSeq, - sourceRunId: projection.sourceRunId, - sourceCommandId: projection.sourceCommandId, - staleMs: projection.staleMs, - blocker: projection.blocker, + projection: responseProjection, + projectionStatus: responseProjection.projectionStatus, + projectionHealth: responseProjection.projectionHealth, + lastProjectedSeq: responseProjection.lastProjectedSeq, + sourceRunId: responseProjection.sourceRunId, + sourceCommandId: responseProjection.sourceCommandId, + staleMs: responseProjection.staleMs, + blocker: responseProjection.blocker, valuesRedacted: true, secretMaterialStored: false }); @@ -956,6 +957,8 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT function traceEventPageFromFacts(sourceEvents, options, metadata = {}) { const rows = factArray(sourceEvents).filter((event) => factProjectedSeq(event)).sort(compareFactTraceEventsAsc); + const collision = projectedSeqCollision(rows); + if (collision) return blockedTraceEventPage(options, metadata, collision); const pageRows = rows.slice(0, options.limit); const events = pageRows.map(factTraceEventDto).filter(Boolean); const toProjectedSeq = events.length ? events.at(-1).projectedSeq : null; @@ -980,6 +983,62 @@ function traceEventPageFromFacts(sourceEvents, options, metadata = {}) { }; } +function projectedSeqCollision(rows = []) { + const seen = new Map(); + for (const row of rows) { + const projectedSeq = factProjectedSeq(row); + if (!projectedSeq) continue; + const id = textValue(row?.id ?? row?.sourceEventId) || null; + const existing = seen.get(projectedSeq); + if (existing && existing.id !== id) { + return { + code: "projected_seq_collision", + category: "workbench-projection", + layer: "workbench-trace-events", + message: "Trace projection contains duplicate projectedSeq values; canonical event order is blocked until projection is rebuilt.", + projectedSeq, + eventIds: [existing.id, id].filter(Boolean), + retryable: false, + valuesPrinted: false + }; + } + if (!existing) seen.set(projectedSeq, { id }); + } + return null; +} + +function blockedTraceEventPage(options, metadata = {}, blocker) { + const total = Number.isFinite(Number(metadata.total)) ? Math.trunc(Number(metadata.total)) : null; + return { + events: [], + eventCount: total ?? options.afterProjectedSeq, + range: { + afterProjectedSeq: options.afterProjectedSeq, + fromProjectedSeq: null, + toProjectedSeq: null, + limit: options.limit, + returned: 0, + total + }, + hasMore: false, + nextProjectedSeq: options.afterProjectedSeq, + nextCursor: null, + traceStatus: metadata.traceStatus ?? "unknown", + updatedAt: null, + blocker + }; +} + +function blockedTraceEventProjection(projection = {}, blocker) { + return { + ...projection, + projectionStatus: "blocked", + projectionHealth: "degraded", + blocker, + valuesRedacted: true + }; +} + async function visibleSessionById(store, sessionId, actor) { const safeId = safeSessionId(sessionId); if (!safeId) return null; diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index 0cc9b73f..b4c97561 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -4,6 +4,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; +import { createCloudRuntimeStore } from "../db/runtime-store.ts"; import { writeWorkbenchProjectionEvent, writeWorkbenchProjectionSession } from "./workbench-projection-writer.ts"; test("workbench projection writer commits terminal owner evidence as sealed durable facts", async () => { @@ -77,10 +78,15 @@ test("workbench projection writer commits terminal owner evidence as sealed dura test("workbench projection event commit writes trace event and checkpoint facts", async () => { const factWrites = []; + const baseStore = createCloudRuntimeStore({ now: () => "2026-06-20T11:00:00.000Z" }); const runtimeStore = { + allocateWorkbenchProjectedSeq(params, requestMeta) { + return baseStore.allocateWorkbenchProjectedSeq(params, requestMeta); + }, async writeWorkbenchFacts(params, requestMeta) { + const result = baseStore.writeWorkbenchFacts(params, requestMeta); factWrites.push({ params, requestMeta }); - return { written: true, facts: params.facts }; + return result; } }; @@ -117,9 +123,45 @@ test("workbench projection event commit writes trace event and checkpoint facts" }); assert.equal(factWrites.length, 2); - assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 7); + assert.equal(factWrites[0].params.facts.traceEvents[0].projectedSeq, 1); assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting"); assert.equal(factWrites[1].params.facts.sessions[0].status, "completed"); assert.equal(factWrites[1].params.facts.traceEvents[0].sealed, true); + assert.equal(factWrites[1].params.facts.traceEvents[0].projectedSeq, 2); assert.equal(factWrites[1].params.facts.checkpoints[0].projectionStatus, "caught_up"); }); + +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", + 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", + 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); +}); diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index d270caf5..638e40a3 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -54,17 +54,36 @@ export async function writeWorkbenchProjectionSession({ accessController, runtim export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event = {}, requestMeta = {} } = {}) { 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 sourceSeq = nonNegativeInteger(event.sourceSeq ?? event.seq); - const projectedSeq = nonNegativeInteger(event.seq ?? event.projectedSeq ?? sourceSeq); const occurredAt = timestampValue(event.createdAt ?? event.occurredAt ?? event.updatedAt); const sessionId = textValue(event.sessionId ?? requestMeta.sessionId) || null; const turnId = textValue(event.turnId) || traceId; const terminal = event.terminal === true || TERMINAL_STATUSES.has(normalizeWorkbenchStatus(event.status)); const status = terminal ? normalizeWorkbenchStatus(event.status) : "running"; - const sourceEventId = textValue(event.sourceEventId ?? event.id) || `${traceId}:${sourceSeq || projectedSeq}`; - const eventId = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId, sourceSeq, projectedSeq, label: event.label, occurredAt }); + const 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: textValue(event.eventType ?? event.type ?? event.label) || "event", + occurredAt, + updatedAt: timestampValue(event.updatedAt ?? occurredAt) + }, { + 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 = textValue(event.id) || stableFactId("wte", { traceId, sourceEventId }); const facts = { sessions: sessionId ? [{ sessionId, @@ -279,6 +298,26 @@ function stableFactId(prefix, value) { return `${prefix}_${createHash("sha256").update(JSON.stringify(sortJson(value))).digest("hex").slice(0, 32)}`; } +function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } = {}) { + const explicit = textValue(event.sourceEventId ?? event.id); + if (explicit) return explicit; + const source = textValue(event.source); + const label = textValue(event.label ?? event.type ?? event.eventType) || "event"; + if (source && sourceSeq > 0) return `${source}:${sourceSeq}:${label}`; + return stableFactId("wte_source", { + traceId, + source: source || null, + sourceSeq, + label, + eventType: textValue(event.eventType ?? event.type) || null, + status: textValue(event.status) || null, + itemId: textValue(event.itemId ?? event.messageId) || null, + runId: textValue(event.runId) || null, + commandId: textValue(event.commandId) || null, + occurredAt + }); +} + function sortJson(value) { if (Array.isArray(value)) return value.map(sortJson); if (value && typeof value === "object") return Object.fromEntries(Object.keys(value).sort().map((key) => [key, sortJson(value[key])])); diff --git a/internal/db/runtime-store.test.ts b/internal/db/runtime-store.test.ts index 4d7e59b6..52bf8f15 100644 --- a/internal/db/runtime-store.test.ts +++ b/internal/db/runtime-store.test.ts @@ -806,6 +806,41 @@ test("configured postgres runtime persists and queries Workbench durable facts", assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT session_json FROM workbench_sessions WHERE session_id = $1"))); }); +test("configured postgres runtime allocates Workbench projectedSeq idempotently by source event", async () => { + const queryClient = createFakePostgresClient({ migrationReady: true }); + const store = createConfiguredCloudRuntimeStore({ + env: { + HWLAB_CLOUD_RUNTIME_ADAPTER: "postgres", + HWLAB_CLOUD_RUNTIME_DURABLE: "true" + }, + dbUrl: "postgres://hwlab_redacted@db.example.invalid:5432/hwlab", + queryClient, + now: () => "2026-06-20T12:30:00.000Z" + }); + const traceId = "trc_pg_projected_seq_idempotent"; + + const first = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" }); + await store.writeWorkbenchFacts({ facts: { + traceEvents: [{ id: "wte_pg_source_a", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a", projectedSeq: first.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:01.000Z" }], + checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: first.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:01.000Z" }] + } }); + const second = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b" }); + await store.writeWorkbenchFacts({ facts: { + traceEvents: [{ id: "wte_pg_source_b", traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-b", projectedSeq: second.projectedSeq, eventType: "backend", occurredAt: "2026-06-20T12:30:02.000Z" }], + checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: second.projectedSeq, sourceSeq: 1, sourceEventId: "source-b", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:02.000Z" }] + } }); + const repeated = await store.allocateWorkbenchProjectedSeq({ traceId, sessionId: "ses_pg_projected_seq_idempotent", sourceSeq: 1, sourceEventId: "source-a" }); + await store.writeWorkbenchFacts({ facts: { checkpoints: [{ traceId, sessionId: "ses_pg_projected_seq_idempotent", projectedSeq: repeated.projectedSeq, sourceSeq: 1, sourceEventId: "source-a", projectionStatus: "projecting", projectionHealth: "healthy", updatedAt: "2026-06-20T12:30:03.000Z" }] } }); + + const loaded = await store.queryWorkbenchFacts({ traceId, families: ["traceEvents", "checkpoints"], limit: 10 }); + assert.equal(first.projectedSeq, 1); + assert.equal(second.projectedSeq, 2); + assert.equal(repeated.projectedSeq, 1); + assert.deepEqual(loaded.facts.traceEvents.map((event) => event.projectedSeq), [1, 2]); + assert.equal(loaded.facts.checkpoints[0].projectedSeq, 2); + assert.ok(queryClient.calls.some((call) => call.sql.startsWith("SELECT pg_advisory_xact_lock"))); +}); + test("configured postgres runtime reuses proven readiness for Workbench durable reads", async () => { const queryClient = createFakePostgresClient({ migrationReady: true }); const store = createConfiguredCloudRuntimeStore({ @@ -1000,6 +1035,9 @@ function createFakePostgresClient({ calls, async query(sql, params = []) { calls.push({ sql, params }); + if (["BEGIN", "COMMIT", "ROLLBACK"].includes(sql)) { + return { rows: [] }; + } if (sql.includes("information_schema.columns")) { return { rows: schemaRows() }; } @@ -1030,6 +1068,23 @@ function createFakePostgresClient({ if (sql.startsWith("CREATE INDEX IF NOT EXISTS idx_workbench_")) { return { rows: [] }; } + if (sql.startsWith("SELECT pg_advisory_xact_lock")) { + return { rows: [{ pg_advisory_xact_lock: null }] }; + } + if (sql.startsWith("SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2")) { + const rows = [...state.workbench_trace_events.values()] + .filter((record) => record.trace_id === params[0] && record.source_event_id === params[1]) + .sort((left, right) => Number(left.projected_seq) - Number(right.projected_seq)); + return { rows: rows.slice(0, 1).map((record) => ({ projected_seq: record.projected_seq })) }; + } + if (sql.startsWith("SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events")) { + const traceId = params[0]; + const eventMax = [...state.workbench_trace_events.values()] + .filter((record) => record.trace_id === traceId) + .reduce((max, record) => Math.max(max, Number(record.projected_seq ?? 0)), 0); + const checkpointMax = Number(state.workbench_projection_checkpoints.get(traceId)?.projected_seq ?? 0); + return { rows: [{ max_projected_seq: Math.max(eventMax, checkpointMax) }] }; + } if (sql.startsWith("SELECT gateway_session_json FROM gateway_sessions")) { return jsonSelect(state.gateway_sessions, params[0], "gateway_session_json"); } @@ -1206,10 +1261,19 @@ function createFakePostgresClient({ return { rows: [] }; } if (sql.startsWith("INSERT INTO workbench_projection_checkpoints")) { + const previous = state.workbench_projection_checkpoints.get(params[0]) ?? null; + const projectedSeq = sql.includes("RETURNING projected_seq") + ? Math.max(Number(previous?.projected_seq ?? 0) + (previous ? 1 : 0), Number(params[5] ?? 0)) + : previous && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0) + ? Number(previous.projected_seq) + : Number(params[5] ?? 0); + if (previous && !sql.includes("RETURNING projected_seq") && Number(params[5] ?? 0) < Number(previous.projected_seq ?? 0)) { + return { rows: [] }; + } state.workbench_projection_checkpoints.set(params[0], { - trace_id: params[0], session_id: params[1], turn_id: params[2], run_id: params[3], command_id: params[4], projected_seq: params[5], source_seq: params[6], source_event_id: params[7], projection_status: params[8], projection_health: params[9], terminal: params[10], sealed: params[11], diagnostic_json: params[12], checkpoint_json: params[13], created_at: params[14], updated_at: params[15] + trace_id: params[0], session_id: params[1], turn_id: params[2], run_id: params[3], command_id: params[4], projected_seq: projectedSeq, source_seq: params[6], source_event_id: params[7], projection_status: params[8], projection_health: params[9], terminal: params[10], sealed: params[11], diagnostic_json: params[12], checkpoint_json: params[13], created_at: params[14], updated_at: params[15] }); - return { rows: [] }; + return { rows: sql.includes("RETURNING projected_seq") ? [{ projected_seq: projectedSeq }] : [] }; } throw new Error(`unexpected sql: ${sql}`); } diff --git a/internal/db/runtime-store.ts b/internal/db/runtime-store.ts index 106e5628..5449fe5a 100644 --- a/internal/db/runtime-store.ts +++ b/internal/db/runtime-store.ts @@ -574,6 +574,33 @@ export class CloudRuntimeStore { }; } + allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) { + const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now()); + const existing = [...this.workbenchTraceEvents.values()].find((record) => + record.traceId === allocation.traceId && record.sourceEventId === allocation.sourceEventId + ); + if (existing) { + return { + ...allocation, + projectedSeq: nonNegativeInteger(existing.projectedSeq), + reused: true, + persistence: this.summary(), + valuesPrinted: false + }; + } + const eventMax = [...this.workbenchTraceEvents.values()] + .filter((record) => record.traceId === allocation.traceId) + .reduce((max, record) => Math.max(max, nonNegativeInteger(record.projectedSeq)), 0); + const checkpointMax = nonNegativeInteger(this.workbenchProjectionCheckpoints.get(allocation.traceId)?.projectedSeq); + return { + ...allocation, + projectedSeq: Math.max(eventMax, checkpointMax) + 1, + reused: false, + persistence: this.summary(), + valuesPrinted: false + }; + } + queryWorkbenchProjectionStates(params = {}) { const dueAt = textOr(params.dueAt, ""); const statuses = Array.isArray(params.projectionStatuses) ? new Set(params.projectionStatuses.map((item) => textOr(item, "")).filter(Boolean)) : null; @@ -596,7 +623,12 @@ export class CloudRuntimeStore { for (const fact of facts.parts) this.workbenchParts.set(fact.partId, fact); for (const fact of facts.turns) this.workbenchTurns.set(fact.turnId, fact); for (const fact of facts.traceEvents) this.workbenchTraceEvents.set(fact.id, fact); - for (const fact of facts.checkpoints) this.workbenchProjectionCheckpoints.set(fact.traceId, fact); + for (const fact of facts.checkpoints) { + const previous = this.workbenchProjectionCheckpoints.get(fact.traceId) ?? null; + if (!previous || nonNegativeInteger(fact.projectedSeq) >= nonNegativeInteger(previous.projectedSeq)) { + this.workbenchProjectionCheckpoints.set(fact.traceId, fact); + } + } return { written: true, facts, @@ -1198,6 +1230,40 @@ export class PostgresCloudRuntimeStore { }; } + async allocateWorkbenchProjectedSeq(params = {}, requestMeta = {}) { + const readiness = this.summary(); + if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) { + await this.assertReadyForWrites(); + } + const allocation = normalizeWorkbenchProjectionAllocation(params, requestMeta, this.now()); + const result = await this.withDurableTransaction("workbench.projected-seq.allocate", async (client) => { + await client.query("SELECT pg_advisory_xact_lock(hashtext($1))", [`workbench-projection:${allocation.traceId}`]); + const existing = await client.query( + "SELECT projected_seq FROM workbench_trace_events WHERE trace_id = $1 AND source_event_id = $2 ORDER BY projected_seq ASC LIMIT 1", + [allocation.traceId, allocation.sourceEventId] + ); + const existingSeq = nonNegativeInteger(existing.rows?.[0]?.projected_seq); + if (existingSeq > 0) return { projectedSeq: existingSeq, reused: true }; + const maxResult = await client.query( + "SELECT GREATEST(COALESCE((SELECT MAX(projected_seq) FROM workbench_trace_events WHERE trace_id = $1), 0), COALESCE((SELECT projected_seq FROM workbench_projection_checkpoints WHERE trace_id = $1), 0))::int AS max_projected_seq", + [allocation.traceId] + ); + const proposedSeq = nonNegativeInteger(maxResult.rows?.[0]?.max_projected_seq) + 1; + const reserve = await client.query( + "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq + 1, EXCLUDED.projected_seq), updated_at = EXCLUDED.updated_at RETURNING projected_seq", + [allocation.traceId, allocation.sessionId, allocation.turnId, allocation.runId, allocation.commandId, proposedSeq, allocation.sourceSeq, allocation.sourceEventId, "projecting", "healthy", false, false, stableJson({ allocator: "workbench-projected-seq", valuesRedacted: true }), stableJson({ ...allocation, projectedSeq: proposedSeq, projectionStatus: "projecting", projectionHealth: "healthy", valuesPrinted: false }), allocation.createdAt, allocation.updatedAt] + ); + return { projectedSeq: nonNegativeInteger(reserve.rows?.[0]?.projected_seq) || proposedSeq, reused: false }; + }); + return { + ...allocation, + projectedSeq: result.projectedSeq, + reused: result.reused, + persistence: this.summary(), + valuesPrinted: false + }; + } + async writeWorkbenchFacts(params = {}, requestMeta = {}) { const readiness = this.summary(); if (readiness.ready !== true || readiness.durable !== true || readiness.liveRuntimeEvidence !== true) { @@ -1566,7 +1632,7 @@ export class PostgresCloudRuntimeStore { async persistWorkbenchProjectionCheckpoint(record) { await this.query( - "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = EXCLUDED.session_id, turn_id = EXCLUDED.turn_id, run_id = EXCLUDED.run_id, command_id = EXCLUDED.command_id, projected_seq = EXCLUDED.projected_seq, source_seq = EXCLUDED.source_seq, source_event_id = EXCLUDED.source_event_id, projection_status = EXCLUDED.projection_status, projection_health = EXCLUDED.projection_health, terminal = EXCLUDED.terminal, sealed = EXCLUDED.sealed, diagnostic_json = EXCLUDED.diagnostic_json, checkpoint_json = EXCLUDED.checkpoint_json, updated_at = EXCLUDED.updated_at", + "INSERT INTO workbench_projection_checkpoints (trace_id, session_id, turn_id, run_id, command_id, projected_seq, source_seq, source_event_id, projection_status, projection_health, terminal, sealed, diagnostic_json, checkpoint_json, created_at, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT (trace_id) DO UPDATE SET session_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.session_id ELSE workbench_projection_checkpoints.session_id END, turn_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.turn_id ELSE workbench_projection_checkpoints.turn_id END, run_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.run_id ELSE workbench_projection_checkpoints.run_id END, command_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.command_id ELSE workbench_projection_checkpoints.command_id END, projected_seq = GREATEST(workbench_projection_checkpoints.projected_seq, EXCLUDED.projected_seq), source_seq = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_seq ELSE workbench_projection_checkpoints.source_seq END, source_event_id = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.source_event_id ELSE workbench_projection_checkpoints.source_event_id END, projection_status = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_status ELSE workbench_projection_checkpoints.projection_status END, projection_health = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.projection_health ELSE workbench_projection_checkpoints.projection_health END, terminal = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.terminal ELSE workbench_projection_checkpoints.terminal END, sealed = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.sealed ELSE workbench_projection_checkpoints.sealed END, diagnostic_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.diagnostic_json ELSE workbench_projection_checkpoints.diagnostic_json END, checkpoint_json = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.checkpoint_json ELSE workbench_projection_checkpoints.checkpoint_json END, updated_at = CASE WHEN EXCLUDED.projected_seq >= workbench_projection_checkpoints.projected_seq THEN EXCLUDED.updated_at ELSE workbench_projection_checkpoints.updated_at END", [record.traceId, record.sessionId, record.turnId, record.runId, record.commandId, record.projectedSeq, record.sourceSeq, record.sourceEventId, record.projectionStatus, record.projectionHealth, record.terminal, record.sealed, stableJson(record.diagnostic), stableJson(record), record.createdAt, record.updatedAt] ); } @@ -1623,6 +1689,33 @@ export class PostgresCloudRuntimeStore { return client.query(sql, params); } + async withDurableTransaction(label, fn) { + const client = await this.getQueryClient(); + if (typeof client.connect === "function") { + const tx = await client.connect(); + try { + await tx.query("BEGIN"); + const result = await fn(tx); + await tx.query("COMMIT"); + return result; + } catch (error) { + try { await tx.query("ROLLBACK"); } catch {} + throw error; + } finally { + tx.release?.(); + } + } + await client.query("BEGIN"); + try { + const result = await fn(client); + await client.query("COMMIT"); + return result; + } catch (error) { + try { await client.query("ROLLBACK"); } catch {} + throw error; + } + } + async getQueryClient() { if (this.queryClient) { return this.queryClient; @@ -1839,6 +1932,33 @@ function normalizeWorkbenchProjectionState(value, requestMeta = {}, now) { }); } +function normalizeWorkbenchProjectionAllocation(value, requestMeta = {}, now) { + const input = normalizeJsonObject(value); + const traceId = textOr(input.traceId ?? requestMeta.traceId, ""); + const sourceEventId = textOr(input.sourceEventId ?? input.eventId, ""); + if (!traceId || !sourceEventId) { + throw new HwlabProtocolError("workbench projectedSeq allocation requires traceId and sourceEventId", { + code: ERROR_CODES.invalidParams, + data: { required: ["traceId", "sourceEventId"] } + }); + } + const timestamp = timestampOr(input.updatedAt ?? input.occurredAt ?? input.createdAt, now); + return pruneUndefined({ + ...input, + traceId, + sessionId: textOr(input.sessionId ?? requestMeta.sessionId, "") || null, + turnId: textOr(input.turnId ?? requestMeta.turnId ?? traceId, "") || null, + runId: textOr(input.runId ?? input.sourceRunId, "") || null, + commandId: textOr(input.commandId ?? input.sourceCommandId, "") || null, + sourceSeq: nonNegativeInteger(input.sourceSeq ?? input.lastSourceSeq ?? input.lastAgentRunSeq), + sourceEventId, + eventType: textOr(input.eventType ?? input.type, "event"), + createdAt: timestampOr(input.createdAt, timestamp), + updatedAt: timestamp, + valuesPrinted: false + }); +} + function normalizeWorkbenchFacts(value, requestMeta = {}, now) { const input = normalizeJsonObject(value); return { diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index ae32e9f3..a9e77d66 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -16,9 +16,6 @@ export function traceDisplayRows(trace: Record = {}, events: Tr const rows: TraceEventRow[] = []; const renderedSourceEvents = new Set(); const renderedToolIdentities = new Set(); - const assistantRows: AssistantRowState[] = []; - let lastAssistantRowIndex = -1; - let completionEvent: TraceEvent | null = null; for (let index = 0; index < events.length; index += 1) { const event = events[index]; const sourceEventKey = traceSourceEventKey(event); @@ -44,30 +41,20 @@ export function traceDisplayRows(trace: Record = {}, events: Tr continue; } if (isTerminalAssistantTraceEvent(event)) { - const assistantRowIndex = upsertAssistantMessageRow(rows, assistantRows, effectiveTrace, event, { terminal: true }); - if (assistantRowIndex >= 0) lastAssistantRowIndex = assistantRowIndex; + rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true })); continue; } if (isAssistantTraceEvent(event)) { - const assistantRowIndex = upsertAssistantMessageRow(rows, assistantRows, effectiveTrace, event, { terminal: false }); - if (assistantRowIndex >= 0) lastAssistantRowIndex = assistantRowIndex; + rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false })); continue; } if (isCompletionTraceEvent(event)) { - completionEvent = event; + const finalResponseText = traceFinalResponseText(effectiveTrace); + rows.push(finalResponseText ? traceFinalResponseRow(event, finalResponseText) : traceCompletionSummaryRow(effectiveTrace, event)); continue; } rows.push(traceDisplayRow(effectiveTrace, event)); } - if (completionEvent) { - const finalResponseText = traceFinalResponseText(effectiveTrace); - if (finalResponseText) { - lastAssistantRowIndex = upsertAuthoritativeFinalResponseRow(rows, assistantRows, completionEvent, finalResponseText); - } - const lastAssistantRow = lastAssistantRowIndex >= 0 ? rows[lastAssistantRowIndex] : undefined; - if (lastAssistantRow) rows[lastAssistantRowIndex] = markAssistantRowTerminal(effectiveTrace, lastAssistantRow, completionEvent); - else rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent)); - } const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, events); if (progressRow) rows.push(progressRow); if (rows.length > 0) return rows; @@ -113,18 +100,6 @@ function stripTraceToolOutputFromSummary(value: string): string { return value.replace(/\s+(stdout:|stderr:|exitCode=).*$/isu, "").trim(); } -interface AssistantRowState { - rowIndex: number; - comparableText: string; - sourceEvent: TraceEvent; - derivedSnapshotSuffix?: boolean; -} - -interface AssistantSnapshotDecision { - action: "replace" | "append-suffix" | "keep-existing"; - suffixText?: string; -} - function traceToolCallRow(trace: Record, event: TraceEvent): TraceEventRow { const command = cleanShellCommand(event.command); const status = traceStatusToken(event); @@ -211,146 +186,23 @@ function traceAssistantMessageRow(trace: Record, event: TraceEv }; } -function upsertAssistantMessageRow(rows: TraceEventRow[], assistantRows: AssistantRowState[], trace: Record, event: TraceEvent, options: { terminal: boolean }): number { - const row = traceAssistantMessageRow(trace, event, options); - const comparableText = comparableAssistantText(row.body); - if (!comparableText) { - rows.push(row); - assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: event }); - return rows.length - 1; - } - - for (let index = assistantRows.length - 1; index >= 0; index -= 1) { - const state = assistantRows[index]; - if (!state) continue; - const existingRow = rows[state.rowIndex]; - if (!existingRow) continue; - if (state.comparableText === comparableText) { - rows[state.rowIndex] = mergeAssistantRows(existingRow, row); - state.sourceEvent = event; - state.derivedSnapshotSuffix = false; - return state.rowIndex; - } - if (comparableText.includes(state.comparableText)) { - const previousIndex = comparableText.indexOf(state.comparableText); - if (state.derivedSnapshotSuffix === true && previousIndex === 0) { - rows[state.rowIndex] = mergeAssistantRows(row, existingRow); - state.comparableText = comparableText; - state.sourceEvent = event; - state.derivedSnapshotSuffix = false; - return state.rowIndex; - } - const snapshotDecision = assistantSnapshotDecision(comparableText, state.comparableText, event, state.sourceEvent); - if (snapshotDecision.action === "append-suffix" && snapshotDecision.suffixText) { - const suffixRow = { ...row, body: snapshotDecision.suffixText }; - rows.push(suffixRow); - assistantRows.push({ rowIndex: rows.length - 1, comparableText: snapshotDecision.suffixText, sourceEvent: event, derivedSnapshotSuffix: true }); - return rows.length - 1; - } - if (snapshotDecision.action === "keep-existing") return state.rowIndex; - rows[state.rowIndex] = mergeAssistantRows(row, existingRow); - state.comparableText = comparableText; - state.sourceEvent = event; - state.derivedSnapshotSuffix = false; - return state.rowIndex; - } - if (state.comparableText.includes(comparableText)) { - rows[state.rowIndex] = mergeAssistantRows(existingRow, row); - return state.rowIndex; - } - } - - rows.push(row); - assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: event }); - return rows.length - 1; -} - -function upsertAuthoritativeFinalResponseRow(rows: TraceEventRow[], assistantRows: AssistantRowState[], completionEvent: TraceEvent, finalText: string): number { - const comparableText = comparableAssistantText(finalText); - let matchedState: AssistantRowState | null = null; - for (const state of assistantRows) { - const row = rows[state.rowIndex]; - if (!row || row.bodyFormat !== "markdown") continue; - row.terminal = undefined; - if (!matchedState && state.comparableText === comparableText) matchedState = state; - } - if (matchedState) { - const row = rows[matchedState.rowIndex]; - if (row) { - rows[matchedState.rowIndex] = { - ...row, - body: finalText, - terminal: true, - tone: "ok", - header: `${traceEventClock(completionEvent)} 助手最终消息` - }; - matchedState.comparableText = comparableText; - matchedState.derivedSnapshotSuffix = false; - return matchedState.rowIndex; - } - } - rows.push({ - rowId: `trace-final-response:${completionEvent.seq ?? "completed"}`, - seq: numberOrNull(completionEvent.seq), +function traceFinalResponseRow(event: TraceEvent, finalText: string): TraceEventRow { + return { + rowId: `trace-final-response:${event.seq ?? "completed"}`, + seq: numberOrNull(event.seq), tone: "ok", - header: `${traceEventClock(completionEvent)} 助手最终消息`, + header: `${traceEventClock(event)} 助手最终消息`, terminal: true, body: finalText, bodyFormat: "markdown" - }); - assistantRows.push({ rowIndex: rows.length - 1, comparableText, sourceEvent: completionEvent }); - return rows.length - 1; -} - -function assistantSnapshotDecision(nextText: string, previousText: string, nextEvent: TraceEvent, previousEvent: TraceEvent): AssistantSnapshotDecision { - if (!sameAssistantSnapshotIdentity(nextEvent, previousEvent)) return { action: "replace" }; - if (isAuthoritativeAssistantEvent(nextEvent) || isAuthoritativeAssistantEvent(previousEvent)) return { action: "replace" }; - const previousIndex = nextText.indexOf(previousText); - if (previousIndex < 0) return { action: "replace" }; - const suffixText = nextText.slice(previousIndex + previousText.length).trim(); - if (suffixText && suffixText.length >= Math.max(80, previousText.length / 3)) return { action: "append-suffix", suffixText }; - if (previousIndex > 0 || !suffixText) return { action: "keep-existing" }; - return { action: "replace" }; -} - -function sameAssistantSnapshotIdentity(left: TraceEvent, right: TraceEvent): boolean { - const leftItem = nonEmptyString(left.itemId ?? left.messageId ?? left.id); - const rightItem = nonEmptyString(right.itemId ?? right.messageId ?? right.id); - if (leftItem && rightItem) return leftItem === rightItem; - return false; -} - -function isAuthoritativeAssistantEvent(event: TraceEvent): boolean { - return event.replyAuthority === true || event.final === true || event.terminal === true || String(event.status ?? "") === "completed"; -} - -function mergeAssistantRows(preferred: TraceEventRow, metadata: TraceEventRow): TraceEventRow { - return { - ...preferred, - terminal: preferred.terminal || metadata.terminal ? true : undefined, - header: metadata.terminal && !preferred.terminal ? metadata.header : preferred.header, - tone: preferred.tone === "ok" || metadata.tone === "ok" ? "ok" : preferred.tone }; } -function comparableAssistantText(value: unknown): string { - return cleanTraceText(value).replace(/\s+/gu, " "); -} - function traceFinalResponseText(trace: Record): string | null { const response = trace.finalResponse && typeof trace.finalResponse === "object" ? trace.finalResponse as Record : null; return nonEmptyString(response?.text ?? response?.content ?? response?.message); } -function markAssistantRowTerminal(trace: Record, row: TraceEventRow, completionEvent: TraceEvent): TraceEventRow { - return { - ...row, - tone: "ok", - terminal: true, - header: `${row.header}，轮次完成（总耗时 ${formatTraceDuration(traceRelativeMs(trace, completionEvent))}）` - }; -} - function traceCompletionSummaryRow(trace: Record, event: TraceEvent): TraceEventRow { return { rowId: `trace-completion:${event.seq ?? "turn"}`, diff --git a/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts b/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts index 36d4f3d6..844e2c41 100644 --- a/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts @@ -78,6 +78,32 @@ test("R1 shared trace renderer suppresses AgentRun reuse diagnostics", () => { assert.doesNotMatch(rows[0]?.header ?? "", /reused|ensured|runner-job/u); }); +test("R1 shared trace renderer keeps terminal assistant at its own event position", () => { + const rows = traceDisplayRows({ traceId: "trc_terminal_position", status: "completed", startedAt: "2026-06-20T12:00:00.000Z" }, [ + { seq: 1, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "FINAL", createdAt: "2026-06-20T12:00:01.000Z" }, + { seq: 2, label: "item/commandExecution:completed", type: "tool_call", toolName: "commandExecution", status: "completed", command: "make test", createdAt: "2026-06-20T12:00:02.000Z" }, + { seq: 3, label: "agentrun:assistant:message", type: "assistant", status: "completed", message: "FINAL", final: true, replyAuthority: true, terminal: true, createdAt: "2026-06-20T12:00:03.000Z" } + ]); + const toolIndex = rows.findIndex((row) => row.seq === 2); + const finalIndex = rows.findIndex((row) => row.seq === 3 && row.terminal === true && row.body === "FINAL"); + assert.ok(toolIndex >= 0); + assert.ok(finalIndex > toolIndex); + assert.equal(rows.some((row) => row.seq === 1 && row.terminal === true), false); +}); + +test("R1 shared trace renderer keeps completion final response at completion event position", () => { + const rows = traceDisplayRows({ traceId: "trc_completion_position", status: "completed", startedAt: "2026-06-20T12:10:00.000Z", finalResponse: { text: "FINAL" } }, [ + { seq: 1, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "FINAL", createdAt: "2026-06-20T12:10:01.000Z" }, + { seq: 2, label: "item/commandExecution:completed", type: "tool_call", toolName: "commandExecution", status: "completed", command: "make test", createdAt: "2026-06-20T12:10:02.000Z" }, + { seq: 3, label: "agentrun:result:completed", type: "result", status: "completed", terminal: true, createdAt: "2026-06-20T12:10:03.000Z" } + ]); + const toolIndex = rows.findIndex((row) => row.seq === 2); + const finalIndex = rows.findIndex((row) => row.seq === 3 && row.terminal === true && row.body === "FINAL"); + assert.ok(toolIndex >= 0); + assert.ok(finalIndex > toolIndex); + assert.equal(rows.some((row) => row.seq === 1 && row.terminal === true), false); +}); + test("R1 trace API snapshots stay authoritative over compact result traces", () => { const previous = { traceId: "trc_merge",