fix: 披露 Workbench Kafka 重放耗时

internal/cloud/server-workbench-realtime-http.ts

@@ -758,17 +758,34 @@ async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response,
     liveBufferLimit: refreshReplay.liveBufferLimit,
     identityWindowLimit: refreshReplay.matchedEventLimit + refreshReplay.liveBufferLimit,
     subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener),
-    queryRetention: ({ signal }) => bridge.queryHwlabEventRetention({
-      sessionId: requestedSessionId,
-      traceId: requestedTraceId,
-      limit: refreshReplay.matchedEventLimit,
-      scanLimit: refreshReplay.scanLimit,
-      timeoutMs: refreshReplay.timeoutMs,
-      groupIdPrefix: refreshReplay.groupIdPrefix,
-      partitionKey: requestedSessionId,
-      fromBeginning: true,
-      signal
-    }),
+    queryRetention: async ({ signal }) => {
+      const result = await bridge.queryHwlabEventRetention({
+        sessionId: requestedSessionId,
+        traceId: requestedTraceId,
+        limit: refreshReplay.matchedEventLimit,
+        scanLimit: refreshReplay.scanLimit,
+        timeoutMs: refreshReplay.timeoutMs,
+        groupIdPrefix: refreshReplay.groupIdPrefix,
+        partitionKey: requestedSessionId,
+        fromBeginning: true,
+        signal
+      });
+      (options.logger ?? console).log(JSON.stringify({
+        code: "workbench-kafka-refresh-query",
+        component: "hwlab-workbench-realtime",
+        sessionId: requestedSessionId,
+        traceId: requestedTraceId,
+        completionReason: result?.completionReason ?? null,
+        complete: result?.completion?.complete === true,
+        scannedCount: result?.scannedCount ?? null,
+        matchedCount: result?.matchedCount ?? null,
+        partitionKeyScoped: result?.partitionKeyScoped === true,
+        targetPartition: result?.targetPartition ?? null,
+        timing: result?.timing ?? null,
+        valuesPrinted: false
+      }));
+      return result;
+    },
     deliverEvent: (envelope, transport) => enqueueEvent("hwlab.event.v1", envelope, transport),
     deliverConnected: (summary) => enqueueEvent("workbench.connected", {
       type: "connected",

internal/workbench/workbench.test.ts

@@ -132,7 +132,20 @@ describe("Workbench native HTTP adapter", () => {
       ], {});
       for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(5);
       for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_sse", traceId: "trc_l0_sse", event: { type: "backend", eventType: "backend", label: "agentrun:backend:command-created" } });
-      expect(await pending).toMatchObject({ ok: true, operation: "events.inspect", connected: true, eventCount: 1, eventTypes: ["backend"] });
+      expect(await pending).toMatchObject({
+        ok: true,
+        operation: "events.inspect",
+        connected: true,
+        connectedContract: {
+          deliverySemantics: "live-only",
+          realtimeSource: "hwlab.event.v1",
+          replay: false,
+          liveOnly: true,
+          refreshReplay: null
+        },
+        eventCount: 1,
+        eventTypes: ["backend"]
+      });
     } finally {
       server.stop(true);
       await app.close();

tools/src/workbench-cli.ts

@@ -89,11 +89,13 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
     clearTimeout(timer);
   }
   const business = businessFrames(frames);
-  const connected = frames.some((entry) => entry.name === "workbench.connected");
+  const connectedFrame = frames.find((entry) => entry.name === "workbench.connected") ?? null;
+  const connected = connectedFrame !== null;
+  const connectedContract = connectedFrame ? workbenchConnectedContract(connectedFrame.data) : null;
   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, 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 };
+  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 };
 }
 
 function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
@@ -148,6 +150,31 @@ function semanticWait(value: unknown): EventSemantic[] {
   return [...new Set(requested)] as EventSemantic[];
 }
 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);
+  return {
+    deliverySemantics: String(data.deliverySemantics ?? "").trim() || null,
+    realtimeSource: String(data.realtimeSource ?? "").trim() || null,
+    replay: data.replay === true,
+    liveOnly: data.liveOnly === true,
+    lossPossible: data.lossPossible === true,
+    refreshReplay: refreshReplay ? {
+      phase: String(refreshReplay.phase ?? "").trim() || null,
+      topic: String(refreshReplay.topic ?? "").trim() || null,
+      barrier: Array.isArray(refreshReplay.barrier) ? refreshReplay.barrier : [],
+      counts: counts ? {
+        matched: Number(counts.matched ?? 0),
+        replayed: Number(counts.replayed ?? 0),
+        buffered: Number(counts.buffered ?? 0),
+        bufferedDelivered: Number(counts.bufferedDelivered ?? 0),
+        liveDelivered: Number(counts.liveDelivered ?? 0),
+        deduplicated: Number(counts.deduplicated ?? 0)
+      } : null
+    } : null,
+    valuesPrinted: false
+  };
+}
 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;