diff --git a/cmd/hwlab-workbench-api/main.ts b/cmd/hwlab-workbench-api/main.ts index f73f659a..c1e11b95 100644 --- a/cmd/hwlab-workbench-api/main.ts +++ b/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) }); diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts index c3944742..c2d00240 100644 --- a/internal/workbench/http.ts +++ b/internal/workbench/http.ts @@ -1,6 +1,6 @@ import type { WorkbenchCommand, WorkbenchMode } from "./contracts.ts"; -export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise; snapshot?: () => Promise<{ sessions: Record>; turns: Record> }>; authorization?: string; mode?: WorkbenchMode; close?: () => Promise }) { +export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise; snapshot?: () => Promise<{ sessions: Record>; turns: Record> }>; authorization?: string; mode?: WorkbenchMode; kafkaEventBridge?: any; close?: () => Promise }) { return { async fetch(request: Request): Promise { 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; 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 }) { 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 | 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" }; diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 7e5e8b9b..8a358931 100644 --- a/internal/workbench/workbench.test.ts +++ b/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 { diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index baa88114..9e3ad3fa 100644 --- a/tools/src/workbench-cli.ts +++ b/tools/src/workbench-cli.ts @@ -6,6 +6,7 @@ export async function runWorkbenchCli(argv: string[], env: Record) { + 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 }> = []; + 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): 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 | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : 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 } | 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) { 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 }); }