diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index 7d38c18b1..1eeee0a4b 100644 --- a/crates/broker/src/listen_api.rs +++ b/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) { 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, 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 diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index dfa2add9c..ed7665e18 100644 --- a/crates/broker/src/runtime/api.rs +++ b/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. diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts index 77ba28767..a42713398 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.test.ts +++ b/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); }); }); diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts index 0097d1470..936cc3404 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.ts +++ b/packages/cli/src/cli/lib/attach-input-recovery.ts @@ -41,7 +41,9 @@ 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 ( @@ -49,6 +51,37 @@ export function isBackpressureRejection(error: unknown): boolean { ); } +/** + * `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)); }; diff --git a/packages/harness-driver/src/pty-input-stream.test.ts b/packages/harness-driver/src/pty-input-stream.test.ts index e313488ec..ab039fb3c 100644 --- a/packages/harness-driver/src/pty-input-stream.test.ts +++ b/packages/harness-driver/src/pty-input-stream.test.ts @@ -174,6 +174,50 @@ describe('PtyInputStream pipelining', () => { expect(stream.closed).toBe(true); }); + it('relay#1544 MUST-FIRE: a worker_timeout on one write must not close the stream', async () => { + // The broker keeps the socket open for `worker_timeout` (the write may + // simply be a busy worker, not a dead one — see + // `pty_input_error_is_connection_fatal` on the broker side). Only that + // one write should reject; the stream itself must stay usable so the + // next keystroke doesn't have to go through a full reconnect. Before the + // fix, `handleMessage` closed on every `pty_input_error` unconditionally + // and this assertion on `stream.closed` failed. + const stream = new PtyInputStream({ url: 'ws://x/api/input/agent/stream' }); + const socket = lastSocket(); + socket.open(); + await stream.waitUntilOpen(); + + const p1 = stream.send('x'); + await Promise.resolve(); + socket.errorFrame('worker_timeout', 'worker_timeout: worker did not respond in time'); + + await expect(p1).rejects.toMatchObject({ code: 'worker_timeout' }); + expect(stream.closed).toBe(false); + + // The stream is still usable for the next keystroke. + const p2 = stream.send('y'); + await Promise.resolve(); + socket.ack(1); + await expect(p2).resolves.toMatchObject({ bytes_written: 1 }); + }); + + it('relay#1544 MUST-NOT-FIRE: a confirmed-dead worker still closes the stream', async () => { + // `worker_disappeared` means the worker was reaped independently — a + // genuine outage, not a slow ack. This must still close the stream so + // the CLI's existing reconnect-on-close recovery still runs. + const stream = new PtyInputStream({ url: 'ws://x/api/input/agent/stream' }); + const socket = lastSocket(); + socket.open(); + await stream.waitUntilOpen(); + + const p = stream.send('x'); + await Promise.resolve(); + socket.errorFrame('worker_disappeared', "worker 'agent' exited before responding"); + + await expect(p).rejects.toMatchObject({ code: 'worker_disappeared' }); + expect(stream.closed).toBe(true); + }); + it('fails all in-flight and queued frames on close', async () => { const stream = new PtyInputStream({ url: 'ws://x/api/input/agent/stream' }); const socket = lastSocket(); diff --git a/packages/harness-driver/src/transport.ts b/packages/harness-driver/src/transport.ts index 031c7d31d..69e5db761 100644 --- a/packages/harness-driver/src/transport.ts +++ b/packages/harness-driver/src/transport.ts @@ -325,6 +325,21 @@ export class PtyInputStream { status: typeof message.statusCode === 'number' ? message.statusCode : undefined, data: message, }); + + // `worker_timeout` means the ONE write this error correlates to + // didn't get an ack in time — the worker may simply be busy, not + // dead (relay#1544). The broker keeps the socket open for exactly + // this code (see `pty_input_error_is_connection_fatal` on the + // broker side); settle only the write it belongs to and leave the + // stream usable for the next one. Every other code means the broker + // is closing (or refused to open) the connection, so it's still + // fatal to the whole stream. + if (error.code === 'worker_timeout') { + const current = this.inFlight.shift(); + if (current) this.settle(current, error); + return; + } + this.rejectOpen(error); this.failAll(error); this.close();