From ff09a06003be099bec5b6df34f13bfffea68b7e4 Mon Sep 17 00:00:00 2001 From: Will Washburn Date: Mon, 27 Jul 2026 12:20:01 -0400 Subject: [PATCH 1/2] fix(cli): keep driven agents receiving relay messages Attaching with `drive` put the agent in `manual_flush` for the whole session, so every inbound relay message parked until Ctrl+] or detach. Watching a team coordinate therefore stopped it: three driven agents exchanged 8 messages in seven minutes where the same unattended run exchanged 21 in three. `drive` now asserts `auto_inject` on attach. Ctrl+] holds delivery for when the screen must stay still while typing, and a second press drains the queue and returns to live delivery. Detach still restores an explicit pre-attach hold. The status line also counted every `delivery_queued` frame, including the ones harness runtimes emit for their own injection queues. With live delivery that count climbed on ordinary traffic and never came back down, so it now counts only the frames the inbound hold actually parked. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 2 + packages/cli/src/cli/lib/attach-drive.test.ts | 175 +++++++++++++----- packages/cli/src/cli/lib/attach-drive.ts | 84 +++++---- 3 files changed, 175 insertions(+), 86 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 683303b99..2653c272a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `@agent-relay/config`, `@agent-relay/cli`, `@agent-relay/fleet`, and `@agent-relay/harness-driver` now build on Zod 4, matching `@agent-relay/sdk`; Relay no longer ships a split Zod 3/Zod 4 install. - `@agent-relay/config` builds `jsonSchemas` with Zod's built-in `z.toJSONSchema` and drops the `zod-to-json-schema` dependency. Output stays draft-7 and still describes config files as authored, but some keywords differ (records gain `propertyNames`, integers gain explicit bounds, and `RegExp` fields render as unconstrained). - `@agent-relay/harness-driver` now exports `SpawnAgentResultSchema` as a Zod 4 schema, and `@agent-relay/fleet` `action()` validates its `input` with Zod 4. Code that pairs either with its own Zod 3 instance — `z.infer`, `.extend()` — needs `zod@^4`. +- `node agent attach --mode drive` keeps the agent in `auto_inject`, so a driven agent goes on receiving relay messages while you watch it instead of having them parked for the whole session. `Ctrl+]` now holds delivery when you want the screen still while typing; a second press drains the queue and returns to live delivery. ### Added @@ -33,6 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - PTY `node agent attach --mode drive|passthrough` sessions now reserve and safely clip their status row, preventing full-screen agent CLIs from tearing, scrolling, or duplicating Relay's controls. - Detaching from `node agent attach --mode drive|passthrough` restores the row and column the status line reserved, so a later `--mode view` session no longer inherits a PTY one row and column short. `POST /api/resize/{name}` applies dimensions sent alongside `release: true`. - Agent Relay MCP instructions now route work with existing named participants through Relay instead of provider-native subagents. +- The `drive` status line counts only messages actually parked by the inbound hold. It previously also counted every delivery a harness queued for injection, so `pending` climbed on ordinary traffic and never came back down. - `node agent attach --mode view` now exits on the first Ctrl-C instead of waiting for a WebSocket close handshake. - The broker now sends its anonymous telemetry id (`X-Agent-Relay-Distinct-Id`) and origin actor with its Relaycast requests, so hosted usage can be attributed to an install instead of only to a workspace. The id header is omitted when telemetry is opted out; requests and origin actor are unaffected. - The broker now reads its telemetry preference and machine-id files from `AGENT_RELAY_DATA_DIR` when set, matching the CLI. It previously only read `~/.agentworkforce/relay/telemetry.json`, so an opt-out written by `agent-relay telemetry disable` under a configured data directory was ignored. diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts index e299b6857..2cec0168d 100644 --- a/packages/cli/src/cli/lib/attach-drive.test.ts +++ b/packages/cli/src/cli/lib/attach-drive.test.ts @@ -484,17 +484,41 @@ describe('classifyWsEvent', () => { ).toEqual({ kind: 'other' }); }); - it('classifies delivery_queued for the targeted agent, carrying its event id', () => { + it('classifies a held delivery_queued for the targeted agent, carrying its event id', () => { expect( - classifyWsEvent(JSON.stringify({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' }), 'Alice') + classifyWsEvent( + JSON.stringify({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e1', + reason: 'inbound_delivery_manual_flush', + }), + 'Alice' + ) ).toEqual({ kind: 'delivery_queued', eventId: 'e1' }); }); - it('classifies delivery_queued without an event id (legacy frame)', () => { - expect(classifyWsEvent(JSON.stringify({ kind: 'delivery_queued', name: 'Alice' }), 'Alice')).toEqual({ - kind: 'delivery_queued', - eventId: undefined, - }); + it('classifies a held delivery_queued without an event id (legacy frame)', () => { + expect( + classifyWsEvent( + JSON.stringify({ + kind: 'delivery_queued', + name: 'Alice', + reason: 'inbound_delivery_manual_flush', + }), + 'Alice' + ) + ).toEqual({ kind: 'delivery_queued', eventId: undefined }); + }); + + it('ignores a delivery_queued the harness emitted for its own injection queue', () => { + // The PTY/headless runtimes emit this kind for every delivery they enqueue + // for injection — ordinary traffic reaching the agent, not a parked + // message. Counting it would make `pending` climb on every message and + // never come back down. + expect( + classifyWsEvent(JSON.stringify({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' }), 'Alice') + ).toEqual({ kind: 'other' }); }); it('classifies agent_pending_drained with optional count', () => { @@ -657,17 +681,19 @@ describe('renderStatusLine', () => { }); describe('runDriveSession', () => { - it('flips to manual_flush delivery mode, renders snapshot, opens WS, then restores prior mode on detach', async () => { - const { deps, sockets, fetchLog, stdin, logs } = createHarness({ initialMode: 'auto_inject' }); + it('asserts auto_inject delivery mode, renders snapshot, opens WS, then restores prior mode on detach', async () => { + // Pre-attach hold, so the detach restore has something to put back. + // Attaching still goes live: watching an agent never pauses its intake. + const { deps, sockets, fetchLog, stdin, logs } = createHarness({ initialMode: 'manual_flush' }); const sessionPromise = runDriveSession('Alice', {}, deps); const socket = await openSocket(sockets); expect(socket.url).toBe('ws://localhost:3889/ws'); expect(socket.headers['X-API-Key']).toBe('k'); expect(logs.some((args) => String(args[0]).includes('driving Alice via'))).toBe(false); - // PUT /delivery-mode body should be { mode: 'manual_flush' }. + // PUT /delivery-mode body should be { mode: 'auto_inject' }. const flipCall = fetchLog.find((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); - expect(flipCall?.body).toEqual({ mode: 'manual_flush' }); + expect(flipCall?.body).toEqual({ mode: 'auto_inject' }); // Raw mode should be on after open. expect(stdin.rawModeCalls.includes(true)).toBe(true); @@ -680,13 +706,13 @@ describe('runDriveSession', () => { // Raw mode restored. expect(stdin.rawModeCalls).toEqual([true, false]); - // Last PUT /delivery-mode call should restore to 'auto_inject' (the prior - // mode) via a compare-and-set guarded by `expected_mode: manual_flush`. + // Last PUT /delivery-mode call should restore to 'manual_flush' (the prior + // mode) via a compare-and-set guarded by `expected_mode: auto_inject`. const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls).toHaveLength(2); expect(modeCalls[1].body).toEqual({ - mode: 'auto_inject', - expected_mode: 'manual_flush', + mode: 'manual_flush', + expected_mode: 'auto_inject', expected_revision: '1', }); }); @@ -793,8 +819,8 @@ describe('runDriveSession', () => { // Best-effort restore PUT should still have fired. const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'auto_inject', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'auto_inject', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); @@ -834,8 +860,24 @@ describe('runDriveSession', () => { const initialPaints = writes.filter((w) => w.includes('drive Alice')).length; expect(initialPaints).toBeGreaterThan(0); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' })); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e2' })); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e1', + reason: 'inbound_delivery_manual_flush', + }) + ); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e2', + reason: 'inbound_delivery_manual_flush', + }) + ); expect(writes.some((w) => w.includes('pending=1'))).toBe(true); expect(writes.some((w) => w.includes('pending=2'))).toBe(true); @@ -852,9 +894,33 @@ describe('runDriveSession', () => { const sessionPromise = runDriveSession('Alice', {}, deps); const socket = await openSocket(sockets); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' })); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e2' })); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e3' })); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e1', + reason: 'inbound_delivery_manual_flush', + }) + ); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e2', + reason: 'inbound_delivery_manual_flush', + }) + ); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e3', + reason: 'inbound_delivery_manual_flush', + }) + ); expect(writes.some((w) => w.includes('pending=3'))).toBe(true); // A failed injection stops a flush mid-queue: only 2 of 3 drained. @@ -866,33 +932,36 @@ describe('runDriveSession', () => { await sessionPromise; }); - it('toggles delivery to auto_inject on Ctrl+], re-holds on a second press, and restores from the latest revision on detach', async () => { + it('holds delivery on Ctrl+], goes live again on a second press, and restores from the latest revision on detach', async () => { const { deps, sockets, writes, stdin, fetchLog, inputStreams } = createHarness({ initialMode: 'auto_inject', }); const sessionPromise = runDriveSession('Alice', {}, deps); await openSocket(sockets); - // First Ctrl+] — go live: CAS from this session's attach flip (rev 1). + // First Ctrl+] — hold so the screen stays still while typing: CAS from + // this session's attach assertion (rev 1). stdin.type(Buffer.from([0x1d])); for (let i = 0; i < 10; i++) await new Promise((resolve) => setImmediate(resolve)); let modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls).toHaveLength(2); expect(modeCalls[1].body).toEqual({ - mode: 'auto_inject', - expected_mode: 'manual_flush', + mode: 'manual_flush', + expected_mode: 'auto_inject', expected_revision: '1', }); - expect(writes.some((w) => w.includes('delivery=auto_inject') && w.includes('Ctrl+] hold'))).toBe(true); + expect(writes.some((w) => w.includes('delivery=manual_flush') && w.includes('Ctrl+] deliver'))).toBe( + true + ); - // Second Ctrl+] — back to hold: CAS from the toggled state (rev 2). + // Second Ctrl+] — drain and go live again: CAS from the toggled state (rev 2). stdin.type(Buffer.from([0x1d])); for (let i = 0; i < 10; i++) await new Promise((resolve) => setImmediate(resolve)); modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls).toHaveLength(3); expect(modeCalls[2].body).toEqual({ - mode: 'manual_flush', - expected_mode: 'auto_inject', + mode: 'auto_inject', + expected_mode: 'manual_flush', expected_revision: '2', }); @@ -908,7 +977,7 @@ describe('runDriveSession', () => { expect(modeCalls).toHaveLength(4); expect(modeCalls[3].body).toEqual({ mode: 'auto_inject', - expected_mode: 'manual_flush', + expected_mode: 'auto_inject', expected_revision: '3', }); }); @@ -969,15 +1038,15 @@ describe('runDriveSession', () => { expect(modeCalls).toHaveLength(3); // The toggle landed (rev 1 → 2)… expect(modeCalls[1].body).toEqual({ - mode: 'auto_inject', - expected_mode: 'manual_flush', + mode: 'manual_flush', + expected_mode: 'auto_inject', expected_revision: '1', }); // …and the restore CASed against the POST-toggle state, not the stale // attach-time revision. expect(modeCalls[2].body).toEqual({ mode: 'auto_inject', - expected_mode: 'auto_inject', + expected_mode: 'manual_flush', expected_revision: '2', }); }); @@ -1003,8 +1072,8 @@ describe('runDriveSession', () => { const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); // The toggle is still mode-guarded — never an unconditional write. expect(modeCalls[1]?.body).toEqual({ - mode: 'auto_inject', - expected_mode: 'manual_flush', + mode: 'manual_flush', + expected_mode: 'auto_inject', }); stdin.type(Buffer.from([0x03])); @@ -1044,17 +1113,25 @@ describe('runDriveSession', () => { const socket = await openSocket(sockets); expect(writes.some((w) => w.includes('pending=2'))).toBe(true); + const held = (eventId: string) => + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: eventId, + reason: 'inbound_delivery_manual_flush', + }); + // Replayed frame for a seeded delivery — deduped, stays at 2. - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e0' })); + socket.emit('message', held('e0')); expect(writes.some((w) => w.includes('pending=3'))).toBe(false); // A genuinely new delivery — counted, goes to 3. - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'brand-new' })); + socket.emit('message', held('brand-new')); expect(writes.some((w) => w.includes('pending=3'))).toBe(true); // The same seeded id re-queued *later* (after being consumed once from the // seed) counts normally — the id was forgotten after its first dedupe. - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e0' })); + socket.emit('message', held('e0')); expect(writes.some((w) => w.includes('pending=4'))).toBe(true); stdin.type(Buffer.from([0x03])); @@ -1178,8 +1255,8 @@ describe('runDriveSession', () => { const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'auto_inject', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'auto_inject', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); @@ -1195,8 +1272,8 @@ describe('runDriveSession', () => { const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'auto_inject', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'auto_inject', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); @@ -1212,8 +1289,8 @@ describe('runDriveSession', () => { // Restore to 'manual_flush' since that was the prior mode, via a // compare-and-set guarded by `expected_mode: manual_flush`. expect(modeCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'manual_flush', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'manual_flush', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); @@ -1612,8 +1689,8 @@ describe('runDriveSession', () => { const modeCalls = fetchLog.filter((c) => c.method === 'PUT' && c.url.endsWith('/delivery-mode')); expect(modeCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'auto_inject', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'auto_inject', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); @@ -1978,8 +2055,8 @@ describe('runDriveSession', () => { // Attach flip (unconditional), then a compare-and-set restore guarded by // `expected_mode`. The restore no-ops broker-side rather than clobbering. expect(putCalls.map((c) => c.body)).toEqual([ - { mode: 'manual_flush' }, - { mode: 'auto_inject', expected_mode: 'manual_flush', expected_revision: '1' }, + { mode: 'auto_inject' }, + { mode: 'auto_inject', expected_mode: 'auto_inject', expected_revision: '1' }, ]); }); }); diff --git a/packages/cli/src/cli/lib/attach-drive.ts b/packages/cli/src/cli/lib/attach-drive.ts index 4274764b1..48dd4e20a 100644 --- a/packages/cli/src/cli/lib/attach-drive.ts +++ b/packages/cli/src/cli/lib/attach-drive.ts @@ -1,29 +1,31 @@ /** * `agent-relay drive ` — interactive read-write take-over client. * - * Attaches to a running agent, flips it into `manual_flush` inbound delivery mode so the - * broker parks new relay messages in a per-worker queue, and forwards your - * keystrokes to the worker's PTY. Parked messages are surfaced in the status - * line (`pending=N`) and released in-band with `Ctrl+]`, which toggles the - * worker between `manual_flush` (hold) and `auto_inject` (deliver): flipping to - * `auto_inject` drains the queue into the PTY and lets later messages inject - * live while you watch — without it, a driven agent can never receive relay - * messages (the reply to anything you ask it to send sits parked until detach). - * The out-of-band commands `local agent message flush`, `local agent message - * hold`, and `local agent message auto` still work from another terminal: a - * bare `flush` during a drive session injects the queued backlog immediately - * (the broker follows the handoff with a one-shot `flush_injections` frame - * that exempts it from the interactive hold) while leaving the mode — and the - * parking of later messages — unchanged. `Ctrl+C` detaches, restores the - * worker's previous inbound delivery mode, and leaves the agent running under - * the broker — `drive` never kills the worker. + * Attaches to a running agent, puts it in `auto_inject` inbound delivery mode so + * relay messages keep reaching it live, and forwards your keystrokes to the + * worker's PTY. Watching an agent never pauses it: a driven agent goes on + * receiving its peers' messages exactly as it would unattended, which is what + * makes it possible to observe a team coordinating instead of freezing it by + * looking at it. + * + * `Ctrl+]` toggles the worker into `manual_flush` when you want the screen to + * hold still while you type — messages park in a per-worker queue, the status + * line counts them (`pending=N`), and the next press drains the queue and + * returns to live delivery. The out-of-band commands `local agent message + * flush`, `local agent message hold`, and `local agent message auto` still work + * from another terminal: a bare `flush` during a drive session injects the + * queued backlog immediately (the broker follows the handoff with a one-shot + * `flush_injections` frame that exempts it from the interactive hold) while + * leaving the mode — and the parking of later messages — unchanged. `Ctrl+C` + * detaches, restores the worker's previous inbound delivery mode, and leaves + * the agent running under the broker — `drive` never kills the worker. * * Sequence of operations on attach (subscribe-first, so no output around * attach time is lost and none is double-painted): * * 1. Discover broker connection (CLI flag → env → connection.json). * 2. `GET /api/spawned/{name}/delivery-mode` → remember the previous mode. - * 3. `PUT /api/spawned/{name}/delivery-mode` → switch to `manual_flush`. + * 3. `PUT /api/spawned/{name}/delivery-mode` → assert `auto_inject`. * 4. `GET /api/events/replay` → capture the durable-event `sinceSeq` cutoff, * then `GET /api/spawned/{name}/pending` → seed the status-line counter * and the set of already-queued `event_id`s. Cutoff-first + id-dedupe @@ -476,6 +478,14 @@ export function classifyWsEvent(rawMessage: string, name: string): DriveWsEvent return { kind: 'worker_stream', chunk, offset }; } if (parsed.kind === 'delivery_queued') { + // Two different layers emit `delivery_queued`. Only the inbound hold means + // "parked in the per-worker pending queue" — the count this status line + // shows and the one `/api/spawned/{name}/pending` seeds. The harness + // runtimes emit the same kind for every delivery they enqueue for + // injection, which is ordinary traffic on its way to the agent; counting + // those would make `pending` climb on every message the agent receives and + // never come back down (nothing drains a queue the message was never in). + if (parsed.reason !== 'inbound_delivery_manual_flush') return { kind: 'other' }; // `event_id` correlates a replayed frame with the pending seed so the // counter isn't double-incremented for a delivery already reflected in // the seed (see the seed-dedup note in `runDriveSession`). Absent on @@ -567,10 +577,11 @@ export function renderStatusLine(opts: { const row = Math.max(opts.rows ?? 24, 1); const scrollTop = Math.max(1, opts.scrollTop ?? 1); const scrollBottom = Math.max(scrollTop + 1, opts.scrollBottom ?? row - 1); - // The Ctrl+] hint names the action the NEXT press performs: in manual_flush - // it delivers (drains the parked queue and goes live); in auto_inject it - // re-holds. Without the hint, a parked message is invisible beyond the - // pending counter and a driven agent looks like it never receives replies. + // The Ctrl+] hint names the action the NEXT press performs: in auto_inject + // (the session default) it holds; in manual_flush it delivers, draining the + // parked queue and going live again. Without the hint, a parked message is + // invisible beyond the pending counter and a held agent looks like it never + // receives replies. const toggleHint = opts.mode === 'manual_flush' ? 'Ctrl+] deliver' : 'Ctrl+] hold'; const text = clampStatusLineText( `[drive ${opts.name} | delivery=${opts.mode} | pending=${opts.pending} | ${toggleHint} | Ctrl+C detach]`, @@ -661,10 +672,10 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies): let reassertTimer: ReturnType | null = null; let pending = state.initialPending; // This session's last-known inbound delivery mode and its broker revision. - // The attach flip set `manual_flush`; `Ctrl+]` toggles it mid-session, and + // The attach asserted `auto_inject`; `Ctrl+]` toggles it mid-session, and // the detach restore compare-and-sets against whatever this session last // wrote so an out-of-band change is never clobbered. - let currentMode: InboundDeliveryMode = 'manual_flush'; + let currentMode: InboundDeliveryMode = 'auto_inject'; let currentRevision = state.sessionRevision; let terminalRows = pickInitialTerminalRows(state.initialLocalSize, undefined); let terminalCols = pickInitialTerminalCols(state.initialLocalSize, undefined); @@ -832,11 +843,9 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies): }; // In-band delivery toggle (`Ctrl+]`). Flips the worker between - // `manual_flush` (messages park so nothing splices into the human's - // typing) and `auto_inject` (the parked queue drains into the PTY and - // later messages inject live). Without this, an agent being driven can - // never receive a relay message — the reply to anything the human asks it - // to send just sits in the queue until detach. + // `auto_inject` (the session default — messages inject live while you + // watch) and `manual_flush` (messages park so nothing splices into what + // you are typing); flipping back drains the parked queue into the PTY. // // Guarded compare-and-set against this session's last-known revision: if // another session or CLI changed the mode out-of-band, the broker no-ops @@ -1047,8 +1056,8 @@ function runDriveSessionLoop(state: DriveSessionState, deps: DriveDependencies): // buffered chunk flushes to stdout after detach. deps.disposeWriter?.(); // Best-effort restore: re-read the mode and only revert if it's still - // what this session set, so we don't leave the worker stuck in - // manual_flush and don't clobber a change another session made. + // what this session set, so an explicit pre-attach `hold` is put back and + // a change another session made is never clobbered. // // Await the release alongside the restore before resolving: `resolve` // typically ends the process, which aborts any still-pending fetch. The @@ -1314,20 +1323,21 @@ export async function runDriveSession( const flipResult = await switchInboundDeliveryModeOrAbort( connection, name, - 'manual_flush', - `switch '${name}' to manual_flush mode`, + 'auto_inject', + `switch '${name}' to auto_inject mode`, deps ); if (!flipResult) return 1; const { previousMode, sessionRevision } = flipResult; - // The mode is now flipped to `manual_flush`, but the terminal is still cooked + // The mode is now asserted to `auto_inject`, but the terminal is still cooked // and we have several awaited HTTP round-trips (pending, cutoff, snapshot, // input stream) before the session loop installs its signal handlers. Ctrl+C // in that window would otherwise kill the process with the worker stranded in - // `manual_flush`, silently queueing every later relay message. Register early - // restore-and-exit handlers immediately; the loop disposes them once its own - // handlers are ready (no double restore — see `disposeEarlySignals`). + // this session's mode, silently discarding an explicit `hold` the operator had + // set before attaching. Register early restore-and-exit handlers immediately; + // the loop disposes them once its own handlers are ready (no double restore — + // see `disposeEarlySignals`). let earlyHandled = false; const earlyCleanups: Array<() => void> = []; const earlyRestore = async (): Promise => { @@ -1344,7 +1354,7 @@ export async function runDriveSession( connection, name, previousMode, - 'manual_flush', + 'auto_inject', sessionRevision, 'drive', deps From 3d128388235d1045244547e0eaa79d41a0b7cce0 Mon Sep 17 00:00:00 2001 From: relay Date: Wed, 29 Jul 2026 21:07:13 -0400 Subject: [PATCH 2/2] fix(cli): abort attach when live delivery cannot resume --- packages/cli/src/cli/lib/attach-drive.test.ts | 58 ++++++++++++++++++- packages/cli/src/cli/lib/attach.ts | 14 ++++- 2 files changed, 68 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/cli/lib/attach-drive.test.ts b/packages/cli/src/cli/lib/attach-drive.test.ts index 2cec0168d..634c89185 100644 --- a/packages/cli/src/cli/lib/attach-drive.test.ts +++ b/packages/cli/src/cli/lib/attach-drive.test.ts @@ -717,6 +717,42 @@ describe('runDriveSession', () => { }); }); + it('aborts when the broker cannot complete the transition to auto_inject', async () => { + const { deps, sockets, fetchLog, errors, stdin } = createHarness({ + initialMode: 'manual_flush', + routes: { + 'PUT /delivery-mode': async () => + new Response( + JSON.stringify({ + mode: 'manual_flush', + flushed: 0, + matched: true, + revision: '1', + }), + { + status: 200, + headers: { 'Content-Type': 'application/json' }, + } + ), + }, + }); + + await expect(runDriveSession('Alice', {}, deps)).resolves.toBe(1); + + expect(sockets).toHaveLength(0); + expect(stdin.rawModeCalls).toEqual([]); + expect( + fetchLog.filter((call) => call.method === 'PUT' && call.url.endsWith('/delivery-mode')) + ).toHaveLength(1); + expect( + errors.some((args) => + String(args[0]).includes( + "could not switch 'Alice' to auto_inject mode: broker remained in manual_flush mode" + ) + ) + ).toBe(true); + }); + it('takes stdin raw before replaying a TUI snapshot', async () => { const { deps, sockets, stdin } = createHarness(); deps.captureAndRenderSnapshot = vi.fn(async () => { @@ -992,7 +1028,7 @@ describe('runDriveSession', () => { // out-of-band change and reports the current (unchanged) state. const body = putCalls === 1 - ? { mode: 'manual_flush', flushed: 0, matched: true, revision: '1' } + ? { mode: 'auto_inject', flushed: 0, matched: true, revision: '1' } : { mode: 'manual_flush', flushed: 0, matched: false, revision: '7' }; return new Response(JSON.stringify(body), { status: 200, @@ -1772,7 +1808,15 @@ describe('runDriveSession', () => { // repaint. Painting now would splice reverse-video controls into the // agent's half-sent sequence. socket.emit('message', jsonMessage({ kind: 'worker_stream', name: 'Alice', chunk: 'data\x1b[' })); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' })); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e1', + reason: 'inbound_delivery_manual_flush', + }) + ); expect(writes.filter((w) => w.includes('drive Alice')).length).toBe(paintsBefore); // Completing the CSI lands at a boundary → the deferred repaint fires. socket.emit('message', jsonMessage({ kind: 'worker_stream', name: 'Alice', chunk: '2J' })); @@ -1806,7 +1850,15 @@ describe('runDriveSession', () => { const paintsBefore = writes.filter((write) => write.includes('[drive Alice')).length; socket.emit('message', jsonMessage({ kind: 'worker_stream', name: 'Alice', chunk: 'data\x1b[' })); - socket.emit('message', jsonMessage({ kind: 'delivery_queued', name: 'Alice', event_id: 'e1' })); + socket.emit( + 'message', + jsonMessage({ + kind: 'delivery_queued', + name: 'Alice', + event_id: 'e1', + reason: 'inbound_delivery_manual_flush', + }) + ); expect(writes.filter((write) => write.includes('[drive Alice'))).toHaveLength(paintsBefore); socket.emit('message', jsonMessage({ kind: 'worker_stream', name: 'Alice', chunk: '2J' })); diff --git a/packages/cli/src/cli/lib/attach.ts b/packages/cli/src/cli/lib/attach.ts index 28e1a967a..9f12163e8 100644 --- a/packages/cli/src/cli/lib/attach.ts +++ b/packages/cli/src/cli/lib/attach.ts @@ -603,7 +603,8 @@ export async function syncInitialPtySize( * Read the worker's prior inbound delivery mode and flip it to * `targetMode`. Returns the previous mode on success so the caller can * restore it on detach; returns `null` (and writes an error) when the - * flip fails so the caller bails before touching the terminal. + * request fails or the broker reports that the requested transition did not + * complete, so the caller bails before touching the terminal. * * Non-404 errors are surfaced as `Error: could not ${actionPhrase}: * ${message}` so callers pass verb-appropriate wording (e.g. "switch to @@ -626,6 +627,17 @@ export async function switchInboundDeliveryModeOrAbort( } try { const result = await createBrokerClient(connection, deps.fetch).setInboundDeliveryMode(name, targetMode); + // A manual_flush → auto_inject transition can fail partway through when + // draining a parked message into the harness fails. The broker deliberately + // keeps and reports manual_flush in that case even though the PUT itself + // succeeded. Starting an attach session as if the target mode landed would + // render false state and make its first guarded toggle a no-op. + if (result.mode !== targetMode) { + deps.error( + `Error: could not ${actionPhrase}: broker remained in ${result.mode} mode while draining queued messages` + ); + return null; + } return { previousMode, sessionRevision: result.revision }; } catch (err: unknown) { const failure = mapBrokerSdkFailure(err);