From 6a9d485794e76d4ee22d65a37a660bae0214ffcd Mon Sep 17 00:00:00 2001
From: Proactive Runtime Bot <agent@agent-relay.com>
Date: Fri, 7 Aug 2026 13:36:58 +0200
Subject: [PATCH 1/7] fix(cli): treat a lost PTY input stream as one liveness
 event, not one error per keystroke
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A drive session whose PTY input stream died logged

  [drive] input stream send failed: PTY input stream is closed

once per inbound stdin chunk, forever, and stayed alive while doing it.
The session was unusable but looked idle, and exited 0 when the human
finally pressed Ctrl+C.

There was no retry loop. The repetition was 1:1 with stdin: the SDK's
PtyInputStream latches `_closed` and rejects every later send with that
exact string (transport.ts:194-203), and the CLI caught the rejection,
logged, and returned without ever nulling the handle, reading the
`closed` getter it already exposes, or ending the session. Because
KeybindParser forwards every byte except Ctrl+C / Ctrl+], a source TUI
with mouse tracking on turned pointer movement into a flood with no
keystroke at all.

The stream dies easily and asymmetrically: the broker pings the events
WebSocket every 30s (listen_api.rs:2932/:2984) but never pings the input
WebSocket, so an idle input socket is silent on the wire and any idle
timeout reaps it alone — the screen keeps updating while input is dead.
A broker-side write error (PTY worker restart) closes it the same way.
The keepalive gap itself is filed separately as #1450.

Both attach modes now share `attach-input-recovery.ts`: report the loss
once, reopen with bounded exponential backoff, and exit non-zero with a
readable message when that is exhausted.

A reopen resolves by agent *name*, and a name is not an identity, so a
successful reopen is not accepted until the worker process is verified
unchanged. The broker exposes no per-instance token — only the harness
pid on GET /api/spawned (worker.rs:242) — so the check is a heuristic
that fails closed: identity missing at attach, unreadable after
reconnect, or changed all refuse the replacement rather than route
keystrokes into a PTY the human did not attach to.

Partial fix for #1419 — acceptance criteria 2 and 3. The event-WS resume
half (criteria 1, 4, 5: replay reconciliation, snapshot re-sync, queued
message repair, connection rotation) is deliberately left open there.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 packages/cli/src/cli/lib/attach-drive.test.ts | 297 +++++++++++++++++-
 packages/cli/src/cli/lib/attach-drive.ts      | 136 +++++++-
 .../cli/src/cli/lib/attach-input-recovery.ts  | 238 ++++++++++++++
 .../src/cli/lib/attach-passthrough.test.ts    |  90 +++++-
 .../cli/src/cli/lib/attach-passthrough.ts     |  80 ++++-
 .../src/pty-input-stream.test.ts              |  33 ++
 6 files changed, 856 insertions(+), 18 deletions(-)
 create mode 100644 packages/cli/src/cli/lib/attach-input-recovery.ts

diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts
index 634c89185..30805269b 100644
--- a/packages/cli/src/cli/lib/attach-drive.test.ts
+++ b/packages/cli/src/cli/lib/attach-drive.test.ts
@@ -152,6 +152,9 @@ class FakeInputStream implements CliPtyInputStream {
   }
 
   async send(data: string): Promise<{ name: string; bytes_written: number }> {
+    // Mirror the real PtyInputStream: once the socket is gone the guard rejects
+    // immediately and permanently, with this exact message (transport.ts:199).
+    if (this.closed) throw new Error('PTY input stream is closed');
     if (this.sendError) throw this.sendError;
     this.writes.push(data);
     return { name: this.name, bytes_written: Buffer.byteLength(data, 'utf8') };
@@ -162,6 +165,15 @@ class FakeInputStream implements CliPtyInputStream {
     this.closeCode = code;
     this.closeReason = reason;
   }
+
+  /**
+   * Test helper: the broker/proxy drops the socket underneath us. This is the
+   * real-world event — an idle-timeout reap or a PTY worker restart — and it
+   * latches `closed` exactly as the SDK stream does, without the CLI being told.
+   */
+  killFromServer(): void {
+    this.closed = true;
+  }
 }
 
 class FakePredictiveEcho {
@@ -203,6 +215,22 @@ interface FetchScript {
   terminalSize?: { rows: number; cols: number } | null;
   inputStreamOpenError?: Error;
   inputStreamSendError?: Error;
+  /**
+   * Open-errors applied to *reopen* attempts only, in order. `undefined` at an
+   * index means that attempt succeeds. Lets a test make the first N reopens
+   * fail and the N+1th succeed, or make every one fail to reach exhaustion.
+   */
+  reopenOpenErrors?: Array<Error | undefined>;
+  /** Reopen attempts before the session gives up. Defaults to 2 for speed. */
+  inputReopenMaxAttempts?: number;
+  /** Reopen backoff base in ms. Defaults to 1 so tests don't wait. */
+  inputReopenBaseDelayMs?: number;
+  /**
+   * Worker identities returned by successive `getWorkerIdentity` calls. Index 0
+   * is the attach-time baseline; later entries answer post-reopen checks.
+   * Defaults to a stable pid, i.e. "same worker throughout".
+   */
+  workerIdentities?: Array<string | null>;
   /** Ownership re-assert interval (ms). Defaults to disabled (0) in tests so
    *  the keep-alive timer doesn't interfere; the re-assert test sets it small. */
   ownershipReassertMs?: number;
@@ -234,6 +262,7 @@ function createHarness(opts: FetchScript = {}): {
     body?: unknown;
     headers: Record<string, string>;
   }> = [];
+  const identityCalls: Array<string | null> = [];
   const stdin = new FakeStdin();
   const terminal = new FakeTerminal(
     opts.terminalSize === undefined ? { rows: 30, cols: 100 } : opts.terminalSize
@@ -414,7 +443,14 @@ function createHarness(opts: FetchScript = {}): {
     stdin,
     terminal,
     openInputStream: vi.fn((_connection, streamName) => {
-      const stream = new FakeInputStream(streamName, opts.inputStreamOpenError, opts.inputStreamSendError);
+      // Index 0 is the initial open; 1..N are reopen attempts, which a test can
+      // script independently via `reopenOpenErrors`.
+      const reopenIndex = inputStreams.length - 1;
+      const openError =
+        reopenIndex >= 0 && opts.reopenOpenErrors
+          ? opts.reopenOpenErrors[reopenIndex]
+          : opts.inputStreamOpenError;
+      const stream = new FakeInputStream(streamName, openError, opts.inputStreamSendError);
       inputStreams.push(stream);
       return stream;
     }),
@@ -424,6 +460,21 @@ function createHarness(opts: FetchScript = {}): {
     // Disable the ownership re-assert timer by default so it can't perturb
     // resize-count assertions; individual tests opt in with a small value.
     ownershipReassertMs: opts.ownershipReassertMs ?? 0,
+    // Small, deterministic reopen policy: real defaults (5 attempts, 250ms
+    // doubling) would make these tests slow without testing anything more.
+    inputReopenMaxAttempts: opts.inputReopenMaxAttempts ?? 2,
+    inputReopenBaseDelayMs: opts.inputReopenBaseDelayMs ?? 1,
+    getWorkerIdentity: vi.fn(async () => {
+      const scripted = opts.workerIdentities;
+      if (!scripted) return 'pid-1';
+      // Index explicitly: a scripted `null` is a meaningful value ("identity
+      // unavailable"), so `??` must not collapse it into the fallback.
+      const index = identityCalls.length;
+      const value =
+        index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
+      identityCalls.push(value);
+      return value;
+    }),
   };
 
   return {
@@ -2112,3 +2163,247 @@ describe('runDriveSession', () => {
     ]);
   });
 });
+
+
+/**
+ * Regression coverage for #1419: a PTY input stream that dies mid-session used
+ * to log `[drive] input stream send failed: PTY input stream is closed` once
+ * per inbound stdin chunk, forever, while the session stayed alive and
+ * eventually exited 0. Each test below pins one half of that contract.
+ */
+describe('runDriveSession — lost PTY input stream', () => {
+  /** Drain enough microtask/timer turns for the reopen backoff to run out. */
+  async function settleRecovery(turns = 60): Promise<void> {
+    for (let i = 0; i < turns; i++) await new Promise((r) => setTimeout(r, 1));
+  }
+
+  function floodLines(logs: unknown[][], errors: unknown[][]): string[] {
+    return [...logs, ...errors]
+      .map((args) => String(args[0]))
+      .filter((line) => line.includes('input stream'));
+  }
+
+  it('reports the loss exactly once no matter how much input arrives', async () => {
+    // THE FLOOD ASSERTION. Fails if the loss is announced more than once.
+    // 200 SGR mouse reports and ZERO keystrokes: KeybindParser forwards every
+    // byte except 0x03/0x1d, so a source TUI with mouse tracking on generates
+    // this load from pointer movement alone — which is how Khaliq hit it
+    // without typing. Before the fix this produced 200 identical lines.
+    const { deps, sockets, stdin, logs, errors, inputStreams } = createHarness({
+      // Never let the reopen succeed, so the only thing that can vary is how
+      // often the *loss* is announced.
+      reopenOpenErrors: [new Error('still down'), new Error('still down')],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+
+    for (let i = 0; i < 200; i++) {
+      stdin.type(Buffer.from(`\x1b[<35;${i};10M`));
+    }
+    await settleRecovery();
+
+    const lost = floodLines(logs, errors).filter((l) => l.includes('input stream lost'));
+    expect(lost).toHaveLength(1);
+    expect(lost[0]).toContain('reconnecting');
+
+    // And nothing was smuggled onto the dead stream.
+    expect(inputStreams[0].writes).toHaveLength(0);
+    await sessionPromise;
+  });
+
+  it('exits non-zero with a readable message when every reopen fails', async () => {
+    // THE EXIT-CODE ASSERTION. Fails if the session resolves 0 (the old
+    // behaviour: degraded forever, then exit 0 when the human pressed Ctrl+C),
+    // and fails if the operator is not told the agent survived.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 2,
+      reopenOpenErrors: [new Error('broker down'), new Error('broker down')],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    const code = await sessionPromise;
+    expect(code).toBe(1);
+
+    const exhausted = errors
+      .map((args) => String(args[0]))
+      .find((line) => line.includes('could not be reopened'));
+    expect(exhausted).toBeDefined();
+    // A readable message names the attempt count, the agent, and the way out.
+    expect(exhausted).toContain('after 2 attempts');
+    expect(exhausted).toContain('Alice is still running');
+    expect(exhausted).toContain('reattach');
+  });
+
+  it('tries exactly the configured number of reopens, then stops', async () => {
+    // Fails if recovery loops unbounded (the failure mode that would turn a
+    // flood of log lines into a flood of sockets).
+    const { deps, sockets, stdin, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 3,
+      reopenOpenErrors: [new Error('x'), new Error('x'), new Error('x'), new Error('x')],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+    await sessionPromise;
+    await settleRecovery();
+
+    // 1 initial open + exactly 3 reopen attempts.
+    expect(inputStreams).toHaveLength(4);
+  });
+
+  it('recovers and routes later keystrokes to the replacement stream', async () => {
+    // Fails if reopen "succeeds" but input still goes nowhere — i.e. if the
+    // session reports recovery it did not actually achieve.
+    const { deps, sockets, stdin, logs, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('lost'));
+    await settleRecovery();
+
+    expect(inputStreams).toHaveLength(2);
+    expect(logs.map((a) => String(a[0])).filter((l) => l.includes('reconnected'))).toHaveLength(1);
+
+    stdin.type(Buffer.from('typed after recovery'));
+    await settleRecovery(5);
+
+    expect(inputStreams[1].writes.join('')).toBe('typed after recovery');
+    // The keystroke sent during the outage was dropped, not replayed: feeding
+    // stale input into a recovered PTY would execute it out of context.
+    expect(inputStreams[1].writes.join('')).not.toContain('lost');
+
+    stdin.type(Buffer.from([0x03]));
+    expect(await sessionPromise).toBe(0);
+  });
+
+  it('a detach during recovery still exits 0 and cancels the reopen', async () => {
+    // Fails if a user detach mid-outage is misreported as a transport failure,
+    // or if a pending backoff timer fires into a torn-down session.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 4,
+      inputReopenBaseDelayMs: 20,
+      reopenOpenErrors: [new Error('x'), new Error('x'), new Error('x'), new Error('x')],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+    stdin.type(Buffer.from([0x03])); // user detaches while reconnecting
+
+    expect(await sessionPromise).toBe(0);
+    const before = inputStreams.length;
+    await settleRecovery();
+    // No further sockets opened after teardown, and no late error printed.
+    expect(inputStreams).toHaveLength(before);
+    expect(errors.map((a) => String(a[0])).filter((l) => l.includes('could not be reopened'))).toEqual(
+      []
+    );
+  });
+
+  it('refuses a reopen that landed on a different worker process', async () => {
+    // THE IDENTITY ASSERTION. The input stream is reopened *by name*, and a
+    // name is not an identity. If the worker was replaced, a socket that opens
+    // successfully would route the human's keystrokes into a different PTY.
+    // Fails if the session accepts the replacement, or exits 0.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+      // Baseline pid-1 at attach; a different process answers after reconnect.
+      workerIdentities: ['pid-1', 'pid-2'],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    expect(await sessionPromise).toBe(1);
+
+    const refusal = errors
+      .map((args) => String(args[0]))
+      .find((line) => line.includes('not the same worker'));
+    expect(refusal).toBeDefined();
+    expect(refusal).toContain('pid-1');
+    expect(refusal).toContain('pid-2');
+
+    // The replacement socket was opened but must have been closed unused —
+    // nothing may be written to a stream we could not vouch for.
+    expect(inputStreams[1].writes).toHaveLength(0);
+    expect(inputStreams[1].closed).toBe(true);
+  });
+
+  it('refuses a reopen when worker identity cannot be read', async () => {
+    // Fails closed on "don't know", not just on "known different". An
+    // unreadable identity is not evidence of sameness.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+      workerIdentities: ['pid-1', null],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    expect(await sessionPromise).toBe(1);
+    expect(
+      errors.map((args) => String(args[0])).find((l) => l.includes('could not be read'))
+    ).toBeDefined();
+    expect(inputStreams[1].writes).toHaveLength(0);
+  });
+
+  it('refuses a reopen when identity was never established at attach', async () => {
+    // If we never learned who we attached to, we cannot claim the replacement
+    // matches. Fails if a null baseline is treated as a wildcard.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+      workerIdentities: [null, 'pid-9'],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    expect(await sessionPromise).toBe(1);
+    expect(
+      errors.map((args) => String(args[0])).find((l) => l.includes('unavailable at attach'))
+    ).toBeDefined();
+    expect(inputStreams[1].writes).toHaveLength(0);
+  });
+
+  it('rolls back predictive echo once for input that never reached the PTY', async () => {
+    // Fails if the screen keeps optimistically-echoed glyphs for keystrokes the
+    // agent never received — a silent lie about what the agent has seen.
+    const echo = new FakePredictiveEcho();
+    const rollback = vi.spyOn(echo, 'rollback');
+    const { deps, sockets, stdin, inputStreams } = createHarness({
+      predictiveEcho: echo,
+      reopenOpenErrors: [undefined],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    // Several separate chunks during one outage. Pre-fix this rolled back once
+    // per chunk; the contract is one rollback per outage, so a per-chunk
+    // implementation fails here.
+    for (let i = 0; i < 5; i++) stdin.type(Buffer.from(`chunk${i}`));
+    await settleRecovery();
+
+    expect(rollback).toHaveBeenCalledTimes(1);
+    stdin.type(Buffer.from([0x03]));
+    await sessionPromise;
+  });
+});
diff --git a/packages/cli/src/cli/lib/attach-drive.ts b/packages/cli/src/cli/lib/attach-drive.ts
index 1be5ffa27..f1071119a 100644
--- a/packages/cli/src/cli/lib/attach-drive.ts
+++ b/packages/cli/src/cli/lib/attach-drive.ts
@@ -52,6 +52,11 @@ import { StringDecoder } from 'node:string_decoder';
 import type { InboundDeliveryMode } from '@agent-relay/harness-driver';
 import WebSocket from 'ws';
 
+import {
+  createInputStreamRecovery,
+  INPUT_REOPEN_BASE_DELAY_MS,
+  INPUT_REOPEN_MAX_ATTEMPTS,
+} from './attach-input-recovery.js';
 import {
   captureAndRenderSnapshot,
   canReserveStatusLine,
@@ -143,6 +148,13 @@ export interface CliPtyInputStream {
   close(code?: number, reason?: string): void;
   /** Smoothed input→ack RTT (ms), or null before the first ack. */
   readonly srttMs?: number | null;
+  /**
+   * True once the underlying socket has closed. The stream never reopens
+   * itself, so this latches: every later `send()` rejects immediately. Read it
+   * before sending so a dead stream is handled as one liveness event rather
+   * than once per keystroke.
+   */
+  readonly closed?: boolean;
 }
 
 export interface DriveDependencies {
@@ -205,6 +217,24 @@ export interface DriveDependencies {
    * refresh (no SIGWINCH). Defaults to 60000. Set `0` to disable (tests).
    */
   ownershipReassertMs?: number;
+  /**
+   * How many times to reopen a dead PTY input stream before giving up and
+   * exiting non-zero. Defaults to 5. Set `0` to disable recovery and fail on
+   * the first loss.
+   */
+  inputReopenMaxAttempts?: number;
+  /**
+   * Base delay (ms) for the input-stream reopen backoff; doubles per attempt up
+   * to {@link INPUT_REOPEN_MAX_DELAY_MS}. Defaults to 250. Tests set a small
+   * value to keep the backoff deterministic and fast.
+   */
+  inputReopenBaseDelayMs?: number;
+  /**
+   * Reads the identity of the worker process behind `name`, or `null` when it
+   * cannot be established. Used to reject a reopen that landed on a different
+   * process. See {@link fetchWorkerIdentity}.
+   */
+  getWorkerIdentity: (connection: BrokerConnection, name: string) => Promise<string | null>;
 }
 
 function withDefaults(overrides: Partial<DriveDependencies> = {}): DriveDependencies {
@@ -248,6 +278,7 @@ function withDefaults(overrides: Partial<DriveDependencies> = {}): DriveDependen
       },
     },
     openInputStream: (connection, name, options) => openPtyInputStream(connection, name, fetchFn, options),
+    getWorkerIdentity: (connection, name) => fetchWorkerIdentity(connection, name, fetchFn),
     createPredictiveEcho,
     ...overrides,
   };
@@ -376,6 +407,38 @@ export async function sendInput(
   }
 }
 
+/**
+ * Best-available identity for the worker process behind `name`, or `null` when
+ * it cannot be established.
+ *
+ * The broker exposes no per-instance token for a worker — no `instance_id`,
+ * `run_id`, `epoch`, or absolute spawn timestamp reaches the wire. The only
+ * restart-discriminating value on `GET /api/spawned` is the harness `pid`
+ * (`crates/broker/src/worker.rs:242`, an `Option<u32>` that is null until the
+ * worker reports ready), so that is what this returns.
+ *
+ * This is a heuristic, not a nonce: the OS can reuse a pid. It is used to
+ * *reject* a reopen that lands on a visibly different process, never to prove
+ * two processes are the same — callers treat `null` as "cannot verify" and
+ * fail closed. The durable fix is for the broker to surface the per-spawn
+ * identity it already holds in memory (`WorkerHandle.spawned_at`,
+ * `PersistedAgent.started_at`, or the `MetricsCollector` spawn counter).
+ */
+export async function fetchWorkerIdentity(
+  connection: BrokerConnection,
+  name: string,
+  fetchFn: typeof globalThis.fetch
+): Promise<string | null> {
+  try {
+    const agents = await createBrokerClient(connection, fetchFn).listAgents();
+    const agent = agents.find((candidate) => candidate.name === name);
+    if (!agent || typeof agent.pid !== 'number') return null;
+    return String(agent.pid);
+  } catch {
+    return null;
+  }
+}
+
 /** Open the SDK-backed raw PTY input stream for interactive CLI sessions. */
 export function openPtyInputStream(
   connection: BrokerConnection,
@@ -899,6 +962,46 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
       return deliveryToggleInFlight;
     };
 
+    // ---- input-stream liveness ----
+    // A closed PTY input stream is a session-liveness event, not a
+    // per-keystroke error. See `attach-input-recovery.ts` for why (#1419).
+    //
+    // Identity of the worker this session attached to, captured once at attach
+    // and compared after any reopen. `null` means the broker could not tell us,
+    // which is treated as "cannot verify" — never as "verified".
+    let attachedWorkerIdentity: string | null = null;
+    const inputRecovery = createInputStreamRecovery({
+      label: 'drive',
+      name,
+      maxAttempts: deps.inputReopenMaxAttempts ?? INPUT_REOPEN_MAX_ATTEMPTS,
+      baseDelayMs: deps.inputReopenBaseDelayMs ?? INPUT_REOPEN_BASE_DELAY_MS,
+      log: (message) => deps.log(message),
+      error: (message) => deps.error(message),
+      isSettled: () => settled,
+      getStream: () => inputStream,
+      setStream: (stream) => {
+        inputStream = stream;
+      },
+      openStream: () => deps.openInputStream(connection, name),
+      onRollback: () => predictiveEcho?.rollback(),
+      onExhausted: () => finish(1),
+      verifyIdentity: async () => {
+        // Fail closed in both directions: if we never learned who we attached
+        // to, we cannot claim the replacement is the same process either.
+        if (attachedWorkerIdentity === null) {
+          return { ok: false, reason: 'worker identity was unavailable at attach' };
+        }
+        const current = await deps.getWorkerIdentity(connection, name);
+        if (current === null) {
+          return { ok: false, reason: 'worker identity could not be read after reconnect' };
+        }
+        if (current !== attachedWorkerIdentity) {
+          return { ok: false, reason: `worker process changed (pid ${attachedWorkerIdentity} → ${current})` };
+        }
+        return { ok: true };
+      },
+    });
+
     // ---- stdin handling ----
     let stdinReady = false;
     const stdinDataHandler = (chunk: Buffer): void => {
@@ -913,8 +1016,14 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
       const outcome = parser.feed(chunk);
       if (outcome.forward.length > 0) {
         const stream = inputStream;
-        if (!stream) {
-          deps.log('[drive] input stream is not ready');
+        // A dead or missing stream is a liveness event, not a per-keystroke
+        // error. Drop this input silently — recovery has already announced
+        // itself — rather than emitting a line for every byte the terminal
+        // sends us. Input during the outage is dropped, not buffered: replaying
+        // stale keystrokes into a recovered PTY would execute them out of
+        // context, which is worse than losing them.
+        if (!inputRecovery.isUsable(stream)) {
+          inputRecovery.recover('stream closed');
           return;
         }
         // Decode through the stateful UTF-8 decoder so a multi-byte character
@@ -923,15 +1032,12 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
         // next chunk completes it.
         const decoded = inputDecoder.write(outcome.forward);
         if (decoded.length > 0) {
-          // Fire-and-forget; surface errors via log but don't block the
-          // event loop on every keystroke.
+          // Fire-and-forget; don't block the event loop on every keystroke.
           void stream.send(decoded).catch((err: unknown) => {
             if (settled) return;
-            const message = describeError(err);
-            deps.log(`[drive] input stream send failed: ${message}`);
-            // The keystroke never reached the PTY — drop any optimistic echo
-            // for it so the screen doesn't show input the agent didn't get.
-            predictiveEcho?.rollback();
+            // Only the first failure speaks; `recover` no-ops while a recovery
+            // is already in flight, and it does the echo rollback.
+            inputRecovery.recover(describeError(err));
           });
         }
         predictiveEcho?.onUserInput(outcome.forward);
@@ -998,6 +1104,9 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
     };
 
     const closeInputStream = (): void => {
+      // Cancel any pending reopen backoff so a detach mid-recovery doesn't
+      // leave a timer holding a reference to a torn-down session.
+      inputRecovery.cancel();
       const stream = inputStream;
       inputStream = null;
       if (!stream) return;
@@ -1116,6 +1225,15 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
           closeInputStream();
           return;
         }
+        // Baseline for the reopen identity gate. Best-effort: a broker that
+        // won't tell us leaves this null, which makes any later reopen refuse
+        // rather than guess. Not fatal here — the initial attach is the seat
+        // the human asked for.
+        attachedWorkerIdentity = await deps.getWorkerIdentity(connection, name);
+        if (settled) {
+          closeInputStream();
+          return;
+        }
         // Register the temporary input handler before raw mode. Ctrl+C is an
         // ordinary byte in raw mode, so this keeps detach available while a
         // snapshot fetch or initial resize is still pending.
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
new file mode 100644
index 000000000..4c882b5d9
--- /dev/null
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -0,0 +1,238 @@
+/**
+ * Recovery for a lost PTY input stream, shared by `attach --mode drive` and
+ * `attach --mode passthrough`.
+ *
+ * The SDK's `PtyInputStream` never reopens itself. Once its socket closes the
+ * `closed` flag latches and every later `send()` rejects immediately with
+ * `PTY input stream is closed`. That close is easy to hit and easy to miss: the
+ * broker pings the *events* WebSocket every 30s but never pings the *input*
+ * WebSocket, so an idle input socket is silent on the wire and any idle timeout
+ * between client and broker kills it alone — the screen keeps updating while
+ * input is permanently dead. A broker-side write error (PTY worker restart)
+ * closes it the same way.
+ *
+ * Before this, both attach modes caught that rejection per keystroke, logged
+ * it, and returned. Nothing tore the session down, so the line repeated for as
+ * long as stdin produced bytes. There was no retry loop — the repetition was
+ * 1:1 with inbound chunks, and since the keybind parser forwards every byte
+ * except Ctrl+C / Ctrl+], a source TUI with mouse tracking enabled flooded the
+ * terminal on pointer movement alone, with the human never touching a key.
+ *
+ * This turns that into one liveness event: report once, reopen with bounded
+ * exponential backoff, and on exhaustion hand control back to the caller so the
+ * session exits non-zero. A dead seat that looks alive is the defect; a
+ * readable exit a supervisor can act on is the contract (#1419).
+ */
+
+import type { CliPtyInputStream } from './attach-drive.js';
+
+/** Default number of reopen attempts before a lost input stream ends the session. */
+export const INPUT_REOPEN_MAX_ATTEMPTS = 5;
+/** Default base delay for the reopen backoff; doubles per attempt. */
+export const INPUT_REOPEN_BASE_DELAY_MS = 250;
+/** Ceiling on the doubled reopen delay. */
+export const INPUT_REOPEN_MAX_DELAY_MS = 4_000;
+
+export interface InputStreamRecoveryOptions {
+  /** Log prefix tag — `drive` or `passthrough`. */
+  label: string;
+  /** Agent name, used in the operator-facing exhaustion message. */
+  name: string;
+  /** Attempts before giving up. `0` disables recovery: the first loss exits. */
+  maxAttempts: number;
+  /** Base backoff delay in ms; doubles per attempt up to the max. */
+  baseDelayMs: number;
+  log: (message: string) => void;
+  error: (message: string) => void;
+  /** True once the session has begun tearing down; stops all recovery work. */
+  isSettled: () => boolean;
+  getStream: () => CliPtyInputStream | null;
+  setStream: (stream: CliPtyInputStream | null) => void;
+  /** Opens a replacement stream. May throw; a throw counts as a failed attempt. */
+  openStream: () => CliPtyInputStream;
+  /** Drop optimistic echo for keystrokes that never reached the PTY. */
+  onRollback: () => void;
+  /** Called when every attempt failed. Callers exit non-zero here. */
+  onExhausted: () => void;
+  /**
+   * Proves the reopened stream reached the SAME worker process the session
+   * originally attached to, and is called after every successful reopen.
+   *
+   * This gate exists because the input stream is reopened *by agent name*, and
+   * a name is not an identity: if the worker died and something else claimed
+   * the name, a "successful" reopen would quietly route the human's keystrokes
+   * into a different agent's PTY. Restarting the same agent is equally wrong
+   * for input safety — keystrokes typed for the old session's context would
+   * land in a fresh shell.
+   *
+   * Must fail closed: anything other than a positive match — identity
+   * unavailable, unreadable, or changed — has to return `ok: false` so the
+   * session exits loudly instead of reattaching on a guess.
+   */
+  verifyIdentity?: () => Promise<{ ok: true } | { ok: false; reason: string }>;
+}
+
+export interface InputStreamRecovery {
+  /**
+   * True when `stream` is present and not known-dead. Callers check this
+   * *before* sending, so a dead stream costs one liveness event instead of one
+   * rejected promise (and one log line) per keystroke. A type predicate so the
+   * caller's handle narrows to non-null on the sending path.
+   */
+  isUsable(stream: CliPtyInputStream | null): stream is CliPtyInputStream;
+  /** True while a reopen is in flight. */
+  isRecovering(): boolean;
+  /** Begin recovery. No-ops if already recovering or settled. */
+  recover(reason: string): void;
+  /** Cancel a pending backoff timer (detach mid-recovery). */
+  cancel(): void;
+}
+
+export function createInputStreamRecovery(
+  options: InputStreamRecoveryOptions
+): InputStreamRecovery {
+  const {
+    label,
+    name,
+    maxAttempts,
+    baseDelayMs,
+    log,
+    error,
+    isSettled,
+    getStream,
+    setStream,
+    openStream,
+    onRollback,
+    onExhausted,
+    verifyIdentity,
+  } = options;
+
+  let inFlight: Promise<void> | null = null;
+  let timer: ReturnType<typeof setTimeout> | null = null;
+
+  const cancel = (): void => {
+    if (timer) {
+      clearTimeout(timer);
+      timer = null;
+    }
+  };
+
+  const isUsable = (stream: CliPtyInputStream | null): stream is CliPtyInputStream =>
+    stream !== null && stream.closed !== true;
+
+  const recover = (reason: string): void => {
+    if (isSettled() || inFlight) return;
+
+    // Drop the dead handle first: `isDead()` then short-circuits every chunk
+    // that arrives mid-recovery, which is what actually silences the flood.
+    const dead = getStream();
+    setStream(null);
+    try {
+      dead?.close(1000, `${label} client replacing input stream`);
+    } catch {
+      // best effort — already closed in the common case
+    }
+    onRollback();
+
+    if (maxAttempts <= 0) {
+      error(
+        `[${label}] input stream lost (${reason}); reconnect is disabled. ` +
+          `Detaching — ${name} is still running; reattach to resume.`
+      );
+      onExhausted();
+      return;
+    }
+
+    // One line for the outage, not one per keystroke.
+    log(`[${label}] input stream lost (${reason}); reconnecting…`);
+
+    const closeQuietly = (stream: CliPtyInputStream, why: string): void => {
+      try {
+        stream.close(1000, why);
+      } catch {
+        // best effort
+      }
+    };
+
+    /**
+     * One reopen attempt. `'settled'` means the session went away mid-attempt,
+     * `'retry'` a transport failure worth another go, and `'rejected'` a
+     * replacement that opened but could not be vouched for — which must not be
+     * retried, because a replaced worker does not become the original one on a
+     * later attempt.
+     */
+    const attemptReopen = async (
+      attempt: number
+    ): Promise<'opened' | 'retry' | 'rejected' | 'settled'> => {
+      let replacement: CliPtyInputStream;
+      try {
+        replacement = openStream();
+        await replacement.waitUntilOpen();
+      } catch {
+        // Stay quiet between attempts. The human saw one line when the outage
+        // started and sees exactly one more when it resolves either way;
+        // narrating each failed retry would rebuild the flood.
+        return 'retry';
+      }
+      if (isSettled()) {
+        closeQuietly(replacement, `${label} client exiting`);
+        return 'settled';
+      }
+
+      // The socket is open, but "open" only proves the name resolved. Do not
+      // hand the human's keystrokes to it until it is the same worker.
+      if (verifyIdentity) {
+        const verdict = await verifyIdentity();
+        if (isSettled()) return 'settled';
+        if (!verdict.ok) {
+          closeQuietly(replacement, `${label} client rejected replacement`);
+          error(
+            `[${label}] input stream reopened but it is not the same worker (${verdict.reason}). ` +
+              `Refusing to forward input — your keystrokes would go somewhere you did not attach to. ` +
+              `Detaching; reattach to ${name} to continue.`
+          );
+          return 'rejected';
+        }
+      }
+
+      setStream(replacement);
+      log(`[${label}] input stream reconnected after ${attempt} attempt(s)`);
+      return 'opened';
+    };
+
+    inFlight = (async () => {
+      for (let attempt = 1; attempt <= maxAttempts; attempt++) {
+        if (isSettled()) return;
+        const delay = Math.min(baseDelayMs * 2 ** (attempt - 1), INPUT_REOPEN_MAX_DELAY_MS);
+        await new Promise<void>((resolve) => {
+          timer = setTimeout(resolve, delay);
+          timer.unref?.();
+        });
+        timer = null;
+        if (isSettled()) return;
+
+        const outcome = await attemptReopen(attempt);
+        if (outcome === 'opened' || outcome === 'settled') return;
+        if (outcome === 'rejected') {
+          onExhausted();
+          return;
+        }
+      }
+      if (isSettled()) return;
+      error(
+        `[${label}] input stream could not be reopened after ${maxAttempts} attempts (${reason}). ` +
+          `Detaching — ${name} is still running; reattach to resume.`
+      );
+      onExhausted();
+    })().finally(() => {
+      inFlight = null;
+    });
+  };
+
+  return {
+    isUsable,
+    isRecovering: () => inFlight !== null,
+    recover,
+    cancel,
+  };
+}
diff --git a/packages/cli/src/cli/lib/attach-passthrough.test.ts b/packages/cli/src/cli/lib/attach-passthrough.test.ts
index f0ee36178..154ac5ff7 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.test.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.test.ts
@@ -138,6 +138,9 @@ class FakeInputStream implements CliPtyInputStream {
   }
 
   async send(data: string): Promise<{ name: string; bytes_written: number }> {
+    // Mirror the real PtyInputStream: once the socket is gone the guard rejects
+    // immediately and permanently (transport.ts:199).
+    if (this.closed) throw new Error('PTY input stream is closed');
     if (this.sendError) throw this.sendError;
     this.writes.push(data);
     return { name: this.name, bytes_written: Buffer.byteLength(data, 'utf8') };
@@ -148,6 +151,11 @@ class FakeInputStream implements CliPtyInputStream {
     this.closeCode = code;
     this.closeReason = reason;
   }
+
+  /** Test helper: the broker/proxy drops the socket without telling the CLI. */
+  killFromServer(): void {
+    this.closed = true;
+  }
 }
 
 type FetchRoute = (init?: RequestInit) => Promise<Response>;
@@ -160,6 +168,16 @@ interface FetchScript {
   terminalSize?: { rows: number; cols: number } | null;
   inputStreamOpenError?: Error;
   inputStreamSendError?: Error;
+  /** Open-errors applied to reopen attempts only, in order. */
+  reopenOpenErrors?: Array<Error | undefined>;
+  inputReopenMaxAttempts?: number;
+  inputReopenBaseDelayMs?: number;
+  /**
+   * Worker identities returned by successive `getWorkerIdentity` calls. Index 0
+   * is the attach-time baseline; later entries answer post-reopen checks.
+   * Defaults to a stable pid, i.e. "same worker throughout".
+   */
+  workerIdentities?: Array<string | null>;
   /** When set, inject this fake engine via the createPredictiveEcho factory. */
   predictiveEcho?: FakePredictiveEcho;
 }
@@ -220,6 +238,7 @@ function createHarness(opts: FetchScript = {}): {
     body?: unknown;
     headers: Record<string, string>;
   }> = [];
+  const identityCalls: Array<string | null> = [];
   const stdin = new FakeStdin();
   const terminal = new FakeTerminal(
     opts.terminalSize === undefined ? { rows: 30, cols: 100 } : opts.terminalSize
@@ -370,13 +389,32 @@ function createHarness(opts: FetchScript = {}): {
     stdin,
     terminal,
     openInputStream: vi.fn((_connection, streamName) => {
-      const stream = new FakeInputStream(streamName, opts.inputStreamOpenError, opts.inputStreamSendError);
+      const reopenIndex = inputStreams.length - 1;
+      const openError =
+        reopenIndex >= 0 && opts.reopenOpenErrors
+          ? opts.reopenOpenErrors[reopenIndex]
+          : opts.inputStreamOpenError;
+      const stream = new FakeInputStream(streamName, openError, opts.inputStreamSendError);
       inputStreams.push(stream);
       return stream;
     }),
     createPredictiveEcho: opts.predictiveEcho ? () => opts.predictiveEcho ?? null : undefined,
     // Immediate, deterministic status repaints in tests (no coalescing timer).
     statusRepaintCoalesceMs: 0,
+    // Small, deterministic reopen policy so tests don't wait on real backoff.
+    inputReopenMaxAttempts: opts.inputReopenMaxAttempts ?? 2,
+    inputReopenBaseDelayMs: opts.inputReopenBaseDelayMs ?? 1,
+    getWorkerIdentity: vi.fn(async () => {
+      const scripted = opts.workerIdentities;
+      if (!scripted) return 'pid-1';
+      // Index explicitly: a scripted `null` is a meaningful value ("identity
+      // unavailable"), so `??` must not collapse it into the fallback.
+      const index = identityCalls.length;
+      const value =
+        index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
+      identityCalls.push(value);
+      return value;
+    }),
   };
 
   return {
@@ -1199,3 +1237,53 @@ describe('runPassthroughSession', () => {
     ]);
   });
 });
+
+/**
+ * Passthrough carries the same lost-input-stream defect drive did (#1419) and
+ * now shares its recovery (`attach-input-recovery.ts`). These pin the two
+ * halves of the contract that matter most; the drive suite covers the rest of
+ * the shared behaviour.
+ */
+describe('runPassthroughSession — lost PTY input stream', () => {
+  async function settleRecovery(turns = 60): Promise<void> {
+    for (let i = 0; i < turns; i++) await new Promise((r) => setTimeout(r, 1));
+  }
+
+  it('reports the loss exactly once however much input arrives', async () => {
+    const { deps, sockets, stdin, logs, errors, inputStreams } = createHarness({
+      reopenOpenErrors: [new Error('still down'), new Error('still down')],
+    });
+    const sessionPromise = runPassthroughSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    // Mouse reports, not keystrokes — the amplifier that made this a flood.
+    for (let i = 0; i < 200; i++) stdin.type(Buffer.from(`\x1b[<35;${i};10M`));
+    await settleRecovery();
+
+    const lost = [...logs, ...errors]
+      .map((args) => String(args[0]))
+      .filter((line) => line.includes('input stream lost'));
+    expect(lost).toHaveLength(1);
+    expect(lost[0]).toContain('[passthrough]');
+    await sessionPromise;
+  });
+
+  it('exits non-zero with a readable message when every reopen fails', async () => {
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 2,
+      reopenOpenErrors: [new Error('broker down'), new Error('broker down')],
+    });
+    const sessionPromise = runPassthroughSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    expect(await sessionPromise).toBe(1);
+    const exhausted = errors
+      .map((args) => String(args[0]))
+      .find((line) => line.includes('could not be reopened'));
+    expect(exhausted).toContain('Alice is still running');
+  });
+});
diff --git a/packages/cli/src/cli/lib/attach-passthrough.ts b/packages/cli/src/cli/lib/attach-passthrough.ts
index 765f34d58..4b4460ad7 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.ts
@@ -57,8 +57,14 @@ import {
   type BrokerConnection,
 } from '../lib/broker-connection.js';
 import { defaultExit, runSignalHandler } from '../lib/exit.js';
+import {
+  createInputStreamRecovery,
+  INPUT_REOPEN_BASE_DELAY_MS,
+  INPUT_REOPEN_MAX_ATTEMPTS,
+} from './attach-input-recovery.js';
 import {
   type CliPtyInputStream,
+  fetchWorkerIdentity,
   openPtyInputStream,
   releaseResizeOwnership,
   resizeWorker,
@@ -150,6 +156,19 @@ export interface PassthroughDependencies {
    * refresh (no SIGWINCH). Defaults to 60000. Set `0` to disable (tests).
    */
   ownershipReassertMs?: number;
+  /**
+   * How many times to reopen a dead PTY input stream before giving up and
+   * exiting non-zero. Defaults to 5. Set `0` to disable recovery.
+   */
+  inputReopenMaxAttempts?: number;
+  /** Base delay (ms) for the input-stream reopen backoff. Defaults to 250. */
+  inputReopenBaseDelayMs?: number;
+  /**
+   * Reads the identity of the worker process behind `name`, or `null` when it
+   * cannot be established. Used to reject a reopen that landed on a different
+   * process. See `fetchWorkerIdentity` in attach-drive.ts.
+   */
+  getWorkerIdentity: (connection: BrokerConnection, name: string) => Promise<string | null>;
 }
 
 function withDefaults(overrides: Partial<PassthroughDependencies> = {}): PassthroughDependencies {
@@ -189,6 +208,7 @@ function withDefaults(overrides: Partial<PassthroughDependencies> = {}): Passthr
       },
     },
     openInputStream: (connection, name) => openPtyInputStream(connection, name, fetchFn),
+    getWorkerIdentity: (connection, name) => fetchWorkerIdentity(connection, name, fetchFn),
     createPredictiveEcho,
     ...overrides,
   };
@@ -538,6 +558,43 @@ export async function runPassthroughSession(
       paintStatus();
     };
 
+    // ---- input-stream liveness ----
+    // Same defect and same contract as drive; see `attach-input-recovery.ts`.
+    // Identity of the worker this session attached to; see attach-drive.ts.
+    let attachedWorkerIdentity: string | null = null;
+    const inputRecovery = createInputStreamRecovery({
+      label: 'passthrough',
+      name,
+      maxAttempts: deps.inputReopenMaxAttempts ?? INPUT_REOPEN_MAX_ATTEMPTS,
+      baseDelayMs: deps.inputReopenBaseDelayMs ?? INPUT_REOPEN_BASE_DELAY_MS,
+      log: (message) => deps.log(message),
+      error: (message) => deps.error(message),
+      isSettled: () => settled,
+      getStream: () => inputStream,
+      setStream: (stream) => {
+        inputStream = stream;
+      },
+      openStream: () => deps.openInputStream(connection, name),
+      onRollback: () => predictiveEcho?.rollback(),
+      onExhausted: () => finish(1),
+      verifyIdentity: async () => {
+        if (attachedWorkerIdentity === null) {
+          return { ok: false, reason: 'worker identity was unavailable at attach' };
+        }
+        const current = await deps.getWorkerIdentity(connection, name);
+        if (current === null) {
+          return { ok: false, reason: 'worker identity could not be read after reconnect' };
+        }
+        if (current !== attachedWorkerIdentity) {
+          return {
+            ok: false,
+            reason: `worker process changed (pid ${attachedWorkerIdentity} → ${current})`,
+          };
+        }
+        return { ok: true };
+      },
+    });
+
     let stdinReady = false;
     const stdinDataHandler = (chunk: Buffer): void => {
       // Raw mode starts before snapshot replay so terminal input reports cannot
@@ -551,8 +608,11 @@ export async function runPassthroughSession(
       const outcome = parser.feed(chunk);
       if (outcome.forward.length > 0) {
         const stream = inputStream;
-        if (!stream) {
-          deps.log('[passthrough] input stream is not ready');
+        // A dead or missing stream is a liveness event, not a per-keystroke
+        // error — see `attach-input-recovery.ts` (#1419). Drop this input
+        // silently; recovery has already announced itself once.
+        if (!inputRecovery.isUsable(stream)) {
+          inputRecovery.recover('stream closed');
           return;
         }
         // Decode through the stateful UTF-8 decoder so a multi-byte character
@@ -561,11 +621,7 @@ export async function runPassthroughSession(
         if (decoded.length > 0) {
           void stream.send(decoded).catch((err: unknown) => {
             if (settled) return;
-            const message = describeError(err);
-            deps.log(`[passthrough] input stream send failed: ${message}`);
-            // The keystroke never reached the PTY — drop any optimistic echo
-            // for it so the screen doesn't show input the agent didn't get.
-            predictiveEcho?.rollback();
+            inputRecovery.recover(describeError(err));
           });
         }
         predictiveEcho?.onUserInput(outcome.forward);
@@ -629,6 +685,9 @@ export async function runPassthroughSession(
     };
 
     const closeInputStream = (): void => {
+      // Cancel any pending reopen backoff so a detach mid-recovery doesn't
+      // leave a timer holding a reference to a torn-down session.
+      inputRecovery.cancel();
       const stream = inputStream;
       inputStream = null;
       if (!stream) return;
@@ -730,6 +789,13 @@ export async function runPassthroughSession(
           closeInputStream();
           return;
         }
+        // Baseline for the reopen identity gate; null makes a later reopen
+        // refuse rather than guess. See attach-drive.ts.
+        attachedWorkerIdentity = await deps.getWorkerIdentity(connection, name);
+        if (settled) {
+          closeInputStream();
+          return;
+        }
         // Register the temporary input handler before raw mode. Ctrl+C is an
         // ordinary byte in raw mode, so this keeps detach available while a
         // snapshot fetch or initial resize is still pending.
diff --git a/packages/harness-driver/src/pty-input-stream.test.ts b/packages/harness-driver/src/pty-input-stream.test.ts
index fd394204f..e313488ec 100644
--- a/packages/harness-driver/src/pty-input-stream.test.ts
+++ b/packages/harness-driver/src/pty-input-stream.test.ts
@@ -189,3 +189,36 @@ describe('PtyInputStream pipelining', () => {
     await expect(p2).rejects.toMatchObject({ code: 'input_stream_closed' });
   });
 });
+
+describe('PtyInputStream after its socket closes', () => {
+  /**
+   * The CLI's recovery logic (attach-input-recovery.ts) is built on the
+   * assumption that this stream never heals itself, so that assumption is
+   * pinned here. If PtyInputStream ever grows its own reconnect, this test
+   * fails and the CLI-side recovery must be revisited rather than silently
+   * doubling up.
+   */
+  it('latches closed and rejects every later send with the same message', async () => {
+    const stream = new PtyInputStream({ url: 'ws://x/api/input/agent/stream' });
+    const socket = lastSocket();
+    socket.open();
+    await stream.waitUntilOpen();
+
+    // An idle-timeout reap or a PTY worker restart: abnormal close, no warning.
+    socket.emit('close', 1006, Buffer.from(''));
+    expect(stream.closed).toBe(true);
+
+    const messages: string[] = [];
+    for (let i = 0; i < 50; i++) {
+      await stream.send('x').catch((err: Error) => messages.push(err.message));
+    }
+
+    // Every send rejects — no retry, no backoff, no self-heal. This exact
+    // string is what used to reach the terminal once per keystroke (#1419).
+    expect(messages).toHaveLength(50);
+    expect([...new Set(messages)]).toEqual(['PTY input stream is closed']);
+    // Nothing was put back on the wire, and no replacement socket was opened.
+    expect(socket.sends).toHaveLength(0);
+    expect(FakeWebSocket.instances).toHaveLength(1);
+  });
+});

From d8f0683fab6be0cf3e9c1fc42f2df347f1362ef6 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]" <github-actions[bot]@users.noreply.github.com>
Date: Fri, 7 Aug 2026 11:38:58 +0000
Subject: [PATCH 2/7] style: auto-format with Prettier

---
 packages/cli/src/cli/lib/attach-drive.test.ts       | 12 +++---------
 packages/cli/src/cli/lib/attach-input-recovery.ts   |  8 ++------
 packages/cli/src/cli/lib/attach-passthrough.test.ts |  3 +--
 3 files changed, 6 insertions(+), 17 deletions(-)

diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts
index 30805269b..1f737c0c8 100644
--- a/packages/cli/src/cli/lib/attach-drive.test.ts
+++ b/packages/cli/src/cli/lib/attach-drive.test.ts
@@ -470,8 +470,7 @@ function createHarness(opts: FetchScript = {}): {
       // Index explicitly: a scripted `null` is a meaningful value ("identity
       // unavailable"), so `??` must not collapse it into the fallback.
       const index = identityCalls.length;
-      const value =
-        index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
+      const value = index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
       identityCalls.push(value);
       return value;
     }),
@@ -2164,7 +2163,6 @@ describe('runDriveSession', () => {
   });
 });
 
-
 /**
  * Regression coverage for #1419: a PTY input stream that dies mid-session used
  * to log `[drive] input stream send failed: PTY input stream is closed` once
@@ -2307,9 +2305,7 @@ describe('runDriveSession — lost PTY input stream', () => {
     await settleRecovery();
     // No further sockets opened after teardown, and no late error printed.
     expect(inputStreams).toHaveLength(before);
-    expect(errors.map((a) => String(a[0])).filter((l) => l.includes('could not be reopened'))).toEqual(
-      []
-    );
+    expect(errors.map((a) => String(a[0])).filter((l) => l.includes('could not be reopened'))).toEqual([]);
   });
 
   it('refuses a reopen that landed on a different worker process', async () => {
@@ -2357,9 +2353,7 @@ describe('runDriveSession — lost PTY input stream', () => {
     stdin.type(Buffer.from('a'));
 
     expect(await sessionPromise).toBe(1);
-    expect(
-      errors.map((args) => String(args[0])).find((l) => l.includes('could not be read'))
-    ).toBeDefined();
+    expect(errors.map((args) => String(args[0])).find((l) => l.includes('could not be read'))).toBeDefined();
     expect(inputStreams[1].writes).toHaveLength(0);
   });
 
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
index 4c882b5d9..657bb02fd 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -88,9 +88,7 @@ export interface InputStreamRecovery {
   cancel(): void;
 }
 
-export function createInputStreamRecovery(
-  options: InputStreamRecoveryOptions
-): InputStreamRecovery {
+export function createInputStreamRecovery(options: InputStreamRecoveryOptions): InputStreamRecovery {
   const {
     label,
     name,
@@ -161,9 +159,7 @@ export function createInputStreamRecovery(
      * retried, because a replaced worker does not become the original one on a
      * later attempt.
      */
-    const attemptReopen = async (
-      attempt: number
-    ): Promise<'opened' | 'retry' | 'rejected' | 'settled'> => {
+    const attemptReopen = async (attempt: number): Promise<'opened' | 'retry' | 'rejected' | 'settled'> => {
       let replacement: CliPtyInputStream;
       try {
         replacement = openStream();
diff --git a/packages/cli/src/cli/lib/attach-passthrough.test.ts b/packages/cli/src/cli/lib/attach-passthrough.test.ts
index 154ac5ff7..a6b842765 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.test.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.test.ts
@@ -410,8 +410,7 @@ function createHarness(opts: FetchScript = {}): {
       // Index explicitly: a scripted `null` is a meaningful value ("identity
       // unavailable"), so `??` must not collapse it into the fallback.
       const index = identityCalls.length;
-      const value =
-        index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
+      const value = index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
       identityCalls.push(value);
       return value;
     }),

From 7b5a1860cb094f3a6967a3cfa3f8b82ea9cb0193 Mon Sep 17 00:00:00 2001
From: Proactive Runtime Bot <agent@agent-relay.com>
Date: Fri, 7 Aug 2026 13:48:27 +0200
Subject: [PATCH 3/7] fix(cli): key worker identity on workerPid, not the
 harness pid
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Verified against a live broker: a PTY worker reports `pid: null` and
`workerPid: 30209`. The harness `pid` (worker.rs:242 =
`handle.harness_pid`) stays null until the worker completes the harness
ready handshake (worker_events.rs:849), while `workerPid` (worker.rs:243
= `handle.child.id()`) is the PTY child itself and is populated as soon
as the worker spawns.

Keying the reopen identity gate on `pid` alone therefore made identity
unverifiable — and so every reopen refused — for exactly the class of
worker drive attaches to. Prefer `workerPid`, fold in the harness pid
when the broker has both, and return null only when neither is present.

`workerPid` is on the wire but absent from the `ListAgent` type, so it is
read off the record defensively rather than widening the contract here.

Found by the live-broker reproduction, not by the unit suite — the fakes
had modelled `pid` as always present.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 packages/cli/src/cli/lib/attach-drive.test.ts | 52 +++++++++++++++++++
 packages/cli/src/cli/lib/attach-drive.ts      | 34 +++++++++---
 .../cli/src/cli/lib/attach-passthrough.ts     |  2 +-
 3 files changed, 79 insertions(+), 9 deletions(-)

diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts
index 30805269b..5a6869366 100644
--- a/packages/cli/src/cli/lib/attach-drive.test.ts
+++ b/packages/cli/src/cli/lib/attach-drive.test.ts
@@ -5,6 +5,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest';
 
 import { LOCAL_TERMINAL_RESET_SEQUENCE } from './attach.js';
 import {
+  fetchWorkerIdentity,
   KeybindParser,
   classifyWsEvent,
   renderStatusLine,
@@ -2407,3 +2408,54 @@ describe('runDriveSession — lost PTY input stream', () => {
     await sessionPromise;
   });
 });
+
+describe('fetchWorkerIdentity', () => {
+  const connection = { url: 'http://localhost:3889', apiKey: 'k' };
+
+  function fetchReturning(agents: unknown[]): typeof globalThis.fetch {
+    return (async () =>
+      new Response(JSON.stringify({ agents }), {
+        status: 200,
+        headers: { 'Content-Type': 'application/json' },
+      })) as unknown as typeof globalThis.fetch;
+  }
+
+  it('uses workerPid when the harness pid is null', async () => {
+    // The shape a live broker actually returns for a plain PTY worker: the
+    // harness `pid` stays null until the ready handshake, while `workerPid`
+    // (the PTY child) is populated immediately. Keying on `pid` alone made
+    // every reopen unverifiable for exactly the workers drive attaches to.
+    const identity = await fetchWorkerIdentity(
+      connection,
+      'Alice',
+      fetchReturning([{ name: 'Alice', runtime: 'pty', channels: [], pid: null, workerPid: 30209 }])
+    );
+    expect(identity).toBe('worker:30209');
+  });
+
+  it('folds in the harness pid when the broker has both', async () => {
+    // A change in *either* process means the thing behind the name changed.
+    const identity = await fetchWorkerIdentity(
+      connection,
+      'Alice',
+      fetchReturning([{ name: 'Alice', runtime: 'pty', channels: [], pid: 99778, workerPid: 30209 }])
+    );
+    expect(identity).toBe('worker:30209/harness:99778');
+  });
+
+  it('returns null when the broker reports no pid of either kind', async () => {
+    // "Cannot verify" — the caller must fail closed rather than treat a
+    // missing identity as a match.
+    const identity = await fetchWorkerIdentity(
+      connection,
+      'Alice',
+      fetchReturning([{ name: 'Alice', runtime: 'pty', channels: [] }])
+    );
+    expect(identity).toBeNull();
+  });
+
+  it('returns null for an agent the broker does not list', async () => {
+    const identity = await fetchWorkerIdentity(connection, 'Ghost', fetchReturning([]));
+    expect(identity).toBeNull();
+  });
+});
diff --git a/packages/cli/src/cli/lib/attach-drive.ts b/packages/cli/src/cli/lib/attach-drive.ts
index f1071119a..733007515 100644
--- a/packages/cli/src/cli/lib/attach-drive.ts
+++ b/packages/cli/src/cli/lib/attach-drive.ts
@@ -412,17 +412,28 @@ export async function sendInput(
  * it cannot be established.
  *
  * The broker exposes no per-instance token for a worker — no `instance_id`,
- * `run_id`, `epoch`, or absolute spawn timestamp reaches the wire. The only
- * restart-discriminating value on `GET /api/spawned` is the harness `pid`
- * (`crates/broker/src/worker.rs:242`, an `Option<u32>` that is null until the
- * worker reports ready), so that is what this returns.
+ * `run_id`, `epoch`, or absolute spawn timestamp reaches the wire (#1454). The
+ * only restart-discriminating values on `GET /api/spawned` are two pids, and
+ * they are not interchangeable:
+ *
+ * - `workerPid` (`crates/broker/src/worker.rs:243` = `handle.child.id()`) is the
+ *   PTY child itself — the process whose terminal we are driving. Present as
+ *   soon as the worker is spawned.
+ * - `pid` (`worker.rs:242` = `handle.harness_pid`) is the *harness* wrapper, and
+ *   stays null until the worker completes the harness ready handshake
+ *   (`worker_events.rs:849`). Verified against a live broker: a plain PTY worker
+ *   reports `pid: null` and `workerPid: 30209`, so keying on `pid` alone would
+ *   make every reopen unverifiable for exactly the workers this path serves.
+ *
+ * So prefer `workerPid` and fold in `pid` when the broker also has it — a change
+ * in either means the process behind the name changed.
  *
  * This is a heuristic, not a nonce: the OS can reuse a pid. It is used to
  * *reject* a reopen that lands on a visibly different process, never to prove
  * two processes are the same — callers treat `null` as "cannot verify" and
  * fail closed. The durable fix is for the broker to surface the per-spawn
  * identity it already holds in memory (`WorkerHandle.spawned_at`,
- * `PersistedAgent.started_at`, or the `MetricsCollector` spawn counter).
+ * `PersistedAgent.started_at`, or the `MetricsCollector` spawn counter) — #1454.
  */
 export async function fetchWorkerIdentity(
   connection: BrokerConnection,
@@ -432,8 +443,15 @@ export async function fetchWorkerIdentity(
   try {
     const agents = await createBrokerClient(connection, fetchFn).listAgents();
     const agent = agents.find((candidate) => candidate.name === name);
-    if (!agent || typeof agent.pid !== 'number') return null;
-    return String(agent.pid);
+    if (!agent) return null;
+    // `workerPid` is on the wire but absent from the typed contract, so read it
+    // off the record defensively rather than widening `ListAgent` here.
+    const workerPid = (agent as { workerPid?: unknown }).workerPid;
+    const parts: string[] = [];
+    if (typeof workerPid === 'number') parts.push(`worker:${workerPid}`);
+    if (typeof agent.pid === 'number') parts.push(`harness:${agent.pid}`);
+    // No pid of either kind means the broker cannot tell us who this is.
+    return parts.length > 0 ? parts.join('/') : null;
   } catch {
     return null;
   }
@@ -996,7 +1014,7 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
           return { ok: false, reason: 'worker identity could not be read after reconnect' };
         }
         if (current !== attachedWorkerIdentity) {
-          return { ok: false, reason: `worker process changed (pid ${attachedWorkerIdentity} → ${current})` };
+          return { ok: false, reason: `worker process changed (${attachedWorkerIdentity} → ${current})` };
         }
         return { ok: true };
       },
diff --git a/packages/cli/src/cli/lib/attach-passthrough.ts b/packages/cli/src/cli/lib/attach-passthrough.ts
index 4b4460ad7..db8a15a0f 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.ts
@@ -588,7 +588,7 @@ export async function runPassthroughSession(
         if (current !== attachedWorkerIdentity) {
           return {
             ok: false,
-            reason: `worker process changed (pid ${attachedWorkerIdentity} → ${current})`,
+            reason: `worker process changed (${attachedWorkerIdentity} → ${current})`,
           };
         }
         return { ok: true };

From 367f3ad5d38d947050c63da58f70f4c1630aed05 Mon Sep 17 00:00:00 2001
From: Proactive Runtime Bot <agent@agent-relay.com>
Date: Fri, 7 Aug 2026 13:49:40 +0200
Subject: [PATCH 4/7] docs(cli): fix stale isDead reference after the isUsable
 rename

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 packages/cli/src/cli/lib/attach-input-recovery.ts | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
index 657bb02fd..56b8b32af 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -121,7 +121,7 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
   const recover = (reason: string): void => {
     if (isSettled() || inFlight) return;
 
-    // Drop the dead handle first: `isDead()` then short-circuits every chunk
+    // Drop the dead handle first: `isUsable()` then short-circuits every chunk
     // that arrives mid-recovery, which is what actually silences the flood.
     const dead = getStream();
     setStream(null);

From d2a2b91727fa3a91a0495112bb78a3654fc3dfe1 Mon Sep 17 00:00:00 2001
From: Proactive Runtime Bot <agent@agent-relay.com>
Date: Fri, 7 Aug 2026 14:31:18 +0200
Subject: [PATCH 5/7] fix(cli): close review gaps in attach input-stream
 recovery
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses the review findings on #1453. Two were defects this PR itself
introduced.

Identity gate is no longer optional (cubic P1). `verifyIdentity` was
optional, so a caller that omitted it reconnected the session to whichever
worker currently owned the agent name and forwarded the user's keystrokes
there — the exact reattach-by-name hole the gate exists to close. It is now
required by type and refused at runtime when absent.

Backpressure no longer tears down a healthy stream (codex P2, coderabbit
Major). `PtyInputStream.send()` rejects `input_backpressure` while the
socket is open and usable (transport.ts:206-214, retryable: true). The
unconditional catch treated that as transport loss, closing a healthy
socket, dropping outstanding input, and potentially detaching non-zero
because the broker was briefly slow. Send rejections now route through
`handleSendFailure`, which rolls back the echo and reports once per
episode for backpressure and only recovers on real loss.

Also fixed:

- A verifier that throws or stalls left the session with no input stream
  and no exhaustion path, so it hung instead of exiting non-zero (cubic
  P1). Both collapse into the existing `{ ok: false, reason }` refusal.
- `attemptReopen` now owns the replacement stream on every exit path
  (codex P1, coderabbit). The abandoned-attempt and settled-during-verify
  paths returned without closing it, leaking a live socket with no owner
  that could keep the CLI alive past a clean detach.
- `cancel()` resolves the pending backoff instead of only clearing the
  timer, so the loop unwinds and `inFlight` clears; it previously left the
  helper permanently `isRecovering()` and detach cleanup incomplete
  (cubic P2).
- Open and identity waits are bounded by `attemptTimeoutMs`, so a stalled
  socket or hung broker call cannot park the session in recovery forever
  and make `maxAttempts` a fiction (cubic P2).
- Ctrl+C sharing a stdin chunk with other bytes now still detaches during
  an outage in both modes; the dead-stream branch returned before the
  keybind actions ran, so "ab\x03" was swallowed and the human could not
  escape a broken session (cubic P2).
- CHANGELOG `[Unreleased - Patch]` entry for the user-visible attach
  behaviour change (codex P1).
- Detach-cancellation test now settles past the full 300ms backoff span
  rather than the first ~60ms, so a timer surviving teardown is actually
  caught (cubic P3).

New assertions, each red-checked against the vulnerable shape: a reopen
without a verifier is refused; backpressure does not tear down a healthy
stream or report more than once per episode; a throwing verifier exits
non-zero; a stalling verifier times out; abandoned and settled attempts
close their sockets; cancel ends the loop; Ctrl+C detaches mid-outage.

Live end-to-end re-verified after the change: real pty, real broker,
worker killed underneath, 200 mouse reports and zero keystrokes — one
line, exit 1.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |   7 +-
 packages/cli/src/cli/lib/attach-drive.test.ts | 128 ++++++++-
 packages/cli/src/cli/lib/attach-drive.ts      |  40 +--
 .../src/cli/lib/attach-input-recovery.test.ts | 269 ++++++++++++++++++
 .../cli/src/cli/lib/attach-input-recovery.ts  | 203 +++++++++++--
 .../src/cli/lib/attach-passthrough.test.ts    |  20 ++
 .../cli/src/cli/lib/attach-passthrough.ts     |  29 +-
 7 files changed, 635 insertions(+), 61 deletions(-)
 create mode 100644 packages/cli/src/cli/lib/attach-input-recovery.test.ts

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2a00cd606..5dae37c05 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,12 @@ All notable changes to Agent Relay will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [Unreleased - Patch]
+
+### Fixed
+
+- `agent-relay node agent attach --mode drive` (and `--mode passthrough`) no longer floods the terminal with `input stream send failed: PTY input stream is closed` when the PTY input stream dies mid-session. The loss is now reported once, the stream is reopened with bounded backoff, and if that fails the command exits non-zero with a readable message instead of leaving a session that looks alive but accepts no input. Because every byte except `Ctrl+C`/`Ctrl+]` is forwarded, a source TUI with mouse tracking enabled could produce this flood from pointer movement alone, without a single keystroke.
+- A reopened attach input stream is verified to belong to the same worker process before any keystroke is forwarded. The stream is reopened by agent name, so without this a replaced worker could silently receive input typed for the session you attached to; the check fails closed when identity cannot be established.
 
 ## [11.4.0] - 2026-08-02
 
diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts
index 2c260a9fa..f7173ffd3 100644
--- a/packages/cli/src/cli/lib/attach-drive.test.ts
+++ b/packages/cli/src/cli/lib/attach-drive.test.ts
@@ -232,6 +232,10 @@ interface FetchScript {
    * Defaults to a stable pid, i.e. "same worker throughout".
    */
   workerIdentities?: Array<string | null>;
+  /** Make `getWorkerIdentity` reject, simulating a verifier that cannot answer. */
+  identityError?: Error;
+  /** Park `getWorkerIdentity` on this promise so a detach can race verification. */
+  identityGate?: Promise<void>;
   /** Ownership re-assert interval (ms). Defaults to disabled (0) in tests so
    *  the keep-alive timer doesn't interfere; the re-assert test sets it small. */
   ownershipReassertMs?: number;
@@ -264,6 +268,7 @@ function createHarness(opts: FetchScript = {}): {
     headers: Record<string, string>;
   }> = [];
   const identityCalls: Array<string | null> = [];
+  const identityCallCount = { value: 0 };
   const stdin = new FakeStdin();
   const terminal = new FakeTerminal(
     opts.terminalSize === undefined ? { rows: 30, cols: 100 } : opts.terminalSize
@@ -466,11 +471,21 @@ function createHarness(opts: FetchScript = {}): {
     inputReopenMaxAttempts: opts.inputReopenMaxAttempts ?? 2,
     inputReopenBaseDelayMs: opts.inputReopenBaseDelayMs ?? 1,
     getWorkerIdentity: vi.fn(async () => {
+      // Count EVERY call, not just scripted ones: call 0 is the attach-time
+      // baseline and calls 1+ are post-reopen checks. Deriving the index from a
+      // list that some tests never populate left the counter stuck at 0, so the
+      // reopen branch below never fired and the session hung instead.
+      const index = identityCallCount.value;
+      identityCallCount.value += 1;
+      // The baseline must resolve normally; only post-reopen checks are made to
+      // fail or stall, since that is where the gate actually runs.
+      const isReopenCheck = index > 0;
+      if (isReopenCheck && opts.identityGate) await opts.identityGate;
+      if (isReopenCheck && opts.identityError) throw opts.identityError;
       const scripted = opts.workerIdentities;
       if (!scripted) return 'pid-1';
       // Index explicitly: a scripted `null` is a meaningful value ("identity
       // unavailable"), so `??` must not collapse it into the fallback.
-      const index = identityCalls.length;
       const value = index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
       identityCalls.push(value);
       return value;
@@ -2303,7 +2318,10 @@ describe('runDriveSession — lost PTY input stream', () => {
 
     expect(await sessionPromise).toBe(0);
     const before = inputStreams.length;
-    await settleRecovery();
+    // Must outlast the WHOLE backoff schedule (20+40+80+160 = 300ms here), not
+    // just the first delay: a timer that survives teardown and fires at 80ms or
+    // 160ms would otherwise land outside the window and go unnoticed.
+    await settleRecovery(400);
     // No further sockets opened after teardown, and no late error printed.
     expect(inputStreams).toHaveLength(before);
     expect(errors.map((a) => String(a[0])).filter((l) => l.includes('could not be reopened'))).toEqual([]);
@@ -2378,6 +2396,112 @@ describe('runDriveSession — lost PTY input stream', () => {
     expect(inputStreams[1].writes).toHaveLength(0);
   });
 
+  it('does NOT tear down a healthy stream when a send hits backpressure', async () => {
+    // Regression guard for the defect this PR introduced. `PtyInputStream.send()`
+    // rejects `input_backpressure` while the socket is open and usable
+    // (transport.ts:206-214, retryable: true). Treating that as stream loss
+    // closes a healthy socket, drops outstanding input, and can detach the
+    // session non-zero just because the broker was briefly slow.
+    // Fails if backpressure starts a recovery, opens a second stream, or ends
+    // the session.
+    const backpressure = Object.assign(
+      new Error('PTY input stream buffered 1048576 bytes; refusing 1 more over high water mark 1048576'),
+      { code: 'input_backpressure', retryable: true }
+    );
+    const { deps, sockets, stdin, logs, errors, inputStreams } = createHarness({
+      inputStreamSendError: backpressure,
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    for (let i = 0; i < 50; i++) stdin.type(Buffer.from(`k${i}`));
+    await settleRecovery();
+
+    const all = [...logs, ...errors].map((a) => String(a[0]));
+    expect(all.filter((l) => l.includes('input stream lost'))).toEqual([]);
+    expect(all.filter((l) => l.includes('could not be reopened'))).toEqual([]);
+    // No replacement stream: the original was never torn down.
+    expect(inputStreams).toHaveLength(1);
+    expect(inputStreams[0].closed).toBe(false);
+    // The user is told once, not fifty times.
+    expect(all.filter((l) => l.includes('faster than'))).toHaveLength(1);
+
+    // And the session is still alive and detachable.
+    stdin.type(Buffer.from([0x03]));
+    expect(await sessionPromise).toBe(0);
+  });
+
+  it('still detaches on Ctrl+C when it shares a chunk with input during an outage', async () => {
+    // The dead-stream branch used to `return` before the keybind actions ran,
+    // so a chunk like "ab\x03" was swallowed whole and the user could not
+    // escape a broken session. Fails if the session does not exit.
+    const { deps, sockets, stdin, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 4,
+      inputReopenBaseDelayMs: 50,
+      reopenOpenErrors: [new Error('x'), new Error('x'), new Error('x'), new Error('x')],
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    // Ordinary bytes AND the detach byte in one chunk — exactly what a paste or
+    // a fast typist produces.
+    stdin.type(Buffer.from([0x61, 0x62, 0x03]));
+
+    expect(await sessionPromise).toBe(0);
+  });
+
+  it('refuses the reopen when the identity verifier throws', async () => {
+    // A verifier that cannot answer has not said yes. Before this, the throw
+    // escaped the attempt loop, leaving the session with no input stream and no
+    // exhaustion path — it hung instead of exiting. Fails on a hang or exit 0.
+    const { deps, sockets, stdin, errors, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+      identityError: new Error('broker unreachable'),
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+
+    expect(await sessionPromise).toBe(1);
+    const refusal = errors.map((a) => String(a[0])).find((l) => l.includes('not the same worker'));
+    expect(refusal).toContain('identity check failed');
+    expect(refusal).toContain('broker unreachable');
+    expect(inputStreams[1].writes).toHaveLength(0);
+    expect(inputStreams[1].closed).toBe(true);
+  });
+
+  it('closes the replacement stream when the user detaches mid-verification', async () => {
+    // Teardown sees `inputStream` as null during recovery, so if the attempt
+    // does not close its own replacement the socket leaks with no owner and can
+    // keep the CLI alive past a clean detach. Fails if it is left open.
+    let releaseVerify: (() => void) | undefined;
+    const gate = new Promise<void>((resolve) => {
+      releaseVerify = resolve;
+    });
+    const { deps, sockets, stdin, inputStreams } = createHarness({
+      reopenOpenErrors: [undefined],
+      identityGate: gate,
+    });
+    const sessionPromise = runDriveSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from('a'));
+    await settleRecovery(30);
+    expect(inputStreams).toHaveLength(2); // replacement opened, verify parked
+
+    stdin.type(Buffer.from([0x03])); // detach while verification is pending
+    releaseVerify?.();
+
+    expect(await sessionPromise).toBe(0);
+    await settleRecovery(20);
+    expect(inputStreams[1].closed).toBe(true);
+    expect(inputStreams[1].writes).toHaveLength(0);
+  });
+
   it('rolls back predictive echo once for input that never reached the PTY', async () => {
     // Fails if the screen keeps optimistically-echoed glyphs for keystrokes the
     // agent never received — a silent lie about what the agent has seen.
diff --git a/packages/cli/src/cli/lib/attach-drive.ts b/packages/cli/src/cli/lib/attach-drive.ts
index 733007515..baaa3a296 100644
--- a/packages/cli/src/cli/lib/attach-drive.ts
+++ b/packages/cli/src/cli/lib/attach-drive.ts
@@ -1040,25 +1040,33 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies):
         // sends us. Input during the outage is dropped, not buffered: replaying
         // stale keystrokes into a recovered PTY would execute them out of
         // context, which is worse than losing them.
+        //
+        // Skip only the *forwarding*; fall through to the action loop below.
+        // Ctrl+C can share a chunk with ordinary bytes, and returning here
+        // would swallow the detach — leaving the human unable to escape a
+        // broken session, which is worse than the flood.
         if (!inputRecovery.isUsable(stream)) {
           inputRecovery.recover('stream closed');
-          return;
-        }
-        // Decode through the stateful UTF-8 decoder so a multi-byte character
-        // split across stdin chunks is forwarded intact rather than as U+FFFD.
-        // An incomplete trailing sequence decodes to '' and is held until the
-        // next chunk completes it.
-        const decoded = inputDecoder.write(outcome.forward);
-        if (decoded.length > 0) {
-          // Fire-and-forget; don't block the event loop on every keystroke.
-          void stream.send(decoded).catch((err: unknown) => {
-            if (settled) return;
-            // Only the first failure speaks; `recover` no-ops while a recovery
-            // is already in flight, and it does the echo rollback.
-            inputRecovery.recover(describeError(err));
-          });
+        } else {
+          // Decode through the stateful UTF-8 decoder so a multi-byte character
+          // split across stdin chunks is forwarded intact rather than as U+FFFD.
+          // An incomplete trailing sequence decodes to '' and is held until the
+          // next chunk completes it.
+          const decoded = inputDecoder.write(outcome.forward);
+          if (decoded.length > 0) {
+            // Fire-and-forget; don't block the event loop on every keystroke.
+            void stream.send(decoded).then(
+              () => inputRecovery.noteSendSuccess(),
+              (err: unknown) => {
+                if (settled) return;
+                // Classified, not assumed: backpressure leaves the stream
+                // healthy and must not trigger a teardown.
+                inputRecovery.handleSendFailure(err);
+              }
+            );
+          }
+          predictiveEcho?.onUserInput(outcome.forward);
         }
-        predictiveEcho?.onUserInput(outcome.forward);
       }
       for (const action of outcome.actions) {
         switch (action) {
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
new file mode 100644
index 000000000..a20b1b362
--- /dev/null
+++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
@@ -0,0 +1,269 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import {
+  createInputStreamRecovery,
+  isBackpressureRejection,
+  type InputStreamRecoveryOptions,
+} from './attach-input-recovery.js';
+import type { CliPtyInputStream } from './attach-drive.js';
+
+class FakeStream implements CliPtyInputStream {
+  closed = false;
+  closeReason?: string;
+  readonly writes: string[] = [];
+
+  constructor(private readonly openError?: Error) {}
+
+  async waitUntilOpen(): Promise<void> {
+    if (this.openError) throw this.openError;
+  }
+
+  async send(data: string): Promise<{ name: string; bytes_written: number }> {
+    if (this.closed) throw new Error('PTY input stream is closed');
+    this.writes.push(data);
+    return { name: 'agent', bytes_written: data.length };
+  }
+
+  close(_code?: number, reason?: string): void {
+    this.closed = true;
+    this.closeReason = reason;
+  }
+}
+
+/** Builds a recovery helper over controllable streams and records its output. */
+function harness(overrides: Partial<InputStreamRecoveryOptions> = {}) {
+  const logs: string[] = [];
+  const errors: string[] = [];
+  const opened: FakeStream[] = [];
+  let current: CliPtyInputStream | null = new FakeStream();
+  const first = current as FakeStream;
+  let settled = false;
+  let exhausted = 0;
+  let rollbacks = 0;
+
+  const recovery = createInputStreamRecovery({
+    label: 'drive',
+    name: 'Alice',
+    maxAttempts: 2,
+    baseDelayMs: 1,
+    attemptTimeoutMs: 50,
+    log: (m) => logs.push(m),
+    error: (m) => errors.push(m),
+    isSettled: () => settled,
+    getStream: () => current,
+    setStream: (s) => {
+      current = s;
+    },
+    openStream: () => {
+      const s = new FakeStream();
+      opened.push(s);
+      return s;
+    },
+    onRollback: () => {
+      rollbacks += 1;
+    },
+    onExhausted: () => {
+      exhausted += 1;
+    },
+    verifyIdentity: async () => ({ ok: true }),
+    ...overrides,
+  });
+
+  return {
+    recovery,
+    logs,
+    errors,
+    opened,
+    first,
+    getCurrent: () => current,
+    settle: () => {
+      settled = true;
+    },
+    counts: () => ({ exhausted, rollbacks }),
+  };
+}
+
+const settle = async (turns = 80): Promise<void> => {
+  for (let i = 0; i < turns; i++) await new Promise((r) => setTimeout(r, 1));
+};
+
+describe('isBackpressureRejection', () => {
+  it('recognises the transport code and nothing else', () => {
+    expect(isBackpressureRejection(Object.assign(new Error('x'), { code: 'input_backpressure' }))).toBe(true);
+    expect(isBackpressureRejection(Object.assign(new Error('x'), { code: 'input_stream_closed' }))).toBe(false);
+    expect(isBackpressureRejection(new Error('write EPIPE'))).toBe(false);
+    expect(isBackpressureRejection(null)).toBe(false);
+    expect(isBackpressureRejection('input_backpressure')).toBe(false);
+  });
+});
+
+describe('handleSendFailure', () => {
+  it('does not start recovery for backpressure, and reports it once per episode', () => {
+    // Backpressure leaves the socket open and usable; recovering would close a
+    // healthy stream and drop outstanding input. Fails if the stream is torn
+    // down, or if fifty rejections produce fifty lines.
+    const h = harness();
+    const backpressure = Object.assign(new Error('over high water mark'), {
+      code: 'input_backpressure',
+    });
+
+    for (let i = 0; i < 50; i++) h.recovery.handleSendFailure(backpressure);
+
+    expect(h.recovery.isRecovering()).toBe(false);
+    expect(h.getCurrent()).toBe(h.first);
+    expect(h.first.closed).toBe(false);
+    expect(h.logs.filter((l) => l.includes('faster than'))).toHaveLength(1);
+    // Every dropped keystroke still has its optimistic echo rolled back.
+    expect(h.counts().rollbacks).toBe(50);
+  });
+
+  it('reports a second episode after the stream recovers', () => {
+    const h = harness();
+    const backpressure = Object.assign(new Error('over'), { code: 'input_backpressure' });
+    h.recovery.handleSendFailure(backpressure);
+    h.recovery.noteSendSuccess();
+    h.recovery.handleSendFailure(backpressure);
+    expect(h.logs.filter((l) => l.includes('faster than'))).toHaveLength(2);
+  });
+
+  it('treats every other rejection as stream loss', () => {
+    const h = harness();
+    h.recovery.handleSendFailure(
+      Object.assign(new Error('PTY input stream is closed'), { code: 'input_stream_closed' })
+    );
+    expect(h.recovery.isRecovering()).toBe(true);
+    expect(h.logs.some((l) => l.includes('input stream lost'))).toBe(true);
+  });
+});
+
+describe('identity verification is mandatory', () => {
+  it('refuses the reopen when no verifier is supplied', async () => {
+    // The type requires it; this proves the runtime refuses too, so a JS
+    // caller or a partial double cannot reach the unguarded path. Fails if the
+    // replacement is adopted — that is the keystroke-misrouting hole.
+    const h = harness({ verifyIdentity: undefined as unknown as InputStreamRecoveryOptions['verifyIdentity'] });
+
+    h.recovery.recover('stream closed');
+    await settle();
+
+    expect(h.getCurrent()).toBeNull();
+    expect(h.counts().exhausted).toBe(1);
+    expect(h.errors.some((e) => e.includes('no worker identity verifier was supplied'))).toBe(true);
+    // The socket it opened was closed rather than left live and unowned.
+    expect(h.opened).toHaveLength(1);
+    expect(h.opened[0].closed).toBe(true);
+  });
+
+  it('refuses when the verifier throws, instead of stranding the session', async () => {
+    const h = harness({
+      verifyIdentity: async () => {
+        throw new Error('broker unreachable');
+      },
+    });
+
+    h.recovery.recover('stream closed');
+    await settle();
+
+    expect(h.getCurrent()).toBeNull();
+    expect(h.counts().exhausted).toBe(1);
+    expect(h.errors.some((e) => e.includes('identity check failed: broker unreachable'))).toBe(true);
+  });
+
+  it('refuses when the verifier stalls past the attempt timeout', async () => {
+    // Without a deadline a hung broker call parks the session in recovery
+    // forever, so neither the attempt count nor the non-zero exit is bounded.
+    const h = harness({
+      attemptTimeoutMs: 20,
+      verifyIdentity: () => new Promise(() => {}),
+    });
+
+    h.recovery.recover('stream closed');
+    await settle(120);
+
+    expect(h.counts().exhausted).toBe(1);
+    expect(h.errors.some((e) => e.includes('timed out'))).toBe(true);
+  });
+});
+
+describe('stream ownership on every exit path', () => {
+  it('closes the replacement when the open attempt fails', async () => {
+    const failing: FakeStream[] = [];
+    const h = harness({
+      openStream: () => {
+        const s = new FakeStream(new Error('refused'));
+        failing.push(s);
+        return s;
+      },
+    });
+
+    h.recovery.recover('stream closed');
+    await settle();
+
+    // Two attempts, both abandoned — neither socket may be left unowned.
+    expect(failing).toHaveLength(2);
+    expect(failing.every((s) => s.closed)).toBe(true);
+    expect(h.counts().exhausted).toBe(1);
+  });
+
+  it('closes the replacement when the session settles during verification', async () => {
+    let release: (() => void) | undefined;
+    const gate = new Promise<void>((r) => {
+      release = r;
+    });
+    const h = harness({
+      verifyIdentity: async () => {
+        await gate;
+        return { ok: true };
+      },
+    });
+
+    h.recovery.recover('stream closed');
+    await settle(20);
+    expect(h.opened).toHaveLength(1);
+
+    h.settle(); // user detaches mid-verification
+    release?.();
+    await settle(20);
+
+    // Teardown can't see this stream (it was never handed over), so the attempt
+    // itself has to close it or it leaks a live socket.
+    expect(h.opened[0].closed).toBe(true);
+    expect(h.getCurrent()).toBeNull();
+  });
+});
+
+describe('cancel', () => {
+  it('ends the recovery loop instead of leaving it permanently recovering', async () => {
+    // Clearing the timer alone left the backoff promise unresolved forever, so
+    // `inFlight` never cleared and detach cleanup could not complete.
+    const h = harness({ baseDelayMs: 10_000 });
+
+    h.recovery.recover('stream closed');
+    expect(h.recovery.isRecovering()).toBe(true);
+
+    h.settle();
+    h.recovery.cancel();
+    await settle(20);
+
+    expect(h.recovery.isRecovering()).toBe(false);
+    // Cancellation is not a failure: no exhaustion exit, no new socket.
+    expect(h.counts().exhausted).toBe(0);
+    expect(h.opened).toHaveLength(0);
+  });
+});
+
+describe('recover', () => {
+  it('adopts a verified replacement and announces it once', async () => {
+    const verify = vi.fn(async () => ({ ok: true as const }));
+    const h = harness({ verifyIdentity: verify });
+
+    h.recovery.recover('stream closed');
+    await settle();
+
+    expect(h.getCurrent()).toBe(h.opened[0]);
+    expect(h.opened[0].closed).toBe(false);
+    expect(verify).toHaveBeenCalledTimes(1);
+    expect(h.logs.filter((l) => l.includes('reconnected'))).toHaveLength(1);
+    expect(h.first.closed).toBe(true); // the dead handle was released
+  });
+});
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
index 56b8b32af..a32697f55 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -32,6 +32,24 @@ export const INPUT_REOPEN_MAX_ATTEMPTS = 5;
 export const INPUT_REOPEN_BASE_DELAY_MS = 250;
 /** Ceiling on the doubled reopen delay. */
 export const INPUT_REOPEN_MAX_DELAY_MS = 4_000;
+/** Ceiling on one attempt's open wait and identity check. */
+export const INPUT_REOPEN_ATTEMPT_TIMEOUT_MS = 15_000;
+
+/**
+ * `PtyInputStream.send()` rejects with `input_backpressure` when queued bytes
+ * would exceed its high-water mark — **while the stream is open and healthy**
+ * (`transport.ts:206-214`, `retryable: true`). That is flow control, not
+ * transport death: tearing the socket down and reopening it would drop every
+ * outstanding keystroke and re-run the identity gate, turning a slow broker
+ * into a detach. Every other rejection means the stream is gone or unusable.
+ */
+export function isBackpressureRejection(error: unknown): boolean {
+  return (
+    typeof error === 'object' &&
+    error !== null &&
+    (error as { code?: unknown }).code === 'input_backpressure'
+  );
+}
 
 export interface InputStreamRecoveryOptions {
   /** Log prefix tag — `drive` or `passthrough`. */
@@ -56,20 +74,36 @@ export interface InputStreamRecoveryOptions {
   onExhausted: () => void;
   /**
    * Proves the reopened stream reached the SAME worker process the session
-   * originally attached to, and is called after every successful reopen.
+   * originally attached to. Called after every successful reopen, before a
+   * single byte is forwarded.
    *
-   * This gate exists because the input stream is reopened *by agent name*, and
-   * a name is not an identity: if the worker died and something else claimed
-   * the name, a "successful" reopen would quietly route the human's keystrokes
-   * into a different agent's PTY. Restarting the same agent is equally wrong
-   * for input safety — keystrokes typed for the old session's context would
-   * land in a fresh shell.
+   * **Required, deliberately.** The input stream is reopened *by agent name*,
+   * and a name is not an identity: if the worker died and something else
+   * claimed the name, a "successful" reopen would quietly route the human's
+   * keystrokes into a different agent's PTY. Restarting the same agent is
+   * equally wrong for input safety — keystrokes typed for the old session's
+   * context would land in a fresh shell. An optional verifier is not a gate,
+   * because the unsafe path is then reachable by omission.
    *
    * Must fail closed: anything other than a positive match — identity
    * unavailable, unreadable, or changed — has to return `ok: false` so the
-   * session exits loudly instead of reattaching on a guess.
+   * session exits loudly instead of reattaching on a guess. Throwing is also
+   * treated as a refusal; a verifier that cannot answer has not said yes.
+   */
+  verifyIdentity: () => Promise<{ ok: true } | { ok: false; reason: string }>;
+  /**
+   * Ceiling on a single reopen attempt's open and identity-verify waits.
+   * Without it a stalled socket or a hung broker call parks the session in
+   * recovery forever, so neither the attempt count nor the non-zero exhaustion
+   * exit is actually bounded. Defaults to
+   * {@link INPUT_REOPEN_ATTEMPT_TIMEOUT_MS}.
    */
-  verifyIdentity?: () => Promise<{ ok: true } | { ok: false; reason: string }>;
+  attemptTimeoutMs?: number;
+}
+
+function describeSendError(error: unknown): string {
+  if (error instanceof Error) return error.message;
+  return String(error);
 }
 
 export interface InputStreamRecovery {
@@ -84,6 +118,15 @@ export interface InputStreamRecovery {
   isRecovering(): boolean;
   /** Begin recovery. No-ops if already recovering or settled. */
   recover(reason: string): void;
+  /**
+   * Classify a rejected `send()`. Backpressure is flow control on a healthy
+   * stream and only costs the optimistic echo; everything else is treated as
+   * transport loss and enters recovery. Callers route every send rejection
+   * here rather than assuming loss.
+   */
+  handleSendFailure(error: unknown): void;
+  /** Clears the backpressure latch so a later episode reports again. */
+  noteSendSuccess(): void;
   /** Cancel a pending backoff timer (detach mid-recovery). */
   cancel(): void;
 }
@@ -105,19 +148,88 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
     verifyIdentity,
   } = options;
 
+  const attemptTimeoutMs = options.attemptTimeoutMs ?? INPUT_REOPEN_ATTEMPT_TIMEOUT_MS;
+
   let inFlight: Promise<void> | null = null;
   let timer: ReturnType<typeof setTimeout> | null = null;
+  /**
+   * Resolves the pending backoff wait on cancel. Clearing the timer alone left
+   * that promise permanently unresolved, so the recovery loop never finished,
+   * `inFlight` never cleared, and `isRecovering()` stayed true forever — detach
+   * cleanup could not complete.
+   */
+  let releaseBackoff: (() => void) | null = null;
+  /** True once a backpressure episode has been reported; reset on the next good send. */
+  let backpressureReported = false;
 
   const cancel = (): void => {
     if (timer) {
       clearTimeout(timer);
       timer = null;
     }
+    // Wake the loop so it observes `isSettled()` and unwinds, rather than
+    // parking on a timer that will never fire.
+    releaseBackoff?.();
+    releaseBackoff = null;
+  };
+
+  /**
+   * Rejects with a timeout rather than waiting forever. `waitUntilOpen()` and
+   * the identity check both cross the network; neither is bounded by this
+   * helper otherwise, and an unbounded await makes `maxAttempts` a fiction.
+   */
+  const withDeadline = async <T>(work: Promise<T>, what: string): Promise<T> => {
+    let handle: ReturnType<typeof setTimeout> | undefined;
+    try {
+      return await Promise.race([
+        work,
+        new Promise<never>((_resolve, reject) => {
+          handle = setTimeout(
+            () => reject(new Error(`${what} timed out after ${attemptTimeoutMs}ms`)),
+            attemptTimeoutMs
+          );
+          handle.unref?.();
+        }),
+      ]);
+    } finally {
+      if (handle !== undefined) clearTimeout(handle);
+    }
   };
 
   const isUsable = (stream: CliPtyInputStream | null): stream is CliPtyInputStream =>
     stream !== null && stream.closed !== true;
 
+  const closeQuietly = (stream: CliPtyInputStream, why: string): void => {
+    try {
+      stream.close(1000, why);
+    } catch {
+      // best effort
+    }
+  };
+
+  const noteSendSuccess = (): void => {
+    backpressureReported = false;
+  };
+
+  const handleSendFailure = (sendError: unknown): void => {
+    if (isBackpressureRejection(sendError)) {
+      // Flow control on a healthy stream. The keystroke did not reach the PTY,
+      // so the optimistic echo still has to come off the screen — but the
+      // socket is fine and must not be torn down. Report once per episode:
+      // per-keystroke reporting here would rebuild the exact flood this module
+      // exists to remove.
+      onRollback();
+      if (!backpressureReported) {
+        backpressureReported = true;
+        log(
+          `[${label}] input is arriving faster than ${name} can accept it; dropping keystrokes until it catches up.`
+        );
+      }
+      return;
+    }
+    recover(describeSendError(sendError));
+  };
+
   const recover = (reason: string): void => {
     if (isSettled() || inFlight) return;
 
@@ -144,30 +256,28 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
     // One line for the outage, not one per keystroke.
     log(`[${label}] input stream lost (${reason}); reconnecting…`);
 
-    const closeQuietly = (stream: CliPtyInputStream, why: string): void => {
-      try {
-        stream.close(1000, why);
-      } catch {
-        // best effort
-      }
-    };
-
     /**
      * One reopen attempt. `'settled'` means the session went away mid-attempt,
      * `'retry'` a transport failure worth another go, and `'rejected'` a
      * replacement that opened but could not be vouched for — which must not be
      * retried, because a replaced worker does not become the original one on a
      * later attempt.
+     *
+     * `attemptReopen` owns `replacement` on **every** exit path. `openStream()`
+     * creates the socket eagerly, so any return that does not hand it to
+     * `setStream` must close it or leak a live WebSocket with no owner — which
+     * can also keep the process alive after a clean detach.
      */
     const attemptReopen = async (attempt: number): Promise<'opened' | 'retry' | 'rejected' | 'settled'> => {
-      let replacement: CliPtyInputStream;
+      let replacement: CliPtyInputStream | null = null;
       try {
         replacement = openStream();
-        await replacement.waitUntilOpen();
+        await withDeadline(replacement.waitUntilOpen(), 'input stream open');
       } catch {
         // Stay quiet between attempts. The human saw one line when the outage
         // started and sees exactly one more when it resolves either way;
         // narrating each failed retry would rebuild the flood.
+        if (replacement) closeQuietly(replacement, `${label} client abandoning attempt`);
         return 'retry';
       }
       if (isSettled()) {
@@ -177,18 +287,45 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
 
       // The socket is open, but "open" only proves the name resolved. Do not
       // hand the human's keystrokes to it until it is the same worker.
-      if (verifyIdentity) {
-        const verdict = await verifyIdentity();
-        if (isSettled()) return 'settled';
-        if (!verdict.ok) {
-          closeQuietly(replacement, `${label} client rejected replacement`);
-          error(
-            `[${label}] input stream reopened but it is not the same worker (${verdict.reason}). ` +
-              `Refusing to forward input — your keystrokes would go somewhere you did not attach to. ` +
-              `Detaching; reattach to ${name} to continue.`
-          );
-          return 'rejected';
-        }
+      //
+      // The type makes `verifyIdentity` mandatory; this enforces it at runtime
+      // too. A JS caller, a partial test double, or a future refactor that
+      // drops the field must not silently reach the unguarded path — the whole
+      // point is that reattaching by name is unsafe without a check.
+      if (typeof verifyIdentity !== 'function') {
+        closeQuietly(replacement, `${label} client rejected replacement`);
+        error(
+          `[${label}] input stream reopened but no worker identity verifier was supplied, ` +
+            `so it cannot be shown to be the same worker. Refusing to forward input. ` +
+            `Detaching; reattach to ${name} to continue.`
+        );
+        return 'rejected';
+      }
+
+      // A verifier that throws or stalls has not said yes, so both collapse
+      // into the same refusal rather than escaping and stranding the session
+      // with no stream and no exit.
+      const verdict = await withDeadline(verifyIdentity(), 'worker identity check').catch(
+        (verifyError: unknown) => ({
+          ok: false as const,
+          reason:
+            verifyError instanceof Error
+              ? `identity check failed: ${verifyError.message}`
+              : `identity check failed: ${String(verifyError)}`,
+        })
+      );
+      if (isSettled()) {
+        closeQuietly(replacement, `${label} client exiting`);
+        return 'settled';
+      }
+      if (!verdict.ok) {
+        closeQuietly(replacement, `${label} client rejected replacement`);
+        error(
+          `[${label}] input stream reopened but it is not the same worker (${verdict.reason}). ` +
+            `Refusing to forward input — your keystrokes would go somewhere you did not attach to. ` +
+            `Detaching; reattach to ${name} to continue.`
+        );
+        return 'rejected';
       }
 
       setStream(replacement);
@@ -201,10 +338,12 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
         if (isSettled()) return;
         const delay = Math.min(baseDelayMs * 2 ** (attempt - 1), INPUT_REOPEN_MAX_DELAY_MS);
         await new Promise<void>((resolve) => {
+          releaseBackoff = resolve;
           timer = setTimeout(resolve, delay);
           timer.unref?.();
         });
         timer = null;
+        releaseBackoff = null;
         if (isSettled()) return;
 
         const outcome = await attemptReopen(attempt);
@@ -229,6 +368,8 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
     isUsable,
     isRecovering: () => inFlight !== null,
     recover,
+    handleSendFailure,
+    noteSendSuccess,
     cancel,
   };
 }
diff --git a/packages/cli/src/cli/lib/attach-passthrough.test.ts b/packages/cli/src/cli/lib/attach-passthrough.test.ts
index a6b842765..49700ec1b 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.test.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.test.ts
@@ -1286,3 +1286,23 @@ describe('runPassthroughSession — lost PTY input stream', () => {
     expect(exhausted).toContain('Alice is still running');
   });
 });
+
+describe('runPassthroughSession — Ctrl+C during an input-stream outage', () => {
+  it('still detaches when Ctrl+C shares a chunk with other input', async () => {
+    // The dead-stream branch used to `return` before the keybind actions ran,
+    // so a chunk like "ab\x03" was swallowed whole and the user could not
+    // escape a broken session. Fails if the session never exits.
+    const { deps, sockets, stdin, inputStreams } = createHarness({
+      inputReopenMaxAttempts: 4,
+      inputReopenBaseDelayMs: 50,
+      reopenOpenErrors: [new Error('x'), new Error('x'), new Error('x'), new Error('x')],
+    });
+    const sessionPromise = runPassthroughSession('Alice', {}, deps);
+    await openSocket(sockets);
+
+    inputStreams[0].killFromServer();
+    stdin.type(Buffer.from([0x61, 0x62, 0x03]));
+
+    expect(await sessionPromise).toBe(0);
+  });
+});
diff --git a/packages/cli/src/cli/lib/attach-passthrough.ts b/packages/cli/src/cli/lib/attach-passthrough.ts
index db8a15a0f..7ec4dcb1f 100644
--- a/packages/cli/src/cli/lib/attach-passthrough.ts
+++ b/packages/cli/src/cli/lib/attach-passthrough.ts
@@ -611,20 +611,27 @@ export async function runPassthroughSession(
         // A dead or missing stream is a liveness event, not a per-keystroke
         // error — see `attach-input-recovery.ts` (#1419). Drop this input
         // silently; recovery has already announced itself once.
+        //
+        // Skip only the *forwarding* and fall through to the action loop:
+        // Ctrl+C can share a chunk with ordinary bytes, and returning here
+        // would swallow the detach mid-outage.
         if (!inputRecovery.isUsable(stream)) {
           inputRecovery.recover('stream closed');
-          return;
-        }
-        // Decode through the stateful UTF-8 decoder so a multi-byte character
-        // split across stdin chunks is forwarded intact rather than as U+FFFD.
-        const decoded = inputDecoder.write(outcome.forward);
-        if (decoded.length > 0) {
-          void stream.send(decoded).catch((err: unknown) => {
-            if (settled) return;
-            inputRecovery.recover(describeError(err));
-          });
+        } else {
+          // Decode through the stateful UTF-8 decoder so a multi-byte character
+          // split across stdin chunks is forwarded intact rather than as U+FFFD.
+          const decoded = inputDecoder.write(outcome.forward);
+          if (decoded.length > 0) {
+            void stream.send(decoded).then(
+              () => inputRecovery.noteSendSuccess(),
+              (err: unknown) => {
+                if (settled) return;
+                inputRecovery.handleSendFailure(err);
+              }
+            );
+          }
+          predictiveEcho?.onUserInput(outcome.forward);
         }
-        predictiveEcho?.onUserInput(outcome.forward);
       }
       for (const action of outcome.actions) {
         switch (action) {

From eae3f9c74dc70001448221a113cdb9cf2d285c01 Mon Sep 17 00:00:00 2001
From: "github-actions[bot]" <github-actions[bot]@users.noreply.github.com>
Date: Fri, 7 Aug 2026 12:32:30 +0000
Subject: [PATCH 6/7] style: auto-format with Prettier

---
 packages/cli/src/cli/lib/attach-input-recovery.test.ts | 8 ++++++--
 packages/cli/src/cli/lib/attach-input-recovery.ts      | 4 +---
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
index a20b1b362..c8662cbb7 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.test.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
@@ -90,7 +90,9 @@ const settle = async (turns = 80): Promise<void> => {
 describe('isBackpressureRejection', () => {
   it('recognises the transport code and nothing else', () => {
     expect(isBackpressureRejection(Object.assign(new Error('x'), { code: 'input_backpressure' }))).toBe(true);
-    expect(isBackpressureRejection(Object.assign(new Error('x'), { code: 'input_stream_closed' }))).toBe(false);
+    expect(isBackpressureRejection(Object.assign(new Error('x'), { code: 'input_stream_closed' }))).toBe(
+      false
+    );
     expect(isBackpressureRejection(new Error('write EPIPE'))).toBe(false);
     expect(isBackpressureRejection(null)).toBe(false);
     expect(isBackpressureRejection('input_backpressure')).toBe(false);
@@ -141,7 +143,9 @@ describe('identity verification is mandatory', () => {
     // The type requires it; this proves the runtime refuses too, so a JS
     // caller or a partial double cannot reach the unguarded path. Fails if the
     // replacement is adopted — that is the keystroke-misrouting hole.
-    const h = harness({ verifyIdentity: undefined as unknown as InputStreamRecoveryOptions['verifyIdentity'] });
+    const h = harness({
+      verifyIdentity: undefined as unknown as InputStreamRecoveryOptions['verifyIdentity'],
+    });
 
     h.recovery.recover('stream closed');
     await settle();
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
index a32697f55..f9993e08d 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -45,9 +45,7 @@ export const INPUT_REOPEN_ATTEMPT_TIMEOUT_MS = 15_000;
  */
 export function isBackpressureRejection(error: unknown): boolean {
   return (
-    typeof error === 'object' &&
-    error !== null &&
-    (error as { code?: unknown }).code === 'input_backpressure'
+    typeof error === 'object' && error !== null && (error as { code?: unknown }).code === 'input_backpressure'
   );
 }
 

From 36aa5fa98762ffa4e58d31d8647da1cc38959644 Mon Sep 17 00:00:00 2001
From: Proactive Runtime Bot <agent@agent-relay.com>
Date: Fri, 7 Aug 2026 15:57:05 +0200
Subject: [PATCH 7/7] fix(cli): route a synchronously throwing identity
 verifier through the refusal path
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Second round of review findings on #1453. Three accepted, one rejected.

A verifier that throws synchronously still stranded the session. The type
permits a non-`async` function, and `verifyIdentity()` was evaluated in
argument position, so the throw escaped before `.catch()` was attached —
skipping the refusal, the non-zero exhaustion exit, and the close of the
replacement stream. Same failure the async-throw fix closed, reachable by
a different route. It is now invoked inside the promise chain.

Also:

- CHANGELOG: qualify the byte-forwarding claim. It described why the flood
  happened but read as a present-tense statement, which the recovery path
  now contradicts by dropping input while the stream is down.
- Drop the dead `identityCalls` array from the drive test harness. The
  earlier fix replaced it as the index source with `identityCallCount`,
  leaving it written but never read.

Rejected, with reasoning on the thread: CodeRabbit asked to lower the
changelog heading from `[Unreleased - Patch]` to `[Unreleased]`. AGENTS.md
requires the opposite — the first pending user-visible change must set a
release level, and the level is monotonic and must never be lowered. That
heading is there because an earlier codex P1 required it, citing the same
section.

New assertion, red-checked against argument-position invocation: "refuses
when the verifier throws synchronously" asserts the exhaustion exit runs
and the replacement socket is closed.

Prettier run over the changed files this time, so CI's formatting check
passes without a bot follow-up commit.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 CHANGELOG.md                                  |  2 +-
 packages/cli/src/cli/lib/attach-drive.test.ts |  2 --
 .../src/cli/lib/attach-input-recovery.test.ts | 22 ++++++++++++++++
 .../cli/src/cli/lib/attach-input-recovery.ts  | 25 ++++++++++++-------
 4 files changed, 39 insertions(+), 12 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5dae37c05..e908c748b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
-- `agent-relay node agent attach --mode drive` (and `--mode passthrough`) no longer floods the terminal with `input stream send failed: PTY input stream is closed` when the PTY input stream dies mid-session. The loss is now reported once, the stream is reopened with bounded backoff, and if that fails the command exits non-zero with a readable message instead of leaving a session that looks alive but accepts no input. Because every byte except `Ctrl+C`/`Ctrl+]` is forwarded, a source TUI with mouse tracking enabled could produce this flood from pointer movement alone, without a single keystroke.
+- `agent-relay node agent attach --mode drive` (and `--mode passthrough`) no longer floods the terminal with `input stream send failed: PTY input stream is closed` when the PTY input stream dies mid-session. The loss is now reported once, the stream is reopened with bounded backoff, and if that fails the command exits non-zero with a readable message instead of leaving a session that looks alive but accepts no input. Because attach forwards every byte except `Ctrl+C`/`Ctrl+]` while the stream is healthy, a source TUI with mouse tracking enabled could previously produce this flood from pointer movement alone, without a single keystroke; input is now dropped rather than forwarded for as long as the stream is down.
 - A reopened attach input stream is verified to belong to the same worker process before any keystroke is forwarded. The stream is reopened by agent name, so without this a replaced worker could silently receive input typed for the session you attached to; the check fails closed when identity cannot be established.
 
 ## [11.4.0] - 2026-08-02
diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts
index f7173ffd3..c00c1870f 100644
--- a/packages/cli/src/cli/lib/attach-drive.test.ts
+++ b/packages/cli/src/cli/lib/attach-drive.test.ts
@@ -267,7 +267,6 @@ function createHarness(opts: FetchScript = {}): {
     body?: unknown;
     headers: Record<string, string>;
   }> = [];
-  const identityCalls: Array<string | null> = [];
   const identityCallCount = { value: 0 };
   const stdin = new FakeStdin();
   const terminal = new FakeTerminal(
@@ -487,7 +486,6 @@ function createHarness(opts: FetchScript = {}): {
       // Index explicitly: a scripted `null` is a meaningful value ("identity
       // unavailable"), so `??` must not collapse it into the fallback.
       const value = index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
-      identityCalls.push(value);
       return value;
     }),
   };
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
index c8662cbb7..77ba28767 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.test.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
@@ -173,6 +173,28 @@ describe('identity verification is mandatory', () => {
     expect(h.errors.some((e) => e.includes('identity check failed: broker unreachable'))).toBe(true);
   });
 
+  it('refuses when the verifier throws synchronously', async () => {
+    // The type allows a non-async function. A synchronous throw evaluated in
+    // the argument position escaped before `.catch()` was attached, so the
+    // session was stranded with no stream, no exhaustion exit, and the
+    // replacement left open — the exact failure the async-throw fix closed,
+    // reachable by a different route. Fails if `onExhausted` is skipped.
+    const h = harness({
+      verifyIdentity: (() => {
+        throw new Error('verifier blew up synchronously');
+      }) as unknown as InputStreamRecoveryOptions['verifyIdentity'],
+    });
+
+    h.recovery.recover('stream closed');
+    await settle();
+
+    expect(h.counts().exhausted).toBe(1);
+    expect(h.getCurrent()).toBeNull();
+    expect(h.errors.some((e) => e.includes('verifier blew up synchronously'))).toBe(true);
+    // And the socket it opened was closed, not leaked.
+    expect(h.opened[0].closed).toBe(true);
+  });
+
   it('refuses when the verifier stalls past the attempt timeout', async () => {
     // Without a deadline a hung broker call parks the session in recovery
     // forever, so neither the attempt count nor the non-zero exit is bounded.
diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts
index f9993e08d..0097d1470 100644
--- a/packages/cli/src/cli/lib/attach-input-recovery.ts
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -303,15 +303,22 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
       // A verifier that throws or stalls has not said yes, so both collapse
       // into the same refusal rather than escaping and stranding the session
       // with no stream and no exit.
-      const verdict = await withDeadline(verifyIdentity(), 'worker identity check').catch(
-        (verifyError: unknown) => ({
-          ok: false as const,
-          reason:
-            verifyError instanceof Error
-              ? `identity check failed: ${verifyError.message}`
-              : `identity check failed: ${String(verifyError)}`,
-        })
-      );
+      //
+      // `verifyIdentity()` is invoked inside the promise chain, not as an
+      // argument to it: the type permits a non-`async` function, and a
+      // synchronous throw evaluated in the argument position would escape
+      // before `.catch()` was ever attached — skipping the refusal, the
+      // exhaustion exit, and the close below.
+      const verdict = await withDeadline(
+        Promise.resolve().then(() => verifyIdentity()),
+        'worker identity check'
+      ).catch((verifyError: unknown) => ({
+        ok: false as const,
+        reason:
+          verifyError instanceof Error
+            ? `identity check failed: ${verifyError.message}`
+            : `identity check failed: ${String(verifyError)}`,
+      }));
       if (isSettled()) {
         closeQuietly(replacement, `${label} client exiting`);
         return 'settled';