Merge pull request #2658 from pikasTech/fix/2657-workbench-semantic-wait

修复 Workbench SSE 语义终态等待

internal/workbench/workbench.test.ts

@@ -125,6 +125,94 @@ describe("Workbench native HTTP adapter", () => {
     }
   });
 
+  test("L0 CLI waits for explicit Kafka event semantics in one SSE window", async () => {
+    const subscribers = new Set<(envelope: any) => void>();
+    const bridge = {
+      capabilities: { liveKafkaSse: true },
+      ready: Promise.resolve(),
+      subscribeLiveHwlabEvents(listener: (envelope: any) => void) { subscribers.add(listener); return () => subscribers.delete(listener); }
+    };
+    const app = createWorkbenchHttpApp({
+      mode: "agentrun-native",
+      kafkaEventBridge: bridge,
+      async dispatch() { return { ok: true }; },
+      async snapshot() { return { sessions: {}, turns: {} }; }
+    });
+    const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
+    try {
+      const pending = runWorkbenchCli([
+        "events", "inspect",
+        "--session-id", "ses_l0_semantics",
+        "--trace-id", "trc_l0_semantics",
+        "--over-api",
+        "--api-url", `http://127.0.0.1:${server.port}`,
+        "--timeout-ms", "1000",
+        "--wait-for", "user,backend,assistant,terminal,final"
+      ], {});
+      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" }
+      ];
+      for (const event of events) {
+        for (const listener of subscribers) listener({ schema: "hwlab.event.v1", sessionId: "ses_l0_semantics", traceId: "trc_l0_semantics", event });
+      }
+      expect(await pending).toMatchObject({
+        ok: true,
+        eventCount: 4,
+        eventTypes: ["user", "backend", "assistant", "terminal"],
+        waitFor: ["user", "backend", "assistant", "terminal", "final"],
+        observedSemantics: ["user", "backend", "assistant", "terminal", "final"],
+        missingSemantics: [],
+        semanticCounts: { user: 1, backend: 1, assistant: 1, terminal: 1, final: 1 },
+        terminalStatuses: ["completed"]
+      });
+    } finally {
+      server.stop(true);
+      await app.close();
+    }
+  });
+
+  test("L0 CLI reports missing Kafka event semantics on timeout", async () => {
+    const subscribers = new Set<(envelope: any) => void>();
+    const app = createWorkbenchHttpApp({
+      mode: "agentrun-native",
+      kafkaEventBridge: {
+        capabilities: { liveKafkaSse: true },
+        ready: Promise.resolve(),
+        subscribeLiveHwlabEvents(listener: (envelope: any) => void) { subscribers.add(listener); return () => subscribers.delete(listener); }
+      },
+      async dispatch() { return { ok: true }; },
+      async snapshot() { return { sessions: {}, turns: {} }; }
+    });
+    const server = Bun.serve({ port: 0, fetch: (request) => app.fetch(request) });
+    try {
+      const pending = runWorkbenchCli([
+        "events", "inspect",
+        "--trace-id", "trc_l0_missing_terminal",
+        "--over-api",
+        "--api-url", `http://127.0.0.1:${server.port}`,
+        "--timeout-ms", "50",
+        "--wait-for", "assistant,terminal"
+      ], {});
+      for (let index = 0; index < 20 && subscribers.size === 0; index += 1) await Bun.sleep(2);
+      for (const listener of subscribers) listener({ schema: "hwlab.event.v1", traceId: "trc_l0_missing_terminal", event: { type: "assistant", eventType: "assistant" } });
+      expect(await pending).toMatchObject({
+        ok: false,
+        status: "event-timeout",
+        observedSemantics: ["assistant"],
+        missingSemantics: ["terminal"],
+        terminalStatuses: [],
+        error: { code: "workbench_kafka_sse_event_missing" }
+      });
+    } finally {
+      server.stop(true);
+      await app.close();
+    }
+  });
+
   test("L0 SSE exposes Kafka startup failure without waiting for a read model", async () => {
     const startupError = Object.assign(new Error("Kafka broker unavailable"), { code: "kafka_broker_unavailable" });
     const response = await createWorkbenchHttpApp({

scripts/workbench-native-smoke.mjs

@@ -1,6 +1,7 @@
 #!/usr/bin/env node
 import { spawn } from "node:child_process";
 import { mkdtemp, rm } from "node:fs/promises";
+import { createServer } from "node:net";
 import os from "node:os";
 import path from "node:path";
 
@@ -10,26 +11,31 @@ const root = path.resolve(import.meta.dirname, "..");
 const stateDir = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-native-"));
 const children = [];
 const logs = [];
+const apiPort = await availablePort();
+const webPort = await availablePort();
+const apiUrl = `http://127.0.0.1:${apiPort}`;
+const webUrl = `http://127.0.0.1:${webPort}/workbench`;
 
 try {
   const nativeEnv = {
     ...process.env,
     WORKBENCH_MODE: "native-test",
-    WORKBENCH_NATIVE_STATE_FILE: path.join(stateDir, "state.json")
+    WORKBENCH_NATIVE_STATE_FILE: path.join(stateDir, "state.json"),
+    WORKBENCH_API_IDLE_TIMEOUT_SECONDS: "60"
   };
   const api = start(["bun", "cmd/hwlab-workbench-api/main.ts"], {
     ...nativeEnv,
     WORKBENCH_API_HOST: "127.0.0.1",
-    WORKBENCH_API_PORT: "6677"
+    WORKBENCH_API_PORT: String(apiPort)
   }, "api");
   children.push(api);
-  await waitFor("http://127.0.0.1:6677/health/ready");
+  await waitFor(`${apiUrl}/health/ready`);
 
-  const web = start(["bun", "run", "workbench:web:dev"], process.env, "web");
+  const web = start(["bun", "run", "workbench:web:dev"], { ...process.env, WORKBENCH_WEB_PORT: String(webPort) }, "web");
   children.push(web);
-  await waitFor("http://127.0.0.1:5173/workbench", { acceptHtml: true });
+  await waitFor(webUrl, { acceptHtml: true });
 
-  const commandHealth = await requiredJson(await fetch("http://127.0.0.1:6677/v1/workbench/commands", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ operation: "health" }) }), "command.health");
+  const commandHealth = await requiredJson(await fetch(`${apiUrl}/v1/workbench/commands`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ operation: "health" }) }), "command.health");
   if (commandHealth.mode !== "native-test") throw new Error("native command transport did not expose mode=native-test");
 
   const localSession = await runWorkbenchCli(["session", "create", "--actor-id", "usr_native", "--provider-profile", "native-test"], nativeEnv);
