From 4bad6ceeb5f0a12118654eae18ad1628914dd132 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Jul 2026 10:26:47 +0200 Subject: [PATCH] fix(workbench): consume assistant progress lifecycle --- internal/cloud/kafka-event-bridge.test.ts | 55 +++++ internal/cloud/kafka-event-bridge.ts | 51 ++++- tools/hwlab-cli/kafka-regenerate.test.ts | 208 +++++++++++++++++- tools/src/hwlab-cli/kafka-regenerate.ts | 2 +- tools/src/hwlab-cli/trace-renderer.ts | 112 +++++++--- .../stores/workbench-live-kafka-event.test.ts | 60 +++++ .../src/stores/workbench-live-kafka-event.ts | 12 +- 7 files changed, 454 insertions(+), 46 deletions(-) diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 8f2f66cf..2954a644 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -103,6 +103,61 @@ test("projects AgentRun assistant_message Kafka event into HWLAB trace event", ( assert.equal(projected.valuesPrinted, false); }); +test("projects AgentRun assistant_progress as an identity-preserving HWLAB assistant lifecycle event", () => { + const text = "p".repeat(500); + const projected = projectAgentRunKafkaEventToHwlabEvent({ + schema: "agentrun.event.v1", + eventType: "agentrun.event.committed", + eventId: "evt_assistant_progress", + sourceSeq: 136, + traceId: "trc_assistant_progress", + hwlabSessionId: "ses_assistant_progress", + runId: "run_assistant_progress", + commandId: "cmd_assistant_progress", + valuesPrinted: false, + run: { runId: "run_assistant_progress", sessionId: "ses_agentrun_progress", hwlabSessionId: "ses_assistant_progress", valuesPrinted: false }, + command: { commandId: "cmd_assistant_progress", runId: "run_assistant_progress", seq: 1, type: "turn", state: "running", payloadHash: "hash", valuesPrinted: false }, + event: { + id: "evt_assistant_progress", + runId: "run_assistant_progress", + seq: 136, + type: "assistant_progress", + createdAt: "2026-07-12T10:00:01.000Z", + payload: { + traceId: "trc_assistant_progress", + commandId: "cmd_assistant_progress", + text, + itemId: "msg_assistant_progress", + source: "agent-message-delta-progress", + progress: true, + progressFlush: false, + replyAuthority: false, + final: false, + textBytes: 500, + outputBytes: 500, + valuesPrinted: false + } + } + }, { source: "hwlab-test", sourceTopic: "agentrun.event.v1", sourceOffset: "136" }); + + assert.equal(projected.context.agentRunEventType, "assistant_progress"); + assert.equal(projected.event.agentRunEventType, "assistant_progress"); + assert.equal(projected.event.type, "assistant"); + assert.equal(projected.event.eventType, "assistant_progress"); + assert.equal(projected.event.label, "agentrun:assistant:progress"); + assert.equal(projected.event.itemId, "msg_assistant_progress"); + assert.equal(projected.event.sourceSeq, 136); + assert.equal(projected.event.assistantSource, "agent-message-delta-progress"); + assert.equal(projected.event.progress, true); + assert.equal(projected.event.progressFlush, false); + assert.equal(projected.event.replyAuthority, false); + assert.equal(projected.event.final, false); + assert.equal(projected.event.textBytes, 500); + assert.equal(projected.event.assistantText, text); + assert.equal(projected.event.finalResponse, null); + assert.equal(projected.event.terminal, false); +}); + test("projects AgentRun user_message into a stable HWLAB user event", () => { const projected = projectAgentRunKafkaEventToHwlabEvent({ schema: "agentrun.event.v1", diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index c38e27d3..31e85107 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -476,7 +476,7 @@ export function shouldEmitLiveKafkaOtelSpan(envelope = {}) { const context = objectValue(envelope.context); const eventType = firstText(context.agentRunEventType, event.eventType, event.type, envelope.eventType)?.toLowerCase(); if (event.terminal === true || eventType === "terminal" || eventType === "terminal_status") return true; - if (["assistant", "assistant_message", "tool", "tool_call"].includes(eventType)) return true; + if (["assistant", "assistant_progress", "assistant_message", "tool", "tool_call"].includes(eventType)) return true; if (eventType !== "command_output" && event.type !== "output") return true; const sourceSeq = integerValue(context.sourceSeq ?? event.sourceSeq); if (Number.isInteger(sourceSeq) && sourceSeq > 0) return sourceSeq % LIVE_KAFKA_COMMAND_OUTPUT_OTEL_SAMPLE_MODULUS === 0; @@ -1309,6 +1309,7 @@ export async function openKafkaEventStream({ env = process.env, stream = "hwlab" } function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, command, traceId, sessionId, sourceSeq, runId, commandId }) { + const type = firstText(sourceEvent.type, payload.type, input.eventType) || "event"; const base = { traceId, sessionId, @@ -1319,10 +1320,10 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, attemptId: firstText(command.attemptId, sourceEvent.attemptId, payload.attemptId), runnerId: firstText(command.runnerId, sourceEvent.runnerId, payload.runnerId), backend: firstText(run.backendProfile, payload.backendProfile, payload.providerProfile), + agentRunEventType: type, createdAt: timestampValue(sourceEvent.createdAt ?? input.producedAt), valuesPrinted: false }; - const type = firstText(sourceEvent.type, payload.type, input.eventType) || "event"; if (type === "user_message") { const userMessageId = firstText(payload.userMessageId); const userText = textPayload(payload, "user message"); @@ -1359,16 +1360,60 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, assistantSource: firstText(payload.source), messageIndex: integerValue(payload.messageIndex), messageCount: integerValue(payload.messageCount), + progress: payload.progress === true, + progressFlush: payload.progressFlush === true, + durableText: payload.durableText === true, replyAuthority: payload.replyAuthority === true, final: payload.final === true, + textBytes: integerValue(payload.textBytes), textTruncated: payload.textTruncated === true, + outputBytes: integerValue(payload.outputBytes), + outputTruncated: payload.outputTruncated === true, + terminal: false + }; + } + if (type === "assistant_progress") { + const assistantText = textPayload(payload, "assistant progress"); + return { + ...base, + type: "assistant", + eventType: "assistant_progress", + status: "running", + label: "agentrun:assistant:progress", + message: assistantText, + text: assistantText, + assistantText, + finalResponse: null, + itemId: firstText(payload.itemId), + assistantSource: firstText(payload.source), + messageIndex: integerValue(payload.messageIndex), + messageCount: integerValue(payload.messageCount), + progress: payload.progress === true, + progressFlush: payload.progressFlush === true, + durableText: false, + replyAuthority: false, + final: false, + textBytes: integerValue(payload.textBytes), + textTruncated: payload.textTruncated === true, + outputBytes: integerValue(payload.outputBytes), outputTruncated: payload.outputTruncated === true, terminal: false }; } if (type === "backend_status") { const phase = firstText(payload.phase) || "status"; - return { ...base, type: "backend", eventType: "backend", status: "running", label: `agentrun:backend:${phase}`, message: textPayload(payload, phase) }; + return { + ...base, + type: "backend", + eventType: "backend", + status: "running", + label: `agentrun:backend:${phase}`, + message: textPayload(payload, phase), + itemId: firstText(payload.itemId), + finalSeal: payload.finalSeal === true, + replyRef: payload.replyRef && typeof payload.replyRef === "object" && !Array.isArray(payload.replyRef) ? payload.replyRef : null, + terminal: false + }; } if (type === "terminal_status") { const terminalStatus = firstText(payload.terminalStatus, sourceEvent.terminalStatus, payload.status, sourceEvent.status) || "failed"; diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index 5d574fea..a8f22469 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -6,7 +6,7 @@ import path from "node:path"; import { test } from "bun:test"; import { projectAgentRunKafkaEventToHwlabEvent, queryKafkaEventStream } from "../../internal/cloud/kafka-event-bridge.ts"; -import { mapAgentRunRecordsToHwlabDebugEvents, renderKafkaCliText, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts"; +import { mapAgentRunRecordsToHwlabDebugEvents, projectHwlabRecordsThroughWorkbench, renderKafkaCliText, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts"; import { renderTraceRowsMarkdown, traceDisplayRows, traceToolSummary } from "../src/hwlab-cli/trace-renderer.ts"; const SESSION_ID = "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105"; @@ -252,6 +252,198 @@ test("shared row model removes late assistant progress after the same item compl assert.equal((result.markdownOutput?.match(new RegExp(duplicateText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1); }); +test("fixed Kafka assistant lifecycle replays 500/1003 byte progress into one 1190 byte final", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-assistant-progress-contract-")); + const inputFile = path.join(cwd, "hwlab-events.jsonl"); + const itemId = "msg_0d2ec_assistant_lifecycle"; + const finalText = "x".repeat(1190); + const progress500 = finalText.slice(0, 500); + const progress1003 = finalText.slice(0, 1003); + assert.deepEqual([Buffer.byteLength(progress500), Buffer.byteLength(progress1003), Buffer.byteLength(finalText)], [500, 1003, 1190]); + + const agentrunRecords = [ + agentrunAssistantLifecycleRecord(136, 1, "assistant_progress", { + text: progress500, + itemId, + source: "agent-message-delta-progress", + progress: true, + progressFlush: false, + replyAuthority: false, + final: false, + textBytes: 500, + outputBytes: 500, + valuesPrinted: false + }), + agentrunAssistantLifecycleRecord(137, 2, "assistant_progress", { + text: progress1003, + itemId, + source: "agent-message-delta-progress", + progress: true, + progressFlush: false, + replyAuthority: false, + final: false, + textBytes: 1003, + outputBytes: 1003, + valuesPrinted: false + }), + agentrunAssistantLifecycleRecord(138, 3, "assistant_message", { + text: finalText, + itemId, + source: "completed-agent-message", + durableText: true, + replyAuthority: false, + final: false, + textBytes: 1190, + outputBytes: 1190, + valuesPrinted: false + }), + agentrunAssistantLifecycleRecord(142, 4, "backend_status", { + phase: "assistant-message-final-sealed", + itemId, + source: "assistant-message-final-seal", + finalSeal: true, + replyAuthority: true, + final: true, + replyRef: { + eventType: "assistant_message", + itemId, + source: "completed-agent-message", + textSha256: "a".repeat(64) + }, + valuesPrinted: false + }), + agentrunAssistantLifecycleRecord(143, 5, "terminal_status", { + terminalStatus: "completed", + valuesPrinted: false + }) + ]; + const hwlabRecords = agentrunRecords.map((record, index) => hwlabOrderRecord(record, index)); + const hwlabEvents = hwlabRecords.map((record) => record.value.event); + + assert.deepEqual(hwlabRecords.map((record) => record.value.context.agentRunEventType), [ + "assistant_progress", + "assistant_progress", + "assistant_message", + "backend_status", + "terminal_status" + ]); + assert.deepEqual(hwlabEvents.slice(0, 3).map((event) => [event.itemId, event.sourceSeq, event.progress, event.assistantText.length]), [ + [itemId, 136, true, 500], + [itemId, 137, true, 1003], + [itemId, 138, false, 1190] + ]); + assert.equal(hwlabEvents[3]?.finalSeal, true); + assert.equal(hwlabEvents[3]?.message, "assistant-message-final-sealed"); + assert.equal(hwlabEvents[3]?.assistantText, undefined, "seal must not copy the completed body"); + assert.equal(hwlabEvents[4]?.assistantText, undefined, "terminal must not copy the completed body"); + + const runningRows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, hwlabEvents.slice(0, 2)); + assert.equal(runningRows.filter((row) => /助手/u.test(row.header)).length, 1); + assert.equal(runningRows.find((row) => /助手/u.test(row.header))?.seq, 137); + assert.equal(runningRows.find((row) => /助手/u.test(row.header))?.body, progress1003); + const completedRows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, hwlabEvents.slice(0, 3)); + assert.equal(completedRows.filter((row) => /助手/u.test(row.header)).length, 1); + assert.equal(completedRows.find((row) => /助手/u.test(row.header))?.seq, 138); + assert.equal(completedRows.find((row) => /助手/u.test(row.header))?.body, finalText); + + const projected = projectHwlabRecordsThroughWorkbench(hwlabRecords, { + traceId: TRACE_ID, + sessionId: HWLAB_SESSION_ID, + observedAt: "2026-07-12T10:01:00.000Z" + }); + assert.equal(projected.appliedCount, 5); + assert.equal(projected.sourceEvents.length, 5); + assert.equal(projected.message?.runnerTrace?.events?.length, 5, "raw/applied observation must retain every unfiltered event"); + assert.equal(projected.message?.text, finalText); + assert.equal(projected.message?.status, "completed"); + assert.equal((projected.message?.finalResponse as { text?: string } | null)?.text, finalText); + + await writeFile(inputFile, hwlabRecords.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + const result = await runKafkaCli([ + "render", "trace", + "--from", "jsonl", + "--trace-id", TRACE_ID, + "--session-id", HWLAB_SESSION_ID, + "--jsonl-file", inputFile, + "--format", "markdown" + ], { cwd, env: {}, now: () => "2026-07-12T10:02:00.000Z" }); + + assert.equal(result.exitCode, 0); + assert.equal(result.payload.projection.appliedCount, 5); + assert.equal(result.payload.projection.eventCount, 5); + assert.equal(result.payload.render.assistantRowCount, 0, "the authoritative outer final replaces the progress lifecycle"); + assert.equal(result.payload.render.finalResponsePresent, true); + assert.equal(result.payload.render.finalResponseInTrace, false); + assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 0, tool: 0, terminal: 1, final: 1 }); + assert.equal(result.payload.replayLineage.layers.source.assistant, 3, "raw source lineage must keep both progress snapshots and the durable completion"); + assert.equal(result.payload.replayLineage.layers.applied.assistant, 3); + assert.equal(result.payload.validation.allMatchedEventsApplied, true); + assert.equal(result.payload.validation.finalResponseOutsideTrace, true); + assert.equal(result.markdownOutput?.split(finalText).length, 2, "the final body must render exactly once"); + const traceMarkdown = (result.markdownOutput ?? "").split("## 运行记录")[1]?.split("## 最终回复")[0] ?? ""; + assert.doesNotMatch(traceMarkdown, /x{500}/u, "progress must not remain as a separate rendered row"); +}); + +test("HWLAB compatibility window coalesces legacy assistant_message progress before AgentRun assistant_progress rollout", () => { + const itemId = "msg_legacy_progress_contract"; + const finalText = "legacy durable body"; + const agentrunRecords = [ + agentrunAssistantLifecycleRecord(201, 1, "assistant_message", { + text: "legacy progress 1", + itemId, + source: "agent-message-delta-progress", + progress: true, + progressFlush: false, + replyAuthority: false, + final: false + }), + agentrunAssistantLifecycleRecord(202, 2, "assistant_message", { + text: "legacy progress 2", + itemId, + source: "agent-message-delta-progress", + progress: true, + progressFlush: true, + replyAuthority: false, + final: false + }), + agentrunAssistantLifecycleRecord(203, 3, "assistant_message", { + text: finalText, + itemId, + source: "completed-agent-message", + durableText: true, + replyAuthority: false, + final: false + }), + agentrunAssistantLifecycleRecord(204, 4, "terminal_status", { terminalStatus: "completed" }) + ]; + const hwlabRecords = agentrunRecords.map((record, index) => hwlabOrderRecord(record, index)); + const events = hwlabRecords.map((record) => record.value.event); + + assert.deepEqual(events.slice(0, 2).map((event) => [event.agentRunEventType, event.eventType, event.progress]), [ + ["assistant_message", "assistant", true], + ["assistant_message", "assistant", true] + ]); + const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, events.slice(0, 3)); + const assistantRows = rows.filter((row) => /助手/u.test(row.header)); + assert.equal(assistantRows.length, 1); + assert.equal(assistantRows[0]?.seq, 203); + assert.equal(assistantRows[0]?.body, finalText); + + const projected = projectHwlabRecordsThroughWorkbench(hwlabRecords, { + traceId: TRACE_ID, + sessionId: HWLAB_SESSION_ID, + observedAt: "2026-07-12T10:03:00.000Z" + }); + assert.equal(projected.message?.text, finalText); + assert.equal(projected.message?.runnerTrace?.events?.length, 4); + assert.equal((projected.message?.finalResponse as { text?: string } | null)?.text, finalText); + const finalRows = traceDisplayRows(projected.message?.runnerTrace ?? {}, projected.message?.runnerTrace?.events ?? [], { + finalResponsePlacement: "outer", + outerFinalResponseText: finalText + }); + assert.equal(finalRows.filter((row) => /助手/u.test(row.header)).length, 0); +}); + test("shared tool summary model keeps status semantic and visible content status-free", () => { const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ { @@ -325,7 +517,7 @@ test("shared tool summary model keeps status semantic and visible content status assert.match(markdown, /aria-label="工具调用失败：/u); }); -test("outer final response stays out of the Trace fallback when no completion event exists", () => { +test("outer final response suppression follows formal assistant identity instead of equal text", () => { const finalText = "Hi. What do you want to work on?"; const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ { @@ -360,7 +552,9 @@ test("outer final response stays out of the Trace fallback when no completion ev outerFinalResponseText: finalText }); - assert.deepEqual(rows, []); + assert.equal(rows.length, 1); + assert.equal(rows[0]?.rowId, "event:evt_same_text"); + assert.equal(rows[0]?.body, finalText, "a distinct itemId with equal text must remain visible"); }); test("trace JSON defaults to a bounded tail while the Markdown artifact stays complete", async () => { @@ -1504,6 +1698,14 @@ function agentrunOrderRecord(seq: number, traceId: string, type: string, payload return record; } +function agentrunAssistantLifecycleRecord(seq: number, ordinal: number, type: string, payload: Record): any { + const record = agentrunOrderRecord(seq, TRACE_ID, type, payload); + const createdAt = `2026-07-12T10:00:0${ordinal}.000Z`; + record.value.committedAt = createdAt; + record.value.event.createdAt = createdAt; + return record; +} + function hwlabOrderRecord(source: any, index: number): any { const valueText = JSON.stringify(source.value); const projected = projectAgentRunKafkaEventToHwlabEvent(source.value, { diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts index fcc83339..3ddd6378 100644 --- a/tools/src/hwlab-cli/kafka-regenerate.ts +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -1502,7 +1502,7 @@ function traceLayerEventKind(event: JsonRecord): "user" | "assistant" | "tool" | const type = text(event.type ?? event.eventType); const label = text(event.label); if (type === "user") return "user"; - if (type === "assistant" || type === "assistant_message" || label === "agentrun:assistant:message" || label === "assistant:message") return "assistant"; + if (type === "assistant" || type === "assistant_progress" || type === "assistant_message" || label === "agentrun:assistant:progress" || label === "agentrun:assistant:message" || label === "assistant:message") return "assistant"; if (type === "tool" || type === "tool_call" || text(event.toolName) === "commandExecution" || /commandExecution|^agentrun:tool:/u.test(label)) return "tool"; return "other"; } diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index e161fa63..7241144b 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -14,7 +14,7 @@ export interface TraceEventRow { export type TraceToolState = "running" | "success" | "failure" | "neutral"; -type TraceEvent = Record; +export type TraceEvent = Record; type TraceClockFormatter = (value: string) => string; export type TraceSequenceAuthority = "projected" | "source"; export type TraceFinalResponsePlacement = "trace" | "outer"; @@ -42,12 +42,9 @@ export function traceDisplayRows(trace: Record = {}, events: Tr ); const effectiveTrace = traceWithInferredStart(trace, orderedEvents); const finalResponseText = nonEmptyString(options.outerFinalResponseText) ?? traceFinalResponseText(effectiveTrace); - const finalResponseFingerprint = traceMessageFingerprint(finalResponseText); - const terminalAssistantFingerprints = new Set(orderedEvents - .filter(isTerminalAssistantTraceEvent) - .map((event) => traceAssistantEventFingerprint(effectiveTrace, event)) - .filter((value): value is string => Boolean(value))); - if (finalResponseFingerprint) terminalAssistantFingerprints.add(finalResponseFingerprint); + const outerFinalAssistantEvent = displayOptions.finalResponsePlacement === "outer" && finalResponseText + ? traceOuterFinalAssistantEvent(orderedEvents) + : null; const hasCompletionEvent = orderedEvents.some(isCompletionTraceEvent); const rows: TraceEventRow[] = []; let completionEvent: TraceEvent | null = null; @@ -78,7 +75,7 @@ export function traceDisplayRows(trace: Record = {}, events: Tr continue; } if (isTerminalAssistantTraceEvent(event)) { - if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue; + if (isOuterFinalResponseTraceEvent(event, displayOptions, outerFinalAssistantEvent)) continue; if (hasCompletionEvent) continue; if (finalResponseText) { rows.push(traceFinalResponseRow(event, finalResponseText, displayOptions)); @@ -88,9 +85,7 @@ export function traceDisplayRows(trace: Record = {}, events: Tr continue; } if (isAssistantTraceEvent(event)) { - const fingerprint = traceAssistantEventFingerprint(effectiveTrace, event); - if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue; - if (finalResponseFingerprint && fingerprint === finalResponseFingerprint) continue; + if (isOuterFinalResponseTraceEvent(event, displayOptions, outerFinalAssistantEvent)) continue; rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, displayOptions)); continue; } @@ -108,21 +103,47 @@ export function traceDisplayRows(trace: Record = {}, events: Tr return orderedEvents .filter((event, index) => !isSupersededTraceItemLifecycleEvent(event, orderedEvents, index)) .filter((event) => !isSuppressedTraceEvent(event)) - .filter((event) => !isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) + .filter((event) => !isOuterFinalResponseTraceEvent(event, displayOptions, outerFinalAssistantEvent)) .map((event) => traceDisplayRow(event, displayOptions)); } function isOuterFinalResponseTraceEvent( - trace: Record, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions, - terminalAssistantFingerprints: Set + outerFinalAssistantEvent: TraceEvent | null ): boolean { if (options.finalResponsePlacement !== "outer") return false; - if (isTerminalAssistantTraceEvent(event)) return true; - if (!isAssistantTraceEvent(event)) return false; - const fingerprint = traceAssistantEventFingerprint(trace, event); - return Boolean(fingerprint && terminalAssistantFingerprints.has(fingerprint)); + return event === outerFinalAssistantEvent; +} + +function traceOuterFinalAssistantEvent(events: TraceEvent[]): TraceEvent | null { + const visible = events + .map((event, index) => ({ event, index })) + .filter(({ event, index }) => isAssistantTraceEvent(event) && !traceAssistantLifecycleEventIsSuperseded(event, events, index)); + if (visible.length === 0) return null; + const authoritative = [...visible].reverse().find(({ event }) => event.replyAuthority === true || event.final === true); + if (authoritative) return authoritative.event; + const seal = [...events].reverse().find((event) => event.finalSeal === true || nonEmptyString(event.label) === "agentrun:backend:assistant-message-final-sealed"); + if (seal) { + const replyRef = isRecord(seal.replyRef) ? seal.replyRef : null; + const itemId = nonEmptyString(replyRef?.itemId ?? seal.itemId); + if (itemId) { + const matched = [...visible].reverse().find(({ event }) => nonEmptyString(event.itemId) === itemId); + if (matched) return matched.event; + } + const sealIndex = events.indexOf(seal); + const preceding = [...visible].reverse().find(({ index }) => index < sealIndex); + if (preceding) return preceding.event; + } + let terminalIndex = -1; + for (let index = events.length - 1; index >= 0; index -= 1) { + const event = events[index]; + if (!event || !(isCompletionTraceEvent(event) || event.terminal === true || event.eventType === "terminal" || event.type === "result")) continue; + terminalIndex = index; + break; + } + const precedingTerminal = terminalIndex >= 0 ? [...visible].reverse().find(({ index }) => index < terminalIndex) : null; + return precedingTerminal?.event ?? visible.at(-1)?.event ?? null; } function traceSequenceAuthority(trace: Record): TraceSequenceAuthority { @@ -344,15 +365,6 @@ function traceAssistantEventText(trace: Record, event: TraceEve return cleanTraceDetailText(event.message ?? event.outputSummary ?? assistantStreamText(trace, event) ?? ""); } -function traceAssistantEventFingerprint(trace: Record, event: TraceEvent): string | null { - return traceMessageFingerprint(traceAssistantEventText(trace, event)); -} - -function traceMessageFingerprint(value: unknown): string | null { - const text = cleanTraceDetailText(value ?? "").replace(/\s+/gu, " ").trim(); - return text || null; -} - function traceFinalResponseRow(event: TraceEvent, finalText: string, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { @@ -452,7 +464,13 @@ function isCompletionTraceEvent(event: TraceEvent): boolean { function isAssistantTraceEvent(event: TraceEvent): boolean { const label = String(event.label ?? ""); - return label === "agentrun:assistant:message" || label === "assistant:message" || event.type === "assistant_message" || event.type === "assistant"; + return label === "agentrun:assistant:message" + || label === "agentrun:assistant:progress" + || label === "assistant:message" + || event.type === "assistant_progress" + || event.eventType === "assistant_progress" + || event.type === "assistant_message" + || event.type === "assistant"; } function isTerminalAssistantTraceEvent(event: TraceEvent): boolean { @@ -498,7 +516,7 @@ function traceSourceEventKey(event: TraceEvent | null | undefined): string | nul } function isSupersededTraceItemLifecycleEvent(event: TraceEvent, events: TraceEvent[], index: number): boolean { - return isSupersededToolStart(event, events, index) || isLateAssistantProgressAfterCompletion(event, events, index); + return isSupersededToolStart(event, events, index) || traceAssistantLifecycleEventIsSuperseded(event, events, index); } function isSupersededToolStart(event: TraceEvent, events: TraceEvent[], index: number): boolean { @@ -508,14 +526,38 @@ function isSupersededToolStart(event: TraceEvent, events: TraceEvent[], index: n return events.slice(index + 1).some((candidate) => candidate && isToolTraceEvent(candidate) && isToolCompleteTraceEvent(candidate) && toolIdentity(candidate) === identity); } -function isLateAssistantProgressAfterCompletion(event: TraceEvent, events: TraceEvent[], index: number): boolean { - if (!isAssistantTraceEvent(event) || nonEmptyString(event.assistantSource) !== "agent-message-delta-progress") return false; +export function traceAssistantLifecycleEventIsSuperseded(event: TraceEvent, events: TraceEvent[], index: number): boolean { + if (!isAssistantTraceEvent(event)) return false; const identity = nonEmptyString(event.itemId); if (!identity) return false; - return events.slice(0, index).some((candidate) => candidate - && isAssistantTraceEvent(candidate) - && nonEmptyString(candidate.itemId) === identity - && nonEmptyString(candidate.assistantSource) === "completed-agent-message"); + const rank = traceAssistantLifecycleRank(event); + const sequence = traceAssistantLifecycleSequence(event); + return events.some((candidate, candidateIndex) => { + if (candidateIndex === index || !candidate || !isAssistantTraceEvent(candidate) || nonEmptyString(candidate.itemId) !== identity) return false; + const candidateRank = traceAssistantLifecycleRank(candidate); + if (candidateRank !== rank) return candidateRank > rank; + const candidateSequence = traceAssistantLifecycleSequence(candidate); + if (sequence !== null && candidateSequence !== null && candidateSequence !== sequence) return candidateSequence > sequence; + return candidateIndex > index; + }); +} + +function traceAssistantLifecycleRank(event: TraceEvent): number { + if (event.replyAuthority === true || event.final === true) return 3; + if (traceAssistantLifecycleProgress(event)) return 1; + return 2; +} + +function traceAssistantLifecycleProgress(event: TraceEvent): boolean { + return event.progress === true + || event.type === "assistant_progress" + || event.eventType === "assistant_progress" + || nonEmptyString(event.label) === "agentrun:assistant:progress" + || nonEmptyString(event.assistantSource) === "agent-message-delta-progress"; +} + +function traceAssistantLifecycleSequence(event: TraceEvent): number | null { + return numberOrNull(event.sourceSeq ?? event.projectedSeq ?? event.seq); } function isNoisyTraceEvent(event: TraceEvent): boolean { diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts index f3a2e022..5f8b8c6a 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts @@ -219,6 +219,66 @@ test("live Kafka projection refreshes lastEventAt from each ingress receipt", () assert.equal(second.runnerTrace?.eventCount, 2); }); +test("assistant item completion stays authoritative over a later progress event while raw events remain observable", () => { + const traceId = "trc_live_assistant_lifecycle"; + const sessionId = "ses_live_assistant_lifecycle"; + const itemId = "msg_live_assistant_lifecycle"; + const progress = projectWorkbenchLiveKafkaMessage({ + previous: null, + traceId, + sessionId, + receivedAt: "2026-07-12T10:00:01.000Z", + event: traceEvent(136, "assistant", { + eventType: "assistant_progress", + itemId, + assistantSource: "agent-message-delta-progress", + progress: true, + assistantText: "partial body" + }) + }); + const completed = projectWorkbenchLiveKafkaMessage({ + previous: progress, + traceId, + sessionId, + receivedAt: "2026-07-12T10:00:02.000Z", + event: traceEvent(138, "assistant", { + eventType: "assistant", + itemId, + assistantSource: "completed-agent-message", + durableText: true, + assistantText: "durable body" + }) + }); + const lateProgress = projectWorkbenchLiveKafkaMessage({ + previous: completed, + traceId, + sessionId, + receivedAt: "2026-07-12T10:00:03.000Z", + event: traceEvent(139, "assistant", { + eventType: "assistant_progress", + itemId, + assistantSource: "agent-message-delta-progress", + progress: true, + assistantText: "stale late progress" + }) + }); + const terminal = projectWorkbenchLiveKafkaMessage({ + previous: lateProgress, + traceId, + sessionId, + receivedAt: "2026-07-12T10:00:04.000Z", + event: traceEvent(143, "result", { eventType: "terminal", terminal: true, status: "completed" }) + }); + + assert.equal(progress.text, "partial body"); + assert.equal(completed.text, "durable body"); + assert.equal(lateProgress.text, "durable body"); + assert.equal(lateProgress.runnerTrace?.events?.length, 3, "visibility keeps the superseded progress source event"); + assert.equal(terminal.text, "durable body"); + assert.equal((terminal.finalResponse as { text?: string } | null)?.text, "durable body"); + assert.equal(terminal.runnerTrace?.events?.length, 4); +}); + test("live Kafka projection keeps terminal lifecycle while accepting a later event", () => { const traceId = "trc_live_reopened"; const sessionId = "ses_live_reopened"; diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts index 679ecd43..abee3c58 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts @@ -4,6 +4,7 @@ import { mergeRunnerTrace } from "../composables/workbench-trace-snapshot"; import type { ChatMessage, TraceEvent, WorkbenchTurnTimingProjection } from "../types"; import { firstNonEmptyString } from "../utils"; +import { traceAssistantLifecycleEventIsSuperseded } from "../../../../tools/src/hwlab-cli/trace-renderer.ts"; export interface WorkbenchLiveMessageState { text: string; @@ -28,11 +29,14 @@ export interface WorkbenchLiveKafkaUserProjectionInput { receivedAt: string; } -export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent): WorkbenchLiveMessageState { +export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent, previousEvents: TraceEvent[] = []): WorkbenchLiveMessageState { if (previous.terminal) return previous; const terminal = workbenchLiveKafkaEventIsTerminal(event); const status = terminal ? firstNonEmptyString(event.status, "completed") ?? "completed" : "running"; - const assistantText = workbenchLiveKafkaAssistantText(event); + const lifecycleEvents = [...previousEvents, event]; + const assistantText = traceAssistantLifecycleEventIsSuperseded(event, lifecycleEvents, lifecycleEvents.length - 1) + ? null + : workbenchLiveKafkaAssistantText(event); return { text: assistantText ?? (terminal ? workbenchLiveKafkaTerminalText(event, previous.text, status) : previous.text), status, @@ -98,7 +102,7 @@ export function workbenchLiveKafkaEventIsTerminal(event: TraceEvent | null | und } export function workbenchLiveKafkaAssistantText(event: TraceEvent | null | undefined): string | null { - if (!event || firstNonEmptyString(event.type, event.eventType) !== "assistant") return null; + if (!event || !["assistant", "assistant_progress"].includes(firstNonEmptyString(event.type, event.eventType) ?? "")) return null; const finalResponse = recordValue(event.finalResponse); return firstNonEmptyString(finalResponse?.text, event.assistantText, event.text, event.message) ?? null; } @@ -121,7 +125,7 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec text: previous?.text ?? "", status: previous?.status ?? "running", terminal: previous?.traceAutoLifecycle === "terminal" - }, input.event); + }, input.event, Array.isArray(previous?.runnerTrace?.events) ? previous.runnerTrace.events : []); const startedAt = timestamp(previous?.timing?.startedAt) ?? timestamp(previous?.startedAt) ?? timestamp(input.event.createdAt)