diff --git a/internal/cloud/workbench-projection-writer.test.ts b/internal/cloud/workbench-projection-writer.test.ts index b34bcc84..f20d6c28 100644 --- a/internal/cloud/workbench-projection-writer.test.ts +++ b/internal/cloud/workbench-projection-writer.test.ts @@ -196,7 +196,7 @@ test("workbench projection writer keeps event writes durable when checkpoint rea assert.equal(factWrites[0].params.facts.checkpoints[0].projectionStatus, "projecting"); }); -test("workbench projection writer keeps terminal session facts durable and skips unsafe trace backfill when reads are blocked", async () => { +test("workbench projection writer keeps terminal session facts durable without trace backfill queries", async () => { const factWrites = []; let allocationSeq = 0; let queryCount = 0; @@ -261,7 +261,7 @@ test("workbench projection writer keeps terminal session facts durable and skips }); assert.equal(owner.id, "ses_writer_blocked_backfill"); - assert.equal(queryCount, 2); + 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"); @@ -269,7 +269,7 @@ test("workbench projection writer keeps terminal session facts durable and skips assert.equal(allocationSeq, 0); }); -test("workbench projection writer backfills terminal final and completion trace events from sealed facts", async () => { +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) { @@ -347,15 +347,14 @@ test("workbench projection writer backfills terminal final and completion trace } }); - const loaded = runtimeStore.queryWorkbenchFacts({ traceId: "trc_writer_terminal_backfill", families: ["traceEvents", "checkpoints"], limit: 20 }); + 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, 4, 5]); - assert.deepEqual(events.slice(-2).map((event) => event.label), ["agentrun:assistant:message", "agentrun:result:completed"]); - assert.equal(events.at(-2).text, "BACKFILL_DONE"); - assert.equal(events.at(-1).terminal, true); - assert.equal(events.at(-1).sealed, true); - assert.equal(loaded.facts.checkpoints[0].projectedSeq, 5); - assert.equal(loaded.facts.checkpoints[0].projectionStatus, "caught_up"); + 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 writer does not duplicate terminal trace events that already exist", async () => { @@ -462,6 +461,85 @@ test("workbench projection writer does not duplicate terminal trace events that assert.equal(loaded.facts.checkpoints[0].projectedSeq, 3); }); +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, + event: { + traceId, + sessionId, + source: "agentrun", + sourceSeq: 1, + type: "backend", + eventType: "backend", + status: "running", + label: "agentrun:backend:running", + createdAt: "2026-06-20T11:31:00.000Z" + } + }); + await writeWorkbenchProjectionEvent({ + runtimeStore, + event: { + traceId, + sessionId, + source: "agentrun", + sourceSeq: 25, + type: "assistant", + eventType: "assistant", + status: "running", + label: "agentrun:assistant:message", + text: "AGGREGATED_FINAL", + createdAt: "2026-06-20T11:31:10.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 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 = { @@ -662,7 +740,14 @@ test("workbench projection writer does not replace prior user prompt with cancel 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 } + 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 }); diff --git a/internal/cloud/workbench-projection-writer.ts b/internal/cloud/workbench-projection-writer.ts index cb10a5ec..452f8c12 100644 --- a/internal/cloud/workbench-projection-writer.ts +++ b/internal/cloud/workbench-projection-writer.ts @@ -121,6 +121,16 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event 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 + }); + 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); @@ -162,7 +172,7 @@ export async function writeWorkbenchProjectionEvent({ runtimeStore = null, event }); 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 eventId = isAssistantMessageTraceEvent(event) ? stableFactId("wte", { traceId, sourceEventId }) : textValue(event.id) || stableFactId("wte", { traceId, sourceEventId }); const previousFinalText = finalResponseTextValue(resolvedPreviousCheckpoint?.finalResponse, resolvedPreviousCheckpoint?.assistantText, resolvedPreviousCheckpoint?.finalText); const messageFinalText = terminal ? finalResponseTextValue(event.finalResponse, event.assistantText, event.reply, previousFinalText) : null; const userMessageFact = sessionId && !suppressedAfterSeal && resolvedPreviousCheckpoint?.userMessage ? normalizeMessageFact(resolvedPreviousCheckpoint.userMessage, 0, { traceId, sessionId, turnId, terminal: false, terminalStatus: "sent", finalText: null, timestamp: projectedAt, timing }) : null; @@ -335,16 +345,15 @@ async function writeWorkbenchProjectionFacts({ runtimeStore = null, traceStore = projectId, valuesPrinted: false }); - const terminalTraceBackfill = await backfillTerminalTraceEvents({ runtimeStore, traceStore, traceId: safeId, facts, payload, status }); emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "ok", { ...workbenchProjectionFactCountAttributes(facts), - "workbench.projection.terminal_trace_backfill_written": terminalTraceBackfill?.written ?? 0, + "workbench.projection.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 }); - return terminalTraceBackfill ? { ...result, terminalTraceBackfill } : result; + return result; } catch (error) { const safeId = safeTraceId(traceId); emitWorkbenchProjectionOtel("workbench.projection.facts.write", safeId, "error", { @@ -505,7 +514,7 @@ function buildWorkbenchProjectionFacts({ traceId = null, ownerUserId = null, own 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 }), { + const messages = normalizeMessages(workbenchProjectionInputMessages({ session, payload, params, traceId: safeId, timestamp, terminal, terminalStatus, finalText, previousCheckpoint }), { traceId: safeId, sessionId: resolvedSessionId, conversationId: resolvedConversationId, @@ -621,9 +630,9 @@ function workbenchProjectionFactCountAttributes(facts = {}) { }; } -function workbenchProjectionInputMessages({ session = {}, payload = {}, params = {}, traceId = null, timestamp = null, terminal = false, terminalStatus = "completed", finalText = null } = {}) { - if (Array.isArray(session?.messages) && session.messages.length > 0) return session.messages; - if (Array.isArray(payload?.messages) && payload.messages.length > 0) return payload.messages; +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); @@ -667,6 +676,35 @@ function workbenchProjectionUserMessage({ payload = {}, params = {}, traceId = n }; } +function preserveTerminalUserMessage(messages = [], { terminal = false, traceId = null, previousCheckpoint = null } = {}) { + if (!terminal || !traceId || !previousCheckpoint?.userMessage) return messages; + const previousUserMessage = previousCheckpoint.userMessage; + const previousText = textValue(previousUserMessage.text ?? previousUserMessage.content ?? previousUserMessage.message); + if (!previousText) return messages; + let replaced = 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; + replaced = true; + 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 + }; + }); + if (replaced) return preserved; + return [previousUserMessage, ...messages]; +} + 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)) { @@ -971,6 +1009,10 @@ function stableFactId(prefix, value) { } function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } = {}) { + if (isAssistantMessageTraceEvent(event)) { + const messageId = eventProjectionAssistantMessageId(traceId, event); + return `${safeTraceId(traceId) ?? traceId}:assistant-message:${messageId}`; + } const explicit = textValue(event.sourceEventId ?? event.id); if (explicit) return explicit; const source = textValue(event.source); @@ -990,6 +1032,12 @@ function workbenchSourceEventId(event = {}, { traceId, sourceSeq, occurredAt } = }); } +function isAssistantMessageTraceEvent(event = {}) { + const type = textValue(event?.eventType ?? event?.type).toLowerCase(); + const label = textValue(event?.label).toLowerCase(); + return type === "assistant" || /assistant:message|assistant_message/u.test(label); +} + 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])]));