mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: connect native Workbench to Kafka SSE

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
root
2026-07-18 05:53:23 +02:00
parent 3fa9e8541e
commit ae24fcb899
4 changed files with 149 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbench/http.ts
+34 -5
View File
@@ -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
+52
View File
@@ -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 {