fix(app): recover failed turn retractions without duplicating drafts

apps/server/src/provider/Layers/CodexAdapter.test.ts

@@ -42,6 +42,7 @@ import type { CodexAdapterShape } from "../Services/CodexAdapter.ts";
 import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
 import {
   type CodexSessionRuntimeOptions,
+  type CodexSessionRuntimeError,
   type CodexSessionRuntimeSendTurnInput,
   type CodexSessionRuntimeShape,
   type CodexThreadSnapshot,
@@ -104,6 +105,14 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
       }),
   );
 
+  public readonly revertThreadImpl = vi.fn(
+    (_beforeTurnId: TurnId): Promise<CodexThreadSnapshot> =>
+      Promise.resolve({
+        threadId: "provider-thread-1",
+        turns: [],
+      }),
+  );
+
   public readonly deleteThreadImpl = vi.fn((): Promise<void> => Promise.resolve(undefined));
   public readonly uploadFeedbackImpl = vi.fn((_reason?: string) =>
     Promise.resolve({ threadId: "provider-thread-1" }),
@@ -147,6 +156,13 @@ class FakeCodexRuntime implements CodexSessionRuntimeShape {
     return Effect.promise(() => this.rollbackThreadImpl(numTurns));
   }
 
+  revertThread(beforeTurnId: TurnId) {
+    return Effect.tryPromise({
+      try: () => this.revertThreadImpl(beforeTurnId),
+      catch: (cause) => cause as CodexSessionRuntimeError,
+    });
+  }
+
   deleteThread = Effect.promise(() => this.deleteThreadImpl());
   uploadFeedback(reason?: string) {
     return Effect.promise(() => this.uploadFeedbackImpl(reason));
@@ -389,21 +405,59 @@ sessionErrorLayer("CodexAdapterLive session errors", (it) => {
         turns: turnIds.map((id) => ({ id: asTurnId(id), items: [] })),
       });
       const targetTurnId = asTurnId("turn-retracted");
-      runtime.readThreadImpl
-        .mockResolvedValueOnce(snapshot(["turn-1", "turn-2", targetTurnId]))
-        .mockResolvedValue(snapshot(["turn-1", "turn-2"]));
-      runtime.rollbackThreadImpl.mockResolvedValue(snapshot(["turn-1", "turn-2"]));
+      runtime.revertThreadImpl.mockResolvedValue(snapshot(["turn-1", "turn-2"]));
 
       const first = yield* adapter.rollbackThreadTo(threadId, 3, targetTurnId);
       NodeAssert.deepStrictEqual(
         first.turns.map((turn) => turn.id),
         [asTurnId("turn-1"), asTurnId("turn-2")],
       );
-      NodeAssert.deepStrictEqual(runtime.rollbackThreadImpl.mock.calls, [[1]]);
+      NodeAssert.deepStrictEqual(runtime.revertThreadImpl.mock.calls, [[targetTurnId]]);
+      NodeAssert.equal(runtime.readThreadImpl.mock.calls.length, 0);
+      NodeAssert.equal(runtime.rollbackThreadImpl.mock.calls.length, 0);
 
       const repeated = yield* adapter.rollbackThreadTo(threadId, 3, targetTurnId);
       NodeAssert.equal(repeated.turns.length, 2);
+      NodeAssert.deepStrictEqual(runtime.revertThreadImpl.mock.calls, [
+        [targetTurnId],
+        [targetTurnId],
+      ]);
+    }),
+  );
+
+  it.effect("verifies a rollback from its response when thread reads still lag", () =>
+    Effect.gen(function* () {
+      const adapter = yield* CodexAdapter;
+      const threadId = asThreadId("lagging-read-rollback-thread");
+      yield* adapter.startSession({
+        provider: ProviderDriverKind.make("codex"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+      const runtime = sessionRuntimeFactory.lastRuntime;
+      NodeAssert.ok(runtime);
+      NodeAssert.ok(adapter.rollbackThreadTo);
+      const snapshot = (turnIds: ReadonlyArray<string>): CodexThreadSnapshot => ({
+        threadId: "provider-thread-1",
+        turns: turnIds.map((id) => ({ id: asTurnId(id), items: [] })),
+      });
+      const targetTurnId = asTurnId("turn-retracted");
+      const staleSnapshot = snapshot(["turn-1", targetTurnId]);
+      const rolledBackSnapshot = snapshot(["turn-1"]);
+      runtime.readThreadImpl.mockResolvedValue(staleSnapshot);
+      runtime.rollbackThreadImpl.mockResolvedValue(rolledBackSnapshot);
+      runtime.revertThreadImpl.mockRejectedValue(
+        CodexErrors.CodexAppServerRequestError.methodNotFound("thread/revert"),
+      );
+
+      const result = yield* adapter.rollbackThreadTo(threadId, 1, targetTurnId);
+
+      NodeAssert.deepStrictEqual(
+        result.turns.map((turn) => turn.id),
+        [asTurnId("turn-1")],
+      );
       NodeAssert.deepStrictEqual(runtime.rollbackThreadImpl.mock.calls, [[1]]);
+      NodeAssert.equal(runtime.readThreadImpl.mock.calls.length, 1);
     }),
   );
 

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -33,6 +33,7 @@ import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
 import * as Queue from "effect/Queue";
+import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
@@ -68,13 +69,17 @@ import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogg
 import { resolveCodexLaunchArgs } from "./codexLaunchArgs.ts";
 const isCodexAppServerProcessExitedError = Schema.is(CodexErrors.CodexAppServerProcessExitedError);
 const isCodexAppServerTransportError = Schema.is(CodexErrors.CodexAppServerTransportError);
+const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);
 const isCodexSessionRuntimeThreadIdMissingError = Schema.is(
   CodexSessionRuntimeThreadIdMissingError,
 );
 const isCodexResumeCursorSchema = Schema.is(CodexResumeCursorSchema);
 
 const PROVIDER = ProviderDriverKind.make("codex");
 
+const isThreadRevertUnavailable = (error: CodexSessionRuntimeError): boolean =>
+  isCodexAppServerRequestError(error) && error.code === -32601;
+
 export interface CodexAdapterLiveOptions {
   readonly instanceId?: ProviderInstanceId;
   readonly environment?: NodeJS.ProcessEnv;
@@ -1925,6 +1930,21 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
       });
     }
     const session = yield* requireSession(threadId);
