From b8fb6681c71fd6ce49770c88917325ea0d7f98bb Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 9 Oct 2026 18:33:12 +0200 Subject: [PATCH 1/3] fix: address #1939 P2/P3 review items (changelog, docs, real-codex guards) - CHANGELOG: move the PTY delivery entry from the released 13.1.1 section to [Unreleased - Minor] / Fixed, impact-first, without the release-only sentence. 6bdec7b2 is not an ancestor of v13.2.0 or main. - AGENTS.md: restore "When the PR is ready" as its own line so items 3 and 4 render as one ordered list with nested bullets. - fleet.rs: the enqueue_delivery_ack comment names process echo as acceptance evidence alongside activity and a cleared composer. - real-codex.mjs: `--broker` without a value falls back to the friendly error instead of path.resolve(undefined); a directory is rejected by the executable-file guard. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: 2a0590dc-a82a-4174-9eff-c23d59571061 --- AGENTS.md | 10 +++++----- CHANGELOG.md | 7 +++++-- crates/broker/src/runtime/fleet.rs | 5 +++-- .../1891-codex-parked-composer-recovery/real-codex.mjs | 6 ++++-- 4 files changed, 17 insertions(+), 11 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9459ebd04..294c2e182 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -334,12 +334,12 @@ agent batches ready PRs into `trunk`, gets that one PR green, and merges it. typecheck, tests and lint this repo uses, and list the exact commands and results in the PR body. -**When the PR is ready** 3. Add the label **`mergeable`** once all of these are true: - -- The change is complete and the local checks above pass. -- Review feedback (human and bot) is addressed or answered. -- It is not a draft and does not depend on an unmerged PR. +**When the PR is ready** +3. Add the label **`mergeable`** once all of these are true: + - The change is complete and the local checks above pass. + - Review feedback (human and bot) is addressed or answered. + - It is not a draft and does not depend on an unmerged PR. 4. Remove `mergeable` if the PR stops being ready (new work, a failing check, a blocking question). The label is read live from GitHub on every sweep. diff --git a/CHANGELOG.md b/CHANGELOG.md index aebbe86cd..676e57e34 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 - Minor] + +### Fixed + +- Broker-managed PTY messages now stay pending until the harness accepts them instead of being acknowledged while still parked in the composer. A delivery that is never accepted blocks further sends to that agent, gets up to two submit-only retries without resending the body, and is then reported as failed; typing into the agent's terminal takes over and cancels the retries. ## [13.2.0] - 2026-10-08 @@ -77,7 +81,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - Broker fleet control keeps a healthy node WebSocket open when Relaycast returns retryable `d1_pressure` for `node.register` or `inventory.sync`, retrying the same frame with bounded exponential backoff while the existing registration and application-liveness deadlines remain fail-closed. -- Broker-managed PTY messages now remain pending until the harness accepts them; failed delivery blocks further sends, with up to two submit-only retries (three total attempts) instead of silently parked input. Codex recovery uses `End`, a distinct carriage return, and a guarded line-feed fallback without replaying the body; human input takes ownership and cancels recovery. Upgrade and restart brokers and PTY agents for the planned 13.2.0 rollout. - `--state-dir` on `node agent` commands, `node status`, and `node down` also accepts a fleet node directory whose broker state lives in `state/`. - An explicit `--state-dir` on `node agent` commands is no longer overridden by `RELAY_BROKER_URL` / `RELAY_BROKER_API_KEY`. - "No broker connection" errors name the path searched and whether it was the project default or `--state-dir`, instead of implying a running broker is down. diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 92d335fe9..418390dab 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -1940,8 +1940,9 @@ pub(super) async fn enqueue_delivery_ack( /// confirmation of a specific PTY injection (relay#1310: the ack must not /// fire before the worker confirms the write landed). Called with the /// `delivery_id`/`event_id` from the worker's own internal `delivery_ack` -/// event (`worker_events.rs`), which pty_worker.rs sends only after activity or -/// a cleared composer proves harness acceptance — never at write-enqueue time. +/// event (`worker_events.rs`), which pty_worker.rs sends only after activity, +/// a cleared composer, or process echo proves harness acceptance — never at +/// write-enqueue time. /// /// Returns the `(agent, up_to_seq)` to send to the engine once resolved, or /// `None` when there is nothing withheld for `delivery_id` (already diff --git a/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs b/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs index ad72a56c5..dcac3761f 100644 --- a/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs +++ b/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs @@ -18,7 +18,7 @@ */ import { constants as fsConstants } from 'node:fs'; -import { access, chmod, copyFile, mkdtemp, mkdir, rm, writeFile } from 'node:fs/promises'; +import { access, chmod, copyFile, mkdtemp, mkdir, rm, stat, writeFile } from 'node:fs/promises'; import { spawn } from 'node:child_process'; import { tmpdir } from 'node:os'; import path from 'node:path'; @@ -415,11 +415,13 @@ function sendFrame(child, frame) { function option(name, fallback) { const index = process.argv.indexOf(name); - return index >= 0 ? process.argv[index + 1] : fallback; + const value = index >= 0 ? process.argv[index + 1] : undefined; + return value === undefined || value.startsWith('--') ? fallback : value; } async function assertExecutable(candidate, name) { try { + if (!(await stat(candidate)).isFile()) throw new Error('not a file'); await access(candidate, fsConstants.X_OK); } catch { throw new Error(`${name} must be an executable file: ${candidate}`); From a80f9202550b165c0c53be1a67bdc66484627029 Mon Sep 17 00:00:00 2001 From: Miya Date: Fri, 9 Oct 2026 21:47:50 +0200 Subject: [PATCH 2/3] fix(broker): keep parked drafts with prompt-like rows detectable current_composer anchored on the last prompt-looking row, so a relay body containing a line such as `> b` cut the live composer short at that row. The tail check then missed a parked draft: observe_visible_composer never saw the echo, and assess_harness_acceptance returned Inconclusive instead of Parked, so the delivery failed without submit-key recovery. composer_holds_tail keeps the latest-anchor containment check and also accepts an earlier prompt anchor, but only when its composer ends with the body's tail at the cursor. A submitted body in the transcript above a fresh prompt does not end at the cursor and is not treated as parked. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: 2a0590dc-a82a-4174-9eff-c23d59571061 --- .../src/broker/delivery_verification.rs | 152 +++++++++++++----- 1 file changed, 114 insertions(+), 38 deletions(-) diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 55c8aa855..ce908654d 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -163,11 +163,7 @@ impl PendingVerification { if self.echo_seen { return; } - let tail = expected_tail(&self.expected_echo); - if !tail.is_empty() - && current_composer(snapshot, cli) - .is_some_and(|composer| compact_render(&composer).contains(&tail)) - { + if composer_holds_tail(snapshot, cli, &expected_tail(&self.expected_echo)) { self.echo_seen = true; self.activity_buffer.clear(); } @@ -250,12 +246,18 @@ fn gemini_composer_row(line: &str) -> &str { .unwrap_or(bordered) } -fn current_composer(snapshot: &Snapshot, cli: &str) -> Option { +/// Live-composer readings for the rows through the cursor, latest prompt +/// anchor first. Earlier anchors only matter when a draft itself contains a +/// prompt-looking row (for example a quoted `> b` line); see +/// [`composer_holds_tail`]. +fn composer_candidates(snapshot: &Snapshot, cli: &str) -> Vec { let plain = snapshot.to_plain_through_cursor(); let lines: Vec<_> = plain.lines().collect(); - let end = snapshot.cursor.0.checked_sub(1)? as usize; + let Some(end) = snapshot.cursor.0.checked_sub(1).map(|row| row as usize) else { + return Vec::new(); + }; if end >= lines.len() { - return None; + return Vec::new(); } let lower = cli.to_ascii_lowercase(); if lower.contains("gemini") { @@ -263,26 +265,29 @@ fn current_composer(snapshot: &Snapshot, cli: &str) -> Option { .get(end) .is_some_and(|line| cursor_row_is_activity(cli, line)) { - return None; + return Vec::new(); } // Gemini renders its live input with a `> ` prefix (inside a `│` // border on terminals that cannot use the background-colour frame). // Anchor to that boundary instead of treating arbitrary transcript // rows near the cursor as the composer. Keeping every row through the // cursor also preserves long, wrapped drafts beyond four rows. - let start = lines.iter().take(end + 1).rposition(|line| { - let input = gemini_composer_row(line); - input == ">" || input.starts_with("> ") - })?; - return Some( - lines[start..=end] - .iter() - .map(|line| gemini_composer_row(line)) - .collect::>() - .join("\n"), - ); + return (0..=end) + .rev() + .filter(|&start| { + let input = gemini_composer_row(lines[start]); + input == ">" || input.starts_with("> ") + }) + .map(|start| { + lines[start..=end] + .iter() + .map(|line| gemini_composer_row(line)) + .collect::>() + .join("\n") + }) + .collect(); } - let is_prompt = |line: &&str| { + let is_prompt = |line: &str| { let trimmed = line.trim_start(); if lower.contains("codex") { trimmed == "›" @@ -309,20 +314,56 @@ fn current_composer(snapshot: &Snapshot, cli: &str) -> Option { || trimmed.starts_with("❯ ") } }; - let start = lines.iter().take(end + 1).rposition(is_prompt)?; - let composer_lines = &lines[start..=end]; - if composer_lines.len() > 1 - && composer_lines - .last() - .is_some_and(|line| cursor_row_is_activity(cli, line)) + let starts: Vec = (0..=end) + .rev() + .filter(|&row| is_prompt(lines[row])) + .collect(); + let Some(&latest) = starts.first() else { + return Vec::new(); + }; + if lines + .get(end) + .is_some_and(|line| cursor_row_is_activity(cli, line)) { - // The latest prompt-looking line is part of the submitted transcript; - // a distinctive harness status row owns the cursor. Do not use the - // broader ActivityDetector patterns here because ordinary multiline - // draft text can contain Tool:, Write(, or shell prompts. - return None; + if latest < end { + // The latest prompt-looking line is part of the submitted + // transcript; a distinctive harness status row owns the cursor. Do + // not use the broader ActivityDetector patterns here because + // ordinary multiline draft text can contain Tool:, Write(, or shell + // prompts. + return Vec::new(); + } + return vec![lines[latest].to_string()]; + } + starts + .into_iter() + .map(|start| lines[start..=end].join("\n")) + .collect() +} + +fn current_composer(snapshot: &Snapshot, cli: &str) -> Option { + composer_candidates(snapshot, cli).into_iter().next() +} + +/// Whether the delivery's tail is still in the live composer. +/// +/// The latest prompt anchor keeps the original containment check. A draft +/// can itself contain prompt-looking rows, which moves that anchor into the +/// body, so an earlier anchor also counts, but only when the composer it +/// yields *ends* with the tail at the cursor. A submitted body in the +/// transcript above a fresh prompt never ends at the cursor. +fn composer_holds_tail(snapshot: &Snapshot, cli: &str, tail: &str) -> bool { + if tail.is_empty() { + return false; + } + let mut candidates = composer_candidates(snapshot, cli) + .into_iter() + .map(|composer| compact_render(&composer)); + match candidates.next() { + Some(latest) if latest.contains(tail) => true, + Some(_) => candidates.any(|composer| composer.ends_with(tail)), + None => false, } - Some(composer_lines.join("\n")) } fn cursor_row_is_activity(cli: &str, line: &str) -> bool { @@ -391,11 +432,7 @@ pub(crate) fn assess_harness_acceptance( snapshot: &Snapshot, ) -> HarnessAcceptance { let tail = expected_tail(&verification.expected_echo); - let parked = !tail.is_empty() - && current_composer(snapshot, cli) - .map(|composer| compact_render(&composer).contains(&tail)) - .unwrap_or(false); - if parked { + if composer_holds_tail(snapshot, cli, &tail) { return HarnessAcceptance::Parked; } if let Some(pattern) = verification.accepted_activity() { @@ -830,6 +867,45 @@ mod tests { pty.shutdown().unwrap(); } + #[cfg(unix)] + #[tokio::test] + async fn prompt_like_body_rows_do_not_hide_a_parked_draft() { + let expected = "Relay message from Lead [evt]: quote this\n> b\nthen continue"; + let screen = "❯ Relay message from Lead [evt]: quote this\n> b\nthen continue"; + let (pty, snapshot) = codex_snapshot(screen).await; + let mut verification = codex_verification(expected); + verification.echo_seen = false; + verification.detector = ActivityDetector::for_cli("claude"); + + verification.observe_visible_composer(&snapshot, "claude"); + assert!( + verification.echo_seen, + "a `> ` row inside the body must not cut the draft off from its prompt" + ); + assert_eq!( + assess_harness_acceptance("claude", &verification, &snapshot), + HarnessAcceptance::Parked + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn submitted_prompt_like_body_above_a_fresh_prompt_is_not_parked() { + let expected = "Relay message from Lead [evt]: quote this\n> b\nthen continue"; + let screen = "❯ Relay message from Lead [evt]: quote this\n> b\nthen continue\n\nDone.\n❯ "; + let (pty, snapshot) = codex_snapshot(screen).await; + let mut verification = codex_verification(expected); + verification.detector = ActivityDetector::for_cli("claude"); + + assert_ne!( + assess_harness_acceptance("claude", &verification, &snapshot), + HarnessAcceptance::Parked, + "the transcript copy of a submitted body is not the live composer" + ); + pty.shutdown().unwrap(); + } + #[cfg(unix)] #[tokio::test] async fn claude_interrupt_status_is_activity_not_a_parked_transcript() { From 5447d54c53d0ba8a9cb61e5ff37439c7ee626046 Mon Sep 17 00:00:00 2001 From: Miya Date: Sat, 10 Oct 2026 02:25:12 +0200 Subject: [PATCH 3/3] fix: address #1955 review (option guard, changelog split, ack comment) - real-codex.mjs: a supplied option with no value now fails with " requires a value" instead of falling back, so `--broker --timeout-seconds 20` can no longer silently run the broker from RELAY_REAL_CODEX_BROKER_BINARY. - CHANGELOG: split the PTY delivery entry into three impact-first bullets and state the limit as three total attempts. - fleet.rs: process echo is acceptance evidence only for a plain `cat` recipient, where it proves transport receipt, not harness acceptance. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: 2a0590dc-a82a-4174-9eff-c23d59571061 --- CHANGELOG.md | 4 +++- crates/broker/src/runtime/fleet.rs | 7 ++++--- .../1891-codex-parked-composer-recovery/real-codex.mjs | 6 ++++-- 3 files changed, 11 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 676e57e34..4641c90cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Broker-managed PTY messages now stay pending until the harness accepts them instead of being acknowledged while still parked in the composer. A delivery that is never accepted blocks further sends to that agent, gets up to two submit-only retries without resending the body, and is then reported as failed; typing into the agent's terminal takes over and cancels the retries. +- Broker-managed PTY messages now stay pending until the harness accepts them, instead of being acknowledged while still parked in the composer. +- A PTY delivery the harness never accepts blocks further sends to that agent and is retried by pressing submit only, never by resending the body, for up to three total attempts before it is reported as failed. +- Typing into an agent's terminal while a delivery is being retried takes over the terminal and cancels the retries. ## [13.2.0] - 2026-10-08 diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 418390dab..faaec8168 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -1940,9 +1940,10 @@ pub(super) async fn enqueue_delivery_ack( /// confirmation of a specific PTY injection (relay#1310: the ack must not /// fire before the worker confirms the write landed). Called with the /// `delivery_id`/`event_id` from the worker's own internal `delivery_ack` -/// event (`worker_events.rs`), which pty_worker.rs sends only after activity, -/// a cleared composer, or process echo proves harness acceptance — never at -/// write-enqueue time. +/// event (`worker_events.rs`), which pty_worker.rs sends only after activity +/// or a cleared composer proves harness acceptance (or, for a plain `cat` +/// recipient with no harness, after its PTY echo proves transport receipt) — +/// never at write-enqueue time. /// /// Returns the `(agent, up_to_seq)` to send to the engine once resolved, or /// `None` when there is nothing withheld for `delivery_id` (already diff --git a/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs b/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs index dcac3761f..a8c47e72b 100644 --- a/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs +++ b/tests/relayflows/cases/1891-codex-parked-composer-recovery/real-codex.mjs @@ -415,8 +415,10 @@ function sendFrame(child, frame) { function option(name, fallback) { const index = process.argv.indexOf(name); - const value = index >= 0 ? process.argv[index + 1] : undefined; - return value === undefined || value.startsWith('--') ? fallback : value; + if (index < 0) return fallback; + const value = process.argv[index + 1]; + if (value === undefined || value.startsWith('--')) throw new Error(`${name} requires a value.`); + return value; } async function assertExecutable(candidate, name) {