@@ -41,7 +47,7 @@ try {
   const localCancel = await runWorkbenchCli(["turn", "cancel", "--actor-id", "usr_native", "--trace-id", localTraceId], nativeEnv);
   if (localCancel.data?.status !== "cancel_requested") throw new Error("CLI local cancel was not accepted");
 
-  const overApiArgs = ["--over-api", "--api-url", "http://127.0.0.1:6677"];
+  const overApiArgs = ["--over-api", "--api-url", apiUrl];
   const apiHealth = await runWorkbenchCli(["health", ...overApiArgs], nativeEnv);
   if (apiHealth.transport !== "api" || apiHealth.httpStatus !== 200) throw new Error("CLI --over-api health transport contract failed");
   const apiSession = await runWorkbenchCli(["session", "create", "--actor-id", "usr_native", "--provider-profile", "native-test", ...overApiArgs], nativeEnv);
@@ -54,18 +60,18 @@ try {
   if (apiCancel.data?.status !== "cancel_requested") throw new Error("CLI --over-api cancel was not accepted");
 
   const actorHeaders = { "content-type": "application/json", cookie: "hwlab_session=native-test" };
-  const sessionResponse = await fetch("http://127.0.0.1:6677/v1/agent/sessions", { method: "POST", headers: actorHeaders, body: JSON.stringify({ providerProfile: "native-test" }) });
+  const sessionResponse = await fetch(`${apiUrl}/v1/agent/sessions`, { method: "POST", headers: actorHeaders, body: JSON.stringify({ providerProfile: "native-test" }) });
   const session = await requiredJson(sessionResponse, "session.create");
   const sessionId = session.session?.sessionId;
   if (!sessionId) throw new Error("native session.create returned no sessionId");
 
   const traceId = "trc_native_smoke";
-  const submit = await requiredJson(await fetch("http://127.0.0.1:6677/v1/agent/chat", { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId, message: "native smoke" }) }), "turn.submit");
+  const submit = await requiredJson(await fetch(`${apiUrl}/v1/agent/chat`, { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId, message: "native smoke" }) }), "turn.submit");
   if (submit.orchestrationMode !== "native-test" || submit.resultSynthesized !== false) throw new Error("native submit mode/authority contract failed");
