Merge pull request #2479 from pikasTech/fix/2474-trace-raw-inspector

[v0.3][#2474] 修复纯 Kafka Trace 并增加原始事件观察窗

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

@@ -1,6 +1,7 @@
 export interface TraceEventRow {
   rowId: string;
   seq: number | null;
+  sequenceAuthority: TraceSequenceAuthority;
   tone: "ok" | "blocked" | "warn" | "source";
   header: string;
   body: string | null;
@@ -11,13 +12,25 @@ export interface TraceEventRow {
 
 type TraceEvent = Record<string, unknown>;
 type TraceClockFormatter = (value: string) => string;
+export type TraceSequenceAuthority = "projected" | "source";
 
 export interface TraceDisplayRowsOptions {
   formatClock?: TraceClockFormatter;
 }
 
+interface ResolvedTraceDisplayRowsOptions extends TraceDisplayRowsOptions {
+  sequenceAuthority: TraceSequenceAuthority;
+}
+
 export function traceDisplayRows(trace: Record<string, unknown> = {}, events: TraceEvent[] = [], options: TraceDisplayRowsOptions = {}): TraceEventRow[] {
-  const orderedEvents = traceEventsForDisplay(events.filter(hasProjectedTraceSequence));
+  const displayOptions: ResolvedTraceDisplayRowsOptions = {
+    ...options,
+    sequenceAuthority: traceSequenceAuthority(trace)
+  };
+  const orderedEvents = traceEventsForDisplay(
+    events.filter((event) => hasTraceSequence(event, displayOptions.sequenceAuthority)),
+    displayOptions.sequenceAuthority
+  );
   const effectiveTrace = traceWithInferredStart(trace, orderedEvents);
   const finalResponseText = traceFinalResponseText(effectiveTrace);
   const finalResponseFingerprint = traceMessageFingerprint(finalResponseText);
@@ -41,7 +54,7 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
         if (renderedToolIdentities.has(identity)) continue;
         renderedToolIdentities.add(identity);
       }
-      rows.push(traceToolCallRow(effectiveTrace, event, options));
+      rows.push(traceToolCallRow(effectiveTrace, event, displayOptions));
       continue;
     }
     if (isRequestTraceEvent(event)) {
@@ -53,33 +66,37 @@ export function traceDisplayRows(trace: Record<string, unknown> = {}, events: Tr
     if (isTerminalAssistantTraceEvent(event)) {
       if (hasCompletionEvent) continue;
       if (finalResponseText) {
-        rows.push(traceFinalResponseRow(event, finalResponseText, options));
+        rows.push(traceFinalResponseRow(event, finalResponseText, displayOptions));
       } else {
-        rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true }, options));
+        rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true }, displayOptions));
       }
       continue;
     }
     if (isAssistantTraceEvent(event)) {
       if (finalResponseFingerprint && traceAssistantEventFingerprint(effectiveTrace, event) === finalResponseFingerprint) continue;
-      rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, options));
+      rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, displayOptions));
       continue;
     }
     if (isCompletionTraceEvent(event)) {
       completionEvent ??= event;
       continue;
     }
-    rows.push(traceDisplayRow(effectiveTrace, event, options));
+    rows.push(traceDisplayRow(effectiveTrace, event, displayOptions));
   }
-  if (completionEvent) rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent, options));
-  const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, orderedEvents, options);
+  if (completionEvent) rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent, displayOptions));
+  const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, orderedEvents, displayOptions);
   if (progressRow) rows.push(progressRow);
   if (rows.length > 0) return rows;
   if (orderedEvents.length === 0) return [];
-  return orderedEvents.filter((event) => !isSuppressedTraceEvent(event)).map((event) => traceDisplayRow(effectiveTrace, event, options));
+  return orderedEvents.filter((event) => !isSuppressedTraceEvent(event)).map((event) => traceDisplayRow(effectiveTrace, event, displayOptions));
 }
 
