diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2a00cd606..e908c748b 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 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 634c89185..c00c1870f 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,
@@ -152,6 +153,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 +166,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 +216,26 @@ 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>;
+  /** 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;
@@ -234,6 +267,7 @@ function createHarness(opts: FetchScript = {}): {
     body?: unknown;
     headers: Record<string, string>;
   }> = [];
+  const identityCallCount = { value: 0 };
   const stdin = new FakeStdin();
   const terminal = new FakeTerminal(
     opts.terminalSize === undefined ? { rows: 30, cols: 100 } : opts.terminalSize
@@ -414,7 +448,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 +465,29 @@ 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 () => {
+      // 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 value = index < scripted.length ? scripted[index] : (scripted[scripted.length - 1] ?? null);
+      return value;
+    }),
   };
 
   return {
@@ -2112,3 +2176,402 @@ 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;
+    // 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([]);
+  });
+
+  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('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.
+    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;
+  });
+});
+
+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 1be5ffa27..baaa3a296 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,56 @@ 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 (#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) — #1454.
+ */
+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) 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;
+  }
+}
+
 /** Open the SDK-backed raw PTY input stream for interactive CLI sessions. */
 export function openPtyInputStream(
   connection: BrokerConnection,
@@ -899,6 +980,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 (${attachedWorkerIdentity} → ${current})` };
+        }
+        return { ok: true };
+      },
+    });
+
     // ---- stdin handling ----
     let stdinReady = false;
     const stdinDataHandler = (chunk: Buffer): void => {
@@ -913,28 +1034,39 @@ 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');
-          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; surface errors via log but 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();
-          });
+        // 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.
+        //
+        // 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');
+        } 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) {
@@ -998,6 +1130,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 +1251,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.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
new file mode 100644
index 000000000..77ba28767
--- /dev/null
+++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts
@@ -0,0 +1,295 @@
+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 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.
+    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
new file mode 100644
index 000000000..0097d1470
--- /dev/null
+++ b/packages/cli/src/cli/lib/attach-input-recovery.ts
@@ -0,0 +1,380 @@
+/**
+ * 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;
+/** 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`. */
+  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. Called after every successful reopen, before a
+   * single byte is forwarded.
+   *
+   * **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. 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}.
+   */
+  attemptTimeoutMs?: number;
+}
+
+function describeSendError(error: unknown): string {
+  if (error instanceof Error) return error.message;
+  return String(error);
+}
+
+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;
+  /**
+   * 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;
+}
+
+export function createInputStreamRecovery(options: InputStreamRecoveryOptions): InputStreamRecovery {
+  const {
+    label,
+    name,
+    maxAttempts,
+    baseDelayMs,
+    log,
+    error,
+    isSettled,
+    getStream,
+    setStream,
+    openStream,
+    onRollback,
+    onExhausted,
+    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;
+
+    // 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);
+    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…`);
+
+    /**
+     * 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 | null = null;
+      try {
+        replacement = openStream();
+        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()) {
+        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.
+      //
+      // 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.
+      //
+      // `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';
+      }
+      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) => {
+          releaseBackoff = resolve;
+          timer = setTimeout(resolve, delay);
+          timer.unref?.();
+        });
+        timer = null;
+        releaseBackoff = 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,
+    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 f0ee36178..49700ec1b 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,31 @@ 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 +1236,73 @@ 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');
+  });
+});
+
+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 765f34d58..7ec4dcb1f 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 (${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,24 +608,30 @@ 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');
-          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;
-            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();
-          });
+        // 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');
+        } 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) {
@@ -629,6 +692,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 +796,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);
+  });
+});