fix(server): Codex rollback works after the app-server restarts (#13787)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: juliusmarminge

apps/server/scripts/record-codex-app-server-replay-fixture.ts

@@ -102,6 +102,7 @@ const SCENARIO_NAMES = [
   "turn_interrupt",
   "turn_interrupt_mid_tool",
   "thread_rollback",
+  "thread_rollback_after_restart",
   "thread_fork_native_continue",
   "thread_fork_native_siblings",
   "thread_merge_back_continue",
@@ -804,6 +805,41 @@ function scenarios(): ReadonlyArray<ReplayScenario> {
         },
       ],
     },
+    {
+      name: "thread_rollback_after_restart",
+      fileName: "thread_rollback_after_restart.ndjson",
+      description:
+        "One thread completes two turns, the app-server restarts, then the thread rolls back its latest turn and starts another turn.",
+      runs: [
+        {
+          name: "rollback-after-restart",
+          description:
+            "Two completed turns, a fresh app-server that has not loaded the thread, thread/revert before the second turn, then a post-rollback turn.",
+          steps: [
+            {
+              type: "turn",
+              label: "first-before-rollback",
+              prompt: THREAD_ROLLBACK_FIRST_PROMPT,
+            },
+            {
+              type: "turn",
+              label: "second-before-rollback",
+              prompt: THREAD_ROLLBACK_SECOND_PROMPT,
+            },
+            {
+              type: "rollback",
+              label: "rollback-latest-turn",
+              numTurns: 1,
+            },
+            {
+              type: "turn",
+              label: "post-rollback",
+              prompt: THREAD_ROLLBACK_AFTER_PROMPT,
+            },
+          ],
+        },
+      ],
+    },
     {
       name: "thread_fork_native_continue",
       fileName: "thread_fork_native_continue.ndjson",
@@ -1416,6 +1452,52 @@ function runReplaySession({
       return;
     }
 
+    if (scenario.name === "thread_rollback_after_restart") {
+      const [first, second, rollback, after] = run.steps;
+      if (
+        first === undefined ||
+        second === undefined ||
+        rollback?.type !== "rollback" ||
+        after === undefined ||
+        first.type === "rollback" ||
+        first.type === "fork" ||
+        second.type === "rollback" ||
+        second.type === "fork" ||
+        after.type === "rollback" ||
+        after.type === "fork"
+      ) {
+        throw new Error(
+          "thread_rollback_after_restart replay recording requires turn, turn, rollback, turn.",
+        );
+      }
+
+      const threadId = yield* Effect.gen(function* () {
+        const client = yield* initializeClient;
+        const thread = yield* client.request("thread/start", threadRuntimeParams);
+        yield* runTurnStep(client, thread.thread.id, first);
+        yield* runTurnStep(client, thread.thread.id, second);
+        return thread.thread.id;
+      }).pipe(Effect.provide(makeCodexLayer({ recorder })));
+
+      yield* recorder.writeRecord({
+        type: "runtime_exit",
+        status: "success",
+      });
+
+      yield* Effect.gen(function* () {
+        const client = yield* initializeClient;
+        const resumeParams = { threadId, excludeTurns: true, ...threadRuntimeParams };
+        // Like the adapter: the fresh app-server reports the thread notLoaded,
+        // so it is resumed before thread/revert, and the next turn resumes it again.
+        yield* client.request("thread/read", { threadId, includeTurns: false });
+        yield* client.request("thread/resume", resumeParams);
+        yield* revertCodexThread(client, threadId, rollback.numTurns);
+        yield* client.request("thread/resume", resumeParams);
+        yield* runTurnStep(client, threadId, after);
+      }).pipe(Effect.provide(makeCodexLayer({ recorder })));
+      return;
+    }
+
     yield* Effect.gen(function* () {
       const client = yield* initializeClient;
       const thread = yield* client.request("thread/start", {

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -924,22 +924,32 @@ export const resolveCodexForkBoundary = Effect.fn("CodexAdapterV2.resolveForkBou
 const CodexThreadHistoryMetadata = Schema.Struct({
   thread: Schema.Struct({
     historyMode: Schema.optionalKey(Schema.Literals(["legacy", "paginated"])),
+    status: Schema.optionalKey(Schema.Struct({ type: Schema.String })),
   }),
 });
 const decodeCodexThreadHistoryMetadata = Schema.decodeUnknownEffect(CodexThreadHistoryMetadata);
 
-const readCodexThreadHistoryMode = Effect.fn("CodexAdapterV2.readThreadHistoryMode")(function* (
-  raw: Pick<CodexClient.CodexAppServerClient["Service"]["raw"], "request">,
-  threadId: string,
-) {
-  const response = yield* raw.request("thread/read", { threadId, includeTurns: false });
-  const metadata = yield* decodeCodexThreadHistoryMetadata(response).pipe(
-    Effect.mapError((error) =>
-      CodexErrors.CodexAppServerRequestError.invalidPayload("thread/read", "decode-payload", error),
-    ),
-  );
-  return metadata.thread.historyMode;
-});
+const readCodexThreadHistoryMetadata = Effect.fn("CodexAdapterV2.readThreadHistoryMetadata")(
+  function* (
+    raw: Pick<CodexClient.CodexAppServerClient["Service"]["raw"], "request">,
+    threadId: string,
+  ) {
+    const response = yield* raw.request("thread/read", { threadId, includeTurns: false });
+    const metadata = yield* decodeCodexThreadHistoryMetadata(response).pipe(
+      Effect.mapError((error) =>
+        CodexErrors.CodexAppServerRequestError.invalidPayload(
+          "thread/read",
+          "decode-payload",
+          error,
+        ),
+      ),
+    );
+    return {
+      historyMode: metadata.thread.historyMode,
+      loaded: metadata.thread.status?.type !== "notLoaded",
+    };
+  },
+);
 
 export const resolveCodexRollbackTurnCount = Effect.fn("CodexAdapterV2.resolveRollbackTurnCount")(
   function* (input: ProviderAdapterV2RollbackThreadInput) {
@@ -6086,8 +6096,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
               }
               // Codex 0.156 can revert paginated history at a turn boundary.
               // Legacy history no longer has a rollback endpoint.
-              const historyMode = yield* ensureInitialized.pipe(
-                Effect.andThen(readCodexThreadHistoryMode(client.raw, threadId)),
+              const { historyMode, loaded } = yield* ensureInitialized.pipe(
+                Effect.andThen(readCodexThreadHistoryMetadata(client.raw, threadId)),
               );
               if (historyMode !== "paginated") {
                 return yield* new ProviderAdapterRollbackThreadError({
@@ -6096,6 +6106,20 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
                   cause: `Cannot roll back Codex thread ${threadId}: the thread uses legacy history, which Codex 0.156 cannot revert.`,
                 });
               }
+              // `thread/revert` only acts on a thread loaded in this app-server
+              // process. After a restart or idle release, load it the same way
+              // the next turn would before reverting.
+              if (!loaded) {
+                yield* client.raw.request("thread/resume", {
+                  threadId,
+                  excludeTurns: true,
+                  ...codexThreadRuntimeParams({
+                    threadId: threadInput.providerThread.appThreadId,
+                    modelSelection: input.modelSelection,
+                    runtimePolicy: input.runtimePolicy,
+                  }),
+                });
+              }
               const response = yield* ensureInitialized.pipe(
                 Effect.andThen(revertCodexThread(client, threadId, numTurns)),
               );
@@ -6163,8 +6187,8 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
                 // Reached only when the selected source turn has no native
                 // turn reference, so the fork had to be taken at head and then
                 // trimmed with the paginated history API.
-                const historyMode = yield* ensureInitialized.pipe(
-                  Effect.andThen(readCodexThreadHistoryMode(client.raw, response.thread.id)),
+                const { historyMode } = yield* ensureInitialized.pipe(
+                  Effect.andThen(readCodexThreadHistoryMetadata(client.raw, response.thread.id)),
                 );
                 if (historyMode !== "paginated") {
                   return yield* new ProviderAdapterForkThreadError({

apps/server/src/orchestration-v2/testkit/CodexReplayFixtures.integration.test.ts

@@ -275,6 +275,22 @@ const scenarioExpectations = {
     turnCompletedCount: 3,
     approvalRequestCount: 0,
   },
+  thread_rollback_after_restart: {
+    outgoing: [
+      "initialize",
+      "initialized",
+      "thread/start",
+      "turn/start",
+      "thread/read",
+      "thread/resume",
+      "thread/turns/list",
+      "thread/revert",
+    ],
+    incoming: ["turn/started", "turn/completed", "item/agentMessage/delta", "thread/reverted"],
+    turnStartCount: 3,
+    turnCompletedCount: 3,
+    approvalRequestCount: 0,
+  },
   thread_fork_native_continue: {
     outgoing: ["initialize", "initialized", "thread/start", "thread/fork", "turn/start"],
     incoming: ["thread/started", "turn/started", "turn/completed", "item/agentMessage/delta"],

apps/server/src/orchestration-v2/testkit/fixtures/index.ts

@@ -89,6 +89,8 @@ import { assertClaudeThreadRollbackOutput } from "./thread_rollback/claude_outpu
 import { assertThreadRollbackOutput } from "./thread_rollback/codex_output.ts";
 import { threadRollbackInput } from "./thread_rollback/input.ts";
 import { assertPiThreadRollbackOutput } from "./thread_rollback/pi_output.ts";
+import { assertThreadRollbackAfterRestartOutput } from "./thread_rollback_after_restart/codex_output.ts";
+import { threadRollbackAfterRestartInput } from "./thread_rollback_after_restart/input.ts";
 import { threadRollbackAfterStopInput } from "./thread_rollback_after_stop/input.ts";
 import { assertPiThreadRollbackAfterStopOutput } from "./thread_rollback_after_stop/pi_output.ts";
 import { assertTodoListOutput } from "./todo_list/codex_output.ts";
@@ -1126,6 +1128,21 @@ export const ORCHESTRATOR_REPLAY_FIXTURES: ReadonlyArray<OrchestratorReplayFixtu
       },
     ],
   },
+  {
+    name: "thread_rollback_after_restart",
+    buildInput: threadRollbackAfterRestartInput,
+    providers: [
+      {
+        driver: ProviderDriverKind.make("codex"),
+        transcriptFile: new URL(
+          "./thread_rollback_after_restart/codex_transcript.ndjson",
+          import.meta.url,
+        ),
+        modelSelection: CODEX_MODEL_SELECTION,
+        assertOutput: assertThreadRollbackAfterRestartOutput,
+      },
+    ],
+  },
   {
     name: "thread_rollback_after_stop",
     buildInput: threadRollbackAfterStopInput,

apps/server/src/orchestration-v2/testkit/fixtures/thread_rollback_after_restart/codex_output.ts

@@ -0,0 +1,31 @@
+import { assert } from "@effect/vitest";
+import type { ProviderReplayTranscript } from "@t3tools/contracts";
+
+import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts";
+import { assertThreadRollbackOutput } from "../thread_rollback/codex_output.ts";
+
+/**
+ * Codex only reverts threads loaded in its current app-server process, so the
+ * released session's successor must resume the thread before `thread/revert`.
+ */
+export function assertThreadRollbackAfterRestartOutput(
+  result: OrchestratorV2ScenarioResult,
+  transcript: ProviderReplayTranscript,
+) {
+  assertThreadRollbackOutput(result, transcript);
+  const methods = transcript.entries.flatMap((entry) =>
+    entry.type === "expect_outbound" &&
+    typeof entry.frame === "object" &&
+    entry.frame !== null &&
+    "method" in entry.frame
+      ? [String(entry.frame.method)]
+      : [],
+  );
+  const revert = methods.indexOf("thread/revert");
+  assert.isAbove(revert, 0, "the rollback must reach thread/revert");
+  assert.equal(
+    methods.lastIndexOf("thread/resume", revert),
+    revert - 2,
+    "the fresh app-server must resume the thread right before listing turns and reverting",
+  );
+}