import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import crypto from "node:crypto"; import { spawn } from "node:child_process"; import { controlCancelAsk, controlCompact, controlConfigure, controlInterrupt, controlRespondAsk, controlSend, controlStatus, controlSubscribe } from "./control-client.mjs"; import { primeControlCompact, primeControlConfigure, primeControlInterrupt, primeControlSend, primeControlStatus, primeControlSubscribe } from "./prime-control.mjs"; import { ClientConnection } from "./client-connection.mjs"; import { promptMatches, StateStore, turnPromptText } from "./store.mjs"; import { inspectOmpSessionActivity } from "./omp-sessions.mjs"; import { createBackendFactory, FEATURED_MODEL_IDS, syncCodexModelsCache } from "./backend/omp.mjs"; import { splitTaggedModel, tagForPersistentState } from "./backend/both.mjs"; import { loadBridgeConfig } from "./config.mjs"; import { readTomlFile } from "./toml.mjs"; import { createIsolatedWorkspace, cleanupWorkspace, getVcsInfo, adaptWorktreeSpawnCommand } from "./workspace.mjs"; import { ERROR, RpcError, errorResponse, initializeResult, makeAgentMessageItem, makeCommandItem, makeContextCompactionItem, makeFileChangeItem, makeDynamicToolCallItem, makeImageViewItem, makeMcpToolCallItem, makeReasoningItem, makeTurn, makeUserMessageItem, makeUserInputResponseItem, makeTokenUsage, makeWebSearchItem, resolveModelContextWindow, successResponse, } from "./protocol.mjs"; import { abortError, atomicWriteJson, canonicalizeForPolicy, debug, decodeCursor, extractInputPromptAndImages, extractInputText, id, encodeCursor, isInside, log, protocolShape, protocolTrace, normalizeApprovalDecision, paginateArray, readJson, safePreview, extractToolOutputText, askAnswersFromResult, sleep, serializeError, unixSeconds, } from "./util.mjs"; const EMPTY_DISCOVERY_METHODS = new Set([ "skills/list", "skill/list", "plugin/list", "plugins/list", "app/list", "apps/list", "mcpServerStatus/list", "mcpServer/list", "connector/list", "collaborationMode/list", "experimentalFeature/list", ]); const EXTERNAL_STREAM_TURN_WINDOW = 3; const EXTERNAL_DURABLE_AUDIT_INTERVAL_MS = 30_000; /** How long turn/start waits for an externally steered turn to materialize. */ const EXTERNAL_STEER_TURN_TIMEOUT_MS = Number(process.env.OMP_CODEX_STEER_TURN_TIMEOUT_MS || 8_000); /** * Order-independent structural fingerprint. * * The same item reaches us through two projections - the locally streamed one * and the one reparsed from the OMP session file - which agree on content but * not on key insertion order. A plain JSON.stringify therefore flip-flops * between two hashes for a byte-identical item and republishes it on every * poll, so the fingerprint has to canonicalize key order. */ function canonicalize(value) { if (Array.isArray(value)) return value.map(canonicalize); if (!value || typeof value !== "object") return value; const out = {}; for (const key of Object.keys(value).sort()) out[key] = canonicalize(value[key]); return out; } function itemSignature(item) { try { return crypto.createHash("sha1").update(JSON.stringify(canonicalize(item))).digest("base64"); } catch { return `${item?.status ?? ""}:${String(item?.id ?? "")}`; } } const FUZZY_SEARCH_IGNORED_DIRECTORIES = new Set([ ".git", ".jj", ".direnv", "node_modules", "target", "dist", "build", ".next", ".svelte-kit", ]); const FUZZY_SEARCH_MAX_ENTRIES = 50_000; const FUZZY_SEARCH_MAX_RESULTS = 100; function fuzzyMatch(candidate, query) { const haystack = String(candidate || "").toLowerCase(); const needle = String(query || "").trim().toLowerCase(); if (!needle) return { score: 1, indices: [] }; const contiguous = haystack.indexOf(needle); if (contiguous >= 0) { return { score: Math.max(1, 10_000 - contiguous * 10 - haystack.length), indices: Array.from({ length: needle.length }, (_, index) => contiguous + index), }; } const indices = []; let cursor = 0; for (const character of needle) { const found = haystack.indexOf(character, cursor); if (found < 0) return null; indices.push(found); cursor = found + 1; } const gaps = indices.at(-1) - indices[0] + 1 - needle.length; return { score: Math.max(1, 5_000 - gaps * 20 - haystack.length), indices }; } async function searchWorkspaceFiles(query, roots, signal) { const results = []; let seenEntries = 0; for (const requestedRoot of roots || []) { if (signal?.aborted || seenEntries >= FUZZY_SEARCH_MAX_ENTRIES) break; if (typeof requestedRoot !== "string" || !path.isAbsolute(requestedRoot)) continue; const root = await fs.realpath(requestedRoot).catch(() => null); if (!root) continue; const stat = await fs.stat(root).catch(() => null); if (!stat?.isDirectory()) continue; const queue = [{ absolute: root, relative: "" }]; while (queue.length && seenEntries < FUZZY_SEARCH_MAX_ENTRIES && !signal?.aborted) { const current = queue.shift(); const entries = await fs.readdir(current.absolute, { withFileTypes: true }).catch(() => []); entries.sort((left, right) => left.name.localeCompare(right.name)); for (const entry of entries) { if (signal?.aborted || seenEntries >= FUZZY_SEARCH_MAX_ENTRIES) break; seenEntries += 1; const relative = current.relative ? `${current.relative}/${entry.name}` : entry.name; if (entry.isDirectory() && FUZZY_SEARCH_IGNORED_DIRECTORIES.has(entry.name)) continue; const match = fuzzyMatch(relative, query); if (match) { results.push({ root, path: relative, match_type: entry.isDirectory() ? "directory" : "file", file_name: entry.name, score: match.score, indices: match.indices.length ? match.indices : null, }); } if (entry.isDirectory()) { queue.push({ absolute: path.join(current.absolute, entry.name), relative }); } } } } return results .sort((left, right) => right.score - left.score || left.path.localeCompare(right.path)) .slice(0, FUZZY_SEARCH_MAX_RESULTS); } /** * Give an item left mid-flight a terminal shape. * * ChatGPT keeps a spinner on any tool item that still says `inProgress`, and a * command with a null exit code reads as a background task that never came * back, so an abandoned call is reported as failed rather than pending. */ function settleAbandonedItem(item) { if (!item || item.status !== "inProgress") return item; if (item.type === "commandExecution") { item.status = "failed"; if (item.exitCode == null) item.exitCode = 1; } else if (item.type === "dynamicToolCall") { item.status = "failed"; item.success = false; } else if (item.type === "mcpToolCall") { item.status = "failed"; } else { item.status = "completed"; } if (item.durationMs === null) item.durationMs = 0; return item; } /** * compare closing blocks with this message's stream, not the turn-wide item * mismatches are dropped because guessing glues separate thinking blocks */ function pendingBlockTail(streamed, complete) { const current = String(streamed || ""); const value = String(complete || ""); if (!value.trim()) return ""; if (!current) return value; if (value.startsWith(current)) return value.slice(current.length); return ""; } function codexErrorInfoFor(message, code = null) { const value = `${message || ""} ${code || ""}`.toLowerCase(); if (/context.{0,20}(window|length|token)|maximum context/.test(value)) return "contextWindowExceeded"; if (/quota|usage.{0,12}limit|rate.{0,8}limit|\\b429\\b/.test(value)) return "usageLimitExceeded"; if (/unauthor|forbidden|credential|api.?key|invalid_grant|\\b40[13]\\b/.test(value)) return "unauthorized"; if (/overload|capacity|\\b503\\b/.test(value)) return "serverOverloaded"; if (/bad request|invalid request|\\b400\\b/.test(value)) return "badRequest"; if (/connect|network|timeout|timed out|econn/.test(value)) { return { httpConnectionFailed: { httpStatusCode: null } }; } return "other"; } function protocolTurnError(error) { const message = String(error?.message || error || "OMP turn failed"); const code = error?.code ?? null; return { message, codexErrorInfo: error?.codexErrorInfo || codexErrorInfoFor(message, code), additionalDetails: error?.additionalDetails ?? (code == null ? null : `code: ${code}`), }; } function lifecycleModelSelector(model) { if (typeof model === "string") return model; if (!model || typeof model !== "object") return ""; if (model.selector) return String(model.selector); const id = model.id || model.model; const provider = model.provider || model.modelProvider; return provider && id ? `${provider}/${id}` : String(id || ""); } function normalizeUserInputQuestions(request = {}) { const rawQuestions = Array.isArray(request.questions) && request.questions.length ? request.questions.slice(0, 3) : [{ id: "answer", header: request.title || "OMP", question: request.prompt || request.message || "Input requested", options: request.options || null, }]; return rawQuestions.map((question, index) => ({ id: String(question.id || `answer-${index + 1}`), header: String(question.header || request.title || "OMP"), question: String(question.question || question.prompt || question.message || "Input requested"), isOther: question.isOther !== false, isSecret: Boolean(question.isSecret), options: Array.isArray(question.options) && question.options.length ? question.options.map((option, optionIndex) => { const rawLabel = String(typeof option === "string" ? option : option?.label || option?.value || ""); const recommended = Number(question.recommended) === optionIndex && !/\(Recommended\)$/i.test(rawLabel); return { label: recommended ? `${rawLabel} (Recommended)` : rawLabel, description: String(typeof option === "string" ? "" : option?.description || ""), }; }) : null, })); } function normalizeUserInputAnswers(answers) { if (!answers || typeof answers !== "object") return {}; return Object.fromEntries(Object.entries(answers).map(([questionId, answer]) => [ questionId, { ...answer, answers: Array.isArray(answer?.answers) ? answer.answers.map((value) => String(value).replace(/\s+\(Recommended\)$/i, "")) : [], }, ])); } function clientNeedsCanonicalModelIds(peer) { return String(peer?.clientInfo?.name || "").toLowerCase().includes("ios"); } const FULL_PROTOCOL_TRACE_METHODS = new Set([ "model/list", "collaborationMode/list", "configRequirements/read", "permissionProfile/list", "experimentalFeature/list", ]); function protocolResultDetails(method, result) { const details = { resultShape: protocolShape(result) }; if (FULL_PROTOCOL_TRACE_METHODS.has(method)) details.result = result; if (method === "config/read") { const config = result?.config || {}; details.selection = { model: config.model ?? null, modelProvider: config.model_provider ?? config.modelProvider ?? null, reasoningEffort: config.model_reasoning_effort ?? config.reasoningEffort ?? null, approvalPolicy: config.approval_policy ?? config.approvalPolicy ?? null, serviceTier: config.service_tier ?? config.serviceTier ?? null, }; } else if (method === "thread/resume" || method === "thread/start") { const thread = result?.thread || result; details.selection = { model: thread?.model ?? null, modelProvider: thread?.modelProvider ?? null, reasoningEffort: thread?.reasoningEffort ?? null, serviceTier: thread?.serviceTier ?? null, status: thread?.status ?? null, }; } return details; } export class CodexAppServer { constructor({ backendFactory, store, stateFile, backend, cwd = process.cwd(), config, configFile } = {}) { this.configPromise = config ? Promise.resolve(config) : loadBridgeConfig(configFile); this.backendFactoryPromise = backendFactory ? Promise.resolve(backendFactory) : this.configPromise.then((loadedConfig) => createBackendFactory({ backend: backend || loadedConfig.backend, sessionOptions: loadedConfig.sessionOptions, })); this.store = store || new StateStore(stateFile); this.defaultCwd = cwd; this.peers = new Set(); this.sessions = new Map(); this.activeTurns = new Map(); this.startingTurns = new Set(); this.modelCacheSyncKey = null; this.modelCacheSyncPromise = Promise.resolve(); this.externalStatusPoll = null; this.externalStatuses = new Map(); this.externalProjections = new Map(); this.externalControlSubscriptions = new Map(); this.externalControlCursors = new Map(); this.externalControlConnecting = new Map(); this.externalControlSyncTimers = new Map(); this.externalControlSyncPromises = new Map(); this.externalControlConnectControllers = new Map(); this.externalRefreshPromise = null; this.fuzzySearchSessions = new Map(); this.fuzzySearchRequests = new Map(); this.externalAskRequests = new Map(); this.closing = false; } createPeer(send, metadata = {}) { const client = new ClientConnection(send); const peer = { id: id("peer"), client, metadata, initialized: false, initializedNotification: false, clientInfo: null, capabilities: {}, subscriptions: new Set(), disabledNotifications: new Set(), processes: new Map(), }; this.peers.add(peer); this.#startExternalStatusPoll(); protocolTrace("peer.created", { peerId: peer.id, transport: metadata.transport ?? null, remoteAddress: metadata.remoteAddress ?? null, }); return peer; } async closePeer(peer, reason = new Error("Client disconnected")) { if (!this.peers.delete(peer)) return; peer.client.close(reason); protocolTrace("peer.closed", { peerId: peer.id, transport: peer.metadata?.transport ?? null, reason: serializeError(reason), }); for (const proc of peer.processes?.values() || []) { try { proc.child.kill("SIGTERM"); } catch {} } peer.processes?.clear(); const fuzzyPrefix = `${peer.id}:`; for (const [key, session] of this.fuzzySearchSessions) { if (!key.startsWith(fuzzyPrefix)) continue; session.controller?.abort(); this.fuzzySearchSessions.delete(key); } for (const [key, controller] of this.fuzzySearchRequests) { if (!key.startsWith(fuzzyPrefix)) continue; controller.abort(); this.fuzzySearchRequests.delete(key); } } async close() { this.closing = true; clearInterval(this.externalStatusPoll); this.externalStatusPoll = null; for (const scheduled of this.externalControlSyncTimers.values()) clearTimeout(scheduled.timer); this.externalControlSyncTimers.clear(); for (const controller of this.externalControlConnectControllers.values()) controller.abort(); this.externalControlConnectControllers.clear(); for (const subscription of this.externalControlSubscriptions.values()) subscription.close?.(); this.externalControlSubscriptions.clear(); this.externalControlConnecting.clear(); for (const session of this.fuzzySearchSessions.values()) session.controller?.abort(); this.fuzzySearchSessions.clear(); for (const controller of this.fuzzySearchRequests.values()) controller.abort(); this.fuzzySearchRequests.clear(); for (const pending of this.externalAskRequests.values()) { pending.controller.abort(); void controlCancelAsk(pending.sessionId, pending.requestId); } this.externalAskRequests.clear(); const runningTurns = []; for (const active of this.activeTurns.values()) { active.controller.abort(abortError("Server closing")); await active.session.interrupt?.().catch(() => {}); if (active.runPromise) runningTurns.push(active.runPromise); } await Promise.allSettled(runningTurns); await Promise.allSettled([ ...this.externalControlSyncPromises.values(), this.externalRefreshPromise, ]); this.externalControlSyncPromises.clear(); this.externalRefreshPromise = null; await this.store.flush().catch(() => {}); for (const peer of [...this.peers]) await this.closePeer(peer); for (const session of this.sessions.values()) await session.dispose?.().catch(() => {}); this.sessions.clear(); } #startExternalStatusPoll() { if (this.externalStatusPoll) return; this.externalStatusPoll = setInterval(() => { if (this.externalRefreshPromise || this.closing) return; const refresh = this.#refreshExternalStatuses().catch((error) => { debug("external refresh failed", error); }); this.externalRefreshPromise = refresh; void refresh.finally(() => { if (this.externalRefreshPromise === refresh) this.externalRefreshPromise = null; }); }, 1_000); this.externalStatusPoll.unref?.(); } async #refreshExternalStatuses() { const threadIds = new Set(); for (const peer of this.peers) { for (const threadId of peer.subscriptions || []) { if (!this.activeTurns.has(threadId)) threadIds.add(threadId); } } for (const known of [...this.externalProjections.keys()]) { if (!threadIds.has(known)) this.externalProjections.delete(known); } for (const [threadId, subscription] of this.externalControlSubscriptions) { if (threadIds.has(threadId) || (this.activeTurns.has(threadId) && this.#hasExternalSubscribers(threadId))) continue; subscription.close?.(); this.externalControlSubscriptions.delete(threadId); const timer = this.externalControlSyncTimers.get(threadId); clearTimeout(timer?.timer); const controller = this.externalControlConnectControllers.get(threadId); controller?.abort(); this.externalControlConnectControllers.delete(threadId); this.externalControlConnecting.delete(threadId); this.externalControlSyncTimers.delete(threadId); } await Promise.all([...threadIds].map(async (threadId) => { const subscription = await this.#ensureExternalControlSubscription(threadId); const needsDurableAudit = !subscription || Date.now() - Number(subscription.lastDurableSyncAt || 0) >= EXTERNAL_DURABLE_AUDIT_INTERVAL_MS; if (needsDurableAudit) { await this.#enqueueExternalStream(threadId).catch((error) => debug("external stream failed", threadId, error)); } const live = subscription?.status || await this.#externalControlStatus(threadId); const status = live ? this.#codexStatusFromControl(live) : await this.store.getExternalThreadStatus?.(threadId); if (status) await this.#publishExternalStatus(threadId, status); })); } #codexStatusFromControl(status) { return status?.idle && !status?.hasPendingMessages ? { type: "idle" } : { type: "active", activeFlags: [] }; } async #publishExternalStatus(threadId, status) { const statusKey = JSON.stringify(status); const previous = this.externalStatuses.get(threadId); this.externalStatuses.set(threadId, statusKey); if (previous !== statusKey && (previous || status.type === "active")) { await this.#broadcast(threadId, "thread/status/changed", { threadId, status }); } } #enqueueExternalStream(threadId, { terminal = false } = {}) { if (this.closing) return Promise.resolve(); const previous = this.externalControlSyncPromises.get(threadId) || Promise.resolve(); const run = previous.catch(() => {}).then(async () => { const delays = terminal ? [0, 25, 75, 200] : [0]; for (const delay of delays) { if (delay) await sleep(delay); await this.#streamExternalThread(threadId); } }); this.externalControlSyncPromises.set(threadId, run); void run.finally(() => { if (this.externalControlSyncPromises.get(threadId) === run) { this.externalControlSyncPromises.delete(threadId); } }); return run; } #scheduleExternalControlSync(threadId, eventType = "") { if (this.closing) return; const terminal = /(?:_end|session_compact|goal_updated)$/.test(String(eventType)); const scheduled = this.externalControlSyncTimers.get(threadId); if (scheduled) { if (!terminal || scheduled.terminal) return; clearTimeout(scheduled.timer); this.externalControlSyncTimers.delete(threadId); } const timer = setTimeout(() => { this.externalControlSyncTimers.delete(threadId); void this.#enqueueExternalStream(threadId, { terminal }).catch((error) => { debug("live control sync failed", threadId, error); }); }, terminal ? 0 : 25); timer.unref?.(); this.externalControlSyncTimers.set(threadId, { timer, terminal }); } async #handleExternalAskRequest(threadId, request = {}) { const requestId = String(request.requestId || ""); if (!requestId) return; const key = `${threadId}:${requestId}`; if (this.externalAskRequests.has(key)) return; const peers = [...this.peers].filter((peer) => peer.initialized && !peer.client.closed && peer.subscriptions?.has(threadId) ); if (!peers.length) return; const { sessionId } = await this.#controlTarget(threadId); const turnId = (await this.store.getThread(threadId))?.turns?.at(-1)?.id; if (!turnId) return; const controller = new AbortController(); this.externalAskRequests.set(key, { controller, sessionId, requestId }); const questions = normalizeUserInputQuestions({ questions: request.questions || [] }); const itemId = String(request.toolCallId || id("ask")); const askPeer = async (peer) => { let serverRequestId = null; try { const result = await peer.client.request("item/tool/requestUserInput", { threadId, turnId, itemId, questions, isBlocking: true, autoResolutionMs: null, }, { signal: controller.signal, onRequestId: (value) => { serverRequestId = value; }, }); if (!result?.answers || typeof result.answers !== "object") { throw new Error("Ask response contained no answers"); } return result; } finally { if (serverRequestId) { peer.client.notify("serverRequest/resolved", { threadId, requestId: serverRequestId }); } } }; try { const result = await Promise.any(peers.map(askPeer)); controller.abort(abortError("Ask answered by another client")); await controlRespondAsk(sessionId, requestId, normalizeUserInputAnswers(result.answers)); } catch (error) { if (!controller.signal.aborted) { debug("external ask had no client answer", threadId, requestId, error); await controlCancelAsk(sessionId, requestId); } } finally { if (this.externalAskRequests.get(key)?.controller === controller) { this.externalAskRequests.delete(key); } } } #settleExternalAsk(threadId, requestId) { const key = `${threadId}:${String(requestId || "")}`; const pending = this.externalAskRequests.get(key); if (!pending) return; pending.controller.abort(abortError("Ask settled by another client")); this.externalAskRequests.delete(key); } async #handleExternalControlEvent(threadId, frame) { const current = this.externalControlCursors.get(threadId); if (current?.streamId === frame.streamId && Number(frame.sequence || 0) <= Number(current.sequence || 0)) return; this.externalControlCursors.set(threadId, { streamId: frame.streamId, sequence: Number(frame.sequence || 0), }); const subscription = this.externalControlSubscriptions.get(threadId); if (subscription && frame.status) subscription.status = frame.status; const eventType = frame.event?.type || frame.eventType || ""; protocolTrace("external.control_event", { threadId, streamId: frame.streamId || null, sequence: frame.sequence ?? null, eventType: eventType || null, }); if (eventType === "ask_request") { void this.#handleExternalAskRequest(threadId, frame.event).catch((error) => { debug("external ask relay failed", threadId, error); }); } else if (eventType === "ask_resolved" || eventType === "ask_cancelled") { this.#settleExternalAsk(threadId, frame.event?.requestId); } if (this.activeTurns.has(threadId)) return; await this.#publishExternalRecoveryEvent(threadId, frame.event || {}); const status = frame.status ? this.#codexStatusFromControl(frame.status) : (/^(?:agent|turn)_start$/.test(eventType) ? { type: "active", activeFlags: [] } : null); if (status) await this.#publishExternalStatus(threadId, status); this.#scheduleExternalControlSync(threadId, eventType); } async #publishExternalRecoveryEvent(threadId, event) { const type = event?.type || ""; if (![ "auto_retry_start", "retry_start", "retryStart", "retry_fallback_applied", "retryFallbackApplied", "retry_fallback_succeeded", "retryFallbackSucceeded", "model_change", "modelChange", "auto_retry_end", "retry_end", "retryEnd", ].includes(type)) return; const thread = await this.store.getThread(threadId); const turnId = thread?.turns?.at(-1)?.id; if (!turnId) return; if (["auto_retry_start", "retry_start", "retryStart"].includes(type)) { const attempt = Number(event.attempt || 0); const maxAttempts = Number(event.maxAttempts || 0); const delayMs = Number(event.delayMs || 0); const details = [ attempt > 0 ? `attempt ${attempt}${maxAttempts > 0 ? ` of ${maxAttempts}` : ""}` : null, delayMs > 0 ? `retrying in ${delayMs}ms` : "retrying", ].filter(Boolean).join(", "); await this.#broadcast(threadId, "error", { threadId, turnId, error: protocolTurnError({ message: event.errorMessage || event.message || "OMP model request failed", additionalDetails: details || null, code: event.errorId ?? null, }), willRetry: true, }); } else if (type === "retry_fallback_applied" || type === "retryFallbackApplied") { const from = lifecycleModelSelector(event.from) || "the requested model"; const to = lifecycleModelSelector(event.to) || "a fallback model"; await this.#applyProjectedModelChange(threadId, { model: to }); await this.#broadcast(threadId, "warning", { threadId, message: `${from} failed, continuing with ${to}`, }); } else if (type === "retry_fallback_succeeded" || type === "retryFallbackSucceeded") { const model = lifecycleModelSelector(event.model); if (model) await this.#applyProjectedModelChange(threadId, { model }); } else if (["model_change", "modelChange"].includes(type)) { const model = lifecycleModelSelector(event.model); if (model) await this.#applyProjectedModelChange(threadId, { model }); } else if ((type === "auto_retry_end" || type === "retry_end" || type === "retryEnd") && event.success === false) { await this.#broadcast(threadId, "error", { threadId, turnId, error: protocolTurnError({ message: event.finalError || event.errorMessage || "OMP exhausted its model retries", additionalDetails: Number(event.attempt || 0) > 0 ? `failed after ${event.attempt} attempts` : null, code: "omp_retry_exhausted", }), willRetry: false, }); } } #hasExternalSubscribers(threadId) { return [...this.peers].some((peer) => peer.subscriptions?.has(threadId)); } async #ensureExternalControlSubscription(threadId) { if (this.closing) return null; const target = await this.#controlTarget(threadId); if (this.closing) return null; const targetId = target.kind === "prime-rpc" ? (target.activeSessionId || target.sessionId) : target.sessionId; const existing = this.externalControlSubscriptions.get(threadId); if (existing?.targetId === targetId) return existing; if (existing) { existing.close?.(); this.externalControlSubscriptions.delete(threadId); } const connecting = this.externalControlConnecting.get(threadId); if (connecting) return connecting; const cursor = this.externalControlCursors.get(threadId) || {}; const controller = new AbortController(); this.externalControlConnectControllers.set(threadId, controller); const subscribe = target.kind === "prime-rpc" ? primeControlSubscribe : controlSubscribe; const promise = subscribe(target.kind === "prime-rpc" ? target : target.sessionId, { streamId: cursor.streamId || null, afterSequence: cursor.sequence ?? null, signal: controller.signal, onEvent: (frame) => { if (this.closing) return; void this.#handleExternalControlEvent(threadId, frame).catch((error) => { debug("external control event failed", threadId, error); }); }, onDisconnect: () => { const active = this.externalControlSubscriptions.get(threadId); if (active?.targetId === targetId) this.externalControlSubscriptions.delete(threadId); }, }).then(async (handle) => { if (!handle) return null; if (!this.#hasExternalSubscribers(threadId)) { handle.close?.(); return null; } if (this.closing) { handle.close?.(); return null; } const subscription = { ...handle, targetId, sessionId: target.sessionId, status: handle.status || null, lastDurableSyncAt: Date.now(), }; this.externalControlSubscriptions.set(threadId, subscription); const observed = this.externalControlCursors.get(threadId); const observedSequence = observed?.streamId === handle.streamId ? Number(observed.sequence || 0) : 0; this.externalControlCursors.set(threadId, { streamId: handle.streamId, sequence: Math.max(Number(handle.sequence || 0), observedSequence), }); if (handle.status) await this.#publishExternalStatus(threadId, this.#codexStatusFromControl(handle.status)); if (handle.reset) this.#scheduleExternalControlSync(threadId, "snapshot_reset"); for (const pendingAsk of handle.pendingAsks || []) { void this.#handleExternalAskRequest(threadId, { type: "ask_request", ...pendingAsk, }).catch((error) => debug("pending external ask relay failed", threadId, error)); } return subscription; }).catch((error) => { debug("external control subscribe failed", threadId, error); return null; }).finally(() => { if (this.externalControlConnecting.get(threadId) === promise) { this.externalControlConnecting.delete(threadId); } if (this.externalControlConnectControllers.get(threadId) === controller) { this.externalControlConnectControllers.delete(threadId); } }); this.externalControlConnecting.set(threadId, promise); return promise; } #projectionSnapshot(thread) { const turns = Array.isArray(thread?.turns) ? thread.turns : []; const snapshot = new Map(); const windowStart = Math.max(0, turns.length - EXTERNAL_STREAM_TURN_WINDOW); turns.forEach((turn, index) => { if (!turn?.id) return; const items = new Map(); if (index >= windowStart) { for (const item of turn.items || []) { if (item?.id) items.set(item.id, itemSignature(item)); } } snapshot.set(turn.id, { status: turn.status || "inProgress", items }); }); return snapshot; } async #seedExternalProjection(threadId) { if (!threadId || this.externalProjections.has(threadId)) return; const thread = await this.store.getThread(threadId).catch(() => null); if (!thread) return; this.externalProjections.set(threadId, this.#projectionSnapshot(thread)); } async #streamExternalThread(threadId) { const thread = await this.store.getThread(threadId); if (!thread) return; const turns = Array.isArray(thread.turns) ? thread.turns : []; const seeded = this.externalProjections.has(threadId); const previous = this.externalProjections.get(threadId) || new Map(); const next = new Map(); const recent = turns.slice(-EXTERNAL_STREAM_TURN_WINDOW); const recentIds = new Set(recent.map((turn) => turn?.id).filter(Boolean)); for (const [turnId, snapshot] of previous) { if (!recentIds.has(turnId)) next.set(turnId, snapshot); } for (const turn of recent) { if (!turn?.id) continue; const priorTurn = previous.get(turn.id); const items = new Map(); const turnStatus = turn.status || "inProgress"; if (!priorTurn && seeded) { await this.#broadcast(threadId, "turn/started", { threadId, turn }); } for (const item of turn.items || []) { if (!item?.id) continue; const signature = itemSignature(item); items.set(item.id, signature); if (!seeded) continue; const priorSignature = priorTurn?.items?.get(item.id); if (priorSignature === signature) continue; const method = priorSignature === undefined ? "item/started" : (item.status === "completed" || item.status === "failed" ? "item/completed" : "item/updated"); protocolTrace("external.item_emit", { threadId, turnId: turn.id, itemId: item.id, itemType: item.type ?? null, itemStatus: item.status ?? null, method, hadPrior: priorSignature !== undefined, pairedCompletion: method === "item/started" && (item.status === "completed" || item.status === "failed"), }); await this.#broadcast(threadId, method, { threadId, turnId: turn.id, item }); if (method === "item/started" && (item.status === "completed" || item.status === "failed")) { await this.#broadcast(threadId, "item/completed", { threadId, turnId: turn.id, item }); } } next.set(turn.id, { status: turnStatus, items }); if (!seeded) continue; const wasComplete = priorTurn?.status === "completed" || priorTurn?.status === "failed"; const isComplete = turnStatus === "completed" || turnStatus === "failed"; if (isComplete && !wasComplete) { await this.#broadcast(threadId, "turn/completed", { threadId, turn }); const usage = turn.tokenUsage; if (usage && Number(usage.totalTokens || 0) > 0) { await this.#broadcast(threadId, "thread/tokenUsage/updated", { threadId, turnId: turn.id, tokenUsage: { total: usage, last: usage, modelContextWindow: resolveModelContextWindow(thread.model) }, }); } } } this.externalProjections.set(threadId, next); const subscription = this.externalControlSubscriptions.get(threadId); if (subscription) subscription.lastDurableSyncAt = Date.now(); } async handle(peer, message) { if (!message || typeof message !== "object" || Array.isArray(message)) { protocolTrace("rpc.invalid_message", { peerId: peer?.id ?? null, messageShape: protocolShape(message) }); peer.client.send(errorResponse(null, new RpcError(ERROR.INVALID_REQUEST, "Message must be an object"))); return; } if (peer.client.acceptResponse(message)) { protocolTrace("rpc.client_response", { peerId: peer.id, id: message.id ?? null, responseShape: protocolShape(message), }); return; } if (typeof message.method !== "string") { protocolTrace("rpc.missing_method", { peerId: peer.id, id: message.id ?? null, messageShape: protocolShape(message), }); if (message.id !== undefined) { peer.client.send(errorResponse(message.id, new RpcError(ERROR.INVALID_REQUEST, "Request is missing method"))); } return; } const isRequest = message.id !== undefined; const startedAt = Date.now(); protocolTrace(isRequest ? "rpc.request" : "rpc.notification", { peerId: peer.id, transport: peer.metadata?.transport ?? null, id: message.id ?? null, method: message.method, paramsShape: protocolShape(message.params || {}), }); try { const outcome = await this.#dispatch(peer, message.method, message.params || {}, isRequest); if (isRequest) { const result = outcome?.result ?? outcome; protocolTrace("rpc.response", { peerId: peer.id, id: message.id, method: message.method, durationMs: Date.now() - startedAt, ok: true, ...protocolResultDetails(message.method, result), }); peer.client.send(successResponse(message.id, result)); } if (outcome?.after) queueMicrotask(() => Promise.resolve(outcome.after()).catch((error) => log("post-response task failed", error))); } catch (error) { protocolTrace("rpc.response", { peerId: peer.id, id: message.id ?? null, method: message.method, durationMs: Date.now() - startedAt, ok: false, error: serializeError(error), }); if (isRequest) peer.client.send(errorResponse(message.id, error)); else log(`notification ${message.method} failed`, error); } } async #dispatch(peer, method, params, isRequest) { if (method === "initialize") return this.#initialize(peer, params); if (method === "initialized") { if (!peer.initialized) throw new RpcError(ERROR.NOT_INITIALIZED, "initialize must be called first"); peer.initializedNotification = true; return {}; } if (!peer.initialized) throw new RpcError(ERROR.NOT_INITIALIZED, "initialize must be called before other methods"); switch (method) { case "fuzzyFileSearch": return this.#fuzzyFileSearch(peer, params); case "fuzzyFileSearch/sessionStart": return this.#fuzzyFileSearchSessionStart(peer, params); case "fuzzyFileSearch/sessionUpdate": return this.#fuzzyFileSearchSessionUpdate(peer, params); case "fuzzyFileSearch/sessionStop": return this.#fuzzyFileSearchSessionStop(peer, params); case "model/list": return this.#modelList(peer, params); case "modelProvider/capabilities/read": return { namespaceTools: true, imageGeneration: false, webSearch: true }; case "permissionProfile/list": return { data: [], nextCursor: null }; case "collaborationMode/list": return { data: [ { name: "Plan", mode: "plan", model: null, reasoning_effort: "medium" }, { name: "Default", mode: "default", model: null, reasoning_effort: null }, ], }; case "getAuthStatus": return this.#getAuthStatus(params); case "experimentalFeature/list": return { data: [], nextCursor: null }; case "experimentalFeature/enablement/set": return { enablement: params?.enablement || {} }; case "account/read": return this.#accountRead(); case "account/login/start": return this.#accountLoginStart(peer, params); case "account/login/cancel": return this.#accountLoginCancel(peer, params); case "account/logout": return this.#accountLogout(peer, params); case "account/consumeRateLimitResetCredit": return { outcome: "success" }; case "account/rateLimits/read": return this.#accountRateLimitsRead(); case "account/usage/read": return this.#accountUsageRead(); case "account/workspaceMessages/read": return this.#accountWorkspaceMessagesRead(); case "configRequirements/read": return this.#configRequirementsRead(); case "plugin/installed": return this.#pluginCatalog(); case "skills/list": { const cwds = Array.isArray(params?.cwds) && params.cwds.length ? params.cwds : [params?.cwd || this.defaultCwd]; return { data: cwds.map((cwd) => ({ cwd, skills: [], errors: [] })), }; } case "plugin/list": return this.#pluginCatalog(); case "plugin/read": return { plugin: null }; case "app/installed": return { apps: [] }; case "app/list": return { apps: [] }; case "app/read": return { app: null }; case "skills/extraRoots/set": return this.#skillsExtraRootsSet(params); case "skills/config/write": return this.#skillsConfigWrite(params); case "hooks/list": { const cwds = Array.isArray(params?.cwds) && params.cwds.length ? params.cwds : [params?.cwd || this.defaultCwd]; return { data: cwds.map((cwd) => ({ cwd: path.resolve(cwd), hooks: [], })), nextCursor: null, }; } case "process/spawn": return this.#processSpawn(peer, params); case "process/writeStdin": return this.#processWriteStdin(peer, params); case "process/kill": return this.#processKill(peer, params); case "process/resizePty": return this.#processResizePty(peer, params); case "mcpServer/list": case "mcpServerStatus/list": return { data: [], nextCursor: null }; case "mcpServer/resource/read": return { data: [] }; case "externalAgentConfig/detect": return this.#externalAgentConfigDetect(); case "externalAgentConfig/import/readHistories": return { data: [], connectors: [], histories: [] }; case "remoteControl/status/read": return { status: "disabled", serverName: "omp-codex-bridge", installationId: "omp", environmentId: null }; case "server/diagnostics": return { ok: true, process: { id: process.pid }, gauges: [] }; case "project/list": return { data: [], nextCursor: null }; case "threadSection/list": return { data: [], nextCursor: null }; case "thread/settings/update": return this.#threadSettingsUpdate(peer, params); case "thread/increment_elicitation": case "thread/decrement_elicitation": return { count: 0, paused: false }; case "windowsSandbox/setupStart": return this.#windowsSandboxSetupStart(); case "feedback/upload": return this.#feedbackUpload(); case "config/value/write": case "config/batchWrite": case "config/write": return this.#configWrite(peer, params); case "remoteControl/enable": return {}; case "config/read": return this.#configRead(peer, params); case "features/list": return this.#featuresList(); case "thread/start": return this.#threadStart(peer, params); case "thread/resume": return this.#threadResume(peer, params); case "thread/fork": return this.#threadFork(peer, params); case "thread/inject_items": return this.#threadInjectItems(params); case "thread/list": return this.#threadList(peer, params); case "thread/read": return this.#threadRead(peer, params); case "thread/turns/list": return this.#threadTurnsList(params); case "thread/items/list": return this.#threadItemsList(params); case "thread/metadata/update": return this.#threadMetadataUpdate(peer, params); case "thread/unsubscribe": return this.#threadUnsubscribe(peer, params); case "thread/name/set": return this.#threadNameSet(peer, params); case "thread/archive": return this.#threadArchive(peer, params, true); case "thread/unarchive": return this.#threadArchive(peer, params, false); case "thread/delete": return this.#threadDelete(peer, params); case "thread/rollback": return this.#threadRollback(peer, params); case "thread/compact/start": return this.#threadCompact(peer, params); case "thread/goal/get": return this.#threadGoalGet(params); case "thread/goal/set": return this.#threadGoalSet(peer, params); case "thread/goal/clear": return this.#threadGoalClear(peer, params); case "thread/loaded/list": return { threadIds: Array.from(this.sessions.keys()) }; case "turn/start": return this.#turnStart(peer, params); case "turn/steer": return this.#turnSteer(peer, params); case "turn/interrupt": return this.#turnInterrupt(peer, params); case "command/exec": return this.#commandExec(peer, params); case "fs/getMetadata": case "fs/stat": return this.#fsGetMetadata(params); case "fs/readDirectory": case "fs/listDirectory": case "fs/list": return this.#fsReadDirectory(params); case "fs/readFile": case "fs/read": return this.#fsRead(params); case "fs/writeFile": case "fs/write": return this.#fsWrite(peer, params); case "fs/createDirectory": case "fs/mkdir": return this.#fsCreateDirectory(peer, params); case "fs/remove": case "fs/delete": return this.#fsRemove(peer, params); case "fs/copy": return this.#fsCopy(peer, params); case "fs/watch": return this.#fsWatch(params); case "fs/unwatch": return this.#fsUnwatch(params); case "ping": return { ok: true, time: Date.now() }; default: if (!isRequest) return {}; log(`unsupported codex method requested: ${method}`, JSON.stringify(params)); throw new RpcError(ERROR.METHOD_NOT_FOUND, `Unsupported Codex app-server method: ${method}`, { method, bridge: "omp-codex-bridge", }); } } #initialize(peer, params) { if (peer.initialized) throw new RpcError(ERROR.ALREADY_INITIALIZED, "Connection is already initialized"); peer.initialized = true; peer.clientInfo = params.clientInfo || params.client || null; peer.capabilities = params.capabilities || {}; const disabled = params.capabilities?.optOutNotificationMethods || params.disableNotifications || params.disabledNotifications || []; peer.disabledNotifications = new Set(Array.isArray(disabled) ? disabled : []); return initializeResult(); } async #fuzzyFileSearch(peer, params = {}) { const token = typeof params.cancellationToken === "string" && params.cancellationToken ? params.cancellationToken : "legacy"; const key = `${peer.id}:${token}`; this.fuzzySearchRequests.get(key)?.abort(); const controller = new AbortController(); this.fuzzySearchRequests.set(key, controller); try { const files = await searchWorkspaceFiles(params.query || "", params.roots || [], controller.signal); return { files }; } finally { if (this.fuzzySearchRequests.get(key) === controller) this.fuzzySearchRequests.delete(key); } } #fuzzyFileSearchSessionStart(peer, params = {}) { const sessionId = requireString(params.sessionId, "sessionId"); const key = `${peer.id}:${sessionId}`; this.fuzzySearchSessions.get(key)?.controller?.abort(); this.fuzzySearchSessions.set(key, { roots: Array.isArray(params.roots) ? params.roots.slice() : [], generation: 0, controller: null, }); return {}; } #fuzzyFileSearchSessionUpdate(peer, params = {}) { const sessionId = requireString(params.sessionId, "sessionId"); const key = `${peer.id}:${sessionId}`; const session = this.fuzzySearchSessions.get(key); if (!session) throw new RpcError(ERROR.NOT_FOUND, `Fuzzy file search session not found: ${sessionId}`); session.controller?.abort(); const controller = new AbortController(); const generation = ++session.generation; session.controller = controller; const query = String(params.query || ""); return { result: {}, after: async () => { const files = await searchWorkspaceFiles(query, session.roots, controller.signal).catch((error) => { debug("fuzzy file search failed", sessionId, error); return []; }); const current = this.fuzzySearchSessions.get(key); if (controller.signal.aborted || current !== session || current.generation !== generation || peer.client.closed) return; if (!peer.disabledNotifications.has("fuzzyFileSearch/sessionUpdated")) { peer.client.notify("fuzzyFileSearch/sessionUpdated", { sessionId, query, files }); } if (!peer.disabledNotifications.has("fuzzyFileSearch/sessionCompleted")) { peer.client.notify("fuzzyFileSearch/sessionCompleted", { sessionId }); } }, }; } #fuzzyFileSearchSessionStop(peer, params = {}) { const sessionId = requireString(params.sessionId, "sessionId"); const key = `${peer.id}:${sessionId}`; this.fuzzySearchSessions.get(key)?.controller?.abort(); this.fuzzySearchSessions.delete(key); return {}; } async #modelList(peer, params) { const [backend, config] = await Promise.all([this.backendFactoryPromise, this.configPromise]); const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex"); const codexToml = (await readTomlFile(path.join(codexHome, "config.toml"))) || {}; const effectiveDefault = config.defaultModel || codexToml.model || "gpt-5.6-luna"; const canonicalCatalog = clientNeedsCanonicalModelIds(peer) && typeof backend.listCanonicalModels === "function"; const backendModels = canonicalCatalog ? await backend.listCanonicalModels(effectiveDefault) : await backend.listModels(); if (backendModels.some((m) => m.id?.startsWith("mock/"))) { const data = backendModels.map((m) => ({ ...m, supportedReasoningEfforts: formatReasoningEfforts(m.supportedReasoningEfforts), defaultReasoningEffort: m.defaultReasoningEffort || "medium", inputModalities: ["text", "image"], supportsPersonality: false, modelSpecialty: m.modelSpecialty ?? null, multiAgentVersion: m.multiAgentVersion ?? null, additionalSpeedTiers: [], serviceTiers: [], defaultServiceTier: null, isDefault: Boolean(m.isDefault), })); return { data, nextCursor: null }; } const unfilteredCatalog = backendCatalogIsUnfiltered(backend); const map = new Map(); for (const m of backendModels) { if (!m.id || m.id === "omp/default") continue; const catalogId = normalizeModelParam(m, m.provider)?.id || m.id; const filterId = m.catalogFilterId || catalogId; if (!catalogRowAllowed(config.modelCatalog, m, filterId)) continue; const configuredVisible = Array.isArray(config.modelCatalog?.include) && config.modelCatalog.include.length > 0; const unfiltered = configuredVisible || unfilteredCatalog || m.catalogPassthrough === true; const hidden = Boolean(m.hidden) || (!unfiltered && !FEATURED_MODEL_IDS.includes(filterId)); const candidate = { ...m, id: catalogId, model: catalogId, hidden }; const existing = map.get(catalogId); const isCanonicalInput = m.id === catalogId; if (!existing || isCanonicalInput) map.set(catalogId, candidate); } const catalogRows = Array.from(map.values()); const clientDefault = canonicalCatalog ? splitTaggedModel(effectiveDefault).id : effectiveDefault; const resolvedDefault = config.resolveModel(effectiveDefault); const exactDefault = map.get(clientDefault); const defaultRow = exactDefault && !exactDefault.hidden ? exactDefault : catalogRows.find((model) => { if (model.hidden) return false; const slug = model.id || model.model || ""; return model.selector === resolvedDefault || model.backendSelector === resolvedDefault || config.resolveModel(slug) === resolvedDefault; }) || exactDefault || null; if (defaultRow?.hidden) defaultRow.hidden = false; const defaultSlug = defaultRow?.id || defaultRow?.model || null; const isDefaultModel = (model) => (model?.id || model?.model || "") === defaultSlug; const sortedRows = catalogRows.sort((a, b) => { const aSlug = a.id || a.model || ""; const bSlug = b.id || b.model || ""; const aConfiguredRank = configuredCatalogRank(config.modelCatalog, a); const bConfiguredRank = configuredCatalogRank(config.modelCatalog, b); if (aConfiguredRank.backend !== bConfiguredRank.backend) return aConfiguredRank.backend - bConfiguredRank.backend; if (aConfiguredRank.model !== bConfiguredRank.model) return aConfiguredRank.model - bConfiguredRank.model; const aDef = isDefaultModel(a); const bDef = isDefaultModel(b); if (aDef && !bDef) return -1; if (!aDef && bDef) return 1; const aIdx = FEATURED_MODEL_IDS.indexOf(aSlug); const bIdx = FEATURED_MODEL_IDS.indexOf(bSlug); if (aIdx !== -1 && bIdx !== -1) return aIdx - bIdx; if (aIdx !== -1) return -1; if (bIdx !== -1) return 1; if (Boolean(a.hidden) !== Boolean(b.hidden)) return a.hidden ? 1 : -1; return 0; }); const rows = sortedRows; const selectableRows = params.includeHidden ? rows : rows.filter((model) => !model.hidden); const cursor = Number(params.cursor || 0) || 0; const requestedLimit = Math.max(1, Math.min(200, Number(params.limit || 100))); const limit = params.includeHidden && cursor === 0 && requestedLimit >= 100 ? Math.max(requestedLimit, selectableRows.length) : requestedLimit; const data = selectableRows.slice(cursor, cursor + limit).map((model) => ({ id: model.id || model.model, model: model.model || model.id, upgrade: null, upgradeInfo: null, availabilityNux: null, displayName: model.displayName || model.name || model.id, description: model.description || "", hidden: Boolean(model.hidden), supportedReasoningEfforts: formatReasoningEfforts(model.supportedReasoningEfforts), defaultReasoningEffort: model.defaultReasoningEffort || "medium", inputModalities: ["text", "image"], supportsPersonality: false, modelSpecialty: model.modelSpecialty ?? null, multiAgentVersion: model.multiAgentVersion ?? null, additionalSpeedTiers: [], serviceTiers: [], defaultServiceTier: null, isDefault: isDefaultModel(model), })); const modelCacheSyncKey = JSON.stringify(rows.map((model) => [ model.id, model.displayName, model.description, model.defaultReasoningEffort, model.supportedReasoningEfforts, model.contextWindow, model.hidden, ])); if (modelCacheSyncKey !== this.modelCacheSyncKey) { const sync = this.modelCacheSyncPromise.then(async () => { if (modelCacheSyncKey === this.modelCacheSyncKey) return; await syncCodexModelsCache(rows); this.modelCacheSyncKey = modelCacheSyncKey; }); this.modelCacheSyncPromise = sync.catch(() => {}); await sync.catch(() => {}); } return { data, nextCursor: cursor + data.length < selectableRows.length ? String(cursor + data.length) : null }; } #getAuthStatus(params = {}) { const includeToken = Boolean(params?.includeToken); return { authMethod: "apikey", authToken: includeToken ? "sk-omp-local" : null, requiresOpenaiAuth: false, }; } #accountRead() { return { account: { type: "apiKey" }, requiresOpenaiAuth: false, }; } async #accountLoginStart(peer, params = {}) { const type = params?.type || "chatgpt"; const loginId = params?.loginId || id("login"); setImmediate(() => { peer?.client?.notify("account/login/completed", { loginId: type === "chatgpt" || type === "chatgptDeviceCode" ? loginId : null, success: true, error: null, onboardingEntrypoint: null, }); peer?.client?.notify("account/updated", { authMode: "apikey", planType: null, }); }); if (type === "apiKey") { return { type: "apiKey" }; } if (type === "chatgptDeviceCode") { return { type: "chatgptDeviceCode", loginId, verificationUrl: "http://127.0.0.1", userCode: "OMP-0000", }; } if (type === "chatgpt") { return { type: "chatgpt", loginId, authUrl: "http://127.0.0.1", }; } return { type: "chatgptAuthTokens" }; } async #accountLoginCancel(peer, params = {}) { const loginId = params?.loginId || null; setImmediate(() => { peer?.client?.notify("account/login/completed", { loginId, success: false, error: "Login cancelled", onboardingEntrypoint: null, }); }); return { status: "canceled" }; } async #accountLogout(peer, _params = {}) { setImmediate(() => { peer?.client?.notify("account/updated", { authMode: null, planType: null, }); }); return {}; } #accountRateLimitsRead() { return { rateLimits: { limitId: "codex-pro", limitName: "Codex Pro", primary: { used: 0, limit: 1000, percentUsed: 0.0, resetsAt: unixSeconds() + 86400, windowDurationMins: 1440, }, secondary: null, credits: null, individualLimit: null, spendControlReached: false, planType: "pro", rateLimitReachedType: null, }, rateLimitsByLimitId: {}, rateLimitResetCredits: null, }; } #accountUsageRead() { return { summary: { lifetimeTokens: 50000, peakDailyTokens: 10000, longestRunningTurnSec: 5, }, dailyUsageBuckets: [ { date: new Date().toISOString().split("T")[0], tokens: 10000, }, ], threadUsage: null, }; } #accountWorkspaceMessagesRead() { return { messages: [] }; } #configRequirementsRead() { return { requirements: null }; } #skillsExtraRootsSet() { return {}; } #skillsConfigWrite() { return {}; } async #processSpawn(peer, params) { const handle = requireString(params.processHandle || params.handle, "processHandle"); if (peer.processes.has(handle)) { throw new RpcError(ERROR.INVALID_PARAMS, `Process handle already running: ${handle}`); } const rawCommand = Array.isArray(params.command) ? params.command.map(String) : [String(params.command || "")]; if (!rawCommand.length || !rawCommand[0]) { throw new RpcError(ERROR.INVALID_PARAMS, "command must not be empty"); } const env = { ...process.env, ...(params.env || {}) }; const command = adaptWorktreeSpawnCommand(rawCommand, env); const cwd = params.cwd ? path.resolve(params.cwd) : this.defaultCwd; const outputBytesCap = Number(params.outputBytesCap) > 0 ? Number(params.outputBytesCap) : 10 * 1024 * 1024; const streamStdoutStderr = Boolean(params.streamStdoutStderr || params.tty); const timeoutMs = Number(params.timeoutMs) > 0 ? Number(params.timeoutMs) : 120_000; const child = spawn(command[0], command.slice(1), { cwd, env, stdio: ["pipe", "pipe", "pipe"], }); let stdoutChunks = []; let stderrChunks = []; let stdoutBytes = 0; let stderrBytes = 0; let stdoutCapReached = false; let stderrCapReached = false; const record = { handle, child }; peer.processes.set(handle, record); const timer = setTimeout(() => { try { child.kill("SIGTERM"); } catch {} }, timeoutMs); timer.unref?.(); child.stdout.on("data", (chunk) => { if (stdoutCapReached) return; if (stdoutBytes + chunk.length > outputBytesCap) { chunk = chunk.subarray(0, Math.max(0, outputBytesCap - stdoutBytes)); stdoutCapReached = true; } stdoutBytes += chunk.length; if (streamStdoutStderr) { peer.client.notify("process/outputDelta", { processHandle: handle, stream: "stdout", deltaBase64: chunk.toString("base64"), capReached: stdoutCapReached, }); } else { stdoutChunks.push(chunk); } }); child.stderr.on("data", (chunk) => { if (stderrCapReached) return; if (stderrBytes + chunk.length > outputBytesCap) { chunk = chunk.subarray(0, Math.max(0, outputBytesCap - stderrBytes)); stderrCapReached = true; } stderrBytes += chunk.length; if (streamStdoutStderr) { peer.client.notify("process/outputDelta", { processHandle: handle, stream: "stderr", deltaBase64: chunk.toString("base64"), capReached: stderrCapReached, }); } else { stderrChunks.push(chunk); } }); const cleanup = (code, signal) => { clearTimeout(timer); peer.processes.delete(handle); const exitCode = typeof code === "number" ? code : signal ? 128 + 15 : 1; const stdout = streamStdoutStderr ? "" : Buffer.concat(stdoutChunks).toString("utf8"); const stderr = streamStdoutStderr ? "" : Buffer.concat(stderrChunks).toString("utf8"); peer.client.notify("process/exited", { processHandle: handle, exitCode, stdout, stdoutCapReached, stderr, stderrCapReached, }); }; child.once("error", (error) => { if (!streamStdoutStderr) stderrChunks.push(Buffer.from(error.message || String(error))); cleanup(1, null); }); child.once("close", (code, signal) => { cleanup(code, signal); }); return {}; } #processWriteStdin(peer, params) { const handle = requireString(params.processHandle || params.handle, "processHandle"); const record = peer.processes?.get(handle); if (!record) throw new RpcError(ERROR.NOT_FOUND, `Process not found: ${handle}`); if (params.deltaBase64) { const buffer = Buffer.from(params.deltaBase64, "base64"); record.child.stdin?.write(buffer); } if (params.closeStdin) { record.child.stdin?.end(); } return {}; } #processKill(peer, params) { const handle = requireString(params.processHandle || params.handle, "processHandle"); const record = peer.processes?.get(handle); if (record) { try { record.child.kill("SIGTERM"); } catch {} } return {}; } #processResizePty() { return {}; } #pluginCatalog() { return { marketplaces: [], featuredPluginIds: [], marketplaceLoadErrors: [], }; } #externalAgentConfigDetect() { return { items: [] }; } #windowsSandboxReadiness() { return { ready: false }; } #windowsSandboxSetupStart() { return { started: false }; } #feedbackUpload() { return { trackingId: id("feedback") }; } async #configWrite(peer, params = {}) { const config = await this.configPromise; const edits = params.edits || (params.keyPath ? [{ keyPath: params.keyPath, value: params.value }] : []); for (const edit of edits) { if (edit.keyPath === "model" && edit.value) { const backend = await this.backendFactoryPromise; config.defaultModel = clientNeedsCanonicalModelIds(peer) && typeof backend?.defaultModelForPersistentState === "function" ? await backend.defaultModelForPersistentState(edit.value) : edit.value; } if (edit.keyPath === "model_reasoning_effort" && edit.value) { config.defaultReasoningEffort = edit.value; } if (edit.keyPath === "approval_policy" && edit.value) { config.defaultApprovalPolicy = edit.value; } } if (this.upstreamCache) this.upstreamCache.clear(); try { if (config.file) { const current = await readJson(config.file, {}); if (config.defaultModel) current.defaultModel = config.defaultModel; if (config.defaultReasoningEffort) current.defaultReasoningEffort = config.defaultReasoningEffort; if (config.defaultApprovalPolicy) current.defaultApprovalPolicy = config.defaultApprovalPolicy; await atomicWriteJson(config.file, current); } } catch {} return { status: "ok", version: "1.0", filePath: config.file || "", overriddenMetadata: null, ok: true }; } async #queryRealCodex(method, params = {}) { if (process.env.OMP_CODEX_NESTED) return null; if (!this.upstreamCache) this.upstreamCache = new Map(); const key = `${method}:${JSON.stringify(params)}`; const cached = this.upstreamCache.get(key); if (cached && Date.now() - cached.time < 300_000) return cached.value; try { let codexBin = process.env.OMP_CODEX_UPSTREAM_BIN; if (!codexBin) { const bak1 = "/Users/dawn/.codex/packages/standalone/current/bin/codex.upstream-bak"; const bak2 = "/Applications/ChatGPT.app/Contents/Resources/codex.upstream-bak"; try { const s1 = await fs.stat(bak1); if (s1.size > 1000000) codexBin = bak1; } catch {} if (!codexBin) { try { const s2 = await fs.stat(bak2); if (s2.size > 1000000) codexBin = bak2; } catch {} } } if (!codexBin) return null; const { spawn } = await import("node:child_process"); const p = spawn(codexBin, ["app-server", "--stdio"], { stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, OMP_CODEX_NESTED: "1" }, }); const result = await new Promise((resolve) => { const timer = setTimeout(() => { try { p.kill(); } catch {} resolve(null); }, 2500); timer.unref?.(); let buffer = ""; p.stdout.on("data", (chunk) => { const lines = buffer.split("\n"); buffer = lines.pop() || ""; for (const line of lines) { try { const json = JSON.parse(line); if (json.id === 2 && json.result) { clearTimeout(timer); try { p.kill(); } catch {} resolve(json.result); return; } } catch {} } }); p.stdin.write(JSON.stringify({ id: 1, method: "initialize", params: { clientInfo: { name: "omp-codex-bridge", version: "1.0" }, capabilities: { experimentalApi: true } } }) + "\n"); p.stdin.write(JSON.stringify({ method: "initialized", params: {} }) + "\n"); p.stdin.write(JSON.stringify({ id: 2, method, params }) + "\n"); p.once("close", () => resolve(null)); p.once("error", () => resolve(null)); }); if (result) { this.upstreamCache.set(key, { time: Date.now(), value: result }); return result; } } catch {} return null; } async #configRead(peer, params) { const config = await this.configPromise; const codexHome = process.env.CODEX_HOME || path.join(os.homedir(), ".codex"); const codexTomlPath = path.join(codexHome, "config.toml"); const codexToml = (await readTomlFile(codexTomlPath)) || {}; const configuredModel = config.defaultModel || codexToml.model || "gpt-5.6-luna"; const canonicalModelIds = clientNeedsCanonicalModelIds(peer); const backend = await this.backendFactoryPromise; const model = typeof backend?.clientModelForState === "function" ? await backend.clientModelForState(null, configuredModel, { canonical: canonicalModelIds }) : (canonicalModelIds ? splitTaggedModel(configuredModel).id : configuredModel); const approvalPolicy = config.defaultApprovalPolicy || codexToml.approval_policy || "never"; const reasoningEffort = config.defaultReasoningEffort || codexToml.model_reasoning_effort || "high"; const base = defaultCodexConfig(model, approvalPolicy, reasoningEffort); const merged = mergeCodexConfig(base, codexToml); merged.model = model; merged.approval_policy = approvalPolicy; merged.model_reasoning_effort = reasoningEffort; if (backend?.modelProviderId) { merged.model_providers = { ...merged.model_providers, [backend.modelProviderId]: { name: "OMP Codex Bridge", base_url: "http://127.0.0.1", wire_api: "responses", }, }; merged.model_provider = backend.modelProviderId; } else { merged.model_provider = resolveConfiguredProvider(merged.model_provider, merged.model_providers); } const result = { config: merged, origins: {} }; if (params?.includeLayers) { result.layers = [ ...(await fs.stat(codexTomlPath).then(() => [{ name: "codex-toml", path: codexTomlPath }]).catch(() => [])), ...(config.file ? [{ name: "omp-codex-bridge", path: config.file }] : []), ]; } return result; } #featuresList() { return { data: [ { name: "threads", enabled: true }, { name: "streaming", enabled: true }, { name: "reasoning", enabled: true }, { name: "goals", enabled: true }, { name: "commandApprovals", enabled: true }, { name: "fileApprovals", enabled: true }, { name: "steering", enabled: true }, { name: "interrupt", enabled: true }, { name: "worktrees", enabled: true }, ], nextCursor: null, }; } async #threadStart(peer, params) { const config = await this.configPromise; const threadId = id("thread"); const requestedCwd = path.resolve(params.cwd || this.defaultCwd); const isolationMode = params.workspaceIsolation || (params.collaborationMode?.mode === "worktree" ? "auto" : config.workspaceIsolation); const workspace = await createIsolatedWorkspace({ threadId, sourceCwd: requestedCwd, isolationMode }); const effectiveCwd = workspace.cwd; const vcsInfo = await getVcsInfo(effectiveCwd); const requestedModel = params.model || params.config?.model || params.collaborationMode?.settings?.model || config.defaultModel; const requestedEffort = reasoningEffortFromParams(params) || config.defaultReasoningEffort; const resolvedModel = config.resolveModel(requestedModel); const model = await this.#persistentThreadModel(null, resolvedModel, params.modelProvider); const thread = await this.store.createThread({ threadId, cwd: effectiveCwd, model, reasoningEffort: requestedEffort, ephemeral: Boolean(params.ephemeral), source: params.source || "appServer", workspace: workspace.isolated ? workspace : null, gitInfo: vcsInfo, }); const session = await this.#openSession(thread, { model: resolvedModel, reasoningEffort: requestedEffort, approvalPolicy: params.approvalPolicy || config.defaultApprovalPolicy, sandboxPolicy: params.sandbox || params.sandboxPolicy, }); await this.#persistSession(thread.id, session); peer.subscriptions.add(thread.id); const effectiveParams = { ...params, model: requestedModel, approvalPolicy: params.approvalPolicy || config.defaultApprovalPolicy, reasoningEffort: requestedEffort }; const projectedThread = await this.#projectThreadModel(peer, thread); const response = threadStartResponse( projectedThread, { ...effectiveParams, model: projectedThread.model }, normalizeModelParam(projectedThread.model, projectedThread.modelProvider), ); return { result: response, after: () => this.#broadcast(thread.id, "thread/started", { thread }), }; } async #threadResume(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); const config = await this.configPromise; const requestedModel = params.model ?? params.config?.model ?? params.collaborationMode?.settings?.model ?? null; const requestedEffort = reasoningEffortFromParams(params); const normalized = requestedModel ? normalizeThreadModel( requestedModel, config.resolveModel(requestedModel), params.modelProvider || thread.modelProvider, ) : null; const resolvedModel = config.resolveModel( normalized?.selector || requestedModel || thread.modelSelector || thread.model, ); const record = await this.store.getRecord(threadId); const persistentModel = requestedModel ? await this.#persistentThreadModel(record?.backend, resolvedModel, params.modelProvider || thread.modelProvider) : null; const resolvedEffort = requestedEffort || thread.reasoningEffort; const live = await this.#externalControlStatus(threadId); if (live) { const configured = await this.#controlConfigure(threadId, { model: resolvedModel, reasoningEffort: resolvedEffort, }); if (!configured && (requestedModel || requestedEffort)) { throw new RpcError(ERROR.INVALID_PARAMS, "The running OMP session rejected the requested model settings"); } thread.status = live.idle && !live.hasPendingMessages ? { type: "idle" } : { type: "active", activeFlags: [] }; } else { await this.#openSession(thread, { model: resolvedModel, reasoningEffort: resolvedEffort, approvalPolicy: params.approvalPolicy || thread.approvalPolicy, sandboxPolicy: params.sandbox || params.sandboxPolicy, }); } await this.#settleAbandonedTurns(threadId, thread, live); if (persistentModel || requestedEffort) { await this.store.updateThread(threadId, (stored) => { if (persistentModel) { stored.model = persistentModel.id; stored.modelProvider = persistentModel.provider || stored.modelProvider; stored.modelSelector = persistentModel.selector || null; } if (requestedEffort) stored.reasoningEffort = requestedEffort; }); if (persistentModel) { thread.model = persistentModel.id; thread.modelProvider = persistentModel.provider || thread.modelProvider; thread.modelSelector = persistentModel.selector || null; } if (requestedEffort) thread.reasoningEffort = requestedEffort; } peer.subscriptions.add(threadId); await this.#seedExternalProjection(threadId); await this.#ensureExternalControlSubscription(threadId); const projectedThread = await this.#projectThreadModel(peer, thread); const response = threadStartResponse( projectedThread, { ...params, model: projectedThread.model }, normalizeModelParam(projectedThread.model, projectedThread.modelProvider), ); return { result: response, after: thread.goal ? () => this.#broadcast(threadId, "thread/goal/updated", { threadId, turnId: null, goal: thread.goal }) : null, }; } /** * Close out turns nobody is running anymore. * * A turn left `inProgress` - our own compaction placeholder, or any turn * whose daemon died mid-flight - keeps ChatGPT's composer in working state * forever: the next message queues behind a turn that already ended, * `turn/steer` fails because no turn is actually active, and only the stop * button breaks the deadlock. Nothing is running here, so say so. */ async #settleAbandonedTurns(threadId, thread, live = null) { if (this.activeTurns.has(threadId) || this.startingTurns.has(threadId)) return false; if (live && !(live.idle && !live.hasPendingMessages)) return false; if (!live) { // No control socket: only the session log itself can prove someone else // is mid-turn. The stored status cannot - it is the very thing that got // stuck, so trusting it would keep the thread wedged forever. const { sessionPath } = await this.#controlTarget(threadId); const activity = sessionPath ? await inspectOmpSessionActivity(sessionPath).catch(() => null) : null; if (activity?.status?.type === "active") return false; } const stale = (thread.turns || []).filter((turn) => turn?.id && turn.status === "inProgress"); if (!stale.length) return false; for (const turn of stale) { const items = turn.items || []; const compactionOnly = items.length > 0 && items.every((item) => item?.type === "contextCompaction"); turn.status = compactionOnly ? "completed" : "interrupted"; turn.completedAt = turn.completedAt || turn.startedAt || unixSeconds(); for (const item of items) settleAbandonedItem(item); const settled = await this.store.updateTurn(threadId, turn.id, (stored) => { stored.status = turn.status; stored.completedAt = turn.completedAt; for (const item of stored.items || []) settleAbandonedItem(item); }); await this.#broadcast(threadId, "turn/completed", { threadId, turn: settled || turn }); } if (thread.status?.type === "active") { thread.status = { type: "idle" }; await this.store.updateThread(threadId, (stored) => { stored.status = { type: "idle" }; }); await this.#broadcast(threadId, "thread/status/changed", { threadId, status: { type: "idle" } }); } return true; } async #threadSettingsUpdate(peer, params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); const config = await this.configPromise; const requestedModel = params.model ?? params.config?.model ?? params.collaborationMode?.settings?.model ?? null; const requestedEffort = reasoningEffortFromParams(params) || params.collaborationMode?.settings?.reasoning_effort || null; const record = await this.store.getRecord(threadId); let normalized = null; if (requestedModel) { normalized = normalizeThreadModel(requestedModel, config.resolveModel(requestedModel), thread.modelProvider); } if (requestedModel) { const currentBackendTag = tagForPersistentState(record?.backend); const requestedBackendTag = splitTaggedModel(config.resolveModel(requestedModel)).tag; if (currentBackendTag && requestedBackendTag && currentBackendTag !== requestedBackendTag) { throw new RpcError( ERROR.INVALID_PARAMS, `Cannot switch an existing ${currentBackendTag === "omp" ? "OMP" : "Prime Agent"} thread to ${requestedBackendTag === "omp" ? "OMP" : "Prime Agent"}; start a new thread instead` ); } } const persistentModel = normalized ? await this.#persistentThreadModel(record?.backend, normalized.selector || requestedModel, normalized.provider || thread.modelProvider) : null; const updatedModel = persistentModel?.id || codexModelId(thread.model); const updatedSelector = persistentModel?.selector || thread.modelSelector || config.resolveModel(updatedModel); const updatedProvider = persistentModel?.provider || thread.modelProvider || "openai"; const updatedCwd = params.cwd ? path.resolve(params.cwd) : thread.cwd; const updatedApprovalPolicy = params.approvalPolicy || thread.approvalPolicy || "on-request"; const updatedEffort = requestedEffort || thread.reasoningEffort || "medium"; const live = await this.#externalControlStatus(threadId); if (live && !await this.#controlConfigure(threadId, { model: normalized ? (normalized.selector || requestedModel) : null, reasoningEffort: requestedEffort, })) { throw new RpcError(ERROR.INVALID_PARAMS, "The running OMP session rejected the requested model settings"); } await this.store.updateThread(threadId, (stored) => { if (persistentModel) { stored.model = updatedModel; stored.modelProvider = updatedProvider; stored.modelSelector = updatedSelector; } if (params.cwd) stored.cwd = updatedCwd; if (params.approvalPolicy) stored.approvalPolicy = updatedApprovalPolicy; if (requestedEffort) stored.reasoningEffort = updatedEffort; }); thread.model = updatedModel; thread.modelProvider = updatedProvider; thread.modelSelector = updatedSelector; thread.cwd = updatedCwd; thread.approvalPolicy = updatedApprovalPolicy; thread.reasoningEffort = updatedEffort; const clientModel = updatedModel; const threadSettings = { cwd: updatedCwd, approvalPolicy: updatedApprovalPolicy, approvalsReviewer: params.approvalsReviewer || "user", sandboxPolicy: params.sandboxPolicy || { type: "workspaceWrite", writableRoots: [], networkAccess: false, excludeTmpdirEnvVar: false, excludeSlashTmp: false, }, activePermissionProfile: null, model: clientModel, modelProvider: updatedProvider, serviceTier: params.serviceTier || "default", effort: updatedEffort, summary: params.summary || "auto", collaborationMode: params.collaborationMode || { mode: "default", settings: { model: clientModel, reasoning_effort: updatedEffort || null, developer_instructions: null, }, }, multiAgentMode: "explicitRequestOnly", personality: params.personality || null, }; await this.#broadcast(threadId, "thread/settings/updated", { threadId, threadSettings }); return {}; } async #threadFork(peer, params) { const config = await this.configPromise; const sourceId = requireString(params.threadId || params.id, "threadId"); const source = await this.store.getThread(sourceId); if (!source) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${sourceId}`); if (params.deferGoalContinuation && params.ephemeral) { throw new RpcError(ERROR.INVALID_PARAMS, "deferGoalContinuation cannot be combined with ephemeral"); } const sourceRecord = await this.store.getRecord(sourceId); const forkId = id("thread"); const rawCwd = path.resolve(params.cwd || source.cwd); const isolationMode = params.workspaceIsolation || config.workspaceIsolation; const workspace = await createIsolatedWorkspace({ threadId: forkId, sourceCwd: rawCwd, isolationMode }); const effectiveCwd = workspace.cwd; const vcsInfo = await getVcsInfo(effectiveCwd); let forkedTurns = params.excludeTurns ? [] : (Array.isArray(source.turns) ? source.turns : []); if (params.lastTurnId) { const cutoff = forkedTurns.findIndex((t) => t.id === params.lastTurnId); if (cutoff >= 0) { forkedTurns = forkedTurns.slice(0, cutoff + 1); } } const requestedModel = params.model || source.model; const resolvedModel = params.model ? config.resolveModel(params.model) : source.modelSelector || config.resolveModel(source.model); const persistentModel = await this.#persistentThreadModel(sourceRecord?.backend, resolvedModel, source.modelProvider); const fork = await this.store.createThread({ threadId: forkId, cwd: effectiveCwd, model: persistentModel, reasoningEffort: params.reasoningEffort || source.reasoningEffort, ephemeral: Boolean(params.ephemeral), preview: source.preview, source: "fork", forkedFromId: source.id, parentThreadId: source.id, goal: params.deferGoalContinuation && source.goal ? { ...source.goal, threadId: forkId } : null, workspace: workspace.isolated ? workspace : null, gitInfo: vcsInfo, turns: structuredClone(forkedTurns), }); const backend = await this.backendFactoryPromise; const session = typeof backend.forkSession === "function" ? await backend.forkSession(sourceRecord?.backend, { cwd: effectiveCwd, model: resolvedModel, reasoningEffort: params.reasoningEffort }) : await backend.openSession({ cwd: effectiveCwd, model: resolvedModel, reasoningEffort: params.reasoningEffort }); this.sessions.set(fork.id, session); await this.#persistSession(fork.id, session); const projectedFork = await this.#projectThreadModel(peer, fork); return { result: threadStartResponse( projectedFork, { ...params, model: projectedFork.model }, normalizeModelParam(projectedFork.model, projectedFork.modelProvider), ), after: () => this.#broadcast(fork.id, "thread/started", { thread: fork }), }; } async #threadInjectItems(params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); if (this.activeTurns.has(threadId) || this.startingTurns.has(threadId)) { throw new RpcError(ERROR.INVALID_PARAMS, "Cannot inject items while the thread has an active turn"); } const session = await this.#openSession(thread); if (typeof session.injectItems !== "function") { throw new RpcError(ERROR.INVALID_PARAMS, "The active backend cannot inject thread history"); } await session.injectItems(params.items); await this.#persistSession(threadId, session); return {}; } async #threadUnsubscribe(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); peer.subscriptions.delete(threadId); return { status: "unsubscribed" }; } async #threadList(peer, params) { const page = await this.store.listThreads({ cursor: params.cursor, limit: params.limit || params.pageSize, archived: Boolean(params.archived), cwd: params.cwd, }); page.data = await Promise.all(page.data.map((thread) => this.#projectThreadModel(peer, thread))); return page; } async #threadRead(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); if (!this.activeTurns.has(threadId)) { const live = await this.#externalControlStatus(threadId); if (live) { thread.status = live.idle && !live.hasPendingMessages ? { type: "idle" } : { type: "active", activeFlags: [] }; } await this.#settleAbandonedTurns(threadId, thread, live); } peer?.subscriptions?.add(threadId); await this.#seedExternalProjection(threadId); await this.#ensureExternalControlSubscription(threadId); const includeTurns = params.includeTurns !== false; if (!includeTurns) thread.turns = []; return { thread: await this.#projectThreadModel(peer, thread) }; } async #threadTurnsList(params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); return makeTurnsPage(thread.turns, params); } async #threadItemsList(params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); const turnId = params.turnId == null ? null : requireString(params.turnId, "turnId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); if (turnId && !thread.turns.some((candidate) => candidate.id === turnId)) { throw new RpcError(ERROR.NOT_FOUND, `Turn not found: ${turnId}`); } const direction = params.sortDirection === "desc" ? "desc" : "asc"; const sourceTurns = turnId ? thread.turns.filter((turn) => turn.id === turnId) : (direction === "desc" ? thread.turns.slice().reverse() : thread.turns); const entries = sourceTurns.flatMap((turn) => { const items = direction === "desc" ? (turn.items || []).slice().reverse() : (turn.items || []); return items.map((item) => ({ turnId: turn.id, item })); }); return paginateStableEntries(entries, params, "thread-item", (entry) => `${entry.turnId}:${entry.item.id}`); } async #threadMetadataUpdate(_peer, params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); await this.store.updateThread(threadId, (t) => { if (params.gitInfo) t.gitInfo = params.gitInfo; if (params.name) t.name = params.name; }); return {}; } async #threadSubscribe(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); if (!await this.store.getThread(threadId)) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); peer.subscriptions.add(threadId); await this.#seedExternalProjection(threadId); await this.#ensureExternalControlSubscription(threadId); return { subscriptionId: `${peer.id}:${threadId}` }; } async #threadNameSet(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.renameThread(threadId, params.name ?? null); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); return { result: { thread: await this.#projectThreadModel(peer, thread) }, after: () => this.#broadcast(threadId, "thread/name/updated", { threadId, name: thread.name }), }; } async #threadArchive(peer, params, archived) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.archiveThread(threadId, archived); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); return { result: archived ? {} : { thread: await this.#projectThreadModel(peer, thread) }, after: () => this.#broadcast(threadId, archived ? "thread/archived" : "thread/unarchived", { threadId }), }; } async #threadDelete(_peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); const active = this.activeTurns.get(threadId); if (active) { active.controller.abort(abortError("Thread deleted")); await active.session.interrupt?.().catch(() => {}); } const session = this.sessions.get(threadId); await session?.dispose?.().catch(() => {}); this.sessions.delete(threadId); if (thread?.workspace) { await cleanupWorkspace(thread.workspace).catch(() => {}); } const deleted = await this.store.deleteThread(threadId); if (!deleted) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); for (const peer of this.peers) peer.subscriptions.delete(threadId); return {}; } async #threadRollback(peer, params = {}) { const threadId = requireString(params.threadId || params.id, "threadId"); const numTurns = Math.max(1, Number(params.numTurns || 1)); const active = this.activeTurns.get(threadId); if (active) { active.controller.abort(abortError("Thread rolled back")); await active.session.interrupt?.().catch(() => {}); this.activeTurns.delete(threadId); } const session = this.sessions.get(threadId); await session?.dispose?.().catch(() => {}); this.sessions.delete(threadId); const thread = await this.store.rollbackThread(threadId, numTurns); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); return { thread: await this.#projectThreadModel(peer, thread) }; } async #threadCompact(peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); const turnId = id("turn"); const itemId = id("item"); const compactionItem = makeContextCompactionItem(itemId); return { result: {}, after: async () => { const turn = makeTurn({ turnId, items: [compactionItem], status: "inProgress", }); await this.store.addTurn(threadId, turn); await this.#broadcast(threadId, "turn/started", { threadId, turn }); await this.#broadcast(threadId, "item/started", { threadId, turnId, item: compactionItem }); const result = await this.#compactThreadContext(threadId, thread); const completedAt = unixSeconds(); // The stored turn is a clone, so completion has to be written through // the store; a placeholder left `inProgress` keeps every client's // composer waiting on a turn that already ended. const completed = await this.store.updateTurn(threadId, turnId, (stored) => { stored.status = "completed"; stored.completedAt = completedAt; }) || { ...turn, status: "completed", completedAt }; await this.#broadcast(threadId, "item/completed", { threadId, turnId, item: compactionItem }); await this.#broadcast(threadId, "thread/compacted", { threadId, turnId, ...result }); await this.#broadcast(threadId, "turn/completed", { threadId, turn: completed }); }, }; } /** * Compact through whoever owns the session. * * Opening our own session for a thread another agent process is driving would * put two agents on one session file - the fastest way to make a thread * diverge - so a live session compacts itself over its control channel. */ async #compactThreadContext(threadId, thread) { const target = await this.#controlTarget(threadId); if (target.sessionId && await this.#externalControlStatus(threadId)) { const started = target.kind === "prime-rpc" ? await primeControlCompact(target) : await controlCompact(target.sessionId); return started ? { compacted: true } : { compacted: false, error: `The running ${target.kind === "prime-rpc" ? "Prime" : "OMP"} session refused to compact` }; } const session = await this.#openSession(thread); return await session.compact?.().catch((error) => ({ compacted: false, error: error.message })) || { compacted: true }; } async #threadGoalGet(params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); return { goal: thread.goal || null }; } async #threadGoalSet(_peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); const goal = mergeThreadGoal(threadId, thread.goal, params); await this.store.updateThread(threadId, (stored) => { stored.goal = goal; }); await this.#broadcast(threadId, "thread/goal/updated", { threadId, turnId: null, goal }); return { goal }; } async #threadGoalClear(_peer, params) { const threadId = requireString(params.threadId || params.id, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); const cleared = Boolean(thread.goal); if (cleared) { await this.store.updateThread(threadId, (stored) => { stored.goal = null; }); await this.#broadcast(threadId, "thread/goal/cleared", { threadId }); } return { cleared }; } /** * Deliver a turn/start into a thread another OMP process is driving. * * Steering a busy session stays in its current turn. Sending to an idle * session opens a new turn, whose real id must appear in the OMP JSONL * before we answer the client. Never associate the prompt with an older * completed turn: iOS renders that mismatch as a permanently queued message. */ async #steerExternalThread(threadId, thread, params, live) { const { prompt, images } = await extractInputPromptAndImages(params.input ?? params.prompt ?? params.message, thread.cwd); if (!prompt.trim() && images.length === 0) throw new RpcError(ERROR.INVALID_PARAMS, "turn/start requires non-empty input"); if (images.length > 0) return null; const config = await this.configPromise; const requestedModel = params.model || params.config?.model || thread.modelSelector || thread.model; const requestedEffort = reasoningEffortFromParams(params) || thread.reasoningEffort; if (!await this.#controlConfigure(threadId, { model: config.resolveModel(requestedModel), reasoningEffort: requestedEffort, })) { throw new RpcError(ERROR.INVALID_PARAMS, "The running OMP session rejected the requested model settings"); } const before = new Set((thread.turns || []).map((turn) => turn?.id).filter(Boolean)); const currentTurn = (thread.turns || []).findLast((turn) => turn?.status === "inProgress"); if (!await this.#controlDeliver(threadId, prompt, "steer")) return null; if (!live.idle && currentTurn) return { result: { turn: currentTurn } }; const deadline = Date.now() + EXTERNAL_STEER_TURN_TIMEOUT_MS; while (Date.now() < deadline) { await sleep(250); const current = await this.store.getThread(threadId).catch(() => null); const appeared = (current?.turns || []).find((turn) => turn?.id && !before.has(turn.id) && promptMatches(turnPromptText(turn), prompt) ); if (appeared) return { result: { turn: appeared } }; if (!live.idle) { const active = (current?.turns || []).findLast((turn) => turn?.status === "inProgress"); if (active) return { result: { turn: active } }; } } throw new RpcError(ERROR.INTERNAL, "OMP accepted the message but did not publish a matching turn"); } async #turnStart(peer, params) { const threadId = requireString(params.threadId, "threadId"); const thread = await this.store.getThread(threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${threadId}`); if (this.activeTurns.has(threadId) || this.startingTurns.has(threadId)) { throw new RpcError(ERROR.BUSY, `Thread already has an active turn: ${threadId}`); } const live = this.sessions.has(threadId) ? null : await this.#externalControlStatus(threadId); if (live) { const steered = await this.#steerExternalThread(threadId, thread, params, live); if (steered) return steered; } if (thread.status?.type === "active") { throw new RpcError(ERROR.BUSY, `Thread is active in another OMP client: ${threadId}`); } this.startingTurns.add(threadId); try { const { prompt, images } = await extractInputPromptAndImages(params.input ?? params.prompt ?? params.message, thread.cwd); if (!prompt.trim() && images.length === 0) throw new RpcError(ERROR.INVALID_PARAMS, "turn/start requires non-empty input"); const userItem = makeUserMessageItem(id("item"), normalizeUserContent(params.input, prompt)); const turn = makeTurn({ status: "inProgress", items: [userItem] }); const backendPrompt = promptWithThreadGoal(prompt, thread.goal || null); const config = await this.configPromise; const requestedModel = params.model || params.config?.model || thread.modelSelector || thread.model?.selector || thread.model?.id || thread.model; const resolvedTurnModel = config.resolveModel(requestedModel); const resolvedTurnEffort = reasoningEffortFromParams(params) || thread.reasoningEffort || config.defaultReasoningEffort; const session = await this.#openSession(thread, { model: resolvedTurnModel, reasoningEffort: resolvedTurnEffort, approvalPolicy: params.approvalPolicy || thread.approvalPolicy || config.defaultApprovalPolicy, sandboxPolicy: params.sandbox || params.sandboxPolicy, }); const previousTokens = (thread.turns || []).reduce((acc, t) => { const u = t.tokenUsage || {}; return { totalTokens: acc.totalTokens + (u.totalTokens || 0), inputTokens: acc.inputTokens + (u.inputTokens || 0), cachedInputTokens: acc.cachedInputTokens + (u.cachedInputTokens || 0), cacheWriteInputTokens: acc.cacheWriteInputTokens + (u.cacheWriteInputTokens || 0), outputTokens: acc.outputTokens + (u.outputTokens || 0), reasoningOutputTokens: acc.reasoningOutputTokens + (u.reasoningOutputTokens || 0), }; }, makeTokenUsage()); let lastKnownUsage = makeTokenUsage(); for (let index = (thread.turns || []).length - 1; index >= 0; index -= 1) { const usage = thread.turns[index]?.tokenUsage; if (usage?.totalTokens > 0) { lastKnownUsage = usage; break; } } const controller = new AbortController(); const active = { threadId, turnId: turn.id, peer, session, controller, params: { ...params, model: resolvedTurnModel, reasoningEffort: resolvedTurnEffort, approvalPolicy: params.approvalPolicy || config.defaultApprovalPolicy }, items: new Map(), threadCwd: thread.cwd, goalAtStart: thread.goal || null, backendGoalUpdated: false, currentModel: resolvedTurnModel, reportedTurnError: null, itemStartTimes: new Map(), ephemeral: Boolean(thread.ephemeral), startedAt: Date.now(), outputText: "", messageText: "", messageThinking: "", projectedThinkingBlocks: new Set(), reasoningSummaryCount: 0, reasoningSummaryIndex: null, aggregateDiff: "", totalTokenUsage: previousTokens, lastTokenUsage: makeTokenUsage(), turnTokenUsage: makeTokenUsage(), lastKnownUsage: structuredClone(lastKnownUsage), runPromise: null, checkpointTimer: null, checkpointDirty: false, checkpointBytes: 0, }; await this.store.updateThread(threadId, (stored, record) => { stored.turns ||= []; stored.turns.push(structuredClone(turn)); if (!stored.preview) stored.preview = safePreview(prompt); stored.status = { type: "active", activeFlags: [] }; record._activePid = process.pid; record._pendingTurnReconciliation = { turnId: turn.id, prompt: backendPrompt, userPrompt: prompt, parsedSize: record._parsedSize ?? null, }; }); peer.subscriptions.add(threadId); this.activeTurns.set(threadId, active); this.store.markThreadActive?.(threadId); return { result: { turn }, after: async () => { await this.#broadcast(threadId, "thread/status/changed", { threadId, status: { type: "active", activeFlags: [] } }); await this.#broadcast(threadId, "turn/started", { threadId, turn }); active.runPromise = this.#runTurn(active, backendPrompt, images).catch((error) => log("turn runner escaped", error)); }, }; } finally { this.startingTurns.delete(threadId); } } /** * Native identity and backend kind for a thread owned by another agent process. */ async #controlTarget(threadId) { const record = await this.store.getRecord?.(threadId).catch(() => null); const sessionId = record?.backend?.resume?.sessionId || record?.backend?.sessionId || record?.thread?.sessionId || threadId; return { kind: record?.backend?.kind || "omp", external: Boolean(record?.backend?.external), sessionId, activeSessionId: record?.backend?.activeSessionId || null, sessionPath: record?.backend?.resume?.sessionPath || record?.backend?.sessionPath || null, model: record?.thread?.modelSelector || record?.thread?.model || null, reasoningEffort: record?.thread?.reasoningEffort || null, }; } async #externalControlStatus(threadId) { const target = await this.#controlTarget(threadId); const status = target.kind === "prime-rpc" ? await primeControlStatus(target) : await controlStatus(target.sessionId); protocolTrace("external.control_status", { threadId, sessionId: target.sessionId, backendKind: target.kind, reachable: Boolean(status), idle: status?.idle ?? null, hasPendingMessages: status?.hasPendingMessages ?? null, }); return status; } async #controlConfigure(threadId, options) { const target = await this.#controlTarget(threadId); const requestedBackendTag = options?.model ? splitTaggedModel(options.model).tag : null; const currentBackendTag = target.kind === "prime-rpc" ? "prime-agent" : "omp"; if (requestedBackendTag && requestedBackendTag !== currentBackendTag) { throw new RpcError( ERROR.INVALID_PARAMS, `Cannot switch an existing ${currentBackendTag === "omp" ? "OMP" : "Prime Agent"} thread to ${requestedBackendTag === "omp" ? "OMP" : "Prime Agent"}; start a new thread instead` ); } const backend = await this.backendFactoryPromise; const resolvedModel = options?.model && typeof backend?.modelForChild === "function" ? await backend.modelForChild(currentBackendTag, options.model) : (options?.model ? splitTaggedModel(options.model).id : options?.model); const backendOptions = { ...options, model: resolvedModel, }; const configured = target.kind === "prime-rpc" ? await primeControlConfigure(target, backendOptions) : await controlConfigure(target.sessionId, backendOptions); protocolTrace("external.control_configure", { threadId, sessionId: target.sessionId, backendKind: target.kind, model: options?.model || null, reasoningEffort: options?.reasoningEffort || null, configured, }); return configured; } /** * Hand a message to whichever agent process owns an externally driven thread. */ async #controlDeliver(threadId, text, deliverAs) { const target = await this.#controlTarget(threadId); const delivered = target.kind === "prime-rpc" ? await primeControlSend(target, text, { deliverAs }) : await controlSend(target.sessionId, text, { deliverAs }); protocolTrace("external.control_send", { threadId, sessionId: target.sessionId, backendKind: target.kind, deliverAs, delivered, }); return delivered; } async #turnSteer(_peer, params) { const threadId = requireString(params.threadId, "threadId"); const text = extractInputText(params.input ?? params.prompt ?? params.message); if (!text.trim()) throw new RpcError(ERROR.INVALID_PARAMS, "turn/steer requires non-empty input"); const active = this.activeTurns.get(threadId); if (!active) { // The turn belongs to another agent process; use its external control // channel instead of refusing. if (await this.#controlDeliver(threadId, text, "steer")) return {}; throw new RpcError(ERROR.NOT_FOUND, `No active turn for thread: ${threadId}`); } if (params.turnId && params.turnId !== active.turnId) throw new RpcError(ERROR.NOT_FOUND, `Active turn does not match ${params.turnId}`); await active.session.steer(text); return {}; } async #turnInterrupt(_peer, params) { const threadId = requireString(params.threadId, "threadId"); const active = this.activeTurns.get(threadId); if (!active) { const target = await this.#controlTarget(threadId); const interrupted = target.kind === "prime-rpc" ? await primeControlInterrupt(target) : await controlInterrupt(target.sessionId); protocolTrace("external.control_interrupt", { threadId, sessionId: target.sessionId, backendKind: target.kind, interrupted, }); return {}; } if (params.turnId && params.turnId !== active.turnId) { throw new RpcError(ERROR.NOT_FOUND, `Turn not found: ${params.turnId}`); } const reason = abortError("Turn interrupted by client"); active.controller.abort(reason); await active.session.interrupt?.(reason).catch(() => {}); await active.session.dispose?.().catch(() => {}); this.sessions.delete(active.threadId); return {}; } async #runTurn(active, backendPrompt, images = []) { const { threadId, turnId, session, controller } = active; const context = { signal: controller.signal, images, options: { model: active.params.model, reasoningEffort: active.params.reasoningEffort, approvalPolicy: active.params.approvalPolicy, sandboxPolicy: active.params.sandbox || active.params.sandboxPolicy, cwd: active.threadCwd, }, emit: (event) => this.#mapBackendEvent(active, event), requestApproval: (request) => this.#requestApproval(active, request), requestInput: (request) => this.#requestInput(active, request), persistSessionState: async (state) => { await this.store.setBackendState(threadId, state); await this.#seedExternalProjection(threadId); if (state?.kind === "prime-rpc") return; for (const delay of [0, 25, 100, 250, 500, 1_000]) { if (delay) await sleep(delay); if (await this.#ensureExternalControlSubscription(threadId)) break; } }, }; let status = "completed"; let turnError = null; try { const result = await session.runTurn(backendPrompt, context); status = result?.status || (controller.signal.aborted ? "interrupted" : "completed"); if (active.reportedTurnError && status !== "interrupted") { status = "failed"; turnError = active.reportedTurnError; } } catch (error) { if (controller.signal.aborted || error?.name === "AbortError" || error?.code === "ABORT_ERR") { status = "interrupted"; } else { status = "failed"; turnError = active.reportedTurnError || protocolTurnError(error); if (!active.reportedTurnError) { await this.#broadcast(threadId, "error", { threadId, turnId, error: turnError, willRetry: false }); } } } finally { await this.#completeOpenItems(active, status === "completed" ? "completed" : status); this.#clearActiveCheckpoint(active); const thread = await this.store.getThread(threadId); const contextWindow = thread?.modelContextWindow || resolveModelContextWindow(active.params?.model || thread?.model); let completed = null; await this.store.updateThread(threadId, (stored, record) => { const turn = stored.turns?.find((candidate) => candidate.id === turnId); if (turn) { turn.status = status; turn.error = turnError; turn.completedAt = unixSeconds(); if (active.turnTokenUsage?.totalTokens > 0) { turn.tokenUsage = active.turnTokenUsage; } completed = structuredClone(turn); } stored.status = { type: "idle" }; record._activePid = null; }, { clone: false }); await this.#accountThreadGoal(active); await this.#persistSession(threadId, session); if (active.ephemeral && this.sessions.get(threadId) === session) { await session.dispose?.().catch(() => {}); this.sessions.delete(threadId); } this.activeTurns.delete(threadId); this.store.markThreadInactive?.(threadId); if (completed) await this.#broadcast(threadId, "turn/completed", { threadId, turn: completed }); const effectiveLastUsage = (active.lastTokenUsage?.totalTokens > 0 || active.lastTokenUsage?.inputTokens > 0) ? active.lastTokenUsage : (active.lastKnownUsage?.totalTokens > 0 ? active.lastKnownUsage : null); const effectiveTotalUsage = active.totalTokenUsage?.totalTokens > 0 ? active.totalTokenUsage : (effectiveLastUsage || null); if (effectiveLastUsage && effectiveTotalUsage) { await this.#broadcast(threadId, "thread/tokenUsage/updated", { threadId, turnId, tokenUsage: { total: effectiveTotalUsage, last: effectiveLastUsage, modelContextWindow: contextWindow, }, }); } await this.#broadcast(threadId, "thread/status/changed", { threadId, status: { type: "idle" } }); this.externalProjections.delete(threadId); } } async #accountThreadGoal(active) { if (!active.goalAtStart || active.backendGoalUpdated || active.goalAtStart.status !== "active") return; const current = await this.store.getThread(active.threadId); if (!current?.goal || current.goal.createdAt !== active.goalAtStart.createdAt || current.goal.status !== "active") return; const elapsed = Math.max(0, Math.floor((Date.now() - active.startedAt) / 1000)); const used = Math.max(0, Number(active.turnTokenUsage?.totalTokens || 0)); const goal = { ...current.goal, tokensUsed: Math.max(0, Number(current.goal.tokensUsed || 0)) + used, timeUsedSeconds: Math.max(0, Number(current.goal.timeUsedSeconds || 0)) + elapsed, updatedAt: unixSeconds(), }; if (goal.tokenBudget != null && goal.tokensUsed >= goal.tokenBudget) goal.status = "budgetLimited"; await this.store.updateThread(active.threadId, (thread) => { thread.goal = goal; }); await this.#broadcast(active.threadId, "thread/goal/updated", { threadId: active.threadId, turnId: active.turnId, goal, }); } async #markActiveDirty(active, bytes = 0) { active.checkpointDirty = true; active.checkpointBytes += Math.max(0, Number(bytes || 0)); if (active.checkpointBytes >= 256 * 1024) { this.#clearActiveCheckpoint(active); await this.store.flush(); return; } if (active.checkpointTimer) return; active.checkpointTimer = setTimeout(() => { active.checkpointTimer = null; if (!active.checkpointDirty) return; active.checkpointDirty = false; active.checkpointBytes = 0; this.store.flush().catch((error) => log("active turn checkpoint failed", error)); }, 2_000); active.checkpointTimer.unref?.(); } #clearActiveCheckpoint(active) { clearTimeout(active.checkpointTimer); active.checkpointTimer = null; active.checkpointDirty = false; active.checkpointBytes = 0; } async #mapBackendEvent(active, event) { if (!event || typeof event !== "object") return; const { threadId, turnId } = active; switch (event.type) { case "textDelta": { const item = await this.#ensureItem(active, "agent", () => makeAgentMessageItem(id("item"), "")); const delta = String(event.delta || ""); active.outputText += delta; active.messageText = `${active.messageText || ""}${delta}`; await this.store.updateItem(threadId, turnId, item.id, (stored) => { stored.text = `${stored.text || ""}${delta}`; }, { flush: false, clone: false }); await this.#markActiveDirty(active, Buffer.byteLength(delta)); await this.#broadcast(threadId, "item/agentMessage/delta", { threadId, turnId, itemId: item.id, delta }); break; } case "reasoningDelta": { const delta = String(event.delta || ""); active.messageThinking = `${active.messageThinking || ""}${delta}`; await this.#appendReasoningSummaryDelta(active, delta); break; } case "assistantText": { const tail = pendingBlockTail(active.messageText, event.text); active.messageText = `${active.messageText || ""}${tail}`; if (!tail) break; const item = await this.#ensureItem(active, "agent", () => makeAgentMessageItem(id("item"), "")); await this.store.updateItem(threadId, turnId, item.id, (stored) => { stored.text = `${stored.text || ""}${tail}`; }, { flush: false, clone: false }); active.outputText += tail; await this.#markActiveDirty(active, Buffer.byteLength(tail)); await this.#broadcast(threadId, "item/agentMessage/delta", { threadId, turnId, itemId: item.id, delta: tail }); break; } case "assistantThinking": { const complete = String(event.thinking || ""); const normalized = complete.trim(); if (!active.messageThinking && normalized && Array.from(active.projectedThinkingBlocks) .some((projected) => normalized === projected || normalized.startsWith(projected))) { break; } const tail = pendingBlockTail(active.messageThinking, complete); active.messageThinking = `${active.messageThinking || ""}${tail}`; await this.#appendReasoningSummaryDelta(active, tail); break; } case "assistantMessageEnd": { this.#endAssistantMessage(active); break; } case "tokenUsage": { if (!event.usage || typeof event.usage !== "object") break; const usage = makeTokenUsage(event.usage); if (usage.totalTokens > 0 || usage.inputTokens > 0) { active.lastTokenUsage = usage; active.lastKnownUsage = usage; active.turnTokenUsage = addTokenUsage(active.turnTokenUsage, usage); active.totalTokenUsage = { totalTokens: active.totalTokenUsage.totalTokens + usage.totalTokens, inputTokens: active.totalTokenUsage.inputTokens + usage.inputTokens, cachedInputTokens: active.totalTokenUsage.cachedInputTokens + usage.cachedInputTokens, cacheWriteInputTokens: active.totalTokenUsage.cacheWriteInputTokens + usage.cacheWriteInputTokens, outputTokens: active.totalTokenUsage.outputTokens + usage.outputTokens, reasoningOutputTokens: active.totalTokenUsage.reasoningOutputTokens + usage.reasoningOutputTokens, }; const thread = await this.store.getThread(threadId); const contextWindow = event.contextUsage?.contextWindow || thread?.modelContextWindow || resolveModelContextWindow(active.params?.model || thread?.model); await this.#broadcast(threadId, "thread/tokenUsage/updated", { threadId, turnId, tokenUsage: { total: active.totalTokenUsage, last: active.lastTokenUsage, modelContextWindow: contextWindow, }, }); } break; } case "toolStart": { this.#endAssistantMessage(active); await this.#closeOpenItems(active, ["agent"]); const itemId = event.toolCallId || id("item"); let item; if (event.itemType === "webSearch") { item = makeWebSearchItem(itemId, event.args); } else if (event.itemType === "imageView") { const imagePath = event.args?.path || event.args?.filePath || event.args?.file || ""; item = makeImageViewItem(itemId, imagePath && !path.isAbsolute(imagePath) ? path.resolve(active.threadCwd, imagePath) : imagePath); } else if (event.itemType === "mcpToolCall") { item = makeMcpToolCallItem(itemId, event.server || "oh-my-pi", event.toolName, event.args); } else if (event.itemType === "dynamicToolCall") { item = makeDynamicToolCallItem(itemId, event.toolName, event.args, event.namespace); } else { item = makeCommandItem(itemId, event.command || event.args?.command || event.toolName, event.cwd || active.threadCwd || this.defaultCwd); if (Array.isArray(event.commandActions) && event.commandActions.length > 0) { item.commandActions = event.commandActions; } } if (event.toolCallId) active.itemStartTimes.set(event.toolCallId, Date.now()); await this.#startItem(active, item, event.toolCallId || itemId); break; } case "toolDelta": { const itemId = active.items.get(event.toolCallId) || event.toolCallId; if (!itemId) break; const delta = typeof event.delta === "string" ? event.delta : extractToolOutputText(event.delta); if (!delta) break; let itemType = null; await this.store.updateItem(threadId, turnId, itemId, (item) => { itemType = item.type; if (item.type === "commandExecution") { item.aggregatedOutput = `${item.aggregatedOutput || ""}${delta}`; } else if (item.type === "dynamicToolCall") { item.contentItems = [{ type: "inputText", text: `${item.contentItems?.[0]?.text || ""}${delta}` }]; } }, { flush: false, clone: false }); await this.#markActiveDirty(active, Buffer.byteLength(delta)); if (itemType === "commandExecution") { await this.#broadcast(threadId, "item/commandExecution/outputDelta", { threadId, turnId, itemId, delta }); } else if (itemType === "mcpToolCall") { await this.#broadcast(threadId, "item/mcpToolCall/progress", { threadId, turnId, itemId, message: delta }); } break; } case "toolEnd": { // The interactive picker comes from the live `item/tool/requestUserInput` // request, whose id only exists in flight. A stored ask item with // `completed: false` renders a second picker the client can never // submit, so persist the card only once the ask has resolved. if (event.itemType === "userInputResponse" && !active.items.get(event.toolCallId)) { const questions = event.args?.questions || []; const resolvedText = event.outputText || extractToolOutputText(event.output); const item = makeUserInputResponseItem(event.toolCallId || id("item"), { requestId: event.toolCallId, turnId, questions, answers: askAnswersFromResult(questions, event.output?.details, resolvedText), completed: true, }); await this.#startItem(active, item, event.toolCallId); await this.#broadcast(threadId, "item/completed", { threadId, turnId, item, completedAtMs: Date.now() }); break; } const itemId = active.items.get(event.toolCallId) || event.toolCallId; if (!itemId) break; const outputText = event.outputText || extractToolOutputText(event.output); const startTime = active.itemStartTimes.get(itemId) || (event.toolCallId ? active.itemStartTimes.get(event.toolCallId) : null); const durationMs = event.durationMs ?? event.details?.wallTimeMs ?? (startTime ? Date.now() - startTime : null); const updated = await this.#finishItem(active, itemId, event.status || "completed", (item) => { if (item.type === "commandExecution") { item.exitCode = event.exitCode ?? (event.status === "completed" ? 0 : null); item.durationMs = durationMs != null ? Math.max(1, Math.round(durationMs)) : 50; if (outputText) item.aggregatedOutput = outputText; } else if (item.type === "dynamicToolCall") { item.contentItems = dynamicToolContentItems(event.output, outputText); item.success = event.status !== "failed"; item.durationMs = durationMs == null ? null : Math.max(1, Math.round(durationMs)); } else if (item.type === "userInputResponse") { item.answers = askAnswersFromResult(item.questions, event.output?.details, outputText); item.completed = true; } else if (item.type === "mcpToolCall") { item.durationMs = durationMs == null ? null : Math.max(1, Math.round(durationMs)); const details = event.output?.details || {}; const rawContent = details.rawContent || event.output?.rawContent || event.output?.content; item.result = event.status === "failed" ? null : { content: Array.isArray(rawContent) ? rawContent : (outputText ? [{ type: "text", text: outputText }] : []), structuredContent: details.structuredContent || event.output?.structuredContent || event.output?.structured_content || null, _meta: details.mcpMeta || event.output?._meta || null, }; item.error = event.status === "failed" ? { message: event.error?.message || outputText || "Tool failed" } : null; } else if (item.type === "webSearch") { const results = event.output?.results || event.output?.details?.results; if (Array.isArray(results)) item.results = results; } }); if (updated) await this.#broadcast(threadId, "item/completed", { threadId, turnId, item: updated, completedAtMs: Date.now() }); break; } case "compactionStart": { const itemId = event.id || id("item"); await this.#startItem(active, makeContextCompactionItem(itemId), "compaction"); break; } case "compactionEnd": { const itemId = active.items.get("compaction") || event.id; if (itemId) { const item = makeContextCompactionItem(itemId); await this.#finishItem(active, itemId, "completed", () => item); await this.#broadcast(threadId, "item/completed", { threadId, turnId, item, completedAtMs: Date.now() }); await this.#broadcast(threadId, "thread/compacted", { threadId, turnId, compacted: true }); } break; } case "fileStart": { this.#endAssistantMessage(active); await this.#closeOpenItems(active, ["agent"]); const itemId = event.toolCallId || id("item"); const item = makeFileChangeItem(itemId, event.changes || []); await this.#startItem(active, item, event.toolCallId || itemId); const diff = (item.changes || []).map((change) => change.diff).filter(Boolean).join("\n"); if (diff) { active.aggregateDiff += `${active.aggregateDiff ? "\n" : ""}${diff}`; await this.#broadcast(threadId, "turn/diff/updated", { threadId, turnId, diff: active.aggregateDiff }); } break; } case "fileEnd": { const itemId = active.items.get(event.toolCallId) || event.toolCallId; if (!itemId) break; const updated = await this.#finishItem(active, itemId, event.status || "completed", (item) => { if (Array.isArray(event.changes) && event.changes.length) item.changes = makeFileChangeItem(item.id, event.changes).changes; }); if (updated?.changes?.length) { await this.#broadcast(threadId, "item/fileChange/patchUpdated", { threadId, turnId, itemId, changes: updated.changes, }); const diff = updated.changes.map((change) => change.diff).filter(Boolean).join("\n"); if (diff && !active.aggregateDiff.includes(diff)) { active.aggregateDiff += `${active.aggregateDiff ? "\n" : ""}${diff}`; await this.#broadcast(threadId, "turn/diff/updated", { threadId, turnId, diff: active.aggregateDiff }); } } if (updated) await this.#broadcast(threadId, "item/completed", { threadId, turnId, item: updated, completedAtMs: Date.now() }); break; } case "planUpdate": { await this.#broadcast(threadId, "turn/plan/updated", { threadId, turnId, explanation: event.explanation || null, plan: Array.isArray(event.plan) ? event.plan : [], }); break; } case "goalUpdate": { active.backendGoalUpdated = true; if (!event.goal) { await this.store.updateThread(threadId, (thread) => { thread.goal = null; }); await this.#broadcast(threadId, "thread/goal/cleared", { threadId }); break; } const goal = { ...event.goal, threadId }; await this.store.updateThread(threadId, (thread) => { thread.goal = goal; }); await this.#broadcast(threadId, "thread/goal/updated", { threadId, turnId, goal }); break; } case "modelChanged": { await this.#applyBackendModelChange(active, event); break; } case "turnError": { const error = protocolTurnError(event); const willRetry = Boolean(event.willRetry); await this.#broadcast(threadId, "error", { threadId, turnId, error, willRetry }); if (!willRetry && event.terminal !== false) active.reportedTurnError = error; break; } case "notice": { const message = String(event.message || ""); if (event.level === "error") { await this.#broadcast(threadId, "error", { threadId, turnId, error: protocolTurnError(event), willRetry: Boolean(event.willRetry), }); } else if (event.level === "warning" && message) { await this.#broadcast(threadId, "warning", { threadId, message }); } else if (message) { debug("OMP notice", threadId, message); } break; } default: debug("ignored backend event", event); } } async #applyBackendModelChange(active, event) { await this.#applyProjectedModelChange(active.threadId, event, active); } async #applyProjectedModelChange(threadId, event, active = null) { const selector = String(event.model || ""); if (!selector) return; const thread = await this.store.getThread(threadId); if (!thread) return; const record = await this.store.getRecord(threadId); const normalized = await this.#persistentThreadModel(record?.backend, selector, thread.modelProvider); if (!normalized) return; const modelSelector = normalized.selector || normalized.id; if (active) { active.currentModel = modelSelector; active.params.model = modelSelector; } if (thread.model === normalized.id && thread.modelSelector === modelSelector) return; await this.store.updateThread(threadId, (stored) => { stored.model = normalized.id; stored.modelProvider = normalized.provider; stored.modelSelector = modelSelector; }); const backend = await this.backendFactoryPromise; const clientModel = typeof backend?.clientModelForState === "function" ? await backend.clientModelForState(record?.backend, modelSelector) : normalized.id; const effort = thread.reasoningEffort || active?.params?.reasoningEffort || "medium"; await this.#broadcast(threadId, "thread/settings/updated", { threadId, threadSettings: { cwd: thread.cwd, approvalPolicy: thread.approvalPolicy || "on-request", approvalsReviewer: "user", sandboxPolicy: { type: "workspaceWrite", writableRoots: [], networkAccess: false, excludeTmpdirEnvVar: false, excludeSlashTmp: false, }, activePermissionProfile: null, model: clientModel, modelProvider: normalized.provider, serviceTier: "default", effort, summary: "auto", collaborationMode: { mode: "default", settings: { model: clientModel, reasoning_effort: effort, developer_instructions: null, }, }, multiAgentMode: "explicitRequestOnly", personality: null, }, }); } #endAssistantMessage(active) { const thinking = String(active.messageThinking || "").trim(); if (thinking) active.projectedThinkingBlocks.add(thinking); active.messageText = ""; active.messageThinking = ""; active.reasoningSummaryIndex = null; } async #appendReasoningSummaryDelta(active, delta) { const value = String(delta || ""); if (!value) return; const { threadId, turnId } = active; const item = await this.#ensureItem(active, "reasoning", () => makeReasoningItem(id("item"), [], [])); if (active.reasoningSummaryIndex == null) { active.reasoningSummaryIndex = active.reasoningSummaryCount; active.reasoningSummaryCount += 1; if (active.reasoningSummaryIndex > 0) { await this.#broadcast(threadId, "item/reasoning/summaryPartAdded", { threadId, turnId, itemId: item.id, summaryIndex: active.reasoningSummaryIndex, }); } } const summaryIndex = active.reasoningSummaryIndex; await this.store.updateItem(threadId, turnId, item.id, (stored) => { stored.summary ||= []; while (stored.summary.length <= summaryIndex) stored.summary.push(""); stored.summary[summaryIndex] = `${stored.summary[summaryIndex] || ""}${value}`; stored.content = []; }, { flush: false, clone: false }); await this.#markActiveDirty(active, Buffer.byteLength(value)); await this.#broadcast(threadId, "item/reasoning/summaryTextDelta", { threadId, turnId, itemId: item.id, delta: value, summaryIndex, }); } async #closeOpenItems(active, keys) { const threadId = active.threadId; const turnId = active.turnId; for (const key of keys) { const openId = active.items.get(key); if (openId) { const completed = await this.#finishItem(active, openId, "completed"); if (completed) { await this.#broadcast(threadId, "item/completed", { threadId, turnId, item: completed, completedAtMs: Date.now() }); } active.items.delete(key); } } } async #startItem(active, item, trackingKey = null) { const threadId = active.threadId; const turnId = active.turnId; if (trackingKey) active.items.set(trackingKey, item.id); active.itemStartTimes.set(item.id, Date.now()); await this.store.addItem(threadId, turnId, item, { flush: false, clone: false }); await this.#markActiveDirty(active, Buffer.byteLength(JSON.stringify(item))); await this.#broadcast(threadId, "item/started", { threadId, turnId, item, startedAtMs: Date.now() }); return item; } async #ensureItem(active, key, create) { const existingId = active.items.get(key); if (existingId) return { id: existingId }; return this.#startItem(active, create(), key); } async #finishItem(active, itemId, status, mutate) { const updated = await this.store.updateItem(active.threadId, active.turnId, itemId, (item) => { if (mutate) mutate(item); if (item.type === "reasoning") { item.summary = (Array.isArray(item.summary) ? item.summary : []) .map((part) => String(part || "").trim()) .filter(Boolean); item.content = []; } const normalizedStatus = normalizeItemStatus(status); if (["mcpToolCall", "dynamicToolCall", "collabAgentToolCall"].includes(item.type)) { item.status = normalizedStatus === "completed" ? "completed" : "failed"; } else if (["webSearch", "imageView", "contextCompaction", "plan"].includes(item.type)) { delete item.status; } else { item.status = normalizedStatus; } }, { flush: false }); if (updated) await this.#markActiveDirty(active, Buffer.byteLength(JSON.stringify(updated))); return updated; } async #completeOpenItems(active, status) { const record = await this.store.getRecord(active.threadId); const turn = record?.thread?.turns?.find((candidate) => candidate.id === active.turnId); for (const item of turn?.items || []) { if (!item.status || item.status === "inProgress") { const completed = await this.#finishItem(active, item.id, status); if (completed && item.type !== "userMessage") { await this.#broadcast(active.threadId, "item/completed", { threadId: active.threadId, turnId: active.turnId, item: completed, completedAtMs: Date.now() }); } } else if (["agentMessage", "reasoning"].includes(item.type) && !item.__completionSent) { const completed = await this.#finishItem(active, item.id, "completed"); if (completed) await this.#broadcast(active.threadId, "item/completed", { threadId: active.threadId, turnId: active.turnId, item: completed, completedAtMs: Date.now() }); } } } async #requestApproval(active, request = {}) { const policy = String(active.params.approvalPolicy || "on-request").toLowerCase(); if (["never", "allow", "auto"].includes(policy)) return "accept"; if (["deny", "forbidden"].includes(policy)) return "decline"; const normalized = normalizeApprovalRequest(active, request); const cacheKey = approvalCacheKey(normalized); const cached = await this.store.getApproval(active.threadId, cacheKey); if (cached === "acceptForSession") return cached; if (active.peer.client.closed) return "decline"; const method = normalized.kind === "file" ? "item/fileChange/requestApproval" : "item/commandExecution/requestApproval"; const result = await active.peer.client.requestApproval(method, normalized.params, { signal: active.controller.signal }); const decision = normalizeApprovalDecision(result); if (decision === "acceptForSession") await this.store.rememberApproval(active.threadId, cacheKey, decision); return decision; } async #requestInput(active, request = {}) { if (active.peer.client.closed) return undefined; const questions = normalizeUserInputQuestions(request); let serverRequestId = null; let result; try { result = await active.peer.client.request("item/tool/requestUserInput", { threadId: active.threadId, turnId: active.turnId, itemId: String(request.itemId || request.toolCallId || id("input")), questions, isBlocking: true, autoResolutionMs: null, }, { signal: active.controller.signal, onRequestId: (requestId) => { serverRequestId = requestId; }, }); } finally { if (serverRequestId) { active.peer.client.notify("serverRequest/resolved", { threadId: active.threadId, requestId: serverRequestId, }); } } if (typeof result === "string") return result; if (result?.answers && typeof result.answers === "object") { const first = Object.values(result.answers)[0]; if (Array.isArray(first?.answers) && first.answers.length) { return String(first.answers[0]).replace(/\s+\(Recommended\)$/i, ""); } if (typeof first === "string") return first; } return result?.answer ?? result?.value ?? null; } async #projectThreadModel(peer, thread) { if (!thread) return thread; const backend = await this.backendFactoryPromise; if (typeof backend?.clientModelForState !== "function") return thread; const record = await this.store.getRecord(thread.id).catch(() => null); const raw = thread.modelSelector || thread.model; const model = await backend.clientModelForState(record?.backend, raw, { canonical: clientNeedsCanonicalModelIds(peer), }); if (!model || (thread.model === model && thread.modelSelector === model)) return thread; return { ...thread, model, modelSelector: model }; } async #projectNotificationForPeer(peer, threadId, method, params) { if (method === "thread/started" && params?.thread) { return { ...params, thread: await this.#projectThreadModel(peer, params.thread) }; } if (method !== "thread/settings/updated" || !params?.threadSettings?.model) return params; const thread = await this.store.getThread(threadId).catch(() => null); const projected = await this.#projectThreadModel(peer, thread); if (!projected?.model) return params; const collaborationMode = params.threadSettings.collaborationMode; return { ...params, threadSettings: { ...params.threadSettings, model: projected.model, ...(collaborationMode?.settings ? { collaborationMode: { ...collaborationMode, settings: { ...collaborationMode.settings, model: projected.model }, }, } : {}), }, }; } async #persistentThreadModel(backendState, model, provider = null) { const backend = await this.backendFactoryPromise; const persistent = typeof backend?.modelForPersistentState === "function" ? await backend.modelForPersistentState(backendState, model) : model; return normalizeThreadModel(persistent, persistent, provider); } async #openSession(thread, options = {}) { if (this.sessions.has(thread.id)) return this.sessions.get(thread.id); const backend = await this.backendFactoryPromise; const record = await this.store.getRecord(thread.id); let resume = record?.backend?.resume || (record?.backend?.sessionId ? { sessionId: record.backend.sessionId, sessionPath: record.backend.sessionPath } : null); if (!resume && (thread.turns?.length > 0 || record?.thread?.turns?.length > 0)) { try { const sessionDir = await findSessionDirForCwd(thread.cwd); const entries = await fs.readdir(sessionDir, { withFileTypes: true }).catch(() => []); const jsonlFiles = entries.filter((e) => e.isFile() && e.name.endsWith(".jsonl")); let best = null; for (const entry of jsonlFiles) { const filePath = path.join(sessionDir, entry.name); const stat = await fs.stat(filePath).catch(() => null); if (!stat) continue; if (!best || stat.mtimeMs > best.mtimeMs) { best = { filePath, mtimeMs: stat.mtimeMs }; } } if (best) { const header = await parseOmpSessionHeader(best.filePath).catch(() => null); const sessionId = header?.session?.id || header?.sessionId || path.basename(best.filePath, ".jsonl").split("_").pop(); resume = { sessionId, sessionPath: best.filePath }; await this.store.setBackendState(thread.id, { kind: "rpc", sessionId, sessionPath: best.filePath, resume }); } } catch {} } const session = await backend.openSession({ cwd: thread.cwd, model: options.model || thread.modelSelector || thread.model, reasoningEffort: options.reasoningEffort, approvalPolicy: options.approvalPolicy, sandboxPolicy: options.sandboxPolicy, resume: resume || null, backendState: record?.backend || null, threadId: thread.id, }); this.sessions.set(thread.id, session); return session; } async #persistSession(threadId, session) { const state = await session.getPersistentState?.(); if (state && typeof state === "object") await this.store.setBackendState(threadId, state); } async #broadcast(threadId, method, params) { const tasks = []; const isGlobalOrTurnLifecycle = method === "turn/completed" || method === "turn/started" || method.startsWith("thread/") || this.peers.size <= 1; for (const peer of this.peers) { if (!peer.initialized || peer.disabledNotifications.has(method)) continue; if (!peer.subscriptions.has(threadId) && !isGlobalOrTurnLifecycle) continue; const peerParams = await this.#projectNotificationForPeer(peer, threadId, method, params); tasks.push(Promise.resolve(peer.client.notify(method, peerParams))); } await Promise.allSettled(tasks); } async #resolveFsPath(params, { write = false } = {}) { const requested = requireString(params.path || params.filePath, "path"); let root = params.cwd || this.defaultCwd; if (params.threadId) { const thread = await this.store.getThread(params.threadId); if (!thread) throw new RpcError(ERROR.NOT_FOUND, `Thread not found: ${params.threadId}`); root = thread.cwd; } const candidate = path.isAbsolute(requested) ? requested : path.join(root, requested); if (process.env.OMP_CODEX_ALLOW_FS_OUTSIDE_WORKSPACE !== "1") { const [canonicalRoot, canonicalCandidate] = await Promise.all([canonicalizeForPolicy(root), canonicalizeForPolicy(candidate)]); if (!isInside(canonicalRoot, canonicalCandidate)) { throw new RpcError(ERROR.POLICY, `${write ? "Write" : "Access"} outside workspace is blocked`, { root: canonicalRoot, path: canonicalCandidate }); } } return path.resolve(candidate); } async #fsGetMetadata(params) { const target = path.resolve(requireString(params.path, "path")); try { const stat = await fs.lstat(target); return { isDirectory: stat.isDirectory(), isFile: stat.isFile(), isSymlink: stat.isSymbolicLink(), createdAtMs: Math.round(stat.birthtimeMs || stat.ctimeMs || 0), modifiedAtMs: Math.round(stat.mtimeMs || 0), }; } catch (err) { if (err.code === "ENOENT") throw new RpcError(ERROR.NOT_FOUND, `Path not found: ${target}`); throw new RpcError(ERROR.INTERNAL, `fs/getMetadata failed: ${err.message}`); } } async #fsReadDirectory(params) { const directory = path.resolve(requireString(params.path, "path")); try { const entries = await fs.readdir(directory, { withFileTypes: true }); return { entries: entries.map((entry) => ({ fileName: entry.name, isDirectory: entry.isDirectory(), isFile: entry.isFile(), })), }; } catch (err) { if (err.code === "ENOENT") throw new RpcError(ERROR.NOT_FOUND, `Directory not found: ${directory}`); throw new RpcError(ERROR.INTERNAL, `fs/readDirectory failed: ${err.message}`); } } async #fsRead(params) { const file = path.resolve(requireString(params.path || params.filePath, "path")); try { const buffer = await fs.readFile(file); const dataBase64 = buffer.toString("base64"); return { data: dataBase64, dataBase64, path: file, content: dataBase64, encoding: "base64" }; } catch (err) { if (err.code === "ENOENT") throw new RpcError(ERROR.NOT_FOUND, `File not found: ${file}`); throw new RpcError(ERROR.INTERNAL, `fs/readFile failed: ${err.message}`); } } async #fsWrite(peer, params) { const file = path.resolve(requireString(params.path || params.filePath, "path")); const rawData = params.dataBase64 || params.data; const data = rawData ? Buffer.from(rawData, "base64") : (params.encoding === "base64" ? Buffer.from(params.content || "", "base64") : Buffer.from(params.content ?? "")); await fs.mkdir(path.dirname(file), { recursive: true }); await fs.writeFile(file, data); return {}; } async #fsCreateDirectory(peer, params) { const directory = path.resolve(requireString(params.path, "path")); await fs.mkdir(directory, { recursive: params.recursive !== false }); return {}; } async #fsRemove(peer, params) { const target = path.resolve(requireString(params.path, "path")); await fs.rm(target, { recursive: Boolean(params.recursive), force: true }); return {}; } async #fsCopy(peer, params) { const source = path.resolve(requireString(params.sourcePath, "sourcePath")); const destination = path.resolve(requireString(params.destinationPath, "destinationPath")); await fs.cp(source, destination, { recursive: Boolean(params.recursive) }); return {}; } async #fsWatch(params) { const watchId = params.watchId || id("watch"); return { watchId }; } async #fsUnwatch() { return {}; } async #standaloneApproval(peer, request) { if (process.env.OMP_CODEX_STANDALONE_APPROVAL === "allow") return "accept"; const cmdStr = Array.isArray(request.command) ? request.command.join(" ") : String(request.command || ""); if (/^\s*(git\s+(rev-parse|status|config|remote|branch|log|describe|check-ref-format)|which|command\s+-v|node\s+-v|npm\s+-v|bun\s+-v|rustc\s+-v|cargo\s+-v|echo|cat|pwd|test)\b/.test(cmdStr)) { return "accept"; } const method = request.kind === "file" ? "item/fileChange/requestApproval" : "item/commandExecution/requestApproval"; return peer.client.requestApproval(method, { threadId: request.threadId || "thread_standalone", turnId: request.turnId || "turn_standalone", itemId: id("standalone"), reason: request.reason, command: request.command, cwd: request.cwd, changes: request.changes, startedAtMs: Date.now(), }); } async #commandExec(peer, params) { const cwd = path.resolve(params.cwd || this.defaultCwd); const command = params.command || params.argv; if (!command) throw new RpcError(ERROR.INVALID_PARAMS, "command/exec requires command or argv"); const decision = await this.#standaloneApproval(peer, { kind: "command", command, cwd, reason: params.reason || "ChatGPT requested command execution" }); if (!decision.startsWith("accept")) throw new RpcError(ERROR.POLICY, "Command execution declined"); const result = await executeCommand(command, { cwd, env: params.env, timeoutMs: params.timeoutMs, maxOutputBytes: params.maxOutputBytes, }); return { ...result, output: `${result.stdout}${result.stderr}` }; } } function threadStartResponse(thread, params, model) { const rawModel = model?.id || (typeof thread.model === "string" ? thread.model : thread.model?.id) || params.model || model?.selector || thread.modelSelector || (typeof model === "string" ? model : null) || "gemini-3.7-flash"; const modelId = codexModelId(rawModel, thread.modelSelector); const provider = codexFacingProvider(params.modelProvider || model?.provider || thread.modelProvider || (typeof rawModel === "string" && rawModel.includes("/") ? rawModel.split("/")[0] : "openai")); const approvalPolicy = params.approvalPolicy || thread.approvalPolicy || "on-request"; const reasoningEffort = reasoningEffortFromParams(params) || thread.reasoningEffort || "medium"; const serviceTier = params.serviceTier || "default"; let sandbox = { type: "workspaceWrite", writableRoots: [], networkAccess: false, excludeTmpdirEnvVar: false, excludeSlashTmp: false, }; if (params.sandbox === "danger-full-access" || params.sandbox === "dangerFullAccess") { sandbox = { type: "dangerFullAccess" }; } else if (params.sandbox === "read-only" || params.sandbox === "readOnly") { sandbox = { type: "readOnly", networkAccess: false }; } else if (typeof params.sandbox === "object" && params.sandbox !== null) { sandbox = params.sandbox; } const responseThread = { ...thread, model: codexModelId(thread.model, thread.modelSelector), ...(params.excludeTurns ? { turns: [] } : {}), }; const initialTurnsPage = params.initialTurnsPage ? makeTurnsPage(thread.turns, params.initialTurnsPage) : null; return { thread: responseThread, model: modelId, modelProvider: provider, serviceTier, cwd: thread.cwd, runtimeWorkspaceRoots: Array.from(new Set([thread.cwd, thread.workspace?.sourceCwd, thread.workspace?.root].filter(Boolean))), instructionSources: [], approvalPolicy, approvalsReviewer: params.approvalsReviewer || "user", sandbox, activePermissionProfile: null, reasoningEffort, multiAgentMode: "explicitRequestOnly", ...(initialTurnsPage ? { initialTurnsPage } : {}), }; } function makeTurnsPage(turns = [], params = {}) { const direction = params.sortDirection === "asc" ? "asc" : "desc"; const ordered = direction === "desc" ? turns.slice().reverse() : turns.slice(); const page = paginateStableEntries(ordered, params, "thread-turn", (turn) => turn.id); const itemsView = params.itemsView || "summary"; const data = page.data.map((turn) => { if (itemsView === "full") return { ...turn, itemsView: "full" }; if (itemsView === "summary") { const user = (turn.items || []).find((item) => item.type === "userMessage"); const agent = (turn.items || []).findLast((item) => item.type === "agentMessage"); return { ...turn, items: [user, agent].filter(Boolean), itemsView: "summary" }; } return { ...turn, items: [], itemsView: "notLoaded" }; }); return { ...page, data }; } function paginateStableEntries(entries, params, kind, identity) { const limit = Math.max(1, Math.min(500, Number(params.limit || 100))); let start = 0; let end = Math.min(entries.length, limit); if (params.cursor) { const cursor = decodeCursor(params.cursor, null); if (!cursor || cursor.kind !== kind || typeof cursor.anchor !== "string" || !["after", "inclusive"].includes(cursor.edge)) { throw new RpcError(ERROR.INVALID_PARAMS, `Invalid ${kind} cursor`); } const anchorIndex = entries.findIndex((entry) => identity(entry) === cursor.anchor); if (anchorIndex < 0) throw new RpcError(ERROR.INVALID_PARAMS, `Expired ${kind} cursor`); start = cursor.edge === "after" ? anchorIndex + 1 : anchorIndex; end = Math.min(entries.length, start + limit); } const data = entries.slice(start, end); const nextCursor = end < entries.length && data.length ? encodeCursor({ kind, edge: "after", anchor: identity(data.at(-1)) }) : null; const backwardsCursor = data.length ? encodeCursor({ kind, edge: "inclusive", anchor: identity(data[0]) }) : null; return { data, nextCursor, backwardsCursor }; } function addTokenUsage(left = {}, right = {}) { return { totalTokens: Number(left.totalTokens || 0) + Number(right.totalTokens || 0), inputTokens: Number(left.inputTokens || 0) + Number(right.inputTokens || 0), cachedInputTokens: Number(left.cachedInputTokens || 0) + Number(right.cachedInputTokens || 0), cacheWriteInputTokens: Number(left.cacheWriteInputTokens || 0) + Number(right.cacheWriteInputTokens || 0), outputTokens: Number(left.outputTokens || 0) + Number(right.outputTokens || 0), reasoningOutputTokens: Number(left.reasoningOutputTokens || 0) + Number(right.reasoningOutputTokens || 0), }; } function dynamicToolContentItems(output, fallbackText = "") { const content = Array.isArray(output?.content) ? output.content : []; const items = content.flatMap((part) => { if (typeof part === "string") return [{ type: "inputText", text: part }]; if (!part || typeof part !== "object") return []; if (part.type === "text" || part.type === "inputText") { return [{ type: "inputText", text: String(part.text || "") }]; } if (part.type === "image" || part.type === "inputImage") { const imageUrl = part.imageUrl || part.image_url || part.url || (part.data && part.mimeType ? `data:${part.mimeType};base64,${part.data}` : null); return imageUrl ? [{ type: "inputImage", imageUrl }] : []; } if (part.type === "audio" || part.type === "inputAudio") { const audioUrl = part.audioUrl || part.audio_url || part.url || (part.data && part.mimeType ? `data:${part.mimeType};base64,${part.data}` : null); return audioUrl ? [{ type: "inputAudio", audioUrl }] : []; } return []; }); return items.length ? items : (fallbackText ? [{ type: "inputText", text: fallbackText }] : []); } function normalizeGoalStatus(status) { const value = String(status || "active").replace(/[-_](.)/g, (_match, char) => char.toUpperCase()); const allowed = new Set(["active", "paused", "blocked", "usageLimited", "budgetLimited", "complete"]); if (!allowed.has(value)) throw new RpcError(ERROR.INVALID_PARAMS, `Unsupported goal status: ${status}`); return value; } function mergeThreadGoal(threadId, current, params) { const now = unixSeconds(); const objective = params.objective == null ? current?.objective : requireString(params.objective, "objective"); if (!objective) throw new RpcError(ERROR.INVALID_PARAMS, "objective is required when creating a goal"); let tokenBudget = current?.tokenBudget ?? null; if (Object.prototype.hasOwnProperty.call(params, "tokenBudget")) { tokenBudget = params.tokenBudget == null ? null : Number(params.tokenBudget); if (tokenBudget != null && (!Number.isSafeInteger(tokenBudget) || tokenBudget <= 0)) { throw new RpcError(ERROR.INVALID_PARAMS, "tokenBudget must be a positive integer or null"); } } const goal = { threadId, objective, status: normalizeGoalStatus(params.status || current?.status || "active"), tokenBudget, tokensUsed: Math.max(0, Number(current?.tokensUsed || 0)), timeUsedSeconds: Math.max(0, Number(current?.timeUsedSeconds || 0)), createdAt: Number(current?.createdAt || now), updatedAt: now, }; if (goal.status === "active" && goal.tokenBudget != null && goal.tokensUsed >= goal.tokenBudget) { goal.status = "budgetLimited"; } return goal; } function promptWithThreadGoal(prompt, goal) { if (!goal || goal.status !== "active" || /^\s*\/goal\b/i.test(prompt)) return prompt; const budget = goal.tokenBudget == null ? "unlimited" : String(goal.tokenBudget); return ` ${goal.objective} Continue working toward this persisted thread goal while directly handling the user's message below. ${prompt}`; } const KNOWN_MODEL_PROVIDERS = new Set([ "anthropic", "google-antigravity", "kimi-code", "laguna-local", "nanbeige-local", "openai", "openai-codex", "openrouter", ]); // a backend that ships its own short catalog would lose every row to the featured // filter, so it has to say so itself, we never guess it from a provider or id prefix function backendCatalogIsUnfiltered(backend) { return backend?.modelCatalogPolicy === "all" || backend?.constructor?.modelCatalogPolicy === "all"; } function catalogPatternMatches(value, pattern) { const source = String(pattern || "") .replace(/[.+?^${}()|[\]\\]/g, "\\$&") .replace(/\*/g, ".*"); return new RegExp(`^${source}$`, "i").test(String(value || "")); } function catalogRowAllowed(policy = {}, model, id) { const include = Array.isArray(policy.include) ? policy.include : []; const exclude = Array.isArray(policy.exclude) ? policy.exclude : []; const excludeProviders = new Set(Array.isArray(policy.excludeProviders) ? policy.excludeProviders.map(String) : []); const provider = String(model?.raw?.provider || model?.provider || ""); if (excludeProviders.has(provider)) return false; if (include.length > 0 && !include.some((pattern) => catalogPatternMatches(id, pattern))) return false; return !exclude.some((pattern) => catalogPatternMatches(id, pattern)); } function configuredCatalogRank(policy = {}, model) { const backendOrder = Array.isArray(policy.backendOrder) ? policy.backendOrder : []; const order = Array.isArray(policy.order) ? policy.order : []; const backendIndex = backendOrder.indexOf(model?.backendTag); const id = String(model?.catalogFilterId || model?.id || model?.model || ""); const modelIndex = order.findIndex((pattern) => catalogPatternMatches(id, pattern)); return { backend: backendIndex === -1 ? backendOrder.length : backendIndex, model: modelIndex === -1 ? order.length : modelIndex, }; } function normalizeModelParam(model, provider) { if (!model) return null; const declaredProvider = typeof model === "object" ? (model.provider || model.modelProvider || provider) : provider; const selector = typeof model === "object" ? (model.selector || model.id || model.model) : String(model); const separator = selector?.indexOf("/") ?? -1; const prefix = separator > 0 ? selector.slice(0, separator) : null; const hasProvider = Boolean(prefix && (prefix === declaredProvider || KNOWN_MODEL_PROVIDERS.has(prefix))); const cleanId = hasProvider ? selector.slice(separator + 1) : selector; const cleanProvider = hasProvider ? prefix : (declaredProvider || "openai"); return { id: cleanId, provider: cleanProvider, ...(hasProvider ? { selector } : {}), }; } function normalizeThreadModel(_clientModel, backendModel, provider) { const normalized = normalizeModelParam(backendModel, provider); if (!normalized) return null; return { ...normalized, id: normalized.id, provider: codexFacingProvider(normalized.provider) }; } function codexModelId(model, selector = null) { if (model && typeof model === "object") { return codexModelId(model.id || model.model || model.selector, model.selector || selector); } const value = String(model || "gemini-3.7-flash"); if (selector && value === selector && value.includes("/")) return value.slice(value.indexOf("/") + 1); return value; } function reasoningEffortFromParams(params = {}) { return params.reasoningEffort || params.effort || params.config?.model_reasoning_effort || params.config?.reasoningEffort || null; } function codexFacingProvider(provider) { const value = String(provider || "openai"); return value === "openai-codex" ? "openai" : value; } function requireString(value, name) { if (typeof value !== "string" || !value.trim()) throw new RpcError(ERROR.INVALID_PARAMS, `${name} must be a non-empty string`); return value; } function normalizeUserContent(input, fallback) { if (Array.isArray(input)) return input; return [{ type: "text", text: fallback }]; } function formatReasoningEfforts(efforts) { const list = Array.isArray(efforts) ? efforts : ["low", "medium", "high"]; return list.map((item) => { if (typeof item === "object" && item !== null && item.reasoningEffort) return item; const effort = String(item); const description = effort === "low" ? "Fast responses with lighter reasoning" : effort === "medium" ? "Balances speed and reasoning depth for everyday tasks" : effort === "high" ? "Greater reasoning depth for complex problems" : effort === "xhigh" ? "Extra high reasoning depth for complex problems" : effort === "max" ? "Maximum reasoning depth for the hardest problems" : `${effort} reasoning`; return { reasoningEffort: effort, description }; }); } function isPlainObject(value) { return typeof value === "object" && value !== null && !Array.isArray(value); } const DYNAMIC_CONFIG_MAPS = new Set([ "desktop", "marketplaces", "mcp_servers", "model_providers", "plugins", "profiles", "projects", ]); const MARKETPLACE_ENTRY_DEFAULTS = { last_updated: null, last_revision: null, source_type: "local", source: "", ref: null, sparse_paths: null, }; const MCP_SERVER_ENTRY_DEFAULTS = { command: "", args: [], environment_id: "local", enabled: true, tool_timeout_sec: null, }; function mergeTypedSection(template, override) { const merged = { ...template }; if (!isPlainObject(override)) return merged; for (const [key, value] of Object.entries(override)) { merged[key] = isPlainObject(template[key]) && isPlainObject(value) ? mergeTypedSection(template[key], value) : value; } return merged; } function mergeEntryMap(template, override, entryDefaults) { const merged = { ...template }; if (!isPlainObject(override)) return merged; for (const [key, value] of Object.entries(override)) { if (!isPlainObject(value)) { merged[key] = value; continue; } merged[key] = entryDefaults ? { ...entryDefaults, ...merged[key], ...value } : { ...merged[key], ...value }; } return merged; } export function mergeCodexConfig(template, overrides) { const merged = { ...template }; if (!isPlainObject(overrides)) return merged; for (const [key, value] of Object.entries(overrides)) { if (!(key in template)) continue; if (key === "marketplaces") { merged[key] = mergeEntryMap(template[key], value, MARKETPLACE_ENTRY_DEFAULTS); } else if (key === "mcp_servers") { merged[key] = mergeEntryMap(template[key], value, MCP_SERVER_ENTRY_DEFAULTS); } else if (DYNAMIC_CONFIG_MAPS.has(key)) { merged[key] = mergeEntryMap(template[key], value, null); } else if (isPlainObject(template[key]) && isPlainObject(value)) { merged[key] = mergeTypedSection(template[key], value); } else { merged[key] = value; } } return merged; } export function resolveConfiguredProvider(provider, providers) { if (typeof provider !== "string" || !provider) return null; const entry = providers?.[provider]; return isPlainObject(entry) && Object.keys(entry).length > 0 ? provider : null; } function defaultTuiKeymap() { return { global: { open_transcript: null, open_external_editor: null, copy: null, clear_terminal: null, submit: null, queue: null, toggle_shortcuts: null, toggle_vim_mode: null, toggle_fast_mode: null, toggle_raw_output: null, }, chat: { interrupt_turn: null, decrease_reasoning_effort: null, increase_reasoning_effort: null, edit_queued_message: null, }, composer: { submit: null, queue: null, toggle_shortcuts: null, history_search_previous: null, history_search_next: null, }, editor: { insert_newline: null, move_left: null, move_right: null, move_up: null, move_down: null, move_word_left: null, move_word_right: null, move_line_start: null, move_line_end: null, delete_backward: null, delete_forward: null, delete_backward_word: null, delete_forward_word: null, kill_line_start: null, kill_whole_line: null, kill_line_end: null, yank: null, }, vim_normal: { enter_insert: null, append_after_cursor: null, append_line_end: null, insert_line_start: null, open_line_below: null, open_line_above: null, move_left: null, move_right: null, move_up: null, move_down: null, move_word_forward: null, move_word_backward: null, move_word_end: null, move_line_start: null, move_line_end: null, delete_char: null, substitute_char: null, delete_to_line_end: null, change_to_line_end: null, yank_line: null, paste_after: null, start_delete_operator: null, start_yank_operator: null, start_change_operator: null, cancel_operator: null, }, vim_operator: { delete_line: null, yank_line: null, motion_left: null, motion_right: null, motion_up: null, motion_down: null, motion_word_forward: null, motion_word_backward: null, motion_word_end: null, motion_line_start: null, motion_line_end: null, select_inner_text_object: null, select_around_text_object: null, cancel: null, }, vim_text_object: { word: null, big_word: null, parentheses: null, brackets: null, braces: null, double_quote: null, single_quote: null, backtick: null, cancel: null, }, pager: { scroll_up: null, scroll_down: null, page_up: null, page_down: null, half_page_up: null, half_page_down: null, jump_top: null, jump_bottom: null, close: null, close_transcript: null, }, list: { move_up: null, move_down: null, move_left: null, move_right: null, page_up: null, page_down: null, jump_top: null, jump_bottom: null, accept: null, cancel: null, }, approval: { open_fullscreen: null, open_thread: null, approve: null, approve_for_session: null, approve_for_prefix: null, deny: null, decline: null, cancel: null, }, }; } function defaultCodexConfig(model = "gemini-3.7-flash", approvalPolicy = "never", reasoningEffort = "high") { return { model, review_model: null, model_context_window: null, model_auto_compact_token_limit: null, model_auto_compact_token_limit_scope: null, model_provider: null, approvals_reviewer: "user", sandbox_mode: "danger-full-access", sandbox_workspace_write: null, forced_chatgpt_workspace_id: null, forced_login_method: null, web_search: null, tools: null, instructions: null, developer_instructions: null, compact_prompt: null, model_reasoning_effort: reasoningEffort, model_reasoning_summary: null, model_verbosity: null, service_tier: "default", analytics: null, apps: null, desktop: { useFontSmoothing: true, "dock-icon-preference": "codex-system", keepRemoteControlAwakeWhilePluggedIn: true, appearanceDarkCodeThemeId: "raycast", usePointerCursors: false, conversationDetailMode: "STEPS_COMMANDS", "hotkey-window-projectless-default-enabled": true, "enabled-reasoning-efforts": ["low", "medium", "high", "xhigh", "ultra", "max"], "selected-avatar-id": "null-signal", "reduced-motion-preference": "on", "show-context-window-usage": true, "ambient-suggestions-enabled": false, followUpQueueMode: "steer", }, experimental_compact_prompt_file: null, permissions: null, show_raw_agent_reasoning: null, check_for_update_on_startup: null, experimental_realtime_webrtc_call_base_url: null, model_instructions_file: null, profile: null, openai_base_url: null, experimental_use_unified_exec_tool: null, projects: {}, windows: null, model_catalog_json: null, experimental_realtime_start_instructions: null, plugins: {}, disable_paste_burst: null, hide_agent_reasoning: false, experimental_thread_config_endpoint: null, tool_suggest: null, auto_review: null, plan_mode_reasoning_effort: null, personality: "friendly", notice: null, include_apps_instructions: null, include_collaboration_mode_instructions: null, agents: null, ghost_snapshot: null, history: { persistence: "save-all", max_bytes: null }, experimental_realtime_ws_backend_prompt: null, otel: null, mcp_oauth_callback_port: null, oss_provider: null, background_terminal_max_timeout: null, mcp_oauth_callback_url: null, experimental_realtime_ws_model: null, tui: { notifications: true, notification_method: "auto", notification_condition: "unfocused", animations: true, show_tooltips: true, vim_mode_default: false, raw_output_mode: false, alternate_screen: "auto", status_line: null, status_line_use_colors: true, terminal_title: null, theme: null, pet: null, pet_anchor: "composer", session_picker_view: null, resume_cwd: null, terminal_resize_reflow_max_rows: null, keymap: defaultTuiKeymap(), }, log_dir: null, sqlite_home: null, experimental_thread_store: null, shell_environment_policy: { inherit: null, ignore_default_excludes: null, exclude: null, set: {}, include_only: null, experimental_use_profile: null, }, mcp_oauth_credentials_store: null, profiles: {}, features: { network_proxy: null, js_repl: false, memories: true, auth_elicitation: true, mentions_v2: true, remote_control: false, remote_plugin: true, tool_suggest: true, }, cli_auth_credentials_store: null, apps_mcp_product_sku: null, chatgpt_base_url: null, orchestrator: null, mcp_servers: {}, debug: null, notify: [], experimental_realtime_ws_base_url: null, model_providers: {}, js_repl_node_module_dirs: null, hooks: { PreToolUse: [], PermissionRequest: [], PostToolUse: [], PreCompact: [], PostCompact: [], SessionStart: [], SessionEnd: [], UserPromptSubmit: [], SubagentStart: [], SubagentStop: [], Stop: [], state: {}, }, suppress_unstable_features_warning: null, project_root_markers: null, default_permissions: null, js_repl_node_path: null, feedback: null, marketplaces: {}, memories: { disable_on_external_context: false, generate_memories: true, use_memories: true, dedicated_tools: null, max_raw_memories_for_consolidation: null, max_unused_days: null, max_rollout_age_days: null, max_rollouts_per_startup: null, min_rollout_idle_hours: null, min_rate_limit_remaining_percent: null, extract_model: null, consolidation_model: null, }, project_doc_max_bytes: 1048576, skills: null, file_opener: null, allow_login_shell: true, include_environment_context: null, realtime: null, tool_output_token_limit: null, audio: null, experimental_thread_store_endpoint: null, include_permissions_instructions: null, experimental_realtime_ws_startup_context: null, project_doc_fallback_filenames: [], }; } function normalizeItemStatus(status) { switch (String(status || "").toLowerCase()) { case "completed": case "success": return "completed"; case "declined": case "denied": return "declined"; case "interrupted": case "cancelled": case "canceled": return "interrupted"; default: return "failed"; } } function normalizeApprovalRequest(active, request) { const kind = request.kind === "file" ? "file" : "command"; const itemId = active.items.get(request.toolCallId) || request.itemId || request.toolCallId || id("approval"); if (kind === "file") { return { kind, params: { threadId: active.threadId, turnId: active.turnId, itemId, reason: request.reason || null, grantRoot: request.grantRoot || null, changes: request.changes || [], startedAtMs: Date.now(), }, }; } return { kind, params: { threadId: active.threadId, turnId: active.turnId, itemId, reason: request.reason || null, command: Array.isArray(request.command) ? request.command.join(" ") : String(request.command || ""), cwd: request.cwd || active.params.cwd || null, risk: request.risk || null, startedAtMs: Date.now(), }, }; } function approvalCacheKey(normalized) { if (normalized.kind === "file") { const roots = (normalized.params.changes || []).map((change) => path.dirname(change.path || "")).sort(); return `file:${roots.join("|")}`; } const command = String(normalized.params.command || "").trim().split(/\s+/).slice(0, 2).join(" "); return `command:${normalized.params.cwd || ""}:${command}`; } async function executeCommand(command, { cwd, env, timeoutMs = 120_000, maxOutputBytes = 8 * 1024 * 1024 } = {}) { const argv = Array.isArray(command) ? command.map(String) : null; const child = argv ? spawn(argv[0], argv.slice(1), { cwd, env: { ...process.env, ...(env || {}) }, stdio: ["ignore", "pipe", "pipe"] }) : spawn(String(command), { cwd, env: { ...process.env, ...(env || {}) }, shell: true, stdio: ["ignore", "pipe", "pipe"] }); let stdout = Buffer.alloc(0); let stderr = Buffer.alloc(0); let truncated = false; const append = (current, chunk) => { if (current.length >= maxOutputBytes) { truncated = true; return current; } const room = maxOutputBytes - current.length; if (chunk.length > room) truncated = true; return Buffer.concat([current, chunk.subarray(0, room)]); }; child.stdout.on("data", (chunk) => { stdout = append(stdout, chunk); }); child.stderr.on("data", (chunk) => { stderr = append(stderr, chunk); }); const startedAt = Date.now(); const timer = setTimeout(() => child.kill("SIGTERM"), Math.max(1, Number(timeoutMs || 120_000))); timer.unref?.(); const { code, signal } = await new Promise((resolve, reject) => { child.once("error", reject); child.once("close", (code, signal) => resolve({ code, signal })); }); clearTimeout(timer); return { exitCode: code, signal, stdout: stdout.toString("utf8"), stderr: stderr.toString("utf8"), truncated, durationMs: Date.now() - startedAt, }; }