root
543 lines
24 KiB
TypeScript
543 lines
24 KiB
TypeScript
import { describe, expect, test } from "bun:test";
|
|
|
|
import type { WorkbenchApplication, WorkbenchTemporalGateway } from "./contracts.ts";
|
|
import { createWorkbenchActivities } from "./activities.ts";
|
|
import { createWorkbenchDispatcher } from "./dispatcher.ts";
|
|
import { createCloudWorkbenchApplication } from "./cloud-application.ts";
|
|
import { createWorkbenchHttpApp } from "./http.ts";
|
|
import { workbenchRuntime } from "./runtime.ts";
|
|
import { runWorkbenchCli } from "../../tools/src/workbench-cli.ts";
|
|
|
|
test("temporal API runtime does not require worker Kafka publisher configuration", async () => {
|
|
const runtime = workbenchRuntime({
|
|
WORKBENCH_MODE: "temporal",
|
|
WORKBENCH_CLOUD_API_URL: "http://cloud.internal",
|
|
WORKBENCH_CLOUD_API_AUTHORIZATION: "internal-key",
|
|
WORKBENCH_TEMPORAL_ADDRESS: "127.0.0.1:7233",
|
|
WORKBENCH_TEMPORAL_NAMESPACE: "unidesk",
|
|
WORKBENCH_TEMPORAL_TASK_QUEUE: "hwlab-v03-workbench",
|
|
WORKBENCH_ACTIVITY_START_TO_CLOSE_TIMEOUT_MS: "30000",
|
|
WORKBENCH_ACTIVITY_RETRY_INITIAL_INTERVAL_MS: "1000",
|
|
WORKBENCH_ACTIVITY_RETRY_MAXIMUM_INTERVAL_MS: "30000",
|
|
WORKBENCH_ACTIVITY_RETRY_MAXIMUM_ATTEMPTS: "5"
|
|
}, { eventPublisher: null });
|
|
expect(runtime.eventPublisher).toBeNull();
|
|
await runtime.close();
|
|
});
|
|
|
|
describe("Workbench dispatcher", () => {
|
|
test("submits through Temporal without synthesizing terminal state", async () => {
|
|
const calls: string[] = [];
|
|
const application: WorkbenchApplication = {
|
|
async health() { return { status: "ready" }; },
|
|
async createSession() { return { session: { sessionId: "ses_test" } }; },
|
|
async admitTurn(input) { calls.push("admit"); return input; },
|
|
async dispatchTurn() { throw new Error("API dispatcher must not dispatch directly"); },
|
|
async cancelTurn() { throw new Error("API dispatcher must not cancel directly"); }
|
|
};
|
|
const temporal: WorkbenchTemporalGateway = {
|
|
async submit(input) { calls.push("temporal"); return { workflowId: `workbench-turn-${input.traceId}`, workflowRunId: "run-1", reused: false }; },
|
|
async cancel() { return { workflowId: "cancel", workflowRunId: "run-2", reused: false }; },
|
|
async close() {}
|
|
};
|
|
const dispatch = createWorkbenchDispatcher({ application, temporal, mode: "temporal" });
|
|
const result = await dispatch({ operation: "turn.submit", actor: { id: "usr_test" }, traceId: "trc_test", params: { sessionId: "ses_test", message: "hi" } });
|
|
expect(calls).toEqual(["admit", "temporal"]);
|
|
expect(result).toMatchObject({ ok: true, mode: "temporal", data: { status: "running", traceId: "trc_test", terminalAuthority: "hwlab.event.v1", resultSynthesized: false } });
|
|
});
|
|
|
|
test("cancel is a durable Temporal command", async () => {
|
|
const application = stubApplication();
|
|
const temporal: WorkbenchTemporalGateway = {
|
|
async submit() { throw new Error("not used"); },
|
|
async cancel(input) { return { workflowId: `workbench-cancel-${input.traceId}`, workflowRunId: "run-cancel", reused: false }; },
|
|
async close() {}
|
|
};
|
|
const result = await createWorkbenchDispatcher({ application, temporal, mode: "temporal" })({ operation: "turn.cancel", actor: { id: "usr_test" }, traceId: "trc_test" });
|
|
expect(result).toMatchObject({ ok: true, data: { status: "cancel_requested", resultSynthesized: false } });
|
|
});
|
|
});
|
|
|
|
describe("Workbench Cloud application adapter", () => {
|
|
test("persists admission before dispatch through distinct internal routes", async () => {
|
|
const calls: Array<{ path: string; authorization: string | null }> = [];
|
|
const application = createCloudWorkbenchApplication({
|
|
baseUrl: "http://cloud.internal",
|
|
authorization: "unified-internal-key",
|
|
fetchImpl: (async (url: string | URL | Request, init?: RequestInit) => {
|
|
calls.push({ path: new URL(String(url)).pathname, authorization: new Headers(init?.headers).get("authorization") });
|
|
return Response.json({ ok: true, accepted: true, status: "admitted" });
|
|
}) as typeof fetch
|
|
});
|
|
const input = { actor: { id: "usr_test" }, traceId: "trc_test", sessionId: "ses_test", params: { sessionId: "ses_test", message: "hi" } };
|
|
await application.admitTurn(input);
|
|
await application.dispatchTurn(input);
|
|
expect(calls).toEqual([
|
|
{ path: "/v1/internal/workbench/admit", authorization: "Bearer unified-internal-key" },
|
|
{ path: "/v1/agent/chat", authorization: "Bearer unified-internal-key" }
|
|
]);
|
|
});
|
|
|
|
test("preserves an existing Bearer authorization scheme", async () => {
|
|
let authorization: string | null = null;
|
|
const application = createCloudWorkbenchApplication({
|
|
baseUrl: "http://cloud.internal",
|
|
authorization: "Bearer unified-internal-key",
|
|
fetchImpl: (async (_url: string | URL | Request, init?: RequestInit) => {
|
|
authorization = new Headers(init?.headers).get("authorization");
|
|
return Response.json({ ok: true, session: { sessionId: "ses_test" } });
|
|
}) as typeof fetch
|
|
});
|
|
await application.createSession({ actor: { id: "usr_test" }, params: {} });
|
|
expect(authorization).toBe("Bearer unified-internal-key");
|
|
});
|
|
|
|
test("preserves Cloud API retryability and billing failure semantics", async () => {
|
|
const application = createCloudWorkbenchApplication({
|
|
baseUrl: "http://cloud.internal",
|
|
authorization: "unified-internal-key",
|
|
fetchImpl: (async () => Response.json({
|
|
ok: false,
|
|
error: { code: "insufficient_credits", layer: "billing", retryable: false, message: "credits exhausted", userMessage: "余额不足" }
|
|
}, { status: 402 })) as typeof fetch
|
|
});
|
|
const input = { actor: { id: "usr_test" }, traceId: "trc_billing", sessionId: "ses_billing", params: { message: "hi" } };
|
|
await expect(application.dispatchTurn(input)).rejects.toMatchObject({
|
|
code: "insufficient_credits",
|
|
layer: "billing",
|
|
retryable: false,
|
|
status: 402,
|
|
userMessage: "余额不足"
|
|
});
|
|
});
|
|
});
|
|
|
|
describe("Workbench Temporal activities", () => {
|
|
const activity = { startToCloseTimeoutMs: 1000, retryInitialIntervalMs: 100, retryMaximumIntervalMs: 500, retryMaximumAttempts: 3 };
|
|
const input = { actor: { id: "usr_test" }, traceId: "trc_activity", sessionId: "ses_activity", params: { message: "hi" } };
|
|
|
|
test("publishes semantic Kafka failure and stops retrying non-retryable billing errors", async () => {
|
|
const published: any[] = [];
|
|
const application = stubApplication();
|
|
application.dispatchTurn = async () => { throw Object.assign(new Error("credits exhausted"), { code: "insufficient_credits", layer: "billing", retryable: false, userMessage: "余额不足" }); };
|
|
const activities = createWorkbenchActivities(application, {
|
|
activity,
|
|
attempt: () => 1,
|
|
eventPublisher: { async publish(event) { published.push(event); return { published: true }; } }
|
|
});
|
|
await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "insufficient_credits", nonRetryable: true });
|
|
expect(published).toHaveLength(2);
|
|
expect(published[0]).toMatchObject({ event: { type: "error", failureDomain: "billing", code: "insufficient_credits", retryable: false, willRetry: false, attempt: 1, maxAttempts: 3, terminal: false } });
|
|
expect(published[1]).toMatchObject({ event: { type: "terminal_status", terminalStatus: "failed", errorCode: "insufficient_credits", message: "余额不足", terminal: true } });
|
|
});
|
|
|
|
test("publishes the YAML-controlled exponential retry without sealing the turn early", async () => {
|
|
const published: any[] = [];
|
|
const application = stubApplication();
|
|
application.dispatchTurn = async () => { throw Object.assign(new Error("billing unavailable"), { code: "user_billing_preflight_failed", layer: "billing", retryable: true }); };
|
|
const activities = createWorkbenchActivities(application, {
|
|
activity,
|
|
attempt: () => 2,
|
|
eventPublisher: { async publish(event) { published.push(event); return { published: true }; } }
|
|
});
|
|
await expect(activities.dispatchWorkbenchTurn(input)).rejects.toMatchObject({ type: "user_billing_preflight_failed", nonRetryable: false });
|
|
expect(published).toHaveLength(1);
|
|
expect(published[0]).toMatchObject({ event: { type: "backend_status", retryPhase: "retryScheduled", attempt: 2, maxAttempts: 3, backoffMs: 200, terminal: false } });
|
|
});
|
|
});
|
|
|
|
describe("Workbench native HTTP adapter", () => {
|
|
test("native session list exposes the first user preview for summary-only rail loading", async () => {
|
|
const app = createWorkbenchHttpApp({
|
|
async dispatch() { return { ok: true }; },
|
|
async snapshot() {
|
|
return {
|
|
sessions: {
|
|
ses_title: {
|
|
sessionId: "ses_title",
|
|
updatedAt: "2026-07-20T01:00:00.000Z",
|
|
messages: [{ role: "user", content: "persisted native title", createdAt: "2026-07-20T00:59:00.000Z" }]
|
|
},
|
|
ses_empty: { sessionId: "ses_empty", updatedAt: "2026-07-20T00:00:00.000Z", messages: [] }
|
|
},
|
|
turns: {}
|
|
};
|
|
}
|
|
});
|
|
|
|
const response = await app.fetch(new Request("http://native.test/v1/workbench/sessions"));
|
|
expect(response.status).toBe(200);
|
|
expect(await response.json()).toMatchObject({
|
|
sessions: [
|
|
{ sessionId: "ses_title", firstUserMessagePreview: "persisted native title" },
|
|
{ sessionId: "ses_empty", firstUserMessagePreview: null }
|
|
]
|
|
});
|
|
});
|
|
|
|
test("AgentRun native session list uses Kafka index summaries instead of admission messages", async () => {
|
|
const app = createWorkbenchHttpApp({
|
|
mode: "agentrun-native",
|
|
kafkaEventBridge: {
|
|
workbenchSessionSummaries() {
|
|
return { ok: true, sessions: [{ sessionId: "ses_title", firstUserMessagePreview: "Kafka title", lastUserMessageAt: "2026-07-20T01:00:00.000Z" }] };
|
|
}
|
|
},
|
|
async dispatch() { return { ok: true }; },
|
|
async snapshot() {
|
|
return {
|
|
sessions: {
|
|
ses_title: {
|
|
sessionId: "ses_title",
|
|
updatedAt: "2026-07-20T00:00:00.000Z",
|
|
messages: [{ role: "user", content: "admission snapshot title" }]
|
|
}
|
|
},
|
|
turns: {}
|
|
};
|
|
}
|
|
});
|
|
|
|
const response = await app.fetch(new Request("http://native.test/v1/workbench/sessions"));
|
|
expect(await response.json()).toMatchObject({
|
|
sessions: [{ sessionId: "ses_title", firstUserMessagePreview: "Kafka title", lastUserMessageAt: "2026-07-20T01:00:00.000Z" }]
|
|
});
|
|
});
|
|
|
|
test("authorizes only the Workbench navigation entry", async () => {
|
|
const app = createWorkbenchHttpApp({
|
|
async dispatch() { return { ok: true }; },
|
|
async snapshot() { return { sessions: {}, turns: {} }; }
|
|
});
|
|
|
|
for (const path of ["/auth/session", "/auth/bootstrap"]) {
|
|
const response = await app.fetch(new Request(`http://native.test${path}`));
|
|
expect(response.status).toBe(200);
|
|
expect(await response.json()).toMatchObject({
|
|
authenticated: true,
|
|
access: { nav: { profileId: "workbench-native", allowedIds: ["workbench.code"] } }
|
|
});
|
|
}
|
|
});
|
|
|
|
test("serves read-only empty states for Workbench auxiliary panels", async () => {
|
|
const app = createWorkbenchHttpApp({
|
|
async dispatch() { return { ok: true }; },
|
|
async snapshot() { return { sessions: {}, turns: {} }; }
|
|
});
|
|
|
|
for (const path of ["/v1/caserun/cases", "/v1/hwpod/specs?probe=1", "/v1/hwpod-node-ops"]) {
|
|
const response = await app.fetch(new Request(`http://native.test${path}`));
|
|
expect(response.status).toBe(200);
|
|
expect(await response.json()).toMatchObject({ ok: true, status: "unavailable", mode: "native-test" });
|
|
}
|
|
const performance = await app.fetch(new Request("http://native.test/v1/web-performance", { method: "POST", body: "{}" }));
|
|
expect(performance.status).toBe(202);
|
|
expect(await performance.json()).toMatchObject({ ok: true, status: "accepted", persisted: false, 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: {} }; }
|
|
});
|
|
let authorization: string | null = null;
|
|
const server = Bun.serve({
|
|
port: 0,
|
|
fetch(request) {
|
|
authorization = request.headers.get("authorization");
|
|
return 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"
|
|
], { HWLAB_API_KEY: "l0-sse-api-key" });
|
|
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,
|
|
connectedContract: {
|
|
deliverySemantics: "live-only",
|
|
realtimeSource: "hwlab.event.v1",
|
|
replay: false,
|
|
liveOnly: true,
|
|
refreshReplay: null
|
|
},
|
|
eventCount: 1,
|
|
eventTypes: ["backend"]
|
|
});
|
|
expect(authorization).toBe("Bearer l0-sse-api-key");
|
|
} finally {
|
|
server.stop(true);
|
|
await app.close();
|
|
}
|
|
});
|
|
|
|
test("L0 CLI reports the Workbench SSE HTTP status when authentication fails", async () => {
|
|
const server = Bun.serve({ port: 0, fetch: () => new Response("unauthorized", { status: 401 }) });
|
|
try {
|
|
await expect(runWorkbenchCli([
|
|
"events", "inspect",
|
|
"--trace-id", "trc_l0_auth_failure",
|
|
"--over-api",
|
|
"--api-url", `http://127.0.0.1:${server.port}`,
|
|
"--timeout-ms", "1000"
|
|
], {})).rejects.toMatchObject({
|
|
code: "workbench_sse_read_failed",
|
|
message: expect.stringContaining("Workbench SSE returned HTTP 401")
|
|
});
|
|
} finally {
|
|
server.stop(true);
|
|
}
|
|
});
|
|
|
|
test("L0 CLI replays retained Kafka events before the same live SSE stream", async () => {
|
|
const sessionId = "ses_l0_refresh";
|
|
const traceId = "trc_l0_refresh";
|
|
const bridge = {
|
|
capabilities: { liveKafkaSse: true, kafkaRefreshReplay: true },
|
|
refreshReplay: { groupIdPrefix: "hwlab-l0-refresh", timeoutMs: 1000, scanLimit: 100, matchedEventLimit: 20, liveBufferLimit: 20 },
|
|
ready: Promise.resolve(),
|
|
subscribeLiveHwlabEvents() { return () => {}; },
|
|
async queryHwlabEventRetention(options: any) {
|
|
expect(options).toMatchObject({ sessionId, traceId, partitionKey: sessionId, fromBeginning: true });
|
|
return {
|
|
topic: "hwlab.event.v1",
|
|
events: Array.from({ length: 100 }, (_, index) => ({
|
|
topic: "hwlab.event.v1",
|
|
partition: 0,
|
|
offset: String(index),
|
|
value: {
|
|
schema: "hwlab.event.v1",
|
|
eventType: "hwlab.trace.event.projected",
|
|
eventId: `hwlab:evt_l0_refresh_${index}`,
|
|
sourceEventId: `evt_l0_refresh_${index}`,
|
|
sessionId,
|
|
hwlabSessionId: sessionId,
|
|
traceId,
|
|
event: { type: "backend", eventType: "backend", sessionId, traceId, sourceEventId: `evt_l0_refresh_${index}` }
|
|
}
|
|
})),
|
|
completionReason: "end-offset",
|
|
completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true },
|
|
reachedEndOffsets: true,
|
|
endOffsetsAvailable: true,
|
|
endOffsets: [{ partition: 0, startOffset: "0", endOffset: "100" }],
|
|
scannedCount: 100,
|
|
parsedCount: 100,
|
|
matchedCount: 100,
|
|
filterRejectedCount: 0,
|
|
keyRejectedCount: 0,
|
|
partitionKeyScoped: true,
|
|
targetPartition: 0,
|
|
timing: { endOffsetSnapshotMs: 0, groupOffsetsMs: 0, consumerConnectMs: 0, consumerSubscribeMs: 0, scanMs: 0, cleanupMs: 0, indexLookupMs: 1, totalMs: 1 },
|
|
index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, valuesRedacted: true }
|
|
};
|
|
}
|
|
};
|
|
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 result = await runWorkbenchCli([
|
|
"events", "inspect",
|
|
"--session-id", sessionId,
|
|
"--trace-id", traceId,
|
|
"--over-api",
|
|
"--api-url", `http://127.0.0.1:${server.port}`,
|
|
"--timeout-ms", "1000"
|
|
], {});
|
|
expect(result).toMatchObject({
|
|
ok: true,
|
|
connected: true,
|
|
connectedContract: {
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
realtimeSource: "hwlab.event.v1",
|
|
replay: true,
|
|
liveOnly: false,
|
|
refreshReplay: {
|
|
phase: "live",
|
|
counts: { replayed: 100 },
|
|
query: {
|
|
complete: true,
|
|
scannedCount: 100,
|
|
parsedCount: 100,
|
|
matchedCount: 100,
|
|
filterRejectedCount: 0,
|
|
keyRejectedCount: 0,
|
|
partitionKeyScoped: true,
|
|
targetPartition: 0,
|
|
index: { hit: true, source: "process-session-index", indexedEventCount: 300, indexedSessionCount: 3, incompleteSessionCount: 0, blocking: false },
|
|
timing: { scanMs: 0, indexLookupMs: 1, totalMs: 1 }
|
|
}
|
|
}
|
|
},
|
|
eventCount: 100,
|
|
eventTypes: ["backend"],
|
|
eventTypeCounts: { backend: 100 }
|
|
});
|
|
expect(JSON.stringify(result).length).toBeLessThan(10_000);
|
|
} finally {
|
|
server.stop(true);
|
|
await app.close();
|
|
}
|
|
});
|
|
|
|
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", createdAt: "2026-07-19T16:00:00.000Z" },
|
|
{ type: "backend", eventType: "backend", createdAt: "2026-07-19T16:00:02.000Z" },
|
|
{ type: "assistant", eventType: "assistant", createdAt: "2026-07-19T16:01:02.000Z", final: true, replyAuthority: true, finalResponse: { text: "done" } },
|
|
{ type: "result", eventType: "terminal", createdAt: "2026-07-19T16:01:03.000Z", 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"],
|
|
timing: {
|
|
clock: "agentrun-source-event-created-at",
|
|
phasesMs: {
|
|
firstEventToBackend: 2000,
|
|
admissionAndDispatch: null,
|
|
runnerProvisioning: null,
|
|
runtimePreparation: null,
|
|
providerExecution: null,
|
|
backendToAssistant: 60000,
|
|
assistantToTerminal: 1000,
|
|
firstEventToTerminal: 63000,
|
|
firstEventToFinal: 62000
|
|
},
|
|
slowestGap: {
|
|
durationMs: 60000,
|
|
toIndex: 2,
|
|
toEventType: "assistant"
|
|
},
|
|
valuesPrinted: false
|
|
}
|
|
});
|
|
} 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", "250",
|
|
"--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({
|
|
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 {
|
|
return {
|
|
async health() { return {}; },
|
|
async createSession() { return {}; },
|
|
async admitTurn(input) { return input; },
|
|
async dispatchTurn() { return {}; },
|
|
async cancelTurn() { return {}; }
|
|
};
|
|
}
|