fix(cli): stop the drive-attach input-stream flood; recover or exit non-zero

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
 

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.