Merge pull request #2502 from pikasTech/fix/2500-assistant-replay

[#2500] 按助手 item 生命周期收敛 Trace 重复

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 事件映射函数；

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" }, [

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, any>): string {
 export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], options: { traceId: string; sessionId?: string; observedAt: string }) {
   let message: any = null;
   const userMessagesById = new Map<string, any>();
+  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<string, JsonRecord>();
+  const countsByFingerprint = new Map<string, number>();
+  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<string, any>) {
     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,

tools/src/hwlab-cli/trace-renderer.ts

@@ -57,8 +57,8 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, 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<string, unknown> = {}, 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;

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",