import { type ChildProcessWithoutNullStreams, spawn } from "node:child_process";import { randomUUID } from "node:crypto";import { EventEmitter } from "node:events";import readline from "node:readline"; import type {  ProviderApprovalDecision,  ProviderEvent,  ProviderRequestKind,  ProviderSendTurnInput,  ProviderSession,  ProviderSessionStartInput,  ProviderTurnStartResult,} from "@t3tools/contracts";import { normalizeModelSlug } 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: string;  jsonRpcId: string | number;  method: "item/commandExecution/requestApproval" | "item/fileChange/requestApproval";  requestKind: ProviderRequestKind;  threadId?: string;  turnId?: string;  itemId?: string;} interface CodexSessionContext {  session: ProviderSession;  child: ChildProcessWithoutNullStreams;  output: readline.Interface;  pending: Map<PendingRequestKey, PendingRequest>;  pendingApprovals: Map<string, 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: string;  items: unknown[];} export interface CodexThreadSnapshot {  threadId: string;  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<string, CodexSessionContext>();   async startSession(input: ProviderSessionStartInput): Promise<ProviderSession> {    const sessionId = 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?.trim() || "codex";      const codexHomePath = input.codexHomePath?.trim();      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,      };       let threadOpenMethod: "thread/start" | "thread/resume" = "thread/start";      let threadOpenResponse: unknown;      if (input.resumeThreadId) {        try {          threadOpenMethod = "thread/resume";          threadOpenResponse = await this.sendRequest(context, "thread/resume", {            ...sessionOverrides,            threadId: input.resumeThreadId,          });        } catch (error) {          if (!isRecoverableThreadResumeError(error)) {            throw error;          }           threadOpenMethod = "thread/start";          this.emitLifecycleEvent(            context,            "session/threadResumeFallback",            `Could not resume thread ${input.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 threadId =        this.readString(this.readObject(threadOpenRecord, "thread"), "id") ??        this.readString(threadOpenRecord, "threadId");      if (!threadId) {        throw new Error(`${threadOpenMethod} response did not include a thread id.`);      }       this.updateSession(context, {        status: "ready",        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: 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: string;      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 turnId = this.readString(turn, "id");    if (!turnId) {      throw new Error("turn/start response did not include a turn id.");    }     this.updateSession(context, {      status: "running",      activeTurnId: turnId,    });     return {      threadId: context.session.threadId,      turnId,    };  }   async interruptTurn(sessionId: string, turnId?: string): 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: string): 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: string, 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: string,    requestId: string,    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: 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: string): 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: string): boolean {    return this.sessions.has(sessionId);  }   stopAll(): void {    for (const sessionId of this.sessions.keys()) {      this.stopSession(sessionId);    }  }   private requireSession(sessionId: string): CodexSessionContext {    const context = this.sessions.get(sessionId);    if (!context) {      throw new Error(`Unknown session: ${sessionId}`);    }     if (context.session.status === "closed") {      throw new Error(`Session is closed: ${sessionId}`);    }     return context;  }   private attachProcessListeners(context: CodexSessionContext): void {    context.output.on("line", (line) => {      this.handleStdoutLine(context, line);    }); 

Lines 1–500 of 943. Load more