Merge pull request #2641 from pikasTech/fix/l1-agentrun-v02-smoke

修复 L1 Workbench 直连 AgentRun v0.2

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) });

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,

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.<namespace>.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());
 }

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);
+});

internal/cloud/workbench-command-proxy.ts

@@ -5,7 +5,11 @@ export function workbenchTemporalProxyEnabled(env: Record<string, string | undef
 }
 
 export function workbenchActivityDispatchRequested(request: any) {
-  return String(request.headers?.["x-workbench-activity-dispatch"] ?? "").trim() === "1";
+  const headers = request?.headers;
+  const value = typeof headers?.get === "function"
+    ? headers.get("x-workbench-activity-dispatch")
+    : headers?.["x-workbench-activity-dispatch"];
+  return String(value ?? "").trim() === "1";
 }
 
 export async function proxyWorkbenchCommandHttp(request: any, response: any, options: any, route: string) {

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<string, unknown> }
@@ -12,7 +14,7 @@ export type WorkbenchCommand =
 export type WorkbenchCommandResult = {
   ok: boolean;
   operation: WorkbenchCommand["operation"];
-  mode: "native-test" | "temporal";
+  mode: WorkbenchMode;
   data?: Record<string, unknown>;
   error?: { code: string; message: string; details?: unknown };
 };

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<WorkbenchCommandResult> {
     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<string, unknown>): WorkbenchCommandResult {
+function success(operation: WorkbenchCommand["operation"], mode: WorkbenchMode, data: Record<string, unknown>): WorkbenchCommandResult {
   return { ok: true, operation, mode, data };
 }
 

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<any>; snapshot?: () => Promise<{ sessions: Record<string, Record<string, unknown>>; turns: Record<string, Record<string, unknown>> }>; authorization?: string; 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; close?: () => Promise<void> }) {
   return {
     async fetch(request: Request): Promise<Response> {
       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<any> }, url: URL) {
+async function nativeSessionList(options: { snapshot?: () => Promise<any>; 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<any> }, sessionId: string, messagesOnly: boolean) {
+async function nativeSessionDetail(options: { snapshot?: () => Promise<any>; 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<any> }, traceId: string) {
+async function nativeTurn(options: { snapshot?: () => Promise<any>; 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<any> }) { 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<typeof setInterval> | 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); }

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<string, any>;
+  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<string, any>;
+  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<string, string> {
+  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;
+}

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<string, Record<string, any>>;
+  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>;
+  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 <T>(fn: (state: NativeState) => T | Promise<T>) => {
+    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<string, unknown>[],
+        };
+        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<NativeState> {
+  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<string, Record<string, any>> : {}; }
+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 }); }

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<string, string | undefined> = 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");

tools/src/workbench-native-service.ts

@@ -72,7 +72,7 @@ function serviceCommand(service: ServiceName, env: Record<string, string | undef
 function serviceEnv(service: ServiceName, env: Record<string, string | undefined>) {
   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")); }