fix(runner): end the detached Codex controller's event stream so graceful shutdown exits promptly (#19483)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - The control plane runs native Codex turns through the Paperclip
runner, and a restart must hand those turns to the next controller
> - On a graceful shutdown (SIGTERM), the server detaches each native
Codex session and leaves the runner alive for restart recovery
> - After the detach, the shutdown waited the full 30 s "bounded
heartbeat execution finalizer drain" every time
> - A supervisor with a 30 s graceful window (for example the runner-e2e
supervisor) then sent SIGKILL, so a graceful restart became a hard kill
> - This pull request ends the detached controller's event stream, so
the old run execution unwinds immediately
> - The benefit is that graceful restarts and deploys exit in about one
second and stay graceful

## Linked Issues or Issue Description

**What happened?**
Every graceful shutdown with an active or restart-suspended native Codex
run waited 30 s before exit. The server log showed `graceful heartbeat
run drain complete` with the run in `restartSuspendedRunIds`, then
`bounded heartbeat execution finalizer drain timed out` 30 s later.
Under a supervisor with a 30 s grace period, the server got SIGKILL.

**Steps to reproduce**
1. Start a native Codex run (runner-e2e cell
`everyday-workflows.runner-codex.local.recover-controller`).
2. Stop the server with SIGTERM while the turn is active
(`PAPERCLIP_RUNNER_E2E_CONTROLLER_STOP_SIGNAL=SIGTERM`).
3. See that the server does not exit until the 30 s drain timeout, or
until SIGKILL.

**Expected behavior**
The server exits promptly. The detached run is left for restart
recovery.

**Root cause**
- `drainRunningRunsForShutdown`
(`server/src/services/heartbeat/recovery.ts`) calls
`detachNativeSessionsForRestart` for the native run. This calls
`CodexHarnessSession.detachControllerForRestart`.
- `CodexHarnessSession.detachControllerForRestart`
(`packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts`)
detached the transport but did not close the session `eventQueue`.
- The transport closed its notification queue. `pumpNotifications` then
returned normally and did not close the session queue.
- `executeNativeSession` waited on `eventIterator.next()` for an event
that could never arrive. The heartbeat execution promise (registered at
`server/src/services/heartbeat/queue.ts` in
`activeRunExecutionPromises`) never settled.
- `drainActiveRunExecutions` awaited that promise until the 30 s bound
in `drainRunExecutionFinalizersForShutdown`.

Temporary diagnostics confirmed this: during the hang, the only pending
entry was the `queue.ts` run execution, in the `execution.runtime`
phase, with no pending wakes. Other drivers (Cursor Cloud, external
runnerd) already end their event stream on detach.

## What Changed

- `CodexHarnessSession.detachControllerForRestart` now clears and closes
the session event queue after the transport detaches. The detached
consumer gets end-of-stream. The executor then throws
`NativeControllerDetachedForRestartError`, and heartbeat leaves the run
for restart recovery.
- Buffered events are not delivered to the detached consumer. The next
controller replays them from the runner journal.
- Pending runtime requests stay open for the adopting controller (no
`cancelPendingRequests`).
- If the transport detach fails, the queue stays open. The existing
uncertain-ownership path is unchanged.
- Added a regression test: after detach, the session event iterator
ends, and a late provider notification does not reach the detached
consumer.

## Verification

- Red, then green: `npx vitest run
src/drivers/codex/codex-app-server-driver.lifecycle.test.ts` in
`packages/paperclip-runner`. Before the fix, the new test failed with
`hung`. After the fix, all 22 tests pass.
- `npx vitest run src/drivers/codex src/backends`: 27 files, 457 tests
pass.
- `tsc --noEmit` for `@paperclipai/paperclip-runner` passes.
- Live: `PAPERCLIP_RUNNER_E2E_CONTROLLER_STOP_SIGNAL=SIGTERM pnpm
test:e2e:runner -- --max-automatic-retries 0 --id
everyday-workflows.runner-codex.local.recover-controller`
- Before (unchanged code, reproduced locally): `Paperclip did not stop
within 30000ms; sending SIGKILL`, `server exited code=null
signal=SIGKILL`.
- After, run 1: PASS. Drain complete with the run in
`restartSuspendedRunIds`, then `server exited code=0 signal=null` within
the same second. The next server resumed the run and the task completed.
  - After, run 2: PASS. Same clean exit, no SIGKILL.
