From 28c6a21d2738a21d85dcb7f7b9501a1966e64f76 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 18 Jul 2026 09:54:06 +0200 Subject: [PATCH] fix: wait for Workbench event semantics Co-Authored-By: Codex --- internal/workbench/workbench.test.ts | 88 ++++++++++++++++++++++++++++ scripts/workbench-native-smoke.mjs | 39 ++++++++---- tools/src/workbench-cli.ts | 60 +++++++++++++++---- 3 files changed, 166 insertions(+), 21 deletions(-) diff --git a/internal/workbench/workbench.test.ts b/internal/workbench/workbench.test.ts index 4b0405f4..6dec557e 100644 --- a/internal/workbench/workbench.test.ts +++ b/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({ diff --git a/scripts/workbench-native-smoke.mjs b/scripts/workbench-native-smoke.mjs index 8c5dd34b..77afabf8 100755 --- a/scripts/workbench-native-smoke.mjs +++ b/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)); + }); + }); +} diff --git a/tools/src/workbench-cli.ts b/tools/src/workbench-cli.ts index 7175761a..6b8c680a 100644 --- a/tools/src/workbench-cli.ts +++ b/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 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 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 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): 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 | 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; } +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 }>) { return frames.filter((entry) => entry.name === "hwlab.event.v1"); } +function inspectionSatisfied(business: Array<{ name: string; data: Record }>, 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 }>) { + const semanticCounts = Object.fromEntries(eventSemantics.map((semantic) => [semantic, 0])) as Record; + 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, 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 }) { return String(frame.data?.event?.eventType ?? frame.data?.event?.type ?? frame.data?.eventType ?? "unknown"); } function parseSseBlock(block: string): { name: string; data: Record } | null { if (!block || block.startsWith(":")) return null; const lines = block.split(/\r?\n/u);