diff --git a/docs/reference/kafka-source-direct-debug.md b/docs/reference/kafka-source-direct-debug.md index a7619c62..3ed316d8 100644 --- a/docs/reference/kafka-source-direct-debug.md +++ b/docs/reference/kafka-source-direct-debug.md @@ -13,6 +13,9 @@ - 事件使用 `reduceWorkbenchRealtimeEvent` 与 `planWorkbenchRealtimeApply` 分类； - 卡片使用 `projectWorkbenchLiveKafkaMessage` 增量归约； - Trace 行使用 `traceDisplayRows` 生成同一 row model； + - assistant、tool 与 terminal 使用同一稳定 item lifecycle 收敛可见行； + - 同一 `itemId` 已出现 `completed-agent-message` 后到达的 `agent-message-delta-progress` 不再生成第二条 assistant row； + - lifecycle 收敛只影响可见行，不删除 source/applied 事件，也不使用正文、时间戳或 DOM 去重； - Web 最后渲染 HTML，CLI 最后使用 `renderTraceRowsMarkdown` 渲染 Markdown； - Final Response 只放在 Trace 外层，Trace 内保留非最终的助手进展消息。 @@ -45,7 +48,10 @@ - 默认 JSON 只返回 `--row-limit` 指定的尾部行，并披露 returned/omitted/window； - `--full` 仅用于显式下钻完整 row payload； - `--format markdown` 输出 Markdown 正文； - - `--json` 输出身份、投影、row model、终态、计数和 artifact SHA 的结构化证据。 + - `--json` 输出身份、投影、row model、终态、计数和 artifact SHA 的结构化证据； + - `replayLineage` 分别披露 source、applied、render 与 outer final 层的计数； + - 助手重复只披露正文 SHA-256、首次 identity、`sourceSeq` 和 Kafka transport，不回显正文； + - source/applied 重复保持可见而 render 重复归零时，表示稳定 item lifecycle 已在共享 row model 收敛。 - `hwlab-cli kafka regenerate hwlab` 用于把指定 AgentRun session 的事件直接映射为 HWLAB debug 事件： - 调用生产路径同源的 AgentRun 解码与 HWLAB 事件映射函数； diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index fb63e466..3da91498 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -68,6 +68,19 @@ test("offline HWLAB JSONL uses the shared Web pipeline and renders final respons assert.equal(result.payload.render.assistantRowCount, 1); assert.equal(result.payload.render.finalResponsePresent, true); assert.equal(result.payload.render.finalResponseInTrace, false); + assert.deepEqual(result.payload.replayLineage.counts, { source: 7, applied: 7, assistant: 1, tool: 1, terminal: 1, final: 1 }); + assert.deepEqual(result.payload.replayLineage.layers.source, { + total: 7, + user: 1, + assistant: 3, + tool: 2, + terminal: 1, + final: 1, + other: 1, + assistantDuplicate: { fingerprintCount: 0, occurrenceCount: 0, first: null, valuesPrinted: false } + }); + assert.equal(result.payload.replayLineage.layers.final.duplicatedInRender, false); + assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, null); assert.equal(result.payload.render.rowsReturned, result.payload.render.rowCount); assert.equal(result.payload.render.rowsOmitted, 0); assert.equal(result.payload.validation.allMatchedEventsApplied, true); @@ -86,6 +99,69 @@ test("offline HWLAB JSONL uses the shared Web pipeline and renders final respons assert.equal(result.markdownOutput?.trim(), markdown.trim()); }); +test("trace lineage reports the first non-final assistant content duplicate without printing message text", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-duplicate-lineage-")); + const duplicateText = "I am checking the workspace."; + const records = [ + hwlabRecord(1, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_first", message: duplicateText, assistantText: duplicateText }), + hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_between", command: "rg target", exitCode: 0 }), + hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_repeated", message: duplicateText, assistantText: duplicateText }), + hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: "Done.", assistantText: "Done.", replyAuthority: true, final: true, finalResponse: { text: "Done." } }), + hwlabRecord(5, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true }) + ]; + const inputFile = path.join(cwd, "hwlab-events.jsonl"); + await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + + const result = await runKafkaCli([ + "render", "trace", + "--from", "jsonl", + "--trace-id", TRACE_ID, + "--jsonl-file", inputFile, + "--format", "markdown" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + + assert.equal(result.exitCode, 0); + assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 2, tool: 1, terminal: 1, final: 1 }); + assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, "source"); + assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.occurrenceCount, 1); + assert.equal(result.payload.replayLineage.layers.applied.assistantDuplicate.occurrenceCount, 1); + assert.equal(result.payload.replayLineage.layers.render.assistantDuplicate.occurrenceCount, 1); + assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.first.first.identity, "evt_render_1"); + assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.first.duplicate.identity, "evt_render_3"); + assert.doesNotMatch(JSON.stringify(result.payload.replayLineage), new RegExp(duplicateText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "u")); + assert.match(renderKafkaCliText(result.payload), /assistant=2 tool=1 terminalRows=1 final=1 duplicateLayer=source/u); +}); + +test("shared row model removes late assistant progress after the same item completed", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-assistant-item-lifecycle-")); + const inputFile = path.join(cwd, "hwlab-events.jsonl"); + const duplicateText = "The completed assistant item must stay visible once."; + const records = [ + hwlabRecord(1, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_item", assistantSource: "completed-agent-message", message: duplicateText, assistantText: duplicateText }), + hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_between", command: "rg target", exitCode: 0 }), + hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_item", assistantSource: "agent-message-delta-progress", message: duplicateText, assistantText: duplicateText }), + hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", assistantSource: "completed-agent-message-final", message: "Done.", assistantText: "Done.", replyAuthority: true, final: true, finalResponse: { text: "Done." } }), + hwlabRecord(5, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true }) + ]; + await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + + const result = await runKafkaCli([ + "render", "trace", + "--from", "jsonl", + "--trace-id", TRACE_ID, + "--jsonl-file", inputFile, + "--format", "markdown" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + + assert.equal(result.exitCode, 0); + assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 1, tool: 1, terminal: 1, final: 1 }); + assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.occurrenceCount, 1); + assert.equal(result.payload.replayLineage.layers.applied.assistantDuplicate.occurrenceCount, 1); + assert.equal(result.payload.replayLineage.layers.render.assistantDuplicate.occurrenceCount, 0); + assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, "source"); + assert.equal((result.markdownOutput?.match(new RegExp(duplicateText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1); +}); + test("outer final response stays out of the Trace fallback when no completion event exists", () => { const finalText = "Hi. What do you want to work on?"; const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts index 49f54997..d10ed36c 100644 --- a/tools/src/hwlab-cli/kafka-regenerate.ts +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -18,12 +18,12 @@ import { import { createWorkbenchKafkaRefreshHandoff } from "../../../internal/cloud/workbench-kafka-refresh-handoff.ts"; import { decodeWorkbenchRealtimeEventFrame } from "../../../web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts"; import { reduceWorkbenchRealtimeEvent } from "../../../web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts"; -import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchLiveKafkaProjectionTarget } from "../../../web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts"; +import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchLiveKafkaAssistantText, workbenchLiveKafkaProjectionTarget } from "../../../web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts"; import { planWorkbenchRealtimeApply } from "../../../web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts"; import { renderTraceRowsMarkdown, traceDisplayRows } from "./trace-renderer.ts"; const CLI_NAME = "hwlab-cli"; -const VERSION = "0.3.4-kafka-partial-scan-terminal-monotonic"; +const VERSION = "0.3.5-kafka-trace-layer-duplicates"; const DEFAULT_LIMIT = 500; const DEFAULT_TIMEOUT_MS = 5000; const DEBUG_OUTPUT_PARTITION = 0; @@ -541,6 +541,13 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re const finalResponseInTrace = Boolean(finalResponseFingerprint && rows.some((row) => /助手/u.test(row.header) && traceTextFingerprint(row.body) === finalResponseFingerprint )); + const replayLineage = traceReplayLineageEvidence({ + sourceEvents: projection.sourceEvents, + appliedEvents: events, + userMessages: projection.userMessages, + rows, + finalResponseText + }); return { markdown, payload: { @@ -634,6 +641,7 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re rows: rowPayload, valuesPrinted: false }, + replayLineage, output: { artifact: { path: artifactPath, byteCount: Buffer.byteLength(`${markdown.trimEnd()}\n`, "utf8"), sha256: sha256(`${markdown.trimEnd()}\n`) }, valuesPrinted: false @@ -747,6 +755,7 @@ function kafkaTransportKey(value: Record): string { export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], options: { traceId: string; sessionId?: string; observedAt: string }) { let message: any = null; const userMessagesById = new Map(); + const sourceEvents: Array<{ index: number; event: JsonRecord; transport: JsonRecord }> = []; let decodedCount = 0; let plannedCount = 0; let appliedCount = 0; @@ -791,6 +800,15 @@ export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], optio rejections.push({ index, code: "session-mismatch", detail: eventSessionId }); continue; } + sourceEvents.push({ + index, + event: traceStep.event as JsonRecord, + transport: { + topic: text(record.topic), + partition: nullableInteger(record.partition), + offset: text(record.offset) || null + } + }); if (workbenchLiveKafkaProjectionTarget(traceStep.event as any) === "user") { const userMessageId = text((traceStep.event as any).userMessageId ?? (traceStep.event as any).messageId); const userMessage = projectWorkbenchLiveKafkaUserMessage({ @@ -817,7 +835,190 @@ export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], optio }); appliedCount += 1; } - return { message, userMessages: [...userMessagesById.values()], decodedCount, plannedCount, appliedCount, rejections }; + return { message, userMessages: [...userMessagesById.values()], sourceEvents, decodedCount, plannedCount, appliedCount, rejections }; +} + +function traceReplayLineageEvidence(input: { + sourceEvents: Array<{ index: number; event: JsonRecord; transport: JsonRecord }>; + appliedEvents: JsonRecord[]; + userMessages: JsonRecord[]; + rows: JsonRecord[]; + finalResponseText: string; +}) { + const source = traceEventLayerSummary(input.sourceEvents.map((item) => ({ + event: item.event, + position: item.index, + identity: traceLayerEventIdentity(item.event), + transport: item.transport + }))); + const applied = traceEventLayerSummary([ + ...input.userMessages.map((message, index) => ({ + event: { ...message, type: "user", eventType: "user" }, + position: index, + identity: text(message.messageId ?? message.id) || null, + transport: null + })), + ...input.appliedEvents.map((event, index) => ({ + event, + position: input.userMessages.length + index, + identity: traceLayerEventIdentity(event), + transport: null + })) + ]); + const render = traceRenderLayerSummary(input.rows); + const finalFingerprint = traceContentFingerprint(input.finalResponseText); + const finalDuplicate = finalFingerprint + ? render.assistantEntries.find((entry) => entry.contentFingerprint === finalFingerprint) ?? null + : null; + const firstVisibleDuplicateLayer = source.duplicate.first + ? "source" + : applied.duplicate.first + ? "applied" + : render.duplicate.first + ? "render" + : finalDuplicate + ? "final" + : null; + return { + counts: { + source: source.counts.total, + applied: applied.counts.total, + assistant: render.counts.assistant, + tool: render.counts.tool, + terminal: render.counts.terminal, + final: finalFingerprint ? 1 : 0 + }, + layers: { + source: { ...source.counts, assistantDuplicate: source.duplicate }, + applied: { ...applied.counts, assistantDuplicate: applied.duplicate }, + render: { ...render.counts, assistantDuplicate: render.duplicate }, + final: { + count: finalFingerprint ? 1 : 0, + contentFingerprint: finalFingerprint, + duplicatedInRender: Boolean(finalDuplicate), + matchingRenderIdentity: finalDuplicate?.identity ?? null + } + }, + duplicatePolicy: "normalized non-final assistant text sha256; final/reply-authority events are compared only at the outer-final layer", + firstVisibleDuplicateLayer, + valuesPrinted: false + }; +} + +function traceEventLayerSummary(entries: Array<{ event: JsonRecord; position: number; identity: string | null; transport: JsonRecord | null }>) { + const counts = { total: entries.length, user: 0, assistant: 0, tool: 0, terminal: 0, final: 0, other: 0 }; + const assistantEntries: JsonRecord[] = []; + for (const entry of entries) { + const kind = traceLayerEventKind(entry.event); + if (kind === "user") counts.user += 1; + else if (kind === "assistant") counts.assistant += 1; + else if (kind === "tool") counts.tool += 1; + else counts.other += 1; + const terminal = traceLayerEventTerminal(entry.event); + const final = traceLayerEventFinal(entry.event); + if (terminal) counts.terminal += 1; + if (final) counts.final += 1; + if (kind !== "assistant" || terminal || final) continue; + const contentFingerprint = traceContentFingerprint(workbenchLiveKafkaAssistantText(entry.event)); + if (!contentFingerprint) continue; + assistantEntries.push({ + position: entry.position, + identity: entry.identity, + sourceSeq: nullableInteger(entry.event.sourceSeq), + contentFingerprint, + transport: entry.transport + }); + } + return { counts, duplicate: traceAssistantDuplicateSummary(assistantEntries) }; +} + +function traceRenderLayerSummary(rows: JsonRecord[]) { + const counts = { total: rows.length, assistant: 0, tool: 0, terminal: 0, other: 0 }; + const assistantEntries: JsonRecord[] = []; + for (const [position, row] of rows.entries()) { + const assistant = /助手/u.test(text(row.header)); + const tool = text(row.rowId).startsWith("tool:"); + const terminal = row.terminal === true || text(row.rowId).startsWith("trace-completion:"); + if (assistant) counts.assistant += 1; + else if (tool) counts.tool += 1; + else if (!terminal) counts.other += 1; + if (terminal) counts.terminal += 1; + if (!assistant || terminal) continue; + const contentFingerprint = traceContentFingerprint(row.body); + if (!contentFingerprint) continue; + assistantEntries.push({ position, identity: text(row.rowId) || null, sourceSeq: nullableInteger(row.seq), contentFingerprint, transport: null }); + } + return { counts, assistantEntries, duplicate: traceAssistantDuplicateSummary(assistantEntries) }; +} + +function traceAssistantDuplicateSummary(entries: JsonRecord[]) { + const firstByFingerprint = new Map(); + const countsByFingerprint = new Map(); + let first: JsonRecord | null = null; + for (const entry of entries) { + const fingerprint = text(entry.contentFingerprint); + countsByFingerprint.set(fingerprint, (countsByFingerprint.get(fingerprint) ?? 0) + 1); + const previous = firstByFingerprint.get(fingerprint); + if (!previous) { + firstByFingerprint.set(fingerprint, entry); + continue; + } + first ??= { + contentFingerprint: fingerprint, + first: traceDuplicatePosition(previous), + duplicate: traceDuplicatePosition(entry), + valuesPrinted: false + }; + } + const duplicateCounts = [...countsByFingerprint.values()].filter((count) => count > 1); + return { + fingerprintCount: duplicateCounts.length, + occurrenceCount: duplicateCounts.reduce((sum, count) => sum + count - 1, 0), + first, + valuesPrinted: false + }; +} + +function traceDuplicatePosition(entry: JsonRecord) { + return { + position: nullableInteger(entry.position), + identity: text(entry.identity) || null, + sourceSeq: nullableInteger(entry.sourceSeq), + transport: entry.transport ?? null + }; +} + +function traceLayerEventKind(event: JsonRecord): "user" | "assistant" | "tool" | "other" { + 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 === "tool" || type === "tool_call" || text(event.toolName) === "commandExecution" || /commandExecution|^agentrun:tool:/u.test(label)) return "tool"; + return "other"; +} + +function traceLayerEventTerminal(event: JsonRecord): boolean { + const type = text(event.type ?? event.eventType); + const label = text(event.label); + return event.terminal === true || type === "result" || text(event.eventType) === "terminal" || label === "agentrun:terminal:completed" || label === "agentrun:result:completed"; +} + +function traceLayerEventFinal(event: JsonRecord): boolean { + return event.final === true || event.replyAuthority === true || Boolean(text(recordObject(event.finalResponse)?.text)); +} + +function traceLayerEventIdentity(event: JsonRecord): string | null { + const sourceEventId = text(event.sourceEventId); + if (sourceEventId) return sourceEventId; + const runId = text(event.runId); + const sourceSeq = nullableInteger(event.sourceSeq); + if (runId && sourceSeq !== null) return `${runId}:${sourceSeq}`; + return text(event.itemId ?? event.messageId ?? event.id) || null; +} + +function traceContentFingerprint(value: unknown): string | null { + const normalized = text(value).replace(/\r\n|\r/gu, "\n").replace(/\s+/gu, " ").trim(); + return normalized ? sha256(normalized) : null; } function renderWorkbenchKafkaTraceMarkdown(input: { traceId: string; sessionId: string; status: string; userMessages: any[]; sourceEventCount: number; rows: any[]; finalResponseText: string | null; sourceScanComplete: boolean; completionReason: string | null }) { @@ -1245,6 +1446,11 @@ export function renderKafkaCliText(payload: Record) { payload.input?.commandScope?.scanComplete !== undefined ? `scopeComplete=${payload.input.commandScope.scanComplete}` : null, payload.input?.readCount !== undefined ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null, payload.projection?.appliedCount !== undefined ? `applied=${payload.projection.appliedCount}` : null, + payload.replayLineage?.counts?.assistant !== undefined ? `assistant=${payload.replayLineage.counts.assistant}` : null, + payload.replayLineage?.counts?.tool !== undefined ? `tool=${payload.replayLineage.counts.tool}` : null, + payload.replayLineage?.counts?.terminal !== undefined ? `terminalRows=${payload.replayLineage.counts.terminal}` : null, + payload.replayLineage?.counts?.final !== undefined ? `final=${payload.replayLineage.counts.final}` : null, + payload.replayLineage?.firstVisibleDuplicateLayer !== undefined ? `duplicateLayer=${payload.replayLineage.firstVisibleDuplicateLayer ?? "none"}` : null, payload.render?.rowCount !== undefined ? `rows=${payload.render.rowCount}` : null, payload.projection?.terminalObserved !== undefined ? `terminal=${payload.projection.terminalObserved}` : null, payload.output ? `output=${payload.output.count}` : null, diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index 1fa2205b..6fce9610 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -57,8 +57,8 @@ export function traceDisplayRows(trace: Record = {}, events: Tr renderedSourceEvents.add(sourceEventKey); } if (!event || isNoisyTraceEvent(event)) continue; + if (isSupersededTraceItemLifecycleEvent(event, orderedEvents, index)) continue; if (isToolTraceEvent(event)) { - if (isSupersededToolStart(event, orderedEvents, index)) continue; const identity = toolIdentity(event); if (identity) { if (renderedToolIdentities.has(identity)) continue; @@ -102,6 +102,7 @@ export function traceDisplayRows(trace: Record = {}, events: Tr if (rows.length > 0) return rows; if (orderedEvents.length === 0) return []; return orderedEvents + .filter((event, index) => !isSupersededTraceItemLifecycleEvent(event, orderedEvents, index)) .filter((event) => !isSuppressedTraceEvent(event)) .filter((event) => !isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) .map((event) => traceDisplayRow(effectiveTrace, event, displayOptions)); @@ -488,6 +489,10 @@ function traceSourceEventKey(event: TraceEvent | null | undefined): string | nul return `${source}:${sourceSeq}:${nonEmptyString(event.label ?? event.type) ?? "event"}`; } +function isSupersededTraceItemLifecycleEvent(event: TraceEvent, events: TraceEvent[], index: number): boolean { + return isSupersededToolStart(event, events, index) || isLateAssistantProgressAfterCompletion(event, events, index); +} + function isSupersededToolStart(event: TraceEvent, events: TraceEvent[], index: number): boolean { if (!isToolStartTraceEvent(event)) return false; const identity = toolIdentity(event); @@ -495,6 +500,16 @@ 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; + 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"); +} + function isNoisyTraceEvent(event: TraceEvent): boolean { const label = String(event.label ?? ""); if (isRequestTraceEvent(event) || isSetupTraceEvent(event) || isCompletionTraceEvent(event) || isTerminalAssistantTraceEvent(event) || isAssistantTraceEvent(event)) return false; 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 014747ec..9efa81b5 100644 --- a/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts @@ -135,6 +135,35 @@ test("pure Kafka Trace renders sourceSeq-only rows in ingress order without proj assert.deepEqual(durableRows, []); }); +test("pure Kafka Trace coalesces late assistant progress by stable item lifecycle", () => { + const events = [ + { sourceEventId: "evt_completed_a", sourceSeq: 44, label: "agentrun:assistant:message", type: "assistant", status: "running", itemId: "msg_a", assistantSource: "completed-agent-message", message: "same visible text" }, + { sourceEventId: "evt_progress_a", sourceSeq: 183, label: "agentrun:assistant:message", type: "assistant", status: "running", itemId: "msg_a", assistantSource: "agent-message-delta-progress", message: "same visible text" }, + { sourceEventId: "evt_completed_b", sourceSeq: 184, label: "agentrun:assistant:message", type: "assistant", status: "running", itemId: "msg_b", assistantSource: "completed-agent-message", message: "same visible text" }, + { sourceEventId: "evt_terminal", sourceSeq: 185, label: "agentrun:terminal:completed", type: "result", status: "completed", terminal: true } + ]; + + const rows = traceDisplayRows({ traceId: "trc_assistant_item_lifecycle", eventSource: "hwlab-kafka-sse", status: "completed" }, events); + const assistantRows = rows.filter((row) => /助手/u.test(row.header)); + + assert.deepEqual(assistantRows.map((row) => row.seq), [44, 184]); + assert.equal(assistantRows.every((row) => row.body === "same visible text"), true, "distinct itemId values must not be content-deduplicated"); + assert.equal(rows.some((row) => row.seq === 183), false); + assert.equal(rows.at(-1)?.rowId.startsWith("trace-completion:"), true); +}); + +test("pure Kafka Trace keeps item lifecycle suppression in the empty-row fallback", () => { + const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ + { sourceEventId: "evt_completed", sourceSeq: 1, label: "agentrun:assistant:message", type: "assistant", itemId: "msg_final", assistantSource: "completed-agent-message", message: "Final response" }, + { sourceEventId: "evt_late_progress", sourceSeq: 2, label: "agentrun:assistant:message", type: "assistant", itemId: "msg_final", assistantSource: "agent-message-delta-progress", message: "Stale late progress" } + ], { + finalResponsePlacement: "outer", + outerFinalResponseText: "Final response" + }); + + assert.deepEqual(rows, []); +}); + test("R1 trace API snapshots stay authoritative over compact result traces", () => { const previous = { traceId: "trc_merge",