Merge pull request #2680 from pikasTech/feat/workbench-l0-performance-timing

feat: 增强 Workbench L0 阶段耗时分析

internal/workbench/workbench.test.ts

@@ -165,10 +165,10 @@ describe("Workbench native HTTP adapter", () => {
       ], {});
       for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5);
       const events = [
-        { type: "user", eventType: "user" },
-        { type: "backend", eventType: "backend" },
-        { type: "assistant", eventType: "assistant", final: true, replyAuthority: true, finalResponse: { text: "done" } },
-        { type: "result", eventType: "terminal", terminal: true, terminalStatus: "completed", status: "completed" }
+        { type: "user", eventType: "user", createdAt: "2026-07-19T16:00:00.000Z" },
+        { type: "backend", eventType: "backend", createdAt: "2026-07-19T16:00:02.000Z" },
+        { type: "assistant", eventType: "assistant", createdAt: "2026-07-19T16:01:02.000Z", final: true, replyAuthority: true, finalResponse: { text: "done" } },
+        { type: "result", eventType: "terminal", createdAt: "2026-07-19T16:01:03.000Z", terminal: true, terminalStatus: "completed", status: "completed" }
       ];
       for (const event of events) {
         for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_semantics", traceId: "trc_l0_semantics", event });
@@ -181,7 +181,27 @@ describe("Workbench native HTTP adapter", () => {
         observedSemantics: ["user", "backend", "assistant", "terminal", "final"],
         missingSemantics: [],
         semanticCounts: { user: 1, backend: 1, assistant: 1, terminal: 1, final: 1 },
-        terminalStatuses: ["completed"]
+        terminalStatuses: ["completed"],
+        timing: {
+          clock: "agentrun-source-event-created-at",
+          phasesMs: {
+            firstEventToBackend: 2000,
+            admissionAndDispatch: null,
+            runnerProvisioning: null,
+            runtimePreparation: null,
+            providerExecution: null,
+            backendToAssistant: 60000,
+            assistantToTerminal: 1000,
+            firstEventToTerminal: 63000,
+            firstEventToFinal: 62000
+          },
+          slowestGap: {
+            durationMs: 60000,
+            toIndex: 2,
+            toEventType: "assistant"
+          },
+          valuesPrinted: false
+        }
       });
     } finally {
       server.stop(true);

tools/src/workbench-cli.ts

@@ -27,7 +27,7 @@ function help() {
       "hwlab-cli workbench health [--over-api]",
       "hwlab-cli workbench session create --actor-id ID [--provider-profile PROFILE] [--over-api]",
       "hwlab-cli workbench turn submit --actor-id ID --session-id ID --message TEXT [--trace-id ID] [--over-api]",
-      "hwlab-cli workbench events inspect --session-id ID [--trace-id ID] --over-api [--timeout-ms MS] [--wait-for user,backend,assistant,terminal,final] [--min-events N]",
+      "hwlab-cli workbench events inspect --session-id ID [--trace-id ID] --over-api [--timeout-ms MS] [--wait-for user,backend,assistant,terminal,final] [--min-events N] [--timing-detail summary|full]",
       "hwlab-cli workbench turn cancel --actor-id ID --trace-id ID [--over-api]",
       "hwlab-cli workbench service api|worker|web start|stop|restart|status|logs"
     ],
@@ -46,6 +46,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
   const traceId = parsed.values["trace-id"] ?? "";
   if (!sessionId && !traceId) throw codedError("workbench_realtime_scope_required", "--session-id or --trace-id is required");
   const timeoutMs = positiveInteger(parsed.values["timeout-ms"], 5000);
+  const timingDetail = timingDetailValue(parsed.values["timing-detail"]);
   const waitFor = semanticWait(parsed.values["wait-for"]);
   const minEvents = parsed.values["min-events"] ? positiveInteger(parsed.values["min-events"], 1) : waitFor.length > 0 ? 0 : 1;
   const params = new URLSearchParams();
@@ -54,7 +55,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
   const path = `/v1/workbench/events?${params.toString()}`;
   const controller = new AbortController();
   const timer = setTimeout(() => controller.abort(), timeoutMs);
-  const frames: Array<{ name: string; data: Record<string, any> }> = [];
+  const frames: SseFrame[] = [];
   let buffer = "";
   try {
     const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal });
