From 7ca5e2dc61c5aa029ed2ec7e85e07dbcfe67916c Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 12:08:25 +0200 Subject: [PATCH 1/9] fix(broker): wait for real prompt before spawn brief --- CHANGELOG.md | 1 + crates/broker/src/pty_worker.rs | 97 +++++++++++++++++++------ crates/broker/src/worker.rs | 18 ++--- tests/e2e/fleet/README.md | 2 +- tests/e2e/fleet/fleet-e2e.test.ts | 103 ++++++++++++++++++++------- tests/e2e/fleet/nodes/node-a.ts | 17 ++++- tests/e2e/fleet/nodes/node-b.ts | 10 ++- tests/e2e/fleet/nodes/stub-agent.cjs | 49 +++++++++++-- 8 files changed, 235 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 37687938f..e256f4a82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Spawned PTY agents keep their initial brief queued until the harness exposes a proven input prompt, preventing slow CLI startup from losing the brief and leaving a registered agent idle. - `agent-relay node agent attach --mode drive` (and `--mode passthrough`) no longer floods the terminal with `input stream send failed: PTY input stream is closed` when the PTY input stream dies mid-session. The loss is now reported once, the stream is reopened with bounded backoff, and if that fails the command exits non-zero with a readable message instead of leaving a session that looks alive but accepts no input. Because attach forwards every byte except `Ctrl+C`/`Ctrl+]` while the stream is healthy, a source TUI with mouse tracking enabled could previously produce this flood from pointer movement alone, without a single keystroke; input is now dropped rather than forwarded for as long as the stream is down. - A reopened attach input stream is verified to belong to the same worker process before any keystroke is forwarded. The stream is reopened by agent name, so without this a replaced worker could silently receive input typed for the session you attached to; the check fails closed when identity cannot be established. diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 069c54c0b..98fe3ae9b 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -149,10 +149,22 @@ fn cli_basename(command: &str) -> &str { .unwrap_or(command) } -const STARTUP_READY_TIMEOUT: Duration = Duration::from_secs(25); +/// Emit one diagnostic when a harness has not reached a proven input prompt in +/// this long. This is deliberately a warning threshold, not a readiness +/// fallback: declaring a booting TUI ready causes its initial task to be typed +/// into startup UI and silently consumed. The broker's independent +/// `WORKER_READY_DEADLINE` reaps a harness that never becomes ready. +const STARTUP_READY_WARNING: Duration = Duration::from_secs(25); const STARTUP_BUFFER_MAX: usize = 12_000; const STARTUP_BUFFER_KEEP: usize = 8_000; const PROMPT_WINDOW_BYTES: usize = 800; + +#[derive(Default)] +struct StartupReadinessState { + ready_sent: bool, + wait_warned: bool, +} + const AGENT_RELAY_BOOT_MARKER: &str = "booting mcp server: agent-relay"; const AGENT_RELAY_SERVER_NAME: &str = "agent-relay"; const LEGACY_RELAY_SERVER_NAME: &str = "relaycast"; @@ -411,32 +423,31 @@ async fn try_emit_worker_ready( child_pid: Option, init_request_id: &mut Option, init_received_at: Option, - worker_ready_sent: &mut bool, + readiness: &mut StartupReadinessState, startup_ready: bool, ) { // init_received_at is Some only after init_worker has been received. // We use it (not init_request_id) as the gate because the broker sends // init_worker without a request_id. - if *worker_ready_sent || init_received_at.is_none() { + if readiness.ready_sent || init_received_at.is_none() { return; } - let timed_out = init_received_at - .map(|started| started.elapsed() >= STARTUP_READY_TIMEOUT) - .unwrap_or(false); - if !startup_ready && !timed_out { + if !startup_ready { + if !readiness.wait_warned + && init_received_at.is_some_and(|started| started.elapsed() >= STARTUP_READY_WARNING) + { + tracing::warn!( + target: "agent_relay::worker::pty", + worker = %worker_name, + warning_secs = STARTUP_READY_WARNING.as_secs(), + "harness prompt not ready yet; preserving queued work until readiness is proven" + ); + readiness.wait_warned = true; + } return; } - if timed_out && !startup_ready { - tracing::warn!( - target: "agent_relay::worker::pty", - worker = %worker_name, - timeout_secs = STARTUP_READY_TIMEOUT.as_secs(), - "startup readiness timed out; emitting worker_ready fallback" - ); - } - let request_id = init_request_id.take(); let _ = send_frame( out_tx, @@ -445,7 +456,7 @@ async fn try_emit_worker_ready( json!({"name": worker_name, "runtime": "pty", "pid": child_pid}), ) .await; - *worker_ready_sent = true; + readiness.ready_sent = true; } pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { @@ -571,7 +582,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut post_boot_output = String::new(); let mut init_request_id: Option = None; let mut init_received_at: Option = None; - let mut worker_ready_sent = false; + let mut startup_readiness = StartupReadinessState::default(); let suppress_multiline_mcp_reminder = cli_basename(&resolved_cli).eq_ignore_ascii_case("agent") || cli_basename(&resolved_cli).eq_ignore_ascii_case("cursor-agent") || cmd.cli.to_ascii_lowercase().contains("cursor"); @@ -756,7 +767,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty.child_pid(), &mut init_request_id, init_received_at, - &mut worker_ready_sent, + &mut startup_readiness, startup_ready, ) .await; @@ -1144,7 +1155,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty.child_pid(), &mut init_request_id, init_received_at, - &mut worker_ready_sent, + &mut startup_readiness, startup_ready, ) .await; @@ -1714,7 +1725,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty.child_pid(), &mut init_request_id, init_received_at, - &mut worker_ready_sent, + &mut startup_readiness, startup_ready, ) .await; @@ -2086,6 +2097,50 @@ mod tests { )); } + #[tokio::test] + async fn startup_warning_preserves_work_until_real_readiness() { + let (tx, mut rx) = mpsc::channel(2); + let mut request_id = None; + let started = Instant::now() - STARTUP_READY_WARNING - Duration::from_secs(1); + let mut readiness = StartupReadinessState::default(); + + try_emit_worker_ready( + &tx, + "slow-worker", + Some(42), + &mut request_id, + Some(started), + &mut readiness, + false, + ) + .await; + + assert!( + readiness.wait_warned, + "slow startup should emit its one diagnostic" + ); + assert!( + !readiness.ready_sent, + "elapsed time is not proof of a ready prompt" + ); + assert!(rx.try_recv().is_err(), "no worker_ready frame may escape"); + + try_emit_worker_ready( + &tx, + "slow-worker", + Some(42), + &mut request_id, + Some(started), + &mut readiness, + true, + ) + .await; + + let frame = rx.try_recv().expect("proven readiness should emit a frame"); + assert_eq!(frame.msg_type, "worker_ready"); + assert!(readiness.ready_sent); + } + #[test] fn should_block_pending_injection_wait_mode_when_suggestion_visible() { let pending = PendingWorkerInjection { diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index bf8710898..bec1edcfd 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -45,12 +45,11 @@ const APP_SERVER_RELEASE_GRACE: Duration = Duration::from_secs(35); /// How long a worker may go without reporting `worker_ready` before the broker /// treats its harness as failed-to-start. /// -/// The worker process emits `worker_ready` itself — the PTY runtime even has a -/// 25s fallback that fires when readiness detection times out -/// (`pty_worker::STARTUP_READY_TIMEOUT`). So silence past this deadline does not -/// mean "slow": it means the worker died, or wedged, before it could report. -/// The margin over that 25s fallback is deliberately generous — reaping a -/// healthy-but-slow agent is far worse than listing a dead one a little longer. +/// The worker process emits `worker_ready` itself, but only after its harness +/// has exposed a proven input prompt. Silence past this deliberately generous +/// deadline means the worker died or wedged before it could become deliverable; +/// reaping a healthy-but-slow agent is far worse than listing a dead one a +/// little longer. const WORKER_READY_DEADLINE: Duration = Duration::from_secs(90); /// Briefly hold the spawn acknowledgement so a wrapper that cannot launch its @@ -2332,9 +2331,10 @@ mod tests { } #[test] - fn the_deadline_clears_the_pty_runtime_startup_fallback() { - // `pty_worker::STARTUP_READY_TIMEOUT` is 25s; the broker must wait - // comfortably longer than the worker's own fallback. + fn the_deadline_allows_slow_pty_startup_before_reaping() { + // The PTY emits a one-shot warning at 25s but keeps waiting for a + // proven prompt. The broker must leave a generous margin before it + // classifies the never-ready wrapper as orphaned. assert!(WORKER_READY_DEADLINE > Duration::from_secs(25) * 3); } } diff --git a/tests/e2e/fleet/README.md b/tests/e2e/fleet/README.md index 72bf1aecb..54438c446 100644 --- a/tests/e2e/fleet/README.md +++ b/tests/e2e/fleet/README.md @@ -21,7 +21,7 @@ engine⇄broker mismatches this E2E surfaced (fixed in relaycast#194). | capability query | `GET /v1/nodes?capability=` returns the right node(s), incl. a shared capability on both | | cross-node dispatch | `echo`→node-a, `ping`→node-b each dispatch over the owning node's control connection and ack | | declarative trigger | a `#general` `/deploy/` message fires the action exactly once; the action-generated reply does **not** re-trigger — asserted by counting the `echo:` **prefix** (a broken guard cascades to `echo:echo:…`, growing the total) | -| spawn completes E2E | targeted spawn mints+injects the agent token, binds the agent via-node, and the node heartbeats the count up — the regression guard for the token-authority handshake | +| spawn completes E2E | five consecutive targeted spawns across both nodes mint+inject agent tokens, bind via-node, wait for proven harness prompts, and cause each PTY child to record its unique brief nonce — registration alone is insufficient | | capability-routed spawn | with no target, placement picks the only node advertising the capability | | scheduled spawn | a shared-capability spawn routes to the least-loaded node (pre-loaded node is skipped) | | resume | a resumable spawn carries `session_ref`; after release, the resume re-targets the **origin** node | diff --git a/tests/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index 2a2700abc..2dda69c8f 100644 --- a/tests/e2e/fleet/fleet-e2e.test.ts +++ b/tests/e2e/fleet/fleet-e2e.test.ts @@ -1,3 +1,5 @@ +import { readFileSync } from 'node:fs'; +import path from 'node:path'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { AgentStream, @@ -373,36 +375,89 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { expect(settled).toBe(before + 1); }, 30_000); - it('spawn completes end-to-end: targeted spawn mints+injects the token, binds the agent via-node, node reports it', async () => { - // Regression guard for the token-authority handshake (engine agent.register - // reply frame ↔ broker). Before the fix this hung to a 30s timeout. - const before = node(await getNodes(engine, workspaceKey), 'node-a')!.active_agents; - const spawn = await invokeAction(engine, driverToken, 'spawn', { - cli: 'claude', - name: 'worker-a', - target_node: 'node-a', - }); - expect(spawn.status).toBe(201); - expect(spawn.body.data.handler_node_id).toBe('node_a'); - - const done = await waitFor( - async () => { - const inv = await getInvocation(engine, driverToken, 'spawn', spawn.invocationId!); - return inv.status === 'completed' || inv.status === 'failed' ? inv : null; - }, - { label: 'spawn settled', timeoutMs: 20_000 } + it('spawn completes end-to-end: targeted spawn delivers its brief after harness readiness and the agent acts', async () => { + // Five consecutive spawns across both nodes prove this is not a lucky + // one-off. Node A's stub delays readiness past the historical 25s fallback + // and discards pre-ready input; node B provides the fast-ready control. + const cases = [ + { agent: 'worker-a', cli: 'claude', nodeName: 'node-a', nodeId: 'node_a', host: nodeA }, + { agent: 'worker-b1', cli: 'codex', nodeName: 'node-b', nodeId: 'node_b', host: nodeB }, + { agent: 'worker-a2', cli: 'claude', nodeName: 'node-a', nodeId: 'node_a', host: nodeA }, + { agent: 'worker-b2', cli: 'codex', nodeName: 'node-b', nodeId: 'node_b', host: nodeB }, + { agent: 'worker-a3', cli: 'claude', nodeName: 'node-a', nodeId: 'node_a', host: nodeA }, + ] as const; + const before = new Map( + (await getNodes(engine, workspaceKey)).map((entry) => [entry.name, entry.active_agents]) ); - expect(done.status).toBe('completed'); // the agent registered + token minted, not a timeout - // The broker bound the agent via-node and heartbeated the count up. + for (const [index, testCase] of cases.entries()) { + const nonce = `spawn-brief-${index + 1}-${Date.now().toString(36)}`; + const observationPath = path.join( + testCase.host.projectDir, + '.agentworkforce', + 'relay', + 'e2e-brief-actions', + `${nonce}.json` + ); + const spawn = await invokeAction(engine, driverToken, 'spawn', { + cli: testCase.cli, + name: testCase.agent, + target_node: testCase.nodeName, + task: `First, act on this brief by recording RELAY_E2E_BRIEF_NONCE=${nonce}`, + }); + expect(spawn.status).toBe(201); + expect(spawn.body.data.handler_node_id).toBe(testCase.nodeId); + + const done = await waitFor( + async () => { + const invocation = await getInvocation(engine, driverToken, 'spawn', spawn.invocationId!); + return invocation.status === 'completed' || invocation.status === 'failed' ? invocation : null; + }, + { label: `${testCase.agent} spawn settled`, timeoutMs: 20_000 } + ); + expect(done.status).toBe('completed'); + + // Registration and heartbeat only prove that a process exists. The nonce + // file is written by the PTY child from the injected task, so it proves + // the brief crossed the harness readiness boundary and caused action. + const observation = await waitFor( + async () => { + try { + return JSON.parse(readFileSync(observationPath, 'utf8')) as { + nonce: string; + agent: string; + node: string; + observedAt: string; + }; + } catch { + return null; + } + }, + { label: `${testCase.agent} acted on nonce-bearing brief`, timeoutMs: 40_000 } + ); + expect(observation).toMatchObject({ + nonce, + agent: testCase.agent, + node: testCase.nodeName, + }); + expect(Number.isNaN(Date.parse(observation.observedAt))).toBe(false); + } + await waitFor( async () => { - const a = node(await getNodes(engine, workspaceKey), 'node-a'); - return a && a.active_agents > before ? a : null; + const nodes = await getNodes(engine, workspaceKey); + const a = node(nodes, 'node-a'); + const b = node(nodes, 'node-b'); + return a && + b && + a.active_agents >= (before.get('node-a') ?? 0) + 3 && + b.active_agents >= (before.get('node-b') ?? 0) + 2 + ? { a, b } + : null; }, - { label: 'node-a active_agents incremented', timeoutMs: 20_000 } + { label: 'both nodes heartbeat all five spawned agents', timeoutMs: 20_000 } ); - }, 45_000); + }, 150_000); it('capability-routed spawn: with no target, placement picks the only node advertising the capability', async () => { const spawn = await invokeAction(engine, driverToken, 'spawn', { cli: 'codex', name: 'worker-codex' }); diff --git a/tests/e2e/fleet/nodes/node-a.ts b/tests/e2e/fleet/nodes/node-a.ts index 7dcd7c23d..75f08c3f3 100644 --- a/tests/e2e/fleet/nodes/node-a.ts +++ b/tests/e2e/fleet/nodes/node-a.ts @@ -15,13 +15,28 @@ import { action, defineNode, spawn } from '@agent-relay/fleet'; */ const stubPath = fileURLToPath(new URL('./stub-agent.cjs', import.meta.url)); const stub = definePtyHarness({ runtime: 'pty', command: process.execPath, args: [stubPath] }); +const delayedReadyStub = definePtyHarness({ + runtime: 'pty', + command: process.execPath, + args: [stubPath], + // Longer than the broker's historical 25s false-readiness fallback. The + // stub discards startup input, so only a delivery after its real prompt can + // produce the nonce observation asserted by the fleet E2E. + env: { + RELAY_E2E_STUB_READY_DELAY_MS: '27000', + RELAY_E2E_NODE_NAME: 'node-a', + // Keep the assertion focused on readiness ordering rather than the generic + // harness's character pacing; real Codex tasks already use bulk injection. + RELAY_INJECT_RATE_MS: '0', + }, +}); const sleepMs = (ms: number) => new Promise((r) => setTimeout(r, ms)); export default defineNode({ name: 'node-a', maxAgents: 8, capabilities: { - 'spawn:claude': spawn(stub), + 'spawn:claude': spawn(delayedReadyStub), 'spawn:pool': spawn(stub), echo: action( { diff --git a/tests/e2e/fleet/nodes/node-b.ts b/tests/e2e/fleet/nodes/node-b.ts index 6646485cc..0f33964ee 100644 --- a/tests/e2e/fleet/nodes/node-b.ts +++ b/tests/e2e/fleet/nodes/node-b.ts @@ -15,7 +15,15 @@ import { action, defineNode, spawn } from '@agent-relay/fleet'; * harness (`stub-agent.cjs`) is a launchable PTY child that idles. */ const stubPath = fileURLToPath(new URL('./stub-agent.cjs', import.meta.url)); -const stub = definePtyHarness({ runtime: 'pty', command: process.execPath, args: [stubPath] }); +const stub = definePtyHarness({ + runtime: 'pty', + command: process.execPath, + args: [stubPath], + env: { + RELAY_E2E_NODE_NAME: 'node-b', + RELAY_INJECT_RATE_MS: '0', + }, +}); const sleepMs = (ms: number) => new Promise((r) => setTimeout(r, ms)); export default defineNode({ diff --git a/tests/e2e/fleet/nodes/stub-agent.cjs b/tests/e2e/fleet/nodes/stub-agent.cjs index 5c45d322e..320500630 100644 --- a/tests/e2e/fleet/nodes/stub-agent.cjs +++ b/tests/e2e/fleet/nodes/stub-agent.cjs @@ -1,14 +1,53 @@ #!/usr/bin/env node 'use strict'; // E2E stub "agent": the broker spawns this as a via-node PTY child after the -// token-authority handshake. It just drains stdin (the broker injects delivered -// messages there, so an undrained pipe could back-pressure delivery) and idles, -// keeping the agent registered via-node for the lifetime of the test. This is a -// proper, launchable PTY child without needing a real AI CLI. +// token-authority handshake. It exposes the same readiness boundary as a real +// interactive harness and records an observable effect when a nonce-bearing +// brief reaches it. Before that boundary it deliberately discards input, which +// models a TUI consuming startup keystrokes before its prompt is ready. +const { mkdirSync, writeFileSync } = require('node:fs'); +const path = require('node:path'); + +const readyDelayMs = Number.parseInt(process.env.RELAY_E2E_STUB_READY_DELAY_MS ?? '0', 10) || 0; +let ready = false; +let input = ''; +const recordedNonces = new Set(); + +function recordBriefNonce(nonce) { + if (recordedNonces.has(nonce)) return; + recordedNonces.add(nonce); + const projectDir = process.env.AGENT_RELAY_PROJECT; + if (!projectDir) return; + const observationDir = path.join(projectDir, '.agentworkforce', 'relay', 'e2e-brief-actions'); + mkdirSync(observationDir, { recursive: true }); + writeFileSync( + path.join(observationDir, `${nonce}.json`), + JSON.stringify({ + nonce, + agent: process.env.RELAY_AGENT_NAME ?? null, + node: process.env.RELAY_E2E_NODE_NAME ?? null, + observedAt: new Date().toISOString(), + }) + ); +} + try { process.stdin.resume(); - process.stdin.on('data', () => {}); + process.stdin.on('data', (chunk) => { + if (!ready) return; + input += chunk.toString(); + for (const match of input.matchAll(/RELAY_E2E_BRIEF_NONCE=([A-Za-z0-9_-]+)/g)) { + recordBriefNonce(match[1]); + } + if (input.length > 32_000) input = input.slice(-16_000); + }); } catch { /* no stdin */ } + +setTimeout(() => { + ready = true; + process.stdout.write('->pty:ready\n'); +}, readyDelayMs); + setInterval(() => {}, 1 << 30); From fc80ee70903dc232ad513083d77f1d35b20438da Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 12:24:08 +0200 Subject: [PATCH 2/9] fix(broker): separate harness liveness from prompt readiness --- crates/broker/src/pty_worker.rs | 44 ++++++++++- crates/broker/src/runtime/worker_events.rs | 90 ++++++++++++++++++++-- crates/broker/src/worker.rs | 12 +-- tests/e2e/fleet/README.md | 2 +- 4 files changed, 133 insertions(+), 15 deletions(-) diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 98fe3ae9b..7ad12d98f 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -152,8 +152,8 @@ fn cli_basename(command: &str) -> &str { /// Emit one diagnostic when a harness has not reached a proven input prompt in /// this long. This is deliberately a warning threshold, not a readiness /// fallback: declaring a booting TUI ready causes its initial task to be typed -/// into startup UI and silently consumed. The broker's independent -/// `WORKER_READY_DEADLINE` reaps a harness that never becomes ready. +/// into startup UI and silently consumed. Harness liveness is reported to the +/// broker independently so a live but unrecognized prompt is not reaped. const STARTUP_READY_WARNING: Duration = Duration::from_secs(25); const STARTUP_BUFFER_MAX: usize = 12_000; const STARTUP_BUFFER_KEEP: usize = 8_000; @@ -459,6 +459,20 @@ async fn try_emit_worker_ready( readiness.ready_sent = true; } +async fn emit_harness_started( + out_tx: &mpsc::Sender>, + worker_name: &str, + child_pid: Option, +) { + let _ = send_frame( + out_tx, + "harness_started", + None, + json!({"name": worker_name, "runtime": "pty", "pid": child_pid}), + ) + .await; +} + pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // Disable Claude Code auto-suggestions to prevent accidental acceptance during injection. #[allow(deprecated)] @@ -752,6 +766,19 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { .unwrap_or_else(|| "pty-worker".to_string()); init_request_id = frame.request_id; init_received_at = Some(Instant::now()); + // Process liveness and safe input readiness are + // separate facts. Report the child pid now so + // the broker can reap a dead harness without + // killing a live one whose prompt takes longer + // than its startup deadline or is unrecognized. + // Initial work remains queued until the later + // worker_ready frame. + emit_harness_started( + &out_tx, + &worker_name, + pty.child_pid(), + ) + .await; let startup_ready = startup_gate_ready( &resolved_cli, &startup_output, @@ -2141,6 +2168,19 @@ mod tests { assert!(readiness.ready_sent); } + #[tokio::test] + async fn harness_liveness_is_reported_without_claiming_input_readiness() { + let (tx, mut rx) = mpsc::channel(1); + + emit_harness_started(&tx, "slow-worker", Some(42)).await; + + let frame = rx.try_recv().expect("harness_started frame"); + assert_eq!(frame.msg_type, "harness_started"); + assert_eq!(frame.payload["name"], "slow-worker"); + assert_eq!(frame.payload["runtime"], "pty"); + assert_eq!(frame.payload["pid"], 42); + } + #[test] fn should_block_pending_injection_wait_mode_when_suggestion_visible() { let pending = PendingWorkerInjection { diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index ae6b76b5b..c0fd698a2 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -34,6 +34,29 @@ fn enqueue_pty_event( } } +fn protocol_pid(value: &Value) -> Option { + value + .get("payload") + .and_then(|payload| payload.get("pid")) + .and_then(Value::as_u64) + .and_then(|pid| u32::try_from(pid).ok()) +} + +fn record_started_harness_pid( + runtime: &AgentRuntime, + harness_pid: &mut Option, + value: &Value, +) -> bool { + if *runtime != AgentRuntime::Pty { + return false; + } + let Some(pid) = protocol_pid(value) else { + return false; + }; + *harness_pid = Some(pid); + true +} + #[cfg(test)] mod pty_observability_tests { use super::*; @@ -109,6 +132,53 @@ mod pty_observability_tests { assert_eq!(hosted.payload["sequence"], 9); assert_eq!(hosted.workspace_id, Some(workspace_id)); } + + #[test] + fn protocol_pid_accepts_only_u32_payload_values() { + assert_eq!(protocol_pid(&json!({"payload": {"pid": 42}})), Some(42)); + assert_eq!( + protocol_pid(&json!({"payload": {"pid": u64::from(u32::MAX) + 1}})), + None + ); + assert_eq!(protocol_pid(&json!({"payload": {"pid": "42"}})), None); + } + + #[test] + fn harness_started_records_liveness_without_readiness() { + let mut harness_pid = None; + assert!(record_started_harness_pid( + &AgentRuntime::Pty, + &mut harness_pid, + &json!({"payload": {"pid": 42}}) + )); + assert_eq!(harness_pid, Some(42)); + + let now = Instant::now(); + let live_pid = std::process::id(); + assert!(record_started_harness_pid( + &AgentRuntime::Pty, + &mut harness_pid, + &json!({"payload": {"pid": live_pid}}) + )); + assert_eq!( + crate::worker::orphaned_worker( + harness_pid, + None, + now - std::time::Duration::from_secs(120), + now, + ), + None, + "reported harness liveness must bypass the never-ready deadline" + ); + + let mut headless_pid = None; + assert!(!record_started_harness_pid( + &AgentRuntime::Headless, + &mut headless_pid, + &json!({"payload": {"pid": 42}}) + )); + assert_eq!(headless_pid, None); + } } pub(super) fn publish_pty_starting( @@ -731,6 +801,19 @@ impl BrokerRuntime { } } let _ = send_event(sdk_out_tx, stream_event).await; + } else if msg_type == "harness_started" { + // A running child process proves liveness but not that + // its TUI is ready for injected input. Record the pid so + // startup maintenance can distinguish a live, slow (or + // unrecognized) prompt from a dead harness. Do not set + // ready_at or release initial_tasks here. + if let Some(handle) = workers.workers.get_mut(&name) { + record_started_harness_pid( + &handle.spec.runtime, + &mut handle.harness_pid, + &value, + ); + } } else if msg_type == "worker_ready" { // If this (re)spawned worker's inbound delivery mode is // already manual_flush — e.g. it crashed and restarted @@ -835,12 +918,7 @@ impl BrokerRuntime { workspace_id, ); } - let payload_pid = value - .get("payload") - .and_then(|p| p.get("pid")) - .and_then(Value::as_u64) - .filter(|pid| *pid <= u32::MAX as u64) - .map(|pid| pid as u32); + let payload_pid = protocol_pid(&value); let (provider_val, cli_val, model_val, session_id_val, pid_val) = workers .workers .get_mut(&name) diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index bec1edcfd..4c953c710 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -46,10 +46,9 @@ const APP_SERVER_RELEASE_GRACE: Duration = Duration::from_secs(35); /// treats its harness as failed-to-start. /// /// The worker process emits `worker_ready` itself, but only after its harness -/// has exposed a proven input prompt. Silence past this deliberately generous -/// deadline means the worker died or wedged before it could become deliverable; -/// reaping a healthy-but-slow agent is far worse than listing a dead one a -/// little longer. +/// has exposed a proven input prompt. PTY wrappers report the child pid earlier, +/// so this deadline only applies when the broker has neither readiness nor +/// separate proof of harness liveness. const WORKER_READY_DEADLINE: Duration = Duration::from_secs(90); /// Briefly hold the spawn acknowledgement so a wrapper that cannot launch its @@ -2333,8 +2332,9 @@ mod tests { #[test] fn the_deadline_allows_slow_pty_startup_before_reaping() { // The PTY emits a one-shot warning at 25s but keeps waiting for a - // proven prompt. The broker must leave a generous margin before it - // classifies the never-ready wrapper as orphaned. + // proven prompt. A reported live child pid bypasses this deadline; + // without either signal, the broker still leaves a generous margin + // before classifying the wrapper as orphaned. assert!(WORKER_READY_DEADLINE > Duration::from_secs(25) * 3); } } diff --git a/tests/e2e/fleet/README.md b/tests/e2e/fleet/README.md index 54438c446..8efa43858 100644 --- a/tests/e2e/fleet/README.md +++ b/tests/e2e/fleet/README.md @@ -63,7 +63,7 @@ BROKER_BINARY_PATH="$PWD/target/release/agent-relay-broker" \ The suite **skips cleanly** (never fails) when prerequisites are missing — the default `npm test` does not run it. The `Fleet E2E` GitHub Actions workflow provisions the engine (pinned to the relaycast#194 SHA) + broker and runs the -full matrix; the matrix itself is ~30s, the wall-clock is build-dominated. +full matrix; the matrix itself is ~2 minutes, the wall-clock is build-dominated. ## Isolation notes From c93e4d4678384cf7294c26e31cf529d601d70f52 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 12:26:57 +0200 Subject: [PATCH 3/9] test(fleet): harden spawn brief regression --- tests/e2e/fleet/fleet-e2e.test.ts | 2 +- tests/e2e/fleet/nodes/stub-agent.cjs | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index 2dda69c8f..5768d0d9b 100644 --- a/tests/e2e/fleet/fleet-e2e.test.ts +++ b/tests/e2e/fleet/fleet-e2e.test.ts @@ -413,7 +413,7 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { const invocation = await getInvocation(engine, driverToken, 'spawn', spawn.invocationId!); return invocation.status === 'completed' || invocation.status === 'failed' ? invocation : null; }, - { label: `${testCase.agent} spawn settled`, timeoutMs: 20_000 } + { label: `${testCase.agent} spawn settled`, timeoutMs: 35_000 } ); expect(done.status).toBe('completed'); diff --git a/tests/e2e/fleet/nodes/stub-agent.cjs b/tests/e2e/fleet/nodes/stub-agent.cjs index 320500630..fe31b8c00 100644 --- a/tests/e2e/fleet/nodes/stub-agent.cjs +++ b/tests/e2e/fleet/nodes/stub-agent.cjs @@ -36,10 +36,12 @@ try { process.stdin.on('data', (chunk) => { if (!ready) return; input += chunk.toString(); - for (const match of input.matchAll(/RELAY_E2E_BRIEF_NONCE=([A-Za-z0-9_-]+)/g)) { + // Require a delimiter after the nonce. PTY chunks can split anywhere, so + // treating the current buffer end as a complete token could record a + // truncated nonce before its remaining characters arrive. + for (const match of input.matchAll(/RELAY_E2E_BRIEF_NONCE=([A-Za-z0-9_-]+)(?=[^A-Za-z0-9_-])/g)) { recordBriefNonce(match[1]); } - if (input.length > 32_000) input = input.slice(-16_000); }); } catch { /* no stdin */ From 6a417840c53ab9865592590c4de7e2b021b27123 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 12:38:39 +0200 Subject: [PATCH 4/9] fix(broker): reject invalid liveness pid --- crates/broker/src/runtime/worker_events.rs | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index c0fd698a2..feb44a79c 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -40,6 +40,7 @@ fn protocol_pid(value: &Value) -> Option { .and_then(|payload| payload.get("pid")) .and_then(Value::as_u64) .and_then(|pid| u32::try_from(pid).ok()) + .filter(|pid| *pid != 0) } fn record_started_harness_pid( @@ -136,6 +137,7 @@ mod pty_observability_tests { #[test] fn protocol_pid_accepts_only_u32_payload_values() { assert_eq!(protocol_pid(&json!({"payload": {"pid": 42}})), Some(42)); + assert_eq!(protocol_pid(&json!({"payload": {"pid": 0}})), None); assert_eq!( protocol_pid(&json!({"payload": {"pid": u64::from(u32::MAX) + 1}})), None @@ -171,6 +173,23 @@ mod pty_observability_tests { "reported harness liveness must bypass the never-ready deadline" ); + let mut zero_pid = None; + assert!(!record_started_harness_pid( + &AgentRuntime::Pty, + &mut zero_pid, + &json!({"payload": {"pid": 0}}) + )); + assert_eq!( + crate::worker::orphaned_worker( + zero_pid, + None, + now - std::time::Duration::from_secs(120), + now, + ), + Some(crate::worker::OrphanedWorker::NeverReady), + "a zero pid must not suppress the never-ready deadline" + ); + let mut headless_pid = None; assert!(!record_started_harness_pid( &AgentRuntime::Headless, From ee7b15442e3bc9d06453cd87e53331a008f55ae9 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 13:04:37 +0200 Subject: [PATCH 5/9] test(fleet): bound stub nonce parser input --- tests/e2e/fleet/nodes/stub-agent.cjs | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/tests/e2e/fleet/nodes/stub-agent.cjs b/tests/e2e/fleet/nodes/stub-agent.cjs index fe31b8c00..02242dcdd 100644 --- a/tests/e2e/fleet/nodes/stub-agent.cjs +++ b/tests/e2e/fleet/nodes/stub-agent.cjs @@ -12,6 +12,9 @@ const readyDelayMs = Number.parseInt(process.env.RELAY_E2E_STUB_READY_DELAY_MS ? let ready = false; let input = ''; const recordedNonces = new Set(); +const nonceMarker = 'RELAY_E2E_BRIEF_NONCE='; +const maxNonceLength = 256; +const noncePattern = new RegExp(`${nonceMarker}([A-Za-z0-9_-]{1,${maxNonceLength}})(?=[^A-Za-z0-9_-])`, 'g'); function recordBriefNonce(nonce) { if (recordedNonces.has(nonce)) return; @@ -39,8 +42,23 @@ try { // Require a delimiter after the nonce. PTY chunks can split anywhere, so // treating the current buffer end as a complete token could record a // truncated nonce before its remaining characters arrive. - for (const match of input.matchAll(/RELAY_E2E_BRIEF_NONCE=([A-Za-z0-9_-]+)(?=[^A-Za-z0-9_-])/g)) { + let consumedThrough = 0; + for (const match of input.matchAll(noncePattern)) { recordBriefNonce(match[1]); + consumedThrough = match.index + match[0].length + 1; + } + if (consumedThrough > 0) input = input.slice(consumedThrough); + + // Keep only a possible partial marker/candidate between chunks. This + // bounds memory and avoids rescanning already-consumed PTY input while + // preserving a nonce whose marker or delimiter straddles a chunk boundary. + const maxCandidateLength = nonceMarker.length + maxNonceLength; + if (input.length > maxCandidateLength) { + const candidateStart = input.lastIndexOf(nonceMarker); + input = + candidateStart >= 0 && input.length - candidateStart <= maxCandidateLength + ? input.slice(candidateStart) + : input.slice(-(nonceMarker.length - 1)); } }); } catch { From 6d6a1a77bdd56b9a99f94f48d44326685127f6ad Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 13:23:24 +0200 Subject: [PATCH 6/9] fix(broker): reject stale worker generation events --- crates/broker/src/runtime/tests.rs | 2 ++ crates/broker/src/runtime/worker_events.rs | 29 +++++++++++++++++++++- crates/broker/src/worker.rs | 18 +++++++++++++- 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 126640603..6425b4136 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -30,6 +30,7 @@ use crate::{ }; use serde_json::{json, Value}; use tokio::sync::mpsc; +use uuid::Uuid; use super::{ apply_exit_after_task_instruction, build_agent_state_transition_event, @@ -85,6 +86,7 @@ async fn make_worker_registry_with_worker(name: &str) -> WorkerRegistry { registry.workers.insert( WorkerName::from(name), WorkerHandle { + generation: Uuid::new_v4(), spec: AgentSpec { name: WorkerName::from(name), runtime: AgentRuntime::Pty, diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index feb44a79c..1d85ad8f1 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -58,6 +58,10 @@ fn record_started_harness_pid( true } +fn worker_event_is_current(current_generation: Option, event_generation: Uuid) -> bool { + current_generation == Some(event_generation) +} + #[cfg(test)] mod pty_observability_tests { use super::*; @@ -145,6 +149,14 @@ mod pty_observability_tests { assert_eq!(protocol_pid(&json!({"payload": {"pid": "42"}})), None); } + #[test] + fn worker_generation_gate_rejects_stale_same_name_events() { + let current = Uuid::new_v4(); + assert!(worker_event_is_current(Some(current), current)); + assert!(!worker_event_is_current(Some(current), Uuid::new_v4())); + assert!(!worker_event_is_current(None, current)); + } + #[test] fn harness_started_records_liveness_without_readiness() { let mut harness_pid = None; @@ -417,7 +429,22 @@ impl BrokerRuntime { let delivery_states = &self.delivery_states; match worker_event { - WorkerEvent::Message { name, value } => { + WorkerEvent::Message { + name, + generation, + value, + } => { + let current_generation = workers.workers.get(&name).map(|handle| handle.generation); + if !worker_event_is_current(current_generation, generation) { + tracing::debug!( + target = "agent_relay::broker", + worker = %name, + event_generation = %generation, + current_generation = ?current_generation, + "ignoring event from stale worker generation" + ); + return; + } if let Some(msg_type) = value.get("type").and_then(Value::as_str) { if msg_type == "delivery_ack" { if let Some(payload) = value.get("payload") { diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index 4c953c710..013ed59b7 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -26,6 +26,7 @@ use tokio::{ sync::mpsc, time::timeout, }; +use uuid::Uuid; use crate::{ cli::command_parse::{normalize_cli_name, parse_cli_command}, @@ -155,6 +156,8 @@ pub(crate) use relay_pty::detection; #[derive(Debug)] pub(crate) struct WorkerHandle { + /// Unique identity for this same-name worker process generation. + pub(crate) generation: Uuid, pub(crate) spec: AgentSpec, pub(crate) parent: Option, pub(crate) workspace_id: Option, @@ -191,7 +194,11 @@ impl AgentWorkState { #[derive(Debug, Clone)] pub(crate) enum WorkerEvent { - Message { name: WorkerName, value: Value }, + Message { + name: WorkerName, + generation: Uuid, + value: Value, + }, } pub(crate) struct WorkerRegistry { @@ -959,9 +966,11 @@ impl WorkerRegistry { let log_file = self.worker_log_path(&spec.name); let startup_log_file = log_file.clone(); + let generation = Uuid::new_v4(); spawn_worker_reader( self.event_tx.clone(), spec.name.clone(), + generation, "stdout", stdout, true, @@ -970,6 +979,7 @@ impl WorkerRegistry { spawn_worker_reader( self.event_tx.clone(), spec.name.clone(), + generation, "stderr", stderr, false, @@ -977,6 +987,7 @@ impl WorkerRegistry { ); let handle = WorkerHandle { + generation, spec: spec.clone(), parent, workspace_id, @@ -1925,6 +1936,7 @@ fn codex_models_json_contains_model(bytes: &[u8], model: &str) -> Option { fn spawn_worker_reader( tx: mpsc::Sender, name: WorkerName, + generation: Uuid, stream_name: &'static str, reader: R, parse_json: bool, @@ -2029,6 +2041,7 @@ fn spawn_worker_reader( if tx .send(WorkerEvent::Message { name: name.clone(), + generation, value, }) .await @@ -2071,6 +2084,7 @@ fn spawn_worker_reader( if tx .send(WorkerEvent::Message { name: name.clone(), + generation, value: fallback, }) .await @@ -2185,6 +2199,7 @@ mod tests { reg.workers.insert( WorkerName::from(name), WorkerHandle { + generation: Uuid::new_v4(), spec: spec_for_test(name), parent: None, workspace_id: None, @@ -2228,6 +2243,7 @@ mod tests { reg.workers.insert( WorkerName::from(name), WorkerHandle { + generation: Uuid::new_v4(), spec: spec_for_test(name), parent: None, workspace_id: None, From 76fbb026f7666d12fcd8e8b62743f783a1619d39 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 13:39:00 +0200 Subject: [PATCH 7/9] fix(broker): clear Codex trust prompt before readiness --- CHANGELOG.md | 1 + crates/broker/src/pty_worker.rs | 1 + crates/broker/src/wrap.rs | 29 ++++++++++++++++++++++++++--- crates/relay-pty/src/terminal.rs | 32 ++++++++++++++++++++++++++++++++ 4 files changed, 60 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e256f4a82..9ed04cb26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Spawned PTY agents keep their initial brief queued until the harness exposes a proven input prompt, preventing slow CLI startup from losing the brief and leaving a registered agent idle. +- Spawned Codex PTY agents accept the CLI's directory-trust interstitial before readiness, so a queued initial brief reaches the real input prompt instead of remaining parked behind the startup menu. - `agent-relay node agent attach --mode drive` (and `--mode passthrough`) no longer floods the terminal with `input stream send failed: PTY input stream is closed` when the PTY input stream dies mid-session. The loss is now reported once, the stream is reopened with bounded backoff, and if that fails the command exits non-zero with a readable message instead of leaving a session that looks alive but accepts no input. Because attach forwards every byte except `Ctrl+C`/`Ctrl+]` while the stream is healthy, a source TUI with mouse tracking enabled could previously produce this flood from pointer movement alone, without a single keystroke; input is now dropped rather than forwarded for as long as the stream is down. - A reopened attach input stream is verified to belong to the same worker process before any keystroke is forwarded. The stream is reopened by agent name, so without this a replaced worker could silently receive input typed for the session you attached to; the check fails closed when identity cannot be established. diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 7ad12d98f..d1fff26fc 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -1244,6 +1244,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty_auto.handle_mcp_approval(&text, &pty).await; pty_auto.handle_bypass_permissions(&text, &pty).await; pty_auto.handle_codex_model_prompt(&text, &pty).await; + pty_auto.handle_codex_trust(&text, &pty).await; pty_auto.handle_opencode_permission(&text, &pty).await; pty_auto.handle_gemini_action(&text, &pty).await; pty_auto.handle_gemini_untrusted_banner(&text, &pty).await; diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 343c0de9a..73aa26af3 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -37,9 +37,9 @@ use crate::util::{ ansi::{floor_char_boundary, strip_ansi, AnsiStripper}, terminal::{ detect_bypass_permissions_prompt, detect_claude_trust_prompt, detect_codex_model_prompt, - detect_gemini_action_required, detect_gemini_trust_prompt, detect_gemini_untrusted_banner, - detect_opencode_permission_prompt, is_auto_suggestion, is_bypass_selection_menu, - is_in_editor_mode, + detect_codex_trust_prompt, detect_gemini_action_required, detect_gemini_trust_prompt, + detect_gemini_untrusted_banner, detect_opencode_permission_prompt, is_auto_suggestion, + is_bypass_selection_menu, is_in_editor_mode, }, }; use crate::worker::detection::ActivityDetector; @@ -153,6 +153,9 @@ pub(crate) struct PtyAutoState { // Codex model upgrade prompt pub(crate) codex_model_prompt_handled: bool, pub(crate) codex_model_buffer: String, + // Codex directory trust prompt + pub(crate) codex_trust_buffer: String, + pub(crate) codex_trust_handled: bool, // Opencode/droid EXECUTE permission prompt pub(crate) opencode_perm_buffer: String, pub(crate) last_opencode_perm_approval: Option, @@ -197,6 +200,8 @@ impl PtyAutoState { bypass_perms_send_count: 0, codex_model_prompt_handled: false, codex_model_buffer: String::new(), + codex_trust_buffer: String::new(), + codex_trust_handled: false, opencode_perm_buffer: String::new(), last_opencode_perm_approval: None, gemini_action_buffer: String::new(), @@ -333,6 +338,23 @@ impl PtyAutoState { } } + /// Detect and accept Codex's startup directory-trust prompt. + /// "Yes, continue" is pre-selected as option 1, so Enter is sufficient. + pub(crate) async fn handle_codex_trust(&mut self, text: &str, pty: &PtySession) { + if self.interactive_hold || self.codex_trust_handled { + return; + } + Self::append_buf(&mut self.codex_trust_buffer, text, 2500, 2000); + let clean = strip_ansi(&self.codex_trust_buffer); + if detect_codex_trust_prompt(&clean) { + tracing::info!("Detected Codex directory trust prompt, auto-accepting"); + tokio::time::sleep(Duration::from_millis(100)).await; + warn_on_auto_response_write(pty.submit_write(b"\r".to_vec()), "codex_trust"); + self.codex_trust_buffer.clear(); + self.codex_trust_handled = true; + } + } + /// Detect and auto-approve opencode/droid EXECUTE permission prompts. /// Selects "Yes, and always allow medium impact commands" (arrow down + Enter). pub(crate) async fn handle_opencode_permission(&mut self, text: &str, pty: &PtySession) { @@ -1217,6 +1239,7 @@ pub(crate) async fn run_wrap( pty_auto.handle_mcp_approval(&text, &pty).await; pty_auto.handle_bypass_permissions(&text, &pty).await; pty_auto.handle_codex_model_prompt(&text, &pty).await; + pty_auto.handle_codex_trust(&text, &pty).await; pty_auto.handle_opencode_permission(&text, &pty).await; pty_auto.handle_gemini_action(&text, &pty).await; pty_auto.handle_gemini_untrusted_banner(&text, &pty).await; diff --git a/crates/relay-pty/src/terminal.rs b/crates/relay-pty/src/terminal.rs index 57fbd4cc5..31662861b 100644 --- a/crates/relay-pty/src/terminal.rs +++ b/crates/relay-pty/src/terminal.rs @@ -86,6 +86,18 @@ pub fn detect_codex_model_prompt(clean_output: &str) -> (bool, bool) { (has_upgrade_ref, has_model_options) } +/// Detect Codex's startup directory-trust prompt. +/// +/// Codex shows this interstitial before its normal input prompt when the +/// selected working directory has not been trusted yet. A spawned worker +/// cannot receive its queued initial task until this menu is dismissed. +pub fn detect_codex_trust_prompt(clean_output: &str) -> bool { + let lower = clean_output.to_lowercase(); + lower.contains("do you trust the contents of this directory?") + && lower.contains("yes, continue") + && lower.contains("no, quit") +} + /// Detect opencode/droid EXECUTE permission prompt in output. /// Returns (has_header, has_allow_option). /// The prompt looks like: @@ -187,6 +199,26 @@ mod tests { assert!(has_options); } + #[test] + fn codex_directory_trust_prompt() { + let output = + "Do you trust the contents of this directory? Working with untrusted contents\n\ + comes with higher risk of prompt injection.\n\ + > 1. Yes, continue\n\ + 2. No, quit"; + assert!(detect_codex_trust_prompt(output)); + } + + #[test] + fn codex_directory_trust_requires_the_complete_menu() { + assert!(!detect_codex_trust_prompt( + "Do you trust the contents of this directory?" + )); + assert!(!detect_codex_trust_prompt( + "The agent said yes, continue, then no, quit." + )); + } + #[test] fn gemini_action_required_allow_once() { let output = "⚠ Action Required\nThe tool wants to execute a command.\nAllow once\nDeny"; From 32229bed21766ed55f4dab28af829948e75d2975 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 13:45:22 +0200 Subject: [PATCH 8/9] fix(broker): inspect Codex trust screen grid --- crates/broker/src/wrap.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 73aa26af3..ccff69171 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -346,7 +346,11 @@ impl PtyAutoState { } Self::append_buf(&mut self.codex_trust_buffer, text, 2500, 2000); let clean = strip_ansi(&self.codex_trust_buffer); - if detect_codex_trust_prompt(&clean) { + // Codex redraws this TUI with cursor motion, so the raw byte stream can + // spell only fragments even though the terminal grid contains the + // complete menu. Check both representations, just like readiness does. + let visible_screen = pty.screen_text(); + if detect_codex_trust_prompt(&clean) || detect_codex_trust_prompt(&visible_screen) { tracing::info!("Detected Codex directory trust prompt, auto-accepting"); tokio::time::sleep(Duration::from_millis(100)).await; warn_on_auto_response_write(pty.submit_write(b"\r".to_vec()), "codex_trust"); From 8e6ca1349cc4522423daddbebe227ec5e7855161 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 13:57:15 +0200 Subject: [PATCH 9/9] fix(broker): keep Codex trust menu behind readiness gate --- crates/broker/src/pty_worker.rs | 35 ++++++++++++++++++++++++++++++++- 1 file changed, 34 insertions(+), 1 deletion(-) diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index d1fff26fc..31a23030a 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -37,6 +37,7 @@ use crate::readiness::{cli_prompt_ready, detect_cli_ready, GridReadinessSnapshot use crate::runtime::{get_terminal_size, send_frame}; use crate::snapshot::Snapshot; use crate::util::ansi::{floor_char_boundary, strip_ansi, AnsiStripper}; +use crate::util::terminal::detect_codex_trust_prompt; use crate::util::utf8_stream::Utf8StreamDecoder; use crate::worker::detection::ActivityDetector; use crate::wrap::{warn_on_auto_response_write, PtyAutoState, AUTO_SUGGESTION_BLOCK_TIMEOUT}; @@ -267,6 +268,15 @@ fn evaluate_startup_gate( post_boot_output: &str, grid: GridReadinessSnapshot<'_>, ) -> bool { + // A menu-selection glyph is not the harness input prompt. In particular, + // Codex's directory-trust interstitial contains the same `›` glyph as its + // composer, so the generic prompt detector would otherwise release the + // queued brief before the auto-responder's Enter takes effect. Every gate + // path (output, init, and timer tick) passes through this exclusion. + if detect_codex_trust_prompt(grid.screen) { + return false; + } + if wait_for_agent_relay_boot { saw_agent_relay_boot && output_has_prompt(resolved_cli, post_boot_output) @@ -1167,6 +1177,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { last_context_low_pct = Some(pct); } } + // Clear startup interstitials before evaluating the + // prompt. The gate below still explicitly rejects the + // trust screen until Codex redraws its real composer. + pty_auto.handle_codex_trust(&text, &pty).await; let startup_ready = startup_gate_ready( &resolved_cli, &startup_output, @@ -1244,7 +1258,6 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty_auto.handle_mcp_approval(&text, &pty).await; pty_auto.handle_bypass_permissions(&text, &pty).await; pty_auto.handle_codex_model_prompt(&text, &pty).await; - pty_auto.handle_codex_trust(&text, &pty).await; pty_auto.handle_opencode_permission(&text, &pty).await; pty_auto.handle_gemini_action(&text, &pty).await; pty_auto.handle_gemini_untrusted_banner(&text, &pty).await; @@ -2125,6 +2138,26 @@ mod tests { )); } + #[test] + fn startup_gate_rejects_codex_directory_trust_menu() { + let trust_screen = "Do you trust the contents of this directory?\n\ + › 1. Yes, continue\n\ + 2. No, quit\n\ + Press enter to continue"; + assert!(!evaluate_startup_gate( + "codex", + trust_screen, + 600, + false, + false, + "", + GridReadinessSnapshot { + screen: trust_screen, + cursor: Some((2, 1)), + }, + )); + } + #[tokio::test] async fn startup_warning_preserves_work_until_real_readiness() { let (tx, mut rx) = mpsc::channel(2);