Merge pull request #2688 from pikasTech/fix/workbench-replay-cli-summary

feat(workbench): 单次 CLI 输出回放性能摘要

internal/cloud/workbench-kafka-refresh-handoff.ts

@@ -30,6 +30,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) {
   let bootstrapSettled = false;
   let scopedTopic = null;
   let scopedPartition = null;
+  let queryDiagnostics = null;
   let deliveryChain = Promise.resolve(true);
   let pendingLiveDeliveries = 0;
   const counts = {
@@ -60,6 +61,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) {
     unsubscribe = options.subscribeLive(onLiveEnvelope);
     try {
       const queryResult = await options.queryRetention({ signal: abortController.signal });
+      queryDiagnostics = boundedQueryDiagnostics(queryResult);
       if (failure) throw failure;
       if (phase === "closed") throw handoffError("workbench_kafka_refresh_stopped", "Kafka refresh handoff stopped before the retention barrier completed.", phase);
       validateQueryCompletion(queryResult);
@@ -362,6 +364,7 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) {
         windowAdvanced: retentionWindowAdvanced
       },
       identityWindowLimit,
+      query: queryDiagnostics,
       valuesRedacted: true
     };
   }
@@ -394,6 +397,34 @@ export function createWorkbenchKafkaRefreshHandoff(options = {}) {
   };
 }
 
+function boundedQueryDiagnostics(result) {
+  const timing = result?.timing && typeof result.timing === "object" ? result.timing : {};
+  return {
+    completionReason: textValue(result?.completion?.reason ?? result?.completionReason),
+    complete: result?.completion?.complete === true,
+    scannedCount: numericValue(result?.scannedCount),
+    matchedCount: numericValue(result?.matchedCount),
+    partitionKeyScoped: result?.partitionKeyScoped === true,
+    targetPartition: partitionValue(result?.targetPartition),
+    timing: {
+      endOffsetSnapshotMs: numericValue(timing.endOffsetSnapshotMs),
+      groupOffsetsMs: numericValue(timing.groupOffsetsMs),
+      consumerConnectMs: numericValue(timing.consumerConnectMs),
+      consumerSubscribeMs: numericValue(timing.consumerSubscribeMs),
+      scanMs: numericValue(timing.scanMs),
+      cleanupMs: numericValue(timing.cleanupMs),
+      totalMs: numericValue(timing.totalMs),
+      valuesPrinted: false
+    },
+    valuesPrinted: false
+  };
+}
+
+function numericValue(value) {
+  const number = Number(value);
+  return Number.isFinite(number) && number >= 0 ? number : null;
+}
+
 export function workbenchKafkaRefreshErrorPayload(error, scope = {}) {
   const typed = asHandoffError(error, textValue(error?.phase) || "failed");
   const details = safeHandoffDiagnosticDetails(typed.details);

internal/workbench/workbench.test.ts

@@ -164,28 +164,31 @@ describe("Workbench native HTTP adapter", () => {
         expect(options).toMatchObject({ sessionId, traceId, partitionKey: sessionId, fromBeginning: true });
         return {
           topic: "hwlab.event.v1",
-          events: [{
+          events: Array.from({ length: 100 }, (_, index) => ({
             topic: "hwlab.event.v1",
             partition: 0,
-            offset: "4",
+            offset: String(index),
             value: {
               schema: "hwlab.event.v1",
               eventType: "hwlab.trace.event.projected",
-              eventId: "hwlab:evt_l0_refresh",
+              eventId: `hwlab:evt_l0_refresh_${index}`,
-              sourceEventId: "evt_l0_refresh",
+              sourceEventId: `evt_l0_refresh_${index}`,
               sessionId,
               hwlabSessionId: sessionId,
               traceId,
-              event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: "evt_l0_refresh" }
+              event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: `evt_l0_refresh_${index}` }
             }
-          }],
+          })),
           completionReason: "end-offset",
           completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true },
           reachedEndOffsets: true,
           endOffsetsAvailable: true,
-          endOffsets: [{ partition: 0, startOffset: "0", endOffset: "5" }],
+          endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }],
-          scannedCount: 5,
+          scannedCount: 100,
-          matchedCount: 1
+          matchedCount: 100,
+          partitionKeyScoped: true,
+          targetPartition: 0,
+          timing: { endOffsetSnapshotMs: 2, groupOffsetsMs: 0, consumerConnectMs: 3, consumerSubscribeMs: 1, scanMs: 7, cleanupMs: 1, totalMs: 14 }
         };
       }
     };
