diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 8a5af9e7..19c81e9f 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -46,6 +46,7 @@ import type { Thread, ThreadGoal, ThreadGoalStatus, + ThreadResumeParams, ThreadSourceKind, TurnCompletedNotification, TurnSteerResponse, @@ -59,8 +60,23 @@ import {arePathBasenamesEqual, arePathsEqual, isAbsolutePathLike} from "./PathUt import {CodexSubagentSubscriptions} from "./subagents/CodexSubagentSubscriptions"; import {forkSession as runForkSession} from "./SessionFork"; import type {SessionMetadata, SessionMetadataWithThread} from "./SessionMetadata"; +import {isMissingRolloutError, isUnknownThreadError} from "./CodexThreadErrors"; export type {SessionMetadata, SessionMetadataWithThread} from "./SessionMetadata"; +/** + * The slice of `thread/resume` the session layer consumes, plus whether Codex + * actually had a rollout for the thread. See {@link CodexAcpClient.resumeThread}. + */ +type ResumedThread = { + thread: Thread; + model: string | null; + modelProvider: string; + reasoningEffort: ReasoningEffort | null; + serviceTier: string | null; + turnsBackwardsCursor: string | null; + materialized: boolean; +}; + /** * Well-known provider id for the client-configurable custom LLM gateway. * This is the only provider exposed through the ACP `providers/*` methods and @@ -516,11 +532,61 @@ export class CodexAcpClient { return settingsModelProvider?.config?.model_provider ?? null; } + /** + * `thread/resume`, with a fallback for a thread Codex has not materialized + * on disk yet. + * + * Codex writes a thread's rollout file on its first user message, so + * `thread/resume` fails with "no rollout found" for a session that was + * created but never prompted. Such a thread is still live in the + * app-server -- and still subscribed, since `thread/start` subscribed it -- + * so `thread/read` answers for it and gives back the same state resume + * would have. A thread id Codex has genuinely never seen fails both calls, + * and the original resume error is what the caller sees. + */ + private async resumeThread(params: ThreadResumeParams): Promise { + try { + const response = await this.codexClient.threadResume(params); + return { + thread: response.thread, + model: response.model, + modelProvider: response.modelProvider, + reasoningEffort: response.reasoningEffort, + serviceTier: response.serviceTier, + turnsBackwardsCursor: response.turnsBackwardsCursor, + materialized: true, + }; + } catch (err) { + if (!isMissingRolloutError(err)) throw err; + let response; + try { + response = await this.codexClient.threadRead({threadId: params.threadId}); + } catch { + throw err; + } + logger.log("Thread has no rollout yet; resumed it from its live app-server state", { + threadId: params.threadId, + }); + return { + thread: response.thread, + model: response.thread.model, + modelProvider: response.thread.modelProvider, + reasoningEffort: response.thread.reasoningEffort, + serviceTier: null, + // An unmaterialized thread has no persisted history to hydrate: + // `thread/turns/list` rejects it outright ("not materialized + // yet"), and there is nothing to list either way. + turnsBackwardsCursor: null, + materialized: false, + }; + } + } + async resumeSession(request: acp.ResumeSessionRequest, onSubscribed?: () => void): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); - const response = await this.codexClient.threadResume({ + const response = await this.resumeThread({ excludeTurns: true, config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), cwd: request.cwd, @@ -560,7 +626,7 @@ export class CodexAcpClient { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); - const response = await this.codexClient.threadResume({ + const response = await this.resumeThread({ excludeTurns: true, config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), cwd: request.cwd, @@ -570,7 +636,9 @@ export class CodexAcpClient { onSubscribed?.(); // Resume cursors bound durable history; later turns arrive through live events. // A null paginated cursor means there was no durable history at resume time. - const thread = response.thread.historyMode === "paginated" + const thread = !response.materialized + ? {...response.thread, turns: []} + : response.thread.historyMode === "paginated" ? { ...response.thread, turns: response.turnsBackwardsCursor === null @@ -632,7 +700,21 @@ export class CodexAcpClient { } async deleteSession(sessionId: string): Promise { - await this.codexClient.threadArchive({threadId: sessionId}); + try { + await this.codexClient.threadArchive({threadId: sessionId}); + } catch (err) { + // Deleting a session is idempotent: an id Codex has no persisted + // thread for has nothing left to archive. That covers a session + // that was created but never prompted (Codex materializes the + // rollout on the first user message), an already-deleted session, + // and an ACP session id that is not a Codex thread id at all -- + // ACP session ids are opaque strings, Codex thread ids are UUIDs. + if (!isUnknownThreadError(err)) throw err; + logger.log("Delete request for a session Codex has no persisted thread for; treating as deleted", { + sessionId, + reason: err instanceof Error ? err.message : String(err), + }); + } } async renameSession(sessionId: string, name: string): Promise { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 694c44df..913c3e51 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -20,6 +20,7 @@ import { type UrlElicitationRequester } from "./CodexAcpClient"; import {CodexAppServerClient, type McpStartupResult} from "./CodexAppServerClient"; +import {isNoActiveTurnError} from "./CodexThreadErrors"; import {type CodexConnection, startCodexConnection} from "./CodexJsonRpcConnection"; import {type AcpClientConnection, ACPSessionConnection, type UpdateSessionEvent} from "./ACPSessionConnection"; import type {InputModality, ReasoningEffort, ServerNotification} from "./app-server"; @@ -215,6 +216,21 @@ export interface SessionFailure { const CODEX_PROCESS_EXITED_ERROR_CODE = 1001; +/** + * How long `session/load` waits for an in-flight title generation to settle + * before answering anyway. Generous enough for a title model round-trip, short + * enough that a wedged generation cannot hold a load open. + */ +const TITLE_GENERATION_SETTLE_TIMEOUT_MS = 10_000; + +/** + * Backoff for re-sending `turn/interrupt` when Codex reports the turn is not + * interruptible yet. Covers the sub-second window between a turn's first + * streamed event -- which is what prompts a client to cancel in the first + * place -- and Codex registering the turn as interruptible. + */ +const NO_ACTIVE_TURN_RETRY_DELAYS_MS = [25, 50, 100, 200, 400]; + function clientSupportsTypedSessionFailures(capabilities: acp.ClientCapabilities | null): boolean { return clientSupportsAirCapability(capabilities, AIR_SESSION_FAILURE_KEY); } @@ -783,6 +799,10 @@ export class CodexAcpServer { await this.providerUpdate; } logger.log("Loading session...", {sessionId: params.sessionId}); + // Captured before the load installs a fresh SessionState: a title + // generation started by an earlier turn on this session belongs to the + // state being replaced, and has to settle before we answer. + const previousTitleGen = this.sessions.get(params.sessionId)?.titleGen; const { sessionId, modelState, @@ -792,6 +812,9 @@ export class CodexAcpServer { await this.streamThreadHistory(sessionId, thread); await this.getSessionState(sessionId).asyncTasks.reconcile(); + // A load response means "the replay is complete"; a late rename echo + // from a still-running title generation would arrive after it. + await previousTitleGen?.waitForIdle(TITLE_GENERATION_SETTLE_TIMEOUT_MS); logger.log("Session loaded", { sessionId: sessionId, @@ -2652,17 +2675,40 @@ export class CodexAcpServer { turn: { threadId: string, turnId: string }, requestName: "Cancel" | "Close", ): Promise { - try { - await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ - threadId: turn.threadId, - turnId: turn.turnId, - })); - logger.log(`${requestName} - turnInterrupt succeeded`, { - sessionId: turn.threadId, - currentTurnId: turn.turnId, - }); - } catch (err) { - logger.error(`${requestName} - turnInterrupt failed`, err); + for (let attempt = 0; ; attempt++) { + try { + await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ + threadId: turn.threadId, + turnId: turn.turnId, + })); + logger.log(`${requestName} - turnInterrupt succeeded`, { + sessionId: turn.threadId, + currentTurnId: turn.turnId, + }); + return; + } catch (err) { + const retryDelay = requestName === "Cancel" + && isNoActiveTurnError(err) + && attempt < NO_ACTIVE_TURN_RETRY_DELAYS_MS.length + && this.activePrompts.has(turn.threadId) + ? NO_ACTIVE_TURN_RETRY_DELAYS_MS[attempt]! + : null; + if (retryDelay === null) { + logger.error(`${requestName} - turnInterrupt failed`, err); + return; + } + // The cancel raced the turn's registration in Codex: the prompt + // is still in flight, so the turn is about to become + // interruptible. Dropping the cancel here would let the turn run + // to completion and answer `end_turn`, which ACP forbids after a + // `session/cancel`. + logger.log(`${requestName} - turn not interruptible yet, retrying`, { + sessionId: turn.threadId, + currentTurnId: turn.turnId, + attempt, + }); + await new Promise(resolve => setTimeout(resolve, retryDelay)); + } } } @@ -2695,16 +2741,7 @@ export class CodexAcpServer { }); } try { - await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ - threadId: sessionState.sessionId, - turnId, - })); - logger.log(`${requestName} - turnInterrupt succeeded`, { - sessionId: sessionState.sessionId, - currentTurnId: turnId, - }); - } catch (err) { - logger.error(`${requestName} - turnInterrupt failed`, err); + await this.requestTurnInterrupt({threadId: sessionState.sessionId, turnId}, requestName); } finally { if (resolveInterruptedTurn) { this.codexAcpClient.resolveTurnInterrupted({ diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index b6b2275f..94c49dad 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -568,6 +568,7 @@ export class CodexEventHandler { this.sessionState.sessionTitleSource = notification.params.threadName == null ? "unset" : "explicit"; + this.sessionState.titleGen?.observeRename(); return { sessionUpdate: "session_info_update", title: notification.params.threadName ?? null, diff --git a/src/CodexThreadErrors.ts b/src/CodexThreadErrors.ts new file mode 100644 index 00000000..97253abb --- /dev/null +++ b/src/CodexThreadErrors.ts @@ -0,0 +1,64 @@ +/** + * Classifiers for the Codex app-server errors that ACP has to translate into + * something other than a bare `-32603 Internal error`. + * + * Codex reports these as plain JSON-RPC error messages with no machine-readable + * discriminator, so matching on the message text is the only option; each + * predicate keeps the match anchored on the stable part of the phrasing. + */ + +function errorText(err: unknown): string { + if (err instanceof Error) return err.message; + if (typeof err === "string") return err; + if (err !== null && typeof err === "object" && "message" in err) { + return String((err as { message: unknown }).message); + } + return ""; +} + +/** + * Codex materializes a thread's rollout file lazily, on the thread's first + * user message. `thread/resume` and `thread/archive` read that file, so both + * fail this way for a thread that was started but never prompted -- and for a + * thread id Codex has simply never seen. + */ +export function isMissingRolloutError(err: unknown): boolean { + return errorText(err).includes("no rollout found for thread id"); +} + +/** + * `thread/read` answers this for a thread id that is well-formed but not + * currently loaded in the app-server process. + */ +export function isThreadNotLoadedError(err: unknown): boolean { + return errorText(err).includes("thread not loaded:"); +} + +/** + * Codex thread ids are UUIDs, so anything else is rejected before lookup. ACP + * session ids are opaque strings, so a client is free to send an id Codex + * cannot even parse. + */ +export function isInvalidThreadIdError(err: unknown): boolean { + const text = errorText(err); + return text.includes("invalid thread id:") || text.includes("invalid session id:"); +} + +/** + * `turn/interrupt` answers this both for a turn that has already finished and + * for one Codex has not registered as interruptible yet -- a `session/cancel` + * that lands in the window between the turn's first streamed event and that + * registration. + */ +export function isNoActiveTurnError(err: unknown): boolean { + return errorText(err).includes("no active turn to interrupt"); +} + +/** + * True when the error means "Codex has no persisted thread under this id" for + * any reason -- unparseable id, unknown id, or an id whose rollout was never + * materialized. + */ +export function isUnknownThreadError(err: unknown): boolean { + return isMissingRolloutError(err) || isThreadNotLoadedError(err) || isInvalidThreadIdError(err); +} diff --git a/src/TitleGenerator.ts b/src/TitleGenerator.ts index b606e9d0..9e001007 100644 --- a/src/TitleGenerator.ts +++ b/src/TitleGenerator.ts @@ -4,6 +4,11 @@ import type { Turn } from "./app-server/v2"; // Use cheap model to generate a title const TITLE_MODEL = "gpt-5.6-luna"; +// thread/name/set acks once Codex accepts the rename, but the thread/name/updated +// notification that echoes it back to the client can lag behind that ack. Cap how +// long generateAndPersist waits for the echo before giving up on it. +const RENAME_ECHO_TIMEOUT_MS = 5_000; + const TITLE_OUTPUT_SCHEMA = { type: "object", properties: { title: { type: "string" } }, @@ -22,6 +27,8 @@ const SYSTEM_PROMPT = export class TitleGenerator { private generated = false; + private inFlight: Promise | null = null; + private renameEchoResolve: (() => void) | null = null; constructor( private readonly client: CodexAppServerClient, @@ -38,6 +45,17 @@ export class TitleGenerator { this.generated = true; } + /** + * Call when a `thread/name/updated` notification arrives for this session. + * Unblocks {@link generateAndPersist}'s wait for the rename it just issued to + * be echoed back, so `waitForIdle` reflects "the client has seen the update" + * rather than just "the rename RPC was acknowledged". + */ + observeRename(): void { + this.renameEchoResolve?.(); + this.renameEchoResolve = null; + } + /** * Fire-and-forget hook — call after each turn completes. * Only acts on the first call for new sessions without an existing title. @@ -52,9 +70,39 @@ export class TitleGenerator { // "unknown": resumed session with indeterminate history — skip if (src === "explicit" || src === "unknown") return; this.generated = true; - this.generateAndPersist(userPromptText).catch(() => { - // title generation is best-effort; never surface errors to the user - }); + const run = this.generateAndPersist(userPromptText) + .catch(() => { + // title generation is best-effort; never surface errors to the user + }) + .finally(() => { + if (this.inFlight === run) this.inFlight = null; + }); + this.inFlight = run; + } + + /** + * Resolves once the fire-and-forget generation started by + * {@link onTurnCompleted} has finished, or after `timeoutMs`. + * + * Generation renames the thread, which Codex echoes back as a + * `session_info_update`. `session/load` has to finish replaying a session + * before it answers, so it awaits this first rather than letting a title + * from an earlier turn surface after the load response. + */ + async waitForIdle(timeoutMs: number): Promise { + const pending = this.inFlight; + if (pending === null) return; + let timer: ReturnType | undefined; + try { + await Promise.race([ + pending, + new Promise(resolve => { + timer = setTimeout(resolve, timeoutMs); + }), + ]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } } private async generateAndPersist(userPromptText: string): Promise { @@ -89,6 +137,15 @@ export class TitleGenerator { threadId: this.mainThreadId, name: title, }); + await this.waitForRenameEcho(RENAME_ECHO_TIMEOUT_MS); + } + + private async waitForRenameEcho(timeoutMs: number): Promise { + await new Promise(resolve => { + this.renameEchoResolve = resolve; + setTimeout(resolve, timeoutMs); + }); + this.renameEchoResolve = null; } } diff --git a/src/__tests__/CodexACPAgent/cancel-turn-registration-race.test.ts b/src/__tests__/CodexACPAgent/cancel-turn-registration-race.test.ts new file mode 100644 index 00000000..6f757942 --- /dev/null +++ b/src/__tests__/CodexACPAgent/cancel-turn-registration-race.test.ts @@ -0,0 +1,125 @@ +import {describe, expect, it, vi} from "vitest"; +import {createCodexMockTestFixture, createTestModel, type CodexMockTestFixture} from "../acp-test-utils"; +import type {CodexAcpServer} from "../../CodexAcpServer"; +import type {CodexAcpClient} from "../../CodexAcpClient"; +import type {TurnCompletedNotification} from "../../app-server/v2"; + +const sessionId = "session-id"; +const turnId = "turn-id"; + +// Codex answers this both for a turn that already finished and for one it has +// not registered as interruptible yet. +const NO_ACTIVE_TURN = new Error("no active turn to interrupt"); + +describe("session/cancel racing Codex's turn registration", () => { + it("retries the interrupt until Codex accepts it", async () => { + const turn = await startPrompt(); + const turnInterrupt = vi.spyOn(turn.codexAcpClient, "turnInterrupt") + .mockRejectedValueOnce(NO_ACTIVE_TURN) + .mockRejectedValueOnce(NO_ACTIVE_TURN) + .mockResolvedValueOnce(undefined); + + await turn.codexAcpAgent.cancel({sessionId}); + + expect(turnInterrupt).toHaveBeenCalledTimes(3); + expect(turnInterrupt).toHaveBeenLastCalledWith({threadId: sessionId, turnId}); + await turn.finish(); + }); + + it("gives up after the retry budget instead of spinning", async () => { + const turn = await startPrompt(); + const turnInterrupt = vi.spyOn(turn.codexAcpClient, "turnInterrupt") + .mockRejectedValue(NO_ACTIVE_TURN); + + await turn.codexAcpAgent.cancel({sessionId}); + + // One initial attempt plus one per configured backoff step. + expect(turnInterrupt).toHaveBeenCalledTimes(6); + await turn.finish(); + }); + + it("does not retry an interrupt that failed for another reason", async () => { + const turn = await startPrompt(); + const turnInterrupt = vi.spyOn(turn.codexAcpClient, "turnInterrupt") + .mockRejectedValue(new Error("codex app-server transport closed")); + + await turn.codexAcpAgent.cancel({sessionId}); + + expect(turnInterrupt).toHaveBeenCalledTimes(1); + await turn.finish(); + }); + + it("does not retry on close, which tears the session down anyway", async () => { + const turn = await startPrompt(); + const turnInterrupt = vi.spyOn(turn.codexAcpClient, "turnInterrupt") + .mockRejectedValue(NO_ACTIVE_TURN); + + const closed = turn.codexAcpAgent.closeSession({sessionId}); + await turn.finish(); + await closed; + + expect(turnInterrupt).toHaveBeenCalledTimes(1); + }); +}); + +/** + * Creates a session and leaves one prompt turn in flight -- the state a + * `session/cancel` actually arrives in. + */ +async function startPrompt(): Promise<{ + codexAcpAgent: CodexAcpServer, + codexAcpClient: CodexAcpClient, + finish: () => Promise, +}> { + const fixture: CodexMockTestFixture = createCodexMockTestFixture(); + const codexAcpAgent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + const appServer = fixture.getCodexAppServerClient(); + + vi.spyOn(codexAcpClient, "authRequired").mockResolvedValue(false); + vi.spyOn(codexAcpClient, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false}); + vi.spyOn(codexAcpClient, "newSession").mockResolvedValue({ + sessionId, + currentModelId: "model-id[medium]", + models: [createTestModel()], + collaborationMode: "default", + currentServiceTier: null, + additionalDirectories: [], + }); + await codexAcpAgent.newSession({cwd: "/test/cwd", mcpServers: []}); + + const inProgressTurn = { + id: turnId, + items: [], + itemsView: "notLoaded" as const, + status: "inProgress" as const, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; + vi.spyOn(appServer, "turnStart").mockResolvedValue({turn: inProgressTurn}); + let completeTurn: (value: TurnCompletedNotification) => void = () => {}; + vi.spyOn(appServer, "awaitTurnCompleted").mockReturnValue( + new Promise(resolve => { + completeTurn = resolve; + }) + ); + + const prompt = codexAcpAgent.prompt({sessionId, prompt: [{type: "text", text: "hello"}]}); + await vi.waitFor(() => { + expect(codexAcpAgent.getSessionState(sessionId).currentTurnId).toBe(turnId); + }); + + return { + codexAcpAgent, + codexAcpClient, + finish: async () => { + completeTurn({ + threadId: sessionId, + turn: {...inProgressTurn, status: "interrupted"}, + }); + await prompt; + }, + }; +} diff --git a/src/__tests__/CodexACPAgent/unmaterialized-thread.test.ts b/src/__tests__/CodexACPAgent/unmaterialized-thread.test.ts new file mode 100644 index 00000000..6004a60c --- /dev/null +++ b/src/__tests__/CodexACPAgent/unmaterialized-thread.test.ts @@ -0,0 +1,175 @@ +import {describe, expect, it, vi} from "vitest"; +import {createCodexMockTestFixture, createTestModel} from "../acp-test-utils"; +import type {Thread} from "../../app-server/v2"; + +// Codex materializes a thread's rollout file on the thread's first user +// message, so every rollout-backed call fails this way for a session that was +// created but never prompted -- the exact wording Codex 0.155 answers with. +const NO_ROLLOUT = (threadId: string) => new Error(`no rollout found for thread id ${threadId}`); +const INVALID_ID = new Error( + "invalid session id: invalid character: expected an optional prefix of `urn:uuid:` followed by [0-9a-fA-F-], found `t` at 1" +); + +const threadId = "01a0c48a-fc81-7b33-8d61-f4e2fd7c9b99"; + +function createLiveThread(): Thread { + return { + id: threadId, + sessionId: threadId, + parentThreadId: null, + threadSource: null, + originator: null, + forkedFromId: null, + preview: "", + ephemeral: false, + modelProvider: "openai", + model: "model-id", + reasoningEffort: "medium", + createdAt: 1, + updatedAt: 1, + recencyAt: null, + status: {type: "idle"}, + path: null, + cwd: "/test/cwd", + cliVersion: "0", + section: null, + sectionEnteredAt: null, + projectId: null, + historyMode: "paginated", + source: "cli", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [], + }; +} + +function createFixture() { + const fixture = createCodexMockTestFixture(); + const client = fixture.getCodexAcpClient(); + const appServer = fixture.getCodexAppServerClient(); + client.authRequired = vi.fn().mockResolvedValue(false); + client.getAccount = vi.fn().mockResolvedValue({account: null, requiresOpenaiAuth: false}); + client.listSkills = vi.fn().mockResolvedValue({data: []}); + appServer.listModels = vi.fn().mockResolvedValue({data: [createTestModel()], nextCursor: null}); + return {fixture, client, appServer}; +} + +describe("sessions Codex has not materialized on disk", () => { + it("resumes from the live thread when thread/resume finds no rollout", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadResume = vi.fn().mockRejectedValue(NO_ROLLOUT(threadId)); + appServer.threadRead = vi.fn().mockResolvedValue({thread: createLiveThread()}); + + const response = await fixture.getCodexAcpAgent().resumeSession({ + sessionId: threadId, + cwd: "/test/cwd", + mcpServers: [], + }); + + expect(appServer.threadRead).toHaveBeenCalledWith({threadId}); + expect(response.models?.currentModelId).toBe("model-id[medium]"); + }); + + it("loads an unmaterialized thread with empty history instead of paging it", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadResume = vi.fn().mockRejectedValue(NO_ROLLOUT(threadId)); + appServer.threadRead = vi.fn().mockResolvedValue({thread: createLiveThread()}); + appServer.threadTurnsList = vi.fn(); + + await expect(fixture.getCodexAcpAgent().loadSession({ + sessionId: threadId, + cwd: "/test/cwd", + mcpServers: [], + })).resolves.toBeDefined(); + + // `thread/turns/list` rejects an unmaterialized thread outright, and + // there is no history to hydrate anyway. + expect(appServer.threadTurnsList).not.toHaveBeenCalled(); + }); + + it("keeps reporting a thread id Codex has genuinely never seen", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadResume = vi.fn().mockRejectedValue(NO_ROLLOUT(threadId)); + appServer.threadRead = vi.fn().mockRejectedValue(new Error(`thread not loaded: ${threadId}`)); + + await expect(fixture.getCodexAcpAgent().resumeSession({ + sessionId: threadId, + cwd: "/test/cwd", + mcpServers: [], + })).rejects.toThrow("no rollout found for thread id"); + }); + + it("does not swallow an unrelated thread/resume failure", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadResume = vi.fn().mockRejectedValue(new Error("codex app-server transport closed")); + appServer.threadRead = vi.fn(); + + await expect(fixture.getCodexAcpAgent().resumeSession({ + sessionId: threadId, + cwd: "/test/cwd", + mcpServers: [], + })).rejects.toThrow("transport closed"); + expect(appServer.threadRead).not.toHaveBeenCalled(); + }); + + it("deletes a session that has no persisted rollout", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadArchive = vi.fn().mockRejectedValue(NO_ROLLOUT(threadId)); + + await expect(fixture.getCodexAcpAgent().deleteSession({sessionId: threadId})).resolves.toEqual({}); + }); + + it("deletes a session id Codex cannot even parse", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadArchive = vi.fn().mockRejectedValue(INVALID_ID); + + await expect(fixture.getCodexAcpAgent().deleteSession({ + sessionId: "tck-never-created-session", + })).resolves.toEqual({}); + }); + + it("still fails a delete that went wrong for any other reason", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadArchive = vi.fn().mockRejectedValue(new Error("disk is full")); + + await expect(fixture.getCodexAcpAgent().deleteSession({sessionId: threadId})) + .rejects.toThrow("disk is full"); + }); +}); + +describe("session/load and a title generation left over from an earlier turn", () => { + it("waits for the rename to land before answering", async () => { + const {fixture, appServer} = createFixture(); + appServer.threadResume = vi.fn().mockRejectedValue(NO_ROLLOUT(threadId)); + appServer.threadRead = vi.fn().mockResolvedValue({thread: createLiveThread()}); + appServer.threadStart = vi.fn().mockResolvedValue({ + thread: createLiveThread(), + model: "model-id", + modelProvider: "openai", + reasoningEffort: "medium", + serviceTier: null, + }); + const agent = fixture.getCodexAcpAgent(); + await agent.newSession({cwd: "/test/cwd", mcpServers: []}); + + let settled = false; + const titleGeneration = new Promise(resolve => { + setTimeout(() => { + settled = true; + resolve(); + }, 20); + }); + agent.getSessionState(threadId).titleGen = { + waitForIdle: () => titleGeneration, + markExistingTitle: () => {}, + } as unknown as NonNullable["titleGen"]>; + + await agent.loadSession({sessionId: threadId, cwd: "/test/cwd", mcpServers: []}); + + // The load response means "the replay is complete"; a rename echo from + // a still-running generation would arrive after it. + expect(settled).toBe(true); + }); +}); diff --git a/src/__tests__/CodexThreadErrors.test.ts b/src/__tests__/CodexThreadErrors.test.ts new file mode 100644 index 00000000..80f6595d --- /dev/null +++ b/src/__tests__/CodexThreadErrors.test.ts @@ -0,0 +1,52 @@ +import {describe, expect, it} from "vitest"; +import { + isInvalidThreadIdError, + isMissingRolloutError, + isThreadNotLoadedError, + isUnknownThreadError, +} from "../CodexThreadErrors"; + +// The literal wordings Codex 0.155 answers with, captured from a live +// app-server; the classifiers only promise to recognise these shapes. +const missingRollout = new Error("no rollout found for thread id 01a0c48a-fc81-7b33-8d61-f4e2fd7c9b99"); +const notLoaded = new Error("thread not loaded: 01a0c000-0000-7000-8000-000000000001"); +const invalidThreadId = new Error( + "invalid thread id: invalid character: expected an optional prefix of `urn:uuid:` followed by [0-9a-fA-F-], found `n` at 1" +); +const invalidSessionId = new Error( + "invalid session id: invalid character: expected an optional prefix of `urn:uuid:` followed by [0-9a-fA-F-], found `t` at 1" +); +const unrelated = new Error("stream disconnected before completion"); + +describe("CodexThreadErrors", () => { + it("recognises a thread whose rollout was never materialized", () => { + expect(isMissingRolloutError(missingRollout)).toBe(true); + expect(isMissingRolloutError(notLoaded)).toBe(false); + expect(isMissingRolloutError(unrelated)).toBe(false); + }); + + it("recognises a thread that is not loaded in the app-server", () => { + expect(isThreadNotLoadedError(notLoaded)).toBe(true); + expect(isThreadNotLoadedError(missingRollout)).toBe(false); + }); + + it("recognises an id Codex cannot parse as a thread id", () => { + expect(isInvalidThreadIdError(invalidThreadId)).toBe(true); + expect(isInvalidThreadIdError(invalidSessionId)).toBe(true); + expect(isInvalidThreadIdError(missingRollout)).toBe(false); + }); + + it("treats every 'no persisted thread' shape as an unknown thread", () => { + for (const err of [missingRollout, notLoaded, invalidThreadId, invalidSessionId]) { + expect(isUnknownThreadError(err)).toBe(true); + } + expect(isUnknownThreadError(unrelated)).toBe(false); + }); + + it("reads the message off non-Error rejections too", () => { + expect(isUnknownThreadError({code: -32600, message: notLoaded.message})).toBe(true); + expect(isUnknownThreadError(missingRollout.message)).toBe(true); + expect(isUnknownThreadError(undefined)).toBe(false); + expect(isUnknownThreadError(null)).toBe(false); + }); +}); diff --git a/src/__tests__/TitleGenerator.test.ts b/src/__tests__/TitleGenerator.test.ts new file mode 100644 index 00000000..9dbcf596 --- /dev/null +++ b/src/__tests__/TitleGenerator.test.ts @@ -0,0 +1,79 @@ +import {describe, expect, it, vi} from "vitest"; +import {TitleGenerator} from "../TitleGenerator"; +import type {CodexAppServerClient} from "../CodexAppServerClient"; + +function deferred(): {promise: Promise; resolve: (value: T) => void} { + let resolve: (value: T) => void = () => {}; + const promise = new Promise(innerResolve => { + resolve = innerResolve; + }); + return {promise, resolve}; +} + +function createGenerator(client: Partial) { + return new TitleGenerator(client as CodexAppServerClient, "thread-id", "/test/cwd", () => "unset"); +} + +describe("TitleGenerator.waitForIdle", () => { + it("returns immediately when nothing is generating", async () => { + const generator = createGenerator({}); + + await expect(generator.waitForIdle(50)).resolves.toBeUndefined(); + }); + + it("waits for the rename echo notification before settling", async () => { + const turn = deferred<{turn: {items: {type: string; text: string}[]}}>(); + const threadSetName = vi.fn().mockResolvedValue({}); + const generator = createGenerator({ + threadStart: vi.fn().mockResolvedValue({thread: {id: "ephemeral"}}), + runTurn: vi.fn().mockReturnValue(turn.promise), + threadSetName, + } as unknown as Partial); + + generator.onTurnCompleted("hello"); + let settled = false; + const idle = generator.waitForIdle(5_000).then(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + + turn.resolve({turn: {items: [{type: "agentMessage", text: '{"title":"A short title"}'}]}}); + // Flush the microtask chain (extract title -> threadSetName -> start + // waiting for the echo) without resolving the echo itself yet. + for (let i = 0; i < 10; i++) { + await Promise.resolve(); + } + expect(threadSetName).toHaveBeenCalledWith({threadId: "thread-id", name: "A short title"}); + expect(settled).toBe(false); + + // The thread/name/updated notification for this rename arrives. + generator.observeRename(); + await idle; + expect(settled).toBe(true); + }); + + it("gives up after the timeout rather than holding the caller open", async () => { + const generator = createGenerator({ + threadStart: vi.fn().mockResolvedValue({thread: {id: "ephemeral"}}), + runTurn: vi.fn().mockReturnValue(new Promise(() => {})), + threadSetName: vi.fn(), + } as unknown as Partial); + + generator.onTurnCompleted("hello"); + + await expect(generator.waitForIdle(20)).resolves.toBeUndefined(); + }); + + it("stops waiting once a failed generation has settled", async () => { + const generator = createGenerator({ + threadStart: vi.fn().mockRejectedValue(new Error("no ephemeral threads")), + runTurn: vi.fn(), + threadSetName: vi.fn(), + } as unknown as Partial); + + generator.onTurnCompleted("hello"); + + await expect(generator.waitForIdle(5_000)).resolves.toBeUndefined(); + }); +});