Merge pull request #2642 from pikasTech/fix/l1-agentrun-native-trace

修复 L1 Workbench Kafka SSE 事件可见性

cmd/hwlab-workbench-api/main.ts

@@ -1,9 +1,11 @@
 #!/usr/bin/env bun
 import { createWorkbenchHttpApp } from "../../internal/workbench/http.ts";
 import { workbenchRuntime } from "../../internal/workbench/runtime.ts";
+import { startHwlabKafkaEventBridge } from "../../internal/cloud/kafka-event-bridge.ts";
 
 const runtime = workbenchRuntime();
-const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, mode: runtime.mode, close: runtime.close });
+const kafkaEventBridge = runtime.mode === "agentrun-native" ? startHwlabKafkaEventBridge({ env: process.env }) : null;
+const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, mode: runtime.mode, kafkaEventBridge, close: async () => { await kafkaEventBridge?.stop?.(); await runtime.close(); } });
 const host = process.env.WORKBENCH_API_HOST || "0.0.0.0";
 const port = positivePort(process.env.WORKBENCH_API_PORT || process.env.PORT, 6677);
 const server = Bun.serve({ hostname: host, port, fetch: (request) => app.fetch(request) });

internal/workbench/http.ts

@@ -1,6 +1,6 @@
 import type { WorkbenchCommand, WorkbenchMode } from "./contracts.ts";
 
