fix: address #1939 P2/P3 review items

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.
 

CHANGELOG.md

@@ -5,7 +5,13 @@ 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 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
 
@@ -77,7 +83,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.

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,39 +246,48 @@ fn gemini_composer_row(line: &str) -> &str {
         .unwrap_or(bordered)
 }
 
-fn current_composer(snapshot: &Snapshot, cli: &str) -> Option<String> {
+/// 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<String> {
     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") {
         if lines
             .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::<Vec<_>>()
-                .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::<Vec<_>>()
+                    .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<String> {
                 || 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<usize> = (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<String> {
+    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() {

crates/broker/src/runtime/fleet.rs

@@ -1940,8 +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 or
-/// a cleared composer 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

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,15 @@ function sendFrame(child, frame) {
 
 function option(name, fallback) {
   const index = process.argv.indexOf(name);
-  return index >= 0 ? process.argv[index + 1] : fallback;
+  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) {
   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}`);