-function hasProjectedTraceSequence(event: TraceEvent | null | undefined): boolean {
-  return traceEventDisplaySeq(event) !== null;
+function traceSequenceAuthority(trace: Record<string, unknown>): TraceSequenceAuthority {
+  return nonEmptyString(trace.eventSource) === "hwlab-kafka-sse" ? "source" : "projected";
+}
+
+function hasTraceSequence(event: TraceEvent | null | undefined, authority: TraceSequenceAuthority): boolean {
+  return traceEventDisplaySeq(event, authority) !== null;
 }
 
 export function renderTraceRowsMarkdown(rows: TraceEventRow[] = []): string {
@@ -106,10 +123,11 @@ function renderTraceMessageRowMarkdown(row: TraceEventRow): string {
   return traceBoldMarkdown(row.header || `_No readable trace row body._`);
 }
 
-function traceEventsForDisplay(events: TraceEvent[] = []): TraceEvent[] {
+function traceEventsForDisplay(events: TraceEvent[] = [], authority: TraceSequenceAuthority): TraceEvent[] {
+  if (authority === "source") return [...events];
   return [...events].sort((left, right) => {
-    const leftSeq = traceEventDisplaySeq(left);
-    const rightSeq = traceEventDisplaySeq(right);
+    const leftSeq = traceEventDisplaySeq(left, authority);
+    const rightSeq = traceEventDisplaySeq(right, authority);
     if (leftSeq !== null || rightSeq !== null) {
       if (leftSeq === null) return 1;
       if (rightSeq === null) return -1;
@@ -126,12 +144,20 @@ function traceEventsForDisplay(events: TraceEvent[] = []): TraceEvent[] {
   });
 }
 
-function traceEventDisplaySeq(event: TraceEvent | null | undefined): number | null {
-  return numberOrNull(event?.projectedSeq);
+function traceEventDisplaySeq(event: TraceEvent | null | undefined, authority: TraceSequenceAuthority): number | null {
+  return numberOrNull(authority === "source" ? event?.sourceSeq : event?.projectedSeq);
 }
 
-function traceEventIdentityToken(event: TraceEvent): string | number | null {
-  const seq = traceEventDisplaySeq(event);
+function traceEventIdentityToken(event: TraceEvent, authority: TraceSequenceAuthority): string | number | null {
+  if (authority === "source") {
+    const sourceEventId = nonEmptyString(event.sourceEventId);
+    if (sourceEventId) return sourceEventId;
+    const runId = nonEmptyString(event.runId);
+    const sourceSeq = traceEventDisplaySeq(event, authority);
+    if (runId && sourceSeq !== null) return `${runId}:${sourceSeq}`;
+    return sourceSeq;
+  }
+  const seq = traceEventDisplaySeq(event, authority);
   return seq ?? nonEmptyString(event.projectedSeq);
 }
 
@@ -149,15 +175,16 @@ function stripTraceToolOutputFromSummary(value: string): string {
   return value.replace(/\s+(stdout:|stderr:|exitCode=).*$/isu, "").trim();
 }
 
-function traceToolCallRow(trace: Record<string, unknown>, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow {
+function traceToolCallRow(trace: Record<string, unknown>, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
   const command = cleanShellCommand(event.command);
   const status = traceStatusToken(event);
   const toolName = traceToolName(event, command);
   const body = traceToolCallBody(event, command);
-  const eventKey = traceEventIdentityToken(event);
+  const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
   return {
     rowId: `tool:${event.itemId ?? eventKey ?? `${event.label ?? event.type ?? "tool"}:${event.createdAt ?? "unknown"}`}`,
-    seq: traceEventDisplaySeq(event),
+    seq: traceEventDisplaySeq(event, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: traceEventTone(event),
     header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${status} ${toolName}`.trim(),
     body,
@@ -221,14 +248,15 @@ export function traceNoiseEventCount(events: TraceEvent[] = []): number {
   return Array.isArray(events) ? events.filter((event) => isNoisyTraceEvent(event)).length : 0;
 }
 
-function traceAssistantMessageRow(trace: Record<string, unknown>, event: TraceEvent, { terminal }: { terminal: boolean }, options: TraceDisplayRowsOptions): TraceEventRow {
+function traceAssistantMessageRow(trace: Record<string, unknown>, event: TraceEvent, { terminal }: { terminal: boolean }, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
   const text = traceAssistantEventText(trace, event);
   const index = Number.isInteger(event.messageIndex) ? Number(event.messageIndex) : null;
   const count = Number.isInteger(event.messageCount) ? Number(event.messageCount) : null;
-  const eventKey = traceEventIdentityToken(event);
+  const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
   return {
     rowId: `event:${eventKey ?? `${event.label ?? event.type ?? "assistant"}:${event.createdAt ?? "unknown"}`}`,
-    seq: traceEventDisplaySeq(event),
+    seq: traceEventDisplaySeq(event, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: "ok",
     header: `${traceEventClock(event, options)}  ${terminal ? "助手最终消息" : "助手消息"}${index ? ` ${index}${count ? `/${count}` : ""}` : ""}`,
     terminal: terminal ? true : undefined,
@@ -250,11 +278,12 @@ function traceMessageFingerprint(value: unknown): string | null {
   return text || null;
 }
 
-function traceFinalResponseRow(event: TraceEvent, finalText: string, options: TraceDisplayRowsOptions): TraceEventRow {
-  const eventKey = traceEventIdentityToken(event);
+function traceFinalResponseRow(event: TraceEvent, finalText: string, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
+  const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
   return {
     rowId: `trace-final-response:${eventKey ?? "completed"}`,
-    seq: traceEventDisplaySeq(event),
+    seq: traceEventDisplaySeq(event, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: "ok",
     header: `${traceEventClock(event, options)}  助手最终消息`,
     terminal: true,
@@ -268,24 +297,26 @@ function traceFinalResponseText(trace: Record<string, unknown>): string | null {
   return nonEmptyString(response?.text ?? response?.content ?? response?.message);
 }
 
-function traceCompletionSummaryRow(trace: Record<string, unknown>, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow {
-  const eventKey = traceEventIdentityToken(event);
+function traceCompletionSummaryRow(trace: Record<string, unknown>, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
+  const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
   return {
     rowId: `trace-completion:${eventKey ?? "turn"}`,
-    seq: traceEventDisplaySeq(event),
+    seq: traceEventDisplaySeq(event, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: traceEventTone(event),
     header: `${traceEventClock(event, options)}  轮次完成（总耗时 ${formatTraceDuration(traceRelativeMs(trace, event))}）`,
     body: null
   };
 }
 
-function traceNoiseSummaryRow(trace: Record<string, unknown>, events: TraceEvent[], options: TraceDisplayRowsOptions): TraceEventRow {
+function traceNoiseSummaryRow(trace: Record<string, unknown>, events: TraceEvent[], options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
   const lastEvent = events.at(-1);
   const status = nonEmptyString(trace.status ?? trace.traceStatus) ?? "running";
   const lastLabel = lastEvent ? readableTraceLabel(lastEvent) : "未观测";
   return {
     rowId: "trace-noise-summary",
-    seq: traceEventDisplaySeq(lastEvent),
+    seq: traceEventDisplaySeq(lastEvent, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: lastEvent ? traceEventTone(lastEvent) : "source",
     header: `Trace ${status}，等待可读事件`,
     body: `已隐藏 ${events.length} 条 AgentRun backend 状态事件。最新原始事件：${lastLabel}。`,
@@ -293,23 +324,24 @@ function traceNoiseSummaryRow(trace: Record<string, unknown>, events: TraceEvent
   };
 }
 
-function traceBackendProgressSummaryRow(trace: Record<string, unknown>, rows: TraceEventRow[], events: TraceEvent[], options: TraceDisplayRowsOptions): TraceEventRow | null {
+function traceBackendProgressSummaryRow(trace: Record<string, unknown>, rows: TraceEventRow[], events: TraceEvent[], options: ResolvedTraceDisplayRowsOptions): TraceEventRow | null {
   if (!Array.isArray(events) || events.length === 0) return null;
   if (rows.some((row) => row.rowId.startsWith("tool:") || /助手|轮次完成/u.test(row.header))) return null;
   const lastRenderedSeq = Math.max(0, ...rows.map((row) => row.seq ?? 0).filter(Number.isFinite));
   const hiddenProgress = events.filter((event) => {
-    const seq = traceEventDisplaySeq(event) ?? 0;
+    const seq = traceEventDisplaySeq(event, options.sequenceAuthority) ?? 0;
     return seq > lastRenderedSeq && (isRequestTraceEvent(event) || isSetupTraceEvent(event) || isNoisyTraceEvent(event));
   });
   return hiddenProgress.length > 0 ? traceNoiseSummaryRow(trace, hiddenProgress, options) : null;
 }
 
-function traceDisplayRow(trace: Record<string, unknown>, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow {
+function traceDisplayRow(trace: Record<string, unknown>, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow {
   const label = readableTraceLabel(event);
-  const eventKey = traceEventIdentityToken(event);
+  const eventKey = traceEventIdentityToken(event, options.sequenceAuthority);
   return {
     rowId: `event:${eventKey ?? `${event.label ?? event.type ?? "event"}:${event.createdAt ?? "unknown"}`}`,
-    seq: traceEventDisplaySeq(event),
+    seq: traceEventDisplaySeq(event, options.sequenceAuthority),
+    sequenceAuthority: options.sequenceAuthority,
     tone: traceEventTone(event),
     header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${traceStatusToken(event)} ${label}`.trim(),
     body: traceDisplayBody(event)
@@ -380,8 +412,12 @@ function toolIdentity(event: TraceEvent): string | null {
 
 function traceSourceEventKey(event: TraceEvent | null | undefined): string | null {
   if (!event || typeof event !== "object") return null;
-  const source = nonEmptyString(event.source);
+  const sourceEventId = nonEmptyString(event.sourceEventId);
+  if (sourceEventId) return `source-event:${sourceEventId}`;
+  const runId = nonEmptyString(event.runId);
   const sourceSeq = nonEmptyString(event.sourceSeq);
+  if (runId && sourceSeq) return `run-source:${runId}:${sourceSeq}`;
+  const source = nonEmptyString(event.source);
   if (!source || !sourceSeq) return null;
   return `${source}:${sourceSeq}:${nonEmptyString(event.label ?? event.type) ?? "event"}`;
 }