diff --git a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts b/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts index 02b7a45f33ad..138e57ced029 100644 --- a/apps/server/src/provider/Layers/CodexCollabRuntime.integration.test.ts +++ b/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(); + const exited = yield* Deferred.make(); + 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>([]); + const exited = yield* Deferred.make(); + 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 = { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 2ec77ecb7000..16bb1baf5610 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/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))); }), ), ),