import { randomUUID } from "node:crypto";import { EventEmitter } from "node:events"; import { ApprovalRequestId, EventId, ProviderItemId, ProviderRequestKind, ProviderSessionId, ProviderThreadId, ProviderTurnId, normalizeModelSlug, type ProviderApprovalDecision, type ProviderEvent, type ProviderSession, type ProviderSessionStartInput, type ProviderTurnStartResult,} from "@t3tools/contracts";import type * as NodeServices from "@effect/platform-node/NodeServices";import { Cause, Effect, Exit, ServiceMap, Scope, Stream } from "effect";import type { ChildProcessSpawner } from "effect/unstable/process";import { makeRuntimeCommand, spawnManagedCommand } from "./processRunner"; type PendingRequestKey = string; interface PendingRequest { method: string; timeout: ReturnType<typeof setTimeout>; resolve: (value: unknown) => void; reject: (error: Error) => void;} interface PendingApprovalRequest { requestId: ApprovalRequestId; jsonRpcId: string | number; method: "item/commandExecution/requestApproval" | "item/fileChange/requestApproval"; requestKind: ProviderRequestKind; threadId?: ProviderThreadId; turnId?: ProviderTurnId; itemId?: ProviderItemId;} interface CodexSessionContext { session: ProviderSession; child: ChildProcessSpawner.ChildProcessHandle; scope: Scope.Closeable; pending: Map<PendingRequestKey, PendingRequest>; pendingApprovals: Map<ApprovalRequestId, PendingApprovalRequest>; nextRequestId: number; stopping: boolean;} interface JsonRpcError { code?: number; message?: string;} interface JsonRpcRequest { id: string | number; method: string; params?: unknown;} interface JsonRpcResponse { id: string | number; result?: unknown; error?: JsonRpcError;} interface JsonRpcNotification { method: string; params?: unknown;} export interface CodexAppServerSendTurnInput { readonly sessionId: ProviderSessionId; readonly input?: string; readonly attachments?: ReadonlyArray<{ type: "image"; url: string }>; readonly model?: string; readonly effort?: string;} export interface CodexThreadTurnSnapshot { id: ProviderTurnId; items: unknown[];} export interface CodexThreadSnapshot { threadId: ProviderThreadId; turns: CodexThreadTurnSnapshot[];} const ANSI_ESCAPE_CHAR = String.fromCharCode(27);const ANSI_ESCAPE_REGEX = new RegExp(`${ANSI_ESCAPE_CHAR}\\[[0-9;]*m`, "g");const CODEX_STDERR_LOG_REGEX = /^\d{4}-\d{2}-\d{2}T\S+\s+(TRACE|DEBUG|INFO|WARN|ERROR)\s+\S+:\s+(.*)$/;const BENIGN_ERROR_LOG_SNIPPETS = [ "state db missing rollout path for thread", "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back",];const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "not found", "missing thread", "no such thread", "unknown thread", "does not exist",]; export function normalizeCodexModelSlug( model: string | undefined | null, preferredId?: string,): string | undefined { const normalized = normalizeModelSlug(model); if (!normalized) { return undefined; } if (preferredId?.endsWith("-codex") && preferredId !== normalized) { return preferredId; } return normalized;} export function classifyCodexStderrLine(rawLine: string): { message: string } | null { const line = rawLine.replaceAll(ANSI_ESCAPE_REGEX, "").trim(); if (!line) { return null; } const match = line.match(CODEX_STDERR_LOG_REGEX); if (match) { const level = match[1]; if (level && level !== "ERROR") { return null; } const isBenignError = BENIGN_ERROR_LOG_SNIPPETS.some((snippet) => line.includes(snippet)); if (isBenignError) { return null; } } return { message: line };} export function isRecoverableThreadResumeError(error: unknown): boolean { const message = (error instanceof Error ? error.message : String(error)).toLowerCase(); if (!message.includes("thread/resume")) { return false; } return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet));} export function messageFromCodexProcessCause(cause: Cause.Cause<unknown>): string { const squashed = Cause.squash(cause); return squashed instanceof Error ? squashed.message : "codex app-server process errored.";} export interface CodexAppServerManagerEvents { event: [event: ProviderEvent];} export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEvents> { private readonly sessions = new Map<ProviderSessionId, CodexSessionContext>(); private readonly runPromise: <A, E>( effect: Effect.Effect<A, E, NodeServices.NodeServices>, options?: Effect.RunOptions | undefined, ) => Promise<A>; constructor(services?: ServiceMap.ServiceMap<NodeServices.NodeServices>) { super(); this.runPromise = services ? Effect.runPromiseWith(services) : ((effect, options) => Effect.runPromise( effect as unknown as Effect.Effect<unknown, never>, options, )) as typeof this.runPromise; } async startSession(input: ProviderSessionStartInput): Promise<ProviderSession> { const sessionId = ProviderSessionId.makeUnsafe(randomUUID()); const now = new Date().toISOString(); let context: CodexSessionContext | undefined; try { const resolvedCwd = input.cwd ?? process.cwd(); const session: ProviderSession = { sessionId, provider: "codex", status: "connecting", model: normalizeCodexModelSlug(input.model), cwd: resolvedCwd, createdAt: now, updatedAt: now, }; const codexBinaryPath = input.codexBinaryPath ?? "codex"; const codexHomePath = input.codexHomePath; const managedProcess = await this.runPromise( spawnManagedCommand( makeRuntimeCommand(codexBinaryPath, ["app-server"], { cwd: resolvedCwd, env: { ...process.env, ...(codexHomePath ? { CODEX_HOME: codexHomePath } : {}), }, extendEnv: false, }), ), ); context = { session, child: managedProcess.handle, scope: managedProcess.scope, pending: new Map(), pendingApprovals: new Map(), nextRequestId: 1, stopping: false, }; this.sessions.set(sessionId, context); await this.attachProcessListeners(context); this.emitLifecycleEvent(context, "session/connecting", "Starting codex app-server"); await this.sendRequest(context, "initialize", { clientInfo: { name: "t3code_desktop", title: "T3 Code Desktop", version: "0.1.0", }, capabilities: { experimentalApi: false, }, }); await this.writeMessage(context, { method: "initialized" }); const normalizedModel = normalizeCodexModelSlug(input.model); const sessionOverrides = { model: normalizedModel ?? null, cwd: input.cwd ?? null, approvalPolicy: input.approvalPolicy, sandbox: input.sandboxMode, }; const threadStartParams = { ...sessionOverrides, experimentalRawEvents: false, }; const resumeThreadId = readResumeThreadId(input); let threadOpenMethod: "thread/start" | "thread/resume" = "thread/start"; let threadOpenResponse: unknown; if (resumeThreadId) { try { threadOpenMethod = "thread/resume"; threadOpenResponse = await this.sendRequest(context, "thread/resume", { ...sessionOverrides, threadId: resumeThreadId, }); } catch (error) { if (!isRecoverableThreadResumeError(error)) { throw error; } threadOpenMethod = "thread/start"; this.emitLifecycleEvent( context, "session/threadResumeFallback", `Could not resume thread ${resumeThreadId}; started a new thread instead.`, ); threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams); } } else { threadOpenMethod = "thread/start"; threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams); } const threadOpenRecord = this.readObject(threadOpenResponse); const threadIdRaw = this.readString(this.readObject(threadOpenRecord, "thread"), "id") ?? this.readString(threadOpenRecord, "threadId"); if (!threadIdRaw) { throw new Error(`${threadOpenMethod} response did not include a thread id.`); } const threadId = ProviderThreadId.makeUnsafe(threadIdRaw); this.updateSession(context, { status: "ready", threadId, resumeCursor: { threadId }, }); this.emitLifecycleEvent(context, "session/ready", `Connected to thread ${threadId}`); return { ...context.session }; } catch (error) { const message = error instanceof Error ? error.message : "Failed to start Codex session."; if (context) { this.updateSession(context, { status: "error", lastError: message, }); this.emitErrorEvent(context, "session/startFailed", message); this.stopSession(sessionId); } else { this.emitEvent({ id: EventId.makeUnsafe(randomUUID()), kind: "error", provider: "codex", sessionId, createdAt: new Date().toISOString(), method: "session/startFailed", message, }); } throw new Error(message, { cause: error }); } } async sendTurn(input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> { const context = this.requireSession(input.sessionId); if (!context.session.threadId) { throw new Error("Session is missing a thread id."); } const turnInput: Array< { type: "text"; text: string; text_elements: [] } | { type: "image"; url: string } > = []; if (input.input) { turnInput.push({ type: "text", text: input.input, text_elements: [], }); } for (const attachment of input.attachments ?? []) { if (attachment.type === "image") { turnInput.push({ type: "image", url: attachment.url, }); } } if (turnInput.length === 0) { throw new Error("Turn input must include text or attachments."); } const turnStartParams: { threadId: ProviderThreadId; input: Array< { type: "text"; text: string; text_elements: [] } | { type: "image"; url: string } >; model?: string; effort?: string; } = { threadId: context.session.threadId, input: turnInput, }; const normalizedModel = normalizeCodexModelSlug(input.model); if (normalizedModel) { turnStartParams.model = normalizedModel; } if (input.effort) { turnStartParams.effort = input.effort; } const response = await this.sendRequest(context, "turn/start", turnStartParams); const turn = this.readObject(this.readObject(response), "turn"); const turnIdRaw = this.readString(turn, "id"); if (!turnIdRaw) { throw new Error("turn/start response did not include a turn id."); } const turnId = ProviderTurnId.makeUnsafe(turnIdRaw); this.updateSession(context, { status: "running", activeTurnId: turnId, resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId }, }); return { threadId: context.session.threadId, turnId, resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId }, }; } async interruptTurn(sessionId: ProviderSessionId, turnId?: ProviderTurnId): Promise<void> { const context = this.requireSession(sessionId); const effectiveTurnId = turnId ?? context.session.activeTurnId; if (!effectiveTurnId || !context.session.threadId) { return; } await this.sendRequest(context, "turn/interrupt", { threadId: context.session.threadId, turnId: effectiveTurnId, }); } async readThread(sessionId: ProviderSessionId): Promise<CodexThreadSnapshot> { const context = this.requireSession(sessionId); const threadId = context.session.threadId; if (!threadId) { throw new Error("Session is missing a thread id."); } const response = await this.sendRequest(context, "thread/read", { threadId, includeTurns: true, }); return this.parseThreadSnapshot("thread/read", response); } async rollbackThread( sessionId: ProviderSessionId, numTurns: number, ): Promise<CodexThreadSnapshot> { const context = this.requireSession(sessionId); const threadId = context.session.threadId; if (!threadId) { throw new Error("Session is missing a thread id."); } if (!Number.isInteger(numTurns) || numTurns < 1) { throw new Error("numTurns must be an integer >= 1."); } const response = await this.sendRequest(context, "thread/rollback", { threadId, numTurns, }); this.updateSession(context, { status: "ready", activeTurnId: undefined, }); return this.parseThreadSnapshot("thread/rollback", response); } async respondToRequest( sessionId: ProviderSessionId, requestId: ApprovalRequestId, decision: ProviderApprovalDecision, ): Promise<void> { const context = this.requireSession(sessionId); const pendingRequest = context.pendingApprovals.get(requestId); if (!pendingRequest) { throw new Error(`Unknown pending approval request: ${requestId}`); } context.pendingApprovals.delete(requestId); await this.writeMessage(context, { id: pendingRequest.jsonRpcId, result: { decision, }, }); this.emitEvent({ id: EventId.makeUnsafe(randomUUID()), kind: "notification", provider: "codex", sessionId: context.session.sessionId, createdAt: new Date().toISOString(), method: "item/requestApproval/decision", threadId: pendingRequest.threadId, turnId: pendingRequest.turnId, itemId: pendingRequest.itemId, requestId: pendingRequest.requestId, requestKind: pendingRequest.requestKind, payload: { requestId: pendingRequest.requestId, requestKind: pendingRequest.requestKind, decision, }, }); } stopSession(sessionId: ProviderSessionId): void { const context = this.sessions.get(sessionId); if (!context) { return; } context.stopping = true; for (const pending of context.pending.values()) { clearTimeout(pending.timeout); pending.reject(new Error("Session stopped before request completed.")); } context.pending.clear(); context.pendingApprovals.clear(); void this.runPromise(Scope.close(context.scope, Exit.void)).catch(() => undefined); Lines 1–500 of 1071. Load more
import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; import { ApprovalRequestId, EventId, ProviderItemId, ProviderRequestKind, ProviderSessionId, ProviderThreadId, ProviderTurnId, normalizeModelSlug, type ProviderApprovalDecision, type ProviderEvent, type ProviderSession, type ProviderSessionStartInput, type ProviderTurnStartResult, } from "@t3tools/contracts"; import type * as NodeServices from "@effect/platform-node/NodeServices"; import { Cause, Effect, Exit, ServiceMap, Scope, Stream } from "effect"; import type { ChildProcessSpawner } from "effect/unstable/process"; import { makeRuntimeCommand, spawnManagedCommand } from "./processRunner"; type PendingRequestKey = string; interface PendingRequest { method: string; timeout: ReturnType<typeof setTimeout>; resolve: (value: unknown) => void; reject: (error: Error) => void; } interface PendingApprovalRequest { requestId: ApprovalRequestId; jsonRpcId: string | number; method: "item/commandExecution/requestApproval" | "item/fileChange/requestApproval"; requestKind: ProviderRequestKind; threadId?: ProviderThreadId; turnId?: ProviderTurnId; itemId?: ProviderItemId; } interface CodexSessionContext { session: ProviderSession; child: ChildProcessSpawner.ChildProcessHandle; scope: Scope.Closeable; pending: Map<PendingRequestKey, PendingRequest>; pendingApprovals: Map<ApprovalRequestId, PendingApprovalRequest>; nextRequestId: number; stopping: boolean; } interface JsonRpcError { code?: number; message?: string; } interface JsonRpcRequest { id: string | number; method: string; params?: unknown; } interface JsonRpcResponse { id: string | number; result?: unknown; error?: JsonRpcError; } interface JsonRpcNotification { method: string; params?: unknown; } export interface CodexAppServerSendTurnInput { readonly sessionId: ProviderSessionId; readonly input?: string; readonly attachments?: ReadonlyArray<{ type: "image"; url: string }>; readonly model?: string; readonly effort?: string; } export interface CodexThreadTurnSnapshot { id: ProviderTurnId; items: unknown[]; } export interface CodexThreadSnapshot { threadId: ProviderThreadId; turns: CodexThreadTurnSnapshot[]; } const ANSI_ESCAPE_CHAR = String.fromCharCode(27); const ANSI_ESCAPE_REGEX = new RegExp(`${ANSI_ESCAPE_CHAR}\\[[0-9;]*m`, "g"); const CODEX_STDERR_LOG_REGEX = /^\d{4}-\d{2}-\d{2}T\S+\s+(TRACE|DEBUG|INFO|WARN|ERROR)\s+\S+:\s+(.*)$/; const BENIGN_ERROR_LOG_SNIPPETS = [ "state db missing rollout path for thread", "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back", ]; const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ "not found", "missing thread", "no such thread", "unknown thread", "does not exist", ]; export function normalizeCodexModelSlug( model: string | undefined | null, preferredId?: string, ): string | undefined { const normalized = normalizeModelSlug(model); if (!normalized) { return undefined; } if (preferredId?.endsWith("-codex") && preferredId !== normalized) { return preferredId; } return normalized; } export function classifyCodexStderrLine(rawLine: string): { message: string } | null { const line = rawLine.replaceAll(ANSI_ESCAPE_REGEX, "").trim(); if (!line) { return null; } const match = line.match(CODEX_STDERR_LOG_REGEX); if (match) { const level = match[1]; if (level && level !== "ERROR") { return null; } const isBenignError = BENIGN_ERROR_LOG_SNIPPETS.some((snippet) => line.includes(snippet)); if (isBenignError) { return null; } } return { message: line }; } export function isRecoverableThreadResumeError(error: unknown): boolean { const message = (error instanceof Error ? error.message : String(error)).toLowerCase(); if (!message.includes("thread/resume")) { return false; } return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet)); } export function messageFromCodexProcessCause(cause: Cause.Cause<unknown>): string { const squashed = Cause.squash(cause); return squashed instanceof Error ? squashed.message : "codex app-server process errored."; } export interface CodexAppServerManagerEvents { event: [event: ProviderEvent]; } export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEvents> { private readonly sessions = new Map<ProviderSessionId, CodexSessionContext>(); private readonly runPromise: <A, E>( effect: Effect.Effect<A, E, NodeServices.NodeServices>, options?: Effect.RunOptions | undefined, ) => Promise<A>; constructor(services?: ServiceMap.ServiceMap<NodeServices.NodeServices>) { super(); this.runPromise = services ? Effect.runPromiseWith(services) : ((effect, options) => Effect.runPromise( effect as unknown as Effect.Effect<unknown, never>, options, )) as typeof this.runPromise; } async startSession(input: ProviderSessionStartInput): Promise<ProviderSession> { const sessionId = ProviderSessionId.makeUnsafe(randomUUID()); const now = new Date().toISOString(); let context: CodexSessionContext | undefined; try { const resolvedCwd = input.cwd ?? process.cwd(); const session: ProviderSession = { sessionId, provider: "codex", status: "connecting", model: normalizeCodexModelSlug(input.model), cwd: resolvedCwd, createdAt: now, updatedAt: now, }; const codexBinaryPath = input.codexBinaryPath ?? "codex"; const codexHomePath = input.codexHomePath; const managedProcess = await this.runPromise( spawnManagedCommand( makeRuntimeCommand(codexBinaryPath, ["app-server"], { cwd: resolvedCwd, env: { ...process.env, ...(codexHomePath ? { CODEX_HOME: codexHomePath } : {}), }, extendEnv: false, }), ), ); context = { session, child: managedProcess.handle, scope: managedProcess.scope, pending: new Map(), pendingApprovals: new Map(), nextRequestId: 1, stopping: false, }; this.sessions.set(sessionId, context); await this.attachProcessListeners(context); this.emitLifecycleEvent(context, "session/connecting", "Starting codex app-server"); await this.sendRequest(context, "initialize", { clientInfo: { name: "t3code_desktop", title: "T3 Code Desktop", version: "0.1.0", }, capabilities: { experimentalApi: false, }, }); await this.writeMessage(context, { method: "initialized" }); const normalizedModel = normalizeCodexModelSlug(input.model); const sessionOverrides = { model: normalizedModel ?? null, cwd: input.cwd ?? null, approvalPolicy: input.approvalPolicy, sandbox: input.sandboxMode, }; const threadStartParams = { ...sessionOverrides, experimentalRawEvents: false, }; const resumeThreadId = readResumeThreadId(input); let threadOpenMethod: "thread/start" | "thread/resume" = "thread/start"; let threadOpenResponse: unknown; if (resumeThreadId) { try { threadOpenMethod = "thread/resume"; threadOpenResponse = await this.sendRequest(context, "thread/resume", { ...sessionOverrides, threadId: resumeThreadId, }); } catch (error) { if (!isRecoverableThreadResumeError(error)) { throw error; } threadOpenMethod = "thread/start"; this.emitLifecycleEvent( context, "session/threadResumeFallback", `Could not resume thread ${resumeThreadId}; started a new thread instead.`, ); threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams); } } else { threadOpenMethod = "thread/start"; threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams); } const threadOpenRecord = this.readObject(threadOpenResponse); const threadIdRaw = this.readString(this.readObject(threadOpenRecord, "thread"), "id") ?? this.readString(threadOpenRecord, "threadId"); if (!threadIdRaw) { throw new Error(`${threadOpenMethod} response did not include a thread id.`); } const threadId = ProviderThreadId.makeUnsafe(threadIdRaw); this.updateSession(context, { status: "ready", threadId, resumeCursor: { threadId }, }); this.emitLifecycleEvent(context, "session/ready", `Connected to thread ${threadId}`); return { ...context.session }; } catch (error) { const message = error instanceof Error ? error.message : "Failed to start Codex session."; if (context) { this.updateSession(context, { status: "error", lastError: message, }); this.emitErrorEvent(context, "session/startFailed", message); this.stopSession(sessionId); } else { this.emitEvent({ id: EventId.makeUnsafe(randomUUID()), kind: "error", provider: "codex", sessionId, createdAt: new Date().toISOString(), method: "session/startFailed", message, }); } throw new Error(message, { cause: error }); } } async sendTurn(input: CodexAppServerSendTurnInput): Promise<ProviderTurnStartResult> { const context = this.requireSession(input.sessionId); if (!context.session.threadId) { throw new Error("Session is missing a thread id."); } const turnInput: Array< { type: "text"; text: string; text_elements: [] } | { type: "image"; url: string } > = []; if (input.input) { turnInput.push({ type: "text", text: input.input, text_elements: [], }); } for (const attachment of input.attachments ?? []) { if (attachment.type === "image") { turnInput.push({ type: "image", url: attachment.url, }); } } if (turnInput.length === 0) { throw new Error("Turn input must include text or attachments."); } const turnStartParams: { threadId: ProviderThreadId; input: Array< { type: "text"; text: string; text_elements: [] } | { type: "image"; url: string } >; model?: string; effort?: string; } = { threadId: context.session.threadId, input: turnInput, }; const normalizedModel = normalizeCodexModelSlug(input.model); if (normalizedModel) { turnStartParams.model = normalizedModel; } if (input.effort) { turnStartParams.effort = input.effort; } const response = await this.sendRequest(context, "turn/start", turnStartParams); const turn = this.readObject(this.readObject(response), "turn"); const turnIdRaw = this.readString(turn, "id"); if (!turnIdRaw) { throw new Error("turn/start response did not include a turn id."); } const turnId = ProviderTurnId.makeUnsafe(turnIdRaw); this.updateSession(context, { status: "running", activeTurnId: turnId, resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId }, }); return { threadId: context.session.threadId, turnId, resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId }, }; } async interruptTurn(sessionId: ProviderSessionId, turnId?: ProviderTurnId): Promise<void> { const context = this.requireSession(sessionId); const effectiveTurnId = turnId ?? context.session.activeTurnId; if (!effectiveTurnId || !context.session.threadId) { return; } await this.sendRequest(context, "turn/interrupt", { threadId: context.session.threadId, turnId: effectiveTurnId, }); } async readThread(sessionId: ProviderSessionId): Promise<CodexThreadSnapshot> { const context = this.requireSession(sessionId); const threadId = context.session.threadId; if (!threadId) { throw new Error("Session is missing a thread id."); } const response = await this.sendRequest(context, "thread/read", { threadId, includeTurns: true, }); return this.parseThreadSnapshot("thread/read", response); } async rollbackThread( sessionId: ProviderSessionId, numTurns: number, ): Promise<CodexThreadSnapshot> { const context = this.requireSession(sessionId); const threadId = context.session.threadId; if (!threadId) { throw new Error("Session is missing a thread id."); } if (!Number.isInteger(numTurns) || numTurns < 1) { throw new Error("numTurns must be an integer >= 1."); } const response = await this.sendRequest(context, "thread/rollback", { threadId, numTurns, }); this.updateSession(context, { status: "ready", activeTurnId: undefined, }); return this.parseThreadSnapshot("thread/rollback", response); } async respondToRequest( sessionId: ProviderSessionId, requestId: ApprovalRequestId, decision: ProviderApprovalDecision, ): Promise<void> { const context = this.requireSession(sessionId); const pendingRequest = context.pendingApprovals.get(requestId); if (!pendingRequest) { throw new Error(`Unknown pending approval request: ${requestId}`); } context.pendingApprovals.delete(requestId); await this.writeMessage(context, { id: pendingRequest.jsonRpcId, result: { decision, }, }); this.emitEvent({ id: EventId.makeUnsafe(randomUUID()), kind: "notification", provider: "codex", sessionId: context.session.sessionId, createdAt: new Date().toISOString(), method: "item/requestApproval/decision", threadId: pendingRequest.threadId, turnId: pendingRequest.turnId, itemId: pendingRequest.itemId, requestId: pendingRequest.requestId, requestKind: pendingRequest.requestKind, payload: { requestId: pendingRequest.requestId, requestKind: pendingRequest.requestKind, decision, }, }); } stopSession(sessionId: ProviderSessionId): void { const context = this.sessions.get(sessionId); if (!context) { return; } context.stopping = true; for (const pending of context.pending.values()) { clearTimeout(pending.timeout); pending.reject(new Error("Session stopped before request completed.")); } context.pending.clear(); context.pendingApprovals.clear(); void this.runPromise(Scope.close(context.scope, Exit.void)).catch(() => undefined);