fix: resolve ACP v1 conformance failures found by acp-tck

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<ResumedThread> {
+        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<SessionMetadata> {
         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<void> {
-        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<void> {

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<void> {
-        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({

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,

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);
+}