From d4b6b93a4cf65d67406fabcd7c4d5a7396762983 Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 19:10:19 +0530 Subject: [PATCH 1/6] fix(broker): confirm PTY deliveries through native MCP acknowledgements --- CHANGELOG.md | 6 +- .../src/broker/delivery_verification.rs | 41 +++++ crates/broker/src/broker/injection_format.rs | 7 +- crates/broker/src/lib.rs | 1 + crates/broker/src/pty_delivery_ack.rs | 153 ++++++++++++++++++ crates/broker/src/pty_worker.rs | 27 +++- crates/broker/src/wrap.rs | 23 ++- .../broker/tests/pty_injection_integrity.rs | 58 ++++++- packages/cli/src/cli/mcp/messaging-tools.ts | 28 ++++ .../cli/src/cli/mcp/pty-delivery-ack.test.ts | 100 ++++++++++++ packages/cli/src/cli/mcp/pty-delivery-ack.ts | 23 +++ packages/cli/src/cli/mcp/telemetry.ts | 1 + tests/e2e/fleet/nodes/stub-agent.cjs | 27 +++- .../run.mjs | 18 ++- .../cases/1961-native-pty-ack/case.json | 15 ++ .../cases/1961-native-pty-ack/run.mjs | 133 +++++++++++++++ 16 files changed, 628 insertions(+), 33 deletions(-) create mode 100644 crates/broker/src/pty_delivery_ack.rs create mode 100644 packages/cli/src/cli/mcp/pty-delivery-ack.test.ts create mode 100644 packages/cli/src/cli/mcp/pty-delivery-ack.ts create mode 100644 tests/relayflows/cases/1961-native-pty-ack/case.json create mode 100644 tests/relayflows/cases/1961-native-pty-ack/run.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index cd3e446790..eb93f6a8d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- PTY delivery confirmation now uses recipient MCP read receipts; terminal repaints alone leave delivery unconfirmed. ## [14.0.0] - 2026-10-13 diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 0ceb1dd080..2117592b08 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -109,6 +109,7 @@ pub(crate) const MAX_ACCEPTANCE_LIFETIME: std::time::Duration = std::time::Durat /// A pending delivery waiting for echo verification in PTY output. #[derive(Debug)] pub(crate) struct PendingVerification { + pub native_ack: Option, pub delivery_id: DeliveryId, pub event_id: EventId, pub expected_echo: String, @@ -143,6 +144,22 @@ pub(crate) struct PendingVerification { } impl PendingVerification { + /// Native receipts can complete immediately. Known MCP harnesses get the + /// full acceptance budget to read, while generic harnesses keep the short fallback. + pub(crate) fn ready_to_assess(&self, cli: &str, window: Duration) -> bool { + self.native_ack.as_ref().is_some_and(|ack| ack.confirmed()) + || if self.native_ack.is_some() + && matches!( + crate::cli::command_parse::normalize_cli_name(cli).as_str(), + "claude" | "codex" + ) + { + self.acceptance_expired() + } else { + self.injected_at.elapsed() >= window + } + } + pub(crate) fn observe_raw_process_receipt(&mut self, output: &VerificationOutput, cli: &str) { if self.echo_seen || !is_cat_process(cli) || self.body.is_empty() { return; @@ -603,6 +620,25 @@ pub(crate) fn assess_harness_acceptance( HarnessAcceptance::Inconclusive } +/// Recipient read receipts prove acceptance; terminal output only guides recovery. +pub(crate) fn assess_delivery_acceptance( + cli: &str, + verification: &PendingVerification, + snapshot: &Snapshot, +) -> HarnessAcceptance { + if verification + .native_ack + .as_ref() + .is_some_and(|ack| ack.confirmed()) + { + return HarnessAcceptance::Accepted("native_mcp_ack".to_string()); + } + match assess_harness_acceptance(cli, verification, snapshot) { + HarnessAcceptance::Parked => HarnessAcceptance::Parked, + _ => HarnessAcceptance::Inconclusive, + } +} + #[derive(Debug)] struct VerificationSegment { sequence: u64, @@ -819,6 +855,7 @@ mod tests { fn codex_verification(expected: &str) -> PendingVerification { PendingVerification { + native_ack: None, delivery_id: "delivery-acceptance".into(), event_id: "event-acceptance".into(), expected_echo: expected.to_string(), @@ -1599,6 +1636,7 @@ mod tests { output.push_str(expected); let output_boundary = output.boundary(); let verification = PendingVerification { + native_ack: None, delivery_id: "delivery-repeat".into(), event_id: "evt-repeat".into(), expected_echo: expected.to_string(), @@ -1636,6 +1674,7 @@ mod tests { // producer watermark rather than the consumer's older boundary. let output_boundary = 2; let verification = PendingVerification { + native_ack: None, delivery_id: "delivery-queued".into(), event_id: "evt-queued".into(), expected_echo: expected.to_string(), @@ -1694,6 +1733,7 @@ mod tests { output.push_str(expected); output.push_str("\nTool: Write(review.md)\n"); let verification = PendingVerification { + native_ack: None, delivery_id: "delivery-activity".into(), event_id: "evt-activity".into(), expected_echo: expected.to_string(), @@ -1735,6 +1775,7 @@ mod tests { let output_boundary = output.boundary(); output.push_str(expected); let verification = PendingVerification { + native_ack: None, delivery_id: "delivery-pending".into(), event_id: "evt-pending".into(), expected_echo: expected.to_string(), diff --git a/crates/broker/src/broker/injection_format.rs b/crates/broker/src/broker/injection_format.rs index 0d3ac2e15f..b9bbd605e4 100644 --- a/crates/broker/src/broker/injection_format.rs +++ b/crates/broker/src/broker/injection_format.rs @@ -297,13 +297,18 @@ pub(crate) fn format_injection_for_worker_with_workspace( ) }; + let read_hint = format!("Before acting on this message, acknowledge receipt with mcp__agent-relay__acknowledge_delivery or agent-relay.acknowledge_delivery (message_id: {}).", serde_json::to_string(event_id).expect("message id is serializable")); if !include_reminder { let short_hint = build_mcp_short_hint(from, target, &relay_line, pre_registered, assigned_name); - return format!("{short_hint}\n{relay_line}"); + return format!("{short_hint}\n{read_hint}\n{relay_line}"); } let mut reminder = build_mcp_reminder(from, target, &relay_line, pre_registered, assigned_name); + reminder = reminder.replace( + "", + &format!("- {read_hint}\n"), + ); if let Some(label) = workspace_label { reminder = reminder.replace( "", diff --git a/crates/broker/src/lib.rs b/crates/broker/src/lib.rs index 14193d4605..3501b6e148 100644 --- a/crates/broker/src/lib.rs +++ b/crates/broker/src/lib.rs @@ -35,6 +35,7 @@ pub(crate) mod node_delivery_probe; #[allow(dead_code)] pub(crate) mod obligation; pub(crate) mod priorities; +pub(crate) mod pty_delivery_ack; pub(crate) mod pty_worker; #[allow(dead_code)] pub(crate) mod redact; diff --git a/crates/broker/src/pty_delivery_ack.rs b/crates/broker/src/pty_delivery_ack.rs new file mode 100644 index 0000000000..1e5b18d26d --- /dev/null +++ b/crates/broker/src/pty_delivery_ack.rs @@ -0,0 +1,153 @@ +//! Recipient-originated MCP read receipts for PTY deliveries. +//! +//! Each worker owns a private, short-lived directory inherited by its harness +//! and MCP subprocesses. A fresh nonce fences each delivery against stale reads. +use std::{fs, io, path::PathBuf}; + +use serde_json::json; +use sha2::{Digest, Sha256}; + +pub(crate) const ACK_DIR_ENV: &str = "RELAY_PTY_ACK_DIR"; + +pub(crate) struct NativeAckInbox(tempfile::TempDir); + +impl NativeAckInbox { + pub(crate) fn new() -> io::Result { + let dir = tempfile::Builder::new() + .prefix("relay-pty-acks-") + .tempdir()?; + Ok(Self(dir)) + } + + pub(crate) fn expose_to_child(&self) { + // PTY workers are dedicated processes; only their descendants inherit this path. + std::env::set_var(ACK_DIR_ENV, self.0.path()); + } + + /// MCP subprocesses do not reliably inherit the harness environment. + /// Forward only this worker's acknowledgement directory in their existing config. + pub(crate) fn configure_harness(&self, cli: &str, args: &mut Vec) -> io::Result<()> { + let cli = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); + let directory = self.0.path().to_string_lossy().to_string(); + if cli == "codex" || cli == "codex.exe" { + args.extend([ + "--config".into(), + format!( + "mcp_servers.agent-relay.env.{ACK_DIR_ENV}={}", + serde_json::to_string(&directory)? + ), + ]); + } else if cli == "claude" || cli == "claude.exe" { + for index in 1..args.len() { + if args[index - 1] != "--mcp-config" { + continue; + } + let raw = if args[index].trim_start().starts_with('{') { + args[index].clone() + } else { + fs::read_to_string(&args[index])? + }; + let mut config: serde_json::Value = serde_json::from_str(&raw)?; + if let Some(servers) = config + .get_mut("mcpServers") + .and_then(|value| value.as_object_mut()) + { + for name in ["agent-relay", "relaycast"] { + if let Some(server) = servers + .get_mut(name) + .and_then(|value| value.as_object_mut()) + { + let env = server.entry("env").or_insert_with(|| json!({})); + if let Some(env) = env.as_object_mut() { + env.insert(ACK_DIR_ENV.into(), json!(directory)); + } + } + } + } + // Replace the argument, never the user's config file. + args[index] = config.to_string(); + } + } + Ok(()) + } + + pub(crate) fn register(&self, event_id: &str) -> io::Result { + let key = format!("{:x}", Sha256::digest(event_id.as_bytes())); + let pending = self.0.path().join(format!("{key}.pending")); + let nonce = uuid::Uuid::new_v4().to_string(); + let receipt = self.0.path().join(format!("{key}.{nonce}.read")); + fs::write( + &pending, + json!({"messageId": event_id, "nonce": nonce}).to_string(), + )?; + Ok(NativeAck { + pending, + receipt, + nonce, + }) + } +} + +#[derive(Debug)] +pub(crate) struct NativeAck { + pending: PathBuf, + receipt: PathBuf, + nonce: String, +} + +impl NativeAck { + pub(crate) fn confirmed(&self) -> bool { + fs::read_to_string(&self.receipt).is_ok_and(|value| value == self.nonce) + } +} + +impl Drop for NativeAck { + fn drop(&mut self) { + let _ = fs::remove_file(&self.pending); + let _ = fs::remove_file(&self.receipt); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn native_receipt_requires_this_delivery_nonce_and_is_cleaned_up() { + // Construct directly so parallel tests do not mutate process environment. + let inbox = NativeAckInbox(tempfile::tempdir().unwrap()); + let ack = inbox.register("msg/../one").unwrap(); + assert!(!ack.confirmed()); + fs::write(&ack.receipt, "old-or-wrong-nonce").unwrap(); + assert!(!ack.confirmed()); + fs::write(&ack.receipt, &ack.nonce).unwrap(); + assert!(ack.confirmed()); + let pending = ack.pending.clone(); + let receipt = ack.receipt.clone(); + drop(ack); + assert!(!pending.exists()); + assert!(!receipt.exists()); + assert!(!inbox.register("msg/../one").unwrap().confirmed()); + } + + #[test] + fn native_ack_directory_is_forwarded_to_claude_and_codex_mcp() { + let inbox = NativeAckInbox::new().unwrap(); + let mut claude = vec!["--mcp-config".into(), json!({"mcpServers": {"agent-relay": {"command": "agent-relay", "args": ["mcp"], "env": {"RELAY_AGENT_NAME": "worker"}}, "other": {"command": "other"}}}).to_string()]; + inbox.configure_harness("claude", &mut claude).unwrap(); + let config: serde_json::Value = serde_json::from_str(&claude[1]).unwrap(); + assert_eq!( + config["mcpServers"]["agent-relay"]["env"][ACK_DIR_ENV], + inbox.0.path().to_str().unwrap() + ); + assert_eq!( + config["mcpServers"]["agent-relay"]["env"]["RELAY_AGENT_NAME"], + "worker" + ); + assert!(config["mcpServers"]["other"].get("env").is_none()); + let mut codex = vec!["--model".into(), "test-model".into()]; + inbox.configure_harness("codex", &mut codex).unwrap(); + assert!(codex[3].starts_with("mcp_servers.agent-relay.env.RELAY_PTY_ACK_DIR=")); + assert_eq!(&codex[..2], ["--model", "test-model"]); + } +} diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 21f2f9ce7c..34440eefae 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -28,7 +28,7 @@ use tokio::{ use crate::broker::{ continuity::parse_continuity_command, delivery_verification::{ - assess_harness_acceptance, current_timestamp_ms, delivery_injected_event_payload, + assess_delivery_acceptance, current_timestamp_ms, delivery_injected_event_payload, delivery_queued_event_payload, draft_visible_at_cursor, failed_draft_released, DeliveryOutcome, HarnessAcceptance, PendingActivity, PendingVerification, ThrottleState, VerificationOutput, ACTIVITY_BUFFER_KEEP_BYTES, ACTIVITY_BUFFER_MAX_BYTES, ACTIVITY_WINDOW, @@ -262,6 +262,7 @@ enum InjectionStage { /// be used as the expected echo when the submit command is confirmed). #[derive(Debug)] struct ActiveInjection { + native_ack: Option, pending: PendingWorkerInjection, stage: InjectionStage, prompt_wait_started: Instant, @@ -1126,6 +1127,9 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut effective_args = inline_cli_args; effective_args.extend(cmd.args.clone()); + let native_acks = crate::pty_delivery_ack::NativeAckInbox::new()?; + native_acks.expose_to_child(); + native_acks.configure_harness(&resolved_cli, &mut effective_args)?; let _devin_state = crate::devin::prepare_worker_config(&resolved_cli).await?; let (init_rows, init_cols) = get_terminal_size().unwrap_or((24, 80)); let (pty, mut pty_rx) = @@ -2092,7 +2096,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); pv.observe_visible_composer(&snapshot, &resolved_cli); if let HarnessAcceptance::Accepted(evidence) = - assess_harness_acceptance(&resolved_cli, pv, &snapshot) + assess_delivery_acceptance(&resolved_cli, pv, &snapshot) { verified_indices.push((i, evidence)); } @@ -2321,6 +2325,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let hold_exempt = targeted_hold_exemption || hold_exempt_injections > 0; hold_exempt_injections = hold_exempt_injections.saturating_sub(1); active_injection = Some(ActiveInjection { + native_ack: None, pending, stage: InjectionStage::Escape, prompt_wait_started: Instant::now(), @@ -2424,6 +2429,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty_auto.auto_suggestion_visible = false; let include_mcp_reminder = !suppress_multiline_mcp_reminder && mcp_reminder_throttle.should_include(Instant::now()); + inj.native_ack = Some(native_acks.register(&inj.pending.delivery.event_id)?); let injection = format_injection_for_worker_with_workspace( &inj.pending.delivery.from, &inj.pending.delivery.event_id, @@ -2621,6 +2627,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // composer before broker custody is released. let injection = inj.injection_text.take().unwrap_or_default(); let mut verification = PendingVerification { + native_ack: inj.native_ack.take(), delivery_id: inj.pending.delivery.delivery_id.clone(), event_id: inj.pending.delivery.event_id.clone(), expected_echo: injection, @@ -2645,7 +2652,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { .since(verification.output_boundary) .into_owned(); verification.observe(&echo_buffer, &raced_output); - let acceptance = assess_harness_acceptance( + let acceptance = assess_delivery_acceptance( &resolved_cli, &verification, &Snapshot::capture(&pty), @@ -2762,7 +2769,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); let snapshot = Snapshot::capture(&pty); pv.observe_visible_composer(&snapshot, &resolved_cli); - match assess_harness_acceptance( + match assess_delivery_acceptance( &resolved_cli, &pv, &snapshot, @@ -2908,14 +2915,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut i = 0; while i < pending_verifications.len() { - if pending_verifications[i].injected_at.elapsed() >= verification_window { + if pending_verifications[i].ready_to_assess(&resolved_cli, verification_window) { let mut pv = pending_verifications.remove(i).unwrap(); let delivery_id = pv.delivery_id.clone(); let event_id = pv.event_id.clone(); pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); let snapshot = Snapshot::capture(&pty); pv.observe_visible_composer(&snapshot, &resolved_cli); - match assess_harness_acceptance( + match assess_delivery_acceptance( &resolved_cli, &pv, &snapshot, @@ -3348,10 +3355,12 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { #[cfg(test)] mod tests { use super::*; + use crate::broker::delivery_verification::assess_harness_acceptance; #[test] fn post_acceptance_activity_is_seeded_only_from_post_echo_output() { let verification = PendingVerification { + native_ack: None, delivery_id: "delivery-current".into(), event_id: "event-current".into(), expected_echo: "current request".into(), @@ -3828,6 +3837,7 @@ mod tests { .expect("PTY drainer exited") .expect("initial injection write failed"); let mut verification = PendingVerification { + native_ack: None, delivery_id: "live-codex-delivery".into(), event_id: "live-codex-event".into(), expected_echo: body.to_string(), @@ -4003,6 +4013,7 @@ mod tests { .unwrap(); ack_rx.await.unwrap().unwrap(); let mut verification = PendingVerification { + native_ack: None, delivery_id: "live-codex-parked".into(), event_id: "live-codex-parked-event".into(), expected_echo: body.to_string(), @@ -4936,6 +4947,7 @@ mod tests { let mut count = 0usize; let mut event_ids = HashSet::new(); let targeted = ActiveInjection { + native_ack: None, pending: test_pending_injection("init_task"), stage: InjectionStage::Escape, prompt_wait_started: Instant::now(), @@ -4952,6 +4964,7 @@ mod tests { // A blanket exemption still goes back to the counter. let blanket = ActiveInjection { + native_ack: None, targeted_hold_exemption: false, ..targeted }; @@ -4962,6 +4975,7 @@ mod tests { // An injection popped without any exemption restores nothing. let unexempt = ActiveInjection { + native_ack: None, hold_exempt: false, ..blanket }; @@ -4977,6 +4991,7 @@ mod tests { // requeue would then hand the credit to whatever else is queued instead // of back to this delivery. let mut injection = ActiveInjection { + native_ack: None, pending: test_pending_injection("init_task"), stage: InjectionStage::Escape, prompt_wait_started: Instant::now(), diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index d23ff3e783..8c3195b5da 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -27,9 +27,9 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use crate::broker::{ delivery_verification::{ - assess_harness_acceptance, draft_visible_at_cursor, failed_draft_released, DeliveryOutcome, - HarnessAcceptance, PendingActivity, PendingVerification, ThrottleState, VerificationOutput, - ACTIVITY_BUFFER_KEEP_BYTES, ACTIVITY_BUFFER_MAX_BYTES, ACTIVITY_WINDOW, + assess_delivery_acceptance, draft_visible_at_cursor, failed_draft_released, + DeliveryOutcome, HarnessAcceptance, PendingActivity, PendingVerification, ThrottleState, + VerificationOutput, ACTIVITY_BUFFER_KEEP_BYTES, ACTIVITY_BUFFER_MAX_BYTES, ACTIVITY_WINDOW, MAX_VERIFICATION_ATTEMPTS, VERIFICATION_WINDOW, }, injection_format::{format_injection_for_worker_with_workspace, McpReminderThrottle}, @@ -272,6 +272,7 @@ pub(crate) struct PendingWrapInjection { /// and its harness-specific submit sequence were written and flushed. enum PendingWrapWrite { Initial { + native_ack: crate::pty_delivery_ack::NativeAck, pending: PendingWrapInjection, injection: String, output_boundary: u64, @@ -425,7 +426,7 @@ fn queue_or_confirm_wrap_verification( verification.observe_raw_process_receipt(output, resolved_cli); verification.observe_visible_composer(&Snapshot::capture(pty), resolved_cli); if let HarnessAcceptance::Accepted(evidence) = - assess_harness_acceptance(resolved_cli, &verification, &Snapshot::capture(pty)) + assess_delivery_acceptance(resolved_cli, &verification, &Snapshot::capture(pty)) { tracing::info!( target = "agent_relay::worker::wrap", @@ -1534,6 +1535,9 @@ pub(crate) async fn run_wrap( std::env::set_var("RELAY_BASE_URL", base); } } + let native_acks = crate::pty_delivery_ack::NativeAckInbox::new()?; + native_acks.expose_to_child(); + native_acks.configure_harness(&resolved_cli, &mut effective_cli_args)?; let _devin_state = crate::devin::prepare_worker_config(&resolved_cli).await?; // --- Spawn CLI in PTY --- @@ -1838,7 +1842,7 @@ pub(crate) async fn run_wrap( pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); pv.observe_visible_composer(&snapshot, &resolved_cli); if let HarnessAcceptance::Accepted(evidence) = - assess_harness_acceptance(&resolved_cli, pv, &snapshot) + assess_delivery_acceptance(&resolved_cli, pv, &snapshot) { verified_indices.push((i, evidence)); } @@ -2321,10 +2325,12 @@ pub(crate) async fn run_wrap( pending.workspace_alias.as_deref(), ); let bytes = crate::injection_wire::injection_bytes(crate::injection_wire::injection_wire(&resolved_cli, &pty), &injection); + let native_ack = native_acks.register(&pending.event_id)?; let write = submit_injection_body(&pty, &resolved_cli, bytes, Duration::ZERO); match write { Ok((ack_rx, output_boundary)) => { let pending_write = PendingWrapWrite::Initial { + native_ack, pending, injection, output_boundary, @@ -2370,6 +2376,7 @@ pub(crate) async fn run_wrap( match pending_write { PendingWrapWrite::Initial { + native_ack, pending, injection, output_boundary, @@ -2411,6 +2418,7 @@ pub(crate) async fn run_wrap( ); pty_auto.note_completed_injection(&resolved_cli); let verification = PendingVerification { + native_ack: Some(native_ack), delivery_id: DeliveryId::new(format!("wrap_{}", pending.event_id)), event_id: pending.event_id, expected_echo: injection, @@ -2494,12 +2502,12 @@ pub(crate) async fn run_wrap( _ = verification_tick.tick() => { let mut i = 0; while i < pending_verifications.len() { - if pending_verifications[i].injected_at.elapsed() >= VERIFICATION_WINDOW { + if pending_verifications[i].ready_to_assess(&resolved_cli, VERIFICATION_WINDOW) { let mut pv = pending_verifications.remove(i).unwrap(); pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); let snapshot = Snapshot::capture(&pty); pv.observe_visible_composer(&snapshot, &resolved_cli); - match assess_harness_acceptance( + match assess_delivery_acceptance( &resolved_cli, &pv, &snapshot, @@ -2973,6 +2981,7 @@ sys.stdout.flush()"#; #[test] fn busy_devin_deferrals_preserve_retry_budget_and_reset_deadline() { let mut verification = PendingVerification { + native_ack: None, delivery_id: DeliveryId::new("delivery"), event_id: EventId::new("event"), expected_echo: String::new(), diff --git a/crates/broker/tests/pty_injection_integrity.rs b/crates/broker/tests/pty_injection_integrity.rs index a8caa171b7..698e89d2b5 100644 --- a/crates/broker/tests/pty_injection_integrity.rs +++ b/crates/broker/tests/pty_injection_integrity.rs @@ -21,18 +21,21 @@ impl Drop for Fixture { } impl Fixture { fn new(mode: &str) -> Self { + Self::new_for_cli(mode, "claude-fixture") + } + fn new_for_cli(mode: &str, cli: &str) -> Self { let dir = tempfile::tempdir().unwrap(); - let script = dir.path().join("claude-fixture"); + let script = dir.path().join(cli); std::fs::write( &script, r#"#!/usr/bin/env python3 -import os, tty, time, select +import os, tty, time, select, json, pathlib # Canonical mode silently discards input after MAX_CANON. Real TUIs use raw mode. tty.setraw(0) mode=os.environ['FIXTURE_MODE'] time.sleep(0.2) os.write(1,b'\x1b[?2004h\x1b[?2004l' if mode != 'typed' else b'') -os.write(1,'Claude Code v2.1.261\r\n❯ '.encode()) +os.write(1,('OpenAI Codex\r\n› ' if os.environ['FIXTURE_CLI']=='codex' else 'Claude Code v2.1.261\r\n❯ ').encode()) data=b'' while True: chunk=os.read(0,65536) @@ -51,8 +54,14 @@ while True: data=b'' if data.endswith(b'\r'): body=data.replace(b'\x1b[200~',b'').replace(b'\x1b[201~',b'').rstrip(b'\r') - os.write(1,b'\r\n'+body+b'\r\n') - os.write(1,'\r\n❯ '.encode()) + if mode != 'echo-only' and os.environ.get('RELAY_PTY_ACK_DIR'): + for pending in pathlib.Path(os.environ['RELAY_PTY_ACK_DIR']).glob('*.pending'): + receipt=json.loads(pending.read_text()) + if receipt['messageId'].encode() in body: + pending.with_name(pending.stem+'.'+receipt['nonce']+'.read').write_text(receipt['nonce']) + if mode != 'native': + os.write(1,b'\r\n'+body+b'\r\n') + os.write(1,'\r\n❯ '.encode()) data=b'' "#, ) @@ -61,6 +70,7 @@ while True: let mut child = Command::new(env!("CARGO_BIN_EXE_agent-relay-broker")) .args(["pty", "--agent-name", "integrity", script.to_str().unwrap()]) .env("FIXTURE_MODE", mode) + .env("FIXTURE_CLI", cli) .env("RELAY_FAILED_DRAFT_ESCALATION_MS", "2000") .env("TRANSCRIPT", dir.path().join("transcript")) .env("AGENT_RELAY_LOCAL_ONLY", "1") @@ -84,6 +94,7 @@ while True: } }); let mut f = Self { child, rx, dir }; + f.send("resize_pty", json!({"rows": 24, "cols": 80})); f.send("init_worker", json!({"agent":{"name":"integrity"}})); assert_eq!(f.wait("worker_ready")["payload"]["readiness_proven"], true); f @@ -158,6 +169,43 @@ fn long_pair() { fn ten_kib_task_and_relay_message_are_intact() { long_pair(); } + +#[test] +fn claude_and_codex_native_reads_confirm_without_prompt_repaints() { + for cli in ["claude", "codex"] { + let mut fixture = Fixture::new_for_cli("native", cli); + fixture.deliver("native_one", "No screen echo or prompt repaint needed"); + let frame = fixture.wait_any( + &[ + "delivery_verified", + "delivery_unconfirmed", + "delivery_failed", + ], + 40, + ); + assert_eq!(frame["type"], "delivery_verified", "{cli}: {frame}"); + assert_eq!( + frame["payload"]["evidence"], "native_mcp_ack", + "{cli}: {frame}" + ); + } +} + +#[test] +fn prompt_repaints_without_native_reads_are_unconfirmed() { + let mut fixture = Fixture::new("echo-only"); + fixture.deliver("echo_one", "Echo alone must not confirm reading"); + let frame = fixture.wait_any( + &[ + "delivery_verified", + "delivery_unconfirmed", + "delivery_failed", + ], + 30, + ); + assert_eq!(frame["type"], "delivery_unconfirmed", "{frame}"); + assert_eq!(frame["payload"]["terminal"], true); +} #[test] fn over_limit_and_typed_fallback_reject_before_writing() { for mode in ["echo", "typed"] { diff --git a/packages/cli/src/cli/mcp/messaging-tools.ts b/packages/cli/src/cli/mcp/messaging-tools.ts index 54d6fc51a4..218db853bb 100644 --- a/packages/cli/src/cli/mcp/messaging-tools.ts +++ b/packages/cli/src/cli/mcp/messaging-tools.ts @@ -25,6 +25,7 @@ import { jsonContent, jsonResult, textContent } from './tool-results.js'; import { identityOverrideInputShape, messageResult } from './tool-shapes.js'; import { McpRequestReplay, replayScopeOf } from './request-replay.js'; import type { AgentClientLike } from './types.js'; +import { acknowledgePtyMessageRead } from './pty-delivery-ack.js'; const directMessageResult = z.looseObject({ id: z.string().optional().describe('Message ID, for "get_message_readers"'), @@ -668,6 +669,32 @@ export function registerMessagingTools( jsonContent(await getAgentClient(as).inbox(limit != null ? { limit } : undefined)) ); + server.registerTool( + 'acknowledge_delivery', + { + title: 'Acknowledge PTY Delivery', + description: + 'Confirm that you received a broker-injected message or task. Call before acting on the message, using its bracketed message ID. Only confirms a pending delivery to this worker.', + inputSchema: { + message_id: z.string().describe('Message ID from the delivery'), + ...identityOverrideInputShape, + }, + outputSchema: messageResult, + annotations: { + readOnlyHint: false, + destructiveHint: false, + idempotentHint: true, + openWorldHint: false, + }, + }, + async ({ message_id, as }) => { + if (!(await acknowledgePtyMessageRead(message_id, as))) { + return { ...textContent('No matching pending PTY delivery for this worker'), isError: true }; + } + return textContent(`Acknowledged delivery ${message_id}`); + } + ); + server.registerTool( 'mark_message_read', { @@ -684,6 +711,7 @@ export function registerMessagingTools( }, async ({ message_id, as }) => { await getAgentClient(as).markRead(message_id); + await acknowledgePtyMessageRead(message_id, as); return textContent(`Marked message ${message_id} as read`); } ); diff --git a/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts b/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts new file mode 100644 index 0000000000..bf2a398811 --- /dev/null +++ b/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts @@ -0,0 +1,100 @@ +import { createHash } from 'node:crypto'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { Client } from '@modelcontextprotocol/sdk/client/index.js'; +import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js'; +import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; + +import { acknowledgePtyMessageRead } from './pty-delivery-ack.js'; +import { registerMessagingTools } from './messaging-tools.js'; + +describe('PTY native inbox receipts', () => { + let directory: string; + const messageId = '../msg/one'; + const key = createHash('sha256').update(messageId).digest('hex'); + + beforeEach(async () => { + directory = await mkdtemp(path.join(os.tmpdir(), 'relay-ack-test-')); + vi.stubEnv('RELAY_PTY_ACK_DIR', directory); + vi.stubEnv('RELAY_AGENT_NAME', 'recipient'); + await writeFile( + path.join(directory, `${key}.pending`), + JSON.stringify({ messageId, nonce: '11111111-1111-4111-8111-111111111111' }) + ); + }); + afterEach(async () => { + vi.unstubAllEnvs(); + await rm(directory, { recursive: true, force: true }); + }); + + it('publishes the nonce for the matching read without using message IDs as paths', async () => { + await acknowledgePtyMessageRead(messageId); + expect( + await readFile(path.join(directory, `${key}.11111111-1111-4111-8111-111111111111.read`), 'utf8') + ).toBe('11111111-1111-4111-8111-111111111111'); + }); + + it('does not acknowledge a read performed as another identity', async () => { + await acknowledgePtyMessageRead(messageId, 'other-agent'); + await expect( + readFile(path.join(directory, `${key}.11111111-1111-4111-8111-111111111111.read`)) + ).rejects.toMatchObject({ code: 'ENOENT' }); + }); + + it('ignores receipts without a matching pending message', async () => { + await writeFile( + path.join(directory, `${key}.pending`), + JSON.stringify({ messageId: 'different', nonce: '11111111-1111-4111-8111-111111111111' }) + ); + await acknowledgePtyMessageRead(messageId); + await expect( + readFile(path.join(directory, `${key}.11111111-1111-4111-8111-111111111111.read`)) + ).rejects.toMatchObject({ code: 'ENOENT' }); + await expect(acknowledgePtyMessageRead('not-pending')).resolves.toBe(false); + }); + + it.each([ + ['mark_message_read', false], + ['mark_message_read', true], + ['acknowledge_delivery', false], + ] as const)('acknowledges through MCP %s (failure=%s)', async (tool, fail) => { + const server = new McpServer({ name: 'native-ack-test', version: '1.0.0' }); + const client = new Client({ name: 'recipient', version: '1.0.0' }); + const markRead = vi.fn(async () => { + if (fail) throw new Error('read rejected'); + }); + registerMessagingTools( + server, + () => ({ markRead }) as never, + async () => [] + ); + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + try { + await server.connect(serverTransport); + await client.connect(clientTransport); + const result = await client.callTool({ + name: tool, + arguments: { message_id: messageId }, + }); + if (tool === 'mark_message_read') expect(markRead).toHaveBeenCalledWith(messageId); + else expect(markRead).not.toHaveBeenCalled(); + const receipt = readFile( + path.join(directory, `${key}.11111111-1111-4111-8111-111111111111.read`), + 'utf8' + ); + if (fail) { + expect(result.isError).toBe(true); + await expect(receipt).rejects.toMatchObject({ code: 'ENOENT' }); + } else { + expect(result.isError).not.toBe(true); + expect(await receipt).toBe('11111111-1111-4111-8111-111111111111'); + } + } finally { + await client.close(); + await server.close(); + } + }); +}); diff --git a/packages/cli/src/cli/mcp/pty-delivery-ack.ts b/packages/cli/src/cli/mcp/pty-delivery-ack.ts new file mode 100644 index 0000000000..1f18dc9961 --- /dev/null +++ b/packages/cli/src/cli/mcp/pty-delivery-ack.ts @@ -0,0 +1,23 @@ +import { createHash, randomUUID } from 'node:crypto'; +import { readFile, rename, unlink, writeFile } from 'node:fs/promises'; +import path from 'node:path'; + +/** Publish a local receipt only for a successful read by this worker's own identity. */ +export async function acknowledgePtyMessageRead(messageId: string, as?: string): Promise { + const directory = process.env.RELAY_PTY_ACK_DIR; + if (!directory || (as && as !== process.env.RELAY_AGENT_NAME)) return false; + const key = createHash('sha256').update(messageId).digest('hex'); + const temporary = path.join(directory, `${key}.${randomUUID()}.tmp`); + try { + const pending = JSON.parse(await readFile(path.join(directory, `${key}.pending`), 'utf8')); + if (pending.messageId !== messageId || !/^[a-f0-9-]{36}$/.test(pending.nonce)) return false; + await writeFile(temporary, pending.nonce, { flag: 'wx', mode: 0o600 }); + await rename(temporary, path.join(directory, `${key}.${pending.nonce}.read`)); + return true; + } catch { + // A completed/expired PTY delivery must not turn a successful remote read into an MCP error. + return false; + } finally { + await unlink(temporary).catch(() => {}); + } +} diff --git a/packages/cli/src/cli/mcp/telemetry.ts b/packages/cli/src/cli/mcp/telemetry.ts index 677a716706..c1cb1f45fb 100644 --- a/packages/cli/src/cli/mcp/telemetry.ts +++ b/packages/cli/src/cli/mcp/telemetry.ts @@ -61,6 +61,7 @@ const AGENT_RELAY_TOOL_CALL_METADATA = { remove_reaction: { toolType: 'reaction.remove', toolCategory: 'reaction' }, check_inbox: { toolType: 'inbox.check', toolCategory: 'inbox' }, mark_message_read: { toolType: 'inbox.mark_read', toolCategory: 'inbox' }, + acknowledge_delivery: { toolType: 'inbox.acknowledge_delivery', toolCategory: 'inbox' }, get_message_readers: { toolType: 'inbox.reader_list', toolCategory: 'inbox' }, } satisfies Record; diff --git a/tests/e2e/fleet/nodes/stub-agent.cjs b/tests/e2e/fleet/nodes/stub-agent.cjs index 1f17624e2e..5086c8d56e 100644 --- a/tests/e2e/fleet/nodes/stub-agent.cjs +++ b/tests/e2e/fleet/nodes/stub-agent.cjs @@ -5,12 +5,13 @@ // 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 { mkdirSync, writeFileSync, readdirSync, readFileSync } = 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 = ''; +let deliveryInput = ''; const recordedNonces = new Set(); const nonceMarker = 'RELAY_E2E_BRIEF_NONCE='; const maxNonceLength = 256; @@ -38,11 +39,10 @@ function recordBriefNonce(nonce) { // Invoked as `codex`, model Codex's composer closely enough for the broker's // chunked initial-task path: raw input (no kernel echo, so discarded // pre-ready keystrokes never appear), typed text rendered after the `›` -// prompt, and Enter submitting the turn and redrawing an empty composer. A -// composer that keeps the submitted text visible looks like a parked task, so -// the broker would retry the submit and report the delivery incomplete. +// prompt, and Enter submitting the turn and redrawing an empty composer. +// These screen changes guide chunking only; the native receipt confirms delivery. const codexComposer = path.basename(process.argv[1]) === 'codex'; -if (codexComposer && process.stdin.isTTY) process.stdin.setRawMode(true); +if (process.stdin.isTTY) process.stdin.setRawMode(true); function renderCodexComposer(text) { for (const ch of text) { @@ -59,9 +59,18 @@ try { process.stdin.on('data', (chunk) => { if (!ready) return; if (codexComposer) renderCodexComposer(chunk.toString()); - // Like a real harness, Enter submits the turn and repaints the prompt, so - // the broker can see the composer clear and confirm acceptance. - else if (/[\r\n]/.test(chunk.toString())) process.stdout.write('\r\n❯ '); + // Acknowledge a submitted delivery from the recipient, independent of the screen. + deliveryInput = (deliveryInput + chunk.toString()).slice(-65536); + const ackDir = process.env.RELAY_PTY_ACK_DIR; + if (ackDir && chunk.toString().includes('\r')) { + for (const name of readdirSync(ackDir).filter((name) => name.endsWith('.pending'))) { + const receipt = JSON.parse(readFileSync(path.join(ackDir, name), 'utf8')); + if (deliveryInput.includes(receipt.messageId)) { + writeFileSync(path.join(ackDir, `${name.slice(0, -8)}.${receipt.nonce}.read`), receipt.nonce); + } + } + deliveryInput = ''; + } input += chunk.toString(); // Require a delimiter after the nonce. PTY chunks can split anywhere, so // treating the current buffer end as a complete token could record a @@ -97,6 +106,8 @@ setTimeout(() => { // MCP boot label. This native stub starts no MCP/AI service. if (path.basename(process.argv[1]) === 'codex') { process.stdout.write('› '); + } else { + process.stdout.write('❯ '); } }, readyDelayMs); diff --git a/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs b/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs index 6739f14655..7573924f39 100644 --- a/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs +++ b/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs @@ -48,15 +48,23 @@ const positiveNonce = `R${expectedSha.slice(0, 8)}${arm[0]}`; const negativeNonce = `U${expectedSha.slice(0, 8)}${arm[0]}`; const selfNonce = `S${expectedSha.slice(0, 8)}${arm[0]}`; const fakeClaudeSource = String.raw`#!/usr/bin/env node +const fs = require('node:fs'); +const path = require('node:path'); process.stdin.setRawMode?.(true); process.stdin.resume(); process.stdout.write('RELAYFLOW_RECIPIENT_READY\r\n❯'); -// Like the real composer, redraw the prompt after Enter submits a line, so -// the broker's pre-write composer check can prove readiness for the next -// message instead of seeing only the previous message's echo at the cursor. +let submitted = ''; +// Confirm consumption natively; no echo or prompt repaint is needed. process.stdin.on('data', (chunk) => { - process.stdout.write(chunk); - if (chunk.includes(13)) process.stdout.write('\r\n❯'); + submitted = (submitted + chunk.toString()).slice(-65536); + const directory = process.env.RELAY_PTY_ACK_DIR; + if (directory && chunk.includes(13)) { + for (const name of fs.readdirSync(directory).filter((name) => name.endsWith('.pending'))) { + const receipt = JSON.parse(fs.readFileSync(path.join(directory, name), 'utf8')); + if (submitted.includes(receipt.messageId)) fs.writeFileSync(path.join(directory, name.slice(0, -8) + '.' + receipt.nonce + '.read'), receipt.nonce); + } + submitted = ''; + } }); `; diff --git a/tests/relayflows/cases/1961-native-pty-ack/case.json b/tests/relayflows/cases/1961-native-pty-ack/case.json new file mode 100644 index 0000000000..f033091943 --- /dev/null +++ b/tests/relayflows/cases/1961-native-pty-ack/case.json @@ -0,0 +1,15 @@ +{ + "version": 1, + "id": "1961-native-pty-ack", + "kind": "bugfix", + "title": "Confirm PTY delivery with recipient acknowledgements without prompt repaints", + "runner": { + "command": ["node", "tests/relayflows/cases/1961-native-pty-ack/run.mjs"] + }, + "requirements": ["broker-linux-x64"], + "timeoutSeconds": 180, + "expected": { + "base": { "outcome": "bug", "signature": "native_ack_not_observed" }, + "head": { "outcome": "fixed", "signature": "native_ack_confirms_without_repaint" } + } +} diff --git a/tests/relayflows/cases/1961-native-pty-ack/run.mjs b/tests/relayflows/cases/1961-native-pty-ack/run.mjs new file mode 100644 index 0000000000..f891f02f33 --- /dev/null +++ b/tests/relayflows/cases/1961-native-pty-ack/run.mjs @@ -0,0 +1,133 @@ +import { execFileSync, spawn } from 'node:child_process'; +import { createInterface } from 'node:readline'; +import { chmod, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const caseId = '1961-native-pty-ack'; +const arm = process.env.RELAY_PR_PROOF_ARM; +const target = process.env.RELAY_PR_PROOF_TARGET_DIR; +const harness = process.env.RELAY_PR_PROOF_HARNESS_DIR; +const resultPath = process.env.RELAY_PR_PROOF_RESULT_PATH; +const broker = process.env.RELAY_PR_PROOF_BROKER_BINARY; +if (!['base', 'head'].includes(arm) || !target || !harness || !resultPath || !broker) + throw new Error('Missing proof environment or exact-SHA broker artifact'); +await rm(resultPath, { force: true }); +const sha = execFileSync('git', ['-C', target, 'rev-parse', 'HEAD'], { encoding: 'utf8' }).trim(); +if (sha !== process.env[arm === 'base' ? 'RELAY_PR_PROOF_BASE_SHA' : 'RELAY_PR_PROOF_HEAD_SHA']) + throw new Error('Target SHA mismatch'); +if (path.relative(harness, fileURLToPath(import.meta.url)).startsWith('..')) + throw new Error('Runner is outside the harness checkout'); + +// The recipient consumes a submitted body and emits a native acknowledgement. +// It never echoes the body, emits activity, or repaints its initial prompt. +const recipient = String.raw`#!/usr/bin/env python3 +import json, os, pathlib, tty +tty.setraw(0) +os.write(1, ('OpenAI Codex\r\n› ' if os.environ['PROOF_CLI']=='codex' else 'Claude Code v2.1.261\r\n❯ ').encode()) +data=b'' +while True: + data+=os.read(0,65536) + if data.endswith(b'\r'): + directory=os.environ.get('RELAY_PTY_ACK_DIR') + if directory: + for pending in pathlib.Path(directory).glob('*.pending'): + entry=json.loads(pending.read_text()) + if entry['messageId'].encode() in data: + pending.with_name(pending.stem+'.'+entry['nonce']+'.read').write_text(entry['nonce']) + data=b'' +`; + +/** Observe a terminal broker verdict from a recipient that never repaints. */ +async function observe(cli) { + const dir = await mkdtemp(path.join(os.tmpdir(), 'relay-native-ack-proof-')); + const executable = path.join(dir, cli); + await writeFile(executable, recipient); + await chmod(executable, 0o755); + const child = spawn(broker, ['pty', '--agent-name', 'proof-recipient', executable], { + cwd: dir, + env: { ...process.env, PROOF_CLI: cli, AGENT_RELAY_TELEMETRY_DISABLED: '1', AGENT_RELAY_LOCAL_ONLY: '1' }, + stdio: ['pipe', 'pipe', 'pipe'], + }); + let stderr = ''; + child.stderr.on('data', (chunk) => { + stderr = (stderr + chunk).slice(-4000); + }); + const lines = createInterface({ input: child.stdout }); + let timer; + try { + return await new Promise((resolve, reject) => { + timer = setTimeout(() => reject(new Error(`${cli} timed out: ${stderr}`)), 65_000); + child.once('error', reject); + child.once('exit', (code) => reject(new Error(`${cli} broker exited ${code}: ${stderr}`))); + lines.on('line', (line) => { + let frame; + try { + frame = JSON.parse(line); + } catch { + return; + } + if (frame.type === 'worker_ready') { + if (!frame.payload.readiness_proven) { + reject(new Error(`${cli} readiness was not proven`)); + return; + } + child.stdin.write( + JSON.stringify({ + v: 2, + type: 'deliver_relay', + payload: { + delivery_id: 'proof-delivery', + event_id: 'proof-message', + from: 'sender', + target: 'proof-recipient', + body: 'Read this message without echoing or repainting.', + priority: 2, + injection_mode: 'wait', + }, + }) + '\n' + ); + } + if (frame.payload?.event_id !== 'proof-message') return; + if (frame.type === 'delivery_failed') + reject(new Error(`${cli} delivery failed: ${JSON.stringify(frame)}`)); + if (frame.type === 'delivery_verified') { + if (frame.payload.evidence !== 'native_mcp_ack') + reject(new Error('Screen evidence was treated as confirmed delivery')); + else resolve('native'); + } + if (frame.type === 'delivery_unconfirmed' && frame.payload.terminal === true) resolve('unconfirmed'); + }); + child.stdin.write(JSON.stringify({ v: 2, type: 'resize_pty', payload: { rows: 24, cols: 80 } }) + '\n'); + child.stdin.write( + JSON.stringify({ v: 2, type: 'init_worker', payload: { agent: { name: 'proof-recipient' } } }) + '\n' + ); + }); + } finally { + clearTimeout(timer); + lines.close(); + child.kill('SIGKILL'); + await new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) resolve(); + else child.once('exit', resolve); + }); + await rm(dir, { recursive: true, force: true }); + } +} + +const observations = { claude: await observe('claude'), codex: await observe('codex') }; +if (observations.claude !== observations.codex) + throw new Error(`Harness outcomes differ: ${JSON.stringify(observations)}`); +const fixed = observations.claude === 'native'; +await writeFile( + resultPath, + JSON.stringify({ + version: 1, + caseId, + arm, + outcome: fixed ? 'fixed' : 'bug', + signature: fixed ? 'native_ack_confirms_without_repaint' : 'native_ack_not_observed', + details: JSON.stringify(observations), + }) + '\n' +); From ab780d687c0b48185f0be9b172b957a6da0a0026 Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 19:16:13 +0530 Subject: [PATCH 2/6] fix(broker): bind native receipts to the recipient identity --- crates/broker/src/pty_delivery_ack.rs | 8 ++++---- crates/broker/src/pty_worker.rs | 2 +- crates/broker/src/wrap.rs | 2 +- packages/cli/src/cli/mcp/pty-delivery-ack.test.ts | 13 +++++++++++-- packages/cli/src/cli/mcp/pty-delivery-ack.ts | 7 ++++++- 5 files changed, 23 insertions(+), 9 deletions(-) diff --git a/crates/broker/src/pty_delivery_ack.rs b/crates/broker/src/pty_delivery_ack.rs index 1e5b18d26d..70d90d86cf 100644 --- a/crates/broker/src/pty_delivery_ack.rs +++ b/crates/broker/src/pty_delivery_ack.rs @@ -71,14 +71,14 @@ impl NativeAckInbox { Ok(()) } - pub(crate) fn register(&self, event_id: &str) -> io::Result { + pub(crate) fn register(&self, event_id: &str, agent_name: &str) -> io::Result { let key = format!("{:x}", Sha256::digest(event_id.as_bytes())); let pending = self.0.path().join(format!("{key}.pending")); let nonce = uuid::Uuid::new_v4().to_string(); let receipt = self.0.path().join(format!("{key}.{nonce}.read")); fs::write( &pending, - json!({"messageId": event_id, "nonce": nonce}).to_string(), + json!({"messageId": event_id, "agentName": agent_name, "nonce": nonce}).to_string(), )?; Ok(NativeAck { pending, @@ -116,7 +116,7 @@ mod tests { fn native_receipt_requires_this_delivery_nonce_and_is_cleaned_up() { // Construct directly so parallel tests do not mutate process environment. let inbox = NativeAckInbox(tempfile::tempdir().unwrap()); - let ack = inbox.register("msg/../one").unwrap(); + let ack = inbox.register("msg/../one", "worker").unwrap(); assert!(!ack.confirmed()); fs::write(&ack.receipt, "old-or-wrong-nonce").unwrap(); assert!(!ack.confirmed()); @@ -127,7 +127,7 @@ mod tests { drop(ack); assert!(!pending.exists()); assert!(!receipt.exists()); - assert!(!inbox.register("msg/../one").unwrap().confirmed()); + assert!(!inbox.register("msg/../one", "worker").unwrap().confirmed()); } #[test] diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 34440eefae..82ac1716db 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -2429,7 +2429,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty_auto.auto_suggestion_visible = false; let include_mcp_reminder = !suppress_multiline_mcp_reminder && mcp_reminder_throttle.should_include(Instant::now()); - inj.native_ack = Some(native_acks.register(&inj.pending.delivery.event_id)?); + inj.native_ack = Some(native_acks.register(&inj.pending.delivery.event_id, &worker_name)?); let injection = format_injection_for_worker_with_workspace( &inj.pending.delivery.from, &inj.pending.delivery.event_id, diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 8c3195b5da..30da5d00b9 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -2325,7 +2325,7 @@ pub(crate) async fn run_wrap( pending.workspace_alias.as_deref(), ); let bytes = crate::injection_wire::injection_bytes(crate::injection_wire::injection_wire(&resolved_cli, &pty), &injection); - let native_ack = native_acks.register(&pending.event_id)?; + let native_ack = native_acks.register(&pending.event_id, &default_workspace.self_name)?; let write = submit_injection_body(&pty, &resolved_cli, bytes, Duration::ZERO); match write { Ok((ack_rx, output_boundary)) => { diff --git a/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts b/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts index bf2a398811..36734da85e 100644 --- a/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts +++ b/packages/cli/src/cli/mcp/pty-delivery-ack.test.ts @@ -22,7 +22,7 @@ describe('PTY native inbox receipts', () => { vi.stubEnv('RELAY_AGENT_NAME', 'recipient'); await writeFile( path.join(directory, `${key}.pending`), - JSON.stringify({ messageId, nonce: '11111111-1111-4111-8111-111111111111' }) + JSON.stringify({ messageId, agentName: 'recipient', nonce: '11111111-1111-4111-8111-111111111111' }) ); }); afterEach(async () => { @@ -44,10 +44,19 @@ describe('PTY native inbox receipts', () => { ).rejects.toMatchObject({ code: 'ENOENT' }); }); + it('rejects an inherited receipt directory belonging to another worker', async () => { + vi.stubEnv('RELAY_AGENT_NAME', 'child-worker'); + expect(await acknowledgePtyMessageRead(messageId)).toBe(false); + }); + it('ignores receipts without a matching pending message', async () => { await writeFile( path.join(directory, `${key}.pending`), - JSON.stringify({ messageId: 'different', nonce: '11111111-1111-4111-8111-111111111111' }) + JSON.stringify({ + messageId: 'different', + agentName: 'recipient', + nonce: '11111111-1111-4111-8111-111111111111', + }) ); await acknowledgePtyMessageRead(messageId); await expect( diff --git a/packages/cli/src/cli/mcp/pty-delivery-ack.ts b/packages/cli/src/cli/mcp/pty-delivery-ack.ts index 1f18dc9961..c893fadfdb 100644 --- a/packages/cli/src/cli/mcp/pty-delivery-ack.ts +++ b/packages/cli/src/cli/mcp/pty-delivery-ack.ts @@ -10,7 +10,12 @@ export async function acknowledgePtyMessageRead(messageId: string, as?: string): const temporary = path.join(directory, `${key}.${randomUUID()}.tmp`); try { const pending = JSON.parse(await readFile(path.join(directory, `${key}.pending`), 'utf8')); - if (pending.messageId !== messageId || !/^[a-f0-9-]{36}$/.test(pending.nonce)) return false; + if ( + pending.messageId !== messageId || + pending.agentName !== process.env.RELAY_AGENT_NAME || + !/^[a-f0-9-]{36}$/.test(pending.nonce) + ) + return false; await writeFile(temporary, pending.nonce, { flag: 'wx', mode: 0o600 }); await rename(temporary, path.join(directory, `${key}.${pending.nonce}.read`)); return true; From 04f436bd6bfeaa0311561c1762af5c8c052bdc7e Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 19:17:00 +0530 Subject: [PATCH 3/6] test: keep native delivery proof scoped to one case --- .../run.mjs | 18 +++++------------- 1 file changed, 5 insertions(+), 13 deletions(-) diff --git a/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs b/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs index 7573924f39..6739f14655 100644 --- a/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs +++ b/tests/relayflows/cases/1615-api-send-recipient-reachability/run.mjs @@ -48,23 +48,15 @@ const positiveNonce = `R${expectedSha.slice(0, 8)}${arm[0]}`; const negativeNonce = `U${expectedSha.slice(0, 8)}${arm[0]}`; const selfNonce = `S${expectedSha.slice(0, 8)}${arm[0]}`; const fakeClaudeSource = String.raw`#!/usr/bin/env node -const fs = require('node:fs'); -const path = require('node:path'); process.stdin.setRawMode?.(true); process.stdin.resume(); process.stdout.write('RELAYFLOW_RECIPIENT_READY\r\n❯'); -let submitted = ''; -// Confirm consumption natively; no echo or prompt repaint is needed. +// Like the real composer, redraw the prompt after Enter submits a line, so +// the broker's pre-write composer check can prove readiness for the next +// message instead of seeing only the previous message's echo at the cursor. process.stdin.on('data', (chunk) => { - submitted = (submitted + chunk.toString()).slice(-65536); - const directory = process.env.RELAY_PTY_ACK_DIR; - if (directory && chunk.includes(13)) { - for (const name of fs.readdirSync(directory).filter((name) => name.endsWith('.pending'))) { - const receipt = JSON.parse(fs.readFileSync(path.join(directory, name), 'utf8')); - if (submitted.includes(receipt.messageId)) fs.writeFileSync(path.join(directory, name.slice(0, -8) + '.' + receipt.nonce + '.read'), receipt.nonce); - } - submitted = ''; - } + process.stdout.write(chunk); + if (chunk.includes(13)) process.stdout.write('\r\n❯'); }); `; From 11c64b35ff05513e1336939faa473ac07d44196e Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 19:44:07 +0530 Subject: [PATCH 4/6] fix(broker): preserve native receipt recovery and MCP configuration --- .../src/broker/delivery_verification.rs | 40 ++-- crates/broker/src/pty_delivery_ack.rs | 171 ++++++++++++++++-- crates/broker/src/pty_worker.rs | 9 +- crates/broker/src/snippets.rs | 13 +- crates/broker/src/wrap.rs | 35 +++- .../broker/tests/pty_injection_integrity.rs | 45 +++++ 6 files changed, 273 insertions(+), 40 deletions(-) diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 2117592b08..0c74847ee0 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -144,20 +144,18 @@ pub(crate) struct PendingVerification { } impl PendingVerification { - /// Native receipts can complete immediately. Known MCP harnesses get the - /// full acceptance budget to read, while generic harnesses keep the short fallback. - pub(crate) fn ready_to_assess(&self, cli: &str, window: Duration) -> bool { + /// Check parked composers on the short interval even while waiting for a receipt. + pub(crate) fn ready_to_assess(&self, _cli: &str, window: Duration) -> bool { self.native_ack.as_ref().is_some_and(|ack| ack.confirmed()) - || if self.native_ack.is_some() - && matches!( - crate::cli::command_parse::normalize_cli_name(cli).as_str(), - "claude" | "codex" - ) - { - self.acceptance_expired() - } else { - self.injected_at.elapsed() >= window - } + || self.injected_at.elapsed() >= window + } + + /// Defer only inconclusive outcomes while a native recipient can still acknowledge. + pub(crate) fn waiting_for_native_ack(&self, cli: &str) -> bool { + let name = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); + self.native_ack.is_some() + && matches!(name.trim_end_matches(".exe"), "claude" | "codex") + && !self.acceptance_expired() } pub(crate) fn observe_raw_process_receipt(&mut self, output: &VerificationOutput, cli: &str) { @@ -1520,6 +1518,22 @@ mod tests { assert!(verification.acceptance_expired()); } + #[test] + fn native_receipt_wait_preserves_short_recovery_checks_and_total_deadline() { + let inbox = crate::pty_delivery_ack::NativeAckInbox::new().unwrap(); + for cli in ["claude", "codex", "Claude.exe", "Codex.exe"] { + let mut verification = codex_verification("bounded delivery"); + verification.native_ack = Some(inbox.register(cli, "worker").unwrap()); + verification.injected_at = Instant::now() - VERIFICATION_WINDOW; + assert!(verification.ready_to_assess(cli, VERIFICATION_WINDOW)); + assert!(verification.waiting_for_native_ack(cli)); + verification.injected_at = Instant::now(); + assert!(!verification.ready_to_assess(cli, VERIFICATION_WINDOW)); + verification.verification_started_at = Instant::now() - MAX_ACCEPTANCE_LIFETIME; + assert!(!verification.waiting_for_native_ack(cli)); + } + } + #[cfg(unix)] #[tokio::test] async fn attachment_delivery_to_no_marker_harness_is_confirmed_by_guarded_rule() { diff --git a/crates/broker/src/pty_delivery_ack.rs b/crates/broker/src/pty_delivery_ack.rs index 70d90d86cf..607f67f5a7 100644 --- a/crates/broker/src/pty_delivery_ack.rs +++ b/crates/broker/src/pty_delivery_ack.rs @@ -2,16 +2,22 @@ //! //! Each worker owns a private, short-lived directory inherited by its harness //! and MCP subprocesses. A fresh nonce fences each delivery against stale reads. -use std::{fs, io, path::PathBuf}; +use std::{ + fs, + io::{self, Write}, + path::PathBuf, +}; use serde_json::json; use sha2::{Digest, Sha256}; pub(crate) const ACK_DIR_ENV: &str = "RELAY_PTY_ACK_DIR"; +/// Owns pending receipts and private harness configs for one PTY session. pub(crate) struct NativeAckInbox(tempfile::TempDir); impl NativeAckInbox { + /// Create a private directory that is removed with the session. pub(crate) fn new() -> io::Result { let dir = tempfile::Builder::new() .prefix("relay-pty-acks-") @@ -19,6 +25,7 @@ impl NativeAckInbox { Ok(Self(dir)) } + /// Let direct child processes discover this session's receipt directory. pub(crate) fn expose_to_child(&self) { // PTY workers are dedicated processes; only their descendants inherit this path. std::env::set_var(ACK_DIR_ENV, self.0.path()); @@ -30,22 +37,36 @@ impl NativeAckInbox { let cli = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); let directory = self.0.path().to_string_lossy().to_string(); if cli == "codex" || cli == "codex.exe" { - args.extend([ - "--config".into(), - format!( - "mcp_servers.agent-relay.env.{ACK_DIR_ENV}={}", - serde_json::to_string(&directory)? - ), - ]); + let names: Vec<_> = ["agent-relay", "relaycast"] + .into_iter() + .filter(|name| { + args.iter() + .any(|arg| arg.contains(&format!("mcp_servers.{name}.command="))) + }) + .collect(); + for name in names { + args.extend([ + "--config".into(), + format!( + "mcp_servers.{name}.env.{ACK_DIR_ENV}={}", + serde_json::to_string(&directory)? + ), + ]); + } } else if cli == "claude" || cli == "claude.exe" { - for index in 1..args.len() { - if args[index - 1] != "--mcp-config" { + for index in 0..args.len() { + let equals = args[index].strip_prefix("--mcp-config="); + let value = if let Some(value) = equals { + value + } else if index > 0 && args[index - 1] == "--mcp-config" { + &args[index] + } else { continue; - } - let raw = if args[index].trim_start().starts_with('{') { - args[index].clone() + }; + let raw = if value.trim_start().starts_with('{') { + value.to_string() } else { - fs::read_to_string(&args[index])? + fs::read_to_string(value)? }; let mut config: serde_json::Value = serde_json::from_str(&raw)?; if let Some(servers) = config @@ -64,13 +85,22 @@ impl NativeAckInbox { } } } - // Replace the argument, never the user's config file. - args[index] = config.to_string(); + // Keep credentials out of argv and leave the user's file intact. + // The inbox owns the private file for the entire harness lifetime. + let mut file = tempfile::NamedTempFile::new_in(self.0.path())?; + file.write_all(config.to_string().as_bytes())?; + let (_, path) = file.keep().map_err(|error| error.error)?; + args[index] = if equals.is_some() { + format!("--mcp-config={}", path.display()) + } else { + path.to_string_lossy().into_owned() + }; } } Ok(()) } + /// Register a fresh recipient-bound nonce before submitting a message body. pub(crate) fn register(&self, event_id: &str, agent_name: &str) -> io::Result { let key = format!("{:x}", Sha256::digest(event_id.as_bytes())); let pending = self.0.path().join(format!("{key}.pending")); @@ -96,6 +126,7 @@ pub(crate) struct NativeAck { } impl NativeAck { + /// A receipt confirms this delivery only when it contains its fresh nonce. pub(crate) fn confirmed(&self) -> bool { fs::read_to_string(&self.receipt).is_ok_and(|value| value == self.nonce) } @@ -135,7 +166,8 @@ mod tests { let inbox = NativeAckInbox::new().unwrap(); let mut claude = vec!["--mcp-config".into(), json!({"mcpServers": {"agent-relay": {"command": "agent-relay", "args": ["mcp"], "env": {"RELAY_AGENT_NAME": "worker"}}, "other": {"command": "other"}}}).to_string()]; inbox.configure_harness("claude", &mut claude).unwrap(); - let config: serde_json::Value = serde_json::from_str(&claude[1]).unwrap(); + let config: serde_json::Value = + serde_json::from_str(&fs::read_to_string(&claude[1]).unwrap()).unwrap(); assert_eq!( config["mcpServers"]["agent-relay"]["env"][ACK_DIR_ENV], inbox.0.path().to_str().unwrap() @@ -145,9 +177,110 @@ mod tests { "worker" ); assert!(config["mcpServers"]["other"].get("env").is_none()); - let mut codex = vec!["--model".into(), "test-model".into()]; + let mut codex = vec![ + "--config".into(), + "mcp_servers.agent-relay.command=\"agent-relay\"".into(), + ]; inbox.configure_harness("codex", &mut codex).unwrap(); assert!(codex[3].starts_with("mcp_servers.agent-relay.env.RELAY_PTY_ACK_DIR=")); - assert_eq!(&codex[..2], ["--model", "test-model"]); + assert_eq!( + &codex[..2], + [ + "--config", + "mcp_servers.agent-relay.command=\"agent-relay\"" + ] + ); + } + + #[test] + fn claude_file_and_equals_configs_stay_private_and_preserve_the_source() { + let inbox = NativeAckInbox::new().unwrap(); + let source = tempfile::NamedTempFile::new().unwrap(); + let original = json!({"mcpServers": { + "relaycast": {"command": "agent-relay", "env": {"RELAY_AGENT_NAME": "worker"}}, + "other": {"command": "other", "env": {"SECRET": "private-test-value"}} + }}) + .to_string(); + fs::write(source.path(), &original).unwrap(); + for equals in [false, true] { + let mut args = if equals { + vec![format!("--mcp-config={}", source.path().display())] + } else { + vec![ + "--mcp-config".into(), + source.path().to_string_lossy().into_owned(), + ] + }; + inbox.configure_harness("claude", &mut args).unwrap(); + assert!(!args.iter().any(|arg| arg.contains("private-test-value"))); + let path = args.last().unwrap().trim_start_matches("--mcp-config="); + let config: serde_json::Value = + serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap(); + assert_eq!( + config["mcpServers"]["relaycast"]["env"][ACK_DIR_ENV], + inbox.0.path().to_str().unwrap() + ); + assert_eq!( + config["mcpServers"]["other"]["env"]["SECRET"], + "private-test-value" + ); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + fs::metadata(path).unwrap().permissions().mode() & 0o777, + 0o600 + ); + } + } + assert_eq!(fs::read_to_string(source.path()).unwrap(), original); + } + + #[test] + fn codex_legacy_server_receives_the_directory_without_a_second_server() { + let inbox = NativeAckInbox::new().unwrap(); + let mut args = vec![ + "--config".into(), + "mcp_servers.relaycast.command=\"agent-relay\"".into(), + ]; + inbox.configure_harness("codex", &mut args).unwrap(); + assert!(args[3].starts_with("mcp_servers.relaycast.env.RELAY_PTY_ACK_DIR=")); + assert!(!args + .iter() + .any(|arg| arg.contains("mcp_servers.agent-relay"))); + let mut no_server = vec!["--model".into(), "test-model".into()]; + let original = no_server.clone(); + inbox.configure_harness("codex", &mut no_server).unwrap(); + assert_eq!( + no_server, original, + "local-only sessions must not get an incomplete MCP server" + ); + } + + #[tokio::test] + async fn wrap_claude_config_forwards_the_connected_identity_and_receipt_directory() { + let inbox = NativeAckInbox::new().unwrap(); + let mut args = crate::snippets::configure_agent_relay_mcp_with_token( + "claude", + "connected-worker", + Some("workspace-key"), + None, + &[], + inbox.0.path(), + Some("agent-token"), + None, + None, + ) + .await + .unwrap(); + inbox.configure_harness("claude", &mut args).unwrap(); + let index = args.iter().position(|arg| arg == "--mcp-config").unwrap() + 1; + let config: serde_json::Value = + serde_json::from_str(&fs::read_to_string(&args[index]).unwrap()).unwrap(); + let env = &config["mcpServers"]["agent-relay"]["env"]; + assert_eq!(env["RELAY_AGENT_NAME"], "connected-worker"); + assert_eq!(env["RELAY_AGENT_TOKEN"], "agent-token"); + assert_eq!(env[ACK_DIR_ENV], inbox.0.path().to_str().unwrap()); + assert!(!args.iter().any(|arg| arg.contains("agent-token"))); } } diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 82ac1716db..131d4d56b3 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -2429,7 +2429,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pty_auto.auto_suggestion_visible = false; let include_mcp_reminder = !suppress_multiline_mcp_reminder && mcp_reminder_throttle.should_include(Instant::now()); - inj.native_ack = Some(native_acks.register(&inj.pending.delivery.event_id, &worker_name)?); + inj.native_ack = native_acks.register(&inj.pending.delivery.event_id, &worker_name) + .inspect_err(|error| tracing::warn!(%error, + "native receipt registration failed; continuing without a receipt")) + .ok(); let injection = format_injection_for_worker_with_workspace( &inj.pending.delivery.from, &inj.pending.delivery.event_id, @@ -3059,6 +3062,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pv.injected_at = Instant::now(); pending_verifications.push_back(pv); } + HarnessAcceptance::Inconclusive if pv.waiting_for_native_ack(&resolved_cli) => { + pv.injected_at = Instant::now(); + pending_verifications.push_back(pv); + } HarnessAcceptance::Inconclusive => { // Unconfirmed: the body was written and is not // visibly parked, but nothing proved the diff --git a/crates/broker/src/snippets.rs b/crates/broker/src/snippets.rs index c50354c3d9..c2670e89a7 100644 --- a/crates/broker/src/snippets.rs +++ b/crates/broker/src/snippets.rs @@ -18,7 +18,12 @@ use crate::types::AgentResultMcpConfig; const AGENT_RELAY_MCP_PACKAGE: &str = "agent-relay"; const AGENT_RELAY_MCP_SUBCOMMAND: &str = "mcp"; -const REQUIRED_AGENT_RELAY_MCP_TOOLS: [&str; 3] = ["send_dm", "post_message", "check_inbox"]; +const REQUIRED_AGENT_RELAY_MCP_TOOLS: [&str; 4] = [ + "send_dm", + "post_message", + "check_inbox", + "acknowledge_delivery", +]; const MCP_FILE: &str = ".mcp.json"; const AGENT_RELAY_MCP_SERVER: &str = "agent-relay"; @@ -3147,7 +3152,7 @@ mod tests { r#"#!/bin/sh cat >/dev/null & printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{},"serverInfo":{"name":"agent-relay","version":"test"}}}' -printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"send_dm"},{"name":"post_message"},{"name":"check_inbox"}]}}' +printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"send_dm"},{"name":"post_message"},{"name":"check_inbox"},{"name":"acknowledge_delivery"}]}}' "#, ) .expect("write fake MCP executable"); @@ -3176,7 +3181,7 @@ printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"send_dm"},{"n r#"#!/bin/sh cat >/dev/null & printf '%s\n' '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":"2025-06-18","capabilities":{},"serverInfo":{"name":"agent-relay","version":"test"}}}' -printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"check_inbox"}]}}' +printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"send_dm"},{"name":"post_message"},{"name":"check_inbox"}]}}' "#, ) .expect("write fake MCP executable"); @@ -3194,7 +3199,7 @@ printf '%s\n' '{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"check_inbox"} .await .expect_err("incomplete MCP tool list must fail"); - assert!(error.to_string().contains("send_dm, post_message")); + assert!(error.to_string().contains("acknowledge_delivery")); } #[cfg(unix)] diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 30da5d00b9..3f59b8a4d1 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -272,7 +272,7 @@ pub(crate) struct PendingWrapInjection { /// and its harness-specific submit sequence were written and flushed. enum PendingWrapWrite { Initial { - native_ack: crate::pty_delivery_ack::NativeAck, + native_ack: Option, pending: PendingWrapInjection, injection: String, output_boundary: u64, @@ -1537,6 +1537,28 @@ pub(crate) async fn run_wrap( } let native_acks = crate::pty_delivery_ack::NativeAckInbox::new()?; native_acks.expose_to_child(); + if matches!( + crate::cli::command_parse::normalize_cli_name(&resolved_cli).as_str(), + "claude" | "codex" + ) { + // Wrap sessions also need an explicit MCP server environment. Claude + // does not reliably forward the parent's environment to .mcp.json servers. + effective_cli_args = crate::snippets::configure_agent_relay_mcp_with_token( + &resolved_cli, + &default_workspace.self_name, + Some(&default_workspace.relay_workspace_key), + child_base_url.as_deref(), + &effective_cli_args, + &runtime_cwd, + default_workspace + .http_client + .cached_agent_token(&default_workspace.self_name) + .as_deref(), + Some(&child_workspaces_json), + Some(&default_workspace.workspace_id), + ) + .await?; + } native_acks.configure_harness(&resolved_cli, &mut effective_cli_args)?; let _devin_state = crate::devin::prepare_worker_config(&resolved_cli).await?; @@ -2325,7 +2347,10 @@ pub(crate) async fn run_wrap( pending.workspace_alias.as_deref(), ); let bytes = crate::injection_wire::injection_bytes(crate::injection_wire::injection_wire(&resolved_cli, &pty), &injection); - let native_ack = native_acks.register(&pending.event_id, &default_workspace.self_name)?; + let native_ack = native_acks.register(&pending.event_id, &default_workspace.self_name) + .inspect_err(|error| tracing::warn!(%error, + "wrap: native receipt registration failed; continuing without a receipt")) + .ok(); let write = submit_injection_body(&pty, &resolved_cli, bytes, Duration::ZERO); match write { Ok((ack_rx, output_boundary)) => { @@ -2418,7 +2443,7 @@ pub(crate) async fn run_wrap( ); pty_auto.note_completed_injection(&resolved_cli); let verification = PendingVerification { - native_ack: Some(native_ack), + native_ack, delivery_id: DeliveryId::new(format!("wrap_{}", pending.event_id)), event_id: pending.event_id, expected_echo: injection, @@ -2598,6 +2623,10 @@ pub(crate) async fn run_wrap( // Parked is the positive evidence. wrap_failed_draft = Some(pv.expected_echo.clone()); } + HarnessAcceptance::Inconclusive if pv.waiting_for_native_ack(&resolved_cli) => { + pv.injected_at = Instant::now(); + pending_verifications.push_back(pv); + } HarnessAcceptance::Inconclusive => { tracing::info!( event_id = %pv.event_id, diff --git a/crates/broker/tests/pty_injection_integrity.rs b/crates/broker/tests/pty_injection_integrity.rs index 698e89d2b5..aca3470b14 100644 --- a/crates/broker/tests/pty_injection_integrity.rs +++ b/crates/broker/tests/pty_injection_integrity.rs @@ -33,10 +33,12 @@ import os, tty, time, select, json, pathlib # Canonical mode silently discards input after MAX_CANON. Real TUIs use raw mode. tty.setraw(0) mode=os.environ['FIXTURE_MODE'] +if mode == 'ack-failure': os.rmdir(os.environ['RELAY_PTY_ACK_DIR']) time.sleep(0.2) os.write(1,b'\x1b[?2004h\x1b[?2004l' if mode != 'typed' else b'') os.write(1,('OpenAI Codex\r\n› ' if os.environ['FIXTURE_CLI']=='codex' else 'Claude Code v2.1.261\r\n❯ ').encode()) data=b'' +swallowed=False while True: chunk=os.read(0,65536) with open(os.environ['TRANSCRIPT'],'ab') as f: f.write(chunk) @@ -49,6 +51,13 @@ while True: if shown: os.write(1,shown) continue data+=chunk + if mode == 'recover' and not swallowed: + shown=chunk.replace(b'\x1b[200~',b'').replace(b'\x1b[201~',b'').replace(b'\r',b'').replace(b'\n',b'') + if shown: os.write(1,shown) + if data.endswith(b'\r'): + swallowed=True + data=data.rstrip(b'\r') + continue if mode == 'hostile' and b'\x1b\x1b' in data: os.write(1,b'\x1b[2J\x1b[HTrust prompt, no composer') data=b'' @@ -206,6 +215,42 @@ fn prompt_repaints_without_native_reads_are_unconfirmed() { assert_eq!(frame["type"], "delivery_unconfirmed", "{frame}"); assert_eq!(frame["payload"]["terminal"], true); } + +#[test] +fn native_harnesses_recover_a_swallowed_submit_before_the_receipt_deadline() { + for cli in ["claude", "codex"] { + let mut fixture = Fixture::new_for_cli("recover", cli); + fixture.deliver("recover_one", "Recover the swallowed submit"); + let recovery = fixture.wait_any(&["delivery_resubmitted", "delivery_failed"], 20); + assert_eq!( + recovery["type"], "delivery_resubmitted", + "{cli}: {recovery}" + ); + let receipt = fixture.wait_any(&["delivery_verified", "delivery_failed"], 20); + assert_eq!(receipt["type"], "delivery_verified", "{cli}: {receipt}"); + assert_eq!(receipt["payload"]["evidence"], "native_mcp_ack"); + assert_eq!(receipt["payload"]["attempts"], 2); + let transcript = String::from_utf8_lossy(&fixture.transcript()).into_owned(); + assert_eq!( + transcript.matches("Recover the swallowed submit").count(), + 1 + ); + } +} + +#[test] +fn missing_receipt_directory_does_not_end_the_worker_or_drop_delivery() { + let mut fixture = Fixture::new("ack-failure"); + for id in ["first_no_receipt", "second_no_receipt"] { + fixture.deliver(id, "The worker must keep delivering"); + let frame = fixture.wait_any( + &["delivery_unconfirmed", "delivery_failed", "worker_exited"], + 20, + ); + assert_eq!(frame["type"], "delivery_unconfirmed", "{frame}"); + assert_eq!(frame["payload"]["terminal"], true); + } +} #[test] fn over_limit_and_typed_fallback_reject_before_writing() { for mode in ["echo", "typed"] { From d87bfd761a6fb146e339fee75f5cc040e4e1933f Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 21:31:21 +0530 Subject: [PATCH 5/6] fix(broker): preserve wrap launch options during MCP setup --- .../src/broker/delivery_verification.rs | 3 +- crates/broker/src/pty_delivery_ack.rs | 16 +- crates/broker/src/wrap.rs | 141 +++++++++++++++--- 3 files changed, 133 insertions(+), 27 deletions(-) diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 0c74847ee0..98f50a1542 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -152,9 +152,8 @@ impl PendingVerification { /// Defer only inconclusive outcomes while a native recipient can still acknowledge. pub(crate) fn waiting_for_native_ack(&self, cli: &str) -> bool { - let name = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); self.native_ack.is_some() - && matches!(name.trim_end_matches(".exe"), "claude" | "codex") + && crate::pty_delivery_ack::native_harness_name(cli).is_some() && !self.acceptance_expired() } diff --git a/crates/broker/src/pty_delivery_ack.rs b/crates/broker/src/pty_delivery_ack.rs index 607f67f5a7..dd814727df 100644 --- a/crates/broker/src/pty_delivery_ack.rs +++ b/crates/broker/src/pty_delivery_ack.rs @@ -13,6 +13,16 @@ use sha2::{Digest, Sha256}; pub(crate) const ACK_DIR_ENV: &str = "RELAY_PTY_ACK_DIR"; +/// Canonical MCP harness name, independent of executable case and `.exe` suffix. +pub(crate) fn native_harness_name(cli: &str) -> Option<&'static str> { + let name = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); + match name.strip_suffix(".exe").unwrap_or(&name) { + "claude" => Some("claude"), + "codex" => Some("codex"), + _ => None, + } +} + /// Owns pending receipts and private harness configs for one PTY session. pub(crate) struct NativeAckInbox(tempfile::TempDir); @@ -34,9 +44,9 @@ impl NativeAckInbox { /// MCP subprocesses do not reliably inherit the harness environment. /// Forward only this worker's acknowledgement directory in their existing config. pub(crate) fn configure_harness(&self, cli: &str, args: &mut Vec) -> io::Result<()> { - let cli = crate::cli::command_parse::normalize_cli_name(cli).to_ascii_lowercase(); + let cli = native_harness_name(cli); let directory = self.0.path().to_string_lossy().to_string(); - if cli == "codex" || cli == "codex.exe" { + if cli == Some("codex") { let names: Vec<_> = ["agent-relay", "relaycast"] .into_iter() .filter(|name| { @@ -53,7 +63,7 @@ impl NativeAckInbox { ), ]); } - } else if cli == "claude" || cli == "claude.exe" { + } else if cli == Some("claude") { for index in 0..args.len() { let equals = args[index].strip_prefix("--mcp-config="); let value = if let Some(value) = equals { diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 3f59b8a4d1..4faea59226 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -1416,6 +1416,36 @@ async fn register_broker_actions(workspace: &RelayWorkspace) { } } +/// Add native MCP configuration while preserving every user-supplied launch argument. +async fn configure_wrap_mcp_args( + cli: &str, + args: &mut Vec, + workspace: &RelayWorkspace, + base_url: Option<&str>, + workspaces_json: &str, + cwd: &std::path::Path, +) -> Result<()> { + if let Some(harness) = crate::pty_delivery_ack::native_harness_name(cli) { + let mcp_args = crate::snippets::configure_agent_relay_mcp_with_token( + harness, + &workspace.self_name, + Some(&workspace.relay_workspace_key), + base_url, + args, + cwd, + workspace + .http_client + .cached_agent_token(&workspace.self_name) + .as_deref(), + Some(workspaces_json), + Some(&workspace.workspace_id), + ) + .await?; + args.extend(mcp_args); + } + Ok(()) +} + /// Interactive wrap mode: wraps a CLI in a PTY with terminal passthrough /// while connecting to Relaycast for relay message injection. /// Usage: `agent-relay codex --full-auto` @@ -1537,28 +1567,15 @@ pub(crate) async fn run_wrap( } let native_acks = crate::pty_delivery_ack::NativeAckInbox::new()?; native_acks.expose_to_child(); - if matches!( - crate::cli::command_parse::normalize_cli_name(&resolved_cli).as_str(), - "claude" | "codex" - ) { - // Wrap sessions also need an explicit MCP server environment. Claude - // does not reliably forward the parent's environment to .mcp.json servers. - effective_cli_args = crate::snippets::configure_agent_relay_mcp_with_token( - &resolved_cli, - &default_workspace.self_name, - Some(&default_workspace.relay_workspace_key), - child_base_url.as_deref(), - &effective_cli_args, - &runtime_cwd, - default_workspace - .http_client - .cached_agent_token(&default_workspace.self_name) - .as_deref(), - Some(&child_workspaces_json), - Some(&default_workspace.workspace_id), - ) - .await?; - } + configure_wrap_mcp_args( + &resolved_cli, + &mut effective_cli_args, + &default_workspace, + child_base_url.as_deref(), + &child_workspaces_json, + &runtime_cwd, + ) + .await?; native_acks.configure_harness(&resolved_cli, &mut effective_cli_args)?; let _devin_state = crate::devin::prepare_worker_config(&resolved_cli).await?; @@ -2751,6 +2768,86 @@ mod tests { use std::io; use std::time::{Duration, Instant}; + #[tokio::test] + async fn wrap_native_mcp_setup_preserves_launch_options_and_custom_servers() { + let (ws_control_tx, _rx) = tokio::sync::mpsc::channel(1); + let workspace = super::RelayWorkspace { + workspace_id: "test-workspace".into(), + workspace_alias: None, + relay_workspace_key: "test-key".into(), + self_name: "worker".into(), + self_agent_id: "test-agent".into(), + self_names: Default::default(), + self_agent_ids: Default::default(), + http_client: crate::relaycast::RelaycastHttpClient::local_only("worker"), + ws_control_tx, + }; + for cli in ["claude", "/opt/Claude.EXE", "codex", "/opt/Codex.exe"] { + let claude = crate::pty_delivery_ack::native_harness_name(cli) == Some("claude"); + for custom in [false, true] { + let inbox = crate::pty_delivery_ack::NativeAckInbox::new().unwrap(); + let cwd = tempfile::tempdir().unwrap(); + let mut args: Vec = if claude { + [ + "--model", + "sonnet", + "--resume", + "session-id", + "--dangerously-skip-permissions", + ] + .into_iter() + .map(String::from) + .collect() + } else { + ["--model", "test-model", "--profile", "work", "--full-auto"] + .into_iter() + .map(String::from) + .collect() + }; + if custom { + if claude { + args.extend(["--mcp-config".into(), serde_json::json!({"mcpServers": { + "relaycast": {"command": "custom-relay", "env": {"RELAY_AGENT_NAME": "worker"}}, + "other": {"command": "other-server"} + }}).to_string()]); + } else { + args.extend([ + "--config".into(), + "mcp_servers.relaycast.command=\"custom-relay\"".into(), + ]); + } + } + let original = args.clone(); + super::configure_wrap_mcp_args(cli, &mut args, &workspace, None, "[]", cwd.path()) + .await + .unwrap(); + assert_eq!(&args[..original.len()], &original, "{cli}, custom={custom}"); + inbox.configure_harness(cli, &mut args).unwrap(); + assert_eq!(&args[..5], &original[..5]); + if claude { + let index = args.iter().position(|arg| arg == "--mcp-config").unwrap() + 1; + let config: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(&args[index]).unwrap()) + .unwrap(); + let server = if custom { "relaycast" } else { "agent-relay" }; + assert!(config["mcpServers"][server]["env"]["RELAY_PTY_ACK_DIR"].is_string()); + assert_eq!( + config["mcpServers"][server]["env"]["RELAY_AGENT_NAME"], + "worker" + ); + if custom { + assert_eq!(config["mcpServers"]["other"]["command"], "other-server"); + assert_eq!(config["mcpServers"]["relaycast"]["command"], "custom-relay"); + } + } else { + let server = if custom { "relaycast" } else { "agent-relay" }; + assert!(args.iter().any(|arg| arg + .starts_with(&format!("mcp_servers.{server}.env.RELAY_PTY_ACK_DIR=")))); + } + } + } + } + #[cfg(unix)] #[tokio::test] async fn devin_disables_generic_prompt_approvals_and_idle_enter() { From 63b1034a69a487aa9923add0399d08a839ed1fa3 Mon Sep 17 00:00:00 2001 From: Akarshkushwaha Date: Sun, 18 Oct 2026 23:10:20 +0530 Subject: [PATCH 6/6] refactor(broker): remove redundant native receipt plumbing --- crates/broker/src/broker/delivery_verification.rs | 6 +++--- crates/broker/src/pty_worker.rs | 4 +--- crates/broker/src/wrap.rs | 2 +- 3 files changed, 5 insertions(+), 7 deletions(-) diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 98f50a1542..1461cc6590 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -145,7 +145,7 @@ pub(crate) struct PendingVerification { impl PendingVerification { /// Check parked composers on the short interval even while waiting for a receipt. - pub(crate) fn ready_to_assess(&self, _cli: &str, window: Duration) -> bool { + pub(crate) fn ready_to_assess(&self, window: Duration) -> bool { self.native_ack.as_ref().is_some_and(|ack| ack.confirmed()) || self.injected_at.elapsed() >= window } @@ -1524,10 +1524,10 @@ mod tests { let mut verification = codex_verification("bounded delivery"); verification.native_ack = Some(inbox.register(cli, "worker").unwrap()); verification.injected_at = Instant::now() - VERIFICATION_WINDOW; - assert!(verification.ready_to_assess(cli, VERIFICATION_WINDOW)); + assert!(verification.ready_to_assess(VERIFICATION_WINDOW)); assert!(verification.waiting_for_native_ack(cli)); verification.injected_at = Instant::now(); - assert!(!verification.ready_to_assess(cli, VERIFICATION_WINDOW)); + assert!(!verification.ready_to_assess(VERIFICATION_WINDOW)); verification.verification_started_at = Instant::now() - MAX_ACCEPTANCE_LIFETIME; assert!(!verification.waiting_for_native_ack(cli)); } diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 131d4d56b3..1e8774a785 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -2918,7 +2918,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut i = 0; while i < pending_verifications.len() { - if pending_verifications[i].ready_to_assess(&resolved_cli, verification_window) { + if pending_verifications[i].ready_to_assess(verification_window) { let mut pv = pending_verifications.remove(i).unwrap(); let delivery_id = pv.delivery_id.clone(); let event_id = pv.event_id.clone(); @@ -4971,7 +4971,6 @@ mod tests { // A blanket exemption still goes back to the counter. let blanket = ActiveInjection { - native_ack: None, targeted_hold_exemption: false, ..targeted }; @@ -4982,7 +4981,6 @@ mod tests { // An injection popped without any exemption restores nothing. let unexempt = ActiveInjection { - native_ack: None, hold_exempt: false, ..blanket }; diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 4faea59226..00a02fa97b 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -2544,7 +2544,7 @@ pub(crate) async fn run_wrap( _ = verification_tick.tick() => { let mut i = 0; while i < pending_verifications.len() { - if pending_verifications[i].ready_to_assess(&resolved_cli, VERIFICATION_WINDOW) { + if pending_verifications[i].ready_to_assess(VERIFICATION_WINDOW) { let mut pv = pending_verifications.remove(i).unwrap(); pv.observe_raw_process_receipt(&echo_buffer, &resolved_cli); let snapshot = Snapshot::capture(&pty);