From fb2e0c0d38088c2223f144d0b5de473620f420e8 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 18 Jul 2026 06:58:21 +0200 Subject: [PATCH] fix: enforce Kafka-only L1 Workbench events --- internal/workbench/dispatcher.ts | 2 +- .../native-agentrun-application.test.ts | 36 +++--- .../workbench/native-agentrun-application.ts | 122 ++++++------------ 3 files changed, 56 insertions(+), 104 deletions(-) diff --git a/internal/workbench/dispatcher.ts b/internal/workbench/dispatcher.ts index e3a7059e..d3bf4a39 100644 --- a/internal/workbench/dispatcher.ts +++ b/internal/workbench/dispatcher.ts @@ -30,7 +30,7 @@ export function createWorkbenchDispatcher(options: { traceId, sessionId, ...workflow, - terminalAuthority: options.mode === "agentrun-native" ? "agentrun-command-result" : "hwlab.event.v1", + terminalAuthority: "hwlab.event.v1", resultSynthesized: false }); } diff --git a/internal/workbench/native-agentrun-application.test.ts b/internal/workbench/native-agentrun-application.test.ts index d2161616..4be12351 100644 --- a/internal/workbench/native-agentrun-application.test.ts +++ b/internal/workbench/native-agentrun-application.test.ts @@ -12,7 +12,7 @@ afterEach(async () => { await Promise.all(temporaryDirectories.splice(0).map((directory) => rm(directory, { force: true, recursive: true }))); }); -test("L1 AgentRun application projects a gpt.pika command through terminal response", async () => { +test("L1 AgentRun application dispatches gpt.pika without polling or projecting a terminal result", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-agentrun-")); temporaryDirectories.push(directory); const stateFile = path.join(directory, "state.json"); @@ -35,30 +35,37 @@ test("L1 AgentRun application projects a gpt.pika command through terminal respo const result = await app.dispatchTurn(admitted); expect(result).toMatchObject({ - status: "completed", - finalResponse: { text: "Hi from gpt.pika." }, + status: "running", + terminal: false, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false, agentRun: { runId: "run_l1_gpt_pika", commandId: "cmd_l1_gpt_pika", runnerJobId: "rjob_l1_gpt_pika", - completed: true, }, }); const state = JSON.parse(await readFile(stateFile, "utf8")); expect(state.turns[input.traceId]).toMatchObject({ - status: "completed", - terminal: true, + status: "running", + terminal: false, runId: "run_l1_gpt_pika", commandId: "cmd_l1_gpt_pika", runnerJobId: "rjob_l1_gpt_pika", - finalResponse: { text: "Hi from gpt.pika." }, }); - expect(state.sessions[session.sessionId]).toMatchObject({ status: "idle", providerProfile: "gpt.pika" }); - expect(state.sessions[session.sessionId].messages.at(-1)).toMatchObject({ role: "assistant", content: "Hi from gpt.pika." }); + expect(state.turns[input.traceId]).not.toHaveProperty("result"); + expect(state.turns[input.traceId]).not.toHaveProperty("finalResponse"); + expect(state.sessions[session.sessionId]).toMatchObject({ status: "running", providerProfile: "gpt.pika" }); + expect(state.sessions[session.sessionId].messages).toHaveLength(1); + expect(calls.map((call) => `${call.method} ${call.path}`)).toEqual([ + "POST /api/v1/sessions", + "POST /api/v1/runs", + "POST /api/v1/runs/run_l1_gpt_pika/commands", + ]); expect(calls.every((call) => call.authorization === "Bearer test-agentrun-key")).toBe(true); }); -test("L1 AgentRun application projects manager failures instead of leaving a running turn", async () => { +test("L1 AgentRun application leaves dispatch failures to Temporal instead of projecting a terminal snapshot", async () => { const directory = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-agentrun-failure-")); temporaryDirectories.push(directory); const stateFile = path.join(directory, "state.json"); @@ -76,8 +83,8 @@ test("L1 AgentRun application projects manager failures instead of leaving a run params: { message: "hi", providerProfile: "gpt.pika" }, }); - expect(await app.dispatchTurn(input)).toMatchObject({ status: "failed", terminal: true, error: { code: "manager-unavailable" } }); - expect((await app.snapshot!()).turns[input.traceId]).toMatchObject({ status: "failed", terminal: true, error: { code: "manager-unavailable" } }); + await expect(app.dispatchTurn(input)).rejects.toMatchObject({ code: "manager-unavailable" }); + expect((await app.snapshot!()).turns[input.traceId]).toMatchObject({ status: "admitted", terminal: false }); }); function agentRunEnvironment(): Record { @@ -100,8 +107,6 @@ function agentRunEnvironment(): Record { HWLAB_RUNTIME_LANE: "v03", HWLAB_RUNTIME_ENDPOINT_LOCKED: "1", HWLAB_CODE_AGENT_ASSEMBLED_RUNTIME: "1", - WORKBENCH_AGENTRUN_TERMINAL_TIMEOUT_MS: "1000", - WORKBENCH_AGENTRUN_POLL_INTERVAL_MS: "1", }; } @@ -130,9 +135,6 @@ function fakeAgentRunManager(calls: Array<{ method: string; path: string; author dispatchIntent: { id: "dispatch_l1_gpt_pika", state: "pending", runnerJobId: "rjob_l1_gpt_pika", attemptCount: 0, durable: true }, }); } - if (method === "GET" && url.pathname === "/api/v1/runs/run_l1_gpt_pika/commands/cmd_l1_gpt_pika/result") { - return send({ runId: "run_l1_gpt_pika", commandId: "cmd_l1_gpt_pika", terminalStatus: "completed", completed: true, reply: "Hi from gpt.pika." }); - } throw new Error(`unexpected AgentRun call ${method} ${url.pathname}`); }) as typeof fetch; } diff --git a/internal/workbench/native-agentrun-application.ts b/internal/workbench/native-agentrun-application.ts index 20f4a476..997541ce 100644 --- a/internal/workbench/native-agentrun-application.ts +++ b/internal/workbench/native-agentrun-application.ts @@ -7,8 +7,6 @@ import { describeAgentRunAdapterAvailability, submitAgentRunChatTurn, } from "../cloud/code-agent-agentrun-adapter.ts"; -import { agentRunJson, resolveAgentRunManagerUrl } from "../cloud/code-agent-agentrun-runtime.ts"; -import { agentRunResultToCodeAgentPayload } from "../cloud/code-agent-agentrun-result.ts"; import { createCodeAgentTraceStore } from "../cloud/code-agent-trace-store.ts"; import type { WorkbenchApplication, WorkbenchTurnInput } from "./contracts.ts"; @@ -17,8 +15,6 @@ type NativeState = { turns: Record>; }; -const TERMINAL = new Set(["completed", "failed", "blocked", "cancelled", "canceled"]); - export function createNativeAgentRunWorkbenchApplication(options: { stateFile: string; env: Record; @@ -101,43 +97,47 @@ export function createNativeAgentRunWorkbenchApplication(options: { }); }, async dispatchTurn(input: WorkbenchTurnInput) { - try { - const state = await load(options.stateFile); - const session = ownedSession(state, input.sessionId, input.actor.id); - const params = { - ...input.params, - sessionId: input.sessionId, - conversationId: session.conversationId, - projectId: session.projectId, - ownerUserId: input.actor.id, - ownerRole: input.actor.role ?? "user", - providerProfile: text(input.params.providerProfile) || session.providerProfile, - shortConnection: true, - }; - const admitted = await submitAgentRunChatTurn({ traceId: input.traceId, params, options: { env, fetchImpl }, traceStore }); - await projectRunning(admitted, input.traceId); - const terminal = await waitForTerminal(admitted, input.traceId); - await projectTerminal(terminal, input.traceId); - return terminal; - } catch (error: any) { - const failed = { - status: "failed", - terminal: true, - error: { code: text(error?.code) || "agentrun_native_dispatch_failed", message: text(error?.message) || String(error) }, - valuesPrinted: false, - }; - await projectTerminal(failed, input.traceId); - return failed; - } + const state = await load(options.stateFile); + const session = ownedSession(state, input.sessionId, input.actor.id); + const params = { + ...input.params, + sessionId: input.sessionId, + conversationId: session.conversationId, + projectId: session.projectId, + ownerUserId: input.actor.id, + ownerRole: input.actor.role ?? "user", + providerProfile: text(input.params.providerProfile) || session.providerProfile, + shortConnection: true, + }; + const admitted = await submitAgentRunChatTurn({ traceId: input.traceId, params, options: { env, fetchImpl }, traceStore }); + await projectRunning(admitted, input.traceId); + return { + status: "running", + terminal: false, + traceId: input.traceId, + sessionId: input.sessionId, + agentRun: admitted.agentRun, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false, + valuesPrinted: false, + }; }, async cancelTurn(input) { const state = await load(options.stateFile); const turn = state.turns[input.traceId]; if (!turn || turn.ownerUserId !== input.actor.id) throw codedError("turn_not_found", "turn is not visible to the actor"); - const canceled = await cancelAgentRunChatTurn({ traceId: input.traceId, currentResult: turn.result ?? null, options: { env, fetchImpl }, traceStore }); + const currentResult = turn.agentRun ? { status: "running", agentRun: turn.agentRun } : null; + const canceled = await cancelAgentRunChatTurn({ traceId: input.traceId, currentResult, options: { env, fetchImpl }, traceStore }); if (!canceled) throw codedError("turn_not_running", "turn has no active AgentRun command"); - await projectTerminal(canceled, input.traceId); - return canceled; + return { + status: "cancel-requested", + terminal: false, + traceId: input.traceId, + agentRun: canceled.agentRun, + terminalAuthority: "hwlab.event.v1", + resultSynthesized: false, + valuesPrinted: false, + }; }, }; @@ -151,58 +151,10 @@ export function createNativeAgentRunWorkbenchApplication(options: { runId: result.agentRun?.runId ?? null, commandId: result.agentRun?.commandId ?? null, runnerJobId: result.agentRun?.runnerJobId ?? null, - result, + agentRun: result.agentRun ?? null, }); }); } - - async function waitForTerminal(base: any, traceId: string) { - const managerUrl = resolveAgentRunManagerUrl(env, base.agentRun?.managerUrl); - const runId = requiredText(base.agentRun?.runId, "agentRun.runId"); - const commandId = requiredText(base.agentRun?.commandId, "agentRun.commandId"); - const timeoutMs = positiveInteger(env.WORKBENCH_AGENTRUN_TERMINAL_TIMEOUT_MS, 600_000); - const intervalMs = positiveInteger(env.WORKBENCH_AGENTRUN_POLL_INTERVAL_MS, 1_000); - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(runId)}/commands/${encodeURIComponent(commandId)}/result`, { - timeoutMs: Math.min(20_000, Math.max(1_000, deadline - Date.now())), - env, - }); - const mapped = agentRunResultToCodeAgentPayload({ base, result, traceStore, traceId }); - if (TERMINAL.has(normalizeStatus(mapped.status)) && (mapped.finalResponse || mapped.status !== "completed")) return mapped; - await Bun.sleep(intervalMs); - } - throw codedError("agentrun_terminal_timeout", `AgentRun command did not reach a terminal result within ${timeoutMs}ms`); - } - - async function projectTerminal(result: any, traceId: string) { - return mutate((state) => { - const turn = state.turns[traceId]; - if (!turn) throw codedError("turn_not_found", "turn was not found for terminal projection"); - const session = state.sessions[turn.sessionId]; - const status = normalizeStatus(result.status); - const timestamp = now(); - Object.assign(turn, { - status, - terminal: TERMINAL.has(status), - updatedAt: timestamp, - finishedAt: timestamp, - runId: result.agentRun?.runId ?? turn.runId ?? null, - commandId: result.agentRun?.commandId ?? turn.commandId ?? null, - runnerJobId: result.agentRun?.runnerJobId ?? turn.runnerJobId ?? null, - finalResponse: result.finalResponse ?? null, - error: result.error ?? null, - result, - }); - if (session) { - session.status = status === "completed" ? "idle" : status; - session.updatedAt = timestamp; - if (result.finalResponse?.text && !session.messages.some((message: any) => message.traceId === traceId && message.role === "assistant")) { - session.messages.push({ ...result.finalResponse, role: "assistant", content: result.finalResponse.text, traceId, status }); - } - } - }); - } } async function load(stateFile: string): Promise { @@ -214,6 +166,4 @@ function text(value: unknown) { return String(value ?? "").trim(); } function requiredText(value: unknown, field: string) { const result = text(value); if (!result) throw codedError("invalid_input", `${field} is required`); return result; } function validOptionalId(value: unknown, prefix: string) { const id = text(value); if (!id) return null; if (!id.startsWith(prefix)) throw codedError("invalid_input", `${prefix.slice(0, -1)} id is invalid`); return id; } function ownedSession(state: NativeState, sessionId: string, ownerUserId: string) { const session = state.sessions[sessionId]; if (!session || session.ownerUserId !== ownerUserId) throw codedError("session_not_found", "session is not visible to the actor"); return session; } -function normalizeStatus(value: unknown) { const status = text(value).toLowerCase().replaceAll("_", "-"); return status === "cancelled" ? "canceled" : status; } -function positiveInteger(value: unknown, fallback: number) { const parsed = Number.parseInt(text(value), 10); return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; } function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }