import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";import { randomUUID } from "node:crypto";import { EventEmitter } from "node:events";import readline from "node:readline"; import { ApprovalRequestId, EventId, ProviderItemId, ProviderRequestKind, ProviderSessionId, ProviderThreadId, ProviderTurnId, normalizeModelSlug, type ProviderApprovalDecision, type ProviderEvent, type ProviderSendTurnInput, type ProviderSession, type ProviderSessionStartInput, type ProviderTurnStartResult,} from "@t3tools/contracts"; 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: ChildProcessWithoutNullStreams; output: readline.Interface; 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 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 interface CodexAppServerManagerEvents { event: [event: ProviderEvent];} export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEvents> { private readonly sessions = new Map<ProviderSessionId, CodexSessionContext>(); 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 child = spawn(codexBinaryPath, ["app-server"], { cwd: resolvedCwd, env: { ...process.env, ...(codexHomePath ? { CODEX_HOME: codexHomePath } : {}), }, stdio: ["pipe", "pipe", "pipe"], }); const output = readline.createInterface({ input: child.stdout }); context = { session, child, output, pending: new Map(), pendingApprovals: new Map(), nextRequestId: 1, stopping: false, }; this.sessions.set(sessionId, context); 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, }, }); 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: ProviderSendTurnInput): 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.dataUrl, }); } } 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); 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(); context.output.close(); if (!context.child.killed) { context.child.kill(); } this.updateSession(context, { status: "closed", activeTurnId: undefined, }); this.emitLifecycleEvent(context, "session/closed", "Session stopped"); this.sessions.delete(sessionId); } listSessions(): ProviderSession[] { return Array.from(this.sessions.values(), ({ session }) => ({ ...session, })); } hasSession(sessionId: ProviderSessionId): boolean { return this.sessions.has(sessionId); } stopAll(): void { for (const sessionId of this.sessions.keys()) { this.stopSession(sessionId); } } private requireSession(sessionId: ProviderSessionId): CodexSessionContext { const context = this.sessions.get(sessionId); if (!context) {Lines 1–500 of 1001. Load more
import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; import readline from "node:readline"; import { ApprovalRequestId, EventId, ProviderItemId, ProviderRequestKind, ProviderSessionId, ProviderThreadId, ProviderTurnId, normalizeModelSlug, type ProviderApprovalDecision, type ProviderEvent, type ProviderSendTurnInput, type ProviderSession, type ProviderSessionStartInput, type ProviderTurnStartResult, } from "@t3tools/contracts"; 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: ChildProcessWithoutNullStreams; output: readline.Interface; 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 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 interface CodexAppServerManagerEvents { event: [event: ProviderEvent]; } export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEvents> { private readonly sessions = new Map<ProviderSessionId, CodexSessionContext>(); 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 child = spawn(codexBinaryPath, ["app-server"], { cwd: resolvedCwd, env: { ...process.env, ...(codexHomePath ? { CODEX_HOME: codexHomePath } : {}), }, stdio: ["pipe", "pipe", "pipe"], }); const output = readline.createInterface({ input: child.stdout }); context = { session, child, output, pending: new Map(), pendingApprovals: new Map(), nextRequestId: 1, stopping: false, }; this.sessions.set(sessionId, context); 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, }, }); 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: ProviderSendTurnInput): 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.dataUrl, }); } } 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); 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(); context.output.close(); if (!context.child.killed) { context.child.kill(); } this.updateSession(context, { status: "closed", activeTurnId: undefined, }); this.emitLifecycleEvent(context, "session/closed", "Session stopped"); this.sessions.delete(sessionId); } listSessions(): ProviderSession[] { return Array.from(this.sessions.values(), ({ session }) => ({ ...session, })); } hasSession(sessionId: ProviderSessionId): boolean { return this.sessions.has(sessionId); } stopAll(): void { for (const sessionId of this.sessions.keys()) { this.stopSession(sessionId); } } private requireSession(sessionId: ProviderSessionId): CodexSessionContext { const context = this.sessions.get(sessionId); if (!context) {