-  const cancel = await requiredJson(await fetch("http://127.0.0.1:6677/v1/agent/chat/cancel", { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId }) }), "turn.cancel");
+  const cancel = await requiredJson(await fetch(`${apiUrl}/v1/agent/chat/cancel`, { method: "POST", headers: actorHeaders, body: JSON.stringify({ sessionId, traceId }) }), "turn.cancel");
   if (cancel.status !== "cancel_requested") throw new Error("native cancel was not accepted");
 
-  process.stdout.write(`${JSON.stringify({ ok: true, mode: "native-test", probeEndpoints: { api: "http://127.0.0.1:6677", web: "http://127.0.0.1:5173/workbench" }, publicEndpoints: null, exposureAuthority: "owning-yaml-service-launcher", sessionId, traceId, cli: { local: "application-dispatcher", overApi: "POST /v1/workbench/commands" }, services: { api: "independent", web: "vite-hmr" }, terminalAuthority: "fixture-only", resultSynthesized: false })}\n`);
+  process.stdout.write(`${JSON.stringify({ ok: true, mode: "native-test", probeEndpoints: { api: apiUrl, web: webUrl }, publicEndpoints: null, exposureAuthority: "owning-yaml-service-launcher", sessionId, traceId, cli: { local: "application-dispatcher", overApi: "POST /v1/workbench/commands" }, services: { api: "independent", web: "vite-hmr" }, terminalAuthority: "fixture-only", resultSynthesized: false })}\n`);
 } catch (error) {
   process.stderr.write(`${JSON.stringify({ ok: false, error: { code: "workbench_native_smoke_failed", message: error instanceof Error ? error.message : String(error) }, logs: logs.slice(-20) })}\n`);
   process.exitCode = 1;
@@ -92,3 +98,14 @@ async function waitFor(url, options = {}) {
   throw new Error(`timed out waiting for ${url}`);
 }
 async function requiredJson(response, operation) { const body = await response.json().catch(() => null); if (!response.ok || !body) throw new Error(`${operation} failed with HTTP ${response.status}`); return body; }
+async function availablePort() {
+  return await new Promise((resolve, reject) => {
+    const server = createServer();
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", () => {
+      const address = server.address();
+      const port = typeof address === "object" && address ? address.port : 0;
+      server.close((error) => error ? reject(error) : resolve(port));
+    });
+  });
+}

tools/src/workbench-cli.ts

@@ -27,13 +27,13 @@ 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] [--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]",
       "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"
     ],
     transportContract: "--over-api only changes transport; --overapi is unsupported",
     localConfig: ["WORKBENCH_MODE", "WORKBENCH_NATIVE_STATE_FILE"],
-    nativeServiceConfig: ["WORKBENCH_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PUBLIC_HOST", "WORKBENCH_API_PORT", "WORKBENCH_WORKER_BIND_HOST", "WORKBENCH_WORKER_PROBE_HOST", "WORKBENCH_WORKER_HEALTH_PORT", "WORKBENCH_WEB_BIND_HOST", "WORKBENCH_WEB_PROBE_HOST", "WORKBENCH_WEB_PUBLIC_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"],
+    nativeServiceConfig: ["WORKBENCH_API_BIND_HOST", "WORKBENCH_API_PROBE_HOST", "WORKBENCH_API_PUBLIC_HOST", "WORKBENCH_API_PORT", "WORKBENCH_API_IDLE_TIMEOUT_SECONDS", "WORKBENCH_WORKER_BIND_HOST", "WORKBENCH_WORKER_PROBE_HOST", "WORKBENCH_WORKER_HEALTH_PORT", "WORKBENCH_WEB_BIND_HOST", "WORKBENCH_WEB_PROBE_HOST", "WORKBENCH_WEB_PUBLIC_HOST", "WORKBENCH_WEB_PORT", "WORKBENCH_NATIVE_API_URL", "WORKBENCH_WEB_RUNTIME_CONFIG"],
     apiConfig: ["WORKBENCH_API_URL", "HWLAB_API_KEY"]
   };
 }
