1008 lines · 28.5 KB · TypeScript
RawDownload
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 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 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 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: 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);    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()) {

Lines 1–500 of 1008.