From 9296c7c56f6c696fa05d02fce20bc56f0d4d5821 Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 17 Aug 2026 01:57:32 +0200 Subject: [PATCH 1/4] fix(broker,cli): stop escalating a single write_pty timeout into a full PTY input reconnect (relay#1544) A busy-but-alive driven coding agent that doesn't ack one keystroke in time was treated identically to a dead worker at three separate layers, each of which tore the whole PTY input channel down and forced a reconnect over a single slow write: - crates/broker/src/listen_api.rs: handle_pty_input_ws closed the WebSocket on ANY write_pty failure, including worker_timeout. New pty_input_error_is_connection_fatal(code) returns false only for worker_timeout (a confirmed-dead worker surfaces independently as worker_disappeared via fail_for_worker, well before this deadline); every other failure code still closes the connection exactly as before. - packages/harness-driver/src/transport.ts: PtyInputStream.handleMessage unconditionally failed+closed the whole stream on any pty_input_error. Now a worker_timeout settles only the in-flight write it correlates to and leaves the stream open for the next one. - packages/cli/src/cli/lib/attach-input-recovery.ts: handleSendFailure treated every non-backpressure rejection as stream death and called recover(). New isWriteTimeoutRejection (mirrors isBackpressureRejection) rolls back the optimistic echo and logs once per episode instead of reconnecting. PTY_INPUT_ACK_TIMEOUT (5s) is left unchanged: 100 live samples against a real driven coding agent (idle-thinking and actively streaming/tool- calling) on a heavily loaded shared node measured p50=123ms, p90=154- 220ms, max=804ms -- never within 6x of the timeout, so there's no local measurement to justify a new number. The fix targets the escalation itself, which is correct regardless of what occasionally trips the existing deadline (busy worker, GC pause, cross-node network jitter). Tests: crates/broker/src/listen_api.rs (listen_api::auth_tests) worker_timeout_does_not_close_the_pty_input_connection (must-fire, confirmed red-then-green by reverting the fix locally) and confirmed_dead_or_missing_worker_still_closes_the_pty_input_connection (must-not-fire). Mirrored must-fire/must-not-fire pairs added in pty-input-stream.test.ts and attach-input-recovery.test.ts. Co-Authored-By: Claude Sonnet 5 Session-Id: 68c2dae6-93d1-4e41-90b6-b35f1819a8e7 --- crates/broker/src/listen_api.rs | 76 +++++++++++++++++-- crates/broker/src/runtime/api.rs | 16 +++- .../src/cli/lib/attach-input-recovery.test.ts | 53 +++++++++++++ .../cli/src/cli/lib/attach-input-recovery.ts | 39 +++++++++- .../src/pty-input-stream.test.ts | 44 +++++++++++ packages/harness-driver/src/transport.ts | 15 ++++ 6 files changed, 235 insertions(+), 8 deletions(-) diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index 7d38c18b1..22ee55b35 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,39 @@ 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..f6f142755 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 @@ -136,6 +152,43 @@ describe('handleSendFailure', () => { expect(h.recovery.isRecovering()).toBe(true); expect(h.logs.some((l) => l.includes('input stream lost'))).toBe(true); }); + + 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); + // Every keystroke that failed to ack still has its optimistic echo rolled back. + expect(h.counts().rollbacks).toBe(50); + }); + + 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); + }); }); describe('identity verification is mandatory', () => { diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts index 0097d1470..282b34d71 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,25 @@ 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; the keystroke itself still failed to land (the + * caller must still roll back its optimistic echo), but the stream survives. + */ +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; @@ -159,6 +180,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 +230,7 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions): const noteSendSuccess = (): void => { backpressureReported = false; + writeTimeoutReported = false; }; const handleSendFailure = (sendError: unknown): void => { @@ -225,6 +249,19 @@ 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. The keystroke still didn't + // land, so roll back its optimistic echo same as backpressure. + onRollback(); + if (!writeTimeoutReported) { + writeTimeoutReported = true; + log(`[${label}] ${name} did not confirm a keystroke in time (worker busy); dropping it and continuing.`); + } + 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(); From 40d947602bcc974f598074e56956d5c50b5b0405 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Sun, 16 Aug 2026 23:58:39 +0000 Subject: [PATCH 2/4] style: auto-format Rust code with cargo fmt --- crates/broker/src/listen_api.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/crates/broker/src/listen_api.rs b/crates/broker/src/listen_api.rs index 22ee55b35..1eeee0a4b 100644 --- a/crates/broker/src/listen_api.rs +++ b/crates/broker/src/listen_api.rs @@ -5312,7 +5312,9 @@ mod auth_tests { // 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")); + assert!(!super::pty_input_error_is_connection_fatal( + "worker_timeout" + )); } #[test] @@ -5320,7 +5322,11 @@ mod auth_tests { // 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"] { + 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" From 5b47d95a6f7bc1c34ed7396aeb77457c59a6080a Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Mon, 17 Aug 2026 01:19:53 +0000 Subject: [PATCH 3/4] style: auto-format with Prettier --- packages/cli/src/cli/lib/attach-input-recovery.test.ts | 2 +- packages/cli/src/cli/lib/attach-input-recovery.ts | 8 ++++++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts index f6f142755..607fbac5e 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.test.ts +++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts @@ -153,7 +153,7 @@ describe('handleSendFailure', () => { expect(h.logs.some((l) => l.includes('input stream lost'))).toBe(true); }); - it('relay#1544 MUST-FIRE: does not start recovery for a busy worker\'s write timeout', () => { + 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 diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts index 282b34d71..8b27bd740 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.ts +++ b/packages/cli/src/cli/lib/attach-input-recovery.ts @@ -67,7 +67,9 @@ export function isBackpressureRejection(error: unknown): boolean { * caller must still roll back its optimistic echo), but the stream survives. */ export function isWriteTimeoutRejection(error: unknown): boolean { - return typeof error === 'object' && error !== null && (error as { code?: unknown }).code === 'worker_timeout'; + return ( + typeof error === 'object' && error !== null && (error as { code?: unknown }).code === 'worker_timeout' + ); } export interface InputStreamRecoveryOptions { @@ -258,7 +260,9 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions): onRollback(); if (!writeTimeoutReported) { writeTimeoutReported = true; - log(`[${label}] ${name} did not confirm a keystroke in time (worker busy); dropping it and continuing.`); + log( + `[${label}] ${name} did not confirm a keystroke in time (worker busy); dropping it and continuing.` + ); } return; } From 8437abb7f61142b6edd992d084f2621c5dbf896c Mon Sep 17 00:00:00 2001 From: Miya Date: Mon, 17 Aug 2026 08:15:11 +0200 Subject: [PATCH 4/4] fix(cli): do not roll back optimistic echo on worker_timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A worker_timeout ack is late, not a failed write: the broker's blocking write_all() onto the PTY master is still pending and very likely lands once the busy worker drains stdin. Rolling back the echo here erased operator input right before it executed, with no way to tell whether retyping was safe — worse than the reconnect flap this module fixes. Defer any rollback to a confirmed write failure, which already falls through to the existing recover() path unchanged. Also fixes the handleSendFailure contract doc, which still claimed only input_backpressure skips recovery after worker_timeout was added as a second non-recovering case. relay#1547 Session-Id: 78cef5cb-0e37-467e-a07f-b685f99fefe6 --- .../src/cli/lib/attach-input-recovery.test.ts | 26 +++++++++++-- .../cli/src/cli/lib/attach-input-recovery.ts | 39 +++++++++++++------ 2 files changed, 50 insertions(+), 15 deletions(-) diff --git a/packages/cli/src/cli/lib/attach-input-recovery.test.ts b/packages/cli/src/cli/lib/attach-input-recovery.test.ts index 607fbac5e..a42713398 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.test.ts +++ b/packages/cli/src/cli/lib/attach-input-recovery.test.ts @@ -144,13 +144,17 @@ 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", () => { @@ -173,8 +177,24 @@ describe('handleSendFailure', () => { 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); - // Every keystroke that failed to ack still has its optimistic echo rolled back. - expect(h.counts().rollbacks).toBe(50); + }); + + 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', () => { diff --git a/packages/cli/src/cli/lib/attach-input-recovery.ts b/packages/cli/src/cli/lib/attach-input-recovery.ts index 8b27bd740..936cc3404 100644 --- a/packages/cli/src/cli/lib/attach-input-recovery.ts +++ b/packages/cli/src/cli/lib/attach-input-recovery.ts @@ -63,8 +63,18 @@ export function isBackpressureRejection(error: unknown): boolean { * 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; the keystroke itself still failed to land (the - * caller must still roll back its optimistic echo), but the stream survives. + * 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 ( @@ -140,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. */ @@ -255,14 +268,16 @@ export function createInputStreamRecovery(options: InputStreamRecoveryOptions): // 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. The keystroke still didn't - // land, so roll back its optimistic echo same as backpressure. - onRollback(); + // 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); dropping it and continuing.` - ); + log(`[${label}] ${name} did not confirm a keystroke in time (worker busy); it may still land.`); } return; }