@@ -197,14 +200,15 @@ describe("Workbench native HTTP adapter", () => {
     });
     const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
     try {
-      expect(await runWorkbenchCli([
+      const result = await runWorkbenchCli([
         "events", "inspect",
         "--session-id", sessionId,
         "--trace-id", traceId,
         "--over-api",
         "--api-url", `http://127.0.0.1:${server.port}`,
         "--timeout-ms", "1000"
-      ], {})).toMatchObject({
+      ], {});
+      expect(result).toMatchObject({
         ok: true,
         connected: true,
         connectedContract: {
@@ -212,11 +216,24 @@ describe("Workbench native HTTP adapter", () => {
           realtimeSource: "hwlab.event.v1",
           replay: true,
           liveOnly: false,
-          refreshReplay: { phase: "live", counts: { replayed: 1 } }
+          refreshReplay: {
+            phase: "live",
+            counts: { replayed: 100 },
+            query: {
+              complete: true,
+              scannedCount: 100,
+              matchedCount: 100,
+              partitionKeyScoped: true,
+              targetPartition: 0,
+              timing: { scanMs: 7, totalMs: 14 }
+            }
+          }
         },
-        eventCount: 1,
+        eventCount: 100,
-        eventTypes: ["backend"]
+        eventTypes: ["backend"],
+        eventTypeCounts: { backend: 100 }
       });
+      expect(JSON.stringify(result).length).toBeLessThan(10_000);
     } finally {
       server.stop(true);
       await app.close();
@@ -312,7 +329,7 @@ describe("Workbench native HTTP adapter", () => {
         "--trace-id", "trc_l0_missing_terminal",
         "--over-api",
         "--api-url", `http://127.0.0.1:${server.port}`,
-        "--timeout-ms", "50",
+        "--timeout-ms", "250",
         "--wait-for", "assistant,terminal"
       ], {});
       for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(2);

tools/src/workbench-cli.ts

@@ -95,7 +95,8 @@ 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, connectedContract, 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 };
+  const allEventTypes = business.map(eventType);
+  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, connectedContract, eventCount: business.length, eventTypes: timingDetail === "full" ? allEventTypes : [...new Set(allEventTypes)], eventTypeCounts: countValues(allEventTypes), 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 {
@@ -153,6 +154,8 @@ type SseFrame = { name: string; data: Record<string, any>; observedAt: string };
 function workbenchConnectedContract(data: Record<string, any>) {
   const refreshReplay = record(data.refreshReplay);
   const counts = record(refreshReplay?.counts);
+  const query = record(refreshReplay?.query);
+  const queryTiming = record(query?.timing);
   return {
     deliverySemantics: String(data.deliverySemantics ?? "").trim() || null,
     realtimeSource: String(data.realtimeSource ?? "").trim() || null,
@@ -170,12 +173,40 @@ function workbenchConnectedContract(data: Record<string, any>) {
         bufferedDelivered: Number(counts.bufferedDelivered ?? 0),
         liveDelivered: Number(counts.liveDelivered ?? 0),
         deduplicated: Number(counts.deduplicated ?? 0)
+      } : null,
+      query: query ? {
+        completionReason: String(query.completionReason ?? "").trim() || null,
+        complete: query.complete === true,
+        scannedCount: nullableNumber(query.scannedCount),
+        matchedCount: nullableNumber(query.matchedCount),
+        partitionKeyScoped: query.partitionKeyScoped === true,
+        targetPartition: nullableNumber(query.targetPartition),
+        timing: queryTiming ? {
+          endOffsetSnapshotMs: nullableNumber(queryTiming.endOffsetSnapshotMs),
+          groupOffsetsMs: nullableNumber(queryTiming.groupOffsetsMs),
+          consumerConnectMs: nullableNumber(queryTiming.consumerConnectMs),
+          consumerSubscribeMs: nullableNumber(queryTiming.consumerSubscribeMs),
+          scanMs: nullableNumber(queryTiming.scanMs),
+          cleanupMs: nullableNumber(queryTiming.cleanupMs),
+          totalMs: nullableNumber(queryTiming.totalMs),
+          valuesPrinted: false
+        } : null,
+        valuesPrinted: false
       } : null
     } : null,
     valuesPrinted: false
   };
 }
 function businessFrames(frames: SseFrame[]) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); }
+function countValues(values: string[]) {
+  const counts: Record<string, number> = {};
+  for (const value of values) counts[value] = (counts[value] ?? 0) + 1;
+  return counts;
+}
+function nullableNumber(value: unknown) {
+  const parsed = Number(value);
+  return Number.isFinite(parsed) && parsed >= 0 ? parsed : null;
+}
 function framesInspectionSatisfied(frames: SseFrame[], minEvents: number, waitFor: EventSemantic[]) {
   return frames.some((entry) => entry.name === "workbench.connected")
     && inspectionSatisfied(businessFrames(frames), minEvents, waitFor);