fix: connect L1 Workbench to AgentRun v0.2
This commit is contained in:
@@ -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) });
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
@@ -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 };
|
||||
};
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
|
||||
+23
-20
@@ -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); }
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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 }); }
|
||||
@@ -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");
|
||||
|
||||
@@ -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")); }
|
||||
|
||||
Reference in New Issue
Block a user