@@ -92,7 +93,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
   const semantics = semanticSummary(business);
   const missingSemantics = waitFor.filter((semantic) => !semantics.observedSemantics.includes(semantic));
   const ok = connected && inspectionSatisfied(business, minEvents, waitFor);
-  return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false };
+  return { ok, operation: "events.inspect", status: ok ? "passed" : "event-timeout", transport: "api-sse", baseUrl, route: `GET ${path}`, identity: { sessionId: sessionId || null, traceId: traceId || null }, connected, eventCount: business.length, eventTypes: business.map(eventType), timeoutMs, minEvents, waitFor, ...semantics, timing: timingSummary(business, timingDetail), missingSemantics, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: `Workbench SSE did not deliver the required Kafka event conditions; missing=${missingSemantics.join(",") || (business.length < minEvents ? `min-events:${minEvents}` : "connection")}` }, valuesPrinted: false };
 }
 
 function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
@@ -146,13 +147,14 @@ function semanticWait(value: unknown): EventSemantic[] {
   if (unsupported.length > 0) throw codedError("workbench_event_semantic_unsupported", `unsupported --wait-for semantics: ${unsupported.join(",")}; supported=${eventSemantics.join(",")}`);
   return [...new Set(requested)] as EventSemantic[];
 }
-function businessFrames(frames: Array<{ name: string; data: Record<string, any> }>) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
-function inspectionSatisfied(business: Array<{ name: string; data: Record<string, any> }>, minEvents: number, waitFor: EventSemantic[]) {
+type SseFrame = { name: string; data: Record<string, any>; observedAt: string };
+function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
+function inspectionSatisfied(business: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
   if (business.length < minEvents) return false;
   const observed = semanticSummary(business).observedSemantics;
   return waitFor.every((semantic) => observed.includes(semantic));
 }
-function semanticSummary(business: Array<{ name: string; data: Record<string, any> }>) {
+function semanticSummary(business: SseFrame[]) {
   const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record<EventSemantic, number>;
   const terminalStatuses: string[] = [];
   for (const frame of business) {
@@ -173,15 +175,105 @@ function matchesSemantic(event: Record<string, any>, semantic: EventSemantic) {
   if (semantic === "terminal") return event.terminal === true || type === "terminal" || type === "terminal_status" || type === "result" || event.agentRunEventType === "terminal_status";
   return event.final === true || event.replyAuthority === true || Boolean(record(event.finalResponse)?.text);
 }
+function timingSummary(business: SseFrame[], detail: "summary" | "full") {
+  const rawTimeline = business.map((frame, index) => {
+    const event = record(frame.data?.event) ?? {};
+    const sourceAt = timestamp(event.createdAt);
+    const producedAt = timestamp(frame.data?.producedAt);
+    const observedAt = timestamp(frame.observedAt);
+    const semantics = eventSemantics.filter((semantic) => matchesSemantic(event, semantic));
+    return {
+      index,
+      eventType: eventType(frame),
+      label: String(event.label ?? "").trim() || null,
+      semantics,
+      sourceAt,
+      producedAt,
+      observedAt,
+      projectionLagMs: elapsedMs(sourceAt, producedAt),
+      deliveryLagMs: elapsedMs(producedAt, observedAt)
+    };
+  });
+  const seen = new Set<string>();
+  const timeline = rawTimeline.filter((entry) => {
+    const key = [entry.eventType, entry.label, entry.sourceAt].join("\u0000");
+    if (seen.has(key)) return false;
+    seen.add(key);
+    return true;
+  }).sort((left, right) => Date.parse(left.sourceAt ?? left.observedAt ?? "") - Date.parse(right.sourceAt ?? right.observedAt ?? ""))
+    .map((entry, index) => ({ ...entry, index }));
+  const sourceTimeline = timeline.filter((entry) => entry.sourceAt !== null);
+  const firstSourceAt = sourceTimeline[0]?.sourceAt ?? null;
+  const withDeltas = timeline.map((entry, index) => ({
+    ...entry,
+    sinceFirstMs: elapsedMs(firstSourceAt, entry.sourceAt),
+    sincePreviousMs: elapsedMs(timeline[index - 1]?.sourceAt ?? null, entry.sourceAt)
+  }));
+  const firstBackendAt = firstSemanticAt(withDeltas, "backend");
+  const firstAssistantAt = firstSemanticAt(withDeltas, "assistant");
+  const terminalAt = firstSemanticAt(withDeltas, "terminal");
+  const finalAt = firstSemanticAt(withDeltas, "final");
+  const runnerDispatchedAt = firstLabelAt(withDeltas, "agentrun:backend:runner-dispatch-completed");
+  const runnerReadyAt = firstLabelAt(withDeltas, "agentrun:backend:run-claimed");
+  const providerStartedAt = firstLabelAt(withDeltas, "agentrun:backend:turn/started");
+  const slowestGap = withDeltas
+    .filter((entry) => entry.sincePreviousMs !== null)
+    .sort((left, right) => (right.sincePreviousMs ?? 0) - (left.sincePreviousMs ?? 0))[0] ?? null;
+  return {
+    clock: "agentrun-source-event-created-at",
+    rawEventCount: rawTimeline.length,
+    analyzedEventCount: timeline.length,
+    duplicateEventCount: rawTimeline.length - timeline.length,
+    milestones: { firstEventAt: firstSourceAt, firstBackendAt, firstAssistantAt, terminalAt, finalAt },
+    phasesMs: {
+      firstEventToBackend: elapsedMs(firstSourceAt, firstBackendAt),
+      admissionAndDispatch: elapsedMs(firstSourceAt, runnerDispatchedAt),
+      runnerProvisioning: elapsedMs(runnerDispatchedAt, runnerReadyAt),
+      runtimePreparation: elapsedMs(runnerReadyAt, providerStartedAt),
+      providerExecution: elapsedMs(providerStartedAt, firstAssistantAt),
+      backendToAssistant: elapsedMs(firstBackendAt, firstAssistantAt),
+      assistantToTerminal: elapsedMs(firstAssistantAt, terminalAt),
+      firstEventToTerminal: elapsedMs(firstSourceAt, terminalAt),
+      firstEventToFinal: elapsedMs(firstSourceAt, finalAt)
+    },
+    slowestGap: slowestGap ? {
+      durationMs: slowestGap.sincePreviousMs,
+      toIndex: slowestGap.index,
+      toEventType: slowestGap.eventType,
+      toLabel: slowestGap.label
+    } : null,
+    ...(detail === "full" ? { timeline: withDeltas } : {}),
+    valuesPrinted: false
+  };
+}
+function timingDetailValue(value: unknown): "summary" | "full" {
+  const detail = String(value ?? "summary").trim().toLowerCase();
+  if (detail === "summary" || detail === "full") return detail;
+  throw codedError("workbench_timing_detail_unsupported", "--timing-detail must be summary or full");
+}
+function firstSemanticAt(timeline: Array<{ semantics: EventSemantic[]; sourceAt: string | null }>, semantic: EventSemantic) {
+  return timeline.find((entry) => entry.semantics.includes(semantic) && entry.sourceAt)?.sourceAt ?? null;
+}
+function firstLabelAt(timeline: Array<{ label: string | null; sourceAt: string | null }>, label: string) {
+  return timeline.find((entry) => entry.label === label && entry.sourceAt)?.sourceAt ?? null;
+}
+function timestamp(value: unknown): string | null {
+  const text = String(value ?? "").trim();
+  return text && Number.isFinite(Date.parse(text)) ? new Date(Date.parse(text)).toISOString() : null;
+}
+function elapsedMs(start: string | null, end: string | null): number | null {
+  if (!start || !end) return null;
+  return Math.max(0, Date.parse(end) - Date.parse(start));
+}
 function eventType(frame: { data: Record<string, any> }) { return String(frame.data?.event?.eventType ?? frame.data?.event?.type ?? frame.data?.eventType ?? "unknown"); }
-function parseSseBlock(block: string): { name: string; data: Record<string, any> } | null {
+function parseSseBlock(block: string): SseFrame | null {
   if (!block || block.startsWith(":")) return null;
   const lines = block.split(/\r?\n/u);
   const name = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() || "message";
   const dataText = lines.filter((line) => line.startsWith("data:" )).map((line) => line.slice(5).trimStart()).join("\n");
   if (!dataText) return null;
   const data = JSON.parse(dataText);
-  return { name, data: record(data) ?? {} };
+  return { name, data: record(data) ?? {}, observedAt: new Date().toISOString() };
 }
 function compact(value: Record<string, unknown>) { return Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined)); }
 function required(parsed: Parsed, key: string) { const value = parsed.values[key]; if (!value) throw codedError("option_required", `--${key} is required`); return value; }