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