diff --git a/apps/app/src/app/lib/openwork-server.ts b/apps/app/src/app/lib/openwork-server.ts index cdc4c8f505..aa00e34b23 100644 --- a/apps/app/src/app/lib/openwork-server.ts +++ b/apps/app/src/app/lib/openwork-server.ts @@ -90,6 +90,37 @@ export type OpenworkRuntimeSnapshot = { services: OpenworkRuntimeServiceSnapshot[]; }; +export type OpenworkAgentRuntime = "opencode" | "codex"; + +export type OpenworkAgentRuntimeStatus = { + runtime: OpenworkAgentRuntime; + available: boolean; + experimental: true; + version: string | null; + error: string | null; + process: { + running: boolean; + healthy: boolean; + transport: "stdio"; + placement: "remote-worker"; + publicPort: false; + platform: string | null; + }; + account: { + connected: boolean; + type: "chatgpt" | "apiKey" | "amazonBedrock" | null; + email: string | null; + planType: string | null; + }; + defaultModel: string | null; +}; + +export type OpenworkCodexDeviceLogin = { + loginId: string; + verificationUrl: string; + userCode: string; +}; + export type OpenworkServerSettings = { urlOverride?: string; portOverride?: number; @@ -1642,6 +1673,42 @@ export function createOpenworkServerClient(options: { baseUrl: string; token?: s `/workspace/${encodeURIComponent(workspaceId)}/runtime-config`, { token, hostToken, timeoutMs: timeouts.config }, ), + getAgentRuntime: (workspaceId: string) => + requestJson( + baseUrl, + `/workspace/${encodeURIComponent(workspaceId)}/agent-runtime`, + { token, hostToken, timeoutMs: timeouts.status }, + ), + setAgentRuntime: (workspaceId: string, runtime: OpenworkAgentRuntime) => + requestJson( + baseUrl, + `/workspace/${encodeURIComponent(workspaceId)}/agent-runtime`, + { + token, + hostToken, + method: "PUT", + body: { runtime }, + timeoutMs: 30_000, + }, + ), + startCodexDeviceLogin: (workspaceId: string) => + requestJson( + baseUrl, + `/workspace/${encodeURIComponent(workspaceId)}/agent-runtime/codex/login`, + { token, hostToken, method: "POST", timeoutMs: 30_000 }, + ), + cancelCodexDeviceLogin: (workspaceId: string, loginId: string) => + requestJson<{ ok: true }>( + baseUrl, + `/workspace/${encodeURIComponent(workspaceId)}/agent-runtime/codex/login/cancel`, + { token, hostToken, method: "POST", body: { loginId }, timeoutMs: timeouts.config }, + ), + logoutCodex: (workspaceId: string) => + requestJson( + baseUrl, + `/workspace/${encodeURIComponent(workspaceId)}/agent-runtime/codex/logout`, + { token, hostToken, method: "POST", timeoutMs: timeouts.config }, + ), patchConfig: (workspaceId: string, payload: { opencode?: Record; openwork?: Record }) => requestJson<{ updatedAt?: number | null }>(baseUrl, `/workspace/${workspaceId}/config`, { token, diff --git a/apps/app/src/react-app/domains/settings/codex-runtime-section.tsx b/apps/app/src/react-app/domains/settings/codex-runtime-section.tsx new file mode 100644 index 0000000000..7075bd5457 --- /dev/null +++ b/apps/app/src/react-app/domains/settings/codex-runtime-section.tsx @@ -0,0 +1,301 @@ +/** @jsxImportSource react */ +import { useCallback, useEffect, useState } from "react"; +import { + Check, + CheckCircle2, + Clipboard, + ExternalLink, + FlaskConical, + Loader2, + LogOut, + Server, + TerminalSquare, +} from "lucide-react"; + +import type { + OpenworkAgentRuntime, + OpenworkAgentRuntimeStatus, + OpenworkCodexDeviceLogin, + OpenworkServerClient, +} from "@/app/lib/openwork-server"; +import { Button } from "@/components/ui/button"; +import { cn } from "@/lib/utils"; +import { SettingsNotice, SettingsStatusBadge } from "./settings-section"; +import { + LayoutSection, + LayoutSectionDescription, + LayoutSectionHeader, + LayoutSectionItem, + LayoutSectionItemDescription, + LayoutSectionItemHeader, + LayoutSectionItemHeaderActions, + LayoutSectionItemTitle, + LayoutSectionTitle, +} from "./settings-layout"; + +type CodexRuntimeSectionProps = { + client: OpenworkServerClient | null; + workspaceId: string | null; + remote: boolean; + onOpenExternal: (url: string) => void; + onRuntimeChanged: (status: OpenworkAgentRuntimeStatus) => void | Promise; +}; + +function runtimeCardClass(selected: boolean) { + return cn( + "rounded-2xl border px-4 py-4 transition-colors", + selected ? "border-blue-7 bg-blue-2/40" : "border-dls-border bg-dls-sidebar/20", + ); +} + +function accountLabel(status: OpenworkAgentRuntimeStatus) { + if (!status.account.connected) return "ChatGPT not connected"; + const identity = status.account.email || status.account.planType; + return identity ? `ChatGPT connected · ${identity}` : "ChatGPT connected"; +} + +export function CodexRuntimeSection(props: CodexRuntimeSectionProps) { + const [status, setStatus] = useState(null); + const [login, setLogin] = useState(null); + const [busy, setBusy] = useState<"load" | "runtime" | "login" | "logout" | "cancel" | null>(null); + const [error, setError] = useState(null); + const [copied, setCopied] = useState(false); + + const refresh = useCallback(async () => { + if (!props.client || !props.workspaceId || !props.remote) return null; + try { + const next = await props.client.getAgentRuntime(props.workspaceId); + setStatus(next); + setError(next.error); + return next; + } catch (refreshError) { + setError(refreshError instanceof Error ? refreshError.message : "Unable to read the worker runtime"); + return null; + } + }, [props.client, props.remote, props.workspaceId]); + + useEffect(() => { + if (!props.remote) return; + setBusy("load"); + void refresh().finally(() => setBusy(null)); + }, [props.remote, refresh]); + + useEffect(() => { + if (!login || !props.client || !props.workspaceId) return; + let cancelled = false; + const poll = async () => { + const next = await refresh(); + if (cancelled || !next?.account.connected) return; + setLogin(null); + await props.onRuntimeChanged(next); + }; + const timer = window.setInterval(() => { void poll(); }, 2_000); + void poll(); + return () => { + cancelled = true; + window.clearInterval(timer); + }; + }, [login, props.client, props.onRuntimeChanged, props.workspaceId, refresh]); + + if (!props.remote) return null; + + const selectRuntime = async (runtime: OpenworkAgentRuntime) => { + if (!props.client || !props.workspaceId || busy) return; + setBusy("runtime"); + setError(null); + try { + const next = await props.client.setAgentRuntime(props.workspaceId, runtime); + setStatus(next); + setLogin(null); + await props.onRuntimeChanged(next); + } catch (selectError) { + setError(selectError instanceof Error ? selectError.message : "Unable to change the worker runtime"); + } finally { + setBusy(null); + } + }; + + const startLogin = async () => { + if (!props.client || !props.workspaceId || busy) return; + setBusy("login"); + setError(null); + try { + const next = await props.client.startCodexDeviceLogin(props.workspaceId); + setLogin(next); + props.onOpenExternal(next.verificationUrl); + } catch (loginError) { + setError(loginError instanceof Error ? loginError.message : "Unable to start ChatGPT sign-in"); + } finally { + setBusy(null); + } + }; + + const cancelLogin = async () => { + if (!props.client || !props.workspaceId || !login || busy) return; + setBusy("cancel"); + try { + await props.client.cancelCodexDeviceLogin(props.workspaceId, login.loginId); + setLogin(null); + } catch (cancelError) { + setError(cancelError instanceof Error ? cancelError.message : "Unable to cancel ChatGPT sign-in"); + } finally { + setBusy(null); + } + }; + + const logout = async () => { + if (!props.client || !props.workspaceId || busy) return; + setBusy("logout"); + setError(null); + try { + const next = await props.client.logoutCodex(props.workspaceId); + setStatus(next); + await props.onRuntimeChanged(next); + } catch (logoutError) { + setError(logoutError instanceof Error ? logoutError.message : "Unable to disconnect ChatGPT"); + } finally { + setBusy(null); + } + }; + + const copyCode = async () => { + if (!login) return; + await navigator.clipboard.writeText(login.userCode); + setCopied(true); + window.setTimeout(() => setCopied(false), 1_500); + }; + + const codexSelected = status?.runtime === "codex"; + const processHealthy = codexSelected && status?.process.healthy === true; + + return ( + + + + Agent runtime + + Experimental + + + + Choose which agent runs tasks on this remote worker. Your laptop only controls the session. + + + +
+ + + +
+ + {busy === "load" && !status ? ( +
+ Reading worker runtime… +
+ ) : null} + + {codexSelected && status ? ( + + +
+ + Codex on remote worker + + + {accountLabel(status)} +
+ + {status.account.connected ? ( + + ) : ( + + )} + + + +
+
Runtime: Codex app-server
+
Location: Remote worker
+
Transport: stdio (private)
+
Public control port: None
+
Version: {status.version || "Unknown"}
+
+ + ) : null} + + {login ? ( + + +
+ Finish signing in to ChatGPT + + Enter this one-time code on the OpenAI device sign-in page. Credentials stay on the worker. + +
+ +
+ + {login.userCode} + + + + +
+
+ Waiting for ChatGPT authorization… +
+ + ) : null} + + {error ? {error} : null} + + ); +} diff --git a/apps/app/src/react-app/domains/settings/pages/ai-view.tsx b/apps/app/src/react-app/domains/settings/pages/ai-view.tsx index 8a27293477..b6d7815406 100644 --- a/apps/app/src/react-app/domains/settings/pages/ai-view.tsx +++ b/apps/app/src/react-app/domains/settings/pages/ai-view.tsx @@ -53,6 +53,7 @@ export type AiSettingsViewProps = { onRefreshOpenWorkModels?: () => void | Promise; onDismissOpenWorkModels?: () => void | Promise; cloudProvidersView?: ReactNode; + agentRuntimeView?: ReactNode; }; function providerSourceLabel(source?: ConnectedProvider["source"]) { @@ -84,6 +85,7 @@ export function AiSettingsView(props: AiSettingsViewProps) { return ( + {props.agentRuntimeView} {/* ---- Providers ---- */} diff --git a/apps/app/src/react-app/shell/session-route.tsx b/apps/app/src/react-app/shell/session-route.tsx index 8121258fad..6f4a9df34c 100644 --- a/apps/app/src/react-app/shell/session-route.tsx +++ b/apps/app/src/react-app/shell/session-route.tsx @@ -965,7 +965,7 @@ export function SessionRoute() { todos, } = useSessionInteractions({ client: opencodeClient, - workspaceId: selectedWorkspaceId, + workspaceId: selectedWorkspaceEndpoint?.workspaceId ?? selectedWorkspaceId, sessionId: selectedSessionId, workspaceRoot: selectedWorkspaceRoot, }); diff --git a/apps/app/src/react-app/shell/settings-route.tsx b/apps/app/src/react-app/shell/settings-route.tsx index 2fa1e378e3..c03287c0a5 100644 --- a/apps/app/src/react-app/shell/settings-route.tsx +++ b/apps/app/src/react-app/shell/settings-route.tsx @@ -68,6 +68,7 @@ import { createProviderAuthStore, useProviderAuthStoreSnapshot } from "@/react-a import ProviderAuthModal from "@/react-app/domains/connections/provider-auth/provider-auth-modal"; import ConnectionsModals from "@/react-app/domains/connections/modals"; import { AiSettingsView } from "@/react-app/domains/settings/pages/ai-view"; +import { CodexRuntimeSection } from "@/react-app/domains/settings/codex-runtime-section"; // Side-effect imports: register extension config components into the registry. import "@/react-app/domains/settings/ollama-config"; import "@/react-app/domains/settings/computer-use-config"; @@ -2232,6 +2233,42 @@ function SettingsRouteContent(props: SettingsSurfaceProps = {}) { onSubscribeOpenWorkModels={subscribeToOpenWorkModels} onRefreshOpenWorkModels={refreshOpenWorkModels} onDismissOpenWorkModels={dismissOpenWorkModelsPromo} + agentRuntimeView={ + platform.openLink(url)} + onRuntimeChanged={async (runtimeStatus) => { + if (runtimeStatus.runtime === "codex" && runtimeStatus.defaultModel) { + local.setPrefs((previous) => ({ + ...previous, + defaultModel: { + providerID: "codex", + modelID: runtimeStatus.defaultModel ?? "", + }, + modelVariant: null, + })); + } else if ( + runtimeStatus.runtime === "opencode" && + local.prefs.defaultModel?.providerID === "codex" + ) { + const provider = providers.find((item) => providerConnectedIdSet.has(item.id)); + const modelID = provider + ? providerDefaults[provider.id] || Object.keys(provider.models ?? {})[0] || "" + : ""; + local.setPrefs((previous) => ({ + ...previous, + defaultModel: provider && modelID + ? { providerID: provider.id, modelID } + : null, + modelVariant: null, + })); + } + await providerAuthStore.refreshProviders({ force: true }); + }} + /> + } cloudProvidersView={ ({ + tableName: "agent_runtime_state", + valueColumn: "state_json", + parse: parseAgentRuntimeState, + serialize: JSON.stringify, +}); + +export async function readAgentRuntimeState( + config: ServerConfig, + workspaceId: string, +): Promise { + return (await agentRuntimeStore.get(config, workspaceId)) ?? DEFAULT_AGENT_RUNTIME_STATE; +} + +export async function writeAgentRuntimeState( + config: ServerConfig, + workspaceId: string, + runtime: AgentRuntime, +): Promise { + const state = { runtime } satisfies AgentRuntimeState; + await agentRuntimeStore.set(config, workspaceId, state); + return state; +} + +export async function workspaceUsesCodexRuntime( + config: ServerConfig, + workspaceId: string, +): Promise { + return (await readAgentRuntimeState(config, workspaceId)).runtime === "codex"; +} diff --git a/apps/server/src/codex-app-server.ts b/apps/server/src/codex-app-server.ts new file mode 100644 index 0000000000..6648118029 --- /dev/null +++ b/apps/server/src/codex-app-server.ts @@ -0,0 +1,338 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { mkdir } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface, type Interface as ReadlineInterface } from "node:readline"; +import { runtimeStorageDir } from "./runtime-db.js"; +import type { ServerConfig, WorkspaceInfo } from "./types.js"; + +export type CodexRpcId = number | string; + +export type CodexRpcMessage = { + id?: CodexRpcId; + method?: string; + params?: unknown; + result?: unknown; + error?: unknown; +}; + +export type CodexAppServerMetadata = { + userAgent: string; + codexHome: string; + platformFamily: string; + platformOs: string; +}; + +type PendingRequest = { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timeout: ReturnType; +}; + +type MessageListener = (message: CodexRpcMessage) => void; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function parseRpcMessage(line: string): CodexRpcMessage | null { + try { + const parsed: unknown = JSON.parse(line); + if (!isRecord(parsed)) return null; + const id = typeof parsed.id === "number" || typeof parsed.id === "string" ? parsed.id : undefined; + const method = typeof parsed.method === "string" ? parsed.method : undefined; + return { + ...(id !== undefined ? { id } : {}), + ...(method ? { method } : {}), + ...(Object.prototype.hasOwnProperty.call(parsed, "params") ? { params: parsed.params } : {}), + ...(Object.prototype.hasOwnProperty.call(parsed, "result") ? { result: parsed.result } : {}), + ...(Object.prototype.hasOwnProperty.call(parsed, "error") ? { error: parsed.error } : {}), + }; + } catch { + return null; + } +} + +function errorMessage(value: unknown): string { + if (value instanceof Error) return value.message; + if (typeof value === "string") return value; + if (isRecord(value) && typeof value.message === "string") return value.message; + try { + return JSON.stringify(value); + } catch { + return "Unknown Codex app-server error"; + } +} + +function metadataFromInitialize(value: unknown): CodexAppServerMetadata { + if (!isRecord(value)) { + throw new Error("Codex app-server returned an invalid initialize response"); + } + const userAgent = typeof value.userAgent === "string" ? value.userAgent : "codex"; + const codexHome = typeof value.codexHome === "string" ? value.codexHome : ""; + const platformFamily = typeof value.platformFamily === "string" ? value.platformFamily : "unknown"; + const platformOs = typeof value.platformOs === "string" ? value.platformOs : "unknown"; + return { userAgent, codexHome, platformFamily, platformOs }; +} + +function safeWorkspaceSegment(workspaceId: string): string { + return workspaceId.replace(/[^A-Za-z0-9._-]/g, "_") || "workspace"; +} + +export class CodexAppServerClient { + private static readonly REQUEST_TIMEOUT_MS = 30_000; + + private process: ChildProcessWithoutNullStreams | null = null; + private lines: ReadlineInterface | null = null; + private startPromise: Promise | null = null; + private requestId = 0; + private pending = new Map(); + private listeners = new Set(); + private stderrTail: string[] = []; + private initializedMetadata: CodexAppServerMetadata | null = null; + private stopped = false; + + constructor( + private readonly input: { + binary: string; + cwd: string; + codexHome: string; + clientVersion: string; + }, + ) {} + + get running(): boolean { + return this.process !== null && this.process.exitCode === null && !this.process.killed; + } + + get metadata(): CodexAppServerMetadata | null { + return this.initializedMetadata; + } + + get recentStderr(): string[] { + return [...this.stderrTail]; + } + + onMessage(listener: MessageListener): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + async start(): Promise { + if (this.running && this.initializedMetadata) return; + if (this.startPromise) return this.startPromise; + this.stopped = false; + this.startPromise = this.startProcess().finally(() => { + this.startPromise = null; + }); + return this.startPromise; + } + + private async startProcess(): Promise { + await mkdir(this.input.codexHome, { recursive: true, mode: 0o700 }); + const child = spawn(this.input.binary, ["app-server", "--listen", "stdio://"], { + cwd: this.input.cwd, + env: { + ...process.env, + CODEX_HOME: this.input.codexHome, + }, + stdio: ["pipe", "pipe", "pipe"], + }); + this.process = child; + this.lines = createInterface({ input: child.stdout }); + this.lines.on("line", (line) => this.handleLine(line)); + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + const lines = chunk.split(/\r?\n/).filter(Boolean); + this.stderrTail.push(...lines); + if (this.stderrTail.length > 40) this.stderrTail.splice(0, this.stderrTail.length - 40); + }); + child.once("error", (error) => this.handleExit(error)); + child.once("exit", (code, signal) => { + const detail = signal ? `signal ${signal}` : `code ${code ?? "unknown"}`; + this.handleExit(new Error(`Codex app-server exited with ${detail}`)); + }); + + try { + const initialized = await this.request("initialize", { + clientInfo: { + name: "openwork", + title: "OpenWork", + version: this.input.clientVersion, + }, + capabilities: null, + }); + this.initializedMetadata = metadataFromInitialize(initialized); + this.notify("initialized", {}); + } catch (error) { + await this.stop(); + throw error; + } + } + + private handleLine(line: string): void { + const message = parseRpcMessage(line); + if (!message) return; + if (message.id !== undefined && !message.method) { + const key = String(message.id); + const pending = this.pending.get(key); + if (!pending) return; + this.pending.delete(key); + clearTimeout(pending.timeout); + if (message.error !== undefined) { + pending.reject(new Error(errorMessage(message.error))); + } else { + pending.resolve(message.result); + } + return; + } + for (const listener of this.listeners) listener(message); + } + + private handleExit(error: Error): void { + if (!this.process) return; + this.lines?.close(); + this.lines = null; + this.process = null; + this.initializedMetadata = null; + for (const pending of this.pending.values()) { + clearTimeout(pending.timeout); + pending.reject(error); + } + this.pending.clear(); + if (!this.stopped) { + for (const listener of this.listeners) { + listener({ method: "openwork/processExited", params: { message: error.message } }); + } + } + } + + private write(message: CodexRpcMessage): void { + const child = this.process; + if (!child || child.stdin.destroyed) { + throw new Error("Codex app-server is not running"); + } + child.stdin.write(`${JSON.stringify(message)}\n`); + } + + notify(method: string, params: unknown): void { + this.write({ method, params }); + } + + async request(method: string, params: unknown): Promise { + if (method !== "initialize") await this.start(); + const id = ++this.requestId; + const key = String(id); + const result = new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + if (!this.pending.delete(key)) return; + reject(new Error(`Codex app-server timed out waiting for ${method}`)); + }, CodexAppServerClient.REQUEST_TIMEOUT_MS); + this.pending.set(key, { resolve, reject, timeout }); + }); + try { + this.write({ id, method, params }); + } catch (error) { + const pending = this.pending.get(key); + if (pending) clearTimeout(pending.timeout); + this.pending.delete(key); + throw error; + } + return result; + } + + respond(id: CodexRpcId, result: unknown): void { + this.write({ id, result }); + } + + respondError(id: CodexRpcId, code: number, message: string): void { + this.write({ id, error: { code, message } }); + } + + async stop(): Promise { + this.stopped = true; + const child = this.process; + this.lines?.close(); + this.lines = null; + this.process = null; + this.initializedMetadata = null; + const error = new Error("Codex app-server stopped"); + for (const pending of this.pending.values()) { + clearTimeout(pending.timeout); + pending.reject(error); + } + this.pending.clear(); + if (!child) return; + child.stdin.end(); + child.kill("SIGTERM"); + } +} + +export class CodexAppServerManager { + private clients = new Map(); + + constructor( + private readonly config: ServerConfig, + private readonly clientVersion: string, + ) {} + + clientFor(workspace: WorkspaceInfo): CodexAppServerClient { + const existing = this.clients.get(workspace.id); + if (existing) return existing; + if (workspace.workspaceType !== "local" || !workspace.path.trim()) { + throw new Error("Codex Server requires a local workspace on the remote worker"); + } + const codexHome = join( + runtimeStorageDir(this.config), + "codex", + safeWorkspaceSegment(workspace.id), + ); + const client = new CodexAppServerClient({ + binary: process.env.OPENWORK_CODEX_BIN?.trim() || "codex", + cwd: workspace.path, + codexHome, + clientVersion: this.clientVersion, + }); + this.clients.set(workspace.id, client); + return client; + } + + async stopWorkspace(workspaceId: string): Promise { + const client = this.clients.get(workspaceId); + if (!client) return; + this.clients.delete(workspaceId); + await client.stop(); + } + + async stop(): Promise { + await Promise.all([...this.clients.values()].map((client) => client.stop())); + this.clients.clear(); + } +} + +export async function probeCodexBinary(binary = process.env.OPENWORK_CODEX_BIN?.trim() || "codex"): Promise { + return new Promise((resolve, reject) => { + const child = spawn(binary, ["--version"], { stdio: ["ignore", "pipe", "pipe"] }); + let stdout = ""; + let stderr = ""; + const timeout = setTimeout(() => { + child.kill("SIGTERM"); + reject(new Error("Timed out while checking the Codex CLI")); + }, 5_000); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { stdout += chunk; }); + child.stderr.on("data", (chunk: string) => { stderr += chunk; }); + child.once("error", (error) => { + clearTimeout(timeout); + reject(error); + }); + child.once("exit", (code) => { + clearTimeout(timeout); + if (code === 0) { + resolve(stdout.trim() || "codex"); + } else { + reject(new Error(stderr.trim() || `Codex CLI exited with code ${code ?? "unknown"}`)); + } + }); + }); +} diff --git a/apps/server/src/codex-opencode-adapter.ts b/apps/server/src/codex-opencode-adapter.ts new file mode 100644 index 0000000000..4d07997587 --- /dev/null +++ b/apps/server/src/codex-opencode-adapter.ts @@ -0,0 +1,1170 @@ +import { randomUUID } from "node:crypto"; +import type { AgentRuntime } from "./agent-runtime-store.js"; +import { + readAgentRuntimeState, + workspaceUsesCodexRuntime, + writeAgentRuntimeState, +} from "./agent-runtime-store.js"; +import { + CodexAppServerManager, + probeCodexBinary, + type CodexAppServerClient, + type CodexRpcId, + type CodexRpcMessage, +} from "./codex-app-server.js"; +import type { ServerConfig, WorkspaceInfo } from "./types.js"; + +type JsonRecord = Record; + +type OpenCodeEvent = { + type: string; + properties: unknown; +}; + +type PendingApproval = { + id: string; + rpcId: CodexRpcId; + sessionId: string; + action: string; + resources: string[]; + metadata: JsonRecord; + permissions: JsonRecord | null; + source: { messageID: string; callID: string }; + method: string; +}; + +export type CodexRuntimeStatus = { + runtime: AgentRuntime; + available: boolean; + experimental: true; + version: string | null; + error: string | null; + process: { + running: boolean; + healthy: boolean; + transport: "stdio"; + placement: "remote-worker"; + publicPort: false; + platform: string | null; + }; + account: { + connected: boolean; + type: "chatgpt" | "apiKey" | "amazonBedrock" | null; + email: string | null; + planType: string | null; + }; + defaultModel: string | null; +}; + +export type CodexDeviceLogin = { + loginId: string; + verificationUrl: string; + userCode: string; +}; + +function isRecord(value: unknown): value is JsonRecord { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function arrayValue(value: unknown): unknown[] { + return Array.isArray(value) ? value : []; +} + +function stringValue(value: unknown): string { + return typeof value === "string" ? value : ""; +} + +function numberValue(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) ? value : null; +} + +function objectField(value: unknown, field: string): JsonRecord | null { + if (!isRecord(value)) return null; + return isRecord(value[field]) ? value[field] : null; +} + +function stringField(value: unknown, field: string): string { + if (!isRecord(value)) return ""; + return stringValue(value[field]); +} + +function jsonResponse(data: unknown, status = 200): Response { + return new Response(JSON.stringify(data), { + status, + headers: { "Content-Type": "application/json" }, + }); +} + +function errorResponse(error: unknown, status = 502): Response { + const message = error instanceof Error ? error.message : stringValue(error) || "Codex runtime request failed"; + return jsonResponse({ + name: "UnknownError", + data: { message }, + }, status); +} + +function normalizeProxyPath(proxyPath: string): string { + const withoutPrefix = proxyPath.startsWith("/opencode") + ? proxyPath.slice("/opencode".length) + : proxyPath; + const normalized = (withoutPrefix || "/").replace(/\/+$/, ""); + return normalized || "/"; +} + +function toMillis(seconds: unknown): number { + const value = numberValue(seconds); + return value === null ? Date.now() : Math.round(value * 1_000); +} + +function firstLine(value: string): string { + return value.split(/\r?\n/, 1)[0]?.trim() || "New task"; +} + +function fileChangePatchText(changes: unknown[]): string { + return changes.flatMap((change) => { + if (!isRecord(change)) return []; + const path = stringField(change, "path"); + if (!path) return []; + + const kindValue = change.kind; + const kind = stringValue(kindValue) || stringField(kindValue, "type"); + const header = kind === "add" + ? `*** Add File: ${path}` + : kind === "delete" + ? `*** Delete File: ${path}` + : `*** Update File: ${path}`; + const movePath = objectField(change, "kind") + ? stringField(kindValue, "move_path") + : ""; + return [[header, ...(movePath ? [`*** Move to: ${movePath}`] : []), stringField(change, "diff")] + .filter(Boolean) + .join("\n")]; + }).join("\n"); +} + +function threadStatusBusy(thread: JsonRecord): boolean { + const status = isRecord(thread.status) ? thread.status : null; + return status?.type === "active"; +} + +function threadToSession(thread: JsonRecord, fallbackDirectory: string, version: string): JsonRecord { + const id = stringField(thread, "id"); + const preview = stringField(thread, "preview"); + const name = stringField(thread, "name"); + const directory = stringField(thread, "cwd") || fallbackDirectory; + const created = toMillis(thread.createdAt); + const updated = toMillis(thread.updatedAt); + return { + id, + slug: id, + projectID: "codex", + directory, + title: name || firstLine(preview), + version, + time: { created, updated }, + metadata: { + runtime: "codex", + execution: "remote-worker", + busy: threadStatusBusy(thread), + }, + }; +} + +function messageInfo(input: { + id: string; + sessionId: string; + role: "user" | "assistant"; + createdAt: number; + completedAt?: number | null; + parentId?: string; + directory: string; + model?: string; +}): JsonRecord { + if (input.role === "user") { + return { + id: input.id, + sessionID: input.sessionId, + role: "user", + time: { created: input.createdAt }, + agent: "openwork", + model: { providerID: "codex", modelID: input.model || "default" }, + }; + } + return { + id: input.id, + sessionID: input.sessionId, + role: "assistant", + time: { + created: input.createdAt, + ...(input.completedAt ? { completed: input.completedAt } : {}), + }, + parentID: input.parentId || input.sessionId, + modelID: input.model || "default", + providerID: "codex", + mode: "build", + agent: "openwork", + path: { cwd: input.directory, root: input.directory }, + cost: 0, + tokens: { + input: 0, + output: 0, + reasoning: 0, + cache: { read: 0, write: 0 }, + }, + }; +} + +function userContentText(content: unknown): string { + return arrayValue(content) + .flatMap((entry) => isRecord(entry) && entry.type === "text" ? [stringField(entry, "text")] : []) + .filter(Boolean) + .join("\n"); +} + +function assistantMessageId(turnId: string): string { + return `assistant:${turnId}`; +} + +function itemToPart(input: { + item: JsonRecord; + sessionId: string; + turnId: string; + startedAt: number; + completedAt?: number | null; +}): JsonRecord | null { + const item = input.item; + const itemId = stringField(item, "id") || randomUUID(); + const messageID = assistantMessageId(input.turnId); + const base = { id: itemId, sessionID: input.sessionId, messageID }; + if (item.type === "agentMessage" || item.type === "plan") { + return { + ...base, + type: "text", + text: stringField(item, "text"), + time: { + start: input.startedAt, + ...(input.completedAt ? { end: input.completedAt } : {}), + }, + }; + } + if (item.type === "reasoning") { + const summary = arrayValue(item.summary).filter((entry): entry is string => typeof entry === "string"); + const content = arrayValue(item.content).filter((entry): entry is string => typeof entry === "string"); + return { + ...base, + type: "reasoning", + text: [...summary, ...content].join("\n"), + time: { + start: input.startedAt, + ...(input.completedAt ? { end: input.completedAt } : {}), + }, + }; + } + if (item.type === "commandExecution") { + const status = stringField(item, "status"); + const completed = status === "completed"; + const failed = status === "failed" || status === "declined"; + const command = stringField(item, "command"); + return { + ...base, + type: "tool", + callID: itemId, + tool: "bash", + state: failed + ? { + status: "error", + input: { command, cwd: stringField(item, "cwd") }, + error: stringField(item, "aggregatedOutput") || `Command ${status}`, + time: { start: input.startedAt, end: input.completedAt || Date.now() }, + } + : completed + ? { + status: "completed", + input: { command, cwd: stringField(item, "cwd") }, + output: stringField(item, "aggregatedOutput"), + title: command || "Command", + metadata: { exitCode: item.exitCode ?? null }, + time: { start: input.startedAt, end: input.completedAt || Date.now() }, + } + : { + status: "running", + input: { command, cwd: stringField(item, "cwd") }, + title: command || "Command", + time: { start: input.startedAt }, + }, + }; + } + if (item.type === "fileChange") { + const changes = arrayValue(item.changes); + const patchText = fileChangePatchText(changes); + const files = changes.flatMap((change) => { + if (!isRecord(change)) return []; + const path = stringField(change, "path"); + return path ? [path] : []; + }); + const status = stringField(item, "status"); + const failed = status === "failed" || status === "declined"; + return { + ...base, + type: "tool", + callID: itemId, + tool: "apply_patch", + state: failed + ? { + status: "error", + input: { patchText }, + error: `File change ${status}`, + time: { start: input.startedAt, end: input.completedAt || Date.now() }, + } + : input.completedAt + ? { + status: "completed", + input: { patchText }, + output: files.length ? `Updated ${files.join(", ")}` : "File changes applied", + title: "File changes", + metadata: { files }, + time: { start: input.startedAt, end: input.completedAt }, + } + : { + status: "running", + input: { patchText }, + title: "File changes", + time: { start: input.startedAt }, + }, + }; + } + if (item.type === "mcpToolCall" || item.type === "dynamicToolCall") { + const tool = stringField(item, "tool") || "tool"; + const status = stringField(item, "status"); + const failed = status === "failed"; + const complete = status === "completed" || input.completedAt; + return { + ...base, + type: "tool", + callID: itemId, + tool, + state: failed + ? { + status: "error", + input: isRecord(item.arguments) ? item.arguments : {}, + error: stringField(item, "error") || `${tool} failed`, + time: { start: input.startedAt, end: input.completedAt || Date.now() }, + } + : complete + ? { + status: "completed", + input: isRecord(item.arguments) ? item.arguments : {}, + output: JSON.stringify(item.result ?? item.contentItems ?? ""), + title: tool, + metadata: {}, + time: { start: input.startedAt, end: input.completedAt || Date.now() }, + } + : { + status: "running", + input: isRecord(item.arguments) ? item.arguments : {}, + title: tool, + time: { start: input.startedAt }, + }, + }; + } + return null; +} + +function threadMessages(thread: JsonRecord, directory: string, model: string): JsonRecord[] { + const sessionId = stringField(thread, "id"); + const messages: JsonRecord[] = []; + for (const rawTurn of arrayValue(thread.turns)) { + if (!isRecord(rawTurn)) continue; + const turnId = stringField(rawTurn, "id"); + if (!turnId) continue; + const startedAt = toMillis(rawTurn.startedAt); + const completedAt = numberValue(rawTurn.completedAt) === null ? null : toMillis(rawTurn.completedAt); + let lastUserId = sessionId; + const assistantParts: JsonRecord[] = []; + for (const rawItem of arrayValue(rawTurn.items)) { + if (!isRecord(rawItem)) continue; + if (rawItem.type === "userMessage") { + const id = stringField(rawItem, "id") || randomUUID(); + const text = userContentText(rawItem.content); + messages.push({ + info: messageInfo({ id, sessionId, role: "user", createdAt: startedAt, directory, model }), + parts: [{ id: `${id}:text`, sessionID: sessionId, messageID: id, type: "text", text }], + }); + lastUserId = id; + continue; + } + const part = itemToPart({ item: rawItem, sessionId, turnId, startedAt, completedAt }); + if (part) assistantParts.push(part); + } + if (assistantParts.length) { + const id = assistantMessageId(turnId); + messages.push({ + info: messageInfo({ + id, + sessionId, + role: "assistant", + createdAt: startedAt, + completedAt, + parentId: lastUserId, + directory, + model, + }), + parts: assistantParts, + }); + } + } + return messages; +} + +function threadFromRpc(value: unknown): JsonRecord { + const thread = objectField(value, "thread"); + if (!thread || !stringField(thread, "id")) { + throw new Error("Codex app-server returned an invalid thread"); + } + return thread; +} + +function modelIdFromBody(body: JsonRecord): string | null { + if (typeof body.model === "string") return body.model.trim() || null; + if (!isRecord(body.model)) return null; + return stringField(body.model, "modelID") || stringField(body.model, "id") || null; +} + +function reasoningEffortFromBody(body: JsonRecord): string | null { + const effort = (stringField(body, "reasoning_effort") || stringField(body, "variant")).trim(); + return effort || null; +} + +function grantedPermissions(value: JsonRecord | null): JsonRecord { + if (!value) return {}; + const network = objectField(value, "network"); + const fileSystem = objectField(value, "fileSystem"); + return { + ...(network ? { network } : {}), + ...(fileSystem ? { fileSystem } : {}), + }; +} + +function inputFromOpenCodeParts(value: unknown): JsonRecord[] { + const input: JsonRecord[] = []; + for (const rawPart of arrayValue(value)) { + if (!isRecord(rawPart)) continue; + if (rawPart.type === "text") { + const text = stringField(rawPart, "text"); + if (text) input.push({ type: "text", text, text_elements: [] }); + continue; + } + if (rawPart.type === "file") { + const url = stringField(rawPart, "url"); + const mime = stringField(rawPart, "mime"); + if (url.startsWith("file://") && mime.startsWith("image/")) { + input.push({ type: "localImage", path: decodeURIComponent(url.slice("file://".length)) }); + } else if (url) { + input.push({ type: "text", text: `Attached file: ${url}`, text_elements: [] }); + } + } + } + return input; +} + +function codexModelToOpenCode(raw: JsonRecord): JsonRecord | null { + const id = stringField(raw, "model") || stringField(raw, "id"); + if (!id) return null; + const effortOptions = arrayValue(raw.supportedReasoningEfforts) + .flatMap((entry) => isRecord(entry) ? [stringField(entry, "reasoningEffort")] : []) + .filter(Boolean); + return { + id, + providerID: "codex", + api: { id, url: "codex-app-server", npm: "@openai/codex" }, + name: stringField(raw, "displayName") || id, + family: "codex", + capabilities: { + temperature: false, + reasoning: effortOptions.length > 0, + attachment: true, + toolcall: true, + input: { text: true, audio: false, image: true, video: false, pdf: false }, + output: { text: true, audio: false, image: false, video: false, pdf: false }, + interleaved: false, + }, + cost: { input: 0, output: 0, cache: { read: 0, write: 0 } }, + limit: { context: 200_000, output: 100_000 }, + status: "active", + options: {}, + headers: {}, + release_date: "", + variants: Object.fromEntries(effortOptions.map((effort) => [effort, {}])), + }; +} + +export class CodexOpenCodeAdapter { + private subscribers = new Set<(event: OpenCodeEvent) => void>(); + private activeTurnByThread = new Map(); + private modelByThread = new Map(); + private pendingApprovals = new Map(); + private unsubscribe: () => void; + + constructor( + readonly workspace: WorkspaceInfo, + readonly client: CodexAppServerClient, + ) { + this.unsubscribe = client.onMessage((message) => this.handleCodexMessage(message)); + } + + dispose(): void { + this.unsubscribe(); + this.subscribers.clear(); + } + + private emit(type: string, properties: unknown): void { + const event = { type, properties } satisfies OpenCodeEvent; + for (const subscriber of this.subscribers) subscriber(event); + } + + private sessionVersion(): string { + return this.client.metadata?.userAgent || "codex"; + } + + private handleCodexMessage(message: CodexRpcMessage): void { + if (!message.method) return; + if (message.id !== undefined) { + this.handleServerRequest(message); + return; + } + const params = isRecord(message.params) ? message.params : {}; + if (message.method === "thread/started") { + const thread = objectField(params, "thread"); + if (thread) this.emit("session.created", { info: threadToSession(thread, this.workspace.path, this.sessionVersion()) }); + return; + } + if (message.method === "thread/name/updated") { + const threadId = stringField(params, "threadId"); + if (threadId) this.emit("session.updated", { info: { id: threadId, title: stringField(params, "threadName") } }); + return; + } + if (message.method === "thread/status/changed") { + const threadId = stringField(params, "threadId"); + const status = objectField(params, "status"); + if (!threadId || !status) return; + const busy = status.type === "active"; + this.emit("session.status", { sessionID: threadId, status: { type: busy ? "busy" : "idle" } }); + if (!busy) this.emit("session.idle", { sessionID: threadId }); + return; + } + if (message.method === "turn/started") { + const threadId = stringField(params, "threadId"); + const turn = objectField(params, "turn"); + const turnId = turn ? stringField(turn, "id") : ""; + if (!threadId || !turn || !turnId) return; + this.activeTurnByThread.set(threadId, turnId); + this.emit("session.status", { sessionID: threadId, status: { type: "busy" } }); + for (const item of arrayValue(turn.items)) { + if (isRecord(item)) this.emitItem(threadId, turnId, item, false, toMillis(turn.startedAt)); + } + return; + } + if (message.method === "item/started" || message.method === "item/completed") { + const item = objectField(params, "item"); + const threadId = stringField(params, "threadId"); + const turnId = stringField(params, "turnId"); + if (!item || !threadId || !turnId) return; + const completed = message.method === "item/completed"; + const at = numberValue(completed ? params.completedAtMs : params.startedAtMs) ?? Date.now(); + this.emitItem(threadId, turnId, item, completed, at); + return; + } + if (message.method === "item/agentMessage/delta") { + const threadId = stringField(params, "threadId"); + const turnId = stringField(params, "turnId"); + const itemId = stringField(params, "itemId"); + const delta = stringField(params, "delta"); + if (!threadId || !turnId || !itemId || !delta) return; + this.emit("message.part.delta", { + sessionID: threadId, + messageID: assistantMessageId(turnId), + partID: itemId, + field: "text", + delta, + }); + return; + } + if (message.method === "turn/completed") { + const threadId = stringField(params, "threadId"); + const turn = objectField(params, "turn"); + const turnId = turn ? stringField(turn, "id") : ""; + if (!threadId) return; + if (turnId && this.activeTurnByThread.get(threadId) === turnId) this.activeTurnByThread.delete(threadId); + if (turn && turn.status === "failed") { + const turnError = objectField(turn, "error"); + this.emit("session.error", { + sessionID: threadId, + error: { name: "UnknownError", data: { message: stringField(turnError, "message") || "Codex turn failed" } }, + }); + } + this.emit("session.status", { sessionID: threadId, status: { type: "idle" } }); + this.emit("session.idle", { sessionID: threadId }); + return; + } + if (message.method === "error") { + const threadId = stringField(params, "threadId"); + const error = objectField(params, "error"); + if (threadId) { + this.emit("session.error", { + sessionID: threadId, + error: { name: "UnknownError", data: { message: stringField(error, "message") || "Codex runtime error" } }, + }); + } + return; + } + if (message.method === "openwork/processExited") { + const detail = stringField(params, "message") || "Codex app-server stopped"; + for (const sessionId of this.activeTurnByThread.keys()) { + this.emit("session.error", { + sessionID: sessionId, + error: { name: "UnknownError", data: { message: detail } }, + }); + } + this.activeTurnByThread.clear(); + } + } + + private emitItem(sessionId: string, turnId: string, item: JsonRecord, completed: boolean, at: number): void { + if (item.type === "userMessage") { + const id = stringField(item, "id") || randomUUID(); + const info = messageInfo({ + id, + sessionId, + role: "user", + createdAt: at, + directory: this.workspace.path, + model: this.modelByThread.get(sessionId), + }); + this.emit("message.updated", { info }); + this.emit("message.part.updated", { + part: { + id: `${id}:text`, + sessionID: sessionId, + messageID: id, + type: "text", + text: userContentText(item.content), + }, + }); + return; + } + const messageID = assistantMessageId(turnId); + this.emit("message.updated", { + info: messageInfo({ + id: messageID, + sessionId, + role: "assistant", + createdAt: at, + completedAt: completed ? at : null, + directory: this.workspace.path, + model: this.modelByThread.get(sessionId), + }), + }); + const part = itemToPart({ + item, + sessionId, + turnId, + startedAt: at, + completedAt: completed ? at : null, + }); + if (part) this.emit("message.part.updated", { part }); + } + + private handleServerRequest(message: CodexRpcMessage): void { + if (message.id === undefined || !message.method) return; + const params = isRecord(message.params) ? message.params : {}; + if (!["item/commandExecution/requestApproval", "item/fileChange/requestApproval", "item/permissions/requestApproval"].includes(message.method)) { + this.client.respondError(message.id, -32601, `OpenWork does not support ${message.method}`); + return; + } + const sessionId = stringField(params, "threadId"); + const turnId = stringField(params, "turnId"); + const itemId = stringField(params, "itemId"); + if (!sessionId || !turnId || !itemId) { + this.client.respondError(message.id, -32602, "Invalid approval request"); + return; + } + const id = `codex-${String(message.id)}`; + const command = stringField(params, "command"); + const cwd = stringField(params, "cwd"); + const reason = stringField(params, "reason"); + const permissions = message.method === "item/permissions/requestApproval" + ? objectField(params, "permissions") + : null; + const action = message.method.includes("commandExecution") + ? "command.execute" + : message.method.includes("fileChange") + ? "file.write" + : "permissions.request"; + const resources = [command, cwd, stringField(params, "grantRoot")].filter(Boolean); + const approval: PendingApproval = { + id, + rpcId: message.id, + sessionId, + action, + resources, + metadata: { + runtime: "codex", + ...(command ? { command } : {}), + ...(cwd ? { cwd } : {}), + ...(reason ? { reason } : {}), + ...(permissions ? { permissions } : {}), + }, + permissions, + source: { messageID: assistantMessageId(turnId), callID: itemId }, + method: message.method, + }; + this.pendingApprovals.set(id, approval); + this.emit("permission.v2.asked", this.openCodePermission(approval)); + } + + private openCodePermission(approval: PendingApproval): JsonRecord { + return { + id: approval.id, + sessionID: approval.sessionId, + action: approval.action, + resources: approval.resources, + save: [approval.action], + metadata: approval.metadata, + source: approval.source, + }; + } + + private async readBody(request: Request): Promise { + const text = await request.text(); + if (!text.trim()) return {}; + const value: unknown = JSON.parse(text); + return isRecord(value) ? value : {}; + } + + private async readThread(threadId: string, includeTurns: boolean): Promise { + return threadFromRpc(await this.client.request("thread/read", { threadId, includeTurns })); + } + + async listSessions(input: { limit?: number; search?: string }): Promise { + const result = await this.client.request("thread/list", { + limit: input.limit ?? 200, + cwd: this.workspace.path, + ...(input.search ? { searchTerm: input.search } : {}), + }); + const data = isRecord(result) ? arrayValue(result.data) : []; + return data.flatMap((thread) => isRecord(thread) + ? [threadToSession(thread, this.workspace.path, this.sessionVersion())] + : []); + } + + async createSession(input: { title?: string; model?: string | null }): Promise { + const params: JsonRecord = { + cwd: this.workspace.path, + approvalPolicy: "on-request", + sandbox: "workspace-write", + serviceName: "openwork", + developerInstructions: "You are running inside OpenWork on a remote worker. Keep all file work inside the current workspace unless the user explicitly approves otherwise.", + }; + if (input.model) params.model = input.model; + const thread = threadFromRpc(await this.client.request("thread/start", params)); + const threadId = stringField(thread, "id"); + if (input.model) this.modelByThread.set(threadId, input.model); + if (input.title?.trim()) { + await this.client.request("thread/name/set", { threadId, name: input.title.trim() }); + thread.name = input.title.trim(); + } + return threadToSession(thread, this.workspace.path, this.sessionVersion()); + } + + async getSession(threadId: string): Promise { + return threadToSession( + await this.readThread(threadId, false), + this.workspace.path, + this.sessionVersion(), + ); + } + + async getMessages(threadId: string): Promise { + const thread = await this.readThread(threadId, true); + return threadMessages(thread, this.workspace.path, this.modelByThread.get(threadId) || "default"); + } + + async getStatus(): Promise> { + const sessions = await this.listSessions({ limit: 200 }); + return Object.fromEntries(sessions.flatMap((session) => { + const id = stringField(session, "id"); + if (!id) return []; + const metadata = isRecord(session.metadata) ? session.metadata : {}; + return [[id, { type: metadata.busy === true ? "busy" : "idle" }]]; + })); + } + + async deleteSession(threadId: string): Promise { + await this.client.request("thread/delete", { threadId }); + this.activeTurnByThread.delete(threadId); + this.modelByThread.delete(threadId); + this.emit("session.deleted", { sessionID: threadId }); + return true; + } + + async renameSession(threadId: string, title: string): Promise { + await this.client.request("thread/name/set", { threadId, name: title }); + return this.getSession(threadId); + } + + async startTurn(threadId: string, body: JsonRecord): Promise { + const input = inputFromOpenCodeParts(body.parts); + if (!input.length) throw new Error("A Codex turn requires text or an image"); + const model = modelIdFromBody(body); + const effort = reasoningEffortFromBody(body); + if (model) this.modelByThread.set(threadId, model); + const params: JsonRecord = { + threadId, + clientUserMessageId: stringField(body, "messageID") || randomUUID(), + input, + cwd: this.workspace.path, + approvalPolicy: "on-request", + sandboxPolicy: { + type: "workspaceWrite", + writableRoots: [this.workspace.path], + networkAccess: false, + excludeTmpdirEnvVar: false, + excludeSlashTmp: false, + }, + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + }; + const result = await this.client.request("turn/start", params); + const turn = objectField(result, "turn"); + const turnId = turn ? stringField(turn, "id") : ""; + if (turnId) this.activeTurnByThread.set(threadId, turnId); + } + + async abortTurn(threadId: string): Promise { + let turnId = this.activeTurnByThread.get(threadId) || ""; + if (!turnId) { + const thread = await this.readThread(threadId, true); + const turns = arrayValue(thread.turns); + const latest = turns[turns.length - 1]; + if (isRecord(latest) && latest.status === "inProgress") turnId = stringField(latest, "id"); + } + if (!turnId) return false; + await this.client.request("turn/interrupt", { threadId, turnId }); + this.activeTurnByThread.delete(threadId); + return true; + } + + pendingPermissions(threadId?: string): JsonRecord[] { + return [...this.pendingApprovals.values()].flatMap((approval) => + !threadId || approval.sessionId === threadId ? [this.openCodePermission(approval)] : [], + ); + } + + replyPermission(requestId: string, reply: string): boolean { + const approval = this.pendingApprovals.get(requestId); + if (!approval) return false; + const decision = reply === "always" ? "acceptForSession" : reply === "reject" ? "decline" : "accept"; + this.pendingApprovals.delete(requestId); + if (approval.method === "item/permissions/requestApproval") { + if (reply === "reject") { + this.client.respondError(approval.rpcId, -32000, "Permission request declined"); + } else { + this.client.respond(approval.rpcId, { + permissions: grantedPermissions(approval.permissions), + scope: reply === "always" ? "session" : "turn", + }); + } + } else { + this.client.respond(approval.rpcId, { decision }); + } + this.emit("permission.v2.replied", { + sessionID: approval.sessionId, + requestID: requestId, + reply, + }); + return true; + } + + async providerList(): Promise { + const result = await this.client.request("model/list", { includeHidden: false }); + const data = isRecord(result) ? arrayValue(result.data) : []; + const models = data.flatMap((entry) => isRecord(entry) ? [codexModelToOpenCode(entry)] : []) + .filter((entry): entry is JsonRecord => entry !== null); + const byId = Object.fromEntries(models.map((model) => [stringField(model, "id"), model])); + const defaultEntry = data.find((entry) => isRecord(entry) && entry.isDefault === true); + const defaultModel = isRecord(defaultEntry) + ? stringField(defaultEntry, "model") || stringField(defaultEntry, "id") + : stringField(models[0], "id"); + const account = await this.client.request("account/read", { refreshToken: false }).catch(() => null); + const connected = Boolean(objectField(account, "account")); + return { + all: [{ id: "codex", name: "Codex (ChatGPT)", source: "custom", env: [], options: {}, models: byId }], + default: defaultModel ? { codex: defaultModel } : {}, + connected: connected ? ["codex"] : [], + }; + } + + private eventStream(request: Request): Response { + const encoder = new TextEncoder(); + let unsubscribe = () => {}; + let heartbeat: ReturnType | null = null; + const stream = new ReadableStream({ + start: (controller) => { + const send = (event: OpenCodeEvent) => { + try { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); + } catch { + // The browser closed the stream. + } + }; + send({ type: "server.connected", properties: { runtime: "codex" } }); + const listener = (event: OpenCodeEvent) => send(event); + this.subscribers.add(listener); + unsubscribe = () => this.subscribers.delete(listener); + heartbeat = setInterval(() => { + try { controller.enqueue(encoder.encode(": keepalive\n\n")); } catch {} + }, 15_000); + request.signal.addEventListener("abort", () => { + unsubscribe(); + if (heartbeat) clearInterval(heartbeat); + try { controller.close(); } catch {} + }, { once: true }); + }, + cancel: () => { + unsubscribe(); + if (heartbeat) clearInterval(heartbeat); + }, + }); + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + }, + }); + } + + async handleProxy(request: Request, proxyPath: string, url: URL): Promise { + const path = normalizeProxyPath(proxyPath); + try { + await this.client.start(); + if (request.method === "GET" && path === "/global/health") { + return jsonResponse({ healthy: true, version: this.sessionVersion(), runtime: "codex" }); + } + if (request.method === "POST" && path === "/global/dispose") return jsonResponse(true); + if (request.method === "GET" && path === "/event") return this.eventStream(request); + if (request.method === "GET" && path === "/provider") return jsonResponse(await this.providerList()); + if (request.method === "GET" && path === "/provider/auth") return jsonResponse({}); + if (request.method === "GET" && path === "/config/providers") { + const providers = await this.providerList(); + return jsonResponse({ providers: providers.all, default: providers.default }); + } + if (request.method === "GET" && path === "/agent") { + return jsonResponse([{ + name: "openwork", + description: "Codex running on the remote OpenWork worker", + mode: "primary", + native: true, + permission: [], + options: {}, + }]); + } + if (request.method === "GET" && path === "/command") return jsonResponse([]); + if (request.method === "GET" && path === "/permission") return jsonResponse(this.pendingPermissions()); + if (request.method === "GET" && path === "/question") return jsonResponse([]); + if (request.method === "GET" && path === "/session/status") return jsonResponse(await this.getStatus()); + if (request.method === "GET" && path === "/session") { + const limit = Number(url.searchParams.get("limit")); + return jsonResponse(await this.listSessions({ + limit: Number.isFinite(limit) && limit > 0 ? limit : undefined, + search: url.searchParams.get("search")?.trim() || undefined, + })); + } + if (request.method === "POST" && path === "/session") { + const body = await this.readBody(request); + return jsonResponse(await this.createSession({ + title: stringField(body, "title") || undefined, + model: modelIdFromBody(body), + })); + } + const messageMatch = path.match(/^\/session\/([^/]+)\/message$/); + if (request.method === "GET" && messageMatch?.[1]) { + return jsonResponse(await this.getMessages(decodeURIComponent(messageMatch[1]))); + } + const todoMatch = path.match(/^\/session\/([^/]+)\/todo$/); + if (request.method === "GET" && todoMatch?.[1]) return jsonResponse([]); + const childrenMatch = path.match(/^\/session\/([^/]+)\/children$/); + if (request.method === "GET" && childrenMatch?.[1]) return jsonResponse([]); + const diffMatch = path.match(/^\/session\/([^/]+)\/diff$/); + if (request.method === "GET" && diffMatch?.[1]) return jsonResponse([]); + const promptMatch = path.match(/^\/session\/([^/]+)\/prompt_async$/); + if (request.method === "POST" && promptMatch?.[1]) { + await this.startTurn(decodeURIComponent(promptMatch[1]), await this.readBody(request)); + return jsonResponse({}); + } + const abortMatch = path.match(/^\/session\/([^/]+)\/abort$/); + if (request.method === "POST" && abortMatch?.[1]) { + return jsonResponse(await this.abortTurn(decodeURIComponent(abortMatch[1]))); + } + const sessionMatch = path.match(/^\/session\/([^/]+)$/); + if (sessionMatch?.[1] && request.method === "GET") { + return jsonResponse(await this.getSession(decodeURIComponent(sessionMatch[1]))); + } + if (sessionMatch?.[1] && request.method === "DELETE") { + return jsonResponse(await this.deleteSession(decodeURIComponent(sessionMatch[1]))); + } + if (sessionMatch?.[1] && request.method === "PATCH") { + const body = await this.readBody(request); + const title = stringField(body, "title"); + if (!title) return errorResponse("title is required", 400); + return jsonResponse(await this.renameSession(decodeURIComponent(sessionMatch[1]), title)); + } + const legacyReply = path.match(/^\/permission\/([^/]+)\/reply$/); + const v2Reply = path.match(/^\/api\/session\/([^/]+)\/permission\/([^/]+)\/reply$/); + if (request.method === "POST" && (legacyReply?.[1] || v2Reply?.[2])) { + const requestId = decodeURIComponent(legacyReply?.[1] || v2Reply?.[2] || ""); + const body = await this.readBody(request); + return jsonResponse(this.replyPermission(requestId, stringField(body, "reply"))); + } + const v2PermissionList = path.match(/^\/api\/session\/([^/]+)\/permission$/); + if (request.method === "GET" && v2PermissionList?.[1]) { + return jsonResponse(this.pendingPermissions(decodeURIComponent(v2PermissionList[1]))); + } + return null; + } catch (error) { + return errorResponse(error); + } + } +} + +export class CodexRuntimeService { + private adapters = new Map(); + + constructor( + private readonly config: ServerConfig, + private readonly manager: CodexAppServerManager, + ) {} + + async isSelected(workspaceId: string): Promise { + return workspaceUsesCodexRuntime(this.config, workspaceId); + } + + adapterFor(workspace: WorkspaceInfo): CodexOpenCodeAdapter { + const existing = this.adapters.get(workspace.id); + if (existing) return existing; + const adapter = new CodexOpenCodeAdapter(workspace, this.manager.clientFor(workspace)); + this.adapters.set(workspace.id, adapter); + return adapter; + } + + async select(workspace: WorkspaceInfo, runtime: AgentRuntime): Promise { + if (runtime === "codex") { + await this.manager.clientFor(workspace).start(); + } else { + this.adapters.get(workspace.id)?.dispose(); + this.adapters.delete(workspace.id); + await this.manager.stopWorkspace(workspace.id); + } + await writeAgentRuntimeState(this.config, workspace.id, runtime); + return this.status(workspace); + } + + async status(workspace: WorkspaceInfo): Promise { + const { runtime } = await readAgentRuntimeState(this.config, workspace.id); + let version: string | null = null; + let error: string | null = null; + let available = false; + let accountResult: unknown = null; + let defaultModel: string | null = null; + let running = false; + let platform: string | null = null; + try { + if (runtime === "codex") { + const client = this.manager.clientFor(workspace); + await client.start(); + running = client.running; + version = client.metadata?.userAgent || null; + platform = client.metadata ? `${client.metadata.platformOs}/${client.metadata.platformFamily}` : null; + available = true; + accountResult = await client.request("account/read", { refreshToken: false }); + const modelResult = await client.request("model/list", { limit: 100, includeHidden: false }); + const models = isRecord(modelResult) ? arrayValue(modelResult.data) : []; + const preferred = models.find((entry) => isRecord(entry) && entry.isDefault === true) ?? models[0]; + if (isRecord(preferred)) defaultModel = stringField(preferred, "model") || stringField(preferred, "id") || null; + } else { + version = await probeCodexBinary(); + available = true; + } + } catch (statusError) { + error = statusError instanceof Error ? statusError.message : stringValue(statusError) || "Codex is unavailable"; + } + const account = objectField(accountResult, "account"); + const accountType = account && ["chatgpt", "apiKey", "amazonBedrock"].includes(stringField(account, "type")) + ? stringField(account, "type") + : null; + return { + runtime, + available, + experimental: true, + version, + error, + process: { + running, + healthy: running && !error, + transport: "stdio", + placement: "remote-worker", + publicPort: false, + platform, + }, + account: { + connected: account !== null, + type: accountType === "chatgpt" || accountType === "apiKey" || accountType === "amazonBedrock" + ? accountType + : null, + email: account ? stringField(account, "email") || null : null, + planType: account ? stringField(account, "planType") || null : null, + }, + defaultModel, + }; + } + + async startDeviceLogin(workspace: WorkspaceInfo): Promise { + const client = this.manager.clientFor(workspace); + await client.start(); + const result = await client.request("account/login/start", { type: "chatgptDeviceCode" }); + if (!isRecord(result) || result.type !== "chatgptDeviceCode") { + throw new Error("Codex did not return a device-code login"); + } + const loginId = stringField(result, "loginId"); + const verificationUrl = stringField(result, "verificationUrl"); + const userCode = stringField(result, "userCode"); + if (!loginId || !verificationUrl || !userCode) { + throw new Error("Codex returned an incomplete device-code login"); + } + return { loginId, verificationUrl, userCode }; + } + + async cancelLogin(workspace: WorkspaceInfo, loginId: string): Promise { + await this.manager.clientFor(workspace).request("account/login/cancel", { loginId }); + } + + async logout(workspace: WorkspaceInfo): Promise { + await this.manager.clientFor(workspace).request("account/logout", undefined); + } + + async handleProxy(input: { + workspace: WorkspaceInfo; + request: Request; + proxyPath: string; + url: URL; + }): Promise { + if (!(await this.isSelected(input.workspace.id))) return null; + return this.adapterFor(input.workspace).handleProxy(input.request, input.proxyPath, input.url); + } + + async stop(): Promise { + for (const adapter of this.adapters.values()) adapter.dispose(); + this.adapters.clear(); + await this.manager.stop(); + } +} diff --git a/apps/server/src/codex-runtime.e2e.test.ts b/apps/server/src/codex-runtime.e2e.test.ts new file mode 100644 index 0000000000..7734577b77 --- /dev/null +++ b/apps/server/src/codex-runtime.e2e.test.ts @@ -0,0 +1,390 @@ +import { afterEach, describe, expect, test } from "bun:test"; +import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { startServer } from "./server.js"; +import type { ServerConfig } from "./types.js"; + +const roots: string[] = []; +const stops: Array<() => void | Promise> = []; +const previousCodexBin = process.env.OPENWORK_CODEX_BIN; +const previousRuntimeDb = process.env.OPENWORK_RUNTIME_DB; + +afterEach(async () => { + while (stops.length) await stops.pop()?.(); + while (roots.length) { + const root = roots.pop(); + if (root) await rm(root, { recursive: true, force: true }); + } + if (previousCodexBin === undefined) delete process.env.OPENWORK_CODEX_BIN; + else process.env.OPENWORK_CODEX_BIN = previousCodexBin; + if (previousRuntimeDb === undefined) delete process.env.OPENWORK_RUNTIME_DB; + else process.env.OPENWORK_RUNTIME_DB = previousRuntimeDb; +}); + +async function createFakeCodex(root: string): Promise { + const path = join(root, "fake-codex.mjs"); + const source = String.raw`#!/usr/bin/env bun +import { createInterface } from "node:readline"; + +if (process.argv.includes("--version")) { + console.log("codex-cli 9.9.9-test"); + process.exit(0); +} + +let account = null; +let thread = null; +let pendingTurn = null; +const epoch = 1_700_000_000; + +function send(value) { + process.stdout.write(JSON.stringify(value) + "\n"); +} + +function makeThread() { + return { + id: "thread_remote_1", + sessionId: "thread_remote_1", + forkedFromId: null, + parentThreadId: null, + preview: "Inspect the remote worker", + ephemeral: false, + modelProvider: "openai", + createdAt: epoch, + updatedAt: epoch, + recencyAt: epoch, + status: { type: pendingTurn ? "active" : "idle", activeFlags: pendingTurn ? ["waitingForApproval"] : [] }, + path: null, + cwd: process.cwd(), + cliVersion: "9.9.9-test", + source: "appServer", + threadSource: null, + agentNickname: null, + agentRole: null, + gitInfo: null, + name: "Remote Codex check", + turns: pendingTurn ? [pendingTurn] : [], + }; +} + +function model() { + return { + id: "gpt-5.6-codex", + model: "gpt-5.6-codex", + upgrade: null, + upgradeInfo: null, + availabilityNux: null, + displayName: "GPT-5.6 Codex", + description: "Test Codex model", + hidden: false, + supportedReasoningEfforts: [{ reasoningEffort: "high", description: "High" }], + defaultReasoningEffort: "high", + inputModalities: ["text", "image"], + supportsPersonality: false, + additionalSpeedTiers: [], + serviceTiers: [], + defaultServiceTier: null, + isDefault: true, + }; +} + +function completeApprovedTurn() { + if (!pendingTurn) return; + const command = { + id: "item_command_1", + type: "commandExecution", + command: "hostname", + cwd: process.cwd(), + processId: null, + status: "completed", + commandActions: [], + aggregatedOutput: "remote-worker-test\n", + exitCode: 0, + durationMs: 5, + }; + const fileChange = { + id: "item_file_1", + type: "fileChange", + changes: [{ path: "server-proof.txt", kind: { type: "add" }, diff: "+created remotely" }], + status: "completed", + }; + const answer = { + id: "item_answer_1", + type: "agentMessage", + text: "The command ran on remote-worker-test.", + phase: "final_answer", + }; + pendingTurn.items = [pendingTurn.items[0], command, fileChange, answer]; + pendingTurn.status = "completed"; + pendingTurn.completedAt = epoch + 1; + pendingTurn.durationMs = 1_000; + send({ method: "item/completed", params: { threadId: "thread_remote_1", turnId: "turn_1", item: command, completedAtMs: (epoch + 1) * 1000 } }); + send({ method: "item/completed", params: { threadId: "thread_remote_1", turnId: "turn_1", item: fileChange, completedAtMs: (epoch + 1) * 1000 } }); + send({ method: "item/started", params: { threadId: "thread_remote_1", turnId: "turn_1", item: answer, startedAtMs: (epoch + 1) * 1000 } }); + send({ method: "item/agentMessage/delta", params: { threadId: "thread_remote_1", turnId: "turn_1", itemId: "item_answer_1", delta: answer.text } }); + send({ method: "item/completed", params: { threadId: "thread_remote_1", turnId: "turn_1", item: answer, completedAtMs: (epoch + 1) * 1000 } }); + send({ method: "turn/completed", params: { threadId: "thread_remote_1", turn: pendingTurn } }); + send({ method: "thread/status/changed", params: { threadId: "thread_remote_1", status: { type: "idle" } } }); +} + +const lines = createInterface({ input: process.stdin }); +lines.on("line", (line) => { + const message = JSON.parse(line); + if (message.id === 900 && !message.method) { + completeApprovedTurn(); + return; + } + if (!message.method || message.method === "initialized") return; + const id = message.id; + switch (message.method) { + case "initialize": + send({ id, result: { userAgent: "codex-cli/9.9.9-test", codexHome: process.env.CODEX_HOME, platformFamily: "unix", platformOs: "linux" } }); + return; + case "account/read": + send({ id, result: { account, requiresOpenaiAuth: true } }); + return; + case "account/login/start": + account = { type: "chatgpt", email: "server@example.test", planType: "plus" }; + send({ id, result: { type: "chatgptDeviceCode", loginId: "login_1", verificationUrl: "https://auth.openai.com/codex/device", userCode: "ABCD-EFGH" } }); + return; + case "account/login/cancel": + send({ id, result: {} }); + return; + case "account/logout": + account = null; + send({ id, result: {} }); + return; + case "model/list": + send({ id, result: { data: [model()], nextCursor: null } }); + return; + case "thread/start": + thread = makeThread(); + send({ id, result: { thread } }); + send({ method: "thread/started", params: { thread } }); + return; + case "thread/name/set": + if (thread) thread.name = message.params.name; + send({ id, result: {} }); + send({ method: "thread/name/updated", params: { threadId: "thread_remote_1", threadName: message.params.name } }); + return; + case "thread/list": + send({ id, result: { data: thread ? [makeThread()] : [], nextCursor: null } }); + return; + case "thread/read": + send({ id, result: { thread: makeThread() } }); + return; + case "thread/delete": + thread = null; + pendingTurn = null; + send({ id, result: {} }); + return; + case "turn/start": { + const userText = message.params.input.find((item) => item.type === "text")?.text || ""; + const user = { id: "item_user_1", type: "userMessage", content: [{ type: "text", text: userText }] }; + pendingTurn = { + id: "turn_1", + items: [user], + itemsView: { type: "full" }, + status: "inProgress", + error: null, + startedAt: epoch, + completedAt: null, + durationMs: null, + }; + send({ id, result: { turn: pendingTurn } }); + send({ method: "thread/status/changed", params: { threadId: "thread_remote_1", status: { type: "active", activeFlags: [] } } }); + send({ method: "turn/started", params: { threadId: "thread_remote_1", turn: pendingTurn } }); + setTimeout(() => { + const command = { + id: "item_command_1", + type: "commandExecution", + command: "hostname", + cwd: process.cwd(), + processId: null, + status: "inProgress", + commandActions: [], + aggregatedOutput: "", + exitCode: null, + durationMs: null, + }; + pendingTurn.items.push(command); + send({ method: "item/started", params: { threadId: "thread_remote_1", turnId: "turn_1", item: command, startedAtMs: epoch * 1000 } }); + send({ id: 900, method: "item/commandExecution/requestApproval", params: { threadId: "thread_remote_1", turnId: "turn_1", itemId: "item_command_1", command: "hostname", cwd: process.cwd(), reason: "Confirm remote execution" } }); + }, 5); + return; + } + case "turn/interrupt": + send({ id, result: {} }); + return; + default: + send({ id, error: { code: -32601, message: "Unsupported fake method " + message.method } }); + } +}); + +process.on("SIGTERM", () => process.exit(0)); +`; + await writeFile(path, source, "utf8"); + await chmod(path, 0o755); + return path; +} + +function auth(token: string): Record { + return { Authorization: `Bearer ${token}` }; +} + +async function waitFor(read: () => Promise, ready: (value: T) => boolean): Promise { + let value = await read(); + for (let index = 0; index < 50 && !ready(value); index++) { + await new Promise((resolve) => setTimeout(resolve, 10)); + value = await read(); + } + return value; +} + +describe("Codex server runtime", () => { + test("runs ChatGPT Codex sessions and approvals on the remote worker", async () => { + const root = await mkdtemp(join(tmpdir(), "openwork-codex-runtime-")); + roots.push(root); + process.env.OPENWORK_CODEX_BIN = await createFakeCodex(root); + process.env.OPENWORK_RUNTIME_DB = join(root, "runtime.sqlite"); + + const config: ServerConfig = { + host: "127.0.0.1", + port: 0, + token: "owt_codex_test", + hostToken: "owt_codex_host", + approval: { mode: "auto", timeoutMs: 1_000 }, + corsOrigins: ["*"], + workspaces: [{ + id: "ws_codex", + name: "Remote workspace", + path: root, + preset: "starter", + workspaceType: "local", + baseUrl: "http://127.0.0.1:1", + }], + authorizedRoots: [root], + readOnly: false, + startedAt: Date.now(), + tokenSource: "cli", + hostTokenSource: "cli", + logFormat: "pretty", + logRequests: false, + configPath: join(root, "server.json"), + }; + const server = await startServer(config); + stops.push(() => server.stop()); + const base = `http://127.0.0.1:${server.port}`; + const headers = { ...auth(config.token), "Content-Type": "application/json" }; + + const initial = await fetch(base + "/workspace/ws_codex/agent-runtime", { headers: auth(config.token) }); + expect(initial.status).toBe(200); + await expect(initial.json()).resolves.toMatchObject({ + runtime: "opencode", + available: true, + version: "codex-cli 9.9.9-test", + process: { running: false, transport: "stdio", placement: "remote-worker", publicPort: false }, + }); + + const selected = await fetch(base + "/workspace/ws_codex/agent-runtime", { + method: "PUT", + headers, + body: JSON.stringify({ runtime: "codex" }), + }); + expect(selected.status).toBe(200); + await expect(selected.json()).resolves.toMatchObject({ + runtime: "codex", + process: { running: true, healthy: true, transport: "stdio", publicPort: false }, + account: { connected: false }, + defaultModel: "gpt-5.6-codex", + }); + + const login = await fetch(base + "/workspace/ws_codex/agent-runtime/codex/login", { + method: "POST", + headers, + body: "{}", + }); + expect(login.status).toBe(201); + await expect(login.json()).resolves.toEqual({ + loginId: "login_1", + verificationUrl: "https://auth.openai.com/codex/device", + userCode: "ABCD-EFGH", + }); + + const connected = await fetch(base + "/workspace/ws_codex/agent-runtime", { headers: auth(config.token) }); + await expect(connected.json()).resolves.toMatchObject({ + account: { connected: true, type: "chatgpt", email: "server@example.test", planType: "plus" }, + }); + + const created = await fetch(base + "/workspace/ws_codex/sessions", { + method: "POST", + headers, + body: JSON.stringify({ title: "Remote Codex check", prompt: "Run hostname on the worker." }), + }); + expect(created.status).toBe(201); + await expect(created.json()).resolves.toMatchObject({ + item: { id: "thread_remote_1", title: "Remote Codex check", directory: root }, + started: true, + }); + + const readPermissions = async () => { + const response = await fetch(base + "/workspace/ws_codex/opencode/permission", { headers: auth(config.token) }); + return response.json(); + }; + const permissions = await waitFor(readPermissions, (value) => Array.isArray(value) && value.length === 1); + expect(permissions).toEqual([{ + id: "codex-900", + sessionID: "thread_remote_1", + action: "command.execute", + resources: ["hostname", root], + save: ["command.execute"], + metadata: { + runtime: "codex", + command: "hostname", + cwd: root, + reason: "Confirm remote execution", + }, + source: { messageID: "assistant:turn_1", callID: "item_command_1" }, + }]); + + const approved = await fetch(base + "/workspace/ws_codex/opencode/permission/codex-900/reply", { + method: "POST", + headers, + body: JSON.stringify({ reply: "once" }), + }); + expect(approved.status).toBe(200); + await expect(approved.json()).resolves.toBe(true); + + const readMessages = async () => { + const response = await fetch(base + "/workspace/ws_codex/sessions/thread_remote_1/messages", { + headers: auth(config.token), + }); + return response.json(); + }; + const messageResult = await waitFor( + readMessages, + (value) => JSON.stringify(value).includes("The command ran on remote-worker-test."), + ); + expect(messageResult).toMatchObject({ + items: [ + { info: { role: "user" }, parts: [{ type: "text", text: "Run hostname on the worker." }] }, + { + info: { role: "assistant", providerID: "codex" }, + parts: [ + { type: "tool", tool: "bash", state: { status: "completed", output: "remote-worker-test\n" } }, + { + type: "tool", + tool: "apply_patch", + state: { + status: "completed", + input: { patchText: "*** Add File: server-proof.txt\n+created remotely" }, + }, + }, + { type: "text", text: "The command ran on remote-worker-test." }, + ], + }, + ], + }); + }); +}); diff --git a/apps/server/src/routes/agent-runtime.ts b/apps/server/src/routes/agent-runtime.ts new file mode 100644 index 0000000000..1313d6db8a --- /dev/null +++ b/apps/server/src/routes/agent-runtime.ts @@ -0,0 +1,137 @@ +import { recordAudit } from "../audit.js"; +import type { AgentRuntime } from "../agent-runtime-store.js"; +import type { CodexRuntimeService } from "../codex-opencode-adapter.js"; +import { ApiError } from "../errors.js"; +import type { ServerConfig, TokenScope, WorkspaceInfo } from "../types.js"; +import { shortId } from "../utils.js"; +import { addRoute, type RequestContext, type Route } from "./registry.js"; + +type JsonResponse = (data: unknown, status?: number) => Response; +type ReadJsonBody = (request: Request) => Promise>; + +type RegisterAgentRuntimeRoutesOptions = { + routes: Route[]; + config: ServerConfig; + codexRuntime: CodexRuntimeService; + jsonResponse: JsonResponse; + readJsonBody: ReadJsonBody; + ensureWritable: (config: ServerConfig) => void; + requireClientScope: (ctx: RequestContext, required: TokenScope) => void; + resolveWorkspace: (config: ServerConfig, id: string) => Promise; +}; + +function requireWorkerWorkspace(workspace: WorkspaceInfo): void { + if (workspace.workspaceType !== "local" || !workspace.path.trim()) { + throw new ApiError( + 400, + "codex_runtime_workspace_unsupported", + "Codex Server must be configured on the OpenWork worker that owns this workspace", + ); + } +} + +function runtimeFromBody(body: Record): AgentRuntime { + if (body.runtime === "opencode" || body.runtime === "codex") return body.runtime; + throw new ApiError(400, "invalid_payload", "runtime must be opencode or codex"); +} + +function requiredString(body: Record, field: string): string { + const value = body[field]; + if (typeof value !== "string" || !value.trim()) { + throw new ApiError(400, "invalid_payload", `${field} is required`); + } + return value.trim(); +} + +function runtimeError(error: unknown): ApiError { + if (error instanceof ApiError) return error; + return new ApiError( + 503, + "codex_runtime_unavailable", + error instanceof Error ? error.message : "Codex runtime is unavailable", + ); +} + +export function registerAgentRuntimeRoutes(options: RegisterAgentRuntimeRoutesOptions): void { + const { + routes, + config, + codexRuntime, + jsonResponse, + readJsonBody, + ensureWritable, + requireClientScope, + resolveWorkspace, + } = options; + + addRoute(routes, "GET", "/workspace/:id/agent-runtime", "client", async (ctx) => { + const workspace = await resolveWorkspace(config, ctx.params.id); + requireWorkerWorkspace(workspace); + return jsonResponse(await codexRuntime.status(workspace)); + }); + + addRoute(routes, "PUT", "/workspace/:id/agent-runtime", "client", async (ctx) => { + ensureWritable(config); + requireClientScope(ctx, "collaborator"); + const workspace = await resolveWorkspace(config, ctx.params.id); + requireWorkerWorkspace(workspace); + const runtime = runtimeFromBody(await readJsonBody(ctx.request)); + try { + const status = await codexRuntime.select(workspace, runtime); + await recordAudit(workspace.path, { + id: shortId(), + workspaceId: workspace.id, + actor: ctx.actor ?? { type: "remote" }, + action: "agent-runtime.select", + target: runtime, + summary: `Selected ${runtime === "codex" ? "Codex Server" : "OpenCode"} runtime`, + timestamp: Date.now(), + }); + return jsonResponse(status); + } catch (error) { + throw runtimeError(error); + } + }); + + addRoute(routes, "POST", "/workspace/:id/agent-runtime/codex/login", "client", async (ctx) => { + ensureWritable(config); + requireClientScope(ctx, "collaborator"); + const workspace = await resolveWorkspace(config, ctx.params.id); + requireWorkerWorkspace(workspace); + if (!(await codexRuntime.isSelected(workspace.id))) { + throw new ApiError(409, "codex_runtime_not_selected", "Select Codex Server before connecting ChatGPT"); + } + try { + return jsonResponse(await codexRuntime.startDeviceLogin(workspace), 201); + } catch (error) { + throw runtimeError(error); + } + }); + + addRoute(routes, "POST", "/workspace/:id/agent-runtime/codex/login/cancel", "client", async (ctx) => { + ensureWritable(config); + requireClientScope(ctx, "collaborator"); + const workspace = await resolveWorkspace(config, ctx.params.id); + requireWorkerWorkspace(workspace); + const loginId = requiredString(await readJsonBody(ctx.request), "loginId"); + try { + await codexRuntime.cancelLogin(workspace, loginId); + return jsonResponse({ ok: true }); + } catch (error) { + throw runtimeError(error); + } + }); + + addRoute(routes, "POST", "/workspace/:id/agent-runtime/codex/logout", "client", async (ctx) => { + ensureWritable(config); + requireClientScope(ctx, "collaborator"); + const workspace = await resolveWorkspace(config, ctx.params.id); + requireWorkerWorkspace(workspace); + try { + await codexRuntime.logout(workspace); + return jsonResponse(await codexRuntime.status(workspace)); + } catch (error) { + throw runtimeError(error); + } + }); +} diff --git a/apps/server/src/routes/sessions.ts b/apps/server/src/routes/sessions.ts index 7c3a16210a..017551f412 100644 --- a/apps/server/src/routes/sessions.ts +++ b/apps/server/src/routes/sessions.ts @@ -1,4 +1,5 @@ import type { createOpencodeClient } from "@opencode-ai/sdk/v2/client"; +import type { CodexRuntimeService } from "../codex-opencode-adapter.js"; import { ApiError } from "../errors.js"; import { buildSession, buildSessionList, buildSessionMessages, buildSessionSnapshot } from "../session-read-model.js"; import { @@ -38,6 +39,7 @@ interface RegisterSessionRoutesOptions { resolveWorkspaceWithoutBootstrap: (config: ServerConfig, id: string) => Promise; createWorkspaceOpencodeClient: (config: ServerConfig, workspace: WorkspaceInfo) => WorkspaceOpencodeClient; unwrapOpencodeResult: UnwrapOpencodeResult; + codexRuntime: CodexRuntimeService; } function isRecord(value: unknown): value is Record { @@ -59,6 +61,7 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo resolveWorkspaceWithoutBootstrap, createWorkspaceOpencodeClient, unwrapOpencodeResult, + codexRuntime, } = options; const sessionGroupEvents = new SessionGroupEventStore(); @@ -82,6 +85,9 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo input: { roots?: boolean; start?: number; search?: string; limit?: number }, ) { try { + if (await codexRuntime.isSelected(workspace.id)) { + return buildSessionList(await codexRuntime.adapterFor(workspace).listSessions(input)); + } const opencode = createWorkspaceOpencodeClient(config, workspace); return buildSessionList( unwrapOpencodeResult( @@ -103,6 +109,14 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo workspace: WorkspaceInfo, input: { title: string; prompt?: string }, ) { + if (await codexRuntime.isSelected(workspace.id)) { + const adapter = codexRuntime.adapterFor(workspace); + const session = buildSession(await adapter.createSession({ title: input.title })); + if (input.prompt) { + await adapter.startTurn(session.id, { parts: [{ type: "text", text: input.prompt }] }); + } + return { item: session, started: Boolean(input.prompt) }; + } const opencode = createWorkspaceOpencodeClient(config, workspace); const session = buildSession( unwrapOpencodeResult( @@ -130,6 +144,9 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo async function readWorkspaceSession(workspace: WorkspaceInfo, sessionId: string) { try { + if (await codexRuntime.isSelected(workspace.id)) { + return buildSession(await codexRuntime.adapterFor(workspace).getSession(sessionId)); + } const opencode = createWorkspaceOpencodeClient(config, workspace); return buildSession( unwrapOpencodeResult( @@ -148,6 +165,9 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo input: { limit?: number }, ) { try { + if (await codexRuntime.isSelected(workspace.id)) { + return buildSessionMessages(await codexRuntime.adapterFor(workspace).getMessages(sessionId)); + } const opencode = createWorkspaceOpencodeClient(config, workspace); return buildSessionMessages( unwrapOpencodeResult( @@ -166,6 +186,15 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo input: { limit?: number }, ) { try { + if (await codexRuntime.isSelected(workspace.id)) { + const adapter = codexRuntime.adapterFor(workspace); + const [session, messages, statuses] = await Promise.all([ + adapter.getSession(sessionId), + adapter.getMessages(sessionId), + adapter.getStatus(), + ]); + return buildSessionSnapshot({ session, messages, todos: [], statuses }); + } const opencode = createWorkspaceOpencodeClient(config, workspace); const [session, messages, todos, statuses] = await Promise.all([ opencode.session @@ -415,11 +444,15 @@ export function registerSessionRoutes(options: RegisterSessionRoutesOptions): vo throw new ApiError(400, "invalid_payload", "sessionId is required"); } - const opencode = createWorkspaceOpencodeClient(config, workspace); - unwrapOpencodeResult( - await opencode.session.delete({ sessionID: sessionId }), - `/session/${encodeURIComponent(sessionId)}`, - ); + if (await codexRuntime.isSelected(workspace.id)) { + await codexRuntime.adapterFor(workspace).deleteSession(sessionId); + } else { + const opencode = createWorkspaceOpencodeClient(config, workspace); + unwrapOpencodeResult( + await opencode.session.delete({ sessionID: sessionId }), + `/session/${encodeURIComponent(sessionId)}`, + ); + } return jsonResponse({ ok: true }); }); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 2546826825..bd05cd8a2a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -65,6 +65,9 @@ import { addRoute, matchRoute, type AuthMode, type RequestContext, type Route } import { registerSessionRoutes } from "./routes/sessions.js"; import { registerWorkspaceRoutes } from "./routes/workspaces.js"; import { registerCloudMcpRoutes } from "./routes/cloud-mcp.js"; +import { registerAgentRuntimeRoutes } from "./routes/agent-runtime.js"; +import { CodexAppServerManager } from "./codex-app-server.js"; +import { CodexRuntimeService } from "./codex-opencode-adapter.js"; import { markOpenworkCloudMcpStale, reconcilePersistedOpenworkCloudMcp, @@ -707,7 +710,7 @@ function logRequest(input: { response: Response; durationMs: number; authMode: AuthMode; - proxyService?: "opencode"; + proxyService?: "opencode" | "codex"; proxyBaseUrl?: string; error?: string; }) { @@ -811,6 +814,10 @@ export async function startServer(config: ServerConfig): Promise { watcherHandle = startReloadWatchers({ config, reloadEvents, logger }); }; const engineMcpServerState = beginEngineMcpServerState(config); + const codexRuntime = new CodexRuntimeService( + config, + new CodexAppServerManager(config, SERVER_VERSION), + ); const routes = createRoutes( config, approvals, @@ -818,6 +825,7 @@ export async function startServer(config: ServerConfig): Promise { env, restartReloadWatchers, engineMcpServerState, + codexRuntime, ); const serverOptions: { @@ -831,7 +839,7 @@ export async function startServer(config: ServerConfig): Promise { const url = new URL(request.url); const startedAt = Date.now(); let authMode: AuthMode = "none"; - let proxyService: "opencode" | undefined; + let proxyService: "opencode" | "codex" | undefined; let proxyBaseUrl: string | undefined; let errorMessage: string | undefined; @@ -858,9 +866,16 @@ export async function startServer(config: ServerConfig): Promise { const actor = await requireClient(request, config, tokens); assertOpencodeProxyAllowed(actor, request.method, mount.restPath); const workspace = await resolveWorkspace(config, mount.workspaceId); - proxyService = "opencode"; + proxyService = await codexRuntime.isSelected(workspace.id) ? "codex" : "opencode"; proxyBaseUrl = workspace.baseUrl?.trim() || undefined; - const response = await proxyOpencodeRequest({ config, request, url, workspace, proxyPath: mount.restPath }); + const response = await proxyOpencodeRequest({ + config, + request, + url, + workspace, + proxyPath: mount.restPath, + codexRuntime, + }); return finalize(response); } catch (error) { const apiError = error instanceof ApiError @@ -908,8 +923,15 @@ export async function startServer(config: ServerConfig): Promise { try { const actor = await requireClient(request, config, tokens); assertOpencodeProxyAllowed(actor, request.method, url.pathname); - proxyService = "opencode"; - const response = await proxyOpencodeRequest({ config, request, url, workspace: config.workspaces[0] }); + const workspace = config.workspaces[0]; + proxyService = workspace && await codexRuntime.isSelected(workspace.id) ? "codex" : "opencode"; + const response = await proxyOpencodeRequest({ + config, + request, + url, + workspace, + codexRuntime, + }); return finalize(response); } catch (error) { const apiError = error instanceof ApiError @@ -995,6 +1017,7 @@ export async function startServer(config: ServerConfig): Promise { invalidateEngineMcpServerState(config, engineMcpServerState); watcherHandle.close(); reloadBaselineRefreshers.delete(config); + await codexRuntime.stop(); await server.stop(); }, }; @@ -1068,14 +1091,25 @@ async function proxyOpencodeRequest(input: { url: URL; workspace?: WorkspaceInfo; proxyPath?: string; + codexRuntime: CodexRuntimeService; }) { const workspace = input.workspace; + const proxyPath = input.proxyPath ?? input.url.pathname; + if (workspace) { + const codexResponse = await input.codexRuntime.handleProxy({ + workspace, + request: input.request, + proxyPath, + url: input.url, + }); + if (codexResponse) return codexResponse; + } + const baseUrl = workspace ? resolveWorkspaceOpencodeConnection(input.config, workspace).baseUrl?.trim() ?? "" : ""; if (!baseUrl) { throw new ApiError(400, "opencode_unconfigured", "OpenCode base URL is missing for this workspace"); } - const proxyPath = input.proxyPath ?? input.url.pathname; const targetUrl = buildOpencodeProxyUrl(baseUrl, proxyPath, input.url.search); const headers = new Headers(input.request.headers); headers.delete("authorization"); @@ -1479,6 +1513,7 @@ function createRoutes( env: EnvService, onWorkspacesChanged: () => void, engineMcpServerState: EngineMcpServerState, + codexRuntime: CodexRuntimeService, ): Route[] { const routes: Route[] = []; registerCoreRoutes({ @@ -1534,6 +1569,18 @@ function createRoutes( resolveWorkspaceWithoutBootstrap, createWorkspaceOpencodeClient, unwrapOpencodeResult, + codexRuntime, + }); + + registerAgentRuntimeRoutes({ + routes, + config, + codexRuntime, + jsonResponse, + readJsonBody, + ensureWritable, + requireClientScope, + resolveWorkspace, }); registerCloudMcpRoutes({ diff --git a/apps/server/src/session-read-model.ts b/apps/server/src/session-read-model.ts index 4f96121978..a4b9286c28 100644 --- a/apps/server/src/session-read-model.ts +++ b/apps/server/src/session-read-model.ts @@ -1,14 +1,4 @@ import { z } from "zod"; -import type { - SessionGetResponse, - SessionListResponse, - // The v1 `/session/{id}/message` route returns a bare message array; the - // `SessionMessagesResponse` name now belongs to the paginated v2 envelope. - SessionMessagesResponse2 as SessionMessagesArrayResponse, - SessionStatusResponse, - SessionTodoResponse, -} from "@opencode-ai/sdk/v2/client"; - import { ApiError } from "./errors.js"; const sessionTimeSchema = z @@ -107,31 +97,31 @@ function parseOrThrow(schema: z.ZodType, value: unknown, label: string): T }); } -export function buildSessionList(value: SessionListResponse): SessionInfoReadModel[] { +export function buildSessionList(value: unknown): SessionInfoReadModel[] { return parseOrThrow(sessionListSchema, value, "session list"); } -export function buildSession(value: SessionGetResponse): SessionInfoReadModel { +export function buildSession(value: unknown): SessionInfoReadModel { return parseOrThrow(sessionInfoSchema, value, "session"); } -export function buildSessionMessages(value: SessionMessagesArrayResponse): SessionMessageReadModel[] { +export function buildSessionMessages(value: unknown): SessionMessageReadModel[] { return parseOrThrow(sessionMessagesSchema, value, "session messages"); } -export function buildSessionTodos(value: SessionTodoResponse): SessionTodoReadModel[] { +export function buildSessionTodos(value: unknown): SessionTodoReadModel[] { return parseOrThrow(sessionTodosSchema, value, "session todos"); } -export function buildSessionStatuses(value: SessionStatusResponse): Record { +export function buildSessionStatuses(value: unknown): Record { return parseOrThrow(sessionStatusesSchema, value, "session statuses"); } export function buildSessionSnapshot(input: { - session: SessionGetResponse; - messages: SessionMessagesArrayResponse; - todos: SessionTodoResponse; - statuses: SessionStatusResponse; + session: unknown; + messages: unknown; + todos: unknown; + statuses: unknown; }): SessionSnapshotReadModel { const session = buildSession(input.session); const messages = buildSessionMessages(input.messages); diff --git a/evals/fixtures/fake-codex-app-server.mjs b/evals/fixtures/fake-codex-app-server.mjs new file mode 100755 index 0000000000..0f6e0f6694 --- /dev/null +++ b/evals/fixtures/fake-codex-app-server.mjs @@ -0,0 +1,332 @@ +#!/usr/bin/env bun +import { createInterface } from "node:readline"; +import { writeFileSync } from "node:fs"; + +if (process.argv.includes("--version")) { + console.log("codex-cli 9.9.9-fraimz"); + process.exit(0); +} + +let account = null; +let loginStarted = false; +let loginReads = 0; +let thread = null; +let pendingTurn = null; +const epoch = 1_700_000_000; + +function send(value) { + process.stdout.write(JSON.stringify(value) + "\n"); +} + +function makeThread() { + return { + id: "thread_remote_1", + sessionId: "thread_remote_1", + forkedFromId: null, + parentThreadId: null, + preview: "Inspect the remote worker", + ephemeral: false, + modelProvider: "openai", + createdAt: epoch, + updatedAt: epoch, + recencyAt: epoch, + status: { + type: pendingTurn?.status === "inProgress" ? "active" : "idle", + activeFlags: pendingTurn?.status === "inProgress" ? ["waitingForApproval"] : [], + }, + path: null, + cwd: process.cwd(), + cliVersion: "9.9.9-fraimz", + source: "appServer", + threadSource: null, + agentNickname: null, + agentRole: null, + gitInfo: null, + name: "Remote Codex check", + turns: pendingTurn ? [pendingTurn] : [], + }; +} + +function model() { + return { + id: "gpt-5.6-codex", + model: "gpt-5.6-codex", + upgrade: null, + upgradeInfo: null, + availabilityNux: null, + displayName: "GPT-5.6 Codex", + description: "Codex on the remote OpenWork worker", + hidden: false, + supportedReasoningEfforts: [ + { reasoningEffort: "high", description: "High" }, + ], + defaultReasoningEffort: "high", + inputModalities: ["text", "image"], + supportsPersonality: false, + additionalSpeedTiers: [], + serviceTiers: [], + defaultServiceTier: null, + isDefault: true, + }; +} + +function completeApprovedTurn() { + if (!pendingTurn) return; + writeFileSync("server-proof.md", "created on the remote worker\n", "utf8"); + const command = { + id: "item_command_1", + type: "commandExecution", + command: "hostname", + cwd: process.cwd(), + processId: null, + status: "completed", + commandActions: [], + aggregatedOutput: "openwork-remote-worker\n", + exitCode: 0, + durationMs: 5, + }; + const fileChange = { + id: "item_file_1", + type: "fileChange", + changes: [{ path: "server-proof.md", kind: "add", diff: "+created on the remote worker" }], + status: "completed", + }; + const answer = { + id: "item_answer_1", + type: "agentMessage", + text: "Done on the remote worker. I ran hostname and created server-proof.md.", + phase: "final_answer", + }; + pendingTurn.items = [pendingTurn.items[0], command, fileChange, answer]; + pendingTurn.status = "completed"; + pendingTurn.completedAt = epoch + 1; + pendingTurn.durationMs = 1_000; + send({ + method: "item/completed", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + item: command, + completedAtMs: (epoch + 1) * 1_000, + }, + }); + send({ + method: "item/completed", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + item: fileChange, + completedAtMs: (epoch + 1) * 1_000, + }, + }); + send({ + method: "item/started", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + item: answer, + startedAtMs: (epoch + 1) * 1_000, + }, + }); + send({ + method: "item/agentMessage/delta", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + itemId: "item_answer_1", + delta: answer.text, + }, + }); + send({ + method: "item/completed", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + item: answer, + completedAtMs: (epoch + 1) * 1_000, + }, + }); + send({ + method: "turn/completed", + params: { threadId: "thread_remote_1", turn: pendingTurn }, + }); + send({ + method: "thread/status/changed", + params: { threadId: "thread_remote_1", status: { type: "idle" } }, + }); +} + +const lines = createInterface({ input: process.stdin }); +lines.on("line", (line) => { + const message = JSON.parse(line); + if (message.id === 900 && !message.method) { + completeApprovedTurn(); + return; + } + if (!message.method || message.method === "initialized") return; + const id = message.id; + switch (message.method) { + case "initialize": + send({ + id, + result: { + userAgent: "codex-cli/9.9.9-fraimz", + codexHome: process.env.CODEX_HOME, + platformFamily: "unix", + platformOs: "linux", + }, + }); + return; + case "account/read": + if (loginStarted && !account) { + loginReads += 1; + if (loginReads >= 5) { + account = { + type: "chatgpt", + email: "server@example.test", + planType: "plus", + }; + } + } + send({ id, result: { account, requiresOpenaiAuth: true } }); + return; + case "account/login/start": + loginStarted = true; + loginReads = 0; + send({ + id, + result: { + type: "chatgptDeviceCode", + loginId: "login_fraimz", + verificationUrl: "https://auth.openai.com/codex/device", + userCode: "OPEN-WORK", + }, + }); + return; + case "account/login/cancel": + loginStarted = false; + loginReads = 0; + send({ id, result: {} }); + return; + case "account/logout": + account = null; + loginStarted = false; + loginReads = 0; + send({ id, result: {} }); + return; + case "model/list": + send({ id, result: { data: [model()], nextCursor: null } }); + return; + case "thread/start": + thread = makeThread(); + send({ id, result: { thread } }); + send({ method: "thread/started", params: { thread } }); + return; + case "thread/name/set": + if (thread) thread.name = message.params.name; + send({ id, result: {} }); + send({ + method: "thread/name/updated", + params: { + threadId: "thread_remote_1", + threadName: message.params.name, + }, + }); + return; + case "thread/list": + send({ + id, + result: { data: thread ? [makeThread()] : [], nextCursor: null }, + }); + return; + case "thread/read": + send({ id, result: { thread: makeThread() } }); + return; + case "thread/delete": + thread = null; + pendingTurn = null; + send({ id, result: {} }); + return; + case "turn/start": { + const userText = + message.params.input.find((item) => item.type === "text")?.text || ""; + const user = { + id: "item_user_1", + type: "userMessage", + content: [{ type: "text", text: userText }], + }; + pendingTurn = { + id: "turn_1", + items: [user], + itemsView: { type: "full" }, + status: "inProgress", + error: null, + startedAt: epoch, + completedAt: null, + durationMs: null, + }; + send({ id, result: { turn: pendingTurn } }); + send({ + method: "thread/status/changed", + params: { + threadId: "thread_remote_1", + status: { type: "active", activeFlags: [] }, + }, + }); + send({ + method: "turn/started", + params: { threadId: "thread_remote_1", turn: pendingTurn }, + }); + setTimeout(() => { + const command = { + id: "item_command_1", + type: "commandExecution", + command: "hostname", + cwd: process.cwd(), + processId: null, + status: "inProgress", + commandActions: [], + aggregatedOutput: "", + exitCode: null, + durationMs: null, + }; + pendingTurn.items.push(command); + send({ + method: "item/started", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + item: command, + startedAtMs: epoch * 1_000, + }, + }); + send({ + id: 900, + method: "item/commandExecution/requestApproval", + params: { + threadId: "thread_remote_1", + turnId: "turn_1", + itemId: "item_command_1", + command: "hostname", + cwd: process.cwd(), + reason: "Confirm execution on the remote worker", + }, + }); + }, 2_500); + return; + } + case "turn/interrupt": + send({ id, result: {} }); + return; + default: + send({ + id, + error: { + code: -32601, + message: "Unsupported fake method " + message.method, + }, + }); + } +}); + +process.on("SIGTERM", () => process.exit(0)); diff --git a/evals/flows/codex-server-runtime.flow.ts b/evals/flows/codex-server-runtime.flow.ts new file mode 100644 index 0000000000..8f702ff454 --- /dev/null +++ b/evals/flows/codex-server-runtime.flow.ts @@ -0,0 +1,287 @@ +import { defineFlow, type FlowContext } from "../runner/flow.ts"; +import { loadVoiceoverParagraphs } from "../runner/voiceover.ts"; + +const FLOW_ID = "codex-server-runtime"; +const TASK_PROMPT = "Run hostname and create server-proof.md on the remote worker."; +const FINAL_ANSWER = "Done on the remote worker. I ran hostname and created server-proof.md."; + +const vo = await loadVoiceoverParagraphs(FLOW_ID); +if (!vo) throw new Error(`Missing approved voice-over script for ${FLOW_ID}.`); + +let workspaceRoute = ""; + +async function dismissUnavailableModelPicker(ctx: FlowContext): Promise { + if (!(await ctx.hasText("No models available. Connect a provider to get started."))) return; + await ctx.clickText("Done"); + await ctx.waitFor( + `!document.body.innerText.includes("No models available. Connect a provider to get started.")`, + { timeoutMs: 10_000, label: "unavailable-model picker closes" }, + ); +} + +async function currentRoute(ctx: FlowContext): Promise { + const value = await ctx.eval("String(window.__openworkControl.snapshot().route || '')"); + ctx.assert(typeof value === "string", "OpenWork control route was not available."); + return typeof value === "string" ? value : ""; +} + +async function disconnectChatGptIfNeeded(ctx: FlowContext): Promise { + if (!(await ctx.hasText("ChatGPT connected ·"))) return; + const clicked = await ctx.eval(`(() => { + const buttons = Array.from(document.querySelectorAll("button")); + const button = buttons.find((candidate) => (candidate.textContent || "").trim() === "Disconnect"); + if (!button) return false; + button.click(); + return true; + })()`); + ctx.assert(clicked === true, "Could not disconnect the existing Codex ChatGPT account."); + await ctx.waitForText("ChatGPT not connected", { timeoutMs: 20_000 }); +} + +export default defineFlow({ + id: FLOW_ID, + title: "ChatGPT Codex executes on a remote OpenWork worker", + kind: "user-facing", + precondition: async (ctx) => { + await ctx.eval("location.reload()"); + await new Promise((resolve) => setTimeout(resolve, 750)); + await ctx.reconnect({ timeoutMs: 30_000 }); + await ctx.waitFor("Boolean(window.__openworkControl)", { + timeoutMs: 60_000, + label: "OpenWork control API", + }); + const route = await currentRoute(ctx); + const match = route.match(/^(\/workspace\/[^/]+)/); + if (!match) return "OpenWork is not currently inside a workspace."; + workspaceRoute = match[1] ?? ""; + return await ctx.hasText("Codex Remote Worker") + ? null + : "The Codex Remote Worker eval fixture is not connected."; + }, + steps: [ + { + name: "Remote worker is the active workspace", + run: async (ctx) => { + await ctx.prove("OpenWork is connected to the remote worker before any Codex task starts", { + voiceover: vo[0], + action: async () => { + await ctx.control("route.session"); + await dismissUnavailableModelPicker(ctx); + await ctx.waitForText("What do you need done?"); + }, + assert: async () => { + await ctx.expectRoute(`${workspaceRoute}/session`, { timeoutMs: 10_000 }); + await ctx.expectText("Codex Remote Worker"); + await ctx.expectText("Connected. No tasks found on this remote workspace."); + }, + screenshot: { + name: "remote-worker-connected", + requireText: ["Codex Remote Worker", "Connected. No tasks found on this remote workspace."], + }, + }); + }, + }, + { + name: "Codex Server runtime is selected", + run: async (ctx) => { + await ctx.prove("The remote worker exposes Codex Server as an experimental runtime alongside OpenCode", { + voiceover: vo[1], + action: async () => { + await ctx.control("route.settings.providers"); + await ctx.expectRoute(`${workspaceRoute}/settings/ai`, { timeoutMs: 10_000 }); + await ctx.waitFor(`(() => { + const button = Array.from(document.querySelectorAll("button")) + .find((candidate) => (candidate.textContent || "").includes("Codex Server")); + return Boolean(button && !button.disabled); + })()`, { timeoutMs: 40_000, label: "Codex runtime card enabled after worker status" }); + if (!(await ctx.hasText("Codex on remote worker"))) { + await ctx.clickText("Codex Server"); + } + await ctx.waitForText("Codex on remote worker"); + await disconnectChatGptIfNeeded(ctx); + }, + assert: async () => { + await ctx.expectText("Agent runtime"); + await ctx.expectText("Experimental"); + await ctx.expectText("OpenCode"); + await ctx.expectText("Codex Server"); + await ctx.expectText("Healthy"); + await ctx.expectText("ChatGPT not connected"); + }, + screenshot: { + name: "codex-runtime-selected", + requireText: ["Agent runtime", "Experimental", "Codex Server", "Healthy"], + }, + }); + }, + }, + { + name: "ChatGPT device sign-in starts on the worker", + run: async (ctx) => { + await ctx.prove("OpenWork starts ChatGPT device-code authentication for the server-hosted Codex runtime", { + voiceover: vo[2], + action: async () => { + // Keep the proof focused on OpenWork's device-code UI. Opening the + // external OpenAI page adds a Cloudflare-heavy tab that can starve + // CDP screenshots without changing the worker-side auth result. + await ctx.eval("window.open = () => null; true"); + await ctx.clickText("Connect ChatGPT"); + await ctx.waitForText("OPEN-WORK", { timeoutMs: 10_000 }); + }, + assert: async () => { + await ctx.expectText("Finish signing in to ChatGPT"); + await ctx.expectText("OPEN-WORK"); + await ctx.expectText("Credentials stay on the worker."); + await ctx.expectText("Waiting for ChatGPT authorization"); + }, + screenshot: { + name: "chatgpt-device-code", + requireText: ["Finish signing in to ChatGPT", "OPEN-WORK", "Credentials stay on the worker."], + }, + }); + }, + }, + { + name: "ChatGPT account is connected remotely", + run: async (ctx) => { + await ctx.prove("The worker reports the ChatGPT subscription and Codex provider as connected", { + voiceover: vo[3], + action: async () => { + await ctx.waitForText("ChatGPT connected · server@example.test", { timeoutMs: 30_000 }); + await ctx.waitForText("1 provider connected", { timeoutMs: 15_000 }); + }, + assert: async () => { + await ctx.expectText("Codex on remote worker"); + await ctx.expectText("ChatGPT connected · server@example.test"); + await ctx.expectText("1 provider connected"); + await ctx.expectText("Codex (ChatGPT)"); + }, + screenshot: { + name: "chatgpt-connected-on-worker", + requireText: ["ChatGPT connected · server@example.test", "1 provider connected", "Codex (ChatGPT)"], + }, + }); + }, + }, + { + name: "Codex starts the task on the server", + run: async (ctx) => { + await ctx.prove("A new OpenWork task reaches Codex and surfaces its remote command approval", { + voiceover: vo[4], + action: async () => { + await ctx.control("route.session"); + await dismissUnavailableModelPicker(ctx); + await ctx.waitForText("What do you need done?", { timeoutMs: 20_000 }); + await ctx.control("session.create_task"); + await ctx.expectRoute(`${workspaceRoute}/session/thread_remote_1`, { timeoutMs: 20_000 }); + await ctx.eval("location.reload()"); + await new Promise((resolve) => setTimeout(resolve, 750)); + await ctx.reconnect({ timeoutMs: 30_000 }); + await ctx.waitFor("Boolean(window.__openworkControl)", { + timeoutMs: 30_000, + label: "OpenWork control API for the remote task", + }); + await ctx.waitFor(`window.__openworkControl.listActions() + .some((candidate) => candidate.id === "composer.set_text" && !candidate.disabled)`, { + timeoutMs: 20_000, + label: "session composer available", + }); + await ctx.control("composer.set_text", { text: TASK_PROMPT }); + await ctx.waitFor(`window.__openworkControl.listActions() + .some((candidate) => candidate.id === "composer.send" && !candidate.disabled)`, { + timeoutMs: 10_000, + label: "session composer send enabled", + }); + await ctx.control("composer.send"); + await ctx.waitForText("Approve command.execute?", { timeoutMs: 20_000 }); + }, + assert: async () => { + await ctx.expectRoute(`${workspaceRoute}/session/thread_remote_1`, { timeoutMs: 20_000 }); + await ctx.expectText(TASK_PROMPT); + await ctx.expectText("hostname"); + await ctx.expectText("Allow once"); + }, + screenshot: { + name: "remote-command-awaits-approval", + requireText: [TASK_PROMPT, "hostname", "Allow once"], + }, + }); + }, + }, + { + name: "Permission continues the remote turn", + run: async (ctx) => { + await ctx.prove("Approving the Codex command in OpenWork lets the remote turn finish with its file change", { + voiceover: vo[5], + action: async () => { + await ctx.clickText("Allow once"); + await ctx.waitForText(FINAL_ANSWER, { timeoutMs: 20_000 }); + }, + assert: async () => { + await ctx.expectText(FINAL_ANSWER); + await ctx.expectText("server-proof.md"); + await ctx.expectNoText("Approve command.execute?"); + }, + screenshot: { + name: "remote-turn-completed", + requireText: [FINAL_ANSWER, "server-proof.md"], + rejectText: ["Approve command.execute?"], + }, + }); + }, + }, + { + name: "Conversation and server file survive a new client load", + run: async (ctx) => { + await ctx.prove("Reloading the client restores the same remote conversation and opens the file created on the worker", { + voiceover: vo[6], + action: async () => { + await ctx.eval("location.reload()"); + await new Promise((resolve) => setTimeout(resolve, 750)); + await ctx.reconnect({ timeoutMs: 30_000 }); + await ctx.waitFor("Boolean(window.__openworkControl)", { + timeoutMs: 30_000, + label: "OpenWork control API after reload", + }); + await ctx.waitForText(FINAL_ANSWER, { timeoutMs: 30_000 }); + await ctx.clickText("server-proof.md", { selector: "button" }); + await ctx.waitForText("created on the remote worker", { timeoutMs: 20_000 }); + }, + assert: async () => { + await ctx.expectRoute(`${workspaceRoute}/session/thread_remote_1`, { timeoutMs: 20_000 }); + await ctx.expectText(FINAL_ANSWER); + await ctx.expectText("created on the remote worker"); + }, + screenshot: { + name: "remote-history-and-file-restored", + requireText: [FINAL_ANSWER, "created on the remote worker"], + }, + }); + }, + }, + { + name: "Diagnostics prove the process lives on the worker", + run: async (ctx) => { + await ctx.prove("Runtime diagnostics show a healthy private stdio app-server on the remote worker", { + voiceover: vo[7], + action: async () => { + await ctx.control("route.settings.providers"); + await ctx.waitForText("Codex on remote worker", { timeoutMs: 20_000 }); + }, + assert: async () => { + await ctx.expectRoute(`${workspaceRoute}/settings/ai`, { timeoutMs: 10_000 }); + await ctx.expectText("Healthy"); + await ctx.expectText("Runtime: Codex app-server"); + await ctx.expectText("Location: Remote worker"); + await ctx.expectText("Transport: stdio (private)"); + await ctx.expectText("Public control port: None"); + }, + screenshot: { + name: "worker-runtime-diagnostics", + requireText: ["Healthy", "Location: Remote worker", "Transport: stdio (private)", "Public control port: None"], + }, + }); + }, + }, + ], +}); diff --git a/evals/voiceovers/codex-server-runtime.md b/evals/voiceovers/codex-server-runtime.md new file mode 100644 index 0000000000..9fdd765436 --- /dev/null +++ b/evals/voiceovers/codex-server-runtime.md @@ -0,0 +1,19 @@ +# codex-server-runtime — Codex runs on the remote OpenWork worker + +This demo proves that OpenWork can use a ChatGPT subscription through a Codex runtime hosted on a remote worker, without depending on the user's laptop. + +1. I connect OpenWork to a remote worker, so agent work can continue on the server even when this laptop is offline. + +2. In the worker settings I select the experimental Codex Server runtime, while OpenCode remains available as the default runtime. + +3. I click Connect ChatGPT and complete device-code sign-in with my ChatGPT subscription in the browser. + +4. OpenWork now shows Codex connected on the remote worker, making it clear where authentication and execution live. + +5. I start a task and see Codex stream its messages, commands, and file changes from the server. + +6. When Codex asks for permission, I approve the action directly in OpenWork and the task continues. + +7. I reopen OpenWork from another client and continue the same remote Codex conversation with its server-side files intact. + +8. The worker diagnostics show a healthy Codex app-server runtime, confirming that no Codex process or public control port is required on my laptop.