fix(broker,cli): stop escalating a single write_pty timeout into a full PTY input reconnect

crates/broker/src/listen_api.rs

@@ -1723,8 +1723,13 @@ fn classify_error(err: &str) -> (axum::http::StatusCode, &'static str) {
         // is with the resource's current capabilities.
         (axum::http::StatusCode::CONFLICT, "unsupported_runtime")
     } else if err.starts_with("worker_timeout") {
-        // Worker died or stalled between accepting the frame and
-        // replying. This is a server-side fault, not a bad request.
+        // The worker didn't ack before the deadline. This does NOT mean the
+        // worker died — a confirmed-dead worker is reaped independently
+        // (`fail_for_worker`) and surfaces as `worker_disappeared`, not this.
+        // `worker_timeout` also fires for a worker that is simply busy and
+        // hasn't drained its stdin yet (relay#1544); see
+        // `pty_input_error_is_connection_fatal`, which callers on the PTY
+        // input path use to avoid treating this code as transport death.
         (axum::http::StatusCode::GATEWAY_TIMEOUT, "worker_timeout")
     } else if err.starts_with("internal_error") {
         (
@@ -1738,6 +1743,22 @@ fn classify_error(err: &str) -> (axum::http::StatusCode, &'static str) {
     }
 }
 
+/// Whether a `write_pty` failure (as classified by [`classify_error`]) should
+/// tear down the whole PTY input connection.
+///
+/// `worker_timeout` means one write's ack didn't arrive before
+/// `PTY_INPUT_ACK_TIMEOUT` — the worker may simply be busy (not draining its
+/// stdin promptly while rendering/thinking), not dead. A confirmed-dead
+/// worker is reaped independently and reaches [`handle_pty_input_ws`] as
+/// `worker_disappeared` (or the target never existed at all:
+/// `agent_not_found` / `unsupported_runtime`), which are genuinely
+/// connection-fatal. Closing the connection on a mere timeout manufactures a
+/// transport failure out of a healthy-but-slow worker, which is exactly what
+/// forced the client-side reconnect loop in relay#1544.
+fn pty_input_error_is_connection_fatal(code: &str) -> bool {
+    code != "worker_timeout"
+}
+
 fn internal_error() -> (axum::http::StatusCode, axum::Json<Value>) {
     api_error(
         axum::http::StatusCode::INTERNAL_SERVER_ERROR,
@@ -1842,8 +1863,15 @@ async fn handle_pty_input_ws(
                         let (status, code) = classify_error(&err);
                         let _ =
                             send_pty_input_ws_error(&mut socket, code, err, status.as_u16()).await;
-                        let _ = socket.send(axum::extract::ws::Message::Close(None)).await;
-                        break;
+                        if pty_input_error_is_connection_fatal(code) {
+                            let _ = socket.send(axum::extract::ws::Message::Close(None)).await;
+                            break;
+                        }
+                        // `worker_timeout`: this one write didn't get an ack in
+                        // time, but the worker hasn't been confirmed dead. Keep
+                        // the connection open so the next keystroke gets a fresh
+                        // chance instead of forcing the client into a reconnect
+                        // it doesn't need (relay#1544).
                     }
                 }
             }
@@ -2084,13 +2112,18 @@ async fn send_pty_input_ws_error(
     message: impl Into<String>,
     status_code: u16,
 ) -> bool {
+    // `retryable` tells the client whether this connection is still usable:
+    // `worker_timeout` is the one code the WS loop doesn't close the socket
+    // for (see `pty_input_error_is_connection_fatal`), so it's the one code
+    // that's actually retryable on THIS stream rather than requiring reopen.
+    let retryable = !pty_input_error_is_connection_fatal(code);
     send_pty_input_ws_payload(
         socket,
         json!({
             "type": "pty_input_error",
             "code": code,
             "message": message.into(),
-            "retryable": false,
+            "retryable": retryable,
             "statusCode": status_code,
         }),
     )
@@ -5262,6 +5295,45 @@ mod auth_tests {
         replier.await.expect("replier should complete");
     }
 
+    // -----------------------------------------------------------------
+    // relay#1544: a busy-but-alive worker must not tear down the PTY input
+    // WebSocket just because one write's ack was slow. `worker_timeout` is
+    // the one `write_pty` failure code that does NOT mean the worker is
+    // confirmed dead (that's `worker_disappeared`, reaped independently),
+    // so `handle_pty_input_ws` must keep the connection open for it and
+    // close for everything else. See `pty_input_error_is_connection_fatal`.
+    // -----------------------------------------------------------------
+
+    #[test]
+    fn worker_timeout_does_not_close_the_pty_input_connection() {
+        // MUST-FIRE: this is the exact code a busy worker's slow write_pty
+        // ack produces (classify_error, api.rs PTY_INPUT_ACK_TIMEOUT). If
+        // this ever flips to `true`, `handle_pty_input_ws` starts closing
+        // the socket on every busy-but-healthy worker again — relay#1544's
+        // flap. Reverting the fix (`code != "worker_timeout"`, i.e. always
+        // fatal) makes this assertion fail.
+        assert!(!super::pty_input_error_is_connection_fatal(
+            "worker_timeout"
+        ));
+    }
+
+    #[test]
+    fn confirmed_dead_or_missing_worker_still_closes_the_pty_input_connection() {
+        // MUST-NOT-FIRE: a genuinely dead/missing/unusable target must still
+        // be treated as connection-fatal so the client's existing
+        // reconnect-on-close recovery still runs for a real outage.
+        for code in [
+            "worker_disappeared",
+            "agent_not_found",
+            "unsupported_runtime",
+        ] {
+            assert!(
+                super::pty_input_error_is_connection_fatal(code),
+                "{code} must still close the PTY input connection"
+            );
+        }
+    }
+
     // -----------------------------------------------------------------
     // Inbound delivery mode: four routes that back the `agent-relay drive`
     // client. The HTTP layer only forwards typed requests over the

crates/broker/src/runtime/api.rs

@@ -13,8 +13,20 @@ const DEFAULT_OBSERVER_TOKEN_NAME: &str = "pear-dashboard-observer";
 /// How long the broker waits for a worker's `write_pty_response` before it
 /// fails a PTY input ack. Keeps `PtyInputStream.send()` from hanging forever
 /// when a worker wedges or dies mid-write; the deadline sweep in `reap_tick`
-/// enforces it. Short because a confirmed PTY write is a local pipe → drainer
-/// round-trip that resolves in well under a second on a healthy worker.
+/// enforces it. Measured (relay#1544) at ~100-150ms p50 / <1s worst case for a
+/// single write's local pipe → drainer round-trip against a real driven
+/// coding agent under load (idle-thinking and actively streaming output
+/// alike, on a machine running two dozen concurrent agents) — 5s leaves
+/// ample headroom for that round trip without being raised on a guess.
+///
+/// Firing this deadline does NOT mean the worker is dead: a confirmed-dead
+/// worker is reaped independently and fails its pending requests immediately
+/// via `fail_for_worker` (`WorkerDisappeared`), well before this deadline
+/// would elapse. A `Timeout` here just means ONE write didn't get an ack in
+/// time — e.g. the worker was transiently busy, or cross-node network jitter
+/// added latency this local measurement can't see — so `handle_pty_input_ws`
+/// deliberately keeps the whole PTY input connection open when this is what
+/// fires (`pty_input_error_is_connection_fatal`); only that one write fails.
 const PTY_INPUT_ACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
 
 /// `/model` writes use the same worker-owned stdin writer as protocol frames.

packages/cli/src/cli/lib/attach-input-recovery.test.ts

@@ -3,6 +3,7 @@ import { describe, expect, it, vi } from 'vitest';
 import {
   createInputStreamRecovery,
   isBackpressureRejection,
+  isWriteTimeoutRejection,
   type InputStreamRecoveryOptions,
 } from './attach-input-recovery.js';
 import type { CliPtyInputStream } from './attach-drive.js';
@@ -99,6 +100,21 @@ describe('isBackpressureRejection', () => {
   });
 });
 
