diff --git a/CONTEXT.md b/CONTEXT.md index 8b15cfb1e..0b065cf1f 100644 --- a/CONTEXT.md +++ b/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 diff --git a/src/adapters/backend/reproduce-command.ts b/src/adapters/backend/reproduce-command.ts index ab9b01bfd..48870424f 100644 --- a/src/adapters/backend/reproduce-command.ts +++ b/src/adapters/backend/reproduce-command.ts @@ -39,6 +39,7 @@ 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) { @@ -46,7 +47,10 @@ export function selectReproduceLaunch(input: { 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 }; } diff --git a/src/adapters/cli/codex.ts b/src/adapters/cli/codex.ts index 002c1ac11..3a270780d 100644 --- a/src/adapters/cli/codex.ts +++ b/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 diff --git a/src/adapters/cli/types.ts b/src/adapters/cli/types.ts index 365572831..3ad89a42c 100644 --- a/src/adapters/cli/types.ts +++ b/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'; diff --git a/src/codex-rpc-engine.ts b/src/codex-rpc-engine.ts index 1661ed379..1af4a0711 100644 --- a/src/codex-rpc-engine.ts +++ b/src/codex-rpc-engine.ts @@ -32,6 +32,20 @@ import { WebSocket } from 'ws'; type Json = Record; 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 { return new Promise((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 void; reject: (e: Error) => void; timer: ReturnType }>(); 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 { + 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 { + 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 { + 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 { + 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 { 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); + } }); } diff --git a/src/core/command-handler.ts b/src/core/command-handler.ts index f03d6487e..310eee392 100644 --- a/src/core/command-handler.ts +++ b/src/core/command-handler.ts @@ -85,10 +85,15 @@ import { readRoleProfileEntry, writeRoleProfileEntry, } from '../services/role-profile-store.js'; -import type { LarkMessage, DaemonToWorker } from '../types.js'; +import type { LarkMessage, DaemonToWorker, FastModeApplyResult } from '../types.js'; import { sessionKey, sessionAnchorId, markRepoCardConsumed, claimCurrentRepoCard } from './types.js'; import type { DaemonSession } from './types.js'; import { t, localeForBot, type Locale } from '../i18n/index.js'; +import { + fastModeSessionSupported, + fastModeStateNeedsReconciliation, + parseFastModeAction, +} from './fast-mode-control.js'; import { runSkillsImCommand } from './skills/im-command.js'; import { fetchDaemonIpc } from './daemon-ipc-auth.js'; import { updateSessionTitle } from './session-title.js'; @@ -133,16 +138,13 @@ export function formatSlashGroupName(name: string, prefix = ''): string { } /** - * Daemon commands that operate on an ALREADY-EXISTING session and must never - * pre-create one. `/rename` renames the current session — with no session there - * is nothing to rename, so the daemon routes must skip their generic - * "createSession + activeSessions.set(worker:null)" pre-create block and let - * handleCommand's `!ds` branch reply no_active_session. Without this, `/rename` - * in a brand-new topic (or a thread with no session) would spawn a phantom - * worker:null session just to rename it, polluting the dashboard. (Same class - * of fix as the `/card` / `/term` special cases in daemon.ts.) + * Commands that normally operate on an ALREADY-EXISTING session and must not + * pass through the generic pre-create path. Without this guard, `/rename …` or + * `/fast status` in a new topic would leave a phantom worker:null session. + * daemon.ts has one deliberate exception: a valid Codex `/fast on` may create + * a session, but only after model-catalog preflight succeeds. */ -export const EXISTING_SESSION_ONLY_DAEMON_COMMANDS = new Set(['/rename']); +export const EXISTING_SESSION_ONLY_DAEMON_COMMANDS = new Set(['/rename', '/fast']); export function resolveAdapterDefaultPassthroughCommands(larkAppId?: string): string[] { if (!larkAppId) return []; @@ -391,6 +393,9 @@ export interface CommandHandlerDeps { lastRepoScan: Map; /** 会前预热文档评论会话：立即启动 CLI、读取文档并进入待命。 */ prewarmDocCommentSession?: (ds: DaemonSession, sub: DocSubscription) => Promise; + /** Apply Fast Mode at the real executor boundary. The implementation waits + * for a worker/app-server ACK and returns the concrete model-catalog tier. */ + applyFastMode?: (ds: DaemonSession, enabled: boolean) => Promise; } // ─── Schedule command ──────────────────────────────────────────────────────── @@ -1911,6 +1916,99 @@ export async function handleCommand( break; } + case '/fast': { + const action = parseFastModeAction(message.content); + if (action === 'invalid') { + await sessionReply(rootId, t('cmd.fast.usage', undefined, loc)); + break; + } + if (!ds) { + await sessionReply(rootId, t('cmd.no_active_session', undefined, loc)); + break; + } + const botConfig = getBot(ds.larkAppId).config; + const sessionCliId = ds.session.cliId ?? botConfig.cliId; + const wrapperCli = ds.session.agentFrozen + ? ds.session.wrapperCli + : (ds.session.wrapperCli ?? botConfig.wrapperCli); + const backendType = resolvePairedSpawnBackendType( + sessionCliId, + ds.session.backendType, + botConfig.backendType, + config.daemon.backendType, + ); + // Aiden rejects every Codex config override, so Botmux cannot pin the + // service tier on cold start and therefore cannot uphold Session-local + // semantics. Other known wrappers either rewrite or pass the override. + if (!fastModeSessionSupported({ + cliId: sessionCliId, + wrapperCli, + adopted: !!ds.adoptedFrom, + backendType, + })) { + await sessionReply(rootId, t('cmd.fast.unsupported', undefined, loc)); + break; + } + + const current = ds.session.fastMode === true; + if (action === 'status') { + await sessionReply(rootId, t(current ? 'cmd.fast.on' : 'cmd.fast.off', undefined, loc)); + break; + } + + const enabled = action === 'on' || (action === 'toggle' && !current); + // A legacy `fastMode:true` record without a catalog tier is not proof + // that the executor actually applied Fast Mode. Re-apply it through the + // acknowledged path so the persisted state self-heals. + const needsApply = enabled !== current || fastModeStateNeedsReconciliation({ + enabled, + serviceTier: ds.session.fastServiceTier, + stateVersion: ds.session.fastModeStateVersion, + }); + if (needsApply) { + const executorWasLive = !!ds.worker && !ds.worker.killed; + const applied = deps.applyFastMode + ? await deps.applyFastMode(ds, enabled) + : ({ ok: false, reason: 'apply_failed' } as const); + if (!applied.ok || applied.enabled !== enabled) { + const key = !applied.ok && applied.reason === 'unsupported_model' + ? 'cmd.fast.unsupported_model' + : 'cmd.fast.apply_failed'; + await sessionReply(rootId, t(key, undefined, loc)); + break; + } + + // Commit only after the executor ACKs. Keep both persisted Session + // state and the daemon's in-memory restart snapshot coherent. + ds.session.fastMode = enabled; + ds.session.fastServiceTier = enabled ? applied.serviceTier : undefined; + // A capability probe can stage a cold Session, but it is not an + // executor ACK. Leave the marker absent so init reconciles before + // trusting any persistent pane. A live worker publishes version 1 + // atomically through its exact fast_mode_result before the waiter resolves. + if (!executorWasLive) ds.session.fastModeStateVersion = undefined; + if (ds.initConfig) { + ds.initConfig.fastMode = enabled; + ds.initConfig.fastServiceTier = enabled ? applied.serviceTier : undefined; + if (!executorWasLive) ds.initConfig.fastModeStateVersion = undefined; + } + sessionStore.updateSession(ds.session); + } else if (ds.session.fastMode === undefined) { + // Materialize the default on first use so resumes no longer depend on + // Codex's process-global persisted service tier. + ds.session.fastMode = enabled; + ds.session.fastServiceTier = undefined; + if (ds.initConfig) { + ds.initConfig.fastMode = enabled; + ds.initConfig.fastServiceTier = undefined; + } + sessionStore.updateSession(ds.session); + } + await sessionReply(rootId, t(enabled ? 'cmd.fast.on' : 'cmd.fast.off', undefined, loc)); + logger.info(`[${logTag}] Fast Mode ${enabled ? 'enabled' : 'disabled'} for Session`); + break; + } + case '/schedule': { const scheduleArgs = message.content.replace(/^\/schedule\s*/, ''); const chatId = ds?.chatId!; @@ -3342,6 +3440,7 @@ export async function handleCommand( t('help.repo_wt', undefined, loc), t('help.rename', undefined, loc), t('help.status', undefined, loc), + t('help.fast', undefined, loc), t('help.card', undefined, loc), t('help.term', undefined, loc), t('help.dashboard', undefined, loc), diff --git a/src/core/fast-mode-apply-transaction.ts b/src/core/fast-mode-apply-transaction.ts new file mode 100644 index 000000000..3d5157b99 --- /dev/null +++ b/src/core/fast-mode-apply-transaction.ts @@ -0,0 +1,55 @@ +export type FastModeCancellationSource = 'worker_deadline' | 'daemon_timeout'; + +export class FastModeApplyCancelledError extends Error { + constructor( + readonly requestId: string, + readonly source: FastModeCancellationSource, + ) { + super(`Fast Mode apply transaction ${requestId} was cancelled (${source})`); + this.name = 'FastModeApplyCancelledError'; + } +} + +/** + * Lifetime guard for one exact Fast Mode request. + * + * The worker creates this object immediately after dequeuing the request. Every + * asynchronous result must pass through waitFor() before it can mutate executor + * or persisted state, so cancellation covers the applying window as well as the + * later native prompt-ACK window. + */ +export class FastModeApplyTransaction { + private readonly abortController = new AbortController(); + private cancellationSource?: FastModeCancellationSource; + + constructor(readonly requestId: string) {} + + get signal(): AbortSignal { + return this.abortController.signal; + } + + get cancelledBy(): FastModeCancellationSource | undefined { + return this.cancellationSource; + } + + cancel(requestId: string, source: FastModeCancellationSource): boolean { + if (requestId !== this.requestId) return false; + if (!this.cancellationSource) { + this.cancellationSource = source; + this.abortController.abort(); + } + return true; + } + + assertActive(): void { + if (this.cancellationSource) { + throw new FastModeApplyCancelledError(this.requestId, this.cancellationSource); + } + } + + async waitFor(operation: Promise): Promise { + const result = await operation; + this.assertActive(); + return result; + } +} diff --git a/src/core/fast-mode-control.ts b/src/core/fast-mode-control.ts new file mode 100644 index 000000000..f4dd5dfef --- /dev/null +++ b/src/core/fast-mode-control.ts @@ -0,0 +1,118 @@ +import { randomUUID } from 'node:crypto'; +import type { ChildProcess } from 'node:child_process'; +import { CodexRpcEngine } from '../codex-rpc-engine.js'; +import type { CliId } from '../adapters/cli/types.js'; +import type { BackendType } from '../adapters/backend/types.js'; +import type { DaemonToWorker, FastModeApplyResult } from '../types.js'; +import { + cancelFastModeResult, + waitForFastModeResult, +} from './fast-mode-handshake.js'; + +export type FastModeAction = 'toggle' | 'on' | 'off' | 'status' | 'invalid'; + +/** A persisted Session state is trustworthy only after the executor confirmed + * that exact ON/OFF target. Enabled state additionally requires the concrete + * model-catalog tier used to launch future native processes. */ +export function fastModeStateNeedsReconciliation(input: { + enabled: boolean; + serviceTier?: string; + stateVersion?: 1; +}): boolean { + return input.stateVersion !== 1 || (input.enabled && !input.serviceTier); +} + +/** Parse the public `/fast` surface once so daemon routing and the command + * handler cannot disagree about which invocations may create a Session. */ +export function parseFastModeAction(content: string): FastModeAction { + const match = content.trim().match(/^\/fast(?:\s+(.*))?$/i); + if (!match) return 'invalid'; + const action = (match[1] ?? '').trim().toLowerCase(); + if (!action) return 'toggle'; + if (action === 'on' || action === 'off' || action === 'status') return action; + return 'invalid'; +} + +/** Static capability gate. Model-level support is deliberately separate and + * comes from app-server model/list. */ +export function fastModeSessionSupported(input: { + cliId: CliId; + wrapperCli?: string; + adopted?: boolean; + backendType?: BackendType; +}): boolean { + if (input.cliId !== 'codex' || input.adopted || input.backendType === 'riff') return false; + return input.wrapperCli?.trim().split(/\s+/)[0] !== 'aiden'; +} + +/** Query the same Codex app-server catalog used by the executor. This is used + * before a cold `/fast on` creates a Session and as a migration fallback for + * old Fast sessions that did not persist a concrete tier id. */ +export async function probeCodexFastServiceTier(input: { + cliBin: string; + cwd: string; + env: NodeJS.ProcessEnv; + model?: string; + log?: (message: string) => void; +}): Promise { + const engine = new CodexRpcEngine({ + cliBin: input.cliBin, + cwd: input.cwd, + env: input.env, + model: input.model, + log: input.log, + }); + try { + await engine.start(); + const resolved = await engine.resolveFastServiceTier(input.model); + if (!resolved) return { ok: false, reason: 'unsupported_model' }; + return { + ok: true, + enabled: true, + serviceTier: resolved.serviceTier, + }; + } catch (error) { + return { + ok: false, + reason: 'apply_failed', + message: error instanceof Error ? error.message : String(error), + }; + } finally { + engine.stop(); + } +} + +/** Send a typed state-change request and wait for the executor ACK. Merely + * handing bytes to child_process.send is not success. */ +export async function requestWorkerFastModeChange( + worker: ChildProcess, + enabled: boolean, + timeoutMs = 120_000, +): Promise { + if (worker.killed || worker.connected === false) { + return { ok: false, reason: 'not_ready' }; + } + const requestId = randomUUID(); + const result = waitForFastModeResult(requestId, timeoutMs, () => { + if (worker.killed || worker.connected === false) return; + try { + worker.send({ + type: 'cancel_fast_mode', + requestId, + } satisfies DaemonToWorker); + } catch { + // The daemon-side waiter is already resolving fail-closed. + } + }); + try { + worker.send( + { type: 'set_fast_mode', requestId, enabled } satisfies DaemonToWorker, + error => { + if (error) cancelFastModeResult(requestId); + }, + ); + } catch { + cancelFastModeResult(requestId); + } + return result; +} diff --git a/src/core/fast-mode-handshake.ts b/src/core/fast-mode-handshake.ts new file mode 100644 index 000000000..819ab346f --- /dev/null +++ b/src/core/fast-mode-handshake.ts @@ -0,0 +1,67 @@ +import type { FastModeApplyResult, WorkerToDaemon } from '../types.js'; + +type PendingFastModeResult = { + resolve: (result: FastModeApplyResult) => void; + timer: ReturnType; +}; + +const pendingResults = new Map(); + +/** Read-only exact-request fence for executor-state persistence. */ +export function isFastModeResultPending(requestId: string): boolean { + return pendingResults.has(requestId); +} + +/** Register before sending set_fast_mode. Timeout is deliberately fail-closed: + * the daemon must never persist or report a state the worker did not ACK. */ +export function waitForFastModeResult( + requestId: string, + timeoutMs: number, + onTimeout?: () => void, +): Promise { + return new Promise(resolve => { + const previous = pendingResults.get(requestId); + if (previous) { + clearTimeout(previous.timer); + previous.resolve({ ok: false, reason: 'not_ready' }); + } + const timer = setTimeout(() => { + pendingResults.delete(requestId); + try { onTimeout?.(); } catch { /* best-effort worker cancellation */ } + resolve({ ok: false, reason: 'not_ready' }); + }, timeoutMs); + pendingResults.set(requestId, { resolve, timer }); + }); +} + +/** Resolve an exact worker ACK. Unknown, duplicate, and stale ids are harmless. */ +export function acknowledgeFastModeResult( + message: Extract, +): boolean { + const pending = pendingResults.get(message.requestId); + if (!pending) return false; + pendingResults.delete(message.requestId); + clearTimeout(pending.timer); + pending.resolve(message.ok + ? { + ok: true, + enabled: message.enabled, + ...(message.serviceTier ? { serviceTier: message.serviceTier } : {}), + } + : { + ok: false, + reason: message.reason, + ...(message.message ? { message: message.message } : {}), + }); + return true; +} + +/** Fail immediately when IPC send itself failed. */ +export function cancelFastModeResult(requestId: string): boolean { + const pending = pendingResults.get(requestId); + if (!pending) return false; + pendingResults.delete(requestId); + clearTimeout(pending.timer); + pending.resolve({ ok: false, reason: 'not_ready' }); + return true; +} diff --git a/src/core/fast-mode-restart-watchdog.ts b/src/core/fast-mode-restart-watchdog.ts new file mode 100644 index 000000000..592f5b287 --- /dev/null +++ b/src/core/fast-mode-restart-watchdog.ts @@ -0,0 +1,46 @@ +export const FAST_MODE_RESTART_ACK_TIMEOUT_MS = 110_000; + +type ActiveWatchdog = { + requestId: string; + timer: ReturnType; +}; + +/** One Fast apply transaction may be active at a time. The worker arms this at + * dequeue, before native probing or RPC work; the request id fences stale + * clears/cancels so transaction N cannot release transaction N+1. */ +export class FastModeRestartWatchdog { + private active?: ActiveWatchdog; + + constructor( + private readonly timeoutMs = FAST_MODE_RESTART_ACK_TIMEOUT_MS, + ) {} + + arm( + requestId: string, + onTimeout: (requestId: string) => void | Promise, + ): void { + if (this.active) { + throw new Error(`Fast Mode restart watchdog already armed for ${this.active.requestId}`); + } + const timer = setTimeout(() => { + if (this.active?.requestId !== requestId) return; + this.active = undefined; + void onTimeout(requestId); + }, this.timeoutMs); + timer.unref?.(); + this.active = { requestId, timer }; + } + + clear(requestId: string): boolean { + if (this.active?.requestId !== requestId) return false; + clearTimeout(this.active.timer); + this.active = undefined; + return true; + } + + dispose(): void { + if (!this.active) return; + clearTimeout(this.active.timer); + this.active = undefined; + } +} diff --git a/src/core/passthrough-commands.ts b/src/core/passthrough-commands.ts index df1a935c4..6028060a8 100644 --- a/src/core/passthrough-commands.ts +++ b/src/core/passthrough-commands.ts @@ -10,7 +10,7 @@ * chat) rather than relayed to the CLI. Used both for routing and to reject * `customPassthroughCommands` entries that would shadow a daemon command. */ -export const DAEMON_COMMANDS = new Set(['/close', '/restart', '/status', '/help', '/cd', '/repo', '/rename', '/schedule', '/role', '/botconfig', '/skills', '/pair', '/login', '/adopt', '/detach', '/disconnect', '/oncall', '/group', '/g', '/relay', '/card', '/term', '/list-slash-command', '/slash', '/subscribe-lark-doc', '/watch-comment', '/vc', '/insight', '/dashboard', '/vc-auth']); +export const DAEMON_COMMANDS = new Set(['/close', '/restart', '/status', '/fast', '/help', '/cd', '/repo', '/rename', '/schedule', '/role', '/botconfig', '/skills', '/pair', '/login', '/adopt', '/detach', '/disconnect', '/oncall', '/group', '/g', '/relay', '/card', '/term', '/list-slash-command', '/slash', '/subscribe-lark-doc', '/watch-comment', '/vc', '/insight', '/dashboard', '/vc-auth']); /** * Slash commands that are forwarded verbatim to the underlying CLI (e.g. diff --git a/src/core/worker-pool.ts b/src/core/worker-pool.ts index d7d3b1c2e..41dc5df63 100644 --- a/src/core/worker-pool.ts +++ b/src/core/worker-pool.ts @@ -111,6 +111,10 @@ import { extractBotmuxLarkNativeSessionTitlePrompt, } from './session-title.js'; import { acknowledgeSessionReady } from './session-ready-handshake.js'; +import { + acknowledgeFastModeResult, + isFastModeResultPending, +} from './fast-mode-handshake.js'; import { recordDispatchInputCommit } from './dispatch.js'; type WindowsForkOptions = ForkOptions & { windowsHide?: boolean }; @@ -2382,6 +2386,11 @@ export function forkWorker( wrapperCli: agentCfg.wrapperCli, launchShell: botCfg.launchShell, model: agentCfg.model, + // Fast Mode is frozen on the Session rather than inherited from Codex's + // CODEX_HOME. Missing means the Session has never opted in → standard. + fastMode: agentCfg.cliId === 'codex' ? ds.session.fastMode === true : undefined, + fastServiceTier: agentCfg.cliId === 'codex' ? ds.session.fastServiceTier : undefined, + fastModeStateVersion: agentCfg.cliId === 'codex' ? ds.session.fastModeStateVersion : undefined, reasoningEffort: agentCfg.reasoningEffort, disableCliBypass: botCfg.disableCliBypass === true, codexRpcInput: botCfg.codexRpcInput === true || config.codexRpcInputDefault, @@ -2714,6 +2723,58 @@ function setupWorkerHandlers( acknowledgeSessionReady(msg.requestId); break; } + case 'fast_mode_result': { + if ( + ds.worker !== worker + || ds.workerGeneration !== workerGeneration + || ds.session.workerGeneration !== workerGeneration + ) { + logger.warn(`[${t}] Ignored fast_mode_result from stale worker generation`); + break; + } + // Persistence and waiter resolution form one exact-request commit. A + // result arriving after the daemon deadline cannot mutate the Session, + // even when it came from the current worker generation. + if (!isFastModeResultPending(msg.requestId)) { + logger.warn(`[${t}] Ignored stale fast_mode_result request=${msg.requestId.slice(0, 8)}`); + break; + } + if (msg.ok) { + ds.session.fastMode = msg.enabled; + ds.session.fastServiceTier = msg.serviceTier; + ds.session.fastModeStateVersion = 1; + if (ds.initConfig) { + ds.initConfig.fastMode = msg.enabled; + ds.initConfig.fastServiceTier = msg.serviceTier; + ds.initConfig.fastModeStateVersion = 1; + } + sessionStore.updateSession(ds.session); + } + acknowledgeFastModeResult(msg); + break; + } + case 'fast_mode_state': { + if ( + ds.worker !== worker + || ds.workerGeneration !== workerGeneration + || ds.session.workerGeneration !== workerGeneration + ) { + logger.warn(`[${t}] Ignored fast_mode_state from stale worker generation`); + break; + } + // Only cold/legacy launch reconciliation has no command requestId. + // Runtime changes commit through the exact fast_mode_result case above. + ds.session.fastMode = msg.enabled; + ds.session.fastServiceTier = msg.serviceTier; + ds.session.fastModeStateVersion = 1; + if (ds.initConfig) { + ds.initConfig.fastMode = msg.enabled; + ds.initConfig.fastServiceTier = msg.serviceTier; + ds.initConfig.fastModeStateVersion = 1; + } + sessionStore.updateSession(ds.session); + break; + } case 'local_process_attestation': { // This message arrives over the private parent<->worker IPC channel; // unlike .botmux-cli-pids it cannot be forged or deleted by the CLI. diff --git a/src/daemon.ts b/src/daemon.ts index 59745574a..2f7b824a6 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -3,7 +3,7 @@ import { createHash, randomUUID } from 'node:crypto'; import { readFileSync, existsSync, mkdirSync, unlinkSync, watch, readdirSync } from 'node:fs'; import { atomicWriteFileSync } from './utils/atomic-write.js'; import { readAllowedUsersResolveCache, writeAllowedUsersResolveCache } from './utils/allowed-users-cache.js'; -import { join, dirname } from 'node:path'; +import { join, dirname, delimiter } from 'node:path'; import { homedir, loadavg, cpus, totalmem, freemem } from 'node:os'; import type { IncomingMessage, ServerResponse } from 'node:http'; import { fileURLToPath } from 'node:url'; @@ -108,6 +108,15 @@ import { isLocalCliOpenReady } from './services/local-cli-opener.js'; import { RECEIVED_REACTION_EMOJI_TYPE, SUBSTITUTE_RECEIVED_REACTION_EMOJI_TYPE } from './core/pending-response.js'; import { t as tr, botLocale, localeForBot } from './i18n/index.js'; import { createCliAdapterSync } from './adapters/cli/registry.js'; +import { sanitizePerBotEnv } from './core/per-bot-env.js'; +import { redactChildEnv } from './utils/child-env.js'; +import { + fastModeSessionSupported, + parseFastModeAction, + probeCodexFastServiceTier, + requestWorkerFastModeChange, +} from './core/fast-mode-control.js'; +import type { FastModeApplyResult } from './types.js'; import { initWorkerPool, setActiveSessionsRegistry, @@ -195,7 +204,7 @@ import { resolvePairedSpawnBackendType, type PersistentBackendType, } from './core/persistent-backend.js'; -import type { PersistentBackendTarget } from './adapters/backend/types.js'; +import type { BackendType, PersistentBackendTarget } from './adapters/backend/types.js'; import { handleCardAction, runAutoWorktreeCommit } from './im/lark/card-handler.js'; import type { CardActionData, CardHandlerDeps } from './im/lark/card-handler.js'; import { @@ -3801,6 +3810,100 @@ async function prewarmDocCommentSession(ds: DaemonSession, sub: DocSubscription) logger.info(`[${tag(ds)}] doc-comment watch prewarm injected file=${sub.fileToken.slice(0, 12)}`); } +function fastModeTargetConfig(larkAppId: string, ds?: DaemonSession): { + cliId: CliId; + cliPathOverride?: string; + wrapperCli?: string; + model?: string; + backendType: BackendType; + env?: Record; +} { + const botCfg = getBot(larkAppId).config; + const frozen = ds?.session.agentFrozen === true; + const cliId = ds?.session.cliId ?? botCfg.cliId; + return { + cliId, + cliPathOverride: frozen + ? ds?.session.cliPathOverride + : (ds?.session.cliPathOverride ?? botCfg.cliPathOverride), + wrapperCli: frozen + ? ds?.session.wrapperCli + : (ds?.session.wrapperCli ?? botCfg.wrapperCli), + model: frozen + ? ds?.session.model + : (ds?.session.model ?? botCfg.model), + backendType: resolvePairedSpawnBackendType( + cliId, + ds?.session.backendType, + botCfg.backendType, + config.daemon.backendType, + ), + env: botCfg.env, + }; +} + +async function probeFastModeForTarget( + larkAppId: string, + workingDir: string, + ds?: DaemonSession, +): Promise { + const target = fastModeTargetConfig(larkAppId, ds); + if (!fastModeSessionSupported({ + cliId: target.cliId, + wrapperCli: target.wrapperCli, + adopted: !!ds?.adoptedFrom, + backendType: target.backendType, + })) { + return { ok: false, reason: 'unsupported_session' }; + } + const env: NodeJS.ProcessEnv = { + ...redactChildEnv(process.env), + ...sanitizePerBotEnv(target.env), + }; + env.PATH = `${join(homedir(), '.botmux', 'bin')}${delimiter}${env.PATH ?? ''}`; + const cliBin = createCliAdapterSync(target.cliId, target.cliPathOverride).resolvedBin; + return probeCodexFastServiceTier({ + cliBin, + cwd: workingDir, + env, + model: target.model, + log: message => logger.info(`[fast-probe:${larkAppId}] ${message}`), + }); +} + +async function applyFastModeForSession( + ds: DaemonSession, + enabled: boolean, +): Promise { + const target = fastModeTargetConfig(ds.larkAppId, ds); + if (!fastModeSessionSupported({ + cliId: target.cliId, + wrapperCli: target.wrapperCli, + adopted: !!ds.adoptedFrom, + backendType: target.backendType, + })) { + return { ok: false, reason: 'unsupported_session' }; + } + if (ds.worker && !ds.worker.killed) { + return requestWorkerFastModeChange(ds.worker, enabled); + } + if (!enabled) return { ok: true, enabled: false }; + if (ds.session.fastServiceTier) { + return { + ok: true, + enabled: true, + serviceTier: ds.session.fastServiceTier, + }; + } + return probeFastModeForTarget(ds.larkAppId, getSessionWorkingDir(ds), ds); +} + +function fastModeFailureReplyKey(result: FastModeApplyResult): string { + if (!result.ok && result.reason === 'unsupported_session') return 'cmd.fast.unsupported'; + if (!result.ok && result.reason === 'unsupported_model') return 'cmd.fast.unsupported_model'; + return 'cmd.fast.apply_failed'; +} + // Dependencies passed to command-handler const commandDeps: CommandHandlerDeps = { activeSessions, @@ -3808,6 +3911,7 @@ const commandDeps: CommandHandlerDeps = { getActiveCount, lastRepoScan, prewarmDocCommentSession, + applyFastMode: applyFastModeForSession, }; /** @@ -15267,6 +15371,47 @@ async function handleNewTopic(data: any, ctx: RoutingContext): Promise { fireSessionlessCommandDetached(cmd, anchor, { ...parsed, content: commandContent, chatId }, larkAppId); return; } + let coldFastServiceTier: string | undefined; + let coldFastWorkingDir: string | undefined; + if (cmd === '/fast') { + const action = parseFastModeAction(commandContent); + // Only a valid enable request may open a Session. Status/off/invalid + // are routed session-less so they can report their result without + // polluting the dashboard. + if (action !== 'toggle' && action !== 'on') { + await handleCommand(cmd, anchor, { ...parsed, content: commandContent }, commandDeps, larkAppId); + return; + } + const target = fastModeTargetConfig(larkAppId); + if (!fastModeSessionSupported({ + cliId: target.cliId, + wrapperCli: target.wrapperCli, + backendType: target.backendType, + })) { + await sessionReply(anchor, tr('cmd.fast.unsupported', undefined, localeForBot(larkAppId)), 'text', larkAppId); + return; + } + const { pinnedWorkingDir } = await resolvePinnedWorkingDir({ + scope, + anchor, + chatId, + chatType, + larkAppId, + }); + coldFastWorkingDir = pinnedWorkingDir + ?? expandHome(effectiveDefaultWorkingDir(botCfg) ?? botCfg.workingDir ?? config.daemon.workingDir); + const preflight = await probeFastModeForTarget(larkAppId, coldFastWorkingDir); + if (!preflight.ok || !preflight.serviceTier) { + await sessionReply( + anchor, + tr(fastModeFailureReplyKey(preflight), undefined, localeForBot(larkAppId)), + 'text', + larkAppId, + ); + return; + } + coldFastServiceTier = preflight.serviceTier; + } // `/rename` renames an EXISTING session; a brand-new topic has none. Route // straight to handleCommand (its `!ds` branch replies no_active_session) // so the pre-create block below doesn't spawn a worker:null phantom @@ -15274,7 +15419,7 @@ async function handleNewTopic(data: any, ctx: RoutingContext): Promise { // and /term special cases, but UNLIKE those (which carry their own // permission gates inside their handlers) this branch MUST stay after // the canOperate gate above — the /rename handler itself has no gate. - if (EXISTING_SESSION_ONLY_DAEMON_COMMANDS.has(cmd)) { + if (EXISTING_SESSION_ONLY_DAEMON_COMMANDS.has(cmd) && cmd !== '/fast') { await handleCommand(cmd, anchor, { ...parsed, content: commandContent }, commandDeps, larkAppId); return; } @@ -15298,6 +15443,10 @@ async function handleNewTopic(data: any, ctx: RoutingContext): Promise { session.lastCallerOpenId = senderOpenId; session.lastMessageAt = new Date(now).toISOString(); session.scope = scope; + if (coldFastServiceTier) { + session.fastServiceTier = coldFastServiceTier; + session.workingDir = coldFastWorkingDir; + } // First-message `/repo`: seed the same pending-repo state the card flow // uses, so the `/repo` handler launches the CLI straight away — @@ -15329,6 +15478,7 @@ async function handleNewTopic(data: any, ctx: RoutingContext): Promise { lastMessageAt: now, hasHistory: false, ownerOpenId: senderOpenId, + ...(coldFastWorkingDir ? { workingDir: coldFastWorkingDir } : {}), ...cmdPending, }); // Pass mention-stripped content so /command argument parsing works. @@ -16109,6 +16259,51 @@ async function handleThreadReply(data: any, ctx: RoutingContext): Promise sessionReply(anchor, tr('daemon.cmd_allowed_users_only', { cmd }, localeForBot(larkAppId)), 'text', larkAppId); return; } + let coldFastServiceTier: string | undefined; + let coldFastWorkingDir: string | undefined; + if (!existingDs && cmd === '/fast') { + const action = parseFastModeAction(commandContent); + if (action !== 'toggle' && action !== 'on') { + await handleCommand( + cmd, + anchor, + { ...parsed, content: commandContent, chatId: threadChatId }, + commandDeps, + larkAppId, + ); + return; + } + const target = fastModeTargetConfig(larkAppId); + if (!fastModeSessionSupported({ + cliId: target.cliId, + wrapperCli: target.wrapperCli, + backendType: target.backendType, + })) { + await sessionReply(anchor, tr('cmd.fast.unsupported', undefined, localeForBot(larkAppId)), 'text', larkAppId); + return; + } + const { pinnedWorkingDir } = await resolvePinnedWorkingDir({ + scope, + anchor, + chatId: threadChatId, + chatType: ctxChatType, + larkAppId, + }); + const fastBotCfg = getBot(larkAppId).config; + coldFastWorkingDir = pinnedWorkingDir + ?? expandHome(effectiveDefaultWorkingDir(fastBotCfg) ?? fastBotCfg.workingDir ?? config.daemon.workingDir); + const preflight = await probeFastModeForTarget(larkAppId, coldFastWorkingDir); + if (!preflight.ok || !preflight.serviceTier) { + await sessionReply( + anchor, + tr(fastModeFailureReplyKey(preflight), undefined, localeForBot(larkAppId)), + 'text', + larkAppId, + ); + return; + } + coldFastServiceTier = preflight.serviceTier; + } // First message of a fresh thread carrying a session-needing daemon command // — e.g. another bot dispatched `/repo ` into a brand-new thread. // Without a session, handleCommand gets ds=undefined and `/repo` (and other @@ -16119,7 +16314,7 @@ async function handleThreadReply(data: any, ctx: RoutingContext): Promise // would be a phantom conversation that only exists to be renamed. Let // handleCommand's `!ds` branch reply no_active_session instead. if (!existingDs && threadChatId && !isSessionlessCommandInvocation(cmd, commandContent) - && !EXISTING_SESSION_ONLY_DAEMON_COMMANDS.has(cmd)) { + && (!EXISTING_SESSION_ONLY_DAEMON_COMMANDS.has(cmd) || cmd === '/fast')) { const session = sessionStore.createSession(threadChatId, anchor, cmdContent.substring(0, 50), ctxChatType); const now = Date.now(); if (ctxChatType === 'p2p') { @@ -16136,6 +16331,10 @@ async function handleThreadReply(data: any, ctx: RoutingContext): Promise session.lastCallerOpenId = threadSenderOpenId; session.lastMessageAt = new Date(now).toISOString(); session.scope = scope; + if (coldFastServiceTier) { + session.fastServiceTier = coldFastServiceTier; + session.workingDir = coldFastWorkingDir; + } let cmdPending: Partial | undefined; if (cmd === '/repo') { const { pinnedWorkingDir } = await resolvePinnedWorkingDir({ scope, anchor, chatId: threadChatId, chatType: ctxChatType, larkAppId }); @@ -16157,6 +16356,7 @@ async function handleThreadReply(data: any, ctx: RoutingContext): Promise lastMessageAt: now, hasHistory: false, ownerOpenId: threadSenderOpenId, + ...(coldFastWorkingDir ? { workingDir: coldFastWorkingDir } : {}), ...cmdPending, }); } diff --git a/src/i18n/en.ts b/src/i18n/en.ts index 77cd77080..d93c86799 100644 --- a/src/i18n/en.ts +++ b/src/i18n/en.ts @@ -272,6 +272,12 @@ export const messages: Record = { 'cmd.status.running': 'running', 'cmd.status.waiting': 'idle', 'cmd.status.fallback_no_session': 'No active session in this topic.\nDaemon active sessions: {count}\n{cliName}: v{version}', + 'cmd.fast.on': '✅ Fast Mode is ON for this Session only.', + 'cmd.fast.off': '⚪ Fast Mode is OFF for this Session only.', + 'cmd.fast.usage': 'Usage: /fast (toggle) | /fast on | /fast off | /fast status', + 'cmd.fast.unsupported': '⚠️ /fast requires a Botmux-managed Codex Session whose service tier can be pinned (Aiden gateways are not supported).', + 'cmd.fast.unsupported_model': '⚠️ The current model does not support Fast Mode. State was not changed.', + 'cmd.fast.apply_failed': '⚠️ Codex did not confirm the Fast Mode change. State was not changed; please retry.', 'cmd.login.no_credentials': '❌ Cannot read app credentials.', 'cmd.login.title': '🔐 Lark User OAuth', 'cmd.login.step1': '1. Click the link below to authorize:', @@ -494,6 +500,7 @@ export const messages: Record = { 'help.repo_wt': '/repo wt [branch] - Create a worktree off the remote default branch and open it (auto semantic name when branch is omitted)', 'help.rename': '/rename - Rename this Botmux session and sync the running Codex/Claude session name', 'help.status': '/status - Show current session status (incl. terminal URL)', + 'help.fast': '/fast [on|off|status] - Toggle or inspect Codex Fast Mode for this Session (off by default)', 'help.card': '/card - Manually post the streaming card for this session (summons it even when streaming is off, and resumes live updates; with private-card mode on, sends a static snapshot visible only to authorized users instead)', 'help.term': '/term - Get the operable (write-enabled) terminal link for this session, delivered privately to the owner (visible-to-you in-chat, falling back to DM in topic/p2p — never exposed in the group)', 'help.dashboard': '/dashboard [module] - Open Dashboard control cards in Feishu (sessions/schedules/groups/settings/help, etc.)', diff --git a/src/i18n/zh.ts b/src/i18n/zh.ts index e9e493450..c25588d8c 100644 --- a/src/i18n/zh.ts +++ b/src/i18n/zh.ts @@ -275,6 +275,12 @@ export const messages: Record<string, string> = { 'cmd.status.running': '运行中', 'cmd.status.waiting': '等待中', 'cmd.status.fallback_no_session': '当前话题没有活跃的会话。\nDaemon active sessions: {count}\n{cliName}: v{version}', + 'cmd.fast.on': '✅ Fast Mode 已开启（仅当前话题 Session）。', + 'cmd.fast.off': '⚪ Fast Mode 已关闭（仅当前话题 Session）。', + 'cmd.fast.usage': '用法：/fast（切换）| /fast on | /fast off | /fast status', + 'cmd.fast.unsupported': '⚠️ /fast 仅支持由 Botmux 管理且能固定服务等级的 Codex Session（Aiden 网关暂不支持）。', + 'cmd.fast.unsupported_model': '⚠️ 当前模型不支持 Fast Mode，状态未修改。', + 'cmd.fast.apply_failed': '⚠️ Fast Mode 未能在 Codex 执行端确认生效，状态未修改；请稍后重试。', 'cmd.login.no_credentials': '❌ 无法获取应用凭证', 'cmd.login.title': '🔐 飞书用户授权', 'cmd.login.step1': '1. 点击下方链接完成授权：', @@ -497,6 +503,7 @@ export const messages: Record<string, string> = { 'help.repo_wt': '/repo wt <编号|项目名> [分支] - 基于远端默认分支新建 worktree 并打开（未指定分支时自动语义命名）', 'help.rename': '/rename <标题> - 重命名当前 Botmux 会话，并同步运行中的 Codex/Claude 原生会话名', 'help.status': '/status - 查看当前会话状态（含终端链接）', + 'help.fast': '/fast [on|off|status] - 切换或查看当前话题 Session 的 Codex Fast Mode（默认关闭）', 'help.card': '/card - 手动弹出当前会话的流式卡片（关流式时也能临时召唤，并恢复实时刷新；开了私密卡片则改发仅授权人可见的静态快照）', 'help.term': '/term - 获取当前会话的「可操作终端」（带写权限）链接，私密发给 owner（群内仅你可见，话题/单聊回退私信，不在群里暴露）', 'help.dashboard': '/dashboard [模块] - 在飞书里打开 Dashboard 控制卡片（sessions/schedules/groups/settings/help 等）', diff --git a/src/setup/cli-selection.ts b/src/setup/cli-selection.ts index ab6fb070f..ffaeac27e 100644 --- a/src/setup/cli-selection.ts +++ b/src/setup/cli-selection.ts @@ -250,11 +250,20 @@ function isCjadkWrapper(tokens: ReadonlyArray<string>): boolean { } /** 是否为 botmux 自己给 Codex 注入的 config 覆盖。精确白名单避免误伤用户自带的 - * `-c key=val`；wrapper 层据此决定剥离（aiden）或改成长参数（cjadk）。 */ -function isBotmuxCodexConfigValue(value: string | undefined): boolean { + * `-c key=val`；wrapper 层据此决定剥离（aiden）或改成长参数（cjadk）。 + * + * Fast tier id 来自 Codex model catalog，不能静态枚举；调用方必须把本次实际注入的 + * tier 显式传入，不能按 `service_tier=*` 泛匹配，否则会误伤用户自己的 config。 */ +function isBotmuxCodexConfigValue(value: string | undefined, codexServiceTier?: string): boolean { + const injectedServiceTier = codexServiceTier + ? `service_tier=${JSON.stringify(codexServiceTier)}` + : undefined; return !!value && ( value.startsWith('shell_environment_policy.set.BOTMUX_') || value === 'check_for_update_on_startup=false' + || value === 'service_tier="default"' + || value === 'service_tier="fast"' + || value === injectedServiceTier ); } @@ -276,7 +285,7 @@ export function stripSettingsArgs(args: ReadonlyArray<string>): string[] { /** * 剥掉 aiden `aiden x <cli>` 网关拒收的、**botmux 注入的**底层 CLI config 覆盖参数： * - `--settings <v>` / `--settings=<v>`（claude 携带 hook/bypass，aiden x claude 历来就剥） - * - botmux 自己注入的 Codex `-c`（session 环境，以及关闭启动更新选择器）； + * - botmux 自己注入的 Codex `-c`（session 环境、关闭启动更新选择器、Session Fast Mode）； * aiden 1.8.38+ 会直接报错拒收 `aiden x codex` 透传的 `-c`/`--config`。 * 这些参数承载的 session 环境已在进程级 env（BOTMUX_SESSION_ID 等）注入、并被 wrapper * 子进程继承（见 worker.ts childEnv），故剥掉只是去掉一条冗余的 belt-and-suspenders @@ -294,13 +303,16 @@ export function stripSettingsArgs(args: ReadonlyArray<string>): string[] { * 注意范围仅限上述两种 botmux 注入形态：coco 的 `--config model.name=…`（承载 model、非 session * 管线，且无内置 `aiden x coco` 选项）等不同形态的 config 覆盖**有意不在此剥离**。 */ -export function stripWrapperUnsafeArgs(args: ReadonlyArray<string>): string[] { +export function stripWrapperUnsafeArgs( + args: ReadonlyArray<string>, + codexServiceTier?: string, +): string[] { // 先剥 claude 的 --settings（单一事实源 stripSettingsArgs），再剥 codex 注入的 -c override。 const afterSettings = stripSettingsArgs(args); const out: string[] = []; for (let i = 0; i < afterSettings.length; i++) { const a = afterSettings[i]!; - if (a === '-c' && isBotmuxCodexConfigValue(afterSettings[i + 1])) { i++; continue; } + if (a === '-c' && isBotmuxCodexConfigValue(afterSettings[i + 1], codexServiceTier)) { i++; continue; } out.push(a); } return out; @@ -324,11 +336,14 @@ export function stripWrapperUnsafeArgs(args: ReadonlyArray<string>): string[] { * 非 codex 的 cjadk 形态（cjadk claude 带 `--settings`、不带这些值）经此函数是 no-op， * `--settings` 照常保留透传。 */ -export function rewriteCjadkCodexConfigArgs(args: ReadonlyArray<string>): string[] { +export function rewriteCjadkCodexConfigArgs( + args: ReadonlyArray<string>, + codexServiceTier?: string, +): string[] { const out: string[] = []; for (let i = 0; i < args.length; i++) { const a = args[i]!; - if (a === '-c' && isBotmuxCodexConfigValue(args[i + 1])) { + if (a === '-c' && isBotmuxCodexConfigValue(args[i + 1], codexServiceTier)) { out.push('--config', args[i + 1]!); i++; continue; @@ -357,6 +372,11 @@ export interface WrappedLaunchOptions { * 用 {@link TTADK_DEFAULT_MODEL} 兜底；不接受 -m 的子命令（CoCo）忽略此项。 */ readonly ttadkModel?: string; + /** + * Botmux 本次给 Codex 注入的 service tier。只用于精确识别该一条 config， + * 以便 aiden 剥离或 cjadk 改写，不会匹配用户自带的其他 service_tier。 + */ + readonly codexServiceTier?: string; } /** @@ -422,8 +442,8 @@ export function buildWrappedLaunch( if (tokens.length === 0) return { bin: '', args: [...cliArgs] }; if (tokens[0] === 'ttadk') return buildTtadkLaunch(tokens, cliArgs, binResolver, opts.ttadkModel); let forwarded: string[]; - if (isAidenWrapper(tokens)) forwarded = stripWrapperUnsafeArgs(cliArgs); - else if (isCjadkWrapper(tokens)) forwarded = rewriteCjadkCodexConfigArgs(cliArgs); + if (isAidenWrapper(tokens)) forwarded = stripWrapperUnsafeArgs(cliArgs, opts.codexServiceTier); + else if (isCjadkWrapper(tokens)) forwarded = rewriteCjadkCodexConfigArgs(cliArgs, opts.codexServiceTier); else forwarded = [...cliArgs]; return { bin: binResolver(tokens[0]), args: [...tokens.slice(1), ...forwarded] }; } diff --git a/src/types.ts b/src/types.ts index f73e17a9a..9bf07cf0e 100644 --- a/src/types.ts +++ b/src/types.ts @@ -356,6 +356,17 @@ export interface Session { wrapperCli?: string; /** Optional model frozen at creation so historical sessions resume with their original model. */ model?: string; + /** Session-scoped Codex Fast Mode. Missing/false means standard service tier; + * true opts only this Session into the faster tier. */ + fastMode?: boolean; + /** Concrete app-server service-tier id resolved from the selected model's + * catalog (for example `priority`). Persisting the id keeps cold starts, + * restarts, and the native TUI on the same acknowledged setting. */ + fastServiceTier?: string; + /** Version of the executor-confirmed Fast Mode state. Missing on legacy or + * cold-staged records, which must be reconciled before a surviving native + * pane can be trusted. */ + fastModeStateVersion?: 1; /** Optional codex reasoning effort frozen at creation (per-turn API override). * Only meaningful for codex/codex-app; injected as model_reasoning_effort at spawn. */ reasoningEffort?: 'low' | 'medium' | 'high' | 'xhigh'; @@ -604,9 +615,19 @@ export interface CliTurnPayload { codexAppInput?: CodexAppTurnInput; } +export type FastModeFailureReason = + | 'unsupported_session' + | 'unsupported_model' + | 'not_ready' + | 'apply_failed'; + +export type FastModeApplyResult = + | { ok: true; enabled: boolean; serviceTier?: string } + | { ok: false; reason: FastModeFailureReason; message?: string }; + /** Messages sent from Daemon to Worker */ export type DaemonToWorker = - | { type: 'init'; sessionId: string; chatId: string; chatType?: 'group' | 'p2p'; rootMessageId: string; workingDir: string; cliId: string; cliPathOverride?: string; wrapperCli?: string; launchShell?: string; model?: string; reasoningEffort?: 'low' | 'medium' | 'high' | 'xhigh'; disableCliBypass?: boolean; codexRpcInput?: boolean; startupCommands?: string[]; env?: Record<string, string>; sandbox?: boolean; sandboxPaths?: { readWrite?: string[]; readOnly?: string[]; deny?: string[] }; sandboxHidePaths?: string[]; sandboxReadonlyPaths?: string[]; sandboxNetwork?: boolean; readIsolation?: boolean; readDenyExtraPaths?: string[]; daemonBootId?: string; backendType: BackendType; persistentBackendTarget?: PersistentBackendTarget; backendConfig?: RiffBackendConfig; riffParentTaskId?: string; riffRepoDirs?: string[]; deferredScheduleRun?: Session['deferredScheduleRun']; nativeSessionTitle?: string; nativeSessionTitlePrompt?: string; prompt: string; promptCodexAppInput?: CodexAppTurnInput; resume?: boolean; cliSessionId?: string; originalSessionId?: string; ownerOpenId?: string; webPort?: number; larkAppId: string; larkAppSecret: string; brand?: 'feishu' | 'lark'; botName?: string; botOpenId?: string; locale?: 'zh' | 'en'; turnId?: string; dispatchAttempt?: number; vcMeetingImTurnOrigin?: VcMeetingImTurnOrigin; pluginBindings?: string[]; skillPolicy?: BotSkillPolicy; skillPluginDir?: string; skillReadonlyRoots?: string[]; adoptMode?: boolean; adoptSource?: 'tmux' | 'herdr' | 'zellij'; adoptTmuxTarget?: string; adoptZellijSession?: string; adoptZellijPaneId?: string; adoptHerdrSessionName?: string; adoptHerdrTarget?: string; adoptHerdrPaneId?: string; adoptPaneCols?: number; adoptPaneRows?: number; bridgeJsonlPath?: string; adoptCliPid?: number; adoptCwd?: string; adoptRestoredFromMetadata?: boolean; runnerBuildId?: string; persistedRunnerBuildId?: string; restartAttemptId?: string } + | { type: 'init'; sessionId: string; chatId: string; chatType?: 'group' | 'p2p'; rootMessageId: string; workingDir: string; cliId: string; cliPathOverride?: string; wrapperCli?: string; launchShell?: string; model?: string; reasoningEffort?: 'low' | 'medium' | 'high' | 'xhigh'; fastMode?: boolean; fastServiceTier?: string; fastModeStateVersion?: 1; disableCliBypass?: boolean; codexRpcInput?: boolean; startupCommands?: string[]; env?: Record<string, string>; sandbox?: boolean; sandboxPaths?: { readWrite?: string[]; readOnly?: string[]; deny?: string[] }; sandboxHidePaths?: string[]; sandboxReadonlyPaths?: string[]; sandboxNetwork?: boolean; readIsolation?: boolean; readDenyExtraPaths?: string[]; daemonBootId?: string; backendType: BackendType; persistentBackendTarget?: PersistentBackendTarget; backendConfig?: RiffBackendConfig; riffParentTaskId?: string; riffRepoDirs?: string[]; deferredScheduleRun?: Session['deferredScheduleRun']; nativeSessionTitle?: string; nativeSessionTitlePrompt?: string; prompt: string; promptCodexAppInput?: CodexAppTurnInput; resume?: boolean; cliSessionId?: string; originalSessionId?: string; ownerOpenId?: string; webPort?: number; larkAppId: string; larkAppSecret: string; brand?: 'feishu' | 'lark'; botName?: string; botOpenId?: string; locale?: 'zh' | 'en'; turnId?: string; dispatchAttempt?: number; vcMeetingImTurnOrigin?: VcMeetingImTurnOrigin; pluginBindings?: string[]; skillPolicy?: BotSkillPolicy; skillPluginDir?: string; skillReadonlyRoots?: string[]; adoptMode?: boolean; adoptSource?: 'tmux' | 'herdr' | 'zellij'; adoptTmuxTarget?: string; adoptZellijSession?: string; adoptZellijPaneId?: string; adoptHerdrSessionName?: string; adoptHerdrTarget?: string; adoptHerdrPaneId?: string; adoptPaneCols?: number; adoptPaneRows?: number; bridgeJsonlPath?: string; adoptCliPid?: number; adoptCwd?: string; adoptRestoredFromMetadata?: boolean; runnerBuildId?: string; persistedRunnerBuildId?: string; restartAttemptId?: string } | { type: 'message'; content: string; codexAppInput?: CodexAppTurnInput; nativeSessionTitle?: string; nativeSessionTitlePrompt?: string; turnId?: string; dispatchAttempt?: number; vcMeetingImTurnOrigin?: VcMeetingImTurnOrigin } /** Literal slash-command passthrough. `followUpContent` rides along so the * worker enqueues it strictly AFTER the slash command's Enter — two separate @@ -614,6 +635,13 @@ export type DaemonToWorker = * raw_input branch awaits 200ms between sendText and Enter, a window where * a separate `message` IPC could write into the PTY first. */ | { type: 'raw_input'; content: string; turnId?: string; followUpContent?: string; followUpTurnId?: string; followUpCodexAppInput?: CodexAppTurnInput } + /** Apply Session-scoped Codex Fast Mode and ACK only after the executor has + * accepted it. Unlike raw_input this is queued across startup/restart gates + * and updates the worker's restart config before success is reported. */ + | { type: 'set_fast_mode'; requestId: string; enabled: boolean } + /** Cancel the exact queued or active Fast transaction after the daemon-side + * waiter expires. The active lifetime begins when the worker dequeues it. */ + | { type: 'cancel_fast_mode'; requestId: string } /** Rename the current CLI-native interactive session. The worker queues this * administrative slash command until the TUI is idle and does not treat it * as a model turn. Only adapters declaring buildSessionRenameCommand handle @@ -690,6 +718,11 @@ export type WorkerToDaemon = /** Worker 已处理 SessionStart 信号并建立 post-hook prompt evidence fence。 * daemon 收到后才结束 `botmux session-ready` HTTP 请求。 */ | { type: 'session_ready_ack'; requestId: string } + | { type: 'fast_mode_result'; requestId: string; ok: true; enabled: boolean; serviceTier?: string } + | { type: 'fast_mode_result'; requestId: string; ok: false; reason: FastModeFailureReason; message?: string } + /** Executor-confirmed state discovered during cold/legacy reconciliation. + * Runtime command state is committed through exact-request fast_mode_result. */ + | { type: 'fast_mode_state'; enabled: boolean; serviceTier?: string } | { type: 'screen_update'; content: string; status: ScreenStatus; usageLimit?: CliUsageLimitState; turnId?: string; dispatchAttempt?: number } | { type: 'error'; message: string; turnId?: string; dispatchAttempt?: number } | { type: 'bridge_source_session'; bridge: 'hermes'; sourceSessionId: string } diff --git a/src/worker.ts b/src/worker.ts index a208a2c84..a3b0f9ae0 100644 --- a/src/worker.ts +++ b/src/worker.ts @@ -215,7 +215,7 @@ import { DEVICE_AUTHORITY_DIRECTORY, DEVICE_CREDENTIAL_FILE, } from './platform/device-paths.js'; -import type { BackendType, SessionBackend } from './adapters/backend/types.js'; +import type { BackendType, PersistentBackendTarget, SessionBackend } from './adapters/backend/types.js'; import { tmuxEnv, probeTmuxFunctionalWithRetry } from './setup/ensure-tmux.js'; import { tmuxRestartJitterMs } from './core/tmux-recovery.js'; import { IdleDetector } from './utils/idle-detector.js'; @@ -262,6 +262,16 @@ import { replaceManagedOriginCapabilityFile, } from './core/managed-origin-capability.js'; import { CodexRpcEngine } from './codex-rpc-engine.js'; +import { + fastModeSessionSupported, + fastModeStateNeedsReconciliation, + probeCodexFastServiceTier, +} from './core/fast-mode-control.js'; +import { + FastModeApplyTransaction, + type FastModeCancellationSource, +} from './core/fast-mode-apply-transaction.js'; +import { FastModeRestartWatchdog } from './core/fast-mode-restart-watchdog.js'; // A worker must never trust an INHERITED session-level CLI home pointer // (CLAUDE_CONFIG_DIR / CODEX_HOME): a stale pm2 dump can resurrect the daemon @@ -371,6 +381,16 @@ async function killPersistentSessionVerified(backendType: PersistentBackendType, }); } +/** Kill the exact persisted backend resource (including an agent-scoped Herdr + * target) and fail closed unless probing confirms it is gone. */ +async function killPersistentBackendTargetVerified(target: PersistentBackendTarget): Promise<boolean> { + return killAndVerifyPersistentPane(target.sessionName, { + kill: () => killPersistentBackendTarget(target), + isLive: () => probePersistentBackendTarget(target) !== 'missing', + wait: delay, + }); +} + /** Poll the codex/traex rollout for POSITIVE evidence that the fresh first turn's * user message was persisted (P1-1). Used only to resolve a dispatched-but- * unacked first turn: a hit means the app-server accepted the turn → 'accepted' @@ -402,6 +422,46 @@ function codexNativeTitleEnv(cfg: Extract<DaemonToWorker, { type: 'init' }>): No return env; } +async function resolveWorkerFastServiceTier( + cfg: Extract<DaemonToWorker, { type: 'init' }>, +): Promise<Awaited<ReturnType<typeof probeCodexFastServiceTier>>> { + if (!fastModeSessionSupported({ + cliId: cfg.cliId as CliId, + wrapperCli: cfg.wrapperCli, + adopted: cfg.adoptMode, + backendType: cfg.backendType, + })) { + return { ok: false, reason: 'unsupported_session' }; + } + const cliBin = createCliAdapterSync(cfg.cliId as CliId, cfg.cliPathOverride).resolvedBin; + return probeCodexFastServiceTier({ + cliBin, + cwd: cfg.workingDir, + env: codexNativeTitleEnv(cfg), + model: cfg.model, + log: message => log(`[fast-probe] ${message}`), + }); +} + +/** Migrate pre-catalog Fast sessions before buildArgs. Failing closed here is + * preferable to launching a process that silently executes at default tier + * while the persisted Session says ON. */ +async function ensureConfiguredFastServiceTier( + cfg: Extract<DaemonToWorker, { type: 'init' }>, +): Promise<void> { + if (cfg.fastMode !== true || cfg.fastServiceTier) return; + const resolved = await resolveWorkerFastServiceTier(cfg); + if (!resolved.ok || !resolved.serviceTier) { + const detail = !resolved.ok && resolved.message ? `: ${resolved.message}` : ''; + throw new Error( + resolved.ok || resolved.reason === 'unsupported_model' + ? `Fast Mode is not supported by model ${cfg.model ?? '(default)'}` + : `Fast Mode capability check failed${detail}`, + ); + } + cfg.fastServiceTier = resolved.serviceTier; +} + function registerNativeTitleForceClose(forceClose: () => void): () => void { nativeSessionTitleSyncForceClosers.add(forceClose); return () => nativeSessionTitleSyncForceClosers.delete(forceClose); @@ -706,7 +766,10 @@ async function engageCodexRpc(cfg: Extract<DaemonToWorker, { type: 'init' }>): P Object.assign(engineEnv, sanitizePerBotEnv(cfg.env)); engine = new CodexRpcEngine({ cliBin, cwd: cfg.workingDir, env: engineEnv, sessionId: cfg.sessionId, - model: cfg.model, reasoningEffort: cfg.reasoningEffort, log: (m: string) => log(m), + model: cfg.model, + reasoningEffort: cfg.reasoningEffort, + fastMode: cfg.fastMode === true, + log: (m: string) => log(m), appServerFeatures: cfg.cliId === 'traex' ? ['default_mode_request_user_input'] : undefined, onRequestUserInput: cfg.cliId === 'traex' ? (params: unknown) => bridgeTraexUserInput(cfg, params) @@ -733,6 +796,7 @@ async function engageCodexRpc(cfg: Extract<DaemonToWorker, { type: 'init' }>): P // per-turn identity instead of falling back to a stale/session-only env. registerRpcEnginePidMarker(engine.appServerPid); const threadId = wantResume ? await engine.resumeThread(cfg.cliSessionId!) : await engine.startThread(); + cfg.fastServiceTier = engine.activeServiceTier; let outcome: EngageOutcome = wantResume ? 'resumed' : 'accepted'; if (!wantResume && cfg.prompt) { // Three-state delivery (P1-1, exactly-once priority): 'accepted' (ack or @@ -1530,6 +1594,35 @@ function shortCorrelationId(value: string | undefined): string { * an owned CLI restart was fenced. Normal raw_input commands are still * delivered immediately (including while busy). */ const pendingRawInputs = freshnessInputQueue.raw; +/** Fast Mode is a Session configuration barrier, not ordinary terminal input. + * Requests survive startup/restart gates and run before later user messages. */ +const pendingFastModeChanges: Array<Extract<DaemonToWorker, { type: 'set_fast_mode' }>> = []; +type ActiveFastModeChange = { + transaction: FastModeApplyTransaction; + request: Extract<DaemonToWorker, { type: 'set_fast_mode' }>; + phase: + | 'applying' + | 'restarting_native' + | 'awaiting_native_prompt' + | 'rolling_back_wait' + | 'rolling_back_prompt'; + serviceTier?: string; + previousFastMode?: boolean; + previousServiceTier?: string; + previousStateVersion?: 1; + previousRpcServiceTier?: string; + resultSettled: boolean; + rollbackPromise?: Promise<void>; + resolveRollback?: () => void; +}; +let activeFastModeChange: ActiveFastModeChange | null = null; +const fastModeApplyWatchdog = new FastModeRestartWatchdog(); +/** A cold or legacy native Fast setting is only executor-confirmed after the + * explicitly-tiered replacement process reaches a real prompt. */ +let pendingInitialFastModeConfirmation: { + enabled: boolean; + serviceTier?: string; +} | null = null; /** Latest requested canonical session title. Unlike a normal prompt this is an * administrative TUI command: never type-ahead while the agent is busy, never * open a model turn, and latest-wins if several renames arrive before idle. */ @@ -5179,6 +5272,317 @@ function markPromptReadyFromPty(): void { } } +function updateWorkerFastModeExecutorState( + enabled: boolean, + serviceTier?: string, +): void { + if (!lastInitConfig) return; + lastInitConfig.fastMode = enabled; + lastInitConfig.fastServiceTier = enabled ? serviceTier : undefined; + lastInitConfig.fastModeStateVersion = 1; +} + +/** Initial/legacy reconciliation has no daemon command waiter. Runtime changes + * are persisted atomically from their exact fast_mode_result requestId. */ +function publishInitialFastModeExecutorState( + enabled: boolean, + serviceTier?: string, +): void { + updateWorkerFastModeExecutorState(enabled, serviceTier); + send({ + type: 'fast_mode_state', + enabled, + ...(enabled && serviceTier ? { serviceTier } : {}), + }); +} + +function commitFastModeChange( + active: ActiveFastModeChange, + serviceTier?: string, +): void { + assertActiveFastModeChange(active); + const { request } = active; + updateWorkerFastModeExecutorState(request.enabled, serviceTier); + active.resultSettled = true; + send({ + type: 'fast_mode_result', + requestId: request.requestId, + ok: true, + enabled: request.enabled, + ...(request.enabled && serviceTier ? { serviceTier } : {}), + }); + finishActiveFastModeChange(active); +} + +function rejectFastModeChange( + request: Extract<DaemonToWorker, { type: 'set_fast_mode' }>, + reason: 'unsupported_session' | 'unsupported_model' | 'not_ready' | 'apply_failed', + message?: string, +): void { + send({ + type: 'fast_mode_result', + requestId: request.requestId, + ok: false, + reason, + ...(message ? { message } : {}), + }); +} + +function continueAfterFastModeChange(): void { + queueMicrotask(() => { + if (pendingFastModeChanges.length > 0) void flushPendingFastModeChanges(); + else void flushPending(); + }); +} + +function beginActiveFastModeChange( + request: Extract<DaemonToWorker, { type: 'set_fast_mode' }>, +): ActiveFastModeChange { + const active: ActiveFastModeChange = { + transaction: new FastModeApplyTransaction(request.requestId), + request, + phase: 'applying', + previousFastMode: lastInitConfig?.fastMode, + previousServiceTier: lastInitConfig?.fastServiceTier, + previousStateVersion: lastInitConfig?.fastModeStateVersion, + resultSettled: false, + }; + activeFastModeChange = active; + // The worker deadline owns the entire transaction, beginning at dequeue — + // model probing and RPC ACK waits are not an untracked pre-watchdog window. + fastModeApplyWatchdog.arm(request.requestId, async requestId => { + await cancelActiveFastModeChange(requestId, 'worker_deadline'); + }); + return active; +} + +function isActiveFastModeChange(active: ActiveFastModeChange): boolean { + const current = activeFastModeChange; + return current === active + && current.request.requestId === active.request.requestId; +} + +function assertActiveFastModeChange(active: ActiveFastModeChange): void { + active.transaction.assertActive(); + if (!isActiveFastModeChange(active)) { + throw new Error(`Stale Fast Mode transaction ${active.request.requestId}`); + } +} + +async function waitForActiveFastModeChange<T>( + active: ActiveFastModeChange, + operation: Promise<T>, +): Promise<T> { + const result = await active.transaction.waitFor(operation); + assertActiveFastModeChange(active); + return result; +} + +function finishActiveFastModeChange(active: ActiveFastModeChange): boolean { + if (!isActiveFastModeChange(active)) return false; + fastModeApplyWatchdog.clear(active.request.requestId); + activeFastModeChange = null; + continueAfterFastModeChange(); + return true; +} + +function rejectActiveFastModeChange( + active: ActiveFastModeChange, + reason: 'unsupported_session' | 'unsupported_model' | 'not_ready' | 'apply_failed', + message?: string, +): void { + if (active.resultSettled) return; + active.resultSettled = true; + rejectFastModeChange(active.request, reason, message); +} + +function restoreActiveFastModeConfig(active: ActiveFastModeChange): void { + if (!lastInitConfig) return; + lastInitConfig.fastMode = active.previousFastMode; + lastInitConfig.fastServiceTier = active.previousServiceTier; + lastInitConfig.fastModeStateVersion = active.previousStateVersion; +} + +/** Roll a staged native replacement back on the same serialized transaction. + * Multiple cancellation/error paths share one promise and cannot double-restart. */ +function rollbackActiveNativeFastModeChange( + active: ActiveFastModeChange, +): Promise<void> { + if (active.rollbackPromise) return active.rollbackPromise; + active.phase = 'rolling_back_wait'; + restoreActiveFastModeConfig(active); + active.rollbackPromise = new Promise<void>(resolve => { + active.resolveRollback = resolve; + }); + void (async () => { + try { + // restartCliProcess resolves after scheduling its replacement, while + // cliRestartInProgress remains true until that spawn settles. Never start + // the compensating restart on top of it: wait, ignore the possibly-new- + // tier prompt, then launch one clean old-tier replacement. + while (cliRestartInProgress && isActiveFastModeChange(active)) { + await delay(50); + } + if (!isActiveFastModeChange(active)) return; + active.phase = 'rolling_back_prompt'; + await restartCliProcess( + 'Fast Mode apply failed; restoring previous executor state', + { immediate: true, preservePending: true, skipRestartBudget: true }, + ); + } catch (error) { + log(`WARN Fast Mode rollback restart failed: ${error instanceof Error ? error.message : String(error)}`); + await sendFatalWorkerErrorAndExit(error); + } + })(); + return active.rollbackPromise; +} + +/** Mark the exact active request even while an async probe/RPC call is applying. + * Its continuation owns cleanup; a transaction already waiting for a native + * prompt can start the compensating restart immediately. */ +async function cancelActiveFastModeChange( + requestId: string, + source: FastModeCancellationSource, +): Promise<boolean> { + const active = activeFastModeChange; + if (!active || !active.transaction.cancel(requestId, source)) return false; + + fastModeApplyWatchdog.clear(requestId); + const message = source === 'worker_deadline' + ? 'Fast Mode replacement did not reach a prompt before the worker deadline' + : 'Fast Mode replacement was cancelled after the daemon deadline'; + log(`WARN ${message}; restoring previous executor state (request=${requestId.slice(0, 8)})`); + rejectActiveFastModeChange(active, 'apply_failed', message); + if (active.phase === 'awaiting_native_prompt') { + await rollbackActiveNativeFastModeChange(active); + } + return true; +} + +/** Apply one Fast Mode request at an idle prompt. RPC mode uses the protocol's + * acknowledged settings update. Native mode restarts/resumes with an explicit + * catalog tier instead of injecting the toggle-only `/fast` command; this makes + * on/off deterministic and gives the replacement prompt a real ACK boundary. */ +async function flushPendingFastModeChanges(): Promise<void> { + if (activeFastModeChange || isFlushing || cliRestartInProgress) return; + if (!backend || !cliAdapter || !isPromptReady) return; + if (commandLineWritesPending > 0 || injectionFlushing || sessionRenameInFlight) return; + const request = pendingFastModeChanges.shift(); + if (!request) return; + const active = beginActiveFastModeChange(request); + + if (!lastInitConfig || !fastModeSessionSupported({ + cliId: lastInitConfig.cliId as CliId, + wrapperCli: lastInitConfig.wrapperCli, + adopted: lastInitConfig.adoptMode, + backendType: lastInitConfig.backendType, + })) { + rejectActiveFastModeChange(active, 'unsupported_session'); + finishActiveFastModeChange(active); + return; + } + + const rpcEngine = codexRpcEngine; + try { + if (rpcEngine) { + active.previousRpcServiceTier = rpcEngine.activeServiceTier; + const applied = await waitForActiveFastModeChange( + active, + rpcEngine.setFastMode(request.enabled, { + signal: active.transaction.signal, + }), + ); + commitFastModeChange(active, applied.serviceTier); + return; + } + + let serviceTier: string | undefined; + if (request.enabled) { + const resolved = await active.transaction.waitFor( + resolveWorkerFastServiceTier(lastInitConfig), + ); + assertActiveFastModeChange(active); + if (!resolved.ok || !resolved.serviceTier) { + rejectActiveFastModeChange( + active, + !resolved.ok ? resolved.reason : 'unsupported_model', + !resolved.ok ? resolved.message : undefined, + ); + finishActiveFastModeChange(active); + return; + } + serviceTier = resolved.serviceTier; + } + + // The launch snapshot is authoritative for this worker generation. An + // idempotent request needs no native toggle; the resolved tier still heals + // a legacy record before ACK. + if (lastInitConfig.fastMode === request.enabled + && (!request.enabled || lastInitConfig.fastModeStateVersion === 1)) { + commitFastModeChange(active, serviceTier ?? lastInitConfig.fastServiceTier); + return; + } + + active.serviceTier = serviceTier; + active.phase = 'restarting_native'; + // Stage the replacement config before teardown. The daemon does not persist + // it until the replacement prompt ACK below. + lastInitConfig.fastMode = request.enabled; + lastInitConfig.fastServiceTier = request.enabled ? serviceTier : undefined; + lastInitConfig.fastModeStateVersion = undefined; + await waitForActiveFastModeChange( + active, + restartCliProcess( + `Fast Mode ${request.enabled ? 'enable' : 'disable'} reconciliation`, + { immediate: true, preservePending: true, skipRestartBudget: true }, + ), + ); + active.phase = 'awaiting_native_prompt'; + log(`Fast Mode replacement launched; waiting for prompt ACK (${request.enabled ? 'on' : 'off'})`); + } catch (error) { + // A prompt may have committed while restartCliProcess was unwinding. + if (!isActiveFastModeChange(active)) return; + if (active.transaction.cancelledBy) { + if (rpcEngine) { + // setFastMode compensates when its own AbortSignal wins. This second + // check closes the narrower race where it returned just before the + // worker observed cancellation. + if (rpcEngine.activeServiceTier !== active.previousRpcServiceTier) { + try { + await rpcEngine.restoreFastModeServiceTier(active.previousRpcServiceTier); + } catch (rollbackError) { + log(`WARN Fast Mode RPC rollback failed: ${rollbackError instanceof Error ? rollbackError.message : String(rollbackError)}`); + } + if (!isActiveFastModeChange(active)) return; + } + finishActiveFastModeChange(active); + } else if (active.phase === 'restarting_native' + || active.phase === 'awaiting_native_prompt' + || active.phase === 'rolling_back_wait' + || active.phase === 'rolling_back_prompt') { + await rollbackActiveNativeFastModeChange(active); + } else { + finishActiveFastModeChange(active); + } + return; + } + + rejectActiveFastModeChange( + active, + 'apply_failed', + error instanceof Error ? error.message : String(error), + ); + if (!rpcEngine && (active.phase === 'restarting_native' + || active.phase === 'awaiting_native_prompt' + || active.phase === 'rolling_back_wait' + || active.phase === 'rolling_back_prompt')) { + await rollbackActiveNativeFastModeChange(active); + } else { + finishActiveFastModeChange(active); + } + } +} + function markPromptReady(): void { if (isPromptReady) return; // guard against duplicate calls if (cliRestartInProgress && !replacementSpawnInProgress) { @@ -5222,6 +5626,31 @@ function markPromptReady(): void { } if (freshnessAction === 'ignore') return; isPromptReady = true; + // A non-RPC Fast change is committed only after its replacement CLI has + // resumed with the explicit tier and reached a real prompt. + if (activeFastModeChange?.phase === 'awaiting_native_prompt' + || activeFastModeChange?.phase === 'restarting_native') { + const active = activeFastModeChange; + if (active.transaction.cancelledBy) { + void rollbackActiveNativeFastModeChange(active); + } else { + commitFastModeChange(active, active.serviceTier); + log(`Fast Mode replacement confirmed at prompt (${active.request.enabled ? 'on' : 'off'})`); + } + } + if (activeFastModeChange?.phase === 'rolling_back_prompt') { + const active = activeFastModeChange; + const resolveRollback = active.resolveRollback; + finishActiveFastModeChange(active); + resolveRollback?.(); + log('Fast Mode rollback confirmed at prompt'); + } + if (pendingInitialFastModeConfirmation) { + const pending = pendingInitialFastModeConfirmation; + pendingInitialFastModeConfirmation = null; + publishInitialFastModeExecutorState(pending.enabled, pending.serviceTier); + log('Initial Fast Mode state confirmed at prompt'); + } clearSessionRenameInFlight(); // An old backend can still report idle while its async teardown is running. // Only a prompt observed after the general restart fence drops may release @@ -5283,7 +5712,9 @@ function markPromptReady(): void { // make the CLI busy, so the idle state is transient and shouldn't appear // in the card. This avoids a false "就绪" flash on daemon restart // (where the initial prompt is queued before the CLI becomes idle). - if (renderer && pendingMessages.length === 0 && pendingRawInputs.length === 0 && pendingSessionRename === null && !isFlushing) { + if (renderer && pendingMessages.length === 0 && pendingRawInputs.length === 0 + && pendingFastModeChanges.length === 0 && !activeFastModeChange + && pendingSessionRename === null && !isFlushing) { const { content } = renderer.snapshot(); send({ type: 'screen_update', content, ...usageLimitTracker.classify(content, 'idle'), turnId: currentBotmuxTurnId, dispatchAttempt: currentBotmuxDispatchAttempt }); } @@ -5292,7 +5723,9 @@ function markPromptReady(): void { // flushPending 不会饿死用户消息：flushPending 自身的 injectionFlushing 守卫 // 会挡住并发写入，flushPendingInjections 的 finally 会在注入完成、CLI 重新 // idle 后补踢一次 flushPending 排空 pendingMessages。 - if (shouldFlushInjectionsFirst(pendingInjections)) { + if (pendingFastModeChanges.length > 0) { + void flushPendingFastModeChanges(); + } else if (shouldFlushInjectionsFirst(pendingInjections)) { void flushPendingInjections(); } else { flushPending(); @@ -5599,6 +6032,12 @@ async function flushPending(): Promise<void> { if (cliRestartInProgress) return; if (shouldHoldCodexRunnerInput(codexRunnerFreshness)) return; if (isFlushing) return; // while loop in active flush will pick up new messages + // A Fast Mode request is a configuration barrier for every later user turn. + // Its own queue owns delivery until the executor ACKs it. + if (pendingFastModeChanges.length > 0 || activeFastModeChange) { + void flushPendingFastModeChanges(); + return; + } if (!backend || !cliAdapter) return; if (pendingMessages.length === 0 && pendingRawInputs.length === 0 && pendingSessionRename === null) return; // nothing to flush — keep isPromptReady if (sessionRenameInFlight) return; // wait for /rename to finish before any user input @@ -6358,7 +6797,7 @@ function scheduleBusyPatternIdleProbe(source: string): void { async function spawnCli( cfg: Extract<DaemonToWorker, { type: 'init' }>, - opts: { pluginGenerationPrepared?: boolean } = {}, + opts: { pluginGenerationPrepared?: boolean; forceFreshPersistent?: boolean } = {}, ): Promise<void> { clearSessionRenameInFlight(); currentCliCredentialIsolated = false; @@ -6839,6 +7278,34 @@ async function spawnCli( // real host path the probe may not be able to stat). Computed here (not at // the spawn site below) so the pre-flight can short-circuit on it. let persistentSessionName = selectedBackend.persistentSessionName; + const persistentBackendType: PersistentBackendType | undefined = + effectiveBackendType === 'tmux' + || effectiveBackendType === 'herdr' + || effectiveBackendType === 'zellij' + ? effectiveBackendType + : undefined; + if (opts.forceFreshPersistent && persistentSessionName && persistentBackendType) { + const target = selectedBackend.persistentBackendTarget ?? { + backendType: persistentBackendType, + sessionName: persistentSessionName, + }; + if (probePersistentBackendTarget(target) !== 'missing') { + log(`[fast-mode] replacing unconfirmed legacy persistent pane for ${cfg.sessionId}`); + const killed = await killPersistentBackendTargetVerified(target); + if (!killed) { + throw new Error( + `[fast-mode] refusing to reattach unconfirmed persistent pane for ${cfg.sessionId}`, + ); + } + selectedBackend = selectBackend(); + isTmuxMode = selectedBackend.isTmuxMode; + isPipeMode = selectedBackend.isPipeMode; + isZellijMode = selectedBackend.isZellijMode; + backend = selectedBackend.backend; + cliLifetimeNonce++; + persistentSessionName = selectedBackend.persistentSessionName; + } + } // [read-isolation] Before we decide to reattach a persistent pane: a pane can // survive a daemon restart still running a CLI that may NOT be isolated (e.g. // spawned before isolation was enabled, or by an old build). Isolation is only @@ -7175,9 +7642,13 @@ async function spawnCli( locale: cfg.locale, model: ttadkGateway ? undefined : cfg.model, reasoningEffort: cfg.reasoningEffort, + fastMode: cfg.fastMode === true, + fastServiceTier: cfg.fastServiceTier, disableCliBypass: cfg.disableCliBypass === true, skillPluginDir: cfg.skillPluginDir, readIsolation: willRedirectCliData, + remoteWsUrl, + remoteThreadId, }); // Pi's deferred long-first-prompt command is implemented by a session-scoped // extension. Keep its launch args across owned process restarts while the @@ -7755,6 +8226,9 @@ async function spawnCli( } else { const launch = buildWrappedLaunch(cfg.wrapperCli, spawnArgs, (b) => locateOnPath(b) ?? b, { ttadkModel: cfg.model, + codexServiceTier: cfg.cliId === 'codex' + ? (cfg.fastMode === true ? cfg.fastServiceTier : 'default') + : undefined, }); if (launch.bin) { spawnBin = launch.bin; @@ -7945,6 +8419,9 @@ async function spawnCli( sandboxOn: sandboxRequested, binResolver: (b) => locateOnPath(b) ?? b, ttadkModel: cfg.model, + codexServiceTier: cfg.cliId === 'codex' + ? (cfg.fastMode === true ? cfg.fastServiceTier : 'default') + : undefined, }); capturedSpawnCommand = buildReproduceCommand({ backendType: effectiveBackendType, @@ -10029,6 +10506,16 @@ process.on('message', async (raw: unknown) => { // then launched/respawned as `codex --remote resume` (codex.ts buildArgs) // against the CURRENT app-server (a fresh port each incarnation). const rpcBackendType = msg.backendType ?? config.daemon.backendType; + const fastModeNeedsReconciliation = fastModeSessionSupported({ + cliId: msg.cliId as CliId, + wrapperCli: msg.wrapperCli, + adopted: msg.adoptMode, + backendType: msg.backendType, + }) && fastModeStateNeedsReconciliation({ + enabled: msg.fastMode === true, + serviceTier: msg.fastServiceTier, + stateVersion: msg.fastModeStateVersion, + }); let rpcPluginGenerationPrepared = false; const rpcDecision = await orchestrateCodexRpcInit(msg, { paneInfo: (sid) => persistentPaneInfo(rpcBackendType, sid), @@ -10050,7 +10537,29 @@ process.on('message', async (raw: unknown) => { // init so the daemon retries a clean incarnation instead of freezing. throw new Error('codex RPC resume: could not replace stale --remote pane; aborting init'); } - await spawnCli(msg, { pluginGenerationPrepared: rpcPluginGenerationPrepared }); + // RPC engage resolves the tier itself; native/migration paths resolve + // it here before buildArgs so a persisted Fast Session can never launch + // at default tier. + await ensureConfiguredFastServiceTier(msg); + if (fastModeNeedsReconciliation) { + if (codexRpcEngine) { + // thread/start|resume already accepted either the concrete Fast + // tier or null for default. + publishInitialFastModeExecutorState(msg.fastMode === true, msg.fastServiceTier); + } else { + // Native launch args are not proof until the replacement reaches a + // real prompt. Arm this before spawn so an early ready event cannot + // race past the confirmation. + pendingInitialFastModeConfirmation = { + enabled: msg.fastMode === true, + ...(msg.fastServiceTier ? { serviceTier: msg.fastServiceTier } : {}), + }; + } + } + await spawnCli(msg, { + pluginGenerationPrepared: rpcPluginGenerationPrepared, + forceFreshPersistent: fastModeNeedsReconciliation && !codexRpcEngine, + }); await prepareCodexNativeTitleGeneration(msg, codexRpcEngine); if (codexRpcEngine) armRpcStartupDialogDismiss(); // boundary #4: keep the --remote pane from freezing on a startup dialog @@ -10299,6 +10808,33 @@ process.on('message', async (raw: unknown) => { break; } + case 'set_fast_mode': { + // Queue even if init/backend is still starting. The prompt-gated flush + // applies this before any later user input and returns an explicit ACK. + pendingFastModeChanges.push(msg); + log(`Queued Fast Mode change (${msg.enabled ? 'on' : 'off'}, request=${msg.requestId.slice(0, 8)})`); + void flushPendingFastModeChanges(); + break; + } + + case 'cancel_fast_mode': { + if (await cancelActiveFastModeChange(msg.requestId, 'daemon_timeout')) { + break; + } + const queuedIndex = pendingFastModeChanges.findIndex( + request => request.requestId === msg.requestId, + ); + if (queuedIndex >= 0) { + const [cancelled] = pendingFastModeChanges.splice(queuedIndex, 1); + rejectFastModeChange(cancelled, 'not_ready', 'Fast Mode request expired before execution'); + log(`Cancelled queued Fast Mode change (request=${msg.requestId.slice(0, 8)})`); + continueAfterFastModeChange(); + } else { + log(`Ignored stale Fast Mode cancellation (request=${msg.requestId.slice(0, 8)})`); + } + break; + } + case 'raw_input': { // Preserve legacy busy delivery (/btw and other steering commands). A // native /rename and an owned CLI restart are the exceptions: never splice @@ -10319,7 +10855,8 @@ process.on('message', async (raw: unknown) => { if (cliRestartInProgress || rawInputRestartGate || sessionRenameInFlight || shouldHoldCodexRunnerInput(codexRunnerFreshness) || injectionFlushing || shouldDeferUserFlush(pendingInjections) - || bareShellCheckInProgress || bareShellLaunchBlocked) { + || bareShellCheckInProgress || bareShellLaunchBlocked + || pendingFastModeChanges.length > 0 || activeFastModeChange) { freshnessInputQueue.enqueueRaw(msg); log(`Deferred passthrough slash command until CLI input gate settles: ${msg.content}`); } else { @@ -10711,6 +11248,7 @@ process.on('message', async (raw: unknown) => { function cleanup(): void { stopNativeSessionTitleSync(); + fastModeApplyWatchdog.dispose(); cleanupPiInitialPromptFiles(); stopSessionMcpGatewayHost(); if (tmuxRestartTimer) { diff --git a/test/cli-adapters.test.ts b/test/cli-adapters.test.ts index 72839e2ed..36c95c8f7 100644 --- a/test/cli-adapters.test.ts +++ b/test/cli-adapters.test.ts @@ -333,7 +333,9 @@ describe('codex buildArgs', () => { // pure --remote viewer: no paste-mode bypass flag, no stale resume path expect(args).toEqual([ '--remote', 'ws://127.0.0.1:9931', 'resume', '--no-alt-screen', - '-c', 'check_for_update_on_startup=false', 'thread-abc', + '-c', 'check_for_update_on_startup=false', + '-c', 'service_tier="default"', + 'thread-abc', ]); // the -c disable must land BEFORE the thread id (a resume-subcommand config) const cIdx = args.indexOf('-c'); @@ -394,6 +396,8 @@ describe('codex buildArgs', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="sess-4"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', '-C', '/repo/root', ]); @@ -407,6 +411,8 @@ describe('codex buildArgs', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="sess-4"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', '-C', '/repo/root', ]); @@ -419,6 +425,28 @@ describe('codex buildArgs', () => { expect(args[idx - 1]).toBe('-c'); }); + it('starts every Codex Session with Fast Mode off unless that Session opted in', () => { + const standard = adapter.buildArgs({ sessionId: 'sess-standard', resume: false }); + const fast = adapter.buildArgs({ + sessionId: 'sess-fast', + resume: false, + fastMode: true, + fastServiceTier: 'priority', + }); + + expect(standard).toContain('service_tier="default"'); + expect(fast).toContain('service_tier="priority"'); + expect(fast).not.toContain('service_tier="default"'); + }); + + it('refuses to guess a Fast protocol tier when the model catalog was not resolved', () => { + expect(() => adapter.buildArgs({ + sessionId: 'sess-fast-unresolved', + resume: false, + fastMode: true, + })).toThrow(/Fast service tier was not resolved/); + }); + it('keeps the startup update override on resume before the Codex session id', () => { const args = adapter.buildArgs({ sessionId: 'sess-4', diff --git a/test/cli-selection.test.ts b/test/cli-selection.test.ts index 647435b5d..d753fad0a 100644 --- a/test/cli-selection.test.ts +++ b/test/cli-selection.test.ts @@ -231,6 +231,22 @@ describe('stripWrapperUnsafeArgs', () => { .toEqual(['--model', 'm']); }); + it('strips the botmux-injected Session service tier override', () => { + expect(stripWrapperUnsafeArgs([ + '-c', 'service_tier="default"', + '-c', 'service_tier="fast"', + '-c', 'service_tier="priority"', + '--model', 'm', + ], 'priority')).toEqual(['--model', 'm']); + }); + + it('does not claim an arbitrary catalog tier unless the caller says Botmux injected it', () => { + expect(stripWrapperUnsafeArgs([ + '-c', 'service_tier="priority"', + '--model', 'm', + ])).toEqual(['-c', 'service_tier="priority"', '--model', 'm']); + }); + it('leaves a user-supplied -c (non-botmux config override) untouched', () => { expect(stripWrapperUnsafeArgs(['-c', 'model_reasoning_effort="high"', '--model', 'm'])) .toEqual(['-c', 'model_reasoning_effort="high"', '--model', 'm']); @@ -304,6 +320,16 @@ describe('buildWrappedLaunch', () => { expect(out.args).toEqual(['x', 'codex', '--no-alt-screen']); }); + it('strips the exact dynamic Fast tier injected by Botmux', () => { + const out = buildWrappedLaunch( + 'aiden x codex', + ['-c', 'service_tier="priority"', '--no-alt-screen'], + (bin) => bin, + { codexServiceTier: 'priority' }, + ); + expect(out.args).toEqual(['x', 'codex', '--no-alt-screen']); + }); + it('does not strip a user-supplied -c that is not a botmux override (aiden x codex)', () => { const out = buildWrappedLaunch('aiden x codex', ['-c', 'model_reasoning_effort="high"', '--model', 'm']); expect(out.args).toEqual(['x', 'codex', '-c', 'model_reasoning_effort="high"', '--model', 'm']); @@ -342,6 +368,16 @@ describe('buildWrappedLaunch', () => { expect(out.args).toEqual(['codex', '--config', 'check_for_update_on_startup=false']); }); + it('rewrites the exact dynamic Fast tier injected by Botmux for cjadk', () => { + const out = buildWrappedLaunch( + 'cjadk codex', + ['-c', 'service_tier="priority"', '--no-alt-screen'], + (bin) => bin, + { codexServiceTier: 'priority' }, + ); + expect(out.args).toEqual(['codex', '--config', 'service_tier="priority"', '--no-alt-screen']); + }); + it('keeps the codex -c override for the bare-passthrough ttadk codex gateway', () => { const out = buildWrappedLaunch('ttadk codex', [ '-c', diff --git a/test/codex-fast-worker-wiring.test.ts b/test/codex-fast-worker-wiring.test.ts new file mode 100644 index 000000000..c4f8c86ee --- /dev/null +++ b/test/codex-fast-worker-wiring.test.ts @@ -0,0 +1,132 @@ +import { readFileSync } from 'node:fs'; +import { describe, expect, it } from 'vitest'; + +const workerSource = readFileSync(new URL('../src/worker.ts', import.meta.url), 'utf8'); +const workerPoolSource = readFileSync(new URL('../src/core/worker-pool.ts', import.meta.url), 'utf8'); + +describe('Codex Fast Mode worker wiring', () => { + it('passes Session Fast Mode into the app-server executor and captures its resolved tier', () => { + const start = workerSource.indexOf('async function engageCodexRpc('); + const end = workerSource.indexOf('/** RPC panes have NO terminal input path', start); + const region = workerSource.slice(start, end); + + expect(region).toContain('fastMode: cfg.fastMode === true'); + expect(region).toContain('cfg.fastServiceTier = engine.activeServiceTier'); + }); + + it('passes the live app-server endpoint and thread to the viewer adapter', () => { + const start = workerSource.indexOf('const args = cliAdapter.buildArgs({'); + const end = workerSource.indexOf('\n });', start); + const region = workerSource.slice(start, end); + + expect(start).toBeGreaterThan(-1); + expect(end).toBeGreaterThan(start); + expect(region.length).toBeLessThan(2_000); + expect(region).toContain('remoteWsUrl'); + expect(region).toContain('remoteThreadId'); + expect(region).toContain('fastServiceTier: cfg.fastServiceTier'); + }); + + it('queues typed runtime changes and persists only an exact pending result', () => { + expect(workerSource).toContain("case 'set_fast_mode':"); + expect(workerSource).toContain('pendingFastModeChanges.push(msg)'); + + const applyStart = workerSource.indexOf('function updateWorkerFastModeExecutorState('); + const applyEnd = workerSource.indexOf('\n}', applyStart); + const apply = workerSource.slice(applyStart, applyEnd); + expect(applyStart).toBeGreaterThan(-1); + expect(apply.indexOf('lastInitConfig.fastMode =')).toBeGreaterThan(-1); + const commitStart = workerSource.indexOf('function commitFastModeChange('); + const commitEnd = workerSource.indexOf('\n}', commitStart); + const commit = workerSource.slice(commitStart, commitEnd); + expect(commit.indexOf("type: 'fast_mode_result'")).toBeGreaterThan( + commit.indexOf('updateWorkerFastModeExecutorState('), + ); + expect(commit).not.toContain("type: 'fast_mode_state'"); + + const resultStart = workerPoolSource.indexOf("case 'fast_mode_result':"); + const resultEnd = workerPoolSource.indexOf("case 'fast_mode_state':", resultStart); + const result = workerPoolSource.slice(resultStart, resultEnd); + expect(result.indexOf('isFastModeResultPending(msg.requestId)')).toBeGreaterThan(-1); + expect(result.indexOf('ds.session.fastMode = msg.enabled')).toBeGreaterThan( + result.indexOf('isFastModeResultPending(msg.requestId)'), + ); + expect(result.indexOf('acknowledgeFastModeResult(msg)')).toBeGreaterThan( + result.indexOf('ds.session.fastMode = msg.enabled'), + ); + }); + + it('passes the resolved backend into every worker-side Fast capability gate', () => { + const resolveStart = workerSource.indexOf('async function resolveWorkerFastServiceTier('); + const resolveEnd = workerSource.indexOf('async function ensureConfiguredFastServiceTier(', resolveStart); + expect(workerSource.slice(resolveStart, resolveEnd)).toContain('backendType: cfg.backendType'); + + const runtimeStart = workerSource.indexOf('async function flushPendingFastModeChanges(): Promise<void>'); + const runtimeEnd = workerSource.indexOf('function markPromptReady(): void', runtimeStart); + expect(workerSource.slice(runtimeStart, runtimeEnd)).toContain('backendType: lastInitConfig.backendType'); + }); + + it('persists the concrete tier across daemon/worker restarts', () => { + expect(workerPoolSource).toContain("fastServiceTier: agentCfg.cliId === 'codex' ? ds.session.fastServiceTier : undefined"); + expect(workerPoolSource).toContain("fastModeStateVersion: agentCfg.cliId === 'codex' ? ds.session.fastModeStateVersion : undefined"); + expect(workerPoolSource).toContain("case 'fast_mode_result':"); + expect(workerPoolSource).toContain("case 'fast_mode_state':"); + expect(workerPoolSource).toContain('ds.session.fastServiceTier = msg.serviceTier'); + expect(workerPoolSource).toContain('ds.session.fastModeStateVersion = 1'); + }); + + it('reconciles an unconfirmed legacy Fast state in either direction exactly once', () => { + const initStart = workerSource.indexOf("case 'init':"); + const initEnd = workerSource.indexOf('// Queue the initial prompt', initStart); + const init = workerSource.slice(initStart, initEnd); + expect(init).toContain('}) && fastModeStateNeedsReconciliation({'); + expect(init).toContain('enabled: msg.fastMode === true'); + expect(init).toContain('forceFreshPersistent: fastModeNeedsReconciliation && !codexRpcEngine'); + expect(init).toContain('pendingInitialFastModeConfirmation ='); + + const spawnStart = workerSource.indexOf('async function spawnCli('); + const spawnEnd = workerSource.indexOf('// The plugin set is stable', spawnStart); + const spawn = workerSource.slice(spawnStart, spawnEnd); + expect(spawn).toContain('opts.forceFreshPersistent'); + expect(spawn).toContain('killPersistentBackendTargetVerified'); + + const readyStart = workerSource.indexOf('function markPromptReady(): void'); + const readyEnd = workerSource.indexOf('clearSessionRenameInFlight();', readyStart); + const ready = workerSource.slice(readyStart, readyEnd); + expect(ready).toContain('pendingInitialFastModeConfirmation'); + expect(ready).toContain('publishInitialFastModeExecutorState'); + }); + + it('owns the exact request from dequeue through async apply and native prompt ACK', () => { + expect(workerSource).toContain("case 'cancel_fast_mode':"); + expect(workerSource).toContain('await cancelActiveFastModeChange(msg.requestId'); + + const beginFnStart = workerSource.indexOf('function beginActiveFastModeChange('); + const beginFnEnd = workerSource.indexOf('function assertActiveFastModeChange(', beginFnStart); + const beginFn = workerSource.slice(beginFnStart, beginFnEnd); + expect(beginFn.indexOf('activeFastModeChange = active')).toBeGreaterThan(-1); + expect(beginFn.indexOf('fastModeApplyWatchdog.arm(request.requestId')).toBeGreaterThan( + beginFn.indexOf('activeFastModeChange = active'), + ); + + const runtimeStart = workerSource.indexOf('async function flushPendingFastModeChanges(): Promise<void>'); + const runtimeEnd = workerSource.indexOf('function markPromptReady(): void', runtimeStart); + const runtime = workerSource.slice(runtimeStart, runtimeEnd); + const dequeue = runtime.indexOf('pendingFastModeChanges.shift()'); + const begin = runtime.indexOf('beginActiveFastModeChange(request)'); + const rpcApply = runtime.indexOf('rpcEngine.setFastMode('); + const nativeProbe = runtime.indexOf('resolveWorkerFastServiceTier('); + expect(dequeue).toBeGreaterThan(-1); + expect(begin).toBeGreaterThan(dequeue); + expect(rpcApply).toBeGreaterThan(begin); + expect(nativeProbe).toBeGreaterThan(begin); + expect(runtime).toContain('active.transaction.waitFor('); + expect(runtime).toContain('signal: active.transaction.signal'); + + const readyStart = workerSource.indexOf('function markPromptReady(): void'); + const readyEnd = workerSource.indexOf('clearSessionRenameInFlight();', readyStart); + const ready = workerSource.slice(readyStart, readyEnd); + expect(ready).toContain("activeFastModeChange?.phase === 'awaiting_native_prompt'"); + expect(ready).toContain('commitFastModeChange(active'); + }); +}); diff --git a/test/codex-rpc-engine.test.ts b/test/codex-rpc-engine.test.ts index 96556c9d4..9a2bc9308 100644 --- a/test/codex-rpc-engine.test.ts +++ b/test/codex-rpc-engine.test.ts @@ -1,10 +1,13 @@ import { describe, it, expect, beforeAll } from 'vitest'; -import { chmodSync, mkdirSync, writeFileSync, existsSync, rmSync, readFileSync } from 'node:fs'; +import { chmodSync, mkdirSync, writeFileSync, existsSync, mkdtempSync, readFileSync, rmSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import { spawn } from 'node:child_process'; import { homedir, tmpdir } from 'node:os'; import { join } from 'node:path'; -import { CodexRpcEngine } from '../src/codex-rpc-engine.js'; +import { + CodexRpcEngine, + CodexRpcFastModeCancelledError, +} from '../src/codex-rpc-engine.js'; const isAlive = (pid: number) => { try { process.kill(pid, 0); return true; } catch { return false; } }; @@ -20,7 +23,136 @@ function makeEngine(over: Partial<ConstructorParameters<typeof CodexRpcEngine>[0 }); } +function readRpcLog(path: string): Array<{ method?: string; params?: Record<string, unknown> }> { + return readFileSync(path, 'utf8') + .trim() + .split('\n') + .filter(Boolean) + .map(line => JSON.parse(line)); +} + +async function waitForRpcCount(path: string, method: string, count: number): Promise<void> { + const deadline = Date.now() + 2_000; + while (Date.now() < deadline) { + if (existsSync(path) && readRpcLog(path).filter(entry => entry.method === method).length >= count) { + return; + } + await new Promise(resolve => setTimeout(resolve, 10)); + } + throw new Error(`Timed out waiting for ${count} ${method} request(s)`); +} + describe('CodexRpcEngine — happy-path lifecycle against a fake app-server', () => { + it('resolves the model catalog Fast tier and pins it on thread/start + turn/start', async () => { + const dir = mkdtempSync(join(tmpdir(), 'botmux-fast-rpc-')); + const rpcLog = join(dir, 'rpc.jsonl'); + const engine = makeEngine({ + model: 'gpt-fast', + fastMode: true, + env: { ...process.env, FAKE_RPC_LOG: rpcLog }, + }); + + try { + await engine.start(); + await engine.startThread(); + await engine.sendTurn('run fast'); + + const requests = readRpcLog(rpcLog); + expect(requests.find(entry => entry.method === 'model/list')).toBeDefined(); + expect(requests.find(entry => entry.method === 'thread/start')?.params).toMatchObject({ + serviceTier: 'priority', + }); + expect(requests.find(entry => entry.method === 'turn/start')?.params).toMatchObject({ + serviceTier: 'priority', + }); + } finally { + engine.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }, 20_000); + + it('pins Fast on thread/resume and applies runtime changes through acknowledged thread settings', async () => { + const dir = mkdtempSync(join(tmpdir(), 'botmux-fast-rpc-update-')); + const rpcLog = join(dir, 'rpc.jsonl'); + const engine = makeEngine({ + model: 'gpt-fast', + env: { ...process.env, FAKE_RPC_LOG: rpcLog }, + }); + + try { + await engine.start(); + await engine.resumeThread('thread-fast-resume'); + await engine.setFastMode(true); + await engine.sendTurn('fast turn'); + await engine.setFastMode(false); + await engine.sendTurn('default turn'); + + const requests = readRpcLog(rpcLog); + const resume = requests.find(entry => entry.method === 'thread/resume'); + expect(resume?.params).toMatchObject({ serviceTier: null }); + const updates = requests.filter(entry => entry.method === 'thread/settings/update'); + expect(updates.map(entry => entry.params?.serviceTier)).toEqual(['priority', null]); + const turns = requests.filter(entry => entry.method === 'turn/start'); + expect(turns.map(entry => entry.params?.serviceTier)).toEqual(['priority', null]); + } finally { + engine.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }, 20_000); + + it('compensates to the previous tier when a settings ACK arrives after cancellation', async () => { + const dir = mkdtempSync(join(tmpdir(), 'botmux-fast-rpc-cancel-')); + const rpcLog = join(dir, 'rpc.jsonl'); + const engine = makeEngine({ + model: 'gpt-fast', + env: { + ...process.env, + FAKE_RPC_LOG: rpcLog, + FAKE_SETTINGS_UPDATE_ACK_DELAY_MS: '100', + }, + }); + const controller = new AbortController(); + + try { + await engine.start(); + await engine.resumeThread('thread-fast-cancel'); + + const applying = engine.setFastMode(true, { signal: controller.signal }); + await waitForRpcCount(rpcLog, 'thread/settings/update', 1); + controller.abort(); + + await expect(applying).rejects.toBeInstanceOf(CodexRpcFastModeCancelledError); + await waitForRpcCount(rpcLog, 'thread/settings/update', 2); + + const updates = readRpcLog(rpcLog) + .filter(entry => entry.method === 'thread/settings/update'); + expect(updates.map(entry => entry.params?.serviceTier)).toEqual(['priority', null]); + expect(engine.activeServiceTier).toBeUndefined(); + } finally { + engine.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }, 20_000); + + it('rejects Fast before thread creation when the selected model has no Fast tier', async () => { + const dir = mkdtempSync(join(tmpdir(), 'botmux-fast-rpc-unsupported-')); + const rpcLog = join(dir, 'rpc.jsonl'); + const engine = makeEngine({ + model: 'gpt-standard', + fastMode: true, + env: { ...process.env, FAKE_RPC_LOG: rpcLog }, + }); + + try { + await engine.start(); + await expect(engine.startThread()).rejects.toThrow(/Fast Mode is not supported/); + expect(readRpcLog(rpcLog).some(entry => entry.method === 'thread/start')).toBe(false); + } finally { + engine.stop(); + rmSync(dir, { recursive: true, force: true }); + } + }, 20_000); + it('start (spawn → /readyz → connect → initialize) then startThread → sendTurn → stop', async () => { const engine = makeEngine(); await engine.start(); diff --git a/test/command-handler.test.ts b/test/command-handler.test.ts index d44c1169f..740ebdfcd 100644 --- a/test/command-handler.test.ts +++ b/test/command-handler.test.ts @@ -458,7 +458,7 @@ vi.mock('../src/services/card-mode-store.js', () => ({ // ─── Imports (after mocks) ────────────────────────────────────────────────── -import { DAEMON_COMMANDS, SESSIONLESS_DAEMON_COMMANDS, PASSTHROUGH_COMMANDS, resolvePassthroughCommands, resolveAdapterDefaultPassthroughCommands, handleCommand, handleCardCommand, handleTermLinkCommand, parseSlashCommandInvocation, parseForceTopicInvocation } from '../src/core/command-handler.js'; +import { DAEMON_COMMANDS, SESSIONLESS_DAEMON_COMMANDS, PASSTHROUGH_COMMANDS, resolveAdapterDefaultPassthroughCommands, resolvePassthroughCommands, handleCommand, handleCardCommand, handleTermLinkCommand, parseSlashCommandInvocation, parseForceTopicInvocation } from '../src/core/command-handler.js'; import { setCardMode } from '../src/services/card-mode-store.js'; import { writeRoleFile, deleteRoleFile, writeTeamRoleFile, deleteTeamRoleFile, resolveRole, resolveRoleFile } from '../src/core/role-resolver.js'; import { setBotCapability, clearBotCapability } from '../src/services/bot-profile-store.js'; @@ -579,6 +579,11 @@ function makeDeps(ds?: DaemonSession): CommandHandlerDeps { getActiveCount: vi.fn(() => activeSessions.size), lastRepoScan: new Map(), prewarmDocCommentSession: vi.fn(async () => {}), + applyFastMode: vi.fn(async (_session, enabled) => ( + enabled + ? { ok: true as const, enabled: true, serviceTier: 'priority' } + : { ok: true as const, enabled: false } + )), }; } @@ -605,7 +610,7 @@ function mockCodexAppBot(): void { describe('DAEMON_COMMANDS set', () => { it('should contain all expected commands', () => { - const expected = ['/close', '/restart', '/status', '/help', '/cd', '/repo', '/rename', '/schedule', '/role', '/botconfig', '/skills', '/pair', '/login', '/adopt', '/detach', '/disconnect', '/oncall', '/group', '/g', '/relay', '/card', '/term', '/list-slash-command', '/slash', '/subscribe-lark-doc', '/watch-comment', '/vc', '/insight', '/dashboard', '/vc-auth']; + const expected = ['/close', '/restart', '/status', '/fast', '/help', '/cd', '/repo', '/rename', '/schedule', '/role', '/botconfig', '/skills', '/pair', '/login', '/adopt', '/detach', '/disconnect', '/oncall', '/group', '/g', '/relay', '/card', '/term', '/list-slash-command', '/slash', '/subscribe-lark-doc', '/watch-comment', '/vc', '/insight', '/dashboard', '/vc-auth']; for (const cmd of expected) { expect(DAEMON_COMMANDS.has(cmd), `Expected DAEMON_COMMANDS to contain ${cmd}`).toBe(true); } @@ -638,10 +643,11 @@ describe('DAEMON_COMMANDS set', () => { }); it('should have the correct size', () => { - // 30 = current master command set without the removed /land command. + // 31 = current master command set without the removed /land command, plus + // the session-scoped /fast control command. // /subscribe-lark-doc remains // as its original per-file API subscription command rather than an alias. - expect(DAEMON_COMMANDS.size).toBe(30); + expect(DAEMON_COMMANDS.size).toBe(31); }); it('contains the /list-slash-command lister and its /slash alias', () => { @@ -983,6 +989,14 @@ describe('PASSTHROUGH_COMMANDS set', () => { expect(resolvePassthroughCommands('app-2').has('/goal')).toBe(true); }); + it('keeps /fast daemon-owned instead of exposing it as raw Codex passthrough', () => { + expect(DAEMON_COMMANDS.has('/fast')).toBe(true); + expect(PASSTHROUGH_COMMANDS.has('/fast')).toBe(false); + expect(resolveAdapterDefaultPassthroughCommands('app-1')).not.toContain('/fast'); + expect(resolveAdapterDefaultPassthroughCommands('app-2')).not.toContain('/fast'); + expect(resolvePassthroughCommands('app-2').has('/fast')).toBe(false); + }); + it('exposes /effort globally to every CLI (best-effort passthrough)', () => { // /effort 放在全局 PASSTHROUGH_COMMANDS,而非某个 adapter 的 defaultPassthroughCommands // ——所有 CLI 都尽力透传(Claude 家族 / Codex 原生支持;其它 CLI 认不得顶多回 @@ -1634,6 +1648,223 @@ describe('handleCommand', () => { }); }); + // ─── /fast ────────────────────────────────────────────────────────────── + + describe('/fast', () => { + function makeCodexFastSession(enabled: boolean | undefined, running = true): DaemonSession { + const session = makeSession({ cliId: 'codex' }); + if (enabled !== undefined) session.fastMode = enabled; + return makeDaemonSession({ + larkAppId: 'app-2', + session, + worker: running ? ({ killed: false, send: vi.fn() } as any) : null, + }); + } + + it('enables Fast Mode for only the current Session and replies in Lark', async () => { + const ds = makeCodexFastSession(false); + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), deps, 'app-2'); + + expect(deps.applyFastMode).toHaveBeenCalledWith(ds, true); + expect(ds.worker?.send).not.toHaveBeenCalled(); + expect(ds.session.fastMode).toBe(true); + expect(ds.session.fastServiceTier).toBe('priority'); + expect(sessionStore.updateSession).toHaveBeenCalledWith(ds.session); + expect(deps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringMatching(/Fast Mode.*已开启.*当前话题/s), + undefined, + 'app-2', + 'msg_001', + ); + }); + + it('stores the setting without spawning a CLI when /fast starts a new topic', async () => { + const ds = makeCodexFastSession(false, false); + ds.session.fastModeStateVersion = 1; + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), deps, 'app-2'); + + expect(deps.applyFastMode).toHaveBeenCalledWith(ds, true); + expect(ds.session.fastMode).toBe(true); + expect(ds.session.fastServiceTier).toBe('priority'); + expect(ds.session.fastModeStateVersion).toBeUndefined(); + expect(sessionStore.updateSession).toHaveBeenCalledWith(ds.session); + expect(deps.sessionReply).toHaveBeenCalled(); + }); + + it('re-applies an enabled legacy Session until an executor confirms it', async () => { + const ds = makeCodexFastSession(true); + ds.session.fastServiceTier = 'priority'; + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), deps, 'app-2'); + + expect(deps.applyFastMode).toHaveBeenCalledWith(ds, true); + }); + + it('re-applies a disabled legacy Session until an executor confirms default tier', async () => { + const ds = makeCodexFastSession(false); + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast off'), deps, 'app-2'); + + expect(deps.applyFastMode).toHaveBeenCalledWith(ds, false); + }); + + it('persists and reports success only after the worker ACKs the change', async () => { + const ds = makeCodexFastSession(false); + const deps = makeDeps(ds); + let resolveApply!: (value: { ok: true; enabled: true; serviceTier: string }) => void; + deps.applyFastMode = vi.fn(() => new Promise(resolve => { resolveApply = resolve; })); + + const pending = handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), deps, 'app-2'); + await Promise.resolve(); + + expect(ds.session.fastMode).toBe(false); + expect(sessionStore.updateSession).not.toHaveBeenCalled(); + expect(deps.sessionReply).not.toHaveBeenCalled(); + + resolveApply({ ok: true, enabled: true, serviceTier: 'priority' }); + await pending; + + expect(ds.session.fastMode).toBe(true); + expect(ds.session.fastServiceTier).toBe('priority'); + expect(sessionStore.updateSession).toHaveBeenCalledWith(ds.session); + expect(deps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringMatching(/Fast Mode.*已开启/s), + undefined, + 'app-2', + 'msg_001', + ); + }); + + it('keeps persisted state unchanged when applying Fast Mode fails', async () => { + const ds = makeCodexFastSession(false); + const deps = makeDeps(ds); + deps.applyFastMode = vi.fn(async () => ({ + ok: false as const, + reason: 'unsupported_model' as const, + })); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), deps, 'app-2'); + + expect(ds.session.fastMode).toBe(false); + expect(ds.session.fastServiceTier).toBeUndefined(); + expect(sessionStore.updateSession).not.toHaveBeenCalled(); + expect(deps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringContaining('当前模型不支持'), + undefined, + 'app-2', + 'msg_001', + ); + }); + + it('reports Session state without toggling Codex', async () => { + const ds = makeCodexFastSession(true); + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast status'), deps, 'app-2'); + + expect(ds.worker?.send).not.toHaveBeenCalled(); + expect(deps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringMatching(/Fast Mode.*已开启.*当前话题/s), + undefined, + 'app-2', + 'msg_001', + ); + }); + + it('keeps /fast off idempotent and does not toggle the native CLI', async () => { + const ds = makeCodexFastSession(false); + ds.session.fastModeStateVersion = 1; + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast off'), deps, 'app-2'); + + expect(deps.applyFastMode).not.toHaveBeenCalled(); + expect(ds.worker?.send).not.toHaveBeenCalled(); + expect(ds.session.fastMode).toBe(false); + expect(deps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringMatching(/Fast Mode.*已关闭.*当前话题/s), + undefined, + 'app-2', + 'msg_001', + ); + }); + + it('uses bare /fast as a toggle', async () => { + const ds = makeCodexFastSession(false); + const deps = makeDeps(ds); + + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast'), deps, 'app-2'); + + expect(ds.session.fastMode).toBe(true); + expect(deps.applyFastMode).toHaveBeenCalledWith(ds, true); + expect(ds.worker?.send).not.toHaveBeenCalled(); + }); + + it('rejects unsupported arguments and non-Codex Sessions explicitly', async () => { + const codex = makeCodexFastSession(false); + const codexDeps = makeDeps(codex); + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast turbo'), codexDeps, 'app-2'); + expect(codexDeps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringContaining('/fast on'), + undefined, + 'app-2', + 'msg_001', + ); + expect(codex.worker?.send).not.toHaveBeenCalled(); + + const claude = makeDaemonSession(); + const claudeDeps = makeDeps(claude); + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), claudeDeps, LARK_APP_ID); + expect(claudeDeps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringContaining('仅支持由 Botmux 管理'), + undefined, + LARK_APP_ID, + 'msg_001', + ); + + const aiden = makeCodexFastSession(false); + aiden.session.wrapperCli = 'aiden x codex'; + aiden.session.agentFrozen = true; + const aidenDeps = makeDeps(aiden); + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), aidenDeps, 'app-2'); + expect(aiden.worker?.send).not.toHaveBeenCalled(); + expect(aidenDeps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringContaining('Aiden'), + undefined, + 'app-2', + 'msg_001', + ); + + const riff = makeCodexFastSession(false); + riff.session.cliId = 'riff'; + riff.session.backendType = 'riff'; + const riffDeps = makeDeps(riff); + await handleCommand('/fast', ROOT_ID, makeLarkMessage('/fast on'), riffDeps, 'app-2'); + expect(riffDeps.applyFastMode).not.toHaveBeenCalled(); + expect(riffDeps.sessionReply).toHaveBeenCalledWith( + ROOT_ID, + expect.stringContaining('仅支持由 Botmux 管理'), + undefined, + 'app-2', + 'msg_001', + ); + }); + }); + // ─── /help ────────────────────────────────────────────────────────────── describe('/help', () => { diff --git a/test/daemon-rename-route.test.ts b/test/daemon-rename-route.test.ts index 7b1897d7c..5edf18e36 100644 --- a/test/daemon-rename-route.test.ts +++ b/test/daemon-rename-route.test.ts @@ -44,6 +44,12 @@ const mocks = vi.hoisted(() => { : undefined )), forkWorker: vi.fn(), + probeCodexFastServiceTier: vi.fn(async () => ({ + ok: true as const, + enabled: true, + serviceTier: 'priority', + })), + requestWorkerFastModeChange: vi.fn(), createSession: vi.fn((chatId: string, rootMessageId: string, title: string, chatType?: 'group' | 'p2p') => ({ sessionId: `sess-fake-${++seq}`, chatId, @@ -88,6 +94,15 @@ vi.mock('../src/core/worker-pool.js', async () => { return { ...actual, forkWorker: (...args: any[]) => mocks.forkWorker(...args) }; }); +vi.mock('../src/core/fast-mode-control.js', async () => { + const actual = await vi.importActual<any>('../src/core/fast-mode-control.js'); + return { + ...actual, + probeCodexFastServiceTier: (...args: any[]) => mocks.probeCodexFastServiceTier(...args), + requestWorkerFastModeChange: (...args: any[]) => mocks.requestWorkerFastModeChange(...args), + }; +}); + import { registerBot } from '../src/bot-registry.js'; import { sessionKey } from '../src/core/types.js'; import { @@ -231,6 +246,11 @@ describe('/rename production routing — must not pre-create a session (review P mocks.sendMessage.mockResolvedValue('om_top'); mocks.getChatMode.mockResolvedValue('group'); mocks.getChatNameAndMode.mockResolvedValue({ name: null, mode: 'group' }); + mocks.probeCodexFastServiceTier.mockResolvedValue({ + ok: true, + enabled: true, + serviceTier: 'priority', + }); activeSessions.clear(); const bot = registerBot({ larkAppId: APP, @@ -315,6 +335,103 @@ describe('/rename production routing — must not pre-create a session (review P expect(activeSessions.has(sessionKey('om_new_2', APP))).toBe(true); }); + it('/fast status and invalid input never create phantom sessions on either route', async () => { + await handleNewTopic( + makeEventData('om_fast_status', '/fast status'), + makeCtx('om_fast_status', 'om_fast_status'), + ); + await handleThreadReply( + makeEventData('om_fast_invalid', '/fast turbo', 'om_fast_root'), + makeCtx('om_fast_root', 'om_fast_invalid'), + ); + + expect(mocks.createSession).not.toHaveBeenCalled(); + expect(activeSessions.size).toBe(0); + expect(repliedText()).toContain('没有活跃的会话'); + expect(repliedText()).toContain('/fast on'); + }); + + it('non-Codex /fast on reports unsupported without creating a session', async () => { + await handleNewTopic( + makeEventData('om_fast_claude', '/fast on'), + makeCtx('om_fast_claude', 'om_fast_claude'), + ); + + expect(mocks.createSession).not.toHaveBeenCalled(); + expect(mocks.probeCodexFastServiceTier).not.toHaveBeenCalled(); + expect(activeSessions.size).toBe(0); + expect(repliedText()).toContain('仅支持由 Botmux 管理'); + }); + + it('Riff-backed /fast on reports unsupported without probing or creating a session', async () => { + const bot = registerBot({ + larkAppId: APP, + larkAppSecret: 's', + cliId: 'riff', + backendType: 'riff', + allowedUsers: [OWNER], + oncallChats: [{ chatId: CHAT, workingDir: '/tmp' }], + }); + bot.resolvedAllowedUsers = [OWNER]; + + await handleNewTopic( + makeEventData('om_fast_riff', '/fast on'), + makeCtx('om_fast_riff', 'om_fast_riff'), + ); + + expect(mocks.probeCodexFastServiceTier).not.toHaveBeenCalled(); + expect(mocks.createSession).not.toHaveBeenCalled(); + expect(activeSessions.size).toBe(0); + expect(repliedText()).toContain('仅支持由 Botmux 管理'); + }); + + it('valid Codex /fast on preflights the model, then creates exactly one Fast session', async () => { + const bot = registerBot({ + larkAppId: APP, + larkAppSecret: 's', + cliId: 'codex', + allowedUsers: [OWNER], + oncallChats: [{ chatId: CHAT, workingDir: '/tmp' }], + }); + bot.resolvedAllowedUsers = [OWNER]; + + await handleNewTopic( + makeEventData('om_fast_on', '/fast on'), + makeCtx('om_fast_on', 'om_fast_on'), + ); + + expect(mocks.probeCodexFastServiceTier).toHaveBeenCalledTimes(1); + expect(mocks.createSession).toHaveBeenCalledTimes(1); + const ds = activeSessions.get(sessionKey('om_fast_on', APP)); + expect(ds?.worker).toBeNull(); + expect(ds?.session.fastMode).toBe(true); + expect(ds?.session.fastServiceTier).toBe('priority'); + }); + + it('unsupported Codex model fails before session creation', async () => { + const bot = registerBot({ + larkAppId: APP, + larkAppSecret: 's', + cliId: 'codex', + allowedUsers: [OWNER], + oncallChats: [{ chatId: CHAT, workingDir: '/tmp' }], + }); + bot.resolvedAllowedUsers = [OWNER]; + mocks.probeCodexFastServiceTier.mockResolvedValue({ + ok: false, + reason: 'unsupported_model', + }); + + await handleThreadReply( + makeEventData('om_fast_unsupported', '/fast on', 'om_fast_unsupported_root'), + makeCtx('om_fast_unsupported_root', 'om_fast_unsupported'), + ); + + expect(mocks.createSession).not.toHaveBeenCalled(); + expect(activeSessions.size).toBe(0); + expect(repliedText()).toContain('当前模型不支持'); + }); + it('new topic: passes the accepted Lark message id into the first worker', async () => { await handleNewTopic( makeEventData('om_workflow_new', '/workflow new 修复首轮授权'), diff --git a/test/dashboard-attention-signals.test.ts b/test/dashboard-attention-signals.test.ts index 956179f31..3309085e5 100644 --- a/test/dashboard-attention-signals.test.ts +++ b/test/dashboard-attention-signals.test.ts @@ -218,10 +218,10 @@ describe('attention signals', () => { const src = readFileSync(new URL('../src/daemon.ts', import.meta.url), 'utf-8'); const start = src.indexOf('async function handleThreadReply('); expect(start).toBeGreaterThanOrEqual(0); - // 20000：窗口需罩住函数头到最后一个拦截点 (findPendingAskByAnchor) 的全部 + // 30000：窗口需罩住函数头到最后一个拦截点 (findPendingAskByAnchor) 的全部 // 源码——passthrough 冷启动等合法插入会把后续 marker 往后推，窗口太紧会误报。 // 语义断言不变：clear 在所有拦截点之前。 - const region = src.slice(start, start + 20000); + const region = src.slice(start, start + 30000); const clearIdx = region.indexOf('clearAgentAttentionForHumanInbound();'); expect(clearIdx).toBeGreaterThanOrEqual(0); for (const marker of [ diff --git a/test/fast-mode-apply-transaction.test.ts b/test/fast-mode-apply-transaction.test.ts new file mode 100644 index 000000000..3f746b385 --- /dev/null +++ b/test/fast-mode-apply-transaction.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest'; +import { + FastModeApplyCancelledError, + FastModeApplyTransaction, +} from '../src/core/fast-mode-apply-transaction.js'; + +describe('Fast Mode apply transaction', () => { + it('rejects a native probe result that arrives after the exact request was cancelled', async () => { + let resolveProbe!: (serviceTier: string) => void; + const probe = new Promise<string>(resolve => { + resolveProbe = resolve; + }); + const transaction = new FastModeApplyTransaction('req-native-late'); + + const guardedProbe = transaction.waitFor(probe); + + expect(transaction.cancel('req-other', 'daemon_timeout')).toBe(false); + expect(transaction.cancel('req-native-late', 'daemon_timeout')).toBe(true); + resolveProbe('priority'); + + await expect(guardedProbe).rejects.toEqual( + new FastModeApplyCancelledError('req-native-late', 'daemon_timeout'), + ); + expect(transaction.signal.aborted).toBe(true); + }); +}); diff --git a/test/fast-mode-control.test.ts b/test/fast-mode-control.test.ts new file mode 100644 index 000000000..08ec1ebfd --- /dev/null +++ b/test/fast-mode-control.test.ts @@ -0,0 +1,129 @@ +import { chmodSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import { beforeAll, describe, expect, it } from 'vitest'; +import { + fastModeStateNeedsReconciliation, + fastModeSessionSupported, + parseFastModeAction, + probeCodexFastServiceTier, + requestWorkerFastModeChange, +} from '../src/core/fast-mode-control.js'; +import { acknowledgeFastModeResult } from '../src/core/fast-mode-handshake.js'; + +const FIXTURE = fileURLToPath(new URL('./fixtures/fake-codex-rpc-server.mjs', import.meta.url)); +beforeAll(() => { chmodSync(FIXTURE, 0o755); }); + +describe('Fast Mode control', () => { + it('parses only the supported command actions', () => { + expect(parseFastModeAction('/fast')).toBe('toggle'); + expect(parseFastModeAction('/fast on')).toBe('on'); + expect(parseFastModeAction('/FAST OFF')).toBe('off'); + expect(parseFastModeAction('/fast status')).toBe('status'); + expect(parseFastModeAction('/fast turbo')).toBe('invalid'); + }); + + it('rejects non-Codex, adopted, Aiden-wrapped, and Riff-backed sessions', () => { + expect(fastModeSessionSupported({ cliId: 'codex' })).toBe(true); + expect(fastModeSessionSupported({ cliId: 'claude-code' })).toBe(false); + expect(fastModeSessionSupported({ cliId: 'codex', adopted: true })).toBe(false); + expect(fastModeSessionSupported({ cliId: 'codex', wrapperCli: 'aiden x codex' })).toBe(false); + expect(fastModeSessionSupported({ cliId: 'codex', backendType: 'riff' })).toBe(false); + }); + + it('requires executor confirmation for both enabled and disabled legacy states', () => { + expect(fastModeStateNeedsReconciliation({ + enabled: false, + })).toBe(true); + expect(fastModeStateNeedsReconciliation({ + enabled: false, + stateVersion: 1, + })).toBe(false); + expect(fastModeStateNeedsReconciliation({ + enabled: true, + serviceTier: 'priority', + stateVersion: 1, + })).toBe(false); + expect(fastModeStateNeedsReconciliation({ + enabled: true, + stateVersion: 1, + })).toBe(true); + }); + + it('resolves the selected model Fast tier from app-server model/list', async () => { + await expect(probeCodexFastServiceTier({ + cliBin: FIXTURE, + cwd: '/tmp', + env: process.env, + model: 'gpt-fast', + })).resolves.toEqual({ + ok: true, + enabled: true, + serviceTier: 'priority', + }); + }, 20_000); + + it('fails closed when the selected model has no Fast tier', async () => { + await expect(probeCodexFastServiceTier({ + cliBin: FIXTURE, + cwd: '/tmp', + env: process.env, + model: 'gpt-standard', + })).resolves.toEqual({ + ok: false, + reason: 'unsupported_model', + }); + }, 20_000); + + it('waits for the worker ACK instead of treating IPC send as success', async () => { + let sent: any; + const worker = { + killed: false, + connected: true, + send(message: unknown, callback: (error: Error | null) => void) { + sent = message; + callback(null); + }, + } as any; + + const pending = requestWorkerFastModeChange(worker, true, 1_000); + await Promise.resolve(); + expect(sent).toMatchObject({ type: 'set_fast_mode', enabled: true }); + + acknowledgeFastModeResult({ + type: 'fast_mode_result', + requestId: sent.requestId, + ok: true, + enabled: true, + serviceTier: 'priority', + }); + await expect(pending).resolves.toEqual({ + ok: true, + enabled: true, + serviceTier: 'priority', + }); + }); + + it('cancels the exact worker transaction when the daemon wait expires', async () => { + const sent: any[] = []; + const worker = { + killed: false, + connected: true, + send(message: unknown, callback?: (error: Error | null) => void) { + sent.push(message); + callback?.(null); + }, + } as any; + + await expect(requestWorkerFastModeChange(worker, true, 5)).resolves.toEqual({ + ok: false, + reason: 'not_ready', + }); + + expect(sent).toHaveLength(2); + expect(sent[0]).toMatchObject({ type: 'set_fast_mode', enabled: true }); + expect(sent[1]).toEqual({ + type: 'cancel_fast_mode', + requestId: sent[0].requestId, + }); + }); +}); diff --git a/test/fast-mode-handshake.test.ts b/test/fast-mode-handshake.test.ts new file mode 100644 index 000000000..da7e18937 --- /dev/null +++ b/test/fast-mode-handshake.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; +import { + acknowledgeFastModeResult, + cancelFastModeResult, + isFastModeResultPending, + waitForFastModeResult, +} from '../src/core/fast-mode-handshake.js'; + +describe('Fast Mode IPC handshake', () => { + it('resolves only the exact request ACK', async () => { + const pending = waitForFastModeResult('req-1', 1_000); + + expect(isFastModeResultPending('req-1')).toBe(true); + expect(isFastModeResultPending('other')).toBe(false); + expect(acknowledgeFastModeResult({ + type: 'fast_mode_result', + requestId: 'other', + ok: true, + enabled: true, + serviceTier: 'priority', + })).toBe(false); + expect(acknowledgeFastModeResult({ + type: 'fast_mode_result', + requestId: 'req-1', + ok: true, + enabled: true, + serviceTier: 'priority', + })).toBe(true); + expect(isFastModeResultPending('req-1')).toBe(false); + await expect(pending).resolves.toEqual({ + ok: true, + enabled: true, + serviceTier: 'priority', + }); + }); + + it('fails closed on timeout or send cancellation', async () => { + const cancelled: string[] = []; + await expect(waitForFastModeResult( + 'req-timeout', + 5, + () => cancelled.push('req-timeout'), + )).resolves.toEqual({ + ok: false, + reason: 'not_ready', + }); + expect(cancelled).toEqual(['req-timeout']); + + const pending = waitForFastModeResult('req-cancel', 1_000); + expect(cancelFastModeResult('req-cancel')).toBe(true); + await expect(pending).resolves.toEqual({ + ok: false, + reason: 'not_ready', + }); + }); +}); diff --git a/test/fast-mode-restart-watchdog.test.ts b/test/fast-mode-restart-watchdog.test.ts new file mode 100644 index 000000000..987a219b1 --- /dev/null +++ b/test/fast-mode-restart-watchdog.test.ts @@ -0,0 +1,49 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + FAST_MODE_RESTART_ACK_TIMEOUT_MS, + FastModeRestartWatchdog, +} from '../src/core/fast-mode-restart-watchdog.js'; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('Fast Mode native restart watchdog', () => { + it('expires only the active request after the bounded worker deadline', async () => { + vi.useFakeTimers(); + const expired: string[] = []; + const watchdog = new FastModeRestartWatchdog(); + + watchdog.arm('req-active', requestId => expired.push(requestId)); + await vi.advanceTimersByTimeAsync(FAST_MODE_RESTART_ACK_TIMEOUT_MS - 1); + expect(expired).toEqual([]); + + await vi.advanceTimersByTimeAsync(1); + expect(expired).toEqual(['req-active']); + expect(watchdog.clear('req-active')).toBe(false); + }); + + it('cannot clear a newer transaction with a stale request id', async () => { + vi.useFakeTimers(); + const expired: string[] = []; + const watchdog = new FastModeRestartWatchdog(25); + + watchdog.arm('req-new', requestId => expired.push(requestId)); + expect(watchdog.clear('req-old')).toBe(false); + await vi.advanceTimersByTimeAsync(25); + + expect(expired).toEqual(['req-new']); + }); + + it('does not expire a transaction confirmed at the prompt', async () => { + vi.useFakeTimers(); + const expired: string[] = []; + const watchdog = new FastModeRestartWatchdog(25); + + watchdog.arm('req-confirmed', requestId => expired.push(requestId)); + expect(watchdog.clear('req-confirmed')).toBe(true); + await vi.advanceTimersByTimeAsync(25); + + expect(expired).toEqual([]); + }); +}); diff --git a/test/fixtures/fake-codex-rpc-server.mjs b/test/fixtures/fake-codex-rpc-server.mjs index 0621aad7c..9c9aa6ba9 100755 --- a/test/fixtures/fake-codex-rpc-server.mjs +++ b/test/fixtures/fake-codex-rpc-server.mjs @@ -11,8 +11,8 @@ // (lets a test assert model/effort are SUPPRESSED // on resume) import { createServer } from 'node:http'; +import { appendFileSync, writeFileSync } from 'node:fs'; import { WebSocketServer } from 'ws'; -import { writeFileSync } from 'node:fs'; const listenArg = process.argv[process.argv.indexOf('--listen') + 1] || ''; const m = listenArg.match(/ws:\/\/127\.0\.0\.1:(\d+)/); @@ -23,6 +23,8 @@ const PREVIEW_DELAY_READS = Number(process.env.FAKE_PREVIEW_DELAY_READS ?? '0'); const UPDATED_DELAY_READS = Number(process.env.FAKE_UPDATED_DELAY_READS ?? '0'); const UPDATED_BEFORE = Number(process.env.FAKE_UPDATED_BEFORE ?? '100'); const UPDATED_AFTER = Number(process.env.FAKE_UPDATED_AFTER ?? '101'); +const SETTINGS_UPDATE_ACK_DELAY_MS = Number(process.env.FAKE_SETTINGS_UPDATE_ACK_DELAY_MS ?? '0'); +const RPC_LOG = process.env.FAKE_RPC_LOG; let threadReadAttempt = 0; let currentThreadName; const REQUEST_USER_INPUT = process.env.FAKE_REQUEST_USER_INPUT === '1'; @@ -36,6 +38,7 @@ wss.on('connection', (ws) => { let pendingTurnReply; ws.on('message', (data) => { let msg; try { msg = JSON.parse(data.toString()); } catch { return; } + if (RPC_LOG) appendFileSync(RPC_LOG, `${JSON.stringify(msg)}\n`); if (REQUEST_USER_INPUT && msg.id === 900 && (msg.result !== undefined || msg.error !== undefined)) { if (!pendingTurnReply) return; // Real traex 0.200.19 normalizes ANY reply to requestUserInput (empty @@ -57,6 +60,25 @@ wss.on('connection', (ws) => { const reply = (result) => ws.send(JSON.stringify({ jsonrpc: '2.0', id: msg.id, result })); switch (msg.method) { case 'initialize': return reply({ ok: true }); + case 'model/list': return reply({ + data: [ + { + id: 'gpt-fast', + model: 'gpt-fast', + isDefault: true, + serviceTiers: [ + { id: 'priority', name: 'Fast', description: 'faster' }, + ], + }, + { + id: 'gpt-standard', + model: 'gpt-standard', + isDefault: false, + serviceTiers: [], + }, + ], + nextCursor: null, + }); case 'thread/start': { if (process.env.FAKE_THREAD_CONFIG_FILE) { try { writeFileSync(process.env.FAKE_THREAD_CONFIG_FILE, JSON.stringify(msg.params ?? {})); } catch { /* test-only */ } @@ -69,6 +91,11 @@ wss.on('connection', (ws) => { } return reply({ thread: { id: msg.params?.threadId ?? 'thread-fake-1' } }); } + case 'thread/settings/update': + if (SETTINGS_UPDATE_ACK_DELAY_MS > 0) { + return setTimeout(() => reply({}), SETTINGS_UPDATE_ACK_DELAY_MS); + } + return reply({}); case 'thread/read': threadReadAttempt += 1; return reply({ thread: { diff --git a/test/session-lifecycle-start.test.ts b/test/session-lifecycle-start.test.ts index 7ac3f9619..2828f4c30 100644 --- a/test/session-lifecycle-start.test.ts +++ b/test/session-lifecycle-start.test.ts @@ -1570,6 +1570,20 @@ describe('forkWorker session agent config freeze', () => { })); }); + it('passes the persisted Session Fast Mode into the Codex worker', () => { + const ds = makeDs(); + ds.session.fastMode = true; + + forkWorker(ds, 'hello', false); + + const worker = forkMock.mock.results.at(-1)!.value; + expect(worker.send).toHaveBeenCalledWith(expect.objectContaining({ + type: 'init', + cliId: 'codex', + fastMode: true, + })); + }); + it('fills wrapper and model on fresh sessions that already stamped cliId', () => { const ds = makeDs(); ds.session.cliId = 'codex' as any; diff --git a/test/write-input.test.ts b/test/write-input.test.ts index c077c8474..bec21dd8c 100644 --- a/test/write-input.test.ts +++ b/test/write-input.test.ts @@ -1089,6 +1089,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', '019dd3e2-f2da-7592-86b5-a43d4cd0772f', ]); }); @@ -1110,6 +1112,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', '019dd3e2-f2da-7592-86b5-a43d4cd0772f', ]); }); @@ -1129,6 +1133,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', 'new-codex-session', ]); }); @@ -1149,6 +1155,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="custom-botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', 'custom-codex-session', ]); @@ -1174,6 +1182,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', ]); }); @@ -1192,6 +1202,8 @@ describe('codex writeInput submission confirmation', () => { 'shell_environment_policy.set.BOTMUX_SESSION_ID="botmux-session"', '-c', 'check_for_update_on_startup=false', + '-c', + 'service_tier="default"', '-C', '/repo/root', ]);