- `pnpm verify` returned exit 2 (broader verification required).
Required CI supplies that coverage.
- Note: 21 tests in `src/live/runnerd-restart-continuation.test.ts`,
`runnerd-controller-outage.test.ts` and
`runnerd-codex-transport-warm-attachment.test.ts` fail on this machine
with and without this change. They use the transport directly, not
`CodexHarnessSession`.

## Risks

- Low risk. The change only affects a session after its controller
detaches for restart. That controller has no authority to settle the
run.
- The 30 s finalizer bound stays in place. It still applies when a run
execution is past the provider stream, for example while it persists an
accepted result or settles a workspace. Those durable writes must finish
(or reach their recovery records) before exit, so waiting for them is
correct.
- Dropped buffered events: the old consumer could not persist them
anyway after the transport closed. The next controller replays from the
runner journal.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with
tool use (shell, file edit, test execution).

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.lifecycle.test.ts

@@ -426,6 +426,61 @@ describe("Codex app-server Codex driver", () => {
     expect(close).not.toHaveBeenCalled();
   });
 
+  it("ends the detached controller's event stream so its consumer can unwind", async () => {
+    const transport = new FakeCodexTransport();
+    Object.assign(transport, { detachControllerForRestart: vi.fn(async () => undefined) });
+    const driver = makeDriver([transport]);
+    const session = await driver.openSession({
+      runId: "run-detach-consumer",
+      normalizedSessionId: "normalized-detach-consumer",
+      workingDirectory: WORKSPACE,
+    });
+    await session.startTurn({ message: { role: "user", text: "Keep working" } });
+    transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } });
+    const iterator = session.events()[Symbol.asyncIterator]();
+    // Drain what the provider published before the restart.
+    for (;;) {
+      const pending = await Promise.race([
+        iterator.next(),
+        new Promise<"waiting">(resolve => setTimeout(() => resolve("waiting"), 20)),
+      ]);
+      if (pending === "waiting") break;
+      expect(pending.done).toBe(false);
+    }
+    const next = iterator.next();
+
+    await session.detachControllerForRestart?.();
+    // The provider keeps running for the next controller; a late notification
+    // must not reach the detached consumer.
+    transport.push("item/agentMessage/delta", { threadId: "thread-1", turnId: "turn-1", itemId: "m", delta: "late" });
+
+    await expect(Promise.race([
+      next,
+      new Promise<"hung">(resolve => setTimeout(() => resolve("hung"), 200)),
+    ])).resolves.toEqual({ value: undefined, done: true });
+  });
+
+  it("drops events the detached controller had not read yet", async () => {
+    const transport = new FakeCodexTransport();
+    Object.assign(transport, { detachControllerForRestart: vi.fn(async () => undefined) });
+    const driver = makeDriver([transport]);
+    const session = await driver.openSession({
+      runId: "run-detach-buffered",
+      normalizedSessionId: "normalized-detach-buffered",
+      workingDirectory: WORKSPACE,
+    });
+    await session.startTurn({ message: { role: "user", text: "Keep working" } });
+    transport.push("turn/started", { threadId: "thread-1", turn: { id: "turn-1", status: "inProgress", items: [] } });
+    // Let the notification pump buffer these events without a reader.
+    await new Promise(resolve => setTimeout(resolve, 20));
+
+    await session.detachControllerForRestart?.();
+
+    // The next controller replays them from the runner journal instead.
+    await expect(session.events()[Symbol.asyncIterator]().next())
+      .resolves.toEqual({ value: undefined, done: true });
+  });
+
   it("sends direct chat as plain text and permits a follow-up turn", async () => {
     const transport = new FakeCodexTransport();
     const driver = makeDriver([transport], { conversationMode: "direct" });

packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts

@@ -895,5 +895,11 @@ export class CodexHarnessSession
   async detachControllerForRestart(): Promise<void> {
     if (this.transport.detachControllerForRestart === undefined) return;
     await this.transport.detachControllerForRestart();
+    // The next controller replays the runner journal. End this controller's
+    // stream without its buffered suffix so the detached consumer unwinds
+    // instead of waiting for provider events that will never arrive here.
+    // Pending runtime requests stay open for the adopting controller.
+    this.eventQueue.clear();
+    this.eventQueue.close();
   }
 }