+describe('isWriteTimeoutRejection', () => {
+  it('recognises the transport code and nothing else', () => {
+    expect(isWriteTimeoutRejection(Object.assign(new Error('x'), { code: 'worker_timeout' }))).toBe(true);
+    expect(isWriteTimeoutRejection(Object.assign(new Error('x'), { code: 'worker_disappeared' }))).toBe(
+      false
+    );
+    expect(isWriteTimeoutRejection(Object.assign(new Error('x'), { code: 'input_backpressure' }))).toBe(
+      false
+    );
+    expect(isWriteTimeoutRejection(new Error('worker_timeout: worker did not respond in time'))).toBe(false);
+    expect(isWriteTimeoutRejection(null)).toBe(false);
+    expect(isWriteTimeoutRejection('worker_timeout')).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
@@ -128,13 +144,70 @@ describe('handleSendFailure', () => {
     expect(h.logs.filter((l) => l.includes('faster than'))).toHaveLength(2);
   });
 
-  it('treats every other rejection as stream loss', () => {
+  it('treats every other rejection as stream loss, and still rolls back the echo', () => {
+    // MUST-NOT-FIRE companion to the worker_timeout no-rollback test above:
+    // confirmed transport loss (unlike a merely-late ack) must still roll
+    // back the optimistic echo, since the queued write is genuinely gone.
     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);
+    expect(h.counts().rollbacks).toBe(1);
+  });
+
+  it("relay#1544 MUST-FIRE: does not start recovery for a busy worker's write timeout", () => {
+    // `worker_timeout` on one write means that write's ack was slow, not that
+    // the transport died — the transport (broker + PtyInputStream) already
+    // keeps the stream open for this exact code. Recovering here would
+    // reconnect a perfectly healthy stream on every busy-but-alive worker,
+    // reproducing relay#1544's self-healing flap. Revert the
+    // `isWriteTimeoutRejection` branch in `handleSendFailure` (fall through to
+    // `recover()`, as every non-backpressure rejection used to) and this
+    // assertion on `isRecovering()` goes red.
+    const h = harness();
+    const workerTimeout = Object.assign(new Error('worker_timeout: worker did not respond in time'), {
+      code: 'worker_timeout',
+    });
+
+    for (let i = 0; i < 50; i++) h.recovery.handleSendFailure(workerTimeout);
+
+    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('did not confirm a keystroke in time'))).toHaveLength(1);
+  });
+
+  it('relay#1547 MUST-FIRE: does not roll back the echo on worker_timeout, because the write can still land', () => {
+    // A late ack is not a failed write: the broker's write to the PTY is
+    // still pending when `worker_timeout` fires and very likely lands once
+    // the busy worker drains stdin. Rolling back here erases the operator's
+    // input right before it executes — a UI lie the operator cannot recover
+    // from, since they have no way to tell whether it's safe to retype. If
+    // `onRollback()` is reinstated in the `isWriteTimeoutRejection` branch of
+    // `handleSendFailure`, this assertion (rollbacks === 0) goes red.
+    const h = harness();
+    const workerTimeout = Object.assign(new Error('worker_timeout: worker did not respond in time'), {
+      code: 'worker_timeout',
+    });
+
+    for (let i = 0; i < 50; i++) h.recovery.handleSendFailure(workerTimeout);
+
+    expect(h.counts().rollbacks).toBe(0);
+  });
+
+  it('relay#1544 MUST-NOT-FIRE: a confirmed-dead worker (worker_disappeared) still recovers', () => {
+    // Distinct from `worker_timeout`: this code means the worker was reaped
+    // as gone, a genuine outage that must still trigger the reconnect loop.
+    const h = harness();
+    h.recovery.handleSendFailure(
+      Object.assign(new Error("worker_disappeared: worker 'Alice' exited before responding"), {
+        code: 'worker_disappeared',
+      })
+    );
+    expect(h.recovery.isRecovering()).toBe(true);
+    expect(h.logs.some((l) => l.includes('input stream lost'))).toBe(true);
   });
 });
 

packages/cli/src/cli/lib/attach-input-recovery.ts

@@ -41,14 +41,47 @@ export const INPUT_REOPEN_ATTEMPT_TIMEOUT_MS = 15_000;
  * (`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.
+ * into a detach. See {@link isWriteTimeoutRejection} for the other rejection
+ * that must not trigger a teardown; every remaining 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'
   );
 }
 
+/**
+ * `PtyInputStream.send()` rejects with `worker_timeout` when the broker's
+ * `write_pty` ack for THAT ONE keystroke didn't arrive before
+ * `PTY_INPUT_ACK_TIMEOUT` (broker `api.rs`). That does not mean the worker is
+ * dead: a confirmed-dead worker is reaped independently and surfaces as
+ * `worker_disappeared`, and the broker keeps this WebSocket open for
+ * `worker_timeout` specifically (`pty_input_error_is_connection_fatal`,
+ * `listen_api.rs`) because a busy-but-alive coding agent that hasn't drained
+ * its stdin yet produces the exact same "no ack yet" as a wedged one — the
+ * broker has no other liveness signal to tell them apart. Tearing the whole
+ * input stream down and reconnecting over one slow ack is what produced the
+ * self-healing "input stream lost / reconnected after 1 attempt(s)" flap
+ * reported in relay#1544.
+ *
+ * Critically, a late ack is not a failed write: the broker's `write_all()`
+ * onto the PTY master is still blocking and pending when the ack times out,
+ * and it very likely completes once the busy child drains its stdin — the
+ * keystroke lands anyway. Rolling back the optimistic echo here would show
+ * the operator input vanishing right before it executes, a UI lie that is
+ * worse than the flap this module exists to fix and that the operator cannot
+ * recover from (they cannot tell whether it is safe to retype). So this code
+ * must NOT roll back the echo. Only a *confirmed* write failure — a different
+ * error code entirely — may roll it back, and that already falls through to
+ * {@link createInputStreamRecovery}'s default `recover()` path below.
+ */
+export function isWriteTimeoutRejection(error: unknown): boolean {
+  return (
+    typeof error === 'object' && error !== null && (error as { code?: unknown }).code === 'worker_timeout'
+  );
+}
+
 export interface InputStreamRecoveryOptions {
   /** Log prefix tag — `drive` or `passthrough`. */
   label: string;
@@ -117,10 +150,13 @@ export interface InputStreamRecovery {
   /** 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.
+   * Classify a rejected `send()`. Two codes never enter recovery:
+   * `input_backpressure` is flow control on a healthy stream and only costs
+   * the optimistic echo, and `worker_timeout` is a late ack on a write that
+   * may still land, so it costs nothing — no rollback, no recovery. Every
+   * other rejection is treated as transport loss and enters recovery, which
+   * does roll back. Callers route every send rejection here rather than
+   * assuming loss.
    */
   handleSendFailure(error: unknown): void;
   /** Clears the backpressure latch so a later episode reports again. */
@@ -159,6 +195,8 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
   let releaseBackoff: (() => void) | null = null;
   /** True once a backpressure episode has been reported; reset on the next good send. */
   let backpressureReported = false;
+  /** True once a write-timeout episode has been reported; reset on the next good send. */
+  let writeTimeoutReported = false;
 
   const cancel = (): void => {
     if (timer) {
@@ -207,6 +245,7 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
 
   const noteSendSuccess = (): void => {
     backpressureReported = false;
+    writeTimeoutReported = false;
   };
 
   const handleSendFailure = (sendError: unknown): void => {
@@ -225,6 +264,23 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions):
       }
       return;
     }
+    if (isWriteTimeoutRejection(sendError)) {
+      // This one write's ack didn't arrive in time — the worker may just be
+      // busy, not dead (relay#1544). The transport already kept the socket
+      // open for this exact code, so recovering here would manufacture the
+      // reconnect this module exists to avoid. A late ack is not a failed
+      // write either: the broker's write to the PTY is still pending and
+      // very likely lands once the worker drains its stdin, so do NOT roll
+      // back the optimistic echo — doing so would erase input right before
+      // it executes, with no way for the operator to tell it happened. Only
+      // a confirmed write failure (a distinct error code) may roll it back,
+      // and that already falls through to `recover()` below.
+      if (!writeTimeoutReported) {
+        writeTimeoutReported = true;
+        log(`[${label}] ${name} did not confirm a keystroke in time (worker busy); it may still land.`);
+      }
+      return;
+    }
     recover(describeSendError(sendError));
   };