+    if (targetTurnId !== undefined) {
+      const revertResult = yield* session.runtime.revertThread(targetTurnId).pipe(Effect.result);
+      if (Result.isSuccess(revertResult)) {
+        if (session.lastStartedTurnId === targetTurnId) {
+          session.lastStartedTurnId = undefined;
+        }
+        return {
+          threadId,
+          turns: revertResult.success.turns,
+        };
+      }
+      if (!isThreadRevertUnavailable(revertResult.failure)) {
+        return yield* mapCodexRuntimeError(threadId, "thread/revert", revertResult.failure);
+      }
+    }
     const current = yield* readThread(threadId);
     const targetIndex =
       targetTurnId === undefined ? -1 : current.turns.findIndex((turn) => turn.id === targetTurnId);
@@ -1950,10 +1970,15 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* (
           : targetTurnId !== undefined
             ? 0
             : current.turns.length - retainedTurnCount;
-    if (remainingDelta > 0) {
-      yield* rollbackThread(threadId, remainingDelta);
-    }
-    const verified = yield* readThread(threadId);
+    // thread/rollback returns the post-rollback snapshot. Use that response
+    // for the first verification instead of immediately calling thread/read:
+    // recent Codex app-server builds can briefly serve the pre-rollback turn
+    // from thread/read after the mutation has already committed. Treating
+    // that lagging read as failure leaves a durable T3 retraction pending and
+    // repeatedly rolls back the same provider turn.
+    const rolledBack =
+      remainingDelta > 0 ? yield* rollbackThread(threadId, remainingDelta) : undefined;
+    const verified = rolledBack ?? (yield* readThread(threadId));
     if (targetTurnId !== undefined) {
       if (verified.turns.some((turn) => turn.id === targetTurnId)) {
         return yield* new ProviderAdapterRequestError({

apps/server/src/provider/Layers/CodexSessionRuntime.test.ts

@@ -4,7 +4,7 @@ import { it } from "@effect/vitest";
 import * as Effect from "effect/Effect";
 import * as Schema from "effect/Schema";
 import { describe } from "vite-plus/test";
-import { DEFAULT_MODEL, ThreadId } from "@t3tools/contracts";
+import { DEFAULT_MODEL, ThreadId, TurnId } from "@t3tools/contracts";
 import * as CodexErrors from "effect-codex-app-server/errors";
 import * as CodexRpc from "effect-codex-app-server/rpc";
 import * as EffectCodexSchema from "effect-codex-app-server/schema";
@@ -23,6 +23,7 @@ import {
   isRecoverableThreadResumeError,
   makeMemoryConsolidationNotificationFilter,
   openCodexThread,
+  revertCodexThread,
   toMcpElicitationResponse,
 } from "./CodexSessionRuntime.ts";
 const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);
@@ -875,3 +876,40 @@ describe("deleteCodexThread", () => {
     }),
   );
 });
+
+describe("revertCodexThread", () => {
+  it.effect("reverts paginated history before the target turn", () =>
+    Effect.gen(function* () {
+      const calls: Array<{ method: string; payload: unknown }> = [];
+      const client = {
+        request: (method: string, payload?: unknown) => {
+          calls.push({ method, payload });
+          return Effect.succeed({
+            thread: {
+              id: "provider-thread-paginated",
+              turns: [],
+            },
+            turnsBackwardsCursor: "turn-cursor",
+          });
+        },
+      };
+
+      const response = yield* revertCodexThread(
+        client,
+        "provider-thread-paginated",
+        TurnId.make("turn-retracted"),
+      );
+
+      NodeAssert.equal(response.thread.id, "provider-thread-paginated");
+      NodeAssert.deepStrictEqual(calls, [
+        {
+          method: "thread/revert",
+          payload: {
+            threadId: "provider-thread-paginated",
+            beforeTurnId: "turn-retracted",
+          },
+        },
+      ]);
+    }),
+  );
+});

apps/server/src/provider/Layers/CodexSessionRuntime.ts

@@ -195,6 +195,9 @@ export interface CodexSessionRuntimeShape {
   readonly rollbackThread: (
     numTurns: number,
   ) => Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
+  readonly revertThread: (
+    beforeTurnId: TurnId,
+  ) => Effect.Effect<CodexThreadSnapshot, CodexSessionRuntimeError>;
   readonly deleteThread: Effect.Effect<void, CodexSessionRuntimeError>;
   readonly uploadFeedback: (
     reason?: string,
@@ -693,12 +696,43 @@ interface CodexThreadDeleteClient {
   >;
 }
 
+interface CodexThreadRevertClient {
+  readonly request: (
+    method: string,
+    payload?: unknown,
+  ) => Effect.Effect<unknown, CodexErrors.CodexAppServerError>;
+}
+
+const CodexThreadRevertResponse = Schema.Struct({
+  thread: Schema.Struct({
+    id: Schema.String,
+  }),
+});
+const decodeCodexThreadRevertResponse = Schema.decodeUnknownEffect(CodexThreadRevertResponse);
+
 export const deleteCodexThread = (
   client: CodexThreadDeleteClient,
   threadId: string,
 ): Effect.Effect<void, CodexErrors.CodexAppServerError> =>
   client.request("thread/delete", { threadId }).pipe(Effect.asVoid);
 
+export const revertCodexThread = Effect.fn("revertCodexThread")(function* (
+  client: CodexThreadRevertClient,
+  threadId: string,
+  beforeTurnId: TurnId,
+) {
+  const response = yield* client.request("thread/revert", { threadId, beforeTurnId });
+  return yield* decodeCodexThreadRevertResponse(response).pipe(
+    Effect.mapError((cause) =>
+      CodexErrors.CodexAppServerProtocolParseError.fromSchemaError(
+        "decode-response-payload",
+        cause,
+        { method: "thread/revert" },
+      ),
+    ),
+  );
+});
+
 export const openCodexThread = (input: {
   readonly client: CodexThreadOpenClient;
   readonly threadId: ThreadId;
@@ -2218,6 +2252,19 @@ export const makeCodexSessionRuntime = (
           });
           return parseThreadSnapshot(response);
         }),
+      revertThread: (beforeTurnId) =>
+        Effect.gen(function* () {
+          const providerThreadId = yield* readProviderThreadId;
+          const response = yield* revertCodexThread(client.raw, providerThreadId, beforeTurnId);
+          yield* updateSession(sessionRef, {
+            status: "ready",
+            activeTurnId: undefined,
+          });
+          return {
+            threadId: response.thread.id,
+            turns: [],
+          };
+        }),
       deleteThread: Effect.flatMap(readProviderThreadId, (providerThreadId) =>
         deleteCodexThread(client, providerThreadId),
       ),

apps/web/src/components/chat/RetractionRecoveryHandoff.test.ts

@@ -367,4 +367,33 @@ describe("retraction recovery handoff", () => {
     expect(signal).toBeNull();
     expect(useComposerDraftStore.getState().getDraftSession(draftId)?.hidden).toBe(true);
   });
+
+  it("does not move an unresolved mid-thread message into a new-thread draft", async () => {
+    await seedRecovery();
+    const persisted = useRetractionRecoveryStore.getState().byRequestId[requestId];
+    if (!persisted) throw new Error("Expected recovery fixture");
+    useRetractionRecoveryStore.setState({
+      byRequestId: {
+        [requestId]: { ...persisted, firstUserMessage: false },
+      },
+    });
+    const recovery = useRetractionRecoveryStore.getState().byRequestId[requestId];
+    if (!recovery) throw new Error("Expected mid-thread recovery fixture");
+
+    const signal = resolveRetractionRecoverySignal({
+      recovery,
+      liveCompletion: null,
+      projectedRetraction: null,
+      activities: [],
+      threadStatus: "live",
+      threadDetailExists: true,
+      shellSnapshotReady: true,
+      sourceThreadInShell: true,
+      nowMs: Date.parse(createdAt) + RETRACTION_RECOVERY_STALE_AFTER_MS * 2,
+    });
+
+    expect(signal).toBeNull();
+    expect(useComposerDraftStore.getState().getDraftSession(draftId)?.hidden).toBe(true);
+    expect(useRetractionRecoveryStore.getState().byRequestId[requestId]).toBeDefined();
+  });
 });

apps/web/src/components/chat/RetractionRecoveryHandoff.tsx

@@ -129,7 +129,14 @@ export function resolveRetractionRecoverySignal(input: {
   const hasCorrelatedPendingRow =
     input.projectedRetraction?.requestId === input.recovery.requestId &&
     input.projectedRetraction.status === "requested";
+  // A mid-thread pop is already restored in the source composer. Never turn
+  // that message into a brand-new-thread draft just because the acknowledgement
+  // is late: keeping the recovery pending also keeps resends gated until the
+  // server reaches a terminal result. The draft fallback is only meaningful
+  // for a first-message retraction, whose successful outcome deletes its
+  // transient source thread, plus legacy records that predate this marker.
   const stale =
+    input.recovery.firstUserMessage !== false &&
     input.threadDetailExists &&
     !hasCorrelatedPendingRow &&
     Number.isFinite(createdAtMs) &&