From 3bb7b67a4e8bfa4ac820b198040d746b21d8d93d Mon Sep 17 00:00:00 2001 From: root Date: Sat, 18 Jul 2026 03:49:47 +0200 Subject: [PATCH 1/2] fix: support L1 Workbench AgentRun activity dispatch --- internal/cloud/workbench-command-proxy.test.ts | 9 +++++++++ internal/cloud/workbench-command-proxy.ts | 6 +++++- 2 files changed, 14 insertions(+), 1 deletion(-) create mode 100644 internal/cloud/workbench-command-proxy.test.ts diff --git a/internal/cloud/workbench-command-proxy.test.ts b/internal/cloud/workbench-command-proxy.test.ts new file mode 100644 index 00000000..298c66e3 --- /dev/null +++ b/internal/cloud/workbench-command-proxy.test.ts @@ -0,0 +1,9 @@ +import { expect, test } from "bun:test"; + +import { workbenchActivityDispatchRequested } from "./workbench-command-proxy.ts"; + +test("accepts activity dispatch header from Node and Fetch request shapes", () => { + expect(workbenchActivityDispatchRequested({ headers: { "x-workbench-activity-dispatch": "1" } })).toBe(true); + expect(workbenchActivityDispatchRequested({ headers: new Headers({ "x-workbench-activity-dispatch": "1" }) })).toBe(true); + expect(workbenchActivityDispatchRequested({ headers: new Headers({ "x-workbench-activity-dispatch": "0" }) })).toBe(false); +}); diff --git a/internal/cloud/workbench-command-proxy.ts b/internal/cloud/workbench-command-proxy.ts index 89931233..f24e1292 100644 --- a/internal/cloud/workbench-command-proxy.ts +++ b/internal/cloud/workbench-command-proxy.ts @@ -5,7 +5,11 @@ export function workbenchTemporalProxyEnabled(env: Record Date: Sat, 18 Jul 2026 04:41:49 +0200 Subject: [PATCH 2/2] fix: connect L1 Workbench to AgentRun v0.2 --- cmd/hwlab-workbench-api/main.ts | 2 +- cmd/hwlab-workbench-worker/main.ts | 2 +- internal/cloud/code-agent-agentrun-runtime.ts | 10 +- internal/workbench/contracts.ts | 4 +- internal/workbench/dispatcher.ts | 8 +- internal/workbench/http.ts | 43 ++-- .../native-agentrun-application.test.ts | 138 +++++++++++ .../workbench/native-agentrun-application.ts | 219 ++++++++++++++++++ internal/workbench/runtime.ts | 7 +- tools/src/workbench-native-service.ts | 2 +- 10 files changed, 404 insertions(+), 31 deletions(-) create mode 100644 internal/workbench/native-agentrun-application.test.ts create mode 100644 internal/workbench/native-agentrun-application.ts diff --git a/cmd/hwlab-workbench-api/main.ts b/cmd/hwlab-workbench-api/main.ts index 039d398d..f73f659a 100644 --- a/cmd/hwlab-workbench-api/main.ts +++ b/cmd/hwlab-workbench-api/main.ts @@ -3,7 +3,7 @@ import { createWorkbenchHttpApp } from "../../internal/workbench/http.ts"; import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; const runtime = workbenchRuntime(); -const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, close: runtime.close }); +const app = createWorkbenchHttpApp({ dispatch: runtime.dispatch, snapshot: runtime.application.snapshot?.bind(runtime.application), authorization: process.env.WORKBENCH_API_AUTHORIZATION, mode: runtime.mode, close: runtime.close }); const host = process.env.WORKBENCH_API_HOST || "0.0.0.0"; const port = positivePort(process.env.WORKBENCH_API_PORT || process.env.PORT, 6677); const server = Bun.serve({ hostname: host, port, fetch: (request) => app.fetch(request) }); diff --git a/cmd/hwlab-workbench-worker/main.ts b/cmd/hwlab-workbench-worker/main.ts index 95a40b77..8d898452 100644 --- a/cmd/hwlab-workbench-worker/main.ts +++ b/cmd/hwlab-workbench-worker/main.ts @@ -6,7 +6,7 @@ import { createWorkbenchActivities } from "../../internal/workbench/activities.t import { workbenchRuntime } from "../../internal/workbench/runtime.ts"; const runtime = workbenchRuntime(); -if (runtime.mode !== "temporal") throw Object.assign(new Error("hwlab-workbench-worker requires WORKBENCH_MODE=temporal"), { code: "workbench_worker_temporal_mode_required" }); +if (runtime.mode === "native-test") throw Object.assign(new Error("hwlab-workbench-worker requires WORKBENCH_MODE=agentrun-native or temporal"), { code: "workbench_worker_temporal_mode_required" }); const connection = await NativeConnection.connect({ address: runtime.temporalAddress }); const worker = await Worker.create({ connection, diff --git a/internal/cloud/code-agent-agentrun-runtime.ts b/internal/cloud/code-agent-agentrun-runtime.ts index 666da70d..71e9fd87 100644 --- a/internal/cloud/code-agent-agentrun-runtime.ts +++ b/internal/cloud/code-agent-agentrun-runtime.ts @@ -3,6 +3,7 @@ import { backendPerformanceRouteTemplate, currentBackendPerformanceContext } from "./backend-performance.ts"; import { truthyFlag } from "./server-http-utils.ts"; +import { isIP } from "node:net"; export const DEFAULT_AGENTRUN_MGR_URL = "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080"; @@ -65,7 +66,8 @@ export function resolveAgentRunManagerUrl(env = process.env, override = null) { const url = new URL(raw); const host = url.hostname.toLowerCase(); const allowedByTest = truthyFlag(env.HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL) && ["127.0.0.1", "localhost"].includes(host); - if (url.protocol !== "http:" || (!isAgentRunManagerInternalServiceHost(host) && !allowedByTest)) { + const allowedByNative = env.WORKBENCH_MODE === "agentrun-native" && truthyFlag(env.HWLAB_CODE_AGENT_AGENTRUN_ALLOW_PRIVATE_URL) && isPrivateIp(host); + if (url.protocol !== "http:" || (!isAgentRunManagerInternalServiceHost(host) && !allowedByTest && !allowedByNative)) { throw Object.assign(new Error(`AGENTRUN_MGR_URL must use internal k3s Service DNS agentrun-mgr..svc.cluster.local; got ${redactUrl(raw)}`), { code: "agentrun_internal_url_required", statusCode: 500 @@ -77,6 +79,12 @@ export function resolveAgentRunManagerUrl(env = process.env, override = null) { return url.toString().replace(/\/+$/u, ""); } +function isPrivateIp(host: string) { + if (isIP(host) !== 4) return false; + const [a, b] = host.split(".").map(Number); + return a === 10 || a === 127 || (a === 172 && b >= 16 && b <= 31) || (a === 192 && b === 168); +} + export function isAgentRunManagerInternalServiceHost(host) { return AGENTRUN_MANAGER_HOST_PATTERN.test(String(host ?? "").trim().toLowerCase()); } diff --git a/internal/workbench/contracts.ts b/internal/workbench/contracts.ts index a355cd2f..e50a387a 100644 --- a/internal/workbench/contracts.ts +++ b/internal/workbench/contracts.ts @@ -3,6 +3,8 @@ export type WorkbenchActor = { role?: string | null; }; +export type WorkbenchMode = "native-test" | "agentrun-native" | "temporal"; + export type WorkbenchCommand = | { operation: "health" } | { operation: "session.create"; actor: WorkbenchActor; params: Record } @@ -12,7 +14,7 @@ export type WorkbenchCommand = export type WorkbenchCommandResult = { ok: boolean; operation: WorkbenchCommand["operation"]; - mode: "native-test" | "temporal"; + mode: WorkbenchMode; data?: Record; error?: { code: string; message: string; details?: unknown }; }; diff --git a/internal/workbench/dispatcher.ts b/internal/workbench/dispatcher.ts index dbdc36ff..e3a7059e 100644 --- a/internal/workbench/dispatcher.ts +++ b/internal/workbench/dispatcher.ts @@ -1,6 +1,6 @@ import { randomUUID } from "node:crypto"; -import type { WorkbenchApplication, WorkbenchCommand, WorkbenchCommandResult, WorkbenchTemporalGateway } from "./contracts.ts"; +import type { WorkbenchApplication, WorkbenchCommand, WorkbenchCommandResult, WorkbenchMode, WorkbenchTemporalGateway } from "./contracts.ts"; const SESSION_ID = /^ses_[A-Za-z0-9._:-]{3,180}$/u; const TRACE_ID = /^trc_[A-Za-z0-9._:-]{3,180}$/u; @@ -8,7 +8,7 @@ const TRACE_ID = /^trc_[A-Za-z0-9._:-]{3,180}$/u; export function createWorkbenchDispatcher(options: { application: WorkbenchApplication; temporal: WorkbenchTemporalGateway; - mode: "native-test" | "temporal"; + mode: WorkbenchMode; }) { return async function dispatch(command: WorkbenchCommand): Promise { try { @@ -30,7 +30,7 @@ export function createWorkbenchDispatcher(options: { traceId, sessionId, ...workflow, - terminalAuthority: "hwlab.event.v1", + terminalAuthority: options.mode === "agentrun-native" ? "agentrun-command-result" : "hwlab.event.v1", resultSynthesized: false }); } @@ -55,7 +55,7 @@ export function createWorkbenchDispatcher(options: { }; } -function success(operation: WorkbenchCommand["operation"], mode: "native-test" | "temporal", data: Record): WorkbenchCommandResult { +function success(operation: WorkbenchCommand["operation"], mode: WorkbenchMode, data: Record): WorkbenchCommandResult { return { ok: true, operation, mode, data }; } diff --git a/internal/workbench/http.ts b/internal/workbench/http.ts index e6fce546..c3944742 100644 --- a/internal/workbench/http.ts +++ b/internal/workbench/http.ts @@ -1,6 +1,6 @@ -import type { WorkbenchCommand } from "./contracts.ts"; +import type { WorkbenchCommand, WorkbenchMode } from "./contracts.ts"; -export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise; snapshot?: () => Promise<{ sessions: Record>; turns: Record> }>; authorization?: string; close?: () => Promise }) { +export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchCommand) => Promise; snapshot?: () => Promise<{ sessions: Record>; turns: Record> }>; authorization?: string; mode?: WorkbenchMode; close?: () => Promise }) { return { async fetch(request: Request): Promise { try { @@ -8,16 +8,19 @@ export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchC if (url.pathname === "/health/live") return json(200, { ok: true, serviceId: "hwlab-workbench-api", status: "live" }); if (url.pathname === "/health/ready") return resultResponse(await options.dispatch({ operation: "health" })); if (options.snapshot) { - if (url.pathname === "/auth/session" || url.pathname === "/auth/bootstrap") return json(200, nativeAuthSession()); - if (url.pathname === "/auth/login" && request.method === "POST") return Response.json(nativeAuthSession(), { status: 200, headers: { "set-cookie": "hwlab_session=native-test; Path=/; HttpOnly; SameSite=Lax", "cache-control": "no-store" } }); + if (url.pathname === "/auth/session" || url.pathname === "/auth/bootstrap") return json(200, nativeAuthSession(options.mode)); + if (url.pathname === "/auth/login" && request.method === "POST") return Response.json(nativeAuthSession(options.mode), { status: 200, headers: { "set-cookie": "hwlab_session=native-test; Path=/; HttpOnly; SameSite=Lax", "cache-control": "no-store" } }); if (url.pathname === "/v1/users/me") return json(200, { ok: true, actor: { id: "usr_native", role: "user" }, authMethod: "native-test" }); if (url.pathname === "/v1/access/status") return json(200, { ok: true, status: "active", actor: { id: "usr_native", role: "user" }, nav: ["workbench.code"] }); - if (url.pathname === "/v1/provider-profiles") return json(200, { ok: true, items: [{ profile: "native-test", backendProfile: "native-test", configured: true }], count: 1, mode: "native-test" }); - if (url.pathname === "/v1/dashboard/summary") return json(200, { ok: true, status: "ready", mode: "native-test" }); - if (url.pathname === "/v1/caserun/cases") return json(200, { ok: true, status: "unavailable", cases: [], count: 0, mode: "native-test" }); - if (url.pathname === "/v1/hwpod/specs") return json(200, { ok: true, status: "unavailable", specs: [], mode: "native-test" }); - if (url.pathname === "/v1/hwpod-node-ops") return json(200, { ok: true, status: "unavailable", events: [], groups: [], mode: "native-test" }); - if (url.pathname === "/v1/workbench/events" && request.method === "GET") return nativeEventStream(); + if (url.pathname === "/v1/provider-profiles") { + const profile = options.mode === "agentrun-native" ? "gpt.pika" : "native-test"; + return json(200, { ok: true, items: [{ profile, backendProfile: profile, configured: true }], count: 1, mode: options.mode ?? "native-test" }); + } + if (url.pathname === "/v1/dashboard/summary") return json(200, { ok: true, status: "ready", mode: options.mode ?? "native-test" }); + 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/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"); @@ -50,13 +53,13 @@ export function createWorkbenchHttpApp(options: { dispatch: (command: WorkbenchC }; } -function nativeAuthSession() { +function nativeAuthSession(mode: WorkbenchMode = "native-test") { return { ok: true, authenticated: true, actor: { id: "usr_native", role: "user" }, access: { nav: { profileId: "workbench-native", allowedIds: ["workbench.code"], valuesRedacted: true } }, - mode: "native-test", + mode, valuesRedacted: true }; } @@ -68,30 +71,30 @@ function requireAuthorization(request: Request, options: { snapshot?: unknown; a if (!expected || actual !== expected) throw Object.assign(new Error("Workbench API internal authentication failed"), { code: "auth_required" }); } -async function nativeSessionList(options: { snapshot?: () => Promise }, url: URL) { +async function nativeSessionList(options: { snapshot?: () => Promise; mode?: WorkbenchMode }, url: URL) { const state = await requiredSnapshot(options); const owner = text(url.searchParams.get("ownerUserId")); const sessions = Object.values(state.sessions).filter((session: any) => !owner || session.ownerUserId === owner).sort((left: any, right: any) => String(right.updatedAt).localeCompare(String(left.updatedAt))); - return json(200, { ok: true, status: "ready", sessions, count: sessions.length, mode: "native-test" }); + return json(200, { ok: true, status: "ready", sessions, count: sessions.length, mode: options.mode ?? "native-test" }); } -async function nativeSessionDetail(options: { snapshot?: () => Promise }, sessionId: string, messagesOnly: boolean) { +async function nativeSessionDetail(options: { snapshot?: () => Promise; mode?: WorkbenchMode }, sessionId: string, messagesOnly: boolean) { const state = await requiredSnapshot(options); const session = state.sessions[sessionId]; if (!session) return json(404, { ok: false, error: { code: "session_not_found", message: "native session was not found" } }); - return json(200, messagesOnly ? { ok: true, sessionId, messages: session.messages ?? [], count: Array.isArray(session.messages) ? session.messages.length : 0, mode: "native-test" } : { ok: true, session, mode: "native-test" }); + return json(200, messagesOnly ? { ok: true, sessionId, messages: session.messages ?? [], count: Array.isArray(session.messages) ? session.messages.length : 0, mode: options.mode ?? "native-test" } : { ok: true, session, mode: options.mode ?? "native-test" }); } -async function nativeTurn(options: { snapshot?: () => Promise }, traceId: string) { +async function nativeTurn(options: { snapshot?: () => Promise; mode?: WorkbenchMode }, traceId: string) { const state = await requiredSnapshot(options); const turn = state.turns[traceId]; - return turn ? json(200, { ok: true, ...turn, mode: "native-test" }) : json(404, { ok: false, error: { code: "turn_not_found", message: "native turn was not found" } }); + 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 }) { if (!options.snapshot) throw Object.assign(new Error("native projection is unavailable"), { code: "native_projection_unavailable" }); return options.snapshot(); } -function nativeEventStream() { +function nativeEventStream(mode: WorkbenchMode = "native-test") { const encoder = new TextEncoder(); let heartbeat: ReturnType | undefined; return new Response(new ReadableStream({ start(controller) { - controller.enqueue(encoder.encode(`event: capability\ndata: ${JSON.stringify({ mode: "native-test", terminalAuthority: "fixture" })}\n\n`)); + controller.enqueue(encoder.encode(`event: capability\ndata: ${JSON.stringify({ mode, terminalAuthority: mode === "agentrun-native" ? "agentrun-command-result" : "fixture" })}\n\n`)); heartbeat = setInterval(() => controller.enqueue(encoder.encode(`: native-test-heartbeat ${Date.now()}\n\n`)), 10_000); }, cancel() { if (heartbeat) clearInterval(heartbeat); } diff --git a/internal/workbench/native-agentrun-application.test.ts b/internal/workbench/native-agentrun-application.test.ts new file mode 100644 index 00000000..d2161616 --- /dev/null +++ b/internal/workbench/native-agentrun-application.test.ts @@ -0,0 +1,138 @@ +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, expect, test } from "bun:test"; + +import { createNativeAgentRunWorkbenchApplication } from "./native-agentrun-application.ts"; + +const temporaryDirectories: string[] = []; + +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 () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-agentrun-")); + temporaryDirectories.push(directory); + const stateFile = path.join(directory, "state.json"); + const calls: Array<{ method: string; path: string; authorization: string | null; body: any }> = []; + const app = createNativeAgentRunWorkbenchApplication({ + stateFile, + env: agentRunEnvironment(), + fetchImpl: fakeAgentRunManager(calls), + }); + + const created = await app.createSession({ actor: { id: "usr_native", role: "user" }, params: { providerProfile: "gpt.pika" } }); + const session = created.session as Record; + const input = { + actor: { id: "usr_native", role: "user" }, + traceId: "trc_l1_gpt_pika_smoke", + sessionId: session.sessionId, + params: { message: "hi", providerProfile: "gpt.pika" }, + }; + const admitted = await app.admitTurn(input); + const result = await app.dispatchTurn(admitted); + + expect(result).toMatchObject({ + status: "completed", + finalResponse: { text: "Hi from gpt.pika." }, + 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, + 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(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 () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "hwlab-workbench-agentrun-failure-")); + temporaryDirectories.push(directory); + const stateFile = path.join(directory, "state.json"); + const app = createNativeAgentRunWorkbenchApplication({ + stateFile, + env: agentRunEnvironment(), + fetchImpl: (async () => Response.json({ ok: false, failureKind: "manager-unavailable", message: "manager unavailable" }, { status: 503 })) as typeof fetch, + }); + const created = await app.createSession({ actor: { id: "usr_native" }, params: { providerProfile: "gpt.pika" } }); + const session = created.session as Record; + const input = await app.admitTurn({ + actor: { id: "usr_native" }, + traceId: "trc_l1_manager_failure", + sessionId: session.sessionId, + 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" } }); +}); + +function agentRunEnvironment(): Record { + return { + WORKBENCH_MODE: "agentrun-native", + AGENTRUN_MGR_URL: "http://127.0.0.1:65535", + AGENTRUN_API_KEY: "test-agentrun-key", + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "NC01", + HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567", + HWLAB_CODE_AGENT_AGENTRUN_REPO_URL: "http://git-mirror-http.devops-infra.svc.cluster.local/pikasTech/HWLAB.git", + HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "gpt.pika", + HWLAB_CODE_AGENT_AGENTRUN_RUNNER_NAMESPACE: "agentrun-v02", + HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_MAX: "1", + HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_BASE_MS: "1", + HWLAB_CODE_AGENT_AGENTRUN_DISPATCH_RETRY_MAX_DELAY_MS: "1", + HWLAB_RUNTIME_API_URL: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667", + HWLAB_RUNTIME_WEB_URL: "http://hwlab-cloud-web.hwlab-v03.svc.cluster.local:8080", + HWLAB_RUNTIME_NAMESPACE: "hwlab-v03", + 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", + }; +} + +function fakeAgentRunManager(calls: Array<{ method: string; path: string; authorization: string | null; body: any }>): typeof fetch { + return (async (input: string | URL | Request, init: RequestInit = {}) => { + const url = new URL(String(input)); + const method = init.method ?? "GET"; + const body = init.body ? JSON.parse(String(init.body)) : null; + calls.push({ method, path: url.pathname, authorization: new Headers(init.headers).get("authorization"), body }); + const send = (data: unknown) => Response.json({ ok: true, data }); + if (method === "POST" && url.pathname === "/api/v1/sessions") { + return send({ session: { id: "ses_agentrun_l1_gpt_pika", metadata: {} } }); + } + if (method === "POST" && url.pathname === "/api/v1/runs") { + expect(body.backendProfile).toBe("gpt-pika"); + return send({ id: "run_l1_gpt_pika", status: "pending", sessionRef: body.sessionRef, backendProfile: body.backendProfile }); + } + if (method === "POST" && url.pathname === "/api/v1/runs/run_l1_gpt_pika/commands") { + expect(body.payload.prompt).toBe("hi"); + return send({ + id: "cmd_l1_gpt_pika", + runId: "run_l1_gpt_pika", + state: "pending", + type: "turn", + seq: 1, + 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 new file mode 100644 index 00000000..20f4a476 --- /dev/null +++ b/internal/workbench/native-agentrun-application.ts @@ -0,0 +1,219 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, rename, writeFile } from "node:fs/promises"; +import path from "node:path"; + +import { + cancelAgentRunChatTurn, + 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"; + +type NativeState = { + sessions: Record>; + turns: Record>; +}; + +const TERMINAL = new Set(["completed", "failed", "blocked", "cancelled", "canceled"]); + +export function createNativeAgentRunWorkbenchApplication(options: { + stateFile: string; + env: Record; + fetchImpl?: typeof fetch; + now?: () => string; +}): WorkbenchApplication { + const env = options.env; + const fetchImpl = options.fetchImpl ?? fetch; + const now = options.now ?? (() => new Date().toISOString()); + const traceStore = createCodeAgentTraceStore(); + const mutate = async (fn: (state: NativeState) => T | Promise) => { + const state = await load(options.stateFile); + const result = await fn(state); + await mkdir(path.dirname(options.stateFile), { recursive: true }); + const temporary = `${options.stateFile}.${process.pid}.tmp`; + await writeFile(temporary, `${JSON.stringify(state, null, 2)}\n`, "utf8"); + await rename(temporary, options.stateFile); + return result; + }; + + return { + async snapshot() { return load(options.stateFile); }, + async health() { + const state = await load(options.stateFile); + const availability = describeAgentRunAdapterAvailability(env, { params: { providerProfile: env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE } }); + return { + serviceId: "hwlab-workbench-native-agentrun", + status: availability.ready ? "ready" : "blocked", + mode: "agentrun-native", + providerProfile: text(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE), + agentRun: availability.agentRun, + sessionCount: Object.keys(state.sessions).length, + turnCount: Object.keys(state.turns).length, + valuesPrinted: false, + }; + }, + async createSession({ actor, params }) { + return mutate((state) => { + const sessionId = validOptionalId(params.sessionId, "ses_") ?? `ses_${randomUUID()}`; + const conversationId = validOptionalId(params.conversationId, "cnv_") ?? `cnv_${randomUUID()}`; + const timestamp = now(); + const session = { + sessionId, + conversationId, + projectId: text(params.projectId) || "pikasTech/HWLAB", + ownerUserId: actor.id, + ownerRole: actor.role ?? "user", + status: "idle", + providerProfile: text(params.providerProfile) || text(env.HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE) || "gpt.pika", + createdAt: timestamp, + updatedAt: timestamp, + mode: "agentrun-native", + messages: [] as Record[], + }; + state.sessions[sessionId] = session; + return { status: "created", session, valuesRedacted: true, secretMaterialStored: false }; + }); + }, + async admitTurn(input) { + return mutate((state) => { + const session = ownedSession(state, input.sessionId, input.actor.id); + const timestamp = now(); + const message = requiredText(input.params.message, "message"); + state.turns[input.traceId] = { + traceId: input.traceId, + sessionId: input.sessionId, + conversationId: session.conversationId, + ownerUserId: input.actor.id, + status: "admitted", + message, + createdAt: timestamp, + updatedAt: timestamp, + terminal: false, + }; + session.messages.push({ messageId: `msg_user_${input.traceId.slice(4)}`, role: "user", content: message, traceId: input.traceId, status: "completed", createdAt: timestamp }); + session.status = "running"; + session.lastTraceId = input.traceId; + session.updatedAt = timestamp; + return { ...input, params: { ...input.params, message, providerProfile: text(input.params.providerProfile) || session.providerProfile } }; + }); + }, + 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; + } + }, + 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 }); + if (!canceled) throw codedError("turn_not_running", "turn has no active AgentRun command"); + await projectTerminal(canceled, input.traceId); + return canceled; + }, + }; + + async function projectRunning(result: any, traceId: string) { + return mutate((state) => { + const turn = state.turns[traceId]; + if (!turn) throw codedError("turn_not_admitted", "turn must be admitted before dispatch"); + Object.assign(turn, { + status: "running", + updatedAt: now(), + runId: result.agentRun?.runId ?? null, + commandId: result.agentRun?.commandId ?? null, + runnerJobId: result.agentRun?.runnerJobId ?? null, + result, + }); + }); + } + + 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 { + const parsed = JSON.parse(await readFile(stateFile, "utf8").catch(() => '{"sessions":{},"turns":{}}')); + return { sessions: object(parsed.sessions), turns: object(parsed.turns) }; +} +function object(value: unknown) { return value && typeof value === "object" && !Array.isArray(value) ? value as Record> : {}; } +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 }); } diff --git a/internal/workbench/runtime.ts b/internal/workbench/runtime.ts index 866c2d0d..108ce6cb 100644 --- a/internal/workbench/runtime.ts +++ b/internal/workbench/runtime.ts @@ -3,16 +3,19 @@ import path from "node:path"; import { createCloudWorkbenchApplication } from "./cloud-application.ts"; import { createWorkbenchDispatcher } from "./dispatcher.ts"; import { createNativeTestWorkbenchApplication } from "./native-application.ts"; +import { createNativeAgentRunWorkbenchApplication } from "./native-agentrun-application.ts"; import { createNativeTestTemporalGateway } from "./native-temporal.ts"; import { createWorkbenchTemporalGateway } from "./temporal.ts"; export function workbenchRuntime(env: Record = process.env) { const mode = String(env.WORKBENCH_MODE ?? "").trim(); - if (mode !== "native-test" && mode !== "temporal") throw codedError("workbench_mode_required", "WORKBENCH_MODE must be native-test or temporal"); + if (mode !== "native-test" && mode !== "agentrun-native" && mode !== "temporal") throw codedError("workbench_mode_required", "WORKBENCH_MODE must be native-test, agentrun-native, or temporal"); const stateFile = path.resolve(String(env.WORKBENCH_NATIVE_STATE_FILE ?? ".state/workbench-native/state.json")); const application = mode === "native-test" ? createNativeTestWorkbenchApplication({ stateFile }) - : createCloudWorkbenchApplication({ baseUrl: String(env.WORKBENCH_CLOUD_API_URL ?? ""), authorization: String(env.WORKBENCH_CLOUD_API_AUTHORIZATION ?? "") }); + : mode === "agentrun-native" + ? createNativeAgentRunWorkbenchApplication({ stateFile, env }) + : createCloudWorkbenchApplication({ baseUrl: String(env.WORKBENCH_CLOUD_API_URL ?? ""), authorization: String(env.WORKBENCH_CLOUD_API_AUTHORIZATION ?? "") }); const temporalAddress = String(env.WORKBENCH_TEMPORAL_ADDRESS ?? env.TEMPORAL_ADDRESS ?? ""); const temporalNamespace = String(env.WORKBENCH_TEMPORAL_NAMESPACE ?? "unidesk"); const taskQueue = String(env.WORKBENCH_TEMPORAL_TASK_QUEUE ?? "hwlab-v03-workbench"); diff --git a/tools/src/workbench-native-service.ts b/tools/src/workbench-native-service.ts index f16a7130..462746cb 100644 --- a/tools/src/workbench-native-service.ts +++ b/tools/src/workbench-native-service.ts @@ -72,7 +72,7 @@ function serviceCommand(service: ServiceName, env: Record) { if (service === "api") return { ...process.env, ...env, WORKBENCH_MODE: env.WORKBENCH_MODE ?? "native-test", WORKBENCH_API_HOST: requiredEnv(env, "WORKBENCH_API_BIND_HOST"), WORKBENCH_API_PORT: requiredEnv(env, "WORKBENCH_API_PORT") }; if (service === "web") return { ...process.env, ...env, WORKBENCH_WEB_HOST: requiredEnv(env, "WORKBENCH_WEB_BIND_HOST"), WORKBENCH_WEB_PORT: requiredEnv(env, "WORKBENCH_WEB_PORT"), WORKBENCH_NATIVE_API_URL: requiredEnv(env, "WORKBENCH_NATIVE_API_URL"), WORKBENCH_WEB_RUNTIME_CONFIG: requiredEnv(env, "WORKBENCH_WEB_RUNTIME_CONFIG") }; - return { ...process.env, ...env, WORKBENCH_MODE: "temporal", WORKBENCH_WORKER_HEALTH_HOST: requiredEnv(env, "WORKBENCH_WORKER_BIND_HOST"), WORKBENCH_WORKER_HEALTH_PORT: requiredEnv(env, "WORKBENCH_WORKER_HEALTH_PORT") }; + return { ...process.env, ...env, WORKBENCH_MODE: env.WORKBENCH_MODE ?? "temporal", WORKBENCH_WORKER_HEALTH_HOST: requiredEnv(env, "WORKBENCH_WORKER_BIND_HOST"), WORKBENCH_WORKER_HEALTH_PORT: requiredEnv(env, "WORKBENCH_WORKER_HEALTH_PORT") }; } function validService(value: string): ServiceName { if (value === "api" || value === "worker" || value === "web") return value; throw codedError("invalid_service", "service must be api, worker, or web"); } async function readState(file: string) { return JSON.parse(await readFile(file, "utf8").catch(() => "null")); }