feat(codex): 增加会话级 Fast Mode 控制

CONTEXT.md

@@ -19,6 +19,10 @@ _Avoid_: agent, app
 A continuing conversation between one chat anchor and one **Agent CLI**.
 _Avoid_: thread, task
 
+**Fast Mode**:
+A per-**Session** opt-in to a faster Agent CLI service tier. Every new Session starts with Fast Mode disabled, and changing it never changes another Session.
+_Avoid_: global fast setting, Bot-level fast mode
+
 **Token Usage**:
 Token counts reported by an **Agent CLI** or its persisted transcript for a
 **Session**. Token In is the Agent CLI's native input-side total, including

src/adapters/backend/reproduce-command.ts

@@ -39,14 +39,18 @@ export function selectReproduceLaunch(input: {
   sandboxOn: boolean;
   binResolver?: (bin: string) => string;
   ttadkModel?: string;
+  codexServiceTier?: string;
 }): { bin: string; args: string[] } {
   const { baseBin, baseArgs, wrapperCli, sandboxOn } = input;
   if (wrapperCli && wrapperCli.trim() && !sandboxOn) {
     const launch = buildWrappedLaunch(
       wrapperCli,
       baseArgs,
       input.binResolver ?? ((b) => b),
-      { ttadkModel: input.ttadkModel },
+      {
+        ttadkModel: input.ttadkModel,
+        codexServiceTier: input.codexServiceTier,
+      },
     );
     if (launch.bin) return { bin: launch.bin, args: launch.args };
   }

src/adapters/cli/codex.ts

@@ -154,7 +154,11 @@ export function createCodexAdapter(pathOverride?: string): CliAdapter {
     authPaths: ['~/.codex'],
     get resolvedBin(): string { return (cachedBin ??= resolveCommand(rawBin)); },
 
-    buildArgs({ sessionId, resume, resumeSessionId, workingDir, model, reasoningEffort, disableCliBypass, readIsolation, remoteWsUrl, remoteThreadId }) {
+    buildArgs({ sessionId, resume, resumeSessionId, workingDir, model, reasoningEffort, fastMode, fastServiceTier, disableCliBypass, readIsolation, remoteWsUrl, remoteThreadId }) {
+      if (fastMode === true && !fastServiceTier) {
+        throw new Error('Fast service tier was not resolved from the Codex model catalog');
+      }
+      const serviceTier = fastMode === true ? fastServiceTier! : 'default';
       // Hybrid RPC input mode: attach this TUI to the botmux-owned app-server
       // thread. User input is delivered out-of-band via JSON-RPC (turn/start,
       // see codex-rpc-engine + worker), so the pane is a pure viewer — no paste
@@ -165,7 +169,12 @@ export function createCodexAdapter(pathOverride?: string): CliAdapter {
         // enter to continue" dialog would block the resume forever and freeze the
         // Web terminal. Disable the check at the PROCESS level (never the user's
         // global config). The bounded startup-dialog watcher is only a fail-safe.
-        return ['--remote', remoteWsUrl, 'resume', '--no-alt-screen', '-c', 'check_for_update_on_startup=false', remoteThreadId];
+        return [
+          '--remote', remoteWsUrl, 'resume', '--no-alt-screen',
+          '-c', 'check_for_update_on_startup=false',
+          '-c', `service_tier=${JSON.stringify(serviceTier)}`,
+          remoteThreadId,
+        ];
       }
       // Read isolation for Codex is enforced by the worker's Seatbelt wrapper,
       // NOT by codex's own profile (codex 0.137 can't express a read blocklist).
@@ -182,6 +191,11 @@ export function createCodexAdapter(pathOverride?: string): CliAdapter {
         // not); the host-side daily monitor reports newer versions to the owner.
         '-c',
         'check_for_update_on_startup=false',
+        // Codex persists its native /fast toggle in CODEX_HOME, which would
+        // otherwise leak one Session's choice into every later Session. Pin the
+        // effective tier at process launch; Botmux owns the per-Session state.
+        '-c',
+        `service_tier=${JSON.stringify(serviceTier)}`,
       ];
       // Under read isolation the worker denies bots.json, so `botmux send` (a shell
       // subprocess) registers this bot from the worker-written cred FILE, keyed by

src/adapters/cli/types.ts

@@ -101,6 +101,11 @@ export interface CliAdapter {
      *  `--model` flag (or equivalent) inject it here; adapters whose CLI has no
      *  such concept simply ignore the field. Empty / undefined → CLI default. */
     model?: string;
+    /** Session-scoped Fast Mode. Codex consumes this as a process-level service
+     * tier override; other adapters ignore it. Missing/false means standard. */
+    fastMode?: boolean;
+    /** Concrete Fast tier id resolved from Codex's model catalog. */
+    fastServiceTier?: string;
     /** Optional per-turn reasoning effort (codex `model_reasoning_effort`).
      *  Only codex/codex-app adapters honor it; others ignore. */
     reasoningEffort?: 'low' | 'medium' | 'high' | 'xhigh';

src/codex-rpc-engine.ts

@@ -32,6 +32,20 @@ import { WebSocket } from 'ws';
 type Json = Record<string, any>;
 type LogFn = (msg: string) => void;
 
+class CodexRpcRequestAbortedError extends Error {
+  constructor(method: string) {
+    super(`codex app-server request '${method}' was cancelled`);
+    this.name = 'CodexRpcRequestAbortedError';
+  }
+}
+
+export class CodexRpcFastModeCancelledError extends Error {
+  constructor(message = 'Codex Fast Mode update was cancelled') {
+    super(message);
+    this.name = 'CodexRpcFastModeCancelledError';
+  }
+}
+
 async function findFreePort(): Promise<number> {
   return new Promise<number>((resolve, reject) => {
     const srv = createServer();
@@ -68,6 +82,9 @@ export interface CodexRpcEngineOpts {
   /** Optional model + reasoning effort forwarded to thread config (P1). */
   model?: string;
   reasoningEffort?: string;
+  /** Session-scoped Fast Mode. The concrete protocol tier id is resolved from
+   * `model/list`; "Fast" is a display name and is not itself a stable tier id. */
+  fastMode?: boolean;
   /** Feature gates owned by the app-server process (the viewer TUI does not
    *  execute model tools in RPC mode). */
   appServerFeatures?: string[];
@@ -111,6 +128,7 @@ export class CodexRpcEngine {
   private pending = new Map<number, { resolve: (v: any) => void; reject: (e: Error) => void; timer: ReturnType<typeof setTimeout> }>();
   private port = 0;
   private threadId?: string;
+  private serviceTier?: string;
   private closed = false;
   private deadNotified = false;
   private lastStderr = '';
@@ -122,6 +140,7 @@ export class CodexRpcEngine {
 
   get wsUrl(): string { return `ws://127.0.0.1:${this.port}`; }
   get activeThreadId(): string | undefined { return this.threadId; }
+  get activeServiceTier(): string | undefined { return this.serviceTier; }
   get appServerPid(): number | undefined { return this.child?.pid; }
 
   /** Spawn the app-server, connect, and complete the initialize handshake. */
@@ -159,6 +178,7 @@ export class CodexRpcEngine {
   /** Create a fresh session thread. Its id (== codex rollout session id) is what
    *  the TUI resumes and what botmux persists for future resume. */
   async startThread(): Promise<string> {
+    await this.prepareInitialServiceTier();
     const r = await this.request('thread/start', this.threadParams());
     this.threadId = String(r?.thread?.id ?? '');
     if (!this.threadId) throw new Error('thread/start returned no thread id');
@@ -169,6 +189,7 @@ export class CodexRpcEngine {
    *  so RPC mode stays engaged across daemon restarts instead of reverting to
    *  the paste path. */
   async resumeThread(threadId: string): Promise<string> {
+    await this.prepareInitialServiceTier();
     // forResume=true: a cold resume must NOT re-send ANY model-related override.
     // The codex/TraeX app-server sees any single override (model OR
     // model_reasoning_effort) as "caller is pinning config" and early-returns out
@@ -207,10 +228,134 @@ export class CodexRpcEngine {
       serviceName: 'botmux',
       ephemeral: false,
       persistExtendedHistory: true,
+      serviceTier: this.serviceTier ?? null,
       config,
     };
   }
 
+  /** Resolve the current model's catalog entry whose user-facing name is Fast.
+   * Codex 0.145 deliberately selects by tier name and sends the catalog's id
+   * (currently `priority` for OpenAI models), so callers must never hardcode
+   * `fast` as the protocol value. */
+  async resolveFastServiceTier(
+    model = this.opts.model,
+    opts?: { signal?: AbortSignal },
+  ): Promise<{ model: string; serviceTier: string } | undefined> {
+    let cursor: string | null | undefined;
+    const models: Json[] = [];
+    do {
+      const result = await this.request('model/list', {
+        cursor: cursor ?? null,
+        limit: 100,
+        includeHidden: true,
+      }, { signal: opts?.signal });
+      if (Array.isArray(result?.data)) models.push(...result.data);
+      cursor = typeof result?.nextCursor === 'string' && result.nextCursor
+        ? result.nextCursor
+        : null;
+    } while (cursor);
+
+    const requested = model?.trim();
+    const selected = requested
+      ? models.find(entry => entry?.id === requested || entry?.model === requested)
+      : models.find(entry => entry?.isDefault === true);
+    if (!selected) return undefined;
+    const fast = Array.isArray(selected.serviceTiers)
+      ? selected.serviceTiers.find((tier: Json) =>
+        typeof tier?.name === 'string' && tier.name.toLowerCase() === 'fast')
+      : undefined;
+    if (!fast || typeof fast.id !== 'string' || !fast.id) return undefined;
+    return {
+      model: String(selected.model ?? selected.id),
+      serviceTier: fast.id,
+    };
+  }
+
+  private async prepareInitialServiceTier(): Promise<void> {
+    if (!this.opts.fastMode) {
+      this.serviceTier = undefined;
+      return;
+    }
+    if (this.serviceTier) return;
+    const resolved = await this.resolveFastServiceTier();
+    if (!resolved) {
+      throw new Error(`Fast Mode is not supported by model ${this.opts.model ?? '(default)'}`);
+    }
+    this.serviceTier = resolved.serviceTier;
+  }
+
+  /** Change the loaded thread's tier for subsequent turns. State is committed
+   * only after app-server acknowledges `thread/settings/update`, so the daemon
+   * can persist exactly what the executor accepted. */
+  async setFastMode(
+    enabled: boolean,
+    opts?: { signal?: AbortSignal },
+  ): Promise<{ enabled: boolean; serviceTier?: string }> {
+    if (!this.threadId) throw new Error('setFastMode before startThread/resumeThread');
+    const previousTier = this.serviceTier;
+    let nextTier: string | undefined;
+    let settingsUpdateDispatched = false;
+    try {
+      if (opts?.signal?.aborted) throw new CodexRpcRequestAbortedError('thread/settings/update');
+      if (enabled) {
+        const resolved = await this.resolveFastServiceTier(this.opts.model, opts);
+        if (!resolved) {
+          throw new Error(`Fast Mode is not supported by model ${this.opts.model ?? '(default)'}`);
+        }
+        nextTier = resolved.serviceTier;
+      }
+      if (opts?.signal?.aborted) throw new CodexRpcRequestAbortedError('thread/settings/update');
+      await this.request('thread/settings/update', {
+        threadId: this.threadId,
+        serviceTier: nextTier ?? null,
+      }, { signal: opts?.signal }, () => {
+        settingsUpdateDispatched = true;
+      });
+      this.serviceTier = nextTier;
+      if (opts?.signal?.aborted) throw new CodexRpcRequestAbortedError('thread/settings/update');
+      return nextTier
+        ? { enabled: true, serviceTier: nextTier }
+        : { enabled: false };
+    } catch (error) {
+      const cancelled = opts?.signal?.aborted || error instanceof CodexRpcRequestAbortedError;
+      if (!cancelled) throw error;
+
+      // Once the settings frame is on the socket, cancellation is ambiguous:
+      // app-server may have accepted it even if its ACK has not arrived. Send
+      // the previous tier on the same ordered connection and wait for that ACK
+      // before reporting cancellation to the worker.
+      if (settingsUpdateDispatched) {
+        try {
+          await this.restoreFastModeServiceTier(previousTier);
+        } catch (rollbackError) {
+          const message = rollbackError instanceof Error
+            ? rollbackError.message
+            : String(rollbackError);
+          throw new CodexRpcFastModeCancelledError(
+            `Codex Fast Mode update was cancelled, but restoring the previous tier failed: ${message}`,
+          );
+        }
+      }
+      throw new CodexRpcFastModeCancelledError();
+    }
+  }
+
+  /** Restore an already-confirmed tier after a cancelled runtime update. */
+  async restoreFastModeServiceTier(serviceTier?: string): Promise<void> {
+    if (!this.threadId) throw new Error('restoreFastModeServiceTier before startThread/resumeThread');
+    try {
+      await this.request('thread/settings/update', {
+        threadId: this.threadId,
+        serviceTier: serviceTier ?? null,
+      });
+      this.serviceTier = serviceTier;
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+      this.failAll(new Error(`Fast Mode rollback failed: ${message}`));
+      throw error;
+    }
+  }
+
   /** Inject one user message as a turn. Resolves when the app-server acks the
    *  turn start (fast); the turn itself streams to the attached TUI.
    *  `clientUserMessageId` (a stable botmux turn id) is forwarded so codex can
@@ -228,6 +373,7 @@ export class CodexRpcEngine {
       cwd: this.opts.cwd,
       approvalPolicy: 'never',
       sandboxPolicy: { type: 'dangerFullAccess' },
+      serviceTier: this.serviceTier ?? null,
     };
     if (clientUserMessageId) params.clientUserMessageId = clientUserMessageId;
     await this.request('turn/start', params, opts);
@@ -314,6 +460,7 @@ export class CodexRpcEngine {
       cwd: this.opts.cwd,
       approvalPolicy: 'never',
       sandboxPolicy: { type: 'dangerFullAccess' },
+      serviceTier: this.serviceTier ?? null,
     };
     if (clientUserMessageId) params.clientUserMessageId = clientUserMessageId;
     try {
@@ -452,13 +599,33 @@ export class CodexRpcEngine {
   private request(
     method: string,
     params: unknown,
-    opts?: { timeoutMs?: number; fatalOnTimeout?: boolean },
+    opts?: { timeoutMs?: number; fatalOnTimeout?: boolean; signal?: AbortSignal },
     onDispatch?: () => void,
   ): Promise<any> {
     const timeoutMs = opts?.timeoutMs ?? this.opts.requestTimeoutMs ?? REQUEST_TIMEOUT_MS;
     const fatalOnTimeout = opts?.fatalOnTimeout !== false; // default fatal
     const id = this.nextId++;
+    if (opts?.signal?.aborted) {
+      return Promise.reject(new CodexRpcRequestAbortedError(method));
+    }
     return new Promise((resolve, reject) => {
+      const cleanupAbort = (): void => {
+        opts?.signal?.removeEventListener('abort', onAbort);
+      };
+      const resolveRequest = (value: unknown): void => {
+        cleanupAbort();
+        resolve(value);
+      };
+      const rejectRequest = (error: Error): void => {
+        cleanupAbort();
+        reject(error);
+      };
+      const onAbort = (): void => {
+        if (!this.pending.has(id)) return;
+        this.pending.delete(id);
+        clearTimeout(timer);
+        rejectRequest(new CodexRpcRequestAbortedError(method));
+      };
       const timer = setTimeout(() => {
         if (!this.pending.has(id)) return;
         const err = new Error(`codex app-server request '${method}' timed out after ${timeoutMs}ms`);
@@ -473,16 +640,21 @@ export class CodexRpcEngine {
           // Non-fatal (the fresh first turn): reject only THIS request and keep
           // the engine alive, so its ambiguity can be resolved against rollout
           // persistence and the viewer can still resume if the turn landed (P1-1).
-          this.pending.delete(id); reject(err);
+          this.pending.delete(id); rejectRequest(err);
         }
       }, timeoutMs);
       timer.unref?.();
-      this.pending.set(id, { resolve, reject, timer });
+      this.pending.set(id, { resolve: resolveRequest, reject: rejectRequest, timer });
+      opts?.signal?.addEventListener('abort', onAbort, { once: true });
       // onDispatch fires ONLY after send() succeeds (ws was OPEN + no throw) — the
       // frame is then on the socket, so any later failure is "dispatched" and must
       // be treated as ambiguous, never not-sent (Codex P1-1 boundary).
       try { this.send({ jsonrpc: '2.0', id, method, params }); onDispatch?.(); }
-      catch (e) { this.pending.delete(id); clearTimeout(timer); reject(e as Error); }
+      catch (e) {
+        this.pending.delete(id);
+        clearTimeout(timer);
+        rejectRequest(e as Error);
+      }
     });
   }