fix(server): recover Codex sessions after signal exits

apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts

@@ -17,13 +17,16 @@ import { it } from "@effect/vitest";
 import { type ProviderApprovalDecision, type ProviderEvent, ThreadId } from "@t3tools/contracts";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
+import * as Ref from "effect/Ref";
 import * as Schema from "effect/Schema";
+import * as Scope from "effect/Scope";
 import * as Stream from "effect/Stream";
 import { assert, describe } from "vite-plus/test";
 
 import wireFixture from "../testFixtures/codexMultiAgentWire.json" with { type: "json" };
-import { makeCodexSessionRuntime } from "./CodexSessionRuntime.ts";
+import { CodexResumeCursorSchema, makeCodexSessionRuntime } from "./CodexSessionRuntime.ts";
 import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
 
 const ROOT = wireFixture.rootThreadId;
@@ -37,6 +40,46 @@ const decodeMcpElicitationResponse = Schema.decodeUnknownEffect(
     }),
   ),
 );
+const decodeResumeCursor = Schema.decodeUnknownEffect(CodexResumeCursorSchema);
+const decodeResumeRequest = Schema.decodeUnknownEffect(
+  Schema.fromJsonString(
+    Schema.Struct({
+      params: Schema.Struct({ threadId: Schema.String, excludeTurns: Schema.Boolean }),
+    }),
+  ),
+);
+
+const recoveryPeerSource = (threadStart: unknown, turnStart: unknown) => `#!/usr/bin/env node
+import * as fs from "node:fs";
+const fixture = ${JSON.stringify({ threadStart, turnStart })};
+const pidPath = process.env.T3_CODEX_RECOVERY_PID;
+const requestsPath = process.env.T3_CODEX_RECOVERY_REQUESTS;
+if (pidPath) fs.writeFileSync(pidPath, String(process.pid));
+const write = (message) => process.stdout.write(JSON.stringify(message) + "\\n");
+const rl = (await import("node:readline")).createInterface({ input: process.stdin });
+rl.on("line", (line) => {
+  let message;
+  try { message = JSON.parse(line); } catch { return; }
+  const { id, method, params } = message;
+  if (method === "initialize") return write({ id, result: { userAgent: "recovery-peer", codexHome: "/tmp", platformFamily: "unix", platformOs: "linux" } });
+  if (method === "thread/start") return write({ id, result: fixture.threadStart });
+  if (method === "thread/resume") {
+    if (requestsPath) fs.appendFileSync(requestsPath, JSON.stringify({ method, params }) + "\\n");
+    return write({ id, result: { ...fixture.threadStart, thread: { ...fixture.threadStart.thread, id: params.threadId, sessionId: params.threadId } } });
+  }
+  if (method === "turn/start") {
+    write({ id, result: fixture.turnStart });
+    write({ jsonrpc: "2.0", method: "turn/started", params: { threadId: fixture.threadStart.thread.id, turn: fixture.turnStart.turn } });
+    if (process.env.T3_CODEX_RECOVERY_COMPLETE === "1") write({ jsonrpc: "2.0", method: "turn/completed", params: { threadId: fixture.threadStart.thread.id, turn: { ...fixture.turnStart.turn, status: "completed" } } });
+    if (process.env.T3_CODEX_RECOVERY_EXIT_CODE !== undefined) {
+      const code = Number(process.env.T3_CODEX_RECOVERY_EXIT_CODE);
+      process.stdout.end(() => process.exit(code));
+    }
+    return;
+  }
+  if (id !== undefined) write({ id, result: {} });
+});
+`;
 
 /**
  * The captured sequence, extended with the shapes the live capture didn't
@@ -166,6 +209,182 @@ const peerPath = NodePath.join(
 );
 
 describe("CodexSessionRuntime collab integration", () => {
+  it.live.skipIf(HostProcessPlatform.defaultValue() === "win32")(
+    "recovers a session after its app-server is killed by SIGKILL",
+    () =>
+      Effect.gen(function* () {
+        const recoveryPeerPath = NodePath.join(
+          NodeOS.tmpdir(),
+          `t3-codex-recovery-peer-${process.pid}.mjs`,
+        );
+        const recoveryPidPath = `${recoveryPeerPath}.pid`;
+        const recoveryRequestsPath = `${recoveryPeerPath}.requests`;
+        NodeFS.writeFileSync(
+          recoveryPeerPath,
+          recoveryPeerSource(wireFixture.responses.threadStart, wireFixture.responses.turnStart),
+          "utf8",
+        );
+        NodeFS.chmodSync(recoveryPeerPath, 0o755);
+        NodeFS.rmSync(recoveryPidPath, { force: true });
+        NodeFS.rmSync(recoveryRequestsPath, { force: true });
+        yield* Effect.addFinalizer(() =>
+          Effect.sync(() => {
+            NodeFS.rmSync(recoveryPeerPath, { force: true });
+            NodeFS.rmSync(recoveryPidPath, { force: true });
+            NodeFS.rmSync(recoveryRequestsPath, { force: true });
+          }),
+        );
+
+        const environment = {
+          ...process.env,
+          T3_CODEX_RECOVERY_PID: recoveryPidPath,
+          T3_CODEX_RECOVERY_REQUESTS: recoveryRequestsPath,
+        };
+        const threadId = ThreadId.make("thread-codex-signal-recovery");
+        const firstScope = yield* Scope.make();
+        yield* Effect.addFinalizer(() => Scope.close(firstScope, Exit.void));
+        const firstRuntime = yield* makeCodexSessionRuntime({
+          threadId,
+          binaryPath: recoveryPeerPath,
+          cwd: NodeOS.tmpdir(),
+          runtimeMode: "full-access",
+          environment,
+        }).pipe(Effect.provideService(Scope.Scope, firstScope));
+        const turnStarted = yield* Deferred.make<void>();
+        const exited = yield* Deferred.make<string>();
+        yield* firstRuntime.events.pipe(
+          Stream.runForEach((event) => {
+            if (event.method === "turn/started") {
+              return Deferred.succeed(turnStarted, undefined).pipe(Effect.asVoid);
+            }
+            if (event.method === "session/exited") {
+              return Deferred.succeed(exited, event.message ?? "").pipe(Effect.asVoid);
+            }
+            return Effect.void;
+          }),
+          Effect.forkScoped,
+        );
+        const firstSession = yield* firstRuntime.start();
+        const cursor = yield* decodeResumeCursor(firstSession.resumeCursor);
+        assert.deepEqual(cursor, { threadId: wireFixture.responses.threadStart.thread.id });
+        yield* firstRuntime.sendTurn({ input: "leave this turn open" });
+        yield* Deferred.await(turnStarted);
+        assert.isDefined((yield* firstRuntime.getSession).activeTurnId);
+
+        const childPid = Number(NodeFS.readFileSync(recoveryPidPath, "utf8"));
+        assert.isTrue(Number.isInteger(childPid));
+        process.kill(childPid, "SIGKILL");
+        const exitMessage = yield* Deferred.await(exited).pipe(Effect.timeout("5 seconds"));
+        assert.include(exitMessage, "Codex App Server exited unexpectedly:");
+        assert.equal((yield* firstRuntime.getSession).status, "error");
+        assert.isUndefined((yield* firstRuntime.getSession).activeTurnId);
+        assert.deepEqual((yield* firstRuntime.getSession).resumeCursor, cursor);
+        yield* firstRuntime.close;
+
+        const secondScope = yield* Scope.make();
+        yield* Effect.addFinalizer(() => Scope.close(secondScope, Exit.void));
+        const secondRuntime = yield* makeCodexSessionRuntime({
+          threadId,
+          binaryPath: recoveryPeerPath,
+          cwd: NodeOS.tmpdir(),
+          runtimeMode: "full-access",
+          resumeCursor: cursor,
+          environment: { ...environment, T3_CODEX_RECOVERY_COMPLETE: "1" },
+        }).pipe(Effect.provideService(Scope.Scope, secondScope));
+        const secondSession = yield* secondRuntime.start();
+        assert.equal(secondSession.status, "ready");
+        assert.deepEqual(secondSession.resumeCursor, cursor);
+        const turn = yield* secondRuntime.sendTurn({ input: "continue without replay" });
+        assert.isDefined(turn.turnId);
+        const { params: resumeRequest } = yield* decodeResumeRequest(
+          NodeFS.readFileSync(recoveryRequestsPath, "utf8").trim(),
+        );
+        assert.equal(resumeRequest.threadId, wireFixture.responses.threadStart.thread.id);
+        assert.equal(resumeRequest.excludeTurns, true);
+        yield* secondRuntime.close;
+      }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
+
+  it.live.skipIf(HostProcessPlatform.defaultValue() === "win32")(
+    "preserves numeric exit statuses and ignores intentional close",
+    () =>
+      Effect.forEach(
+        [
+          { exitCode: 0, expectedStatus: "closed" as const },
+          { exitCode: 7, expectedStatus: "error" as const },
+          { intentionalClose: true, expectedStatus: "closed" as const },
+        ],
+        (testCase, index) =>
+          Effect.gen(function* () {
+            const peerPath = NodePath.join(
+              NodeOS.tmpdir(),
+              `t3-codex-exit-peer-${process.pid}-${index}.mjs`,
+            );
+            NodeFS.writeFileSync(
+              peerPath,
+              recoveryPeerSource(
+                wireFixture.responses.threadStart,
+                wireFixture.responses.turnStart,
+              ),
+              "utf8",
+            );
+            NodeFS.chmodSync(peerPath, 0o755);
+            yield* Effect.addFinalizer(() =>
+              Effect.sync(() => NodeFS.rmSync(peerPath, { force: true })),
+            );
+
+            const runtimeScope = yield* Scope.make();
+            yield* Effect.addFinalizer(() => Scope.close(runtimeScope, Exit.void));
+            const runtime = yield* makeCodexSessionRuntime({
+              threadId: ThreadId.make(`thread-codex-exit-${index}`),
+              binaryPath: peerPath,
+              cwd: NodeOS.tmpdir(),
+              runtimeMode: "full-access",
+              environment: {
+                ...process.env,
+                ...(testCase.exitCode === undefined
+                  ? {}
+                  : { T3_CODEX_RECOVERY_EXIT_CODE: String(testCase.exitCode) }),
+              },
+            }).pipe(Effect.provideService(Scope.Scope, runtimeScope));
+            const terminalEvents = yield* Ref.make<Array<string>>([]);
+            const exited = yield* Deferred.make<string>();
+            const eventConsumer = yield* runtime.events.pipe(
+              Stream.runForEach((event) => {
+                const record = Ref.update(terminalEvents, (events) => [...events, event.method]);
+                return event.method === "session/exited"
+                  ? record.pipe(Effect.andThen(Deferred.succeed(exited, event.message ?? "")))
+                  : record;
+              }),
+              Effect.forkScoped,
+            );
+            yield* runtime.start();
+            if (testCase.intentionalClose) {
+              yield* runtime.close;
+              yield* Fiber.join(eventConsumer).pipe(Effect.exit);
+              const events = yield* Ref.get(terminalEvents);
+              assert.include(events, "session/closed");
+              assert.notInclude(events, "session/exited");
+              assert.equal((yield* runtime.getSession).status, "closed");
+              return;
+            }
+            yield* runtime.sendTurn({ input: "exit" });
+            const exitMessage = yield* Deferred.await(exited).pipe(Effect.timeout("5 seconds"));
+            assert.equal((yield* runtime.getSession).status, testCase.expectedStatus);
+            assert.isUndefined((yield* runtime.getSession).activeTurnId);
+            const events = yield* Ref.get(terminalEvents);
+            assert.include(events, "session/exited");
+            assert.equal(
+              exitMessage,
+              testCase.exitCode === 0
+                ? "Codex App Server exited."
+                : "Codex App Server exited with code 7.",
+            );
+          }),
+        { concurrency: 1 },
+      ).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
+
   it.effect("looks up child model metadata once after activity registration", () =>
     Effect.gen(function* () {
       const script = {

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

@@ -2237,26 +2237,32 @@ export const makeCodexSessionRuntime = (
     );
 
     yield* child.exitCode.pipe(
-      Effect.flatMap((exitCode) =>
+      Effect.match({
+        onFailure: (cause) => {
+          const diagnostic = cause instanceof Error ? cause.message : String(cause);
+          return {
+            status: "error" as const,
+            message: `Codex App Server exited unexpectedly: ${diagnostic}.`,
+          };
+        },
+        onSuccess: (exitCode) => ({
+          status: exitCode === 0 ? ("closed" as const) : ("error" as const),
+          message:
+            exitCode === 0
+              ? "Codex App Server exited."
+              : `Codex App Server exited with code ${exitCode}.`,
+        }),
+      }),
+      Effect.flatMap(({ status, message }) =>
         Ref.get(closedRef).pipe(
           Effect.flatMap((closed) => {
             if (closed) {
               return Effect.void;
             }
-            const nextStatus = exitCode === 0 ? "closed" : "error";
             return updateSession(sessionRef, {
-              status: nextStatus,
+              status,
               activeTurnId: undefined,
-            }).pipe(
-              Effect.andThen(
-                emitSessionEvent(
-                  "session/exited",
-                  exitCode === 0
-                    ? "Codex App Server exited."
-                    : `Codex App Server exited with code ${exitCode}.`,
-                ),
-              ),
-            );
+            }).pipe(Effect.andThen(emitSessionEvent("session/exited", message)));
           }),
         ),
       ),