@@ -46,7 +46,8 @@ 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 minEvents = positiveInteger(parsed.values["min-events"], 1);
+  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();
   if (sessionId) params.set("sessionId", sessionId);
   if (traceId) params.set("traceId", traceId);
@@ -70,7 +71,7 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
         buffer = buffer.slice(boundary + 2);
         const frame = parseSseBlock(block);
         if (frame) frames.push(frame);
-        if (frames.filter((entry) => entry.name === "hwlab.event.v1").length >= minEvents) {
+        if (inspectionSatisfied(businessFrames(frames), minEvents, waitFor)) {
           await reader.cancel().catch(() => undefined);
           break readStream;
         }
@@ -78,17 +79,20 @@ async function inspectEvents(parsed: Parsed, env: Record<string, string | undefi
       }
     }
   } catch (error: any) {
-    const businessCount = frames.filter((entry) => entry.name === "hwlab.event.v1").length;
-    if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && businessCount >= minEvents)) {
-      throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${businessCount}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`);
+    const business = businessFrames(frames);
+    if (error?.name !== "AbortError" && !(error?.code === "ECONNRESET" && inspectionSatisfied(business, minEvents, waitFor))) {
+      const observed = semanticSummary(business);
+      throw codedError("workbench_sse_read_failed", `Workbench SSE read failed: ${error?.code ?? error?.name ?? "unknown"}; frames=${frames.length}; businessEvents=${business.length}; observed=${observed.observedSemantics.join(",")}; connected=${frames.some((entry) => entry.name === "workbench.connected")}`);
     }
   } finally {
     clearTimeout(timer);
   }
-  const business = frames.filter((entry) => entry.name === "hwlab.event.v1");
+  const business = businessFrames(frames);
   const connected = frames.some((entry) => entry.name === "workbench.connected");
-  const ok = connected && business.length >= minEvents;
-  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((entry) => String(entry.data?.event?.eventType ?? entry.data?.event?.type ?? entry.data?.eventType ?? "unknown")), timeoutMs, minEvents, error: ok ? undefined : { code: "workbench_kafka_sse_event_missing", message: "Workbench SSE did not deliver the required Kafka business events within the bounded window" }, valuesPrinted: false };
+  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 };
 }
 
 function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
@@ -134,6 +138,42 @@ function parse(argv: string[]): Parsed {
 function identity(data: any) { return { sessionId: data?.sessionId ?? data?.session?.sessionId ?? null, traceId: data?.traceId ?? null, workflowId: data?.workflowId ?? null, workflowRunId: data?.workflowRunId ?? null }; }
 function record(value: unknown): Record<string, any> | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, any> : null; }
 function positiveInteger(value: unknown, fallback: number): number { const parsed = Number.parseInt(String(value ?? ""), 10); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; }
+const eventSemantics = ["user", "backend", "assistant", "terminal", "final"] as const;
+type EventSemantic = typeof eventSemantics[number];
+function semanticWait(value: unknown): EventSemantic[] {
+  const requested = String(value ?? "").split(",").map((item) => item.trim().toLowerCase()).filter(Boolean);
+  const unsupported = requested.filter((item) => !eventSemantics.includes(item as 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[]) {
+  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> }>) {
+  const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record<EventSemantic, number>;
+  const terminalStatuses: string[] = [];
+  for (const frame of business) {
+    const event = record(frame.data?.event) ?? {};
+    for (const semantic of eventSemantics) if (matchesSemantic(event, semantic)) semanticCounts[semantic] += 1;
+    if (matchesSemantic(event, "terminal")) {
+      const status = String(event.terminalStatus ?? event.status ?? "").trim();
+      if (status && !terminalStatuses.includes(status)) terminalStatuses.push(status);
+    }
+  }
+  return { observedSemantics: eventSemantics.filter((semantic) => semanticCounts[semantic] > 0), semanticCounts, terminalStatuses };
+}
+function matchesSemantic(event: Record<string, any>, semantic: EventSemantic) {
+  const type = String(event.eventType ?? event.type ?? "").trim().toLowerCase();
+  if (semantic === "user") return type === "user" || type === "user_message";
+  if (semantic === "backend") return type === "backend" || type === "backend_status";
+  if (semantic === "assistant") return type === "assistant" || type === "assistant_message";
+  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 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 {
   if (!block || block.startsWith(":")) return null;
   const lines = block.split(/\r?\n/u);