-export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise<any>; snapshot?: () => Promise<{ sessions: Record<string, Record<string, unknown>>; turns: Record<string, Record<string, unknown>> }>; authorization?: string; mode?: WorkbenchMode; close?: () => Promise<void> }) {
+export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise<any>; snapshot?: () => Promise<{ sessions: Record<string, Record<string, unknown>>; turns: Record<string, Record<string, unknown>> }>; authorization?: string; mode?: WorkbenchMode; kafkaEventBridge?: any; close?: () => Promise<void> }) {
   return {
     async fetch(request: Request): Promise<Response> {
       try {
@@ -20,7 +20,7 @@ export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchC
           if (url.pathname === "/v1/caserun/cases") return json(200, { ok: true, status: "unavailable", cases: [], count: 0, mode: options.mode ?? "native-test" });
           if (url.pathname === "/v1/hwpod/specs") return json(200, { ok: true, status: "unavailable", specs: [], mode: options.mode ?? "native-test" });
           if (url.pathname === "/v1/hwpod-node-ops") return json(200, { ok: true, status: "unavailable", events: [], groups: [], mode: options.mode ?? "native-test" });
-          if (url.pathname === "/v1/workbench/events" && request.method === "GET") return nativeEventStream(options.mode);
+          if (url.pathname === "/v1/workbench/events" && request.method === "GET") return nativeEventStream(options, url);
           if (url.pathname === "/v1/workbench/sessions" && request.method === "GET") return nativeSessionList(options, url);
           const sessionMatch = /^\/v1\/workbench\/sessions\/([^/]+)(?:\/(messages))?$/u.exec(url.pathname);
           if (sessionMatch && request.method === "GET") return nativeSessionDetail(options, decodeURIComponent(sessionMatch[1]), sessionMatch[2] === "messages");
@@ -89,18 +89,47 @@ async function nativeTurn(options: { snapshot?: () => Promise<any>; mode?: Workb
   return turn ? json(200, { ok: true, ...turn, mode: options.mode ?? "native-test" }) : json(404, { ok: false, error: { code: "turn_not_found", message: "native turn was not found" } });
 }
 async function requiredSnapshot(options: { snapshot?: () => Promise<any> }) { if (!options.snapshot) throw Object.assign(new Error("native projection is unavailable"), { code: "native_projection_unavailable" }); return options.snapshot(); }
-function nativeEventStream(mode: WorkbenchMode = "native-test") {
+async function nativeEventStream(options: { mode?: WorkbenchMode; kafkaEventBridge?: any }, url: URL) {
+  const mode = options.mode ?? "native-test";
+  const bridge = options.kafkaEventBridge;
+  const sessionId = text(url.searchParams.get("sessionId"));
+  const traceId = text(url.searchParams.get("traceId"));
+  if (!sessionId && !traceId) return json(400, { ok: false, error: { code: "workbench_realtime_scope_required", message: "sessionId or traceId is required" } });
+  if (mode === "agentrun-native" && (bridge?.capabilities?.liveKafkaSse !== true || typeof bridge?.subscribeLiveHwlabEvents !== "function")) {
+    return json(503, { ok: false, error: { code: "workbench_live_kafka_unconfigured", message: "Native Workbench requires the Kafka SSE bridge" } });
+  }
   const encoder = new TextEncoder();
   let heartbeat: ReturnType<typeof setInterval> | undefined;
+  let unsubscribe: (() => void) | undefined;
+  let closed = false;
   return new Response(new ReadableStream({
     start(controller) {
-      controller.enqueue(encoder.encode(`event: capability\ndata: ${JSON.stringify({ mode, terminalAuthority: mode === "agentrun-native" ? "agentrun-command-result" : "fixture" })}\n\n`));
+      const write = (name: string, payload: unknown) => { if (!closed) controller.enqueue(encoder.encode(`event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`)); };
+      write("workbench.connected", { type: "connected", status: "connected", mode, realtimeSource: mode === "agentrun-native" ? "hwlab.event.v1" : "fixture", deliverySemantics: "live-only", liveOnly: true, replay: false, filters: { sessionId: sessionId || null, traceId: traceId || null }, valuesPrinted: false });
+      if (typeof bridge?.subscribeLiveHwlabEvents === "function") unsubscribe = bridge.subscribeLiveHwlabEvents((envelope: any) => {
+        if (!nativeEnvelopeMatches(envelope, sessionId, traceId)) return;
+        write("hwlab.event.v1", envelope);
+      });
+      void Promise.resolve(bridge?.liveReady ?? bridge?.ready).catch((error: any) => {
+        write("workbench.error", { type: "error", status: "blocked", error: { code: error?.code ?? "workbench_live_kafka_start_failed", message: error?.message ?? String(error) }, realtimeSource: "hwlab.event.v1", valuesRedacted: true });
+        closed = true;
+        if (heartbeat) clearInterval(heartbeat);
+        unsubscribe?.();
+        controller.close();
+      });
       heartbeat = setInterval(() => controller.enqueue(encoder.encode(`: native-test-heartbeat ${Date.now()}\n\n`)), 10_000);
     },
-    cancel() { if (heartbeat) clearInterval(heartbeat); }
+    cancel() { closed = true; if (heartbeat) clearInterval(heartbeat); unsubscribe?.(); }
   }), { status: 200, headers: { "content-type": "text/event-stream", "cache-control": "no-store", connection: "keep-alive" } });
 }
 
+function nativeEnvelopeMatches(envelope: any, sessionId: string, traceId: string) {
+  const event = envelope?.event && typeof envelope.event === "object" ? envelope.event : {};
+  const envelopeSessionId = text(envelope?.sessionId ?? envelope?.hwlabSessionId ?? event.sessionId);
+  const envelopeTraceId = text(envelope?.traceId ?? event.traceId);
+  return (!sessionId || envelopeSessionId === sessionId) && (!traceId || envelopeTraceId === traceId);
+}
+
 function actorFrom(request: Request, nativeTest: boolean) {
   const id = text(request.headers.get("x-hwlab-actor-id"));
   if (!id && nativeTest) return { id: "usr_native", role: "user" };

internal/workbench/workbench.test.ts

@@ -4,6 +4,7 @@ import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts
 import { createWorkbenchDispatcher } from "./dispatcher.ts";
 import { createCloudWorkbenchApplication } from "./cloud-application.ts";
 import { createWorkbenchHttpApp } from "./http.ts";
+import { runWorkbenchCli } from "../../tools/src/workbench-cli.ts";
 
 describe("Workbench dispatcher", () => {
   test("submits through Temporal without synthesizing terminal state", async () => {
@@ -88,6 +89,57 @@ describe("Workbench native HTTP adapter", () => {
       expect(await response.json()).toMatchObject({ ok: true, status: "unavailable", mode: "native-test" });
     }
   });
+
+  test("L0 CLI observes AgentRun-derived events only through the Kafka SSE route", 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_sse",
+        "--trace-id", "trc_l0_sse",
+        "--over-api",
+        "--api-url", `http://127.0.0.1:${server.port}`,
+        "--timeout-ms", "1000"
+      ], {});
+      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"] });
+    } 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({
+      mode: "agentrun-native",
+      kafkaEventBridge: {
+        capabilities: { liveKafkaSse: true },
+        liveReady: Promise.reject(startupError),
+        subscribeLiveHwlabEvents() { return () => {}; }
+      },
+      async dispatch() { return { ok: true }; },
+      async snapshot() { return { sessions: {}, turns: {} }; }
+    }).fetch(new Request("http://localhost/v1/workbench/events?traceId=trc_l0_kafka_error"));
+    expect(response.status).toBe(200);
+    const body = await response.text();
+    expect(body).toContain("event: workbench.connected");
+    expect(body).toContain("event: workbench.error");
+    expect(body).toContain("kafka_broker_unavailable");
+  });
 });
 
 function stubApplication(): WorkbenchApplication {

tools/src/workbench-cli.ts

@@ -6,6 +6,7 @@ export async function runWorkbenchCli(argv: string[], env: Record<string, string
   const parsed = parse(argv);
   if (parsed.help || parsed.positionals.length === 0) return help();
   if (parsed.positionals[0] === "service") return workbenchNativeServiceCommand({ service: parsed.positionals[1], action: parsed.positionals[2] ?? "status", cwd: process.cwd(), env });
+  if (parsed.positionals[0] === "events" && parsed.positionals[1] === "inspect") return inspectEvents(parsed, env);
   const command = commandFrom(parsed, env);
   if (parsed.overApi) return overApi(command, parsed, env);
   const runtime = workbenchRuntime(env);
@@ -26,6 +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] [--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"
     ],
@@ -36,6 +38,53 @@ function help() {
   };
 }
 
+async function inspectEvents(parsed: Parsed, env: Record<string, string | undefined>) {
+  if (!parsed.overApi) throw codedError("workbench_events_over_api_required", "workbench events inspect requires --over-api so it exercises the product SSE route");
+  const baseUrl = parsed.values["api-url"] || env.WORKBENCH_API_URL;
+  if (!baseUrl) throw codedError("workbench_api_url_required", "WORKBENCH_API_URL or --api-url is required with --over-api");
+  const sessionId = parsed.values["session-id"] ?? "";
+  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 params = new URLSearchParams();
+  if (sessionId) params.set("sessionId", sessionId);
+  if (traceId) params.set("traceId", traceId);
+  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> }> = [];
+  let buffer = "";
+  try {
+    const response = await fetch(`${baseUrl.replace(/\/$/u, "")}${path}`, { signal: controller.signal });
+    if (!response.ok || !response.body) throw codedError("workbench_sse_open_failed", `Workbench SSE returned HTTP ${response.status}`);
+    const reader = response.body.getReader();
+    const decoder = new TextDecoder();
+    while (true) {
+      const item = await reader.read();
+      if (item.done) break;
+      buffer += decoder.decode(item.value, { stream: true });
+      let boundary = buffer.indexOf("\n\n");
+      while (boundary >= 0) {
+        const block = buffer.slice(0, boundary);
+        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) controller.abort();
+        boundary = buffer.indexOf("\n\n");
+      }
+    }
+  } catch (error: any) {
+    if (error?.name !== "AbortError") throw error;
+  } finally {
+    clearTimeout(timer);
+  }
+  const business = frames.filter((entry) => entry.name === "hwlab.event.v1");
+  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 };
+}
+
 function commandFrom(parsed: Parsed, env: Record<string, string | undefined>): WorkbenchCommand {
   const [group, action] = parsed.positionals;
   if (group === "health") return { operation: "health" };
@@ -77,6 +126,17 @@ function parse(argv: string[]): Parsed {
   return 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; }
+function parseSseBlock(block: string): { name: string; data: Record<string, any> } | 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) ?? {} };
+}
 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; }
 function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }