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

fix: enforce Kafka-only L1 Workbench events

Browse Source
This commit is contained in:
root
2026-07-18 06:58:21 +02:00
parent 43d1027cf0
commit fb2e0c0d38
3 changed files with 56 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/workbench/dispatcher.ts
+1 -1
View File
@@ -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
});
}
internal/workbench/native-agentrun-application.test.ts
+19 -17
View File
@@ -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<string, string> {
@@ -100,8 +107,6 @@ function agentRunEnvironment(): Record<string, string> {
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;
}
internal/workbench/native-agentrun-application.ts
+36 -86
View File
@@ -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<string, Record<string, any>>;
};
const TERMINAL = new Set(["completed", "failed", "blocked", "cancelled", "canceled"]);
export function createNativeAgentRunWorkbenchApplication(options: {
stateFile: string;
env: Record<string, string | undefined>;
@@ -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<NativeState> {
@@ -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 }); }