diff --git a/.changeset/codex-process-owner-pool.md b/.changeset/codex-process-owner-pool.md new file mode 100644 index 0000000..52a1b80 --- /dev/null +++ b/.changeset/codex-process-owner-pool.md @@ -0,0 +1,25 @@ +--- +"@pwrdrvr/agent-client": minor +--- + +Codex process owner + per-surface backend views + pool + +Adds `CodexProcessOwner` — one owned `codex app-server` process/connection that +hands out lightweight per-surface `CodexBackendView`s (each an `AgentBackend`), +so many chat surfaces share ONE Codex process. Every inbound notification and +every tool-call/approval server-request is demultiplexed by `threadId` to the +view that owns that thread, so sibling surfaces never clobber each other's +handlers (plain `CodexThreadClient` exposes a single global handler slot — that +clobber is exactly what this fixes). The owner also exposes `listModels()` and a +structured `runOneShot()` (outputSchema, local images, per-turn rollback, token +usage, abort) over the SAME connection, so model-picker refreshes and capture +enrichment no longer each spawn their own App Server. + +`CodexProcessOwnerPool` mirrors `AcpAgentClientPool` (`acquire`/`warm`/`has`/ +`release`/`closeAll`), keyed by the host on connection identity (command, +`CODEX_HOME`/env), giving one process per key with deduped concurrent warm-ups +and app-level shutdown. + +`CodexThreadClient` and `CodexOneShotClient` are now thin shims over the owner +(a single default view, and `runOneShot`/`listModels` respectively). Their public +APIs and behavior are unchanged — additive release. diff --git a/packages/agent-client/src/codex-oneshot-client.ts b/packages/agent-client/src/codex-oneshot-client.ts index 3afa4f0..e7cf1c8 100644 --- a/packages/agent-client/src/codex-oneshot-client.ts +++ b/packages/agent-client/src/codex-oneshot-client.ts @@ -1,67 +1,39 @@ // One-shot Codex App Server client for structured-output enrichment turns. // -// Drives a single short turn against a persistent worker thread: build a turn -// with caller-supplied prompt + JSON Schema (`outputSchema`), feed optional -// local images as file-path inputs, refuse any tool calls, await the assistant -// message, then `thread/rollback` the turn so the worker thread stays clean for -// the next request (a prompt-cache experiment from PwrSnap). +// Now a THIN SHIM over `CodexProcessOwner.runOneShot()` / `.listModels()`: it +// owns one Codex process and drives short structured turns against a persistent +// worker thread (build a turn with caller-supplied prompt + JSON Schema +// `outputSchema`, feed optional local images as file-path inputs, refuse any +// tool call, await the assistant message, then `thread/rollback` the turn so the +// worker thread stays clean for the next request — a prompt-cache experiment). // -// Ported from PwrSnap's CodexAppServerClient with the product specifics removed: -// • the prompt + JSON Schema + base instructions are CALLER-SUPPLIED per -// request (no CAPTURE_ENRICHMENT_SCHEMA / capture prompt baked in); -// • the logger is injected (agent-core `Logger`); -// • identity (`clientInfo.name`, default `serviceName`) is parameterized; -// • the binary is resolved via codex-discovery and spawned `["app-server"]`. +// The result is the RAW assistant text plus token usage — parsing/validating the +// JSON against the caller's schema is the caller's job. // -// The result is the RAW assistant text plus token usage — parsing/validating -// the JSON against the caller's schema is the caller's job. +// When a host already has a pooled `CodexProcessOwner` for chat, prefer +// `owner.runOneShot()` / `owner.listModels()` directly so enrichment + the model +// picker ride the SAME process instead of spawning this standalone one. -import { mkdir } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { join } from "node:path"; import { - noopLogger, - type Logger, - type NormalizedTokenUsage + type Logger } from "@pwrdrvr/agent-core"; import { - JsonRpcConnection, - StdioJsonRpcTransport, - type JsonRpcTransport -} from "@pwrdrvr/agent-transport"; -import { resolveCodexCommand } from "@pwrdrvr/codex-discovery"; -import type { - InitializeParams, - InitializeResponse, - ResponseItem, - ServerNotification -} from "@pwrdrvr/codex-app-server-protocol"; -import type { - DynamicToolCallResponse, - ItemCompletedNotification, - Model, - ModelListResponse, - ThreadStartResponse, - ThreadTokenUsage, - ThreadTokenUsageUpdatedNotification, - TurnCompletedNotification, - TurnStartResponse, - UserInput -} from "@pwrdrvr/codex-app-server-protocol/v2"; -import { normalizeTokenUsage } from "./normalize"; - -export type CodexOneShotTransportFactory = (command: string) => JsonRpcTransport; - -export type CodexModelOption = { - id: string; - model: string; - displayName: string; - description: string; - hidden: boolean; - inputModalities: Model["inputModalities"]; - defaultServiceTier: string | null; - isDefault: boolean; -}; + CodexProcessOwner, + type CodexProcessOwnerOptions, + type CodexModelOption, + type CodexOneShotRequest, + type CodexOneShotResponse, + type CodexTransportFactory +} from "./codex-process-owner"; + +export type CodexOneShotTransportFactory = CodexTransportFactory; + +// Re-exported from the owner (the source of truth). +export type { + CodexModelOption, + CodexOneShotRequest, + CodexOneShotResponse +} from "./codex-process-owner"; export type CodexOneShotClientOptions = { command?: string; @@ -84,522 +56,46 @@ export type CodexOneShotClientOptions = { logger?: Logger; }; -export type CodexOneShotRequest = { - /** The user-message text (the caller's prompt) for this turn. */ - prompt: string; - /** Local image file paths fed as `localImage` inputs (not inlined as base64). */ - imagePaths?: readonly string[]; - /** JSON Schema constraining the final assistant message (`outputSchema`). */ - outputSchema?: unknown; - /** Base instructions for the worker thread (set once; changing it re-creates - * the worker thread). */ - baseInstructions?: string; - /** Reasoning effort for the turn. Defaults to "low". */ - effort?: string; - model?: string | null; - /** Model provider (ThreadStartParams.modelProvider) for the worker thread. - * Omit for the Codex default. Part of the worker-thread cache key. */ - modelProvider?: string | null; - abortSignal?: AbortSignal; -}; - -export type CodexOneShotResponse = { - /** The raw assistant message text. Caller parses/validates against its schema. */ - rawText: string; - threadId: string; - turnId: string; - userAgent: string; - model: string; - modelProvider: string; - serviceTier: string | null; - tokenUsage: NormalizedTokenUsage | null; -}; - -type PendingTurn = { - threadId: string; - turnId: string; - agentMessages: string[]; - tokenUsage: ThreadTokenUsage | null; - resolve: (value: { rawText: string; tokenUsage: ThreadTokenUsage | null }) => void; - reject: (error: Error) => void; - timer: ReturnType; -}; - -type WorkerThread = { - threadId: string; - modelKey: string; - baseInstructions: string; - model: string; - modelProvider: string; - serviceTier: string | null; -}; - -const DEFAULT_CLIENT_NAME = "agent-kit"; -const DEFAULT_SERVICE_NAME = "agent-kit"; +/** Translate the one-shot options into owner options (folding the worker config + * under `owner.worker`). */ +function toOwnerOptions(options: CodexOneShotClientOptions): CodexProcessOwnerOptions { + const owner: CodexProcessOwnerOptions = {}; + if (options.command !== undefined) owner.command = options.command; + if (options.clientName !== undefined) owner.clientName = options.clientName; + if (options.clientTitle !== undefined) owner.clientTitle = options.clientTitle; + if (options.clientVersion !== undefined) owner.clientVersion = options.clientVersion; + if (options.serviceName !== undefined) owner.serviceName = options.serviceName; + if (options.requestTimeoutMs !== undefined) owner.requestTimeoutMs = options.requestTimeoutMs; + if (options.turnTimeoutMs !== undefined) owner.turnTimeoutMs = options.turnTimeoutMs; + if (options.env !== undefined) owner.env = options.env; + if (options.transportFactory !== undefined) owner.transportFactory = options.transportFactory; + if (options.logger !== undefined) owner.logger = options.logger; + const worker: NonNullable = {}; + if (options.workspaceDir !== undefined) worker.workspaceDir = options.workspaceDir; + if (options.workerThreadName !== undefined) worker.threadName = options.workerThreadName; + if (options.threadConfig !== undefined) worker.threadConfig = options.threadConfig; + if (Object.keys(worker).length > 0) owner.worker = worker; + return owner; +} export class CodexOneShotClient { - private readonly requestTimeoutMs: number; - private readonly turnTimeoutMs: number; - private readonly logger: Logger; - private readonly transportFactory: CodexOneShotTransportFactory | null; - private resolvedCommand: string | null = null; - private connection: JsonRpcConnection | null = null; - private initializeResponse: InitializeResponse | null = null; - private pendingTurn: PendingTurn | null = null; - private workerThread: WorkerThread | null = null; - private queue: Promise = Promise.resolve(); + private readonly owner: CodexProcessOwner; - constructor(private readonly options: CodexOneShotClientOptions = {}) { - this.requestTimeoutMs = options.requestTimeoutMs ?? 20_000; - this.turnTimeoutMs = options.turnTimeoutMs ?? 120_000; - this.logger = options.logger ?? noopLogger; - this.transportFactory = options.transportFactory ?? null; + constructor(options: CodexOneShotClientOptions = {}) { + this.owner = new CodexProcessOwner(toOwnerOptions(options)); } /** Run one structured-output turn. Calls are serialized — only one turn is in * flight at a time against the shared worker thread. */ async run(request: CodexOneShotRequest): Promise { - const run = this.queue.catch(() => undefined).then(() => this.runInner(request)); - this.queue = run.then( - () => undefined, - () => undefined - ); - return run; - } - - private async runInner(request: CodexOneShotRequest): Promise { - const connection = await this.getConnection(); - const initialized = await this.initialize(); - let thread: WorkerThread | null = null; - let turnId: string | null = null; - let rolledBack = false; - let aborted = false; - - const abortHandler = (): void => { - aborted = true; - if (thread && turnId) { - void connection - .request("turn/interrupt", { threadId: thread.threadId, turnId }, this.requestTimeoutMs) - .catch((error: unknown) => { - this.logger.warn("turn interrupt failed", { - error: error instanceof Error ? error.message : String(error) - }); - }); - } - }; - - request.abortSignal?.addEventListener("abort", abortHandler, { once: true }); - - try { - if (request.abortSignal?.aborted) { - throw new DOMException("one-shot turn aborted", "AbortError"); - } - - thread = await this.getWorkerThread( - request.model ?? null, - request.modelProvider ?? null, - request.baseInstructions ?? "" - ); - - const input: UserInput[] = [ - { type: "text", text: request.prompt, text_elements: [] }, - ...imagePathsToLocalImageInputs(request.imagePaths ?? []) - ]; - - const turnResponse = (await connection.request( - "turn/start", - { - threadId: thread.threadId, - model: request.model ?? null, - input, - effort: request.effort ?? "low", - ...(request.outputSchema !== undefined ? { outputSchema: request.outputSchema } : {}) - }, - this.requestTimeoutMs - )) as TurnStartResponse; - turnId = turnResponse.turn.id; - - if (request.abortSignal?.aborted || aborted) { - throw new DOMException("one-shot turn aborted", "AbortError"); - } - - const { rawText, tokenUsage } = await this.waitForTurn(thread.threadId, turnId); - await this.rollbackWorkerThread(thread.threadId); - rolledBack = true; - return { - rawText, - threadId: thread.threadId, - turnId, - userAgent: initialized.userAgent, - model: thread.model, - modelProvider: thread.modelProvider, - serviceTier: thread.serviceTier, - tokenUsage: tokenUsage === null ? null : normalizeTokenUsage(tokenUsage) - }; - } finally { - request.abortSignal?.removeEventListener("abort", abortHandler); - if (thread && turnId && !rolledBack) { - await this.rollbackWorkerThread(thread.threadId).catch((error: unknown) => { - this.logger.warn("worker thread rollback failed", { - threadId: thread?.threadId, - error: error instanceof Error ? error.message : String(error) - }); - }); - } - } + return this.owner.runOneShot(request); } async listModels(input: { includeHidden?: boolean } = {}): Promise { - const connection = await this.getConnection(); - await this.initialize(); - const models: CodexModelOption[] = []; - let cursor: string | null = null; - do { - const response = (await connection.request( - "model/list", - { cursor, limit: 100, includeHidden: input.includeHidden ?? false }, - this.requestTimeoutMs - )) as ModelListResponse; - models.push(...response.data.map(modelToOption)); - cursor = response.nextCursor; - } while (cursor !== null); - return models; + return this.owner.listModels(input); } async close(): Promise { - const connection = this.connection; - const thread = this.workerThread; - this.connection = null; - this.initializeResponse = null; - this.workerThread = null; - this.queue = Promise.resolve(); - if (connection) { - if (thread) { - await connection - .request("thread/archive", { threadId: thread.threadId }, this.requestTimeoutMs) - .catch((error: unknown) => { - this.logger.warn("thread archive failed", { - threadId: thread.threadId, - error: error instanceof Error ? error.message : String(error) - }); - }); - } - await connection.close(); - } - } - - // ---- worker-thread management ---- - - private async getWorkerThread( - model: string | null, - modelProvider: string | null, - baseInstructions: string - ): Promise { - const modelKey = `${model ?? "__default__"}@${modelProvider ?? "__default__"}::${baseInstructions}`; - if (this.workerThread?.modelKey === modelKey) { - return this.workerThread; - } - if (this.workerThread) { - const stale = this.workerThread; - this.workerThread = null; - const connection = await this.getConnection(); - await connection - .request("thread/archive", { threadId: stale.threadId }, this.requestTimeoutMs) - .catch((error: unknown) => { - this.logger.warn("thread archive failed", { - threadId: stale.threadId, - error: error instanceof Error ? error.message : String(error) - }); - }); - } - - const connection = await this.getConnection(); - const workspaceDir = await this.prepareWorkspace(); - const threadResponse = (await connection.request( - "thread/start", - { - model, - ...(modelProvider !== null ? { modelProvider } : {}), - ephemeral: false, - cwd: workspaceDir, - runtimeWorkspaceRoots: [workspaceDir], - serviceName: this.options.serviceName ?? DEFAULT_SERVICE_NAME, - approvalPolicy: "never", - sandbox: "read-only", - ...(baseInstructions.length > 0 ? { baseInstructions } : {}), - // Persistent worker thread for a prompt-cache experiment: keep the - // thread id stable across requests, then roll back each turn. The - // dedicated cwd keeps the worker out of any host repo/worktree. - ...(this.options.threadConfig !== undefined ? { config: this.options.threadConfig } : {}), - environments: [], - experimentalRawEvents: false, - persistExtendedHistory: false - }, - this.requestTimeoutMs - )) as ThreadStartResponse; - await this.clearThreadGitInfo(threadResponse.thread.id); - await this.setWorkerThreadName(threadResponse.thread.id); - this.workerThread = { - threadId: threadResponse.thread.id, - modelKey, - baseInstructions, - model: threadResponse.model, - modelProvider: threadResponse.modelProvider, - serviceTier: threadResponse.serviceTier - }; - return this.workerThread; - } - - private async rollbackWorkerThread(threadId: string): Promise { - const connection = await this.getConnection(); - await connection.request("thread/rollback", { threadId, numTurns: 1 }, this.requestTimeoutMs); - } - - private async prepareWorkspace(): Promise { - const workspaceDir = - this.options.workspaceDir ?? join(tmpdir(), "agent-kit", "oneshot-worker"); - await mkdir(workspaceDir, { recursive: true }); - return workspaceDir; - } - - private async clearThreadGitInfo(threadId: string): Promise { - const connection = await this.getConnection(); - await connection - .request( - "thread/metadata/update", - { threadId, gitInfo: { sha: null, branch: null, originUrl: null } }, - this.requestTimeoutMs - ) - .catch((error: unknown) => { - this.logger.warn("thread git metadata clear failed", { - threadId, - error: error instanceof Error ? error.message : String(error) - }); - }); + await this.owner.close(); } - - private async setWorkerThreadName(threadId: string): Promise { - const connection = await this.getConnection(); - await connection - .request( - "thread/name/set", - { threadId, name: this.options.workerThreadName ?? "agent-kit One-Shot Worker" }, - this.requestTimeoutMs - ) - .catch((error: unknown) => { - this.logger.warn("worker thread name set failed", { - threadId, - error: error instanceof Error ? error.message : String(error) - }); - }); - } - - // ---- connection / turn plumbing ---- - - private async resolveCommand(): Promise { - if (this.resolvedCommand !== null) return this.resolvedCommand; - const resolved = await resolveCodexCommand({ - command: this.options.command ?? "codex", - env: this.options.env ?? process.env - }); - this.resolvedCommand = resolved.command; - return resolved.command; - } - - private async initialize(): Promise { - if (this.initializeResponse) return this.initializeResponse; - const connection = await this.getConnection(); - const name = this.options.clientName ?? DEFAULT_CLIENT_NAME; - const params: InitializeParams = { - clientInfo: { - name, - title: this.options.clientTitle ?? name, - version: this.options.clientVersion ?? "0.0.0" - }, - capabilities: { - experimentalApi: true, - requestAttestation: false - } - }; - const response = (await connection.request( - "initialize", - params, - this.requestTimeoutMs - )) as InitializeResponse; - this.initializeResponse = response; - return response; - } - - private async getConnection(): Promise { - if (this.connection) return this.connection; - - let transport: JsonRpcTransport; - if (this.transportFactory !== null) { - transport = this.transportFactory(this.options.command ?? "codex"); - } else { - const command = await this.resolveCommand(); - transport = new StdioJsonRpcTransport({ - command, - args: ["app-server"], - ...(this.options.env !== undefined ? { env: this.options.env } : {}), - logger: this.logger - }); - } - - const connection = new JsonRpcConnection(transport, this.requestTimeoutMs, undefined, { - logger: this.logger, - logContext: { owner: "codex-oneshot-client" } - }); - connection.setNotificationHandler((method, params) => { - this.handleNotification(method, params); - }); - connection.setRequestHandler((method, params) => this.handleServerRequest(method, params)); - await connection.connect(); - this.connection = connection; - return connection; - } - - private waitForTurn( - threadId: string, - turnId: string - ): Promise<{ rawText: string; tokenUsage: ThreadTokenUsage | null }> { - if (this.pendingTurn) { - throw new Error("codex one-shot client already has an active turn"); - } - - return new Promise<{ rawText: string; tokenUsage: ThreadTokenUsage | null }>( - (resolve, reject) => { - const timer = setTimeout(() => { - this.pendingTurn = null; - reject(new Error("codex one-shot turn timed out")); - }, this.turnTimeoutMs); - this.pendingTurn = { - threadId, - turnId, - agentMessages: [], - tokenUsage: null, - resolve, - reject, - timer - }; - } - ); - } - - private handleNotification(method: string, params: unknown): void { - if (method === "item/completed") { - this.handleItemCompleted(params as ItemCompletedNotification); - return; - } - if (method === "rawResponseItem/completed") { - this.handleRawResponseItemCompleted(params as ServerNotification["params"]); - return; - } - if (method === "thread/tokenUsage/updated") { - this.handleThreadTokenUsageUpdated(params as ThreadTokenUsageUpdatedNotification); - return; - } - if (method === "turn/completed") { - this.handleTurnCompleted(params as TurnCompletedNotification); - } - } - - private handleItemCompleted(params: ItemCompletedNotification): void { - const pending = this.pendingTurn; - if (!pending || params.threadId !== pending.threadId || params.turnId !== pending.turnId) { - return; - } - if (params.item.type === "agentMessage") { - pending.agentMessages.push(params.item.text); - } - } - - private handleRawResponseItemCompleted(params: ServerNotification["params"]): void { - const pending = this.pendingTurn; - if (!pending || typeof params !== "object" || params === null) { - return; - } - const maybe = params as { threadId?: unknown; turnId?: unknown; item?: ResponseItem }; - if (maybe.threadId !== pending.threadId || maybe.turnId !== pending.turnId) { - return; - } - const item = maybe.item; - if (item?.type !== "message" || item.role !== "assistant") { - return; - } - const text = item.content - .filter((content) => content.type === "output_text") - .map((content) => content.text) - .join(""); - if (text) { - pending.agentMessages.push(text); - } - } - - private handleTurnCompleted(params: TurnCompletedNotification): void { - const pending = this.pendingTurn; - if (!pending || params.threadId !== pending.threadId || params.turn.id !== pending.turnId) { - return; - } - - clearTimeout(pending.timer); - this.pendingTurn = null; - - if (params.turn.status === "failed") { - pending.reject(new Error(params.turn.error?.message ?? "codex one-shot turn failed")); - return; - } - if (params.turn.status === "interrupted") { - pending.reject(new DOMException("one-shot turn aborted", "AbortError")); - return; - } - - const rawText = pending.agentMessages.at(-1)?.trim(); - if (!rawText) { - pending.reject(new Error("codex one-shot turn returned no assistant message")); - return; - } - pending.resolve({ rawText, tokenUsage: pending.tokenUsage }); - } - - private handleThreadTokenUsageUpdated(params: ThreadTokenUsageUpdatedNotification): void { - const pending = this.pendingTurn; - if (!pending || params.threadId !== pending.threadId || params.turnId !== pending.turnId) { - return; - } - pending.tokenUsage = params.tokenUsage; - } - - private async handleServerRequest(method: string, _params: unknown): Promise { - if (method === "item/tool/call") { - // One-shot enrichment exposes no tools — refuse any tool call. - return { - contentItems: [ - { type: "inputText", text: "This one-shot run does not expose tools." } - ], - success: false - } satisfies DynamicToolCallResponse; - } - this.logger.debug("unhandled codex server request", { method }); - return {}; - } -} - -function modelToOption(model: Model): CodexModelOption { - return { - id: model.id, - model: model.model, - displayName: model.displayName, - description: model.description, - hidden: model.hidden, - inputModalities: model.inputModalities, - defaultServiceTier: model.defaultServiceTier, - isDefault: model.isDefault - }; -} - -function imagePathsToLocalImageInputs(imagePaths: readonly string[]): UserInput[] { - // `localImage` lets App Server read the file as an image input. Do NOT inline a - // base64 data URL — the bridge can account that payload like fresh text. - return imagePaths.map((path) => ({ type: "localImage", path })); } diff --git a/packages/agent-client/src/codex-process-owner-pool.ts b/packages/agent-client/src/codex-process-owner-pool.ts new file mode 100644 index 0000000..66a90be --- /dev/null +++ b/packages/agent-client/src/codex-process-owner-pool.ts @@ -0,0 +1,128 @@ +// Lifecycle pool for `CodexProcessOwner` — the Codex counterpart of +// `AcpAgentClientPool`. A Codex App Server is a long-lived OS process with a +// stdio connection, and ONE owner hosts many concurrent threads (one per chat +// surface, plus model-listing and enrichment one-shots over the same socket). +// Without pooling, every consumer (Library chat, Sizzle chat, capture +// enrichment, a model-picker refresh) news up its own client → its own process; +// a careless caller can spawn several Codex processes for what should be one. +// The pool gives every caller the SAME warmed owner for a key, dedups +// concurrent warm-ups onto one spawn, and owns teardown. +// +// const pool = new CodexProcessOwnerPool(); +// // key on the connection identity the host varies — e.g. `(command, CODEX_HOME)`: +// const key = `${command}::${codexHome ?? "default"}`; +// pool.warm(key, () => new CodexProcessOwner({ command, env })); // startup +// const owner = await pool.acquire(key, factory); // surfaces — same instance +// const view = owner.createBackendView(); // per-surface backend + +import { noopLogger, type Logger } from "@pwrdrvr/agent-core"; +import type { CodexProcessOwner } from "./codex-process-owner"; + +export type CodexProcessOwnerFactory = () => CodexProcessOwner; + +export type CodexProcessOwnerPoolOptions = { + /** Max time to wait for an owner to warm (spawn + `initialize`) before an + * `acquire` rejects. Default 30s. */ + warmTimeoutMs?: number; + logger?: Logger; +}; + +type PoolEntry = { + owner: CodexProcessOwner; + /** Resolves to the owner once warmed; shared by concurrent acquires. */ + ready: Promise; +}; + +function errorMessage(cause: unknown): string { + return cause instanceof Error ? cause.message : String(cause); +} + +export class CodexProcessOwnerPool { + private readonly entries = new Map(); + private readonly warmTimeoutMs: number; + private readonly logger: Logger; + + constructor(options: CodexProcessOwnerPoolOptions = {}) { + this.warmTimeoutMs = options.warmTimeoutMs ?? 30_000; + this.logger = options.logger ?? noopLogger; + } + + /** + * Return the shared, warmed owner for `key`, creating it via `factory` if it + * doesn't exist. Concurrent acquires for the same key share ONE spawn — they + * receive the same in-flight promise, which resolves when Codex is ready or + * rejects if warm-up fails / times out. A failed warm-up evicts the entry so a + * later acquire retries with a fresh owner. + */ + async acquire(key: string, factory: CodexProcessOwnerFactory): Promise { + const existing = this.entries.get(key); + if (existing !== undefined) return existing.ready; + const owner = factory(); + const ready = this.warmWithTimeout(key, owner); + this.entries.set(key, { owner, ready }); + return ready; + } + + /** Fire-and-forget warm-up for app startup. Errors are logged, not thrown. */ + warm(key: string, factory: CodexProcessOwnerFactory): void { + void this.acquire(key, factory).catch((cause) => { + this.logger.warn?.("codex pool: background warm-up failed", { + key, + message: errorMessage(cause) + }); + }); + } + + /** Whether a (warming or warm) entry exists for `key`. */ + has(key: string): boolean { + return this.entries.has(key); + } + + /** Close + evict ONE key (e.g. its config changed). Safe if absent. */ + async release(key: string): Promise { + const entry = this.entries.get(key); + if (entry === undefined) return; + this.entries.delete(key); + // Don't let a still-warming entry surface an unhandled rejection on release. + entry.ready.catch(() => undefined); + await entry.owner.close().catch(() => undefined); + } + + /** Close + evict EVERY entry (app quit). */ + async closeAll(): Promise { + const entries = [...this.entries.values()]; + this.entries.clear(); + for (const entry of entries) entry.ready.catch(() => undefined); + await Promise.allSettled(entries.map((entry) => entry.owner.close())); + } + + private async warmWithTimeout( + key: string, + owner: CodexProcessOwner + ): Promise { + let timer: ReturnType | undefined; + try { + await Promise.race([ + owner.connect(), + new Promise((_, reject) => { + timer = setTimeout( + () => + reject( + new Error(`Codex owner "${key}" warm-up timed out after ${this.warmTimeoutMs}ms`) + ), + this.warmTimeoutMs + ); + }) + ]); + return owner; + } catch (cause) { + // Evict so a later acquire retries with a fresh owner, and tear down the + // half-spawned process. + if (this.entries.get(key)?.owner === owner) this.entries.delete(key); + await owner.close().catch(() => undefined); + throw cause; + } finally { + if (timer !== undefined) clearTimeout(timer); + } + } +} diff --git a/packages/agent-client/src/codex-process-owner.ts b/packages/agent-client/src/codex-process-owner.ts new file mode 100644 index 0000000..73efe4b --- /dev/null +++ b/packages/agent-client/src/codex-process-owner.ts @@ -0,0 +1,1076 @@ +// Codex App Server process OWNER + per-surface backend VIEWS. +// +// One `CodexProcessOwner` owns exactly ONE `codex app-server` child process and +// its JSON-RPC connection (keyed, by the host, on `(command, CODEX_HOME/env)`). +// Multiple chat surfaces — Library chat, Sizzle chat, capture enrichment, +// model-picker refresh — share that single process instead of each spawning +// their own. The owner multiplexes: +// +// • N `CodexBackendView`s — each a lightweight `AgentBackend` a surface +// drives (its own `onEvent` / `onToolCall` / `onApprovalRequest`). Inbound +// notifications AND tool-call/approval server-requests are demultiplexed by +// `threadId` to the view that owns that thread, so sibling surfaces never +// clobber each other's handlers. (Plain `CodexThreadClient` exposes a single +// global handler slot — that is exactly the clobber this fixes.) +// • `listModels()` — model listing over the shared connection (no +// extra process, no one-shot client needed). +// • `runOneShot()` — structured-output enrichment turns over a +// persistent worker thread on the SAME connection (outputSchema, local +// images, per-turn rollback, token usage, abort), so enrichment doesn't +// spawn a second App Server either. +// +// Every Codex notification and every server-request carries a `threadId` +// (`DynamicToolCallParams`, the v2 `*RequestApprovalParams`, and notifications +// all do; legacy v1 approvals carry `conversationId`) — that is what makes +// per-thread routing possible. + +import { mkdir } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + noopLogger, + type AgentBackend, + type AgentBackendApprovalHandler, + type AgentBackendStartThreadResult, + type AgentBackendToolCall, + type AgentBackendToolCallHandler, + type AgentForkThreadOptions, + type AgentStartThreadOptions, + type AgentStartTurnOptions, + type Logger, + type NormalizedApprovalDecision, + type NormalizedThreadEvent, + type NormalizedTokenUsage, + type Unsubscribe +} from "@pwrdrvr/agent-core"; +import { + JsonRpcConnection, + StdioJsonRpcTransport, + type JsonRpcTransport +} from "@pwrdrvr/agent-transport"; +import { resolveCodexCommand } from "@pwrdrvr/codex-discovery"; +import type { + InitializeParams, + InitializeResponse, + Personality, + ReasoningEffort, + ResponseItem, + ServerNotification +} from "@pwrdrvr/codex-app-server-protocol"; +import type { + AskForApproval, + DynamicToolCallResponse, + DynamicToolSpec, + ItemCompletedNotification, + Model, + ModelListResponse, + SandboxMode, + ThreadForkParams, + ThreadForkResponse, + ThreadResumeParams, + ThreadResumeResponse, + ThreadStartParams, + ThreadStartResponse, + ThreadTokenUsage, + ThreadTokenUsageUpdatedNotification, + TurnCompletedNotification, + TurnStartParams, + TurnStartResponse, + UserInput +} from "@pwrdrvr/codex-app-server-protocol/v2"; +import { + CODEX_APPROVAL_METHODS, + CODEX_TOOL_CALL_METHOD, + normalizeNotification, + normalizeTokenUsage +} from "./normalize"; + +export type { Unsubscribe } from "@pwrdrvr/agent-core"; + +export type CodexTransportFactory = (command: string) => JsonRpcTransport; + +/** Identity + lifecycle options for an owned Codex process. The connection + * identity (`clientName`/`env`/`command`) is also what a pool keys on. */ +export type CodexProcessOwnerOptions = { + /** Configured command to resolve. `"codex"` (or empty) triggers discovery. */ + command?: string; + /** Identity sent as `clientInfo.name` at `initialize`. Defaults to "agent-kit". */ + clientName?: string; + /** Human title sent as `clientInfo.title`. Defaults to `clientName`. */ + clientTitle?: string; + /** Version sent as `clientInfo.version`. Defaults to "0.0.0". */ + clientVersion?: string; + /** Default `serviceName` applied at `thread/start` when a caller omits one. */ + serviceName?: string; + requestTimeoutMs?: number; + turnTimeoutMs?: number; + /** Process env passed to the spawned codex. Defaults to `process.env`. */ + env?: NodeJS.ProcessEnv; + /** Override the transport (tests inject an in-memory fake). When supplied, no + * discovery/spawn happens. */ + transportFactory?: CodexTransportFactory; + logger?: Logger; + /** Config for the `runOneShot` persistent worker thread. */ + worker?: CodexOneShotWorkerOptions; +}; + +export type CodexOneShotWorkerOptions = { + /** Working dir for the persistent worker thread (keeps it out of any repo). */ + workspaceDir?: string; + /** Human-readable name set on the worker thread. */ + threadName?: string; + /** Per-thread Codex config overlay applied to the worker thread. */ + threadConfig?: Record; +}; + +/** + * Codex-NATIVE thread/start options. The neutral `AgentBackend.startThread` + * surface maps `AgentStartThreadOptions` onto this; it is also exposed via + * `CodexBackendView.startThreadNative` for hosts wanting full Codex control. + */ +export type CodexStartThreadOptions = { + approvalPolicy?: string; + sandbox?: string; + baseInstructions?: string; + dynamicTools?: DynamicToolSpec[]; + cwd?: string; + runtimeWorkspaceRoots?: string[]; + serviceName?: string; + personality?: string; + /** Model id to start the thread on. Omit for the Codex default. */ + model?: string; + /** Model provider id. Omit for the Codex default. */ + modelProvider?: string; + /** Service tier, when the host pins one. */ + serviceTier?: string; + /** Per-thread Codex config overlay (the `-c key=value` mechanism). */ + config?: Record; + /** Thread environments. **Empty array disables exec-environment access**. */ + environments?: unknown[]; +}; + +/** Codex-NATIVE turn/start options (the native mapping target of `startTurn`). */ +export type CodexStartTurnOptions = { + threadId: string; + input: UserInput[]; + effort?: string; +}; + +/** Tool-call server-request handler (canonical neutral shape). */ +export type CodexToolCallHandler = AgentBackendToolCallHandler; +/** Approval server-request handler (canonical neutral shape). */ +export type CodexApprovalHandler = AgentBackendApprovalHandler; + +export type StartThreadResult = { + threadId: string; + model: string; + modelProvider: string; + serviceTier: string | null; +}; + +export type CodexModelOption = { + id: string; + model: string; + displayName: string; + description: string; + hidden: boolean; + inputModalities: Model["inputModalities"]; + defaultServiceTier: string | null; + isDefault: boolean; +}; + +export type CodexOneShotRequest = { + /** The user-message text (the caller's prompt) for this turn. */ + prompt: string; + /** Local image file paths fed as `localImage` inputs (not inlined as base64). */ + imagePaths?: readonly string[]; + /** JSON Schema constraining the final assistant message (`outputSchema`). */ + outputSchema?: unknown; + /** Base instructions for the worker thread (changing it re-creates it). */ + baseInstructions?: string; + /** Reasoning effort for the turn. Defaults to "low". */ + effort?: string; + model?: string | null; + /** Model provider for the worker thread. Omit for the Codex default. */ + modelProvider?: string | null; + abortSignal?: AbortSignal; +}; + +export type CodexOneShotResponse = { + /** The raw assistant message text. Caller parses/validates against its schema. */ + rawText: string; + threadId: string; + turnId: string; + userAgent: string; + model: string; + modelProvider: string; + serviceTier: string | null; + tokenUsage: NormalizedTokenUsage | null; +}; + +const DEFAULT_CLIENT_NAME = "agent-kit"; +const DEFAULT_SERVICE_NAME = "agent-kit"; + +/** Map a neutral approval decision onto the generic Codex approval response. */ +function approvalResponseFor(decision: NormalizedApprovalDecision): unknown { + switch (decision) { + case "approved": + return { decision: "approved" }; + case "abort": + return { decision: "abort" }; + case "denied": + default: + return { decision: "denied" }; + } +} + +/** A tool call arriving with no handler registered for its thread. */ +function noToolHandlerResponse(): DynamicToolCallResponse { + return { + contentItems: [ + { type: "inputText", text: "No tool handler is registered for this thread." } + ], + success: false + }; +} + +/** The threadId a normalized event belongs to (for view routing). Most events + * carry `threadId`; `thread_settings` nests it under `settings`. */ +function eventThreadId(event: NormalizedThreadEvent): string | undefined { + if (event.kind === "thread_settings") return event.settings.threadId; + if ("threadId" in event) return event.threadId; + return undefined; +} + +/** The threadId a server-request belongs to. v2 params carry `threadId`; legacy + * v1 approvals (`applyPatchApproval`/`execCommandApproval`) carry `conversationId`. */ +function requestThreadId(params: unknown): string | undefined { + if (typeof params !== "object" || params === null) return undefined; + const p = params as Record; + if (typeof p.threadId === "string") return p.threadId; + if (typeof p.conversationId === "string") return p.conversationId; + return undefined; +} + +function modelToOption(model: Model): CodexModelOption { + return { + id: model.id, + model: model.model, + displayName: model.displayName, + description: model.description, + hidden: model.hidden, + inputModalities: model.inputModalities, + defaultServiceTier: model.defaultServiceTier, + isDefault: model.isDefault + }; +} + +function imagePathsToLocalImageInputs(imagePaths: readonly string[]): UserInput[] { + // `localImage` lets App Server read the file as an image input. Do NOT inline a + // base64 data URL — the bridge can account that payload like fresh text. + return imagePaths.map((path) => ({ type: "localImage", path })); +} + +type PendingOneShotTurn = { + threadId: string; + turnId: string; + agentMessages: string[]; + tokenUsage: ThreadTokenUsage | null; + resolve: (value: { rawText: string; tokenUsage: ThreadTokenUsage | null }) => void; + reject: (error: Error) => void; + timer: ReturnType; +}; + +type WorkerThread = { + threadId: string; + modelKey: string; + baseInstructions: string; + model: string; + modelProvider: string; + serviceTier: string | null; +}; + +/** + * A per-surface `AgentBackend` over a shared `CodexProcessOwner`. Each view owns + * the threads it starts/forks/resumes/turns on; the owner routes that thread's + * events + tool-call/approval requests back to THIS view's handlers, so two + * surfaces sharing one process never clobber each other. + * + * Created via `owner.createBackendView()`. `close()` detaches the view (drops + * its handlers + releases its threads from routing) — it does NOT stop the + * shared process. Tear the process down via `owner.close()`. + */ +export class CodexBackendView implements AgentBackend { + private readonly eventListeners = new Set<(event: NormalizedThreadEvent) => void>(); + private readonly ownedThreads = new Set(); + private closed = false; + /** @internal — read by the owner's server-request router. */ + toolCallHandler: CodexToolCallHandler | null = null; + /** @internal */ approvalHandler: CodexApprovalHandler | null = null; + + constructor(private readonly owner: CodexProcessOwner) {} + + onEvent(cb: (event: NormalizedThreadEvent) => void): Unsubscribe { + this.eventListeners.add(cb); + return () => { + this.eventListeners.delete(cb); + }; + } + + onToolCall(handler: CodexToolCallHandler): Unsubscribe { + this.toolCallHandler = handler; + return () => { + if (this.toolCallHandler === handler) this.toolCallHandler = null; + }; + } + + onApprovalRequest(handler: CodexApprovalHandler): Unsubscribe { + this.approvalHandler = handler; + return () => { + if (this.approvalHandler === handler) this.approvalHandler = null; + }; + } + + /** @internal — the owner delivers a routed event to this view. */ + emit(event: NormalizedThreadEvent): void { + for (const listener of this.eventListeners) listener(event); + } + + /** Neutral `AgentBackend.startThread`. Maps onto `CodexStartThreadOptions`. */ + async startThread(opts: AgentStartThreadOptions = {}): Promise { + const native: CodexStartThreadOptions = {}; + if (opts.instructions !== undefined) native.baseInstructions = opts.instructions; + if (opts.cwd !== undefined) native.cwd = opts.cwd; + if (opts.workspaceRoots !== undefined) native.runtimeWorkspaceRoots = [...opts.workspaceRoots]; + if (opts.model !== undefined) native.model = opts.model; + if (opts.modelProvider !== undefined) native.modelProvider = opts.modelProvider; + if (opts.serviceTier != null) native.serviceTier = opts.serviceTier; + if (opts.approvalPolicy !== undefined) native.approvalPolicy = opts.approvalPolicy; + if (opts.sandbox !== undefined) native.sandbox = opts.sandbox; + if (opts.serviceName !== undefined) native.serviceName = opts.serviceName; + if (opts.config !== undefined) native.config = opts.config; + if (opts.environments !== undefined) native.environments = opts.environments; + if (opts.tools !== undefined) native.dynamicTools = opts.tools as DynamicToolSpec[]; + return this.startThreadNative(native); + } + + /** Codex-native thread/start. */ + async startThreadNative(opts: CodexStartThreadOptions = {}): Promise { + const result = await this.owner.startThreadNative(opts); + this.claim(result.threadId); + return result; + } + + async forkThread(opts: AgentForkThreadOptions): Promise { + const result = await this.owner.forkThread(opts); + this.claim(result.threadId); + return result; + } + + async resumeThread(threadId: string): Promise { + this.claim(threadId); + await this.owner.resumeThread(threadId); + } + + async clearThreadGitInfo(threadId: string): Promise { + await this.owner.clearThreadGitInfo(threadId); + } + + /** Neutral `AgentBackend.startTurn`. Maps text + image paths onto `UserInput[]`. */ + async startTurn(opts: AgentStartTurnOptions): Promise<{ turnId: string }> { + const input: UserInput[] = [{ type: "text", text: opts.input.text, text_elements: [] }]; + for (const path of opts.input.imagePaths ?? []) input.push({ type: "localImage", path }); + const native: CodexStartTurnOptions = { threadId: opts.threadId, input }; + if (opts.reasoning !== undefined) native.effort = opts.reasoning; + return this.startTurnNative(native); + } + + /** Codex-native turn/start. */ + async startTurnNative(opts: CodexStartTurnOptions): Promise<{ turnId: string }> { + // Claim the thread we're turning on — covers threads resumed from + // persistence (the host never called startThread on this view for them). + this.claim(opts.threadId); + return this.owner.startTurnNative(opts); + } + + async interruptTurn(threadId: string): Promise { + await this.owner.interruptTurn(threadId); + } + + async archiveThread(threadId: string): Promise { + await this.owner.archiveThread(threadId); + this.release(threadId); + } + + /** Detach this view: drop handlers + release its threads from routing. Does + * NOT stop the shared process. Idempotent. */ + async close(): Promise { + if (this.closed) return; + this.closed = true; + this.owner.detachView(this); + this.eventListeners.clear(); + this.toolCallHandler = null; + this.approvalHandler = null; + } + + /** @internal */ ownedThreadIds(): ReadonlySet { + return this.ownedThreads; + } + + private claim(threadId: string): void { + this.ownedThreads.add(threadId); + this.owner.registerThread(threadId, this); + } + + private release(threadId: string): void { + this.ownedThreads.delete(threadId); + this.owner.unregisterThread(threadId, this); + } +} + +/** + * Owns one Codex App Server process + connection and hands out per-surface + * `CodexBackendView`s, model listing, and structured one-shot turns over the + * single shared connection. + */ +export class CodexProcessOwner { + private readonly requestTimeoutMs: number; + private readonly turnTimeoutMs: number; + private readonly logger: Logger; + private readonly transportFactory: CodexTransportFactory | null; + private resolvedCommand: string | null = null; + private connection: JsonRpcConnection | null = null; + private initializeResponse: InitializeResponse | null = null; + + // threadId → owning view, for routing notifications + server-requests. + private readonly threadOwners = new Map(); + private readonly views = new Set(); + private readonly loadedThreadIds = new Set(); + + // one-shot worker-thread machinery (shares the connection). + private pendingTurn: PendingOneShotTurn | null = null; + private workerThread: WorkerThread | null = null; + private oneShotQueue: Promise = Promise.resolve(); + + constructor(private readonly options: CodexProcessOwnerOptions = {}) { + this.requestTimeoutMs = options.requestTimeoutMs ?? 20_000; + this.turnTimeoutMs = options.turnTimeoutMs ?? 120_000; + this.logger = options.logger ?? noopLogger; + this.transportFactory = options.transportFactory ?? null; + } + + /** Warm the process: spawn + `initialize` without opening any thread. Lets a + * pool hold an owner ready from startup. */ + async connect(): Promise { + await this.getConnection(); + await this.initialize(); + } + + /** Mint a lightweight per-surface backend. */ + createBackendView(): CodexBackendView { + const view = new CodexBackendView(this); + this.views.add(view); + return view; + } + + /** List available models over the shared connection. */ + async listModels(input: { includeHidden?: boolean } = {}): Promise { + const connection = await this.getConnection(); + await this.initialize(); + const models: CodexModelOption[] = []; + let cursor: string | null = null; + do { + const response = (await connection.request( + "model/list", + { cursor, limit: 100, includeHidden: input.includeHidden ?? false }, + this.requestTimeoutMs + )) as ModelListResponse; + models.push(...response.data.map(modelToOption)); + cursor = response.nextCursor; + } while (cursor !== null); + return models; + } + + /** Run one structured-output turn against a persistent worker thread on the + * shared connection. Calls are serialized — one in flight at a time. */ + async runOneShot(request: CodexOneShotRequest): Promise { + const run = this.oneShotQueue.catch(() => undefined).then(() => this.runOneShotInner(request)); + this.oneShotQueue = run.then( + () => undefined, + () => undefined + ); + return run; + } + + async close(): Promise { + const connection = this.connection; + const worker = this.workerThread; + this.connection = null; + this.initializeResponse = null; + this.workerThread = null; + this.oneShotQueue = Promise.resolve(); + this.threadOwners.clear(); + this.views.clear(); + this.loadedThreadIds.clear(); + if (connection) { + if (worker) { + await connection + .request("thread/archive", { threadId: worker.threadId }, this.requestTimeoutMs) + .catch((error: unknown) => { + this.logger.warn("worker thread archive failed", { + threadId: worker.threadId, + error: error instanceof Error ? error.message : String(error) + }); + }); + } + await connection.close(); + } + } + + // ---- view-driven thread ops (called by CodexBackendView) ---- + + /** @internal */ + async startThreadNative(opts: CodexStartThreadOptions = {}): Promise { + const connection = await this.getConnection(); + await this.initialize(); + + const params: ThreadStartParams = { + experimentalRawEvents: false, + persistExtendedHistory: false + }; + if (opts.cwd !== undefined) params.cwd = opts.cwd; + if (opts.model !== undefined) params.model = opts.model; + if (opts.modelProvider !== undefined) params.modelProvider = opts.modelProvider; + if (opts.serviceTier !== undefined) params.serviceTier = opts.serviceTier; + if (opts.runtimeWorkspaceRoots !== undefined) { + params.runtimeWorkspaceRoots = opts.runtimeWorkspaceRoots; + } + params.serviceName = opts.serviceName ?? this.options.serviceName ?? DEFAULT_SERVICE_NAME; + if (opts.approvalPolicy !== undefined) params.approvalPolicy = opts.approvalPolicy as AskForApproval; + if (opts.sandbox !== undefined) params.sandbox = opts.sandbox as SandboxMode; + if (opts.baseInstructions !== undefined) params.baseInstructions = opts.baseInstructions; + if (opts.personality !== undefined) params.personality = opts.personality as Personality; + if (opts.dynamicTools !== undefined) params.dynamicTools = opts.dynamicTools; + if (opts.config !== undefined) { + params.config = opts.config as NonNullable; + } + if (opts.environments !== undefined) { + params.environments = opts.environments as NonNullable; + } + + const response = (await connection.request( + "thread/start", + params, + this.requestTimeoutMs + )) as ThreadStartResponse; + const threadId = response.thread.id; + this.loadedThreadIds.add(threadId); + this.logger.debug("thread started", { threadId }); + return { + threadId, + model: response.model, + modelProvider: response.modelProvider, + serviceTier: response.serviceTier + }; + } + + /** @internal */ + async forkThread(opts: AgentForkThreadOptions): Promise { + const connection = await this.getConnection(); + await this.initialize(); + + const params: ThreadForkParams = { + threadId: opts.sourceThreadId, + persistExtendedHistory: false + }; + if (opts.cwd !== undefined) params.cwd = opts.cwd; + if (opts.workspaceRoots !== undefined) params.runtimeWorkspaceRoots = [...opts.workspaceRoots]; + if (opts.instructions !== undefined) params.baseInstructions = opts.instructions; + if (opts.approvalPolicy !== undefined) params.approvalPolicy = opts.approvalPolicy as AskForApproval; + if (opts.sandbox !== undefined) params.sandbox = opts.sandbox as SandboxMode; + if (opts.config !== undefined) { + params.config = opts.config as NonNullable; + } + + const response = (await connection.request( + "thread/fork", + params, + this.requestTimeoutMs + )) as ThreadForkResponse; + const threadId = response.thread.id; + this.loadedThreadIds.add(threadId); + return { + threadId, + model: response.model, + modelProvider: response.modelProvider, + serviceTier: response.serviceTier + }; + } + + /** @internal */ + async resumeThread(threadId: string): Promise { + if (this.loadedThreadIds.has(threadId)) return; + const connection = await this.getConnection(); + await this.initialize(); + const params: ThreadResumeParams = { threadId, persistExtendedHistory: false }; + const response = (await connection.request( + "thread/resume", + params, + this.requestTimeoutMs + )) as ThreadResumeResponse; + this.loadedThreadIds.add(response.thread.id); + this.logger.debug("thread resumed", { threadId: response.thread.id }); + } + + /** @internal */ + async clearThreadGitInfo(threadId: string): Promise { + const connection = await this.getConnection(); + await this.initialize(); + await connection.request( + "thread/metadata/update", + { threadId, gitInfo: { sha: null, branch: null, originUrl: null } }, + this.requestTimeoutMs + ); + } + + /** @internal */ + async startTurnNative(opts: CodexStartTurnOptions): Promise<{ turnId: string }> { + await this.resumeThread(opts.threadId); + const connection = await this.getConnection(); + await this.initialize(); + + const params: TurnStartParams = { threadId: opts.threadId, input: opts.input }; + if (opts.effort !== undefined) params.effort = opts.effort as ReasoningEffort; + + const response = (await connection.request( + "turn/start", + params, + this.turnTimeoutMs + )) as TurnStartResponse; + const turnId = response.turn.id; + this.logger.debug("turn started", { threadId: opts.threadId, turnId }); + return { turnId }; + } + + /** @internal */ + async interruptTurn(threadId: string): Promise { + const connection = await this.getConnection(); + await connection.request("turn/interrupt", { threadId }, this.requestTimeoutMs); + } + + /** @internal */ + async archiveThread(threadId: string): Promise { + const connection = await this.getConnection(); + await connection.request("thread/archive", { threadId }, this.requestTimeoutMs); + } + + // ---- routing registration (called by CodexBackendView) ---- + + /** @internal */ registerThread(threadId: string, view: CodexBackendView): void { + this.threadOwners.set(threadId, view); + } + + /** @internal */ unregisterThread(threadId: string, view: CodexBackendView): void { + if (this.threadOwners.get(threadId) === view) this.threadOwners.delete(threadId); + } + + /** @internal */ detachView(view: CodexBackendView): void { + for (const threadId of view.ownedThreadIds()) { + if (this.threadOwners.get(threadId) === view) this.threadOwners.delete(threadId); + } + this.views.delete(view); + } + + // ---- connection / routing internals ---- + + private async resolveCommand(): Promise { + if (this.resolvedCommand !== null) return this.resolvedCommand; + const resolved = await resolveCodexCommand({ + command: this.options.command ?? "codex", + env: this.options.env ?? process.env + }); + this.resolvedCommand = resolved.command; + return resolved.command; + } + + private async initialize(): Promise { + if (this.initializeResponse) return this.initializeResponse; + const connection = await this.getConnection(); + const name = this.options.clientName ?? DEFAULT_CLIENT_NAME; + const params: InitializeParams = { + clientInfo: { + name, + title: this.options.clientTitle ?? name, + version: this.options.clientVersion ?? "0.0.0" + }, + capabilities: { + experimentalApi: true, + // We don't proxy through OpenAI's edge attestation flow, so opting in + // would add per-turn latency for an unused round-trip. + requestAttestation: false + } + }; + const response = (await connection.request( + "initialize", + params, + this.requestTimeoutMs + )) as InitializeResponse; + this.initializeResponse = response; + return response; + } + + private async getConnection(): Promise { + if (this.connection) return this.connection; + + let transport: JsonRpcTransport; + if (this.transportFactory !== null) { + transport = this.transportFactory(this.options.command ?? "codex"); + } else { + const command = await this.resolveCommand(); + transport = new StdioJsonRpcTransport({ + command, + args: ["app-server"], + ...(this.options.env !== undefined ? { env: this.options.env } : {}), + logger: this.logger + }); + } + + const connection = new JsonRpcConnection(transport, this.requestTimeoutMs, undefined, { + logger: this.logger, + logContext: { owner: "codex-process-owner" } + }); + connection.setNotificationHandler((method, params) => { + this.handleNotification(method, params); + }); + connection.setRequestHandler((method, params) => this.handleServerRequest(method, params)); + await connection.connect(); + this.connection = connection; + return connection; + } + + private handleNotification(method: string, params: unknown): void { + // Feed the one-shot machinery first (raw); it only acts on its pending + // worker turn and ignores everything else, so view threads pass through. + this.consumeOneShotNotification(method, params); + + const event = normalizeNotification(method, params); + if (event === null) return; + const threadId = eventThreadId(event); + if (threadId === undefined) { + // Connection-level event (e.g. a threadless error) — broadcast to all views. + for (const view of this.views) view.emit(event); + return; + } + const view = this.threadOwners.get(threadId); + // No owning view → the worker thread, or a stale/raced thread. Drop it: the + // worker thread's stream is consumed above and must never leak to a surface. + if (view) view.emit(event); + } + + private async handleServerRequest(method: string, params: unknown): Promise { + const threadId = requestThreadId(params); + const view = threadId !== undefined ? this.threadOwners.get(threadId) : undefined; + + if (method === CODEX_TOOL_CALL_METHOD) { + const handler = view?.toolCallHandler; + if (!handler) { + this.logger.warn("tool call received with no handler for thread", { threadId }); + return noToolHandlerResponse(); + } + const call: AgentBackendToolCall = { method, params }; + return await handler(call); + } + + if (CODEX_APPROVAL_METHODS.has(method)) { + const handler = view?.approvalHandler; + if (!handler) { + this.logger.warn("approval request received with no handler for thread", { + method, + threadId + }); + return approvalResponseFor("denied"); + } + const decision = await handler(method, params); + return approvalResponseFor(decision); + } + + this.logger.debug("unhandled codex server request", { method }); + return {}; + } + + // ---- one-shot worker-thread machinery ---- + + private async runOneShotInner(request: CodexOneShotRequest): Promise { + const connection = await this.getConnection(); + const initialized = await this.initialize(); + let thread: WorkerThread | null = null; + let turnId: string | null = null; + let rolledBack = false; + let aborted = false; + + const abortHandler = (): void => { + aborted = true; + if (thread && turnId) { + void connection + .request("turn/interrupt", { threadId: thread.threadId, turnId }, this.requestTimeoutMs) + .catch((error: unknown) => { + this.logger.warn("turn interrupt failed", { + error: error instanceof Error ? error.message : String(error) + }); + }); + } + }; + + request.abortSignal?.addEventListener("abort", abortHandler, { once: true }); + + try { + if (request.abortSignal?.aborted) { + throw new DOMException("one-shot turn aborted", "AbortError"); + } + + thread = await this.getWorkerThread( + request.model ?? null, + request.modelProvider ?? null, + request.baseInstructions ?? "" + ); + + const input: UserInput[] = [ + { type: "text", text: request.prompt, text_elements: [] }, + ...imagePathsToLocalImageInputs(request.imagePaths ?? []) + ]; + + const turnResponse = (await connection.request( + "turn/start", + { + threadId: thread.threadId, + model: request.model ?? null, + input, + effort: request.effort ?? "low", + ...(request.outputSchema !== undefined ? { outputSchema: request.outputSchema } : {}) + }, + this.requestTimeoutMs + )) as TurnStartResponse; + turnId = turnResponse.turn.id; + + if (request.abortSignal?.aborted || aborted) { + throw new DOMException("one-shot turn aborted", "AbortError"); + } + + const { rawText, tokenUsage } = await this.waitForOneShotTurn(thread.threadId, turnId); + await this.rollbackWorkerThread(thread.threadId); + rolledBack = true; + return { + rawText, + threadId: thread.threadId, + turnId, + userAgent: initialized.userAgent, + model: thread.model, + modelProvider: thread.modelProvider, + serviceTier: thread.serviceTier, + tokenUsage: tokenUsage === null ? null : normalizeTokenUsage(tokenUsage) + }; + } finally { + request.abortSignal?.removeEventListener("abort", abortHandler); + if (thread && turnId && !rolledBack) { + await this.rollbackWorkerThread(thread.threadId).catch((error: unknown) => { + this.logger.warn("worker thread rollback failed", { + threadId: thread?.threadId, + error: error instanceof Error ? error.message : String(error) + }); + }); + } + } + } + + private async getWorkerThread( + model: string | null, + modelProvider: string | null, + baseInstructions: string + ): Promise { + const modelKey = `${model ?? "__default__"}@${modelProvider ?? "__default__"}::${baseInstructions}`; + if (this.workerThread?.modelKey === modelKey) return this.workerThread; + if (this.workerThread) { + const stale = this.workerThread; + this.workerThread = null; + const connection = await this.getConnection(); + await connection + .request("thread/archive", { threadId: stale.threadId }, this.requestTimeoutMs) + .catch((error: unknown) => { + this.logger.warn("thread archive failed", { + threadId: stale.threadId, + error: error instanceof Error ? error.message : String(error) + }); + }); + } + + const connection = await this.getConnection(); + const workspaceDir = await this.prepareWorkerWorkspace(); + const threadResponse = (await connection.request( + "thread/start", + { + model, + ...(modelProvider !== null ? { modelProvider } : {}), + ephemeral: false, + cwd: workspaceDir, + runtimeWorkspaceRoots: [workspaceDir], + serviceName: this.options.serviceName ?? DEFAULT_SERVICE_NAME, + approvalPolicy: "never", + sandbox: "read-only", + ...(baseInstructions.length > 0 ? { baseInstructions } : {}), + // Persistent worker thread for a prompt-cache experiment: keep the thread + // id stable across requests, then roll back each turn. The dedicated cwd + // keeps the worker out of any host repo/worktree. + ...(this.options.worker?.threadConfig !== undefined + ? { config: this.options.worker.threadConfig } + : {}), + environments: [], + experimentalRawEvents: false, + persistExtendedHistory: false + }, + this.requestTimeoutMs + )) as ThreadStartResponse; + await this.clearWorkerThreadGitInfo(threadResponse.thread.id); + await this.setWorkerThreadName(threadResponse.thread.id); + this.workerThread = { + threadId: threadResponse.thread.id, + modelKey, + baseInstructions, + model: threadResponse.model, + modelProvider: threadResponse.modelProvider, + serviceTier: threadResponse.serviceTier + }; + return this.workerThread; + } + + private async rollbackWorkerThread(threadId: string): Promise { + const connection = await this.getConnection(); + await connection.request("thread/rollback", { threadId, numTurns: 1 }, this.requestTimeoutMs); + } + + private async prepareWorkerWorkspace(): Promise { + const workspaceDir = + this.options.worker?.workspaceDir ?? join(tmpdir(), "agent-kit", "oneshot-worker"); + await mkdir(workspaceDir, { recursive: true }); + return workspaceDir; + } + + private async clearWorkerThreadGitInfo(threadId: string): Promise { + const connection = await this.getConnection(); + await connection + .request( + "thread/metadata/update", + { threadId, gitInfo: { sha: null, branch: null, originUrl: null } }, + this.requestTimeoutMs + ) + .catch((error: unknown) => { + this.logger.warn("thread git metadata clear failed", { + threadId, + error: error instanceof Error ? error.message : String(error) + }); + }); + } + + private async setWorkerThreadName(threadId: string): Promise { + const connection = await this.getConnection(); + await connection + .request( + "thread/name/set", + { threadId, name: this.options.worker?.threadName ?? "agent-kit One-Shot Worker" }, + this.requestTimeoutMs + ) + .catch((error: unknown) => { + this.logger.warn("worker thread name set failed", { + threadId, + error: error instanceof Error ? error.message : String(error) + }); + }); + } + + private waitForOneShotTurn( + threadId: string, + turnId: string + ): Promise<{ rawText: string; tokenUsage: ThreadTokenUsage | null }> { + if (this.pendingTurn) { + throw new Error("codex one-shot already has an active turn"); + } + return new Promise<{ rawText: string; tokenUsage: ThreadTokenUsage | null }>( + (resolve, reject) => { + const timer = setTimeout(() => { + this.pendingTurn = null; + reject(new Error("codex one-shot turn timed out")); + }, this.turnTimeoutMs); + this.pendingTurn = { + threadId, + turnId, + agentMessages: [], + tokenUsage: null, + resolve, + reject, + timer + }; + } + ); + } + + private consumeOneShotNotification(method: string, params: unknown): void { + if (this.pendingTurn === null) return; + if (method === "item/completed") { + this.onOneShotItemCompleted(params as ItemCompletedNotification); + } else if (method === "rawResponseItem/completed") { + this.onOneShotRawResponseItemCompleted(params as ServerNotification["params"]); + } else if (method === "thread/tokenUsage/updated") { + this.onOneShotTokenUsage(params as ThreadTokenUsageUpdatedNotification); + } else if (method === "turn/completed") { + this.onOneShotTurnCompleted(params as TurnCompletedNotification); + } + } + + private onOneShotItemCompleted(params: ItemCompletedNotification): void { + const pending = this.pendingTurn; + if (!pending || params.threadId !== pending.threadId || params.turnId !== pending.turnId) return; + if (params.item.type === "agentMessage") pending.agentMessages.push(params.item.text); + } + + private onOneShotRawResponseItemCompleted(params: ServerNotification["params"]): void { + const pending = this.pendingTurn; + if (!pending || typeof params !== "object" || params === null) return; + const maybe = params as { threadId?: unknown; turnId?: unknown; item?: ResponseItem }; + if (maybe.threadId !== pending.threadId || maybe.turnId !== pending.turnId) return; + const item = maybe.item; + if (item?.type !== "message" || item.role !== "assistant") return; + const text = item.content + .filter((content) => content.type === "output_text") + .map((content) => content.text) + .join(""); + if (text) pending.agentMessages.push(text); + } + + private onOneShotTokenUsage(params: ThreadTokenUsageUpdatedNotification): void { + const pending = this.pendingTurn; + if (!pending || params.threadId !== pending.threadId || params.turnId !== pending.turnId) return; + pending.tokenUsage = params.tokenUsage; + } + + private onOneShotTurnCompleted(params: TurnCompletedNotification): void { + const pending = this.pendingTurn; + if (!pending || params.threadId !== pending.threadId || params.turn.id !== pending.turnId) return; + + clearTimeout(pending.timer); + this.pendingTurn = null; + + if (params.turn.status === "failed") { + pending.reject(new Error(params.turn.error?.message ?? "codex one-shot turn failed")); + return; + } + if (params.turn.status === "interrupted") { + pending.reject(new DOMException("one-shot turn aborted", "AbortError")); + return; + } + const rawText = pending.agentMessages.at(-1)?.trim(); + if (!rawText) { + pending.reject(new Error("codex one-shot turn returned no assistant message")); + return; + } + pending.resolve({ rawText, tokenUsage: pending.tokenUsage }); + } +} diff --git a/packages/agent-client/src/codex-thread-client.ts b/packages/agent-client/src/codex-thread-client.ts index d8131d9..58ff4cb 100644 --- a/packages/agent-client/src/codex-thread-client.ts +++ b/packages/agent-client/src/codex-thread-client.ts @@ -1,66 +1,48 @@ // Long-lived, multi-turn Codex App Server client. // -// Keeps a single codex child process + JSON-RPC connection alive and lets the -// caller open MULTIPLE threads on it, each carrying its own dynamic tools. It -// is a thin transport client: it owns the connection lifecycle and routes -// notifications / server-requests, but bakes in no chat or idle-timing logic. +// Now a THIN SHIM over `CodexProcessOwner` + a single `CodexBackendView`: it +// owns one Codex process and exposes one backend whose threads all route to its +// own handlers — exactly the long-standing behavior (a single global +// `onToolCall` / `onApprovalRequest` slot shared by every thread it opens). // -// Ported from PwrSnap's CodexThreadClient with three seams broken: -// • the logger is injected (agent-core `Logger`); -// • `clientInfo.name` + the default `serviceName` are options (no hardcoded -// "pwrsnap"); -// • every native notification is routed through `normalizeNotification`, so -// subscribers receive agent-core `NormalizedThreadEvent` — never a raw -// protocol shape. +// Reach for `CodexProcessOwner.createBackendView()` directly when MULTIPLE chat +// surfaces must share one Codex process with INDEPENDENT handlers — that's the +// per-surface routing `CodexThreadClient`'s single-view shape can't give you. import { - noopLogger, type AgentBackend, - type AgentBackendApprovalHandler, type AgentBackendStartThreadResult, - type AgentBackendToolCall, - type AgentBackendToolCallHandler, type AgentForkThreadOptions, type AgentStartThreadOptions, type AgentStartTurnOptions, type Logger, - type NormalizedThreadEvent, - type NormalizedApprovalDecision + type NormalizedThreadEvent } from "@pwrdrvr/agent-core"; import { - JsonRpcConnection, - StdioJsonRpcTransport, - type JsonRpcTransport -} from "@pwrdrvr/agent-transport"; -import { resolveCodexCommand } from "@pwrdrvr/codex-discovery"; -import type { - InitializeParams, - InitializeResponse, - Personality, - ReasoningEffort -} from "@pwrdrvr/codex-app-server-protocol"; -import type { - AskForApproval, - DynamicToolCallResponse, - DynamicToolSpec, - SandboxMode, - ThreadForkParams, - ThreadForkResponse, - ThreadResumeParams, - ThreadResumeResponse, - ThreadStartParams, - ThreadStartResponse, - TurnStartParams, - TurnStartResponse, - UserInput -} from "@pwrdrvr/codex-app-server-protocol/v2"; -import { - CODEX_APPROVAL_METHODS, - CODEX_TOOL_CALL_METHOD, - normalizeNotification -} from "./normalize"; - -export type CodexThreadClientTransportFactory = (command: string) => JsonRpcTransport; + CodexBackendView, + CodexProcessOwner, + type CodexProcessOwnerOptions, + type CodexStartThreadOptions, + type CodexStartTurnOptions, + type CodexToolCallHandler, + type CodexApprovalHandler, + type CodexTransportFactory, + type StartThreadResult, + type Unsubscribe +} from "./codex-process-owner"; + +export type CodexThreadClientTransportFactory = CodexTransportFactory; + +// Re-exported from the owner (the source of truth) so existing imports from +// `./codex-thread-client` keep resolving. +export type { + CodexStartThreadOptions, + CodexStartTurnOptions, + CodexToolCallHandler, + CodexApprovalHandler, + StartThreadResult, + Unsubscribe +} from "./codex-process-owner"; export type CodexThreadClientOptions = { /** Configured command to resolve. `"codex"` (or empty) triggers discovery. */ @@ -77,447 +59,87 @@ export type CodexThreadClientOptions = { turnTimeoutMs?: number; /** Process env passed to the spawned codex. Defaults to `process.env`. */ env?: NodeJS.ProcessEnv; - /** Override the transport (tests inject an in-memory fake). When supplied, - * no discovery/spawn happens. */ + /** Override the transport (tests inject an in-memory fake). When supplied, no + * discovery/spawn happens. */ transportFactory?: CodexThreadClientTransportFactory; logger?: Logger; }; -/** - * Codex-NATIVE thread/start options. **No longer the public `startThread` - * surface** — `CodexThreadClient` now implements the non-generic `AgentBackend` - * and its public `startThread` takes neutral `AgentStartThreadOptions`. This - * type is retained as the internal mapping target (and is still exported for - * hosts that want to construct the native shape directly via - * `startThreadNative`). The neutral→native mapping lives in `startThread`. - */ -export type CodexStartThreadOptions = { - approvalPolicy?: string; - sandbox?: string; - baseInstructions?: string; - dynamicTools?: DynamicToolSpec[]; - cwd?: string; - runtimeWorkspaceRoots?: string[]; - serviceName?: string; - personality?: string; - /** Model id to start the thread on (ThreadStartParams.model). Omit for the - * Codex default. The host drives this from its settings (per-surface default). */ - model?: string; - /** Model provider id (ThreadStartParams.modelProvider) — the "provider" a host - * picks when more than one is configured. Omit for the Codex default. */ - modelProvider?: string; - /** Service tier (ThreadStartParams.serviceTier), when the host pins one. */ - serviceTier?: string; - /** Per-thread Codex config overlay (the `-c key=value` mechanism). Used to - * disable Codex prompt/tool scaffolding that belongs to coding-agent threads. */ - config?: Record; - /** Thread environments. **Empty array disables exec-environment access**, - * dropping Codex's built-in shell / unified_exec / apply_patch tools. Dynamic - * tools are added before that gate, so they survive. */ - environments?: unknown[]; -}; - -/** Codex-NATIVE turn/start options. Internal mapping target (see - * `startThreadNative`'s sibling `startTurnNative`); the public `startTurn` - * takes neutral `AgentStartTurnOptions`. */ -export type CodexStartTurnOptions = { - threadId: string; - input: UserInput[]; - effort?: string; -}; - -/** - * Handles a tool-call ServerRequest. Reconciled to the canonical - * `AgentBackendToolCallHandler` shape so the controller drives Codex and ACP - * identically: the handler receives a neutral `AgentBackendToolCall` - * (`{ method, params }`) and returns an `unknown` payload the client forwards - * back to Codex verbatim. For Codex the `params` is a `DynamicToolCallParams` - * (cast at the dispatch site) and the returned payload is a - * `DynamicToolCallResponse`. - */ -export type CodexToolCallHandler = AgentBackendToolCallHandler; - -/** Handles an approval ServerRequest; resolves to a neutral decision. Identical - * to the canonical `AgentBackendApprovalHandler`. */ -export type CodexApprovalHandler = AgentBackendApprovalHandler; - -export type Unsubscribe = () => void; - -export type StartThreadResult = { - threadId: string; - model: string; - modelProvider: string; - serviceTier: string | null; -}; - -const DEFAULT_CLIENT_NAME = "agent-kit"; -const DEFAULT_SERVICE_NAME = "agent-kit"; - -/** Map a neutral approval decision onto the generic Codex approval response. */ -function approvalResponseFor(decision: NormalizedApprovalDecision): unknown { - switch (decision) { - case "approved": - return { decision: "approved" }; - case "abort": - return { decision: "abort" }; - case "denied": - default: - return { decision: "denied" }; - } +/** Translate the (additive) thread-client options into owner options. */ +function toOwnerOptions(options: CodexThreadClientOptions): CodexProcessOwnerOptions { + const owner: CodexProcessOwnerOptions = {}; + if (options.command !== undefined) owner.command = options.command; + if (options.clientName !== undefined) owner.clientName = options.clientName; + if (options.clientTitle !== undefined) owner.clientTitle = options.clientTitle; + if (options.clientVersion !== undefined) owner.clientVersion = options.clientVersion; + if (options.serviceName !== undefined) owner.serviceName = options.serviceName; + if (options.requestTimeoutMs !== undefined) owner.requestTimeoutMs = options.requestTimeoutMs; + if (options.turnTimeoutMs !== undefined) owner.turnTimeoutMs = options.turnTimeoutMs; + if (options.env !== undefined) owner.env = options.env; + if (options.transportFactory !== undefined) owner.transportFactory = options.transportFactory; + if (options.logger !== undefined) owner.logger = options.logger; + return owner; } export class CodexThreadClient implements AgentBackend { - private readonly requestTimeoutMs: number; - private readonly turnTimeoutMs: number; - private readonly logger: Logger; - private readonly transportFactory: CodexThreadClientTransportFactory | null; - private resolvedCommand: string | null = null; - private connection: JsonRpcConnection | null = null; - private initializeResponse: InitializeResponse | null = null; + private readonly owner: CodexProcessOwner; + private readonly view: CodexBackendView; - private readonly eventListeners = new Set<(event: NormalizedThreadEvent) => void>(); - private toolCallHandler: CodexToolCallHandler | null = null; - private approvalHandler: CodexApprovalHandler | null = null; - private readonly loadedThreadIds = new Set(); - - constructor(private readonly options: CodexThreadClientOptions = {}) { - this.requestTimeoutMs = options.requestTimeoutMs ?? 20_000; - this.turnTimeoutMs = options.turnTimeoutMs ?? 120_000; - this.logger = options.logger ?? noopLogger; - this.transportFactory = options.transportFactory ?? null; + constructor(options: CodexThreadClientOptions = {}) { + this.owner = new CodexProcessOwner(toOwnerOptions(options)); + this.view = this.owner.createBackendView(); } - /** Subscribe to the normalized event stream. Every native notification is - * routed through `normalizeNotification` before listeners see it. */ onEvent(cb: (event: NormalizedThreadEvent) => void): Unsubscribe { - this.eventListeners.add(cb); - return () => { - this.eventListeners.delete(cb); - }; + return this.view.onEvent(cb); } - /** Register the dynamic-tool ServerRequest handler (one at a time). */ onToolCall(handler: CodexToolCallHandler): Unsubscribe { - this.toolCallHandler = handler; - return () => { - if (this.toolCallHandler === handler) this.toolCallHandler = null; - }; + return this.view.onToolCall(handler); } - /** Register the approval ServerRequest handler (one at a time). */ onApprovalRequest(handler: CodexApprovalHandler): Unsubscribe { - this.approvalHandler = handler; - return () => { - if (this.approvalHandler === handler) this.approvalHandler = null; - }; + return this.view.onApprovalRequest(handler); } - /** - * Public `AgentBackend.startThread`: accepts NEUTRAL `AgentStartThreadOptions` - * and maps them onto Codex-native `CodexStartThreadOptions` before delegating - * to `startThreadNative`. Mapping: - * instructions→baseInstructions, cwd, workspaceRoots→runtimeWorkspaceRoots, - * model/modelProvider, serviceTier (drop `null`), approvalPolicy, sandbox, - * serviceName, config, environments, tools (cast to DynamicToolSpec[])→ - * dynamicTools. - */ async startThread(opts: AgentStartThreadOptions = {}): Promise { - const native: CodexStartThreadOptions = {}; - if (opts.instructions !== undefined) native.baseInstructions = opts.instructions; - if (opts.cwd !== undefined) native.cwd = opts.cwd; - if (opts.workspaceRoots !== undefined) { - native.runtimeWorkspaceRoots = [...opts.workspaceRoots]; - } - if (opts.model !== undefined) native.model = opts.model; - if (opts.modelProvider !== undefined) native.modelProvider = opts.modelProvider; - // Codex's serviceTier is a plain string; a neutral `null` means "don't pin". - if (opts.serviceTier != null) native.serviceTier = opts.serviceTier; - if (opts.approvalPolicy !== undefined) native.approvalPolicy = opts.approvalPolicy; - if (opts.sandbox !== undefined) native.sandbox = opts.sandbox; - if (opts.serviceName !== undefined) native.serviceName = opts.serviceName; - if (opts.config !== undefined) native.config = opts.config; - if (opts.environments !== undefined) native.environments = opts.environments; - if (opts.tools !== undefined) native.dynamicTools = opts.tools as DynamicToolSpec[]; - return this.startThreadNative(native); + return this.view.startThread(opts); } - /** Fork an existing thread into a fresh one carrying its history (Codex - * `thread/fork`). The new thread inherits the source's model/provider. */ - async forkThread(opts: AgentForkThreadOptions): Promise { - const connection = await this.getConnection(); - await this.initialize(); - - const params: ThreadForkParams = { - threadId: opts.sourceThreadId, - persistExtendedHistory: false - }; - if (opts.cwd !== undefined) params.cwd = opts.cwd; - if (opts.workspaceRoots !== undefined) params.runtimeWorkspaceRoots = [...opts.workspaceRoots]; - if (opts.instructions !== undefined) params.baseInstructions = opts.instructions; - if (opts.approvalPolicy !== undefined) params.approvalPolicy = opts.approvalPolicy as AskForApproval; - if (opts.sandbox !== undefined) params.sandbox = opts.sandbox as SandboxMode; - if (opts.config !== undefined) { - params.config = opts.config as NonNullable; - } - - const response = (await connection.request( - "thread/fork", - params, - this.requestTimeoutMs - )) as ThreadForkResponse; - const threadId = response.thread.id; - this.loadedThreadIds.add(threadId); - return { - threadId, - model: response.model, - modelProvider: response.modelProvider, - serviceTier: response.serviceTier - }; - } - - /** Codex-native thread/start. Builds `ThreadStartParams` directly. Exposed for - * hosts that want full Codex control; the neutral `startThread` delegates here. */ async startThreadNative(opts: CodexStartThreadOptions = {}): Promise { - const connection = await this.getConnection(); - await this.initialize(); - - // exactOptionalPropertyTypes: only attach a key when the caller supplied it. - const params: ThreadStartParams = { - experimentalRawEvents: false, - persistExtendedHistory: false - }; - if (opts.cwd !== undefined) params.cwd = opts.cwd; - if (opts.model !== undefined) params.model = opts.model; - if (opts.modelProvider !== undefined) params.modelProvider = opts.modelProvider; - if (opts.serviceTier !== undefined) params.serviceTier = opts.serviceTier; - if (opts.runtimeWorkspaceRoots !== undefined) { - params.runtimeWorkspaceRoots = opts.runtimeWorkspaceRoots; - } - const serviceName = opts.serviceName ?? this.options.serviceName ?? DEFAULT_SERVICE_NAME; - params.serviceName = serviceName; - if (opts.approvalPolicy !== undefined) { - params.approvalPolicy = opts.approvalPolicy as AskForApproval; - } - if (opts.sandbox !== undefined) params.sandbox = opts.sandbox as SandboxMode; - if (opts.baseInstructions !== undefined) params.baseInstructions = opts.baseInstructions; - if (opts.personality !== undefined) params.personality = opts.personality as Personality; - if (opts.dynamicTools !== undefined) params.dynamicTools = opts.dynamicTools; - if (opts.config !== undefined) { - params.config = opts.config as NonNullable; - } - if (opts.environments !== undefined) { - params.environments = opts.environments as NonNullable; - } + return this.view.startThreadNative(opts); + } - const response = (await connection.request( - "thread/start", - params, - this.requestTimeoutMs - )) as ThreadStartResponse; - const threadId = response.thread.id; - this.loadedThreadIds.add(threadId); - this.logger.debug("thread started", { threadId }); - return { - threadId, - model: response.model, - modelProvider: response.modelProvider, - serviceTier: response.serviceTier - }; + async forkThread(opts: AgentForkThreadOptions): Promise { + return this.view.forkThread(opts); } async resumeThread(threadId: string): Promise { - if (this.loadedThreadIds.has(threadId)) return; - const connection = await this.getConnection(); - await this.initialize(); - - const params: ThreadResumeParams = { - threadId, - persistExtendedHistory: false - }; - const response = (await connection.request( - "thread/resume", - params, - this.requestTimeoutMs - )) as ThreadResumeResponse; - this.loadedThreadIds.add(response.thread.id); - this.logger.debug("thread resumed", { threadId: response.thread.id }); + return this.view.resumeThread(threadId); } async clearThreadGitInfo(threadId: string): Promise { - const connection = await this.getConnection(); - await this.initialize(); - await connection.request( - "thread/metadata/update", - { threadId, gitInfo: { sha: null, branch: null, originUrl: null } }, - this.requestTimeoutMs - ); + return this.view.clearThreadGitInfo(threadId); } - /** - * Public `AgentBackend.startTurn`: accepts NEUTRAL `AgentStartTurnOptions` and - * maps them onto Codex-native `UserInput[]`. The neutral `input.text` becomes a - * leading `{ type: "text" }` item; each `input.imagePaths` entry becomes a - * `{ type: "localImage", path }` item appended after the text. `reasoning` - * maps to Codex's `effort`. - */ async startTurn(opts: AgentStartTurnOptions): Promise<{ turnId: string }> { - const input: UserInput[] = [{ type: "text", text: opts.input.text, text_elements: [] }]; - for (const path of opts.input.imagePaths ?? []) { - input.push({ type: "localImage", path }); - } - const native: CodexStartTurnOptions = { threadId: opts.threadId, input }; - if (opts.reasoning !== undefined) native.effort = opts.reasoning; - return this.startTurnNative(native); + return this.view.startTurn(opts); } - /** Codex-native turn/start. Takes pre-built `UserInput[]`. The neutral - * `startTurn` delegates here after mapping text + image paths. */ async startTurnNative(opts: CodexStartTurnOptions): Promise<{ turnId: string }> { - await this.resumeThread(opts.threadId); - const connection = await this.getConnection(); - await this.initialize(); - - const params: TurnStartParams = { - threadId: opts.threadId, - input: opts.input - }; - if (opts.effort !== undefined) params.effort = opts.effort as ReasoningEffort; - - const response = (await connection.request( - "turn/start", - params, - this.turnTimeoutMs - )) as TurnStartResponse; - const turnId = response.turn.id; - this.logger.debug("turn started", { threadId: opts.threadId, turnId }); - return { turnId }; + return this.view.startTurnNative(opts); } async interruptTurn(threadId: string): Promise { - const connection = await this.getConnection(); - await connection.request("turn/interrupt", { threadId }, this.requestTimeoutMs); + return this.view.interruptTurn(threadId); } async archiveThread(threadId: string): Promise { - const connection = await this.getConnection(); - await connection.request("thread/archive", { threadId }, this.requestTimeoutMs); + return this.view.archiveThread(threadId); } + /** Tear down the owned Codex process (a `CodexThreadClient` owns its own). */ async close(): Promise { - const connection = this.connection; - this.connection = null; - this.initializeResponse = null; - this.loadedThreadIds.clear(); - if (connection) await connection.close(); - } - - // ---- internals ---- - - private emit(event: NormalizedThreadEvent): void { - for (const listener of this.eventListeners) listener(event); - } - - private async resolveCommand(): Promise { - if (this.resolvedCommand !== null) return this.resolvedCommand; - const resolved = await resolveCodexCommand({ - command: this.options.command ?? "codex", - env: this.options.env ?? process.env - }); - this.resolvedCommand = resolved.command; - return resolved.command; - } - - private async initialize(): Promise { - if (this.initializeResponse) return this.initializeResponse; - const connection = await this.getConnection(); - const name = this.options.clientName ?? DEFAULT_CLIENT_NAME; - const params: InitializeParams = { - clientInfo: { - name, - title: this.options.clientTitle ?? name, - version: this.options.clientVersion ?? "0.0.0" - }, - capabilities: { - experimentalApi: true, - // We don't proxy through OpenAI's edge attestation flow, so opting in - // would add per-turn latency for an unused round-trip. - requestAttestation: false - } - }; - const response = (await connection.request( - "initialize", - params, - this.requestTimeoutMs - )) as InitializeResponse; - this.initializeResponse = response; - return response; - } - - private async getConnection(): Promise { - if (this.connection) return this.connection; - - let transport: JsonRpcTransport; - if (this.transportFactory !== null) { - // Tests / hosts supplying a fake transport bypass discovery + spawn. - transport = this.transportFactory(this.options.command ?? "codex"); - } else { - const command = await this.resolveCommand(); - transport = new StdioJsonRpcTransport({ - command, - args: ["app-server"], - ...(this.options.env !== undefined ? { env: this.options.env } : {}), - logger: this.logger - }); - } - - const connection = new JsonRpcConnection(transport, this.requestTimeoutMs, undefined, { - logger: this.logger, - logContext: { owner: "codex-thread-client" } - }); - connection.setNotificationHandler((method, params) => { - this.handleNotification(method, params); - }); - connection.setRequestHandler((method, params) => this.handleServerRequest(method, params)); - await connection.connect(); - this.connection = connection; - return connection; - } - - private handleNotification(method: string, params: unknown): void { - const event = normalizeNotification(method, params); - if (event !== null) this.emit(event); - } - - private async handleServerRequest(method: string, params: unknown): Promise { - if (method === CODEX_TOOL_CALL_METHOD) { - const handler = this.toolCallHandler; - if (!handler) { - this.logger.warn("tool call received with no handler registered"); - return { - contentItems: [ - { type: "inputText", text: "No tool handler is registered for this thread." } - ], - success: false - } satisfies DynamicToolCallResponse; - } - // Canonical `AgentBackendToolCall` shape: the host's handler reads - // `call.params` (a `DynamicToolCallParams` for Codex) and returns the - // `DynamicToolCallResponse` we forward back to Codex verbatim. - const call: AgentBackendToolCall = { method, params }; - return await handler(call); - } - - if (CODEX_APPROVAL_METHODS.has(method)) { - const handler = this.approvalHandler; - if (!handler) { - this.logger.warn("approval request received with no handler registered", { method }); - return approvalResponseFor("denied"); - } - const decision = await handler(method, params); - return approvalResponseFor(decision); - } - - this.logger.debug("unhandled codex server request", { method }); - return {}; + await this.owner.close(); } } diff --git a/packages/agent-client/src/index.ts b/packages/agent-client/src/index.ts index 7cd4641..d61fb9a 100644 --- a/packages/agent-client/src/index.ts +++ b/packages/agent-client/src/index.ts @@ -1,16 +1,38 @@ // @pwrdrvr/agent-client — Codex App Server adapter for agent-kit. // -// Three surfaces, all normalizing into the agent-core neutral schema: -// • CodexThreadClient — long-lived, multi-turn thread client (normalized -// event stream + injected tool/approval handlers); -// • CodexOneShotClient — one-shot structured-output enrichment turns; -// • ChatThreadController — surface-agnostic chat controller over the thread -// client, with persistence + prompt + catalog seams -// injected by the host. +// Surfaces, all normalizing into the agent-core neutral schema: +// • CodexProcessOwner — owns ONE codex app-server process + connection and +// hands out per-surface backend VIEWS (each an +// AgentBackend whose threads route to its own +// handlers), plus model listing + structured one-shot +// over the shared connection. Pool it for the app. +// • CodexBackendView — a lightweight per-surface AgentBackend over an owner; +// sibling surfaces sharing one process never clobber. +// • CodexProcessOwnerPool — lifecycle pool of owners keyed on connection +// identity (command, CODEX_HOME/env). One process per key. +// • CodexThreadClient — thin single-view shim over an owner (one process, one +// global handler slot) — the historical surface. +// • CodexOneShotClient — thin shim over owner.runOneShot()/listModels(). +// • ChatThreadController — surface-agnostic chat controller over any +// AgentBackend (a view or a thread client). // // Plus the normalization layer (Codex v2 notifications → NormalizedThreadEvent) // and the chat tool-definition primitives. +export { + CodexProcessOwner, + CodexBackendView, + type CodexProcessOwnerOptions, + type CodexOneShotWorkerOptions, + type CodexTransportFactory +} from "./codex-process-owner"; + +export { + CodexProcessOwnerPool, + type CodexProcessOwnerFactory, + type CodexProcessOwnerPoolOptions +} from "./codex-process-owner-pool"; + export { CodexThreadClient, type CodexThreadClientOptions, diff --git a/packages/agent-client/test/codex-process-owner.test.ts b/packages/agent-client/test/codex-process-owner.test.ts new file mode 100644 index 0000000..894afb9 --- /dev/null +++ b/packages/agent-client/test/codex-process-owner.test.ts @@ -0,0 +1,382 @@ +import { describe, it, expect } from "vitest"; +import type { JsonRpcTransport } from "@pwrdrvr/agent-transport"; +import type { + NormalizedApprovalDecision, + NormalizedThreadEvent +} from "@pwrdrvr/agent-core"; +import { CodexProcessOwner } from "../src/codex-process-owner"; +import { CodexProcessOwnerPool } from "../src/codex-process-owner-pool"; +import { CODEX_NOTIFICATION_METHODS } from "../src/normalize"; + +/** + * In-memory JsonRpcTransport that auto-answers the requests an owner makes and + * lets a test (a) push inbound notifications and (b) inject server-requests + * (tool-call / approval) from "Codex" and read back the client's response. + */ +class FakeTransport implements JsonRpcTransport { + sent: Array<{ method?: string; id?: unknown; params?: unknown; result?: unknown }> = []; + private messageHandler: (message: string) => void = () => undefined; + private closeHandler: (error?: Error) => void = () => undefined; + private threadCounter = 0; + private turnCounter = 0; + private serverReqId = 0; + private readonly pendingServerReqs = new Map void>(); + lastThreadId = ""; + lastTurnId = ""; + closed = false; + + async connect(): Promise {} + async close(): Promise { + this.closed = true; + this.closeHandler(); + } + setMessageHandler(handler: (message: string) => void): void { + this.messageHandler = handler; + } + setCloseHandler(handler: (error?: Error) => void): void { + this.closeHandler = handler; + } + + send(message: string): void { + const env = JSON.parse(message) as { + method?: string; + id?: unknown; + params?: unknown; + result?: unknown; + }; + this.sent.push(env); + // A client→server REQUEST (has both id and method): auto-respond. + if (env.id !== undefined && env.method !== undefined) { + const result = this.respond(env.method, env.params); + queueMicrotask(() => { + this.messageHandler(JSON.stringify({ jsonrpc: "2.0", id: env.id, result })); + }); + return; + } + // The client's RESPONSE to a server-initiated request (id, no method). + if (env.id !== undefined && env.method === undefined) { + const resolve = this.pendingServerReqs.get(env.id as number); + if (resolve) { + this.pendingServerReqs.delete(env.id as number); + resolve(env.result); + } + } + } + + private respond(method: string, _params: unknown): unknown { + switch (method) { + case "initialize": + return { userAgent: "fake/1.0", capabilities: {} }; + case "thread/start": { + this.lastThreadId = `thread-${++this.threadCounter}`; + return { + thread: { id: this.lastThreadId }, + model: "gpt-5-codex", + modelProvider: "openai", + serviceTier: "default" + }; + } + case "thread/resume": + return { thread: { id: "thread-resumed" } }; + case "turn/start": { + this.lastTurnId = `turn-${++this.turnCounter}`; + return { turn: { id: this.lastTurnId } }; + } + case "model/list": + return { + data: [ + { + id: "gpt-5-codex", + model: "gpt-5-codex", + displayName: "GPT-5 Codex", + description: "", + hidden: false, + inputModalities: ["text"], + defaultServiceTier: null, + isDefault: true + } + ], + nextCursor: null + }; + default: + return {}; + } + } + + notify(method: string, params: unknown): void { + this.messageHandler(JSON.stringify({ jsonrpc: "2.0", method, params })); + } + + /** Inject a server-initiated request and resolve with the client's response. */ + serverRequest(method: string, params: unknown): Promise { + const id = `srv-${++this.serverReqId}`; + return new Promise((resolve) => { + this.pendingServerReqs.set(id as unknown as number, resolve); + this.messageHandler(JSON.stringify({ jsonrpc: "2.0", id, method, params })); + }); + } + + sentMethods(): string[] { + return this.sent.filter((e) => e.method !== undefined).map((e) => e.method as string); + } + + async waitForSent(method: string): Promise { + for (let i = 0; i < 50; i++) { + if (this.sentMethods().includes(method)) return; + await tick(); + } + throw new Error(`timed out waiting for ${method} to be sent`); + } +} + +const tick = (): Promise => new Promise((r) => setTimeout(r, 0)); + +describe("CodexProcessOwner — per-view routing", () => { + it("routes a thread's events only to the view that owns it", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const viewA = owner.createBackendView(); + const viewB = owner.createBackendView(); + const a: NormalizedThreadEvent[] = []; + const b: NormalizedThreadEvent[] = []; + viewA.onEvent((e) => a.push(e)); + viewB.onEvent((e) => b.push(e)); + + const ta = await viewA.startThread(); + const tb = await viewB.startThread(); + expect(ta.threadId).not.toBe(tb.threadId); + + fake.notify(CODEX_NOTIFICATION_METHODS.agentMessageDelta, { + threadId: ta.threadId, + turnId: "t1", + itemId: "m", + delta: "AAA" + }); + fake.notify(CODEX_NOTIFICATION_METHODS.agentMessageDelta, { + threadId: tb.threadId, + turnId: "t2", + itemId: "m", + delta: "BBB" + }); + await tick(); + + expect(a).toEqual([ + { kind: "agent_message_delta", threadId: ta.threadId, turnId: "t1", itemId: "m", delta: "AAA" } + ]); + expect(b).toEqual([ + { kind: "agent_message_delta", threadId: tb.threadId, turnId: "t2", itemId: "m", delta: "BBB" } + ]); + await owner.close(); + }); + + it("routes a tool call to the owning view's handler, not a sibling's", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const viewA = owner.createBackendView(); + const viewB = owner.createBackendView(); + const calledOn: string[] = []; + viewA.onToolCall(async () => { + calledOn.push("A"); + return { contentItems: [{ type: "inputText", text: "from A" }], success: true }; + }); + viewB.onToolCall(async () => { + calledOn.push("B"); + return { contentItems: [{ type: "inputText", text: "from B" }], success: true }; + }); + + const ta = await viewA.startThread(); + await viewB.startThread(); + + const response = (await fake.serverRequest("item/tool/call", { + threadId: ta.threadId, + turnId: "t1", + callId: "c1", + namespace: null, + tool: "echo", + arguments: {} + })) as { success: boolean }; + + expect(calledOn).toEqual(["A"]); + expect(response.success).toBe(true); + await owner.close(); + }); + + it("routes an approval to the owning view and maps the neutral decision", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const view = owner.createBackendView(); + let seenMethod = ""; + view.onApprovalRequest(async (method): Promise => { + seenMethod = method; + return "approved"; + }); + const t = await view.startThread(); + + const decision = (await fake.serverRequest("item/commandExecution/requestApproval", { + threadId: t.threadId, + turnId: "t1", + itemId: "i1" + })) as { decision: string }; + + expect(seenMethod).toBe("item/commandExecution/requestApproval"); + expect(decision.decision).toBe("approved"); + await owner.close(); + }); + + it("denies a tool call / approval for a thread with no registered handler", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const view = owner.createBackendView(); + const t = await view.startThread(); // no onToolCall/onApprovalRequest registered + + const tool = (await fake.serverRequest("item/tool/call", { + threadId: t.threadId, + turnId: "x", + callId: "c", + namespace: null, + tool: "echo", + arguments: {} + })) as { success: boolean }; + expect(tool.success).toBe(false); + + const appr = (await fake.serverRequest("item/fileChange/requestApproval", { + threadId: t.threadId, + turnId: "x", + itemId: "i" + })) as { decision: string }; + expect(appr.decision).toBe("denied"); + await owner.close(); + }); + + it("detaching a view stops routing its threads without killing the process", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const viewA = owner.createBackendView(); + const viewB = owner.createBackendView(); + const a: NormalizedThreadEvent[] = []; + viewA.onEvent((e) => a.push(e)); + const ta = await viewA.startThread(); + const tb = await viewB.startThread(); + + await viewA.close(); // detach A — process stays up for B + + fake.notify(CODEX_NOTIFICATION_METHODS.agentMessageDelta, { + threadId: ta.threadId, + turnId: "t", + itemId: "m", + delta: "late" + }); + await tick(); + expect(a).toEqual([]); // A no longer receives its thread's events + expect(fake.closed).toBe(false); // process not torn down + + // B still works on the same process. + const turn = await viewB.startTurn({ threadId: tb.threadId, input: { text: "hi" } }); + expect(turn.turnId).toBeTruthy(); + await owner.close(); + expect(fake.closed).toBe(true); + }); +}); + +describe("CodexProcessOwner — model listing + one-shot over the shared connection", () => { + it("lists models without opening a thread", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const models = await owner.listModels(); + expect(models.map((m) => m.id)).toEqual(["gpt-5-codex"]); + expect(fake.sentMethods()).toContain("model/list"); + expect(fake.sentMethods()).not.toContain("thread/start"); + await owner.close(); + }); + + it("runs a structured one-shot turn (outputSchema + rollback) without leaking to a view", async () => { + const fake = new FakeTransport(); + const owner = new CodexProcessOwner({ transportFactory: () => fake }); + const view = owner.createBackendView(); + const viewEvents: NormalizedThreadEvent[] = []; + view.onEvent((e) => viewEvents.push(e)); + + const schema = { type: "object", properties: { ok: { type: "boolean" } } }; + const pending = owner.runOneShot({ prompt: "enrich this", outputSchema: schema }); + + // Let the worker thread + turn get established, then drive completion. + await fake.waitForSent("turn/start"); + const workerThreadId = fake.lastThreadId; + const turnId = fake.lastTurnId; + fake.notify("item/completed", { + threadId: workerThreadId, + turnId, + item: { type: "agentMessage", text: '{"ok":true}' } + }); + fake.notify("thread/tokenUsage/updated", { + threadId: workerThreadId, + turnId, + tokenUsage: { + last: { + inputTokens: 10, + cachedInputTokens: 0, + outputTokens: 5, + reasoningOutputTokens: 0, + totalTokens: 15 + }, + modelContextWindow: 200000 + } + }); + fake.notify("turn/completed", { + threadId: workerThreadId, + turn: { id: turnId, items: [], itemsView: "full", status: "completed", error: null } + }); + + const res = await pending; + expect(res.rawText).toBe('{"ok":true}'); + expect(res.tokenUsage?.totalTokens).toBe(15); + + // outputSchema rode the turn, and the turn was rolled back afterward. + const turnStart = fake.sent.find((e) => e.method === "turn/start")?.params as Record< + string, + unknown + >; + expect(turnStart.outputSchema).toEqual(schema); + expect(fake.sentMethods()).toContain("thread/rollback"); + + // The worker thread's stream never reached the surface view. + expect(viewEvents).toEqual([]); + await owner.close(); + expect(fake.sentMethods()).toContain("thread/archive"); // worker archived on close + }); +}); + +describe("CodexProcessOwnerPool", () => { + it("hands the same warmed owner to concurrent acquires for a key", async () => { + const pool = new CodexProcessOwnerPool(); + let built = 0; + const factory = (): CodexProcessOwner => { + built++; + return new CodexProcessOwner({ transportFactory: () => new FakeTransport() }); + }; + const [a, b] = await Promise.all([ + pool.acquire("codex::default", factory), + pool.acquire("codex::default", factory) + ]); + expect(a).toBe(b); + expect(built).toBe(1); + expect(pool.has("codex::default")).toBe(true); + await pool.closeAll(); + expect(pool.has("codex::default")).toBe(false); + }); + + it("keys distinct identities to distinct owners and releases one", async () => { + const pool = new CodexProcessOwnerPool(); + const factory = (): CodexProcessOwner => + new CodexProcessOwner({ transportFactory: () => new FakeTransport() }); + const one = await pool.acquire("codex::home-a", factory); + const two = await pool.acquire("codex::home-b", factory); + expect(one).not.toBe(two); + + await pool.release("codex::home-a"); + expect(pool.has("codex::home-a")).toBe(false); + expect(pool.has("codex::home-b")).toBe(true); + await pool.release("codex::missing"); // safe no-op + await pool.closeAll(); + }); +});