fix(server): Codex threads survive protocol drift on resume

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

@@ -43,27 +43,66 @@ describe("CodexSessionRuntimeIdentifierGenerationError", () => {
   });
 });
 
-function makeThreadOpenResponse(
-  threadId: string,
-): CodexRpc.ClientRequestResponsesByMethod["thread/start"] {
+/**
+ * Raw `thread/start` / `thread/resume` payload as Codex puts it on the wire.
+ * `items` accepts arbitrary history entries so tests can replay shapes newer
+ * than the generated bindings.
+ */
+function makeThreadOpenResponse(threadId: string, items: ReadonlyArray<unknown> = []): unknown {
   return {
     cwd: "/tmp/project",
     model: "gpt-5.3-codex",
     modelProvider: "openai",
     approvalPolicy: "never",
     approvalsReviewer: "user",
-    sandbox: { type: "danger-full-access" },
+    sandbox: { type: "dangerFullAccess" },
     thread: {
       id: threadId,
-      createdAt: "2026-04-18T00:00:00.000Z",
-      source: { session: "cli" },
-      turns: [],
-      status: {
-        state: "idle",
-        activeFlags: [],
-      },
+      cliVersion: "0.150.0",
+      createdAt: 0,
+      updatedAt: 0,
+      cwd: "/tmp/project",
+      ephemeral: false,
+      modelProvider: "openai",
+      preview: "",
+      sessionId: "session-1",
+      source: "cli",
+      status: { type: "idle" },
+      turns: items.length === 0 ? [] : [{ id: "turn-1", status: "completed", items }],
     },
-  } as unknown as CodexRpc.ClientRequestResponsesByMethod["thread/start"];
+  };
+}
+
+/**
+ * Mirrors the real client: params keep their generated types, and the raw
+ * payload is decoded with whichever response schema the caller supplied,
+ * failing exactly as the client would.
+ */
+function makeThreadOpenClient(
+  respond: (
+    method: "thread/start" | "thread/resume",
+  ) => Effect.Effect<unknown, CodexErrors.CodexAppServerError>,
+) {
+  return {
+    request: <M extends "thread/start" | "thread/resume", A>(
+      method: M,
+      _payload: CodexRpc.ClientRequestParamsByMethod[M],
+      responseSchema: Schema.Codec<A, unknown>,
+    ) =>
+      respond(method).pipe(
+        Effect.flatMap((raw) =>
+          Schema.decodeUnknownEffect(responseSchema)(raw).pipe(
+            Effect.mapError((cause) =>
+              CodexErrors.CodexAppServerRequestError.invalidPayload(
+                method,
+                "decode-payload",
+                cause,
+              ),
+            ),
+          ),
+        ),
+      ),
+  };
 }
 
 describe("buildTurnStartParams", () => {
@@ -752,6 +791,20 @@ describe("isRecoverableThreadResumeError", () => {
     );
   });
 
+  it("matches responses this build cannot decode", () => {
+    NodeAssert.equal(
+      isRecoverableThreadResumeError(
+        new CodexErrors.CodexAppServerRequestError({
+          code: -32602,
+          errorMessage: "Invalid payload for method 'thread/resume' during 'decode-payload'",
+          method: "thread/resume",
+          operation: "decode-payload",
+        }),
+      ),
+      true,
+    );
+  });
+
   it("ignores unrelated missing-resource errors that do not mention threads", () => {
     NodeAssert.equal(
       isRecoverableThreadResumeError(
@@ -775,27 +828,56 @@ describe("isRecoverableThreadResumeError", () => {
 });
 
 describe("openCodexThread", () => {
-  it.effect("falls back to thread/start when resume fails recoverably", () =>
+  it.effect("resumes a thread whose history uses a newer protocol variant", () =>
     Effect.gen(function* () {
-      const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = [];
-      const started = makeThreadOpenResponse("fresh-thread");
-      const client = {
-        request: <M extends "thread/start" | "thread/resume">(
-          method: M,
-          payload: CodexRpc.ClientRequestParamsByMethod[M],
-        ) => {
-          calls.push({ method, payload });
-          if (method === "thread/resume") {
-            return Effect.fail(
-              new CodexErrors.CodexAppServerRequestError({
-                code: -32603,
-                errorMessage: "thread not found",
-              }),
-            );
-          }
-          return Effect.succeed(started as CodexRpc.ClientRequestResponsesByMethod[M]);
+      // Opening a session must not depend on history this build cannot name
+      // (#8322). The kind is deliberately fictional so the test keeps
+      // exercising drift after the bindings learn today's values.
+      const resumed = makeThreadOpenResponse("resumed-thread", [
+        {
+          id: "item-18",
+          type: "subAgentActivity",
+          agentPath: "/root/child",
+          agentThreadId: "child-thread",
+          kind: "escalated",
         },
-      };
+      ]);
+      const calls: Array<string> = [];
+      const client = makeThreadOpenClient((method) => {
+        calls.push(method);
+        return Effect.succeed(resumed);
+      });
+
+      const opened = yield* openCodexThread({
+        client,
+        threadId: ThreadId.make("thread-1"),
+        runtimeMode: "full-access",
+        cwd: "/tmp/project",
+        requestedModel: "gpt-5.3-codex",
+        serviceTier: undefined,
+        resumeThreadId: "resumed-thread",
+      });
+
+      NodeAssert.equal(opened.thread.id, "resumed-thread");
+      NodeAssert.deepStrictEqual(calls, ["thread/resume"]);
+    }),
+  );
+
+  it.effect("falls back to thread/start when resume fails recoverably", () =>
+    Effect.gen(function* () {
+      const calls: Array<string> = [];
+      const client = makeThreadOpenClient((method) => {
+        calls.push(method);
+        if (method === "thread/resume") {
+          return Effect.fail(
+            new CodexErrors.CodexAppServerRequestError({
+              code: -32603,
+              errorMessage: "thread not found",
+            }),
+          );
+        }
+        return Effect.succeed(makeThreadOpenResponse("fresh-thread"));
+      });
 
       const opened = yield* openCodexThread({
         client,
@@ -808,33 +890,48 @@ describe("openCodexThread", () => {
       });
 
       NodeAssert.equal(opened.thread.id, "fresh-thread");
-      NodeAssert.deepStrictEqual(
-        calls.map((call) => call.method),
-        ["thread/resume", "thread/start"],
-      );
+      NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/start"]);
+    }),
+  );
+
+  it.effect("falls back to thread/start when the resume response cannot be decoded", () =>
+    Effect.gen(function* () {
+      const calls: Array<string> = [];
+      const client = makeThreadOpenClient((method) => {
+        calls.push(method);
+        if (method === "thread/resume") {
+          return Effect.succeed({ thread: {} });
+        }
+        return Effect.succeed(makeThreadOpenResponse("fresh-thread"));
+      });
+
+      const opened = yield* openCodexThread({
+        client,
+        threadId: ThreadId.make("thread-1"),
+        runtimeMode: "full-access",
+        cwd: "/tmp/project",
+        requestedModel: "gpt-5.3-codex",
+        serviceTier: undefined,
+        resumeThreadId: "stale-thread",
+      });
+
+      NodeAssert.equal(opened.thread.id, "fresh-thread");
+      NodeAssert.deepStrictEqual(calls, ["thread/resume", "thread/start"]);
     }),
   );
 
   it.effect("propagates non-recoverable resume failures", () =>
     Effect.gen(function* () {
-      const client = {
-        request: <M extends "thread/start" | "thread/resume">(
-          method: M,
-          _payload: CodexRpc.ClientRequestParamsByMethod[M],
-        ) => {
-          if (method === "thread/resume") {
-            return Effect.fail(
+      const client = makeThreadOpenClient((method) =>
+        method === "thread/resume"
+          ? Effect.fail(
               new CodexErrors.CodexAppServerRequestError({
                 code: -32603,
                 errorMessage: "timed out waiting for server",
               }),
-            );
-          }
-          return Effect.succeed(
-            makeThreadOpenResponse("fresh-thread") as CodexRpc.ClientRequestResponsesByMethod[M],
-          );
-        },
-      };
+            )
+          : Effect.succeed(makeThreadOpenResponse("fresh-thread")),
+      );
 
       const error = yield* openCodexThread({
         client,

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

@@ -661,25 +661,47 @@ function classifyCodexStderrLine(rawLine: string): { readonly message: string }
   return { message: line };
 }
 
+const isCodexAppServerRequestError = Schema.is(CodexErrors.CodexAppServerRequestError);
+
 export function isRecoverableThreadResumeError(error: unknown): boolean {
+  // A response we cannot decode means Codex resumed the thread but described it
+  // in a protocol shape this build does not know. Retrying will never help, so
+  // treat it as recoverable and open a fresh Codex thread instead of leaving
+  // the T3 thread permanently unusable.
+  if (isCodexAppServerRequestError(error) && error.operation === "decode-payload") {
+    return true;
+  }
   const message = (error instanceof Error ? error.message : String(error)).toLowerCase();
   if (!message.includes("thread")) {
     return false;
   }
   return RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS.some((snippet) => message.includes(snippet));
 }
 
-type CodexThreadOpenResponse =
-  | CodexRpc.ClientRequestResponsesByMethod["thread/start"]
-  | CodexRpc.ClientRequestResponsesByMethod["thread/resume"];
+/**
+ * The only parts of a `thread/start` or `thread/resume` response this runtime
+ * consumes. Codex replays the whole thread history in those responses, and
+ * decoding it against generated bindings makes opening a session fail whenever
+ * upstream adds a protocol variant we have not regenerated yet (see #8322).
+ * Session state is rebuilt from notifications anyway, so read the handful of
+ * fields we need and let the rest pass through undecoded.
+ */
+const CodexThreadOpenResponse = Schema.Struct({
+  thread: Schema.Struct({ id: Schema.String }),
+  cwd: Schema.String,
+  model: Schema.String,
+});
+
+type CodexThreadOpenResponse = typeof CodexThreadOpenResponse.Type;
 
 type CodexThreadOpenMethod = "thread/start" | "thread/resume";
 
 interface CodexThreadOpenClient {
   readonly request: <M extends CodexThreadOpenMethod>(
     method: M,
     payload: CodexRpc.ClientRequestParamsByMethod[M],
-  ) => Effect.Effect<CodexRpc.ClientRequestResponsesByMethod[M], CodexErrors.CodexAppServerError>;
+    responseSchema: typeof CodexThreadOpenResponse,
+  ) => Effect.Effect<CodexThreadOpenResponse, CodexErrors.CodexAppServerError>;
 }
 
 export const openCodexThread = (input: {
@@ -700,14 +722,18 @@ export const openCodexThread = (input: {
   });
 
   if (resumeThreadId === undefined) {
-    return input.client.request("thread/start", startParams);
+    return input.client.request("thread/start", startParams, CodexThreadOpenResponse);
   }
 
   return input.client
-    .request("thread/resume", {
-      threadId: resumeThreadId,
-      ...startParams,
-    })
+    .request(
+      "thread/resume",
+      {
+        threadId: resumeThreadId,
+        ...startParams,
+      },
+      CodexThreadOpenResponse,
+    )
     .pipe(
       Effect.catchIf(isRecoverableThreadResumeError, (error) =>
         Effect.logWarning("codex app-server thread resume fell back to fresh start", {
@@ -716,7 +742,11 @@ export const openCodexThread = (input: {
           resumeThreadId,
           recoverable: true,
           cause: error,
-        }).pipe(Effect.andThen(input.client.request("thread/start", startParams))),
+        }).pipe(
+          Effect.andThen(
+            input.client.request("thread/start", startParams, CodexThreadOpenResponse),
+          ),
+        ),
       ),
     );
 };

docs/internals/providers.md

@@ -39,6 +39,32 @@ directory to route session and turn operations for a thread, so callers name a t
 Adding a driver means writing the driver plus adapter and adding it to `BUILT_IN_DRIVERS`. No
 orchestration, contract, or client change is required for the common case.
 
+## Generated Codex bindings
+
+`packages/effect-codex-app-server` holds Effect/Schema bindings generated from a pinned
+`openai/codex` revision (`scripts/generate.ts`). Codex ships far more often than we regenerate, so
+at any moment an installed CLI may describe itself with protocol variants those bindings do not
+name — a new enum member, a new item type, a newly required field.
+
+Two rules keep that drift from breaking sessions:
+
+- **Decode only what you consume.** `thread/start` and `thread/resume` replay an entire thread
+  history, and the session runtime needs three fields from it. Pass a narrow response schema as the
+  third argument to `client.request` rather than accepting the generated one; anything the runtime
+  does not read cannot then fail the request.
+- **Never treat drift as fatal.** A response we cannot decode counts as a recoverable resume error,
+  so the thread falls back to a fresh Codex session instead of becoming permanently unopenable.
+  Notifications we cannot decode are dropped with a warning, never silently.
+
+When a live event carries a value the bindings reject, teach them that one value: the generator's
+definition overrides (`Codex0150DefinitionSchemas` in `scripts/generate.ts`) widen a named
+definition without moving the pin. Prefer that to a full refresh, which drags in unrelated changes
+and can break older CLIs — upstream adds _required_ fields too, so pinning forward breaks anyone
+who has not upgraded.
+
+None of that replaces the two rules. Overrides fix the variants we already know about; the rules
+are what keep the ones we do not know about yet from being fatal.
+
 ## Model manifest
 
 The model picker's legacy section is driven by `apps/server/src/provider/model-manifest.json`, which