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;} 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();     const session: ProviderSession = {      sessionId,      provider: "codex",      status: "connecting",      model: normalizeCodexModelSlug(input.model),      cwd: input.cwd,      createdAt: now,      updatedAt: now,    };     const child = spawn("codex", ["app-server"], {      cwd: input.cwd,      env: process.env,      stdio: ["pipe", "pipe", "pipe"],    });    const output = readline.createInterface({ input: child.stdout });     const context: CodexSessionContext = {      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");     try {      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.";      this.updateSession(context, {        status: "error",        lastError: message,      });      this.emitErrorEvent(context, "session/startFailed", message);      this.stopSession(sessionId);      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 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);    });     context.child.stderr.on("data", (chunk: Buffer) => {      const raw = chunk.toString();      const lines = raw.split(/\r?\n/g);      for (const rawLine of lines) {        const classified = classifyCodexStderrLine(rawLine);        if (!classified) {          continue;        }         this.emitErrorEvent(context, "process/stderr", classified.message);      }    });     context.child.on("error", (error) => {      const message = error.message || "codex app-server process errored.";      this.updateSession(context, {        status: "error",        lastError: message,      });      this.emitErrorEvent(context, "process/error", message);    });     context.child.on("exit", (code, signal) => {      if (context.stopping) {        return;      }       const message = `codex app-server exited (code=${code ?? "null"}, signal=${signal ?? "null"}).`;      this.updateSession(context, {        status: "closed",        activeTurnId: undefined,        lastError: code === 0 ? context.session.lastError : message,      });      this.emitLifecycleEvent(context, "session/exited", message);      this.sessions.delete(context.session.sessionId);    });  }   private handleStdoutLine(context: CodexSessionContext, line: string): void {    let parsed: unknown;    try {      parsed = JSON.parse(line);    } catch {      this.emitErrorEvent(        context,        "protocol/parseError",        "Received invalid JSON from codex app-server.",      );      return;    }     if (!parsed || typeof parsed !== "object") {      this.emitErrorEvent(        context,        "protocol/invalidMessage",        "Received non-object protocol message.",      );      return;    }     if (this.isServerRequest(parsed)) {      this.handleServerRequest(context, parsed);      return;

Lines 1–500 of 845. Load more