diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 4fa5768d6..ac4ce1e51 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -334,7 +334,7 @@ jobs: # Build all packages once, version them, and upload build: name: Build & Version - runs-on: ubuntu-latest + runs-on: ${{ vars.CI_LINUX_RUNNER || 'ubuntu-latest' }} outputs: new_version: ${{ steps.bump.outputs.new_version }} is_prerelease: ${{ steps.bump.outputs.is_prerelease }} diff --git a/.github/workflows/rust-ci.yml b/.github/workflows/rust-ci.yml index ba1860ce3..cecf6c814 100644 --- a/.github/workflows/rust-ci.yml +++ b/.github/workflows/rust-ci.yml @@ -26,7 +26,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.rust_changed == 'true') - runs-on: ${{ matrix.os }} + runs-on: ${{ matrix.os == 'ubuntu-latest' && (vars.CI_LINUX_RUNNER || 'ubuntu-latest') || matrix.os }} strategy: fail-fast: false matrix: @@ -70,7 +70,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.rust_changed == 'true') - runs-on: ${{ matrix.os }} + runs-on: ${{ matrix.os == 'ubuntu-latest' && (vars.CI_LINUX_RUNNER || 'ubuntu-latest') || matrix.os }} strategy: fail-fast: false matrix: @@ -106,7 +106,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.rust_changed == 'true') - runs-on: ${{ matrix.os }} + runs-on: ${{ matrix.os == 'ubuntu-latest' && (vars.CI_LINUX_RUNNER || 'ubuntu-latest') || matrix.os }} strategy: fail-fast: false matrix: @@ -127,7 +127,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.rust_changed == 'true') - runs-on: ubuntu-latest + runs-on: ${{ vars.CI_LINUX_RUNNER || 'ubuntu-latest' }} steps: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml index a267819bf..b33d16647 100644 --- a/.github/workflows/test-build.yml +++ b/.github/workflows/test-build.yml @@ -23,7 +23,7 @@ jobs: test-build-script: if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository name: Test build-bun.sh - runs-on: ${{ matrix.os }} + runs-on: ${{ matrix.os == 'ubuntu-latest' && (vars.CI_LINUX_RUNNER || 'ubuntu-latest') || matrix.os }} strategy: fail-fast: false matrix: @@ -33,6 +33,12 @@ jobs: - name: Checkout code uses: actions/checkout@v4 + # StarSling Linux runners (vars.CI_LINUX_RUNNER) preinstall rustc 1.92, older + # than GitHub's ubuntu-24.04 image (1.99); 1.92 cannot compile the + # broker (stable-only atomic try_update). Install the toolchain explicitly. + - name: Setup Rust toolchain + uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable + - name: Setup Bun uses: oven-sh/setup-bun@v2 with: @@ -111,7 +117,7 @@ jobs: test-build-edge-cases: if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository name: Test edge cases - runs-on: ubuntu-latest + runs-on: ${{ vars.CI_LINUX_RUNNER || 'ubuntu-latest' }} steps: - name: Checkout code diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b1dfe2de7..c7872844b 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -29,7 +29,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.node_changed == 'true') - runs-on: ${{ matrix.os }} + runs-on: ${{ matrix.os == 'ubuntu-latest' && (vars.CI_LINUX_RUNNER || 'ubuntu-latest') || matrix.os }} strategy: matrix: os: [ubuntu-latest, macos-latest] @@ -40,6 +40,12 @@ jobs: - name: Checkout repository uses: actions/checkout@v4 + # StarSling Linux runners (vars.CI_LINUX_RUNNER) preinstall rustc 1.92, older + # than GitHub's ubuntu-24.04 image (1.99); 1.92 cannot compile the + # broker (stable-only atomic try_update). Install the toolchain explicitly. + - name: Setup Rust toolchain + uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable + - name: Setup Node.js ${{ matrix.node-version }} uses: actions/setup-node@v4 with: @@ -61,6 +67,49 @@ jobs: - name: Run tests run: npm test + # The PR-proof Landlock / devpts / no_new_privs sandbox tests skip themselves + # (it.skipIf(!HAS_TRUSTED_LANDLOCK_RUNTIME)) on kernels that cannot host the + # sandbox, which includes the StarSling runners that `test` above may run on + # via vars.CI_LINUX_RUNNER. This job stays pinned to GitHub's ubuntu-latest + # (never the runner variable) and fails if any test in the file is skipped, + # so those tests can never silently stop running. + sandbox-proof: + name: PR-proof sandbox tests (Landlock, ubuntu-latest) + needs: changes + if: >- + (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && + (needs.changes.outputs.node_changed == 'true') + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: '22.14.0' + cache: 'npm' + + - name: Install dependencies + run: npm ci + + - name: Ensure rollup optional dependencies are installed + run: npm install --no-save rollup || true + + - name: Run PR-proof sandbox tests + run: npx vitest run tests/fixtures/pr-proof-contract.test.ts --reporter=default --reporter=json --outputFile.json=pr-proof-results.json + + - name: Require every PR-proof test to have run + run: | + node -e ' + const r = require("./pr-proof-results.json"); + const all = r.testResults.flatMap((f) => f.assertionResults); + const notRun = all.filter((t) => t.status !== "passed" && t.status !== "failed"); + console.log("pr-proof-contract: " + all.length + " tests, " + notRun.length + " not run"); + for (const t of notRun) console.log("::error::not run (" + t.status + "): " + t.fullName); + if (all.length === 0 || notRun.length > 0) process.exit(1); + ' + windows-credential-acl: name: Windows credential ACL validation needs: changes @@ -100,11 +149,17 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.node_changed == 'true') - runs-on: ubuntu-latest + runs-on: ${{ vars.CI_LINUX_RUNNER || 'ubuntu-latest' }} steps: - name: Checkout repository uses: actions/checkout@v4 + # StarSling Linux runners (vars.CI_LINUX_RUNNER) preinstall rustc 1.92, older + # than GitHub's ubuntu-24.04 image (1.99); 1.92 cannot compile the + # broker (stable-only atomic try_update). Install the toolchain explicitly. + - name: Setup Rust toolchain + uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c # stable + - name: Setup Node.js uses: actions/setup-node@v4 with: @@ -130,7 +185,7 @@ jobs: if: >- (github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository) && (needs.changes.outputs.node_changed == 'true') - runs-on: ubuntu-latest + runs-on: ${{ vars.CI_LINUX_RUNNER || 'ubuntu-latest' }} steps: - name: Checkout repository uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index 6732aa392..9d80a118b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,11 +10,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - `agent-relay integration subscribe --to self` routes provider events to the broker worker's registered identity, allowing fleet workers to subscribe without desktop session detection. +- Broker events report PTY submit recovery: `delivery_unconfirmed` (no acceptance evidence for a write, with `attempts` / `max_attempts`; `terminal: true` when the delivery ends unconfirmed) and `delivery_resubmitted` (a submit-only retry, with `attempt` and `strategy`); `delivery_verified` adds `evidence` and `attempts`. `@agent-relay/harness-driver` and the Python and Swift SDK protocol types include both new events. + +### Changed + +- A PTY delivery the broker can neither confirm nor see still parked in the agent's input box is now delivered as unconfirmed (a terminal `delivery_unconfirmed` with `outcome: "unconfirmed"`) instead of failing, and later messages to that agent are no longer held behind it. Agents whose terminal shows no acceptance signal keep receiving messages. +- A confirmed `fleet spawn --task` whose task was written but never visibly accepted now succeeds (`spawned: true`) with `warning.code: "spawn_task_unconfirmed"`, instead of failing with `spawn_task_unconfirmed` and exit 8. Only a task that could not be delivered fails the spawn, with `spawn_task_failed`. ### Fixed - `agent-relay fleet spawn` explains how to reuse a retained agent name with `fleet release --delete-agent --wait` when registration fails with `agent_already_exists`. - The installer's Quick Start now prints `agent-relay node up --background` and `agent-relay node down`, which exist, instead of the removed `agent-relay up` form that failed with "unknown command 'up'". An install pinned with `AGENT_RELAY_VERSION` to a release older than 9.2.2 still prints the top-level `up`/`down` forms that release has. +- A PTY message still visible in the agent's input box after submit is retried by pressing submit only, never by resending the body, for up to three total attempts. Later messages queue behind it instead of being typed on top of it, and if it is still stuck after 60 seconds the broker reports a terminal `agent_blocked_on_send` with reason `failed_draft_latched`. +- Typing into an agent's terminal while a delivery is being retried takes over the terminal and cancels the retries. ## [13.3.0] - 2026-10-10 diff --git a/crates/broker/src/attachments.rs b/crates/broker/src/attachments.rs index 473e2c2f7..9d2260ae2 100644 --- a/crates/broker/src/attachments.rs +++ b/crates/broker/src/attachments.rs @@ -1347,9 +1347,23 @@ mod tests { std::fs::create_dir_all(&primary).unwrap(); std::fs::set_permissions(&primary, std::fs::Permissions::from_mode(0o555)).unwrap(); let fallback = temp.path().join("fallback"); + // Some CI runners bypass directory permission bits (root, a granted + // CAP_DAC_OVERRIDE, or a filesystem that ignores modes). There the + // 0o555 directory really is writable, so accepting it is correct and + // refusal must be exercised with a root that cannot be written at all. + let mode_enforced = std::fs::write(primary.join(".mode-probe"), b"").is_err(); let refused = prepare_root(&primary).await.is_err(); std::fs::set_permissions(&primary, std::fs::Permissions::from_mode(0o755)).unwrap(); - assert!(refused, "a read-only attachments root must be refused"); + if mode_enforced { + assert!(refused, "a read-only attachments root must be refused"); + } else { + let not_a_directory = temp.path().join("not-a-directory"); + std::fs::write(¬_a_directory, b"").unwrap(); + assert!( + prepare_root(¬_a_directory).await.is_err(), + "an unwritable attachments root must be refused" + ); + } assert!(prepare_root(&fallback).await.is_ok()); } diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 591092618..0ceb1dd08 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(); } @@ -186,6 +182,12 @@ impl PendingVerification { if let Some(index) = clean.rfind(&expected_echo) { self.activity_buffer .push_str(&clean[index + expected_echo.len()..]); + } else if let Some(end) = + end_of_compact_match(&clean, &expected_tail(&self.expected_echo)) + { + // A wrapped echo matched only by its compact tail still + // carries any same-read turn marker after that tail. + self.activity_buffer.push_str(&clean[end..]); } } else if self.echo_seen { self.activity_buffer.push_str(text); @@ -225,6 +227,73 @@ fn compact_render(text: &str) -> String { .collect() } +/// Byte offset in `text` just past the last occurrence of `compact_needle` +/// when whitespace in `text` is ignored, matching [`compact_render`]. +fn end_of_compact_match(text: &str, compact_needle: &str) -> Option { + if compact_needle.is_empty() { + return None; + } + let mut compact = String::with_capacity(text.len()); + let mut ends = Vec::with_capacity(text.len()); + for (index, character) in text.char_indices() { + if !character.is_whitespace() { + compact.push(character); + ends.push(index + character.len_utf8()); + } + } + let start = compact.rfind(compact_needle)?; + // `ends` is indexed by compact character, so count characters (not bytes) + // through the final matched one. + let matched_chars = compact[..start + compact_needle.len()].chars().count(); + ends.get(matched_chars - 1).copied() +} + +/// Whether the delivery's compact tail may still be in an editor around the +/// cursor: ending just before it, straddling it, or anywhere after it (the +/// cursor can sit at the start of a draft longer than the tail). Any of these +/// means the editor may still hold the draft, regardless of which prompt +/// glyph (or body line that looks like one) the composer parser anchored on. +fn tail_near_cursor(snapshot: &Snapshot, tail: &str) -> bool { + if tail.is_empty() { + return false; + } + let before = compact_render(&snapshot.to_plain_through_cursor()); + let after = compact_render(&snapshot.to_plain_from_cursor()); + if before.ends_with(tail) || after.contains(tail) { + return true; + } + tail.char_indices() + .skip(1) + .any(|(split, _)| before.ends_with(&tail[..split]) && after.starts_with(&tail[split..])) +} + +/// Prompt glyphs and box-drawing borders that frame a composer row. +const CURSOR_ROW_CHROME: &[char] = &[ + '›', '❯', '❭', '>', '$', '│', '┃', '║', '▌', '▎', '▏', '╎', '╏', '┆', '┇', +]; + +/// Whether the cells right of the cursor on its row are part of this body. +/// A cursor parked at the start of a draft whose tail has scrolled out of +/// view shows body text there; harness chrome (a placeholder or status hint) +/// does not appear in the body and is ignored. A placeholder whose exact text +/// also appears in the body still reads as a draft: that errs inconclusive, +/// never toward confirming an unsent message. +fn body_text_right_of_cursor(snapshot: &Snapshot, expected_echo: &str) -> bool { + let from_cursor = snapshot.to_plain_from_cursor(); + let compact_row = compact_render(from_cursor.lines().next().unwrap_or_default()); + // Prompt glyphs and box borders can bracket the text under the cursor. + // Strip only those: a draft can itself start with punctuation or symbols. + let row = compact_row.trim_matches(CURSOR_ROW_CHROME); + if row.is_empty() { + return false; + } + let probe_end = row + .char_indices() + .nth(32) + .map_or(row.len(), |(index, _)| index); + compact_render(expected_echo).contains(&row[..probe_end]) +} + fn is_cat_process(cli: &str) -> bool { cli.rsplit(['/', '\\']) .next() @@ -251,12 +320,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") { @@ -264,26 +339,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 == "›" @@ -310,20 +388,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 { @@ -380,25 +494,73 @@ fn composer_is_idle(snapshot: &Snapshot, cli: &str) -> bool { .unwrap_or(false) } +/// Whether the cursor row's text left of the cursor is the end of this body. +/// A partially written or unsubmitted draft leaves its latest characters just +/// before the cursor; prompt glyphs and borders are ignored. Short rows are not +/// evidence: a handful of characters can match any body by coincidence. +fn body_text_left_of_cursor(snapshot: &Snapshot, expected_echo: &str) -> bool { + let through_cursor = snapshot.to_plain_through_cursor(); + let compact_row = compact_render(through_cursor.lines().last().unwrap_or_default()); + let row = compact_row.trim_matches(CURSOR_ROW_CHROME); + if row.chars().count() < 8 { + return false; + } + let probe_start = row + .char_indices() + .rev() + .nth(31) + .map_or(0, |(index, _)| index); + compact_render(expected_echo).contains(&row[probe_start..]) +} + +/// Positive evidence that a delivery's body is still typed in the composer: +/// its tail is in the parsed composer or around the cursor, or body text sits +/// directly left or right of the cursor. Only this arms the failed-draft +/// latch; a delivery the broker merely cannot verify never does. +pub(crate) fn draft_visible_at_cursor(snapshot: &Snapshot, cli: &str, expected_echo: &str) -> bool { + let tail = expected_tail(expected_echo); + if tail.is_empty() { + return false; + } + composer_holds_tail(snapshot, cli, &tail) + || tail_near_cursor(snapshot, &tail) + || body_text_right_of_cursor(snapshot, expected_echo) + || body_text_left_of_cursor(snapshot, expected_echo) +} + +/// Whether a terminally failed delivery's body has provably left the composer. +/// +/// A failed body can remain typed in the editor. Injecting the next delivery +/// on top of it would append to that draft or submit both as one turn, so the +/// worker holds injection until the composer is proven idle. The tail being +/// absent from the viewport is not proof: a long draft can scroll it out of +/// view. An operator `flush_injections` is the explicit override. +/// +/// An idle-looking composer is not enough on its own: a body whose last line +/// is a bare prompt glyph parses as an idle prompt while the whole draft is +/// still in the editor, so the tail must also be absent from every composer +/// candidate that ends at the cursor. +pub(crate) fn failed_draft_released(cli: &str, expected_echo: &str, snapshot: &Snapshot) -> bool { + let tail = expected_tail(expected_echo); + tail.is_empty() + || (composer_is_idle(snapshot, cli) && !composer_holds_tail(snapshot, cli, &tail)) +} + /// Distinguish terminal echo from actual harness acceptance. /// /// Activity is definitive acceptance. A body is considered parked only when /// its tail is still in the live composer at the cursor. Once echo was seen, /// a proven empty composer is also acceptance. Every other state is /// inconclusive and must fail or wait — never blindly press a key. Harnesses -/// without explicit activity markers fall back to echo-plus-body-departure: -/// once their echoed body is no longer parked, the turn was accepted. +/// without explicit activity markers need post-echo output once the body has +/// left the cursor; see the generic branch below. pub(crate) fn assess_harness_acceptance( cli: &str, verification: &PendingVerification, 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() { @@ -413,8 +575,30 @@ pub(crate) fn assess_harness_acceptance( if verification.echo_seen && is_cat_process(cli) { return HarnessAcceptance::Accepted("process_echo".to_string()); } - if verification.echo_seen && !verification.detector.has_explicit_patterns() { - return HarnessAcceptance::Accepted("echo_left_composer".to_string()); + // Harnesses without a turn-start marker fall back to post-echo output, but + // a repaint of a still-unsent draft is output too. Only accept it once the + // body has left the cursor: a tail ending at, straddling, or anywhere after + // the cursor, or any text to the cursor's right (a draft whose tail has + // scrolled out of view), stays inconclusive. Placeholder or status text + // on the cursor row is not body text and does not block acceptance. + if verification.echo_seen + && !verification.detector.has_explicit_patterns() + && !tail_near_cursor(snapshot, &tail) + && !body_text_right_of_cursor(snapshot, &verification.expected_echo) + { + if let Some(pattern) = verification + .detector + .detect_activity(&verification.activity_buffer, &verification.expected_echo) + { + return HarnessAcceptance::Accepted(format!("activity:{pattern}")); + } + } + // A draft still visible at the cursor is StuckInComposer even when the + // through-cursor composer parse missed it (the cursor sits at the start or + // middle of the draft). Treating it as Unconfirmed would leave it unlatched + // and let the next delivery be typed onto it. + if draft_visible_at_cursor(snapshot, cli, &verification.expected_echo) { + return HarnessAcceptance::Parked; } HarnessAcceptance::Inconclusive } @@ -732,7 +916,7 @@ mod tests { verification.echo_seen = false; assert_eq!( assess_harness_acceptance("cat", &verification, &snapshot), - HarnessAcceptance::Inconclusive, + HarnessAcceptance::Parked, ); let mut output = VerificationOutput::default(); output.push_str("raw PTY echo: LOCAL_WORK_PROOF\r\n"); @@ -850,6 +1034,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() { @@ -923,25 +1146,201 @@ mod tests { #[cfg(unix)] #[tokio::test] - async fn text_right_of_cursor_cannot_look_like_a_cleared_composer() { - let expected = "Relay message from Lead [evt]: hidden draft"; - let screen = format!("› {expected}\x1b[1;3H"); - let (pty, snapshot) = codex_snapshot(&screen).await; - let verification = codex_verification(expected); + async fn a_visible_draft_with_the_cursor_at_its_start_or_middle_is_stuck() { + let expected = "Relay message from Lead [evt]: hidden draft that is still parked here"; + for cursor in ["\x1b[1;3H", "\x1b[1;25H"] { + let screen = format!("› {expected}{cursor}"); + let (pty, snapshot) = codex_snapshot(&screen).await; + let verification = codex_verification(expected); + + assert_eq!( + assess_harness_acceptance("codex", &verification, &snapshot), + HarnessAcceptance::Parked, + "a draft still in the composer is StuckInComposer, never Unconfirmed ({cursor:?})" + ); + pty.shutdown().unwrap(); + } + } + + #[cfg(unix)] + #[tokio::test] + async fn generic_post_echo_output_accepts_after_body_leaves_composer() { + let expected = "Relay message from Lead [evt]: fix idle injection"; + let (pty, snapshot) = codex_snapshot("Processing accepted turn").await; + let mut verification = codex_verification(expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification + .activity_buffer + .push_str("Processing accepted turn"); assert_eq!( - assess_harness_acceptance("codex", &verification, &snapshot), - HarnessAcceptance::Inconclusive, - "right-side draft text prevents composer-cleared acceptance while recovery stays fail-closed" + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Accepted("activity:any_output".to_string()) ); pty.shutdown().unwrap(); } #[cfg(unix)] #[tokio::test] - async fn generic_post_echo_output_accepts_after_body_leaves_composer() { + async fn generic_repaint_cannot_confirm_a_draft_straddling_the_cursor() { let expected = "Relay message from Lead [evt]: fix idle injection"; - let (pty, snapshot) = codex_snapshot("Processing accepted turn").await; + // The cursor sits inside the draft, so the parser's through-cursor + // composer cannot see the tail, yet the body is still unsent. + let (pty, snapshot) = codex_snapshot(&format!("› {expected}\x1b[1;20H")).await; + let mut verification = codex_verification(expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification + .activity_buffer + .push_str("composer repaint after echo"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Parked, + "generic output must not confirm a body that still sits at the cursor" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn opencode_output_cannot_confirm_a_draft_at_the_cursor() { + let expected = "Relay message from Lead [evt]: opencode draft"; + let (pty, snapshot) = codex_snapshot(&format!("┃ {expected}")).await; + let mut verification = codex_verification(expected); + verification.detector = ActivityDetector::for_cli("opencode"); + verification + .activity_buffer + .push_str("Thinking about the tool"); + + assert_eq!( + assess_harness_acceptance("opencode", &verification, &snapshot), + HarnessAcceptance::Parked, + "OpenCode transcript words must not confirm a typed-but-unsent body" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn failed_draft_ending_in_a_bare_prompt_line_stays_latched() { + // The failed body's last line is a bare prompt glyph. The composer + // parser anchors on that line and sees an idle prompt, but the whole + // draft is still in the editor and ends at the cursor. + for (cli, glyph) in [("codex", "›"), ("claude", ">")] { + let expected = format!("Relay message from Lead [evt]: quote this\n{glyph}"); + let (pty, snapshot) = codex_snapshot(&format!( + "{glyph} Relay message from Lead [evt]: quote this\n{glyph}" + )) + .await; + assert!( + !failed_draft_released(cli, &expected, &snapshot), + "{cli}: a draft ending in a bare `{glyph}` row must not release the latch" + ); + pty.shutdown().unwrap(); + } + } + + #[cfg(unix)] + #[tokio::test] + async fn failed_draft_latch_holds_until_the_composer_releases_it() { + let expected = "Relay message from Lead [evt]: parked after recovery"; + let (parked_pty, parked) = codex_snapshot(&format!("› {expected}")).await; + assert!(!failed_draft_released("codex", expected, &parked)); + parked_pty.shutdown().unwrap(); + + let (idle_pty, idle) = codex_snapshot("› Ask Codex to do anything").await; + assert!(failed_draft_released("codex", expected, &idle)); + idle_pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn generic_output_cannot_confirm_a_long_draft_with_cursor_at_its_start() { + let expected = format!("Relay message from Lead [evt]: {}", "x".repeat(170)); + let (pty, snapshot) = codex_snapshot(&format!("› {expected}\x1b[1;3H")).await; + let mut verification = codex_verification(&expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification.activity_buffer.push_str("repaint"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Parked, + "a tail far to the right of the cursor is still an unsent draft" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn generic_output_cannot_confirm_a_draft_whose_tail_scrolled_away() { + // Only the draft's head is in the viewport, right of the cursor; its + // tail has scrolled out of view, so no tail match is possible. + let expected = format!( + "Relay message from Lead [evt]: head {} TAIL_SCROLLED_OFF", + "y".repeat(160) + ); + let (pty, snapshot) = + codex_snapshot("› Relay message from Lead [evt]: head\x1b[1;3H").await; + let mut verification = codex_verification(&expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification.activity_buffer.push_str("repaint"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Parked, + "body text right of the cursor is an unsent draft even without its tail" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn bordered_draft_head_right_of_cursor_is_stuck_in_the_composer() { + // The cursor sits on a box border; the draft's head follows it and its + // tail has scrolled out of view. + let expected = format!( + "Relay message from Lead [evt]: head {} TAIL_SCROLLED_OFF", + "y".repeat(160) + ); + let (pty, snapshot) = + codex_snapshot("┃ Relay message from Lead [evt]: head ┃\x1b[1;1H").await; + let mut verification = codex_verification(&expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification.activity_buffer.push_str("repaint"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Parked, + "border chrome around the cursor must not hide an unsent draft" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn symbolic_draft_head_right_of_cursor_is_stuck_in_the_composer() { + // The visible head is punctuation only; the tail has scrolled away. + let expected = format!("~~~ *** ### {} TAIL_SCROLLED_OFF", "y".repeat(160)); + let (pty, snapshot) = codex_snapshot("┃ ~~~ *** ### ┃\x1b[1;1H").await; + let mut verification = codex_verification(&expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification.activity_buffer.push_str("repaint"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Parked, + "a draft head made of symbols is still body text" + ); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn cursor_row_placeholder_does_not_block_generic_acceptance() { + let expected = "Relay message from Lead [evt]: fix idle injection"; + let (pty, snapshot) = + codex_snapshot("Processing accepted turn\n› Type a message\x1b[2;3H").await; + assert!(snapshot.has_visible_text_at_or_after_cursor()); let mut verification = codex_verification(expected); verification.detector = ActivityDetector::for_cli("muse"); verification @@ -950,11 +1349,75 @@ mod tests { assert_eq!( assess_harness_acceptance("muse", &verification, &snapshot), - HarnessAcceptance::Accepted("echo_left_composer".to_string()) + HarnessAcceptance::Accepted("activity:any_output".to_string()), + "placeholder chrome right of the cursor is not an unsent draft" ); pty.shutdown().unwrap(); } + #[cfg(unix)] + #[tokio::test] + async fn a_tail_missing_from_the_viewport_does_not_release_the_latch() { + // The failed draft's head is visible but its tail has scrolled away; + // only a provably idle composer may release the latch. + let expected = "Relay message from Lead [evt]: head ... TAIL_SCROLLED_OFF"; + let (pty, snapshot) = codex_snapshot("› Relay message from Lead [evt]: head").await; + assert!(!failed_draft_released("codex", expected, &snapshot)); + pty.shutdown().unwrap(); + } + + #[cfg(unix)] + #[tokio::test] + async fn draft_visibility_is_positive_evidence_only() { + let expected = "Relay message from Lead [evt]: please finish the parked draft"; + let (typed_pty, typed) = codex_snapshot(&format!("› {expected}")).await; + assert!( + draft_visible_at_cursor(&typed, "codex", expected), + "draft left of the cursor" + ); + typed_pty.shutdown().unwrap(); + + let (cleared_pty, cleared) = codex_snapshot("› Ask Codex to do anything\x1b[1;3H").await; + assert!( + !draft_visible_at_cursor(&cleared, "codex", expected), + "an idle composer" + ); + cleared_pty.shutdown().unwrap(); + + let (other_pty, other) = codex_snapshot("› an unrelated human draft here").await; + assert!( + !draft_visible_at_cursor(&other, "codex", expected), + "someone else's text" + ); + other_pty.shutdown().unwrap(); + } + + #[test] + fn wrapped_echo_preserves_same_chunk_acceptance_activity() { + let expected = "Relay message from Lead [evt]: Reply with exactly WRAPPED_CODEX_ACK"; + let mut verification = codex_verification(expected); + verification.echo_seen = false; + let mut output = VerificationOutput::default(); + let chunk = "Relay message from Lead [evt]: Reply with exactly\r\n WRAPPED_CODEX_ACK\r\nWorking (1s • esc to interrupt)"; + output.push_str(chunk); + verification.observe(&output, chunk); + + assert!(verification.echo_seen); + assert_eq!( + verification.accepted_activity(), + Some("Working+esc to interrupt".to_string()), + "a turn marker in the same read as a wrapped echo must be retained" + ); + } + + #[test] + fn compact_match_end_maps_back_to_raw_offsets() { + let text = "a b\r\n cé d tail"; + let end = end_of_compact_match(text, "abcéd").unwrap(); + assert_eq!(&text[end..], " tail"); + assert_eq!(end_of_compact_match(text, "zzz"), None); + } + #[test] fn split_echo_preserves_same_chunk_acceptance_activity() { let expected = "Relay message from Lead [evt]: split echo"; @@ -1020,6 +1483,36 @@ mod tests { assert!(verification.acceptance_expired()); } + #[cfg(unix)] + #[tokio::test] + async fn attachment_delivery_to_no_marker_harness_is_confirmed_by_guarded_rule() { + // #1945 attachments ride the same PTY injection with an `Attachments:` + // block appended; the long path wraps on screen. A harness without + // turn markers must still confirm once the body has left the cursor + // and the harness produced post-echo output. + let body = crate::attachments::append_attachment_block( + "see this", + Some("Attachments:\n- shot.png (image/png, 153.1 KB) saved to /w/.agent-relay/attachments/f1/a-very-long-directory-name/shot.png"), + ); + let expected = + crate::broker::injection_format::format_injection("Alice", "evt_1", &body, "Lead"); + let (pty, snapshot) = + codex_snapshot("› see this\n Attachments:\n - shot.png\nProcessing accepted turn") + .await; + let mut verification = codex_verification(&expected); + verification.detector = ActivityDetector::for_cli("muse"); + verification + .activity_buffer + .push_str("Processing accepted turn"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + HarnessAcceptance::Accepted("activity:any_output".to_string()), + "attachment deliveries need no echo_left_composer shortcut" + ); + pty.shutdown().unwrap(); + } + #[test] fn check_echo_matches_injection_with_attachment_block() { // The attachment block is part of the body, so the expected echo is diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 717879e52..21f2f9ce7 100644 --- a/crates/broker/src/pty_worker.rs +++ b/crates/broker/src/pty_worker.rs @@ -29,9 +29,10 @@ use crate::broker::{ continuity::parse_continuity_command, delivery_verification::{ assess_harness_acceptance, current_timestamp_ms, delivery_injected_event_payload, - delivery_queued_event_payload, DeliveryOutcome, HarnessAcceptance, PendingActivity, - PendingVerification, ThrottleState, VerificationOutput, ACTIVITY_BUFFER_KEEP_BYTES, - ACTIVITY_BUFFER_MAX_BYTES, ACTIVITY_WINDOW, MAX_VERIFICATION_ATTEMPTS, VERIFICATION_WINDOW, + 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, + MAX_VERIFICATION_ATTEMPTS, VERIFICATION_WINDOW, }, injection_format::{format_injection_for_worker_with_workspace, McpReminderThrottle}, }; @@ -49,9 +50,8 @@ use crate::util::terminal::{ use crate::util::utf8_stream::Utf8StreamDecoder; use crate::worker::detection::{is_codex_busy_status_line, ActivityDetector}; use crate::wrap::{ - injection_recovery_followup_delay, submit_injection_body, submit_injection_recovery, - submit_injection_recovery_followup, warn_on_auto_response_write, PtyAutoState, - AUTO_SUGGESTION_BLOCK_TIMEOUT, + cancel_recovery_writes, new_recovery_cancel, submit_injection_body, submit_injection_recovery, + warn_on_auto_response_write, PtyAutoState, AUTO_SUGGESTION_BLOCK_TIMEOUT, }; use base64::Engine; @@ -103,6 +103,101 @@ impl CompletedWorkerDeliveries { } } +/// A delivery whose body was written to the composer but never proven +/// accepted. Its text may still be in the editor, so no further delivery is +/// injected until the composer is observed without it; otherwise the next body +/// would append to the failed draft or submit both as one turn. +#[derive(Debug, Clone)] +struct FailedComposerLatch { + delivery_id: DeliveryId, + event_id: EventId, + expected_echo: String, + latched_at: Instant, + /// The terminal stuck signal has been sent for this latch. + escalated: bool, +} + +/// How long a visibly stuck draft may hold injection before the worker sends +/// the terminal `failed_draft_latched` signal to the sender and operator. The +/// draft is still never typed over; queued deliveries keep waiting. +const FAILED_DRAFT_ESCALATION_TIMEOUT: Duration = Duration::from_secs(60); + +/// [`FAILED_DRAFT_ESCALATION_TIMEOUT`], overridable in milliseconds through +/// `RELAY_FAILED_DRAFT_ESCALATION_MS` so integration tests need not wait. +fn failed_draft_escalation_timeout() -> Duration { + std::env::var("RELAY_FAILED_DRAFT_ESCALATION_MS") + .ok() + .and_then(|value| value.trim().parse::().ok()) + .map(Duration::from_millis) + .unwrap_or(FAILED_DRAFT_ESCALATION_TIMEOUT) +} + +/// Whether a still-latched draft has outlived the escalation timeout and has +/// not been reported yet. +fn failed_draft_escalation_due(latch: &FailedComposerLatch, timeout: Duration) -> bool { + !latch.escalated && latch.latched_at.elapsed() >= timeout +} + +/// Record a terminal failure for a delivery whose body was written: fence its +/// id against replay for the PTY's lifetime and, only on positive evidence +/// that the body is still in the composer (`draft_visible`), latch injection +/// until the composer releases the draft. A failure without that evidence +/// never blocks later deliveries. +fn latch_failed_written_delivery( + failed_written_deliveries: &mut HashMap, + failed_composer_latch: &mut Option, + delivery_id: &DeliveryId, + event_id: &EventId, + expected_echo: &str, + draft_visible: bool, +) { + failed_written_deliveries.insert(delivery_id.clone(), event_id.clone()); + if !draft_visible { + return; + } + *failed_composer_latch = Some(FailedComposerLatch { + delivery_id: delivery_id.clone(), + event_id: event_id.clone(), + expected_echo: expected_echo.to_string(), + latched_at: Instant::now(), + escalated: false, + }); +} + +/// Whether the composer is owned by an unresolved delivery: one awaiting +/// acceptance, one whose submit-key recovery is in flight, or a failed body that +/// may still be typed there. The next injection must wait in every case. +fn composer_owned_by_delivery( + pending_verifications_empty: bool, + pending_recovery_writes_empty: bool, + failed_composer_latch: Option<&FailedComposerLatch>, +) -> bool { + !pending_verifications_empty + || !pending_recovery_writes_empty + || failed_composer_latch.is_some() +} + +/// Whether the failed-draft latch alone holds a newly queued delivery. Another +/// blocker (a verification, a recovery write, an active injection) already +/// explains the wait; reporting the latch then would be redundant. +fn failed_draft_latch_is_sole_blocker( + failed_composer_latch: Option<&FailedComposerLatch>, + pending_verifications_empty: bool, + pending_recovery_writes_empty: bool, + active_injection_present: bool, +) -> bool { + failed_composer_latch.is_some() + && pending_verifications_empty + && pending_recovery_writes_empty + && !active_injection_present +} + +/// Only a non-empty `write_pty` that the drainer admitted can change the +/// composer, so only that write takes ownership away from relay deliveries. +fn human_write_takes_ownership(byte_len: usize, admitted: bool) -> bool { + byte_len > 0 && admitted +} + /// Default per-atom gap for escape-aware paced injection, in milliseconds. /// Overridable at runtime via `RELAY_INJECT_RATE_MS`; `0` disables pacing and /// restores the single bulk write. Kept small because a full injection @@ -211,20 +306,10 @@ type RecoveryWriteAck = ( PendingVerification, u64, u64, - RecoveryWriteStage, Result, tokio::sync::oneshot::error::RecvError>, ); type RecoveryWriteAckFuture = Pin + Send>>; -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum RecoveryWriteStage { - Primary, - Followup, -} - -type RecoveryFollowup = (PendingVerification, u64); -type RecoveryFollowupFuture = Pin + Send>>; - fn queue_post_acceptance_activity( verification: &PendingVerification, detector: Option<&ActivityDetector>, @@ -246,66 +331,6 @@ fn queue_post_acceptance_activity( }); } -#[allow(clippy::too_many_arguments)] -async fn confirm_harness_acceptance( - out_tx: &mpsc::Sender>, - pv: PendingVerification, - evidence: String, - detector: Option<&ActivityDetector>, - pending_activities: &mut VecDeque, - throttle: &mut ThrottleState, - pending_delivery_ids: &mut HashSet, - completed_deliveries: &mut CompletedWorkerDeliveries, -) { - let delivery_id = pv.delivery_id.clone(); - let event_id = pv.event_id.clone(); - tracing::debug!( - delivery_id = %delivery_id, - attempts = pv.attempts, - evidence = %evidence, - "delivery accepted by harness" - ); - let activity_pattern = evidence.strip_prefix("activity:").map(str::to_string); - let _ = send_frame( - out_tx, - "delivery_ack", - pv.request_id.clone(), - json!({ "delivery_id": delivery_id, "event_id": event_id }), - ) - .await; - let _ = send_frame( - out_tx, - "delivery_verified", - None, - json!({ - "delivery_id": delivery_id, - "event_id": event_id, - "verification": crate::broker::delivery_verification::HARNESS_ACCEPTANCE, - "evidence": evidence, - "attempts": pv.attempts, - }), - ) - .await; - if let Some(pattern) = activity_pattern { - let _ = send_frame( - out_tx, - "delivery_active", - None, - json!({ - "delivery_id": delivery_id, - "event_id": event_id, - "pattern": pattern, - }), - ) - .await; - } else { - queue_post_acceptance_activity(&pv, detector, pending_activities); - } - throttle.record(DeliveryOutcome::Success); - pending_delivery_ids.remove(&delivery_id); - completed_deliveries.insert(delivery_id, event_id); -} - fn cli_basename(command: &str) -> &str { command .rsplit(['/', '\\']) @@ -871,13 +896,6 @@ fn injected_output_command_detection_allowed( !active_injection_present && pending_verifications_empty && pending_recovery_writes_empty } -/// A recovery key already admitted to the PTY FIFO cannot be cancelled. Hold -/// later human input at the boundary so ownership is never reported as having -/// preempted a key that may already have reached the child. -fn human_input_admissible(pending_recovery_writes_empty: bool) -> bool { - pending_recovery_writes_empty -} - /// What to do with an in-flight injection once its combined Body+Enter write /// ack resolves. Keeps the ack-resolution policy pure and unit-testable, /// separate from the frame-sending / verification side effects in the select @@ -1175,6 +1193,12 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut pending_worker_injections: VecDeque = VecDeque::new(); let mut pending_worker_delivery_ids: HashSet = HashSet::new(); let mut completed_worker_deliveries = CompletedWorkerDeliveries::default(); + // Deliveries written to this PTY that then failed terminally. A replay of + // one of these ids is refused rather than pasted over the failed draft. + // Unbounded on purpose: entries exist only for failures, and an evicted id + // would let a late retry paste the same body again. + let mut failed_written_deliveries: HashMap = HashMap::new(); + let mut failed_composer_latch: Option = None; // The injection currently being written across paced stages, if any. Only // one injection is in flight at a time; the pending-injection interval arm // starts the next one once this returns to `None`. @@ -1220,11 +1244,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // the drainer proves the recovery key reached the child. let mut pending_recovery_writes: FuturesUnordered = FuturesUnordered::new(); - // A delayed recovery follow-up is kept outside the PTY write queue. This - // lets admitted human input advance the generation before the CR is sent. - let mut pending_recovery_followups: FuturesUnordered = - FuturesUnordered::new(); let mut human_input_generation = 0u64; + // Cancellation flags for queued submit-key recoveries. Human input sets + // them so a delayed recovery key cannot submit the human's draft. + let mut recovery_cancels: Vec> = Vec::new(); + let failed_draft_escalation_timeout = failed_draft_escalation_timeout(); let mut startup_output = String::new(); let mut startup_total_bytes = 0usize; let mut init_request_id: Option = None; @@ -1485,6 +1509,18 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { "retryable": false, })).await; } + } else if failed_written_deliveries.contains_key(&delivery.delivery_id) { + tracing::warn!( + delivery_id = %delivery.delivery_id, + event_id = %delivery.event_id, + "refusing to re-inject a delivery that already failed in this composer" + ); + let _ = send_frame(&out_tx, "delivery_failed", None, json!({ + "delivery_id": delivery.delivery_id, + "event_id": delivery.event_id, + "reason": "delivery was already written to this PTY and failed; refusing to inject it again", + "attempts": 0, + })).await; } else if pending_worker_delivery_ids.insert(delivery.delivery_id.clone()) { let _ = send_frame( &out_tx, @@ -1503,6 +1539,21 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { request_id: frame.request_id, queued_at: Instant::now(), }); + if failed_draft_latch_is_sole_blocker( + failed_composer_latch.as_ref(), + pending_verifications.is_empty(), + pending_recovery_writes.is_empty(), + active_injection.is_some(), + ) { + // The broker marks a worker Working on + // delivery_queued, but this delivery + // will wait behind the failed draft. + let _ = send_frame(&out_tx, "agent_blocked_on_send", None, json!({ + "reason": "failed_draft_latched", + "blocked_secs": 0, + "pending_delivery_count": pending_worker_injections.len(), + })).await; + } } else { tracing::debug!( delivery_id = %delivery.delivery_id, @@ -1550,63 +1601,57 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // write those bytes verbatim to the PTY so the // child process sees the keystrokes. match frame.payload.get("data").and_then(Value::as_str) { - Some("") => { - // An empty input neither changes the - // composer nor transfers ownership away - // from automatic delivery recovery. - let _ = send_frame(&out_tx, "write_pty_response", frame.request_id, json!({ - "bytes_written": 0, - })).await; - } Some(data) => { - // Once an automated recovery key is in - // the PTY drainer, its ordering is - // fixed. Do not admit a human write - // behind it and then retroactively - // label that recovery a cancellation: - // the key may already have reached the - // child. A retryable refusal lets the - // caller resend after the short write - // acknowledgement settles. - if !human_input_admissible(pending_recovery_writes.is_empty()) { - let _ = send_frame(&out_tx, "write_pty_response", frame.request_id, json!({ - "error": { - "code": "pty_write_queue_full", - "message": "automatic submit recovery is already in flight; retry the input", - } - })).await; - continue; - } // Non-blocking submission: never parks the // select loop even if the drainer is wedged // behind a child that stopped reading stdin. let bytes = data.as_bytes().to_vec(); let byte_len = bytes.len(); - match pty.submit_write(bytes) { + let admitted = pty.submit_write(bytes); + // An empty frame changes nothing in the + // composer, and a rejected write never + // reached it: neither takes ownership. + if human_write_takes_ownership(byte_len, admitted.is_ok()) { + // Withdraw queued recovery keys first so + // a delayed submit cannot send the + // human's draft. + cancel_recovery_writes(&mut recovery_cancels); + human_input_generation = human_input_generation.saturating_add(1); + if let Some(cancel) = &initial_injection_cancel { cancel.store(true, Ordering::Relaxed); } + // Once a human writes into the PTY we can + // no longer distinguish the broker body + // from their partial composer text. Give + // the human ownership and terminate every + // pending automatic verification without + // pressing a recovery key. Only an admitted + // write transfers ownership: a rejected one + // never reached the composer. + let mut cancelled_verification = false; + while let Some(pv) = pending_verifications.pop_front() { + cancelled_verification = true; + let delivery_id = pv.delivery_id.clone(); + let _ = send_frame(&out_tx, "delivery_failed", None, json!({ + "delivery_id": delivery_id, + "event_id": pv.event_id, + "reason": "human PTY input took ownership before harness acceptance", + "attempts": pv.attempts, + })).await; + pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &pv.event_id, + &pv.expected_echo, + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo), + ); + } + if cancelled_verification { + throttle.record(DeliveryOutcome::Failed); + } + } + match admitted { Ok(ack_rx) => { - // Only an admitted, non-empty - // human write takes ownership. - // A full/closed queue has not - // changed the composer and must - // leave Relay's pending delivery - // and recovery state intact. - human_input_generation = human_input_generation.saturating_add(1); - if let Some(cancel) = &initial_injection_cancel { cancel.store(true, Ordering::Relaxed); } - let mut cancelled_verification = false; - while let Some(pv) = pending_verifications.pop_front() { - cancelled_verification = true; - let delivery_id = pv.delivery_id.clone(); - let _ = send_frame(&out_tx, "delivery_failed", None, json!({ - "delivery_id": delivery_id, - "event_id": pv.event_id, - "reason": "human PTY input took ownership before harness acceptance", - "attempts": pv.attempts, - })).await; - pending_worker_delivery_ids.remove(&delivery_id); - } - if cancelled_verification { - throttle.record(DeliveryOutcome::Failed); - } // Defer the ack until the drainer confirms the // write landed (or failed). The broker holds the // SendInput reply / pty_input_ack until the @@ -1729,6 +1774,19 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { .and_then(Value::as_str) .filter(|event_id| !event_id.is_empty()) .map(str::to_string); + // An operator-wide flush is the explicit override + // for a failed draft the worker cannot prove has + // left the composer. + if targeted_event_id.is_none() { + if let Some(latch) = failed_composer_latch.take() { + tracing::warn!( + target: "agent_relay::worker::pty", + worker = %worker_name, + delivery_id = %latch.delivery_id, + "operator flush released the failed-draft injection latch" + ); + } + } if pty_auto.interactive_hold { if let Some(event_id) = targeted_event_id { if let Some(inj) = active_injection.as_mut() { @@ -1926,8 +1984,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { if injected_output_command_detection_allowed( active_injection.is_some(), pending_verifications.is_empty(), - pending_recovery_writes.is_empty() - && pending_recovery_followups.is_empty(), + pending_recovery_writes.is_empty(), ) && clean_text.lines().any(|line| line.trim() == "/exit") { @@ -1981,8 +2038,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { if injected_output_command_detection_allowed( active_injection.is_some(), pending_verifications.is_empty(), - pending_recovery_writes.is_empty() - && pending_recovery_followups.is_empty(), + pending_recovery_writes.is_empty(), ) { continuity_buffer.push_str(&clean_text); if continuity_buffer.len() > CONTINUITY_BUFFER_MAX { @@ -2045,17 +2101,62 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // Remove verified entries in reverse order to preserve indices. for (i, evidence) in verified_indices.into_iter().rev() { let pv = pending_verifications.remove(i).unwrap(); - confirm_harness_acceptance( + let delivery_id = pv.delivery_id.clone(); + let event_id = pv.event_id.clone(); + tracing::debug!( + delivery_id = %delivery_id, + attempts = pv.attempts, + evidence = %evidence, + "delivery accepted by harness" + ); + let activity_pattern = evidence + .strip_prefix("activity:") + .map(str::to_string); + let _ = send_frame( &out_tx, - pv, - evidence, - activity_detector.as_ref(), - &mut pending_activities, - &mut throttle, - &mut pending_worker_delivery_ids, - &mut completed_worker_deliveries, + "delivery_ack", + pv.request_id.clone(), + json!({ + "delivery_id": delivery_id, + "event_id": event_id + }), ) .await; + let _ = send_frame( + &out_tx, + "delivery_verified", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "verification": "harness_acceptance", + "evidence": evidence, + "attempts": pv.attempts, + }), + ) + .await; + throttle.record(DeliveryOutcome::Success); + if let Some(pattern) = activity_pattern { + let _ = send_frame( + &out_tx, + "delivery_active", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "pattern": pattern, + }), + ) + .await; + } else { + queue_post_acceptance_activity( + &pv, + activity_detector.as_ref(), + &mut pending_activities, + ); + } + pending_worker_delivery_ids.remove(&delivery_id); + completed_worker_deliveries.insert(delivery_id, event_id); } if activity_detector.as_ref().is_some() { @@ -2200,9 +2301,11 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { if let Some(index) = next_injection_index( &pending_worker_injections, active_injection.is_some(), - !pending_verifications.is_empty() - || !pending_recovery_writes.is_empty() - || !pending_recovery_followups.is_empty(), + composer_owned_by_delivery( + pending_verifications.is_empty(), + pending_recovery_writes.is_empty(), + failed_composer_latch.as_ref(), + ), pty_auto.interactive_hold, hold_exempt_injections, &hold_exempt_event_ids, @@ -2459,6 +2562,16 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { "event_id": inj.pending.delivery.event_id, "reason": "initial_injection_incomplete; body will not be replayed" })).await; + // A partial chunked body may sit in the composer; the + // next delivery must not be pasted onto it. + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &inj.pending.delivery.delivery_id, + &inj.pending.delivery.event_id, + inj.injection_text.as_deref().unwrap_or_default(), + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, inj.injection_text.as_deref().unwrap_or_default()), + ); let _ = send_frame(&out_tx, "worker_error", inj.pending.request_id, json!({ "code": "initial_injection_incomplete", "retryable": false, "message": "Initial task delivery interrupted or timed out; inspect the composer before retrying" @@ -2473,6 +2586,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { "reason": "human PTY input took ownership during injection", })).await; pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &inj.pending.delivery.event_id, + inj.injection_text.as_deref().unwrap_or_default(), + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, inj.injection_text.as_deref().unwrap_or_default()), + ); throttle.record(DeliveryOutcome::Failed); continue; } @@ -2531,17 +2652,62 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { ); if let HarnessAcceptance::Accepted(evidence) = acceptance { let pv = verification; - confirm_harness_acceptance( + let delivery_id = pv.delivery_id.clone(); + let event_id = pv.event_id.clone(); + tracing::debug!( + delivery_id = %delivery_id, + attempts = pv.attempts, + evidence = %evidence, + "delivery accepted before write ack was processed" + ); + let activity_pattern = evidence + .strip_prefix("activity:") + .map(str::to_string); + let _ = send_frame( &out_tx, - pv, - evidence, - activity_detector.as_ref(), - &mut pending_activities, - &mut throttle, - &mut pending_worker_delivery_ids, - &mut completed_worker_deliveries, + "delivery_ack", + pv.request_id.clone(), + json!({ + "delivery_id": delivery_id, + "event_id": event_id + }), + ) + .await; + let _ = send_frame( + &out_tx, + "delivery_verified", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "verification": "harness_acceptance", + "evidence": evidence, + "attempts": pv.attempts, + }), ) .await; + throttle.record(DeliveryOutcome::Success); + if let Some(pattern) = activity_pattern { + let _ = send_frame( + &out_tx, + "delivery_active", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "pattern": pattern, + }), + ) + .await; + } else { + queue_post_acceptance_activity( + &pv, + activity_detector.as_ref(), + &mut pending_activities, + ); + } + pending_worker_delivery_ids.remove(&delivery_id); + completed_worker_deliveries.insert(delivery_id, event_id); } else { pending_verifications.push_back(verification); } @@ -2576,40 +2742,18 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { } // Settle submit-key-only recovery writes without blocking PTY - // output. Human writes receive a retryable refusal until this key - // settles, because an admitted key cannot be retroactively - // cancelled. A failed recovery is terminal because the body may - // already be in the composer and must never be replayed. - Some((mut pv, output_boundary, submitted_generation, stage, ack)) = pending_recovery_writes.next(), + // output or human input. A failed recovery is terminal because the + // body may already be in the composer and must never be replayed. + Some((mut pv, output_boundary, submitted_generation, ack)) = pending_recovery_writes.next(), if !pending_recovery_writes.is_empty() => { let delivery_id = pv.delivery_id.clone(); let event_id = pv.event_id.clone(); - if submitted_generation != human_input_generation { - let _ = send_frame(&out_tx, "delivery_failed", None, json!({ - "delivery_id": delivery_id, - "event_id": event_id, - "reason": "human PTY input took ownership during submit recovery", - "attempts": pv.attempts, - })).await; - throttle.record(DeliveryOutcome::Failed); - pending_worker_delivery_ids.remove(&delivery_id); - continue; - } + // Human input cancels any part of the recovery that had not yet + // reached the child. A key that did reach it may still have + // started the turn, so assess acceptance before failing. + let human_took_over = submitted_generation != human_input_generation; match ack { Ok(Ok(())) => { - if stage == RecoveryWriteStage::Primary { - let completed_attempts = pv.attempts.saturating_sub(1); - if let Some(delay) = injection_recovery_followup_delay( - &resolved_cli, - completed_attempts, - ) { - pending_recovery_followups.push(Box::pin(async move { - tokio::time::sleep(delay).await; - (pv, submitted_generation) - })); - continue; - } - } pv.output_boundary = output_boundary; pv.injected_at = Instant::now(); pv.activity_buffer.clear(); @@ -2624,25 +2768,85 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { &snapshot, ) { HarnessAcceptance::Accepted(evidence) => { - confirm_harness_acceptance( + let activity_pattern = evidence + .strip_prefix("activity:") + .map(str::to_string); + let _ = send_frame( &out_tx, - pv, - evidence, - activity_detector.as_ref(), - &mut pending_activities, - &mut throttle, - &mut pending_worker_delivery_ids, - &mut completed_worker_deliveries, + "delivery_ack", + pv.request_id.clone(), + json!({ "delivery_id": delivery_id, "event_id": event_id }), ) .await; + let _ = send_frame( + &out_tx, + "delivery_verified", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "verification": "harness_acceptance", + "evidence": evidence, + "attempts": pv.attempts, + }), + ) + .await; + if let Some(pattern) = activity_pattern { + let _ = send_frame( + &out_tx, + "delivery_active", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "pattern": pattern, + }), + ) + .await; + } else { + queue_post_acceptance_activity( + &pv, + activity_detector.as_ref(), + &mut pending_activities, + ); + } + throttle.record(DeliveryOutcome::Success); + pending_worker_delivery_ids.remove(&delivery_id); + completed_worker_deliveries.insert(delivery_id, event_id); } - HarnessAcceptance::Parked | HarnessAcceptance::Inconclusive => { + HarnessAcceptance::Parked | HarnessAcceptance::Inconclusive + if !human_took_over => + { pending_verifications.push_back(pv); } + HarnessAcceptance::Parked | HarnessAcceptance::Inconclusive => { + let _ = send_frame(&out_tx, "delivery_failed", None, json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "reason": "human PTY input took ownership during submit recovery", + "attempts": pv.attempts, + })).await; + throttle.record(DeliveryOutcome::Failed); + pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo), + ); + } } } Ok(Err(error)) => { - let reason = format!("submit recovery write failed: {error}"); + let reason = if human_took_over { + format!( + "human PTY input took ownership during submit recovery; queued submit key withheld: {error}" + ) + } else { + format!("submit recovery write failed: {error}") + }; let _ = send_frame(&out_tx, "delivery_failed", None, json!({ "delivery_id": delivery_id, "event_id": event_id, @@ -2651,6 +2855,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { })).await; throttle.record(DeliveryOutcome::Failed); pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo), + ); } Err(_) => { let reason = "submit recovery drainer exited before acknowledging the key"; @@ -2662,50 +2874,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { })).await; throttle.record(DeliveryOutcome::Failed); pending_worker_delivery_ids.remove(&delivery_id); - } - } - } - - // Codex recovery positions the cursor first, then waits before CR. - // Re-check ownership after that wait so an admitted human write - // cancels the submit key instead of splicing it into human text. - Some((pv, submitted_generation)) = pending_recovery_followups.next(), - if !pending_recovery_followups.is_empty() => { - let delivery_id = pv.delivery_id.clone(); - let event_id = pv.event_id.clone(); - if submitted_generation != human_input_generation { - let _ = send_frame(&out_tx, "delivery_failed", None, json!({ - "delivery_id": delivery_id, - "event_id": event_id, - "reason": "human PTY input took ownership before delayed submit recovery", - "attempts": pv.attempts, - })).await; - throttle.record(DeliveryOutcome::Failed); - pending_worker_delivery_ids.remove(&delivery_id); - continue; - } - match submit_injection_recovery_followup(&pty) { - Ok((ack_rx, output_boundary)) => { - pending_recovery_writes.push(Box::pin(async move { - ( - pv, - output_boundary, - submitted_generation, - RecoveryWriteStage::Followup, - ack_rx.await, - ) - })); - } - Err(error) => { - let reason = format!("failed to queue delayed submit recovery: {error}"); - let _ = send_frame(&out_tx, "delivery_failed", None, json!({ - "delivery_id": delivery_id, - "event_id": event_id, - "reason": reason, - "attempts": pv.attempts, - })).await; - throttle.record(DeliveryOutcome::Failed); - pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo), + ); } } } @@ -2745,17 +2921,51 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { &snapshot, ) { HarnessAcceptance::Accepted(evidence) => { - confirm_harness_acceptance( + let activity_pattern = evidence + .strip_prefix("activity:") + .map(str::to_string); + let _ = send_frame( &out_tx, - pv, - evidence, - activity_detector.as_ref(), - &mut pending_activities, - &mut throttle, - &mut pending_worker_delivery_ids, - &mut completed_worker_deliveries, + "delivery_ack", + pv.request_id.clone(), + json!({ "delivery_id": delivery_id, "event_id": event_id }), + ) + .await; + let _ = send_frame( + &out_tx, + "delivery_verified", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "verification": "harness_acceptance", + "evidence": evidence, + "attempts": pv.attempts, + }), ) .await; + if let Some(pattern) = activity_pattern { + let _ = send_frame( + &out_tx, + "delivery_active", + None, + json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "pattern": pattern, + }), + ) + .await; + } else { + queue_post_acceptance_activity( + &pv, + activity_detector.as_ref(), + &mut pending_activities, + ); + } + throttle.record(DeliveryOutcome::Success); + pending_worker_delivery_ids.remove(&delivery_id); + completed_worker_deliveries.insert(delivery_id, event_id); } HarnessAcceptance::Parked if pv.attempts < pv.max_attempts @@ -2763,8 +2973,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { && !pty_auto.interactive_hold && pending_pty_writes.is_empty() && active_injection.is_none() - && pending_recovery_writes.is_empty() - && pending_recovery_followups.is_empty() => + && pending_recovery_writes.is_empty() => { let _ = send_frame( &out_tx, @@ -2779,7 +2988,12 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { }), ) .await; - match submit_injection_recovery(&pty, &resolved_cli, pv.attempts) { + match submit_injection_recovery( + &pty, + &resolved_cli, + pv.attempts, + new_recovery_cancel(&mut recovery_cancels), + ) { Ok((ack_rx, output_boundary)) => { pv.attempts += 1; let _ = send_frame( @@ -2795,13 +3009,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { ) .await; pending_recovery_writes.push(Box::pin(async move { - ( - pv, - output_boundary, - human_input_generation, - RecoveryWriteStage::Primary, - ack_rx.await, - ) + (pv, output_boundary, human_input_generation, ack_rx.await) })); } Err(error) => { @@ -2814,6 +3022,14 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { })).await; throttle.record(DeliveryOutcome::Failed); pending_worker_delivery_ids.remove(&delivery_id); + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo), + ); } } } @@ -2822,8 +3038,7 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { && (pty_auto.interactive_hold || !pending_pty_writes.is_empty() || active_injection.is_some() - || !pending_recovery_writes.is_empty() - || !pending_recovery_followups.is_empty()) => + || !pending_recovery_writes.is_empty()) => { // A human or another writer owns the input FIFO. // Keep the delivery pending and retry the check @@ -2837,24 +3052,53 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { pv.injected_at = Instant::now(); pending_verifications.push_back(pv); } - state @ (HarnessAcceptance::Parked - | HarnessAcceptance::Inconclusive) => { - let reason = match state { - HarnessAcceptance::Parked => { - "body remained parked after bounded submit-key recovery" - } - HarnessAcceptance::Inconclusive => { - crate::broker::delivery_verification::HARNESS_ACCEPTANCE_UNPROVEN - } - HarnessAcceptance::Accepted(_) => unreachable!(), - }; + HarnessAcceptance::Inconclusive => { + // Unconfirmed: the body was written and is not + // visibly parked, but nothing proved the + // harness took the turn. That is a delivery, + // not a failure: release custody, report it + // unconfirmed, and keep delivering. Never latch. + let _ = send_frame( + &out_tx, + "delivery_ack", + pv.request_id.clone(), + json!({ "delivery_id": delivery_id, "event_id": event_id }), + ) + .await; + let _ = send_frame(&out_tx, "delivery_unconfirmed", None, json!({ + "delivery_id": delivery_id, + "event_id": event_id, + "outcome": "unconfirmed", + "terminal": true, + "reason": crate::broker::delivery_verification::HARNESS_ACCEPTANCE_UNPROVEN, + "attempts": pv.attempts, + })).await; + throttle.record(DeliveryOutcome::Unverified); + pending_worker_delivery_ids.remove(&delivery_id); + completed_worker_deliveries.insert(delivery_id, event_id); + } + HarnessAcceptance::Parked => { + // StuckInComposer: the body is still visibly + // parked after bounded submit-key recovery. let _ = send_frame(&out_tx, "delivery_failed", None, json!({ "delivery_id": delivery_id, "event_id": event_id, - "reason": reason, + "reason": "body remained parked after bounded submit-key recovery", "attempts": pv.attempts, })).await; throttle.record(DeliveryOutcome::Failed); + // Latch before removing the pending id: the body + // is still in the composer, so neither a replay + // nor the next queued delivery may be typed on + // top of it. Parked is the positive evidence. + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + true, + ); pending_worker_delivery_ids.remove(&delivery_id); } } @@ -2863,6 +3107,49 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { } } + // Release the failed-draft latch once the composer no longer + // holds that body (the harness took it late, or a human + // submitted or cleared it). Until then queued deliveries wait. + if let Some(latch) = failed_composer_latch.as_ref() { + if pending_verifications.is_empty() + && pending_recovery_writes.is_empty() + && active_injection.is_none() + && failed_draft_released( + &resolved_cli, + &latch.expected_echo, + &Snapshot::capture(&pty), + ) + { + tracing::info!( + delivery_id = %latch.delivery_id, + event_id = %latch.event_id, + "failed delivery left the composer; resuming injection" + ); + failed_composer_latch = None; + } + } + // A draft still stuck after the bounded timeout gets one + // terminal signal to the sender and operator. It is still + // never typed over; queued deliveries keep waiting. + if let Some(latch) = failed_composer_latch.as_mut() { + if failed_draft_escalation_due(latch, failed_draft_escalation_timeout) { + latch.escalated = true; + tracing::warn!( + delivery_id = %latch.delivery_id, + event_id = %latch.event_id, + "failed draft still stuck in the composer; escalating" + ); + let _ = send_frame(&out_tx, "agent_blocked_on_send", None, json!({ + "reason": "failed_draft_latched", + "terminal": true, + "delivery_id": latch.delivery_id, + "event_id": latch.event_id, + "blocked_secs": latch.latched_at.elapsed().as_secs(), + "pending_delivery_count": pending_worker_injections.len(), + })).await; + } + } + if activity_detector.is_some() { let mut i = 0; while i < pending_activities.len() { @@ -2888,7 +3175,6 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { if active_injection.is_none() && pending_verifications.is_empty() && pending_recovery_writes.is_empty() - && pending_recovery_followups.is_empty() { pty_auto.try_auto_enter(&pty); } @@ -2899,7 +3185,6 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { && pending_verifications.is_empty() && pending_activities.is_empty() && pending_recovery_writes.is_empty() - && pending_recovery_followups.is_empty() && active_injection.is_none() { if let Some(threshold) = idle_threshold { @@ -3580,28 +3865,18 @@ mod tests { if attempt_started.elapsed() >= VERIFICATION_WINDOW && verification.attempts < verification.max_attempts => { - let completed_attempts = verification.attempts; - let (recovery_ack, mut boundary) = - submit_injection_recovery(pty, "codex", verification.attempts) - .expect("queue submit-key recovery"); + let (recovery_ack, boundary) = submit_injection_recovery( + pty, + "codex", + verification.attempts, + Default::default(), + ) + .expect("queue submit-key recovery"); tokio::time::timeout(Duration::from_secs(5), recovery_ack) .await .expect("submit-key recovery ack timed out") .expect("PTY drainer exited during recovery") .expect("submit-key recovery failed"); - if let Some(delay) = - injection_recovery_followup_delay("codex", completed_attempts) - { - tokio::time::sleep(delay).await; - let (followup_ack, followup_boundary) = - submit_injection_recovery_followup(pty) - .expect("queue delayed submit recovery"); - followup_ack - .await - .expect("PTY drainer exited during delayed recovery") - .expect("delayed submit recovery failed"); - boundary = followup_boundary; - } verification.attempts += 1; verification.output_boundary = boundary; verification.activity_buffer.clear(); @@ -3766,16 +4041,10 @@ mod tests { let mut accepted_after = None; for completed_attempts in [1, 2] { - let (recovery_ack, mut boundary) = - submit_injection_recovery(&pty, "codex", completed_attempts).unwrap(); + let (recovery_ack, boundary) = + submit_injection_recovery(&pty, "codex", completed_attempts, Default::default()) + .unwrap(); recovery_ack.await.unwrap().unwrap(); - if let Some(delay) = injection_recovery_followup_delay("codex", completed_attempts) { - tokio::time::sleep(delay).await; - let (followup_ack, followup_boundary) = - submit_injection_recovery_followup(&pty).unwrap(); - followup_ack.await.unwrap().unwrap(); - boundary = followup_boundary; - } verification.attempts += 1; verification.output_boundary = boundary; verification.activity_buffer.clear(); @@ -4457,12 +4726,6 @@ mod tests { )); } - #[test] - fn human_input_waits_until_an_admitted_recovery_write_settles() { - assert!(!human_input_admissible(false)); - assert!(human_input_admissible(true)); - } - #[test] fn injection_pop_allowed_through_hold_with_flush_exemption() { // An explicit `flush_injections` allowance lets pops pierce the hold… @@ -4541,6 +4804,130 @@ mod tests { ); } + #[test] + fn latch_arms_only_on_visible_draft_and_escalates_once_after_timeout() { + let mut failed = HashMap::new(); + let mut latch = None; + let delivery_id = DeliveryId::from("del_unproven"); + let event_id = EventId::from("evt_unproven"); + latch_failed_written_delivery( + &mut failed, + &mut latch, + &delivery_id, + &event_id, + "Relay message from Lead [evt]: unproven", + false, + ); + assert!(failed.contains_key(&delivery_id), "replay stays fenced"); + assert!(latch.is_none(), "no positive evidence, no latch"); + + latch_failed_written_delivery( + &mut failed, + &mut latch, + &delivery_id, + &event_id, + "Relay message from Lead [evt]: stuck", + true, + ); + let latch = latch.as_mut().expect("visible draft latches"); + assert!(!failed_draft_escalation_due( + latch, + Duration::from_secs(3600) + )); + assert!(failed_draft_escalation_due(latch, Duration::ZERO)); + latch.escalated = true; + assert!( + !failed_draft_escalation_due(latch, Duration::ZERO), + "escalates once" + ); + } + + #[test] + fn failed_written_delivery_latches_injection_and_fences_replay() { + let mut failed = HashMap::new(); + let mut latch = None; + let delivery_id = DeliveryId::from("del_parked_failure"); + let event_id = EventId::from("evt_parked_failure"); + assert!(!composer_owned_by_delivery(true, true, latch.as_ref())); + + latch_failed_written_delivery( + &mut failed, + &mut latch, + &delivery_id, + &event_id, + "Relay message from Lead [evt]: parked", + true, + ); + + assert_eq!(failed.get(&delivery_id), Some(&event_id)); + assert!( + composer_owned_by_delivery(true, true, latch.as_ref()), + "a failed body still in the composer must block the next injection" + ); + let mut pending = VecDeque::new(); + pending.push_back(test_pending_injection("evt_next")); + assert_eq!( + next_injection_index( + &pending, + false, + composer_owned_by_delivery(true, true, latch.as_ref()), + false, + 0, + &HashSet::new(), + ), + None + ); + } + + #[test] + fn latched_worker_reports_blocked_only_when_the_latch_is_the_sole_blocker() { + let latch = FailedComposerLatch { + delivery_id: DeliveryId::from("del_failed"), + event_id: EventId::from("evt_failed"), + expected_echo: "Relay message from Lead [evt]: failed".to_string(), + latched_at: Instant::now(), + escalated: false, + }; + assert!(failed_draft_latch_is_sole_blocker( + Some(&latch), + true, + true, + false + )); + assert!(!failed_draft_latch_is_sole_blocker(None, true, true, false)); + assert!(!failed_draft_latch_is_sole_blocker( + Some(&latch), + false, + true, + false + )); + assert!(!failed_draft_latch_is_sole_blocker( + Some(&latch), + true, + false, + false + )); + assert!(!failed_draft_latch_is_sole_blocker( + Some(&latch), + true, + true, + true + )); + } + + #[test] + fn only_an_admitted_nonempty_human_write_takes_ownership() { + assert!( + !human_write_takes_ownership(0, true), + "empty write_pty frame" + ); + assert!( + !human_write_takes_ownership(5, false), + "write rejected by a full queue" + ); + assert!(human_write_takes_ownership(5, true)); + } + #[test] fn requeued_targeted_exemption_returns_to_its_event_id() { // A failed write requeues the injection. A targeted exemption must go diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index 34a043531..1286e0de3 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -249,31 +249,41 @@ pub(super) fn verified_spawn_ready_result( } } -/// A live, ready agent whose initial task was never confirmed as fully -/// received. +/// A live, ready agent whose initial task was delivered but not confirmed. /// -/// This is deliberately neither a success nor an ordinary failure. The spawn -/// did happen — the agent is registered, ready and holding the node's only -/// slot for that name — so the caller must not retry, or it duplicates the -/// agent. But the task it was spawned to run cannot be shown to have arrived -/// intact, and `spawned:true, ready:true` is exactly the proof `fleet spawn` -/// uses to report success, so the only honest answer is an error that names -/// the live agent and says what to do instead (relay#1893 review, P1). +/// Per the #1959 acceptance policy, only a real delivery failure fails a +/// verified spawn. An unconfirmed task (written, not visibly parked, but no +/// harness evidence either way, or bound to no verdict at all) is a success: +/// the agent is registered, ready and holding the name, so a retry would +/// duplicate it. The result carries `warning.code = "spawn_task_unconfirmed"` +/// so callers can tell it apart from a task the harness was seen to accept. pub(super) fn verified_spawn_task_unconfirmed_result( invocation_id: String, name: &WorkerName, verification: Option<&str>, ) -> ActionResult { - verified_spawn_failed_result( + ActionResult { + task: None, + v: FLEET_WIRE_VERSION, + id: None, invocation_id, - &format!( - "spawn_task_unconfirmed: agent '{name}' is live and ready, but full receipt of its \ - initial task was not confirmed (verification: {}). Do not retry this spawn — it \ - would duplicate the agent. Resend the task, or write the brief to a file and send a \ - short pointer, then check the agent's transcript.", - verification.unwrap_or("none") - ), - ) + result: ActionResultPayload::Output(ActionResultOutput { + output: json!({ + "spawned": true, + "ready": true, + "name": name.as_str(), + "warning": { + "code": "spawn_task_unconfirmed", + "verification": verification.unwrap_or("none"), + "message": format!( + "agent '{name}' is live and ready and its initial task was delivered, but \ + the harness never confirmed accepting it. Do not retry this spawn; check \ + the agent's transcript." + ), + }, + }), + }), + } } /// Result for a verified spawn whose worker had already reported @@ -2248,9 +2258,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 the explicitly supported `cat` process echo exception -/// 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/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 6663d1b56..8f80d356e 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -884,9 +884,39 @@ fn delivery_verified_worker_event( /// looking at *what* confirmed the delivery. A worker that acked through the /// timeout fallback — i.e. an agent that may have swallowed the whole task — /// produced a result identical to one that echoed every byte. +/// #1959 acceptance policy: an unconfirmed initial task is a delivered task, so +/// the verified spawn succeeds and names the live agent with a warning. +fn assert_unconfirmed_spawn_success( + result: &crate::fleet_wire::ActionResultPayload, + worker_name: &str, + verification: &str, +) { + let crate::fleet_wire::ActionResultPayload::Output(output) = result else { + panic!("an unconfirmed task must resolve the spawn as success: {result:?}"); + }; + assert_eq!(output.output["spawned"], true, "{:?}", output.output); + assert_eq!(output.output["ready"], true, "{:?}", output.output); + assert_eq!(output.output["name"], worker_name, "{:?}", output.output); + assert_eq!( + output.output["warning"]["code"], "spawn_task_unconfirmed", + "{:?}", + output.output + ); + assert_eq!( + output.output["warning"]["verification"], verification, + "{:?}", + output.output + ); + let message = output.output["warning"]["message"] + .as_str() + .unwrap_or_default(); + assert!(message.contains(worker_name), "{message}"); + assert!(message.contains("Do not retry"), "{message}"); +} + #[tokio::test] -async fn a_spawn_task_acked_without_proof_of_receipt_never_reports_success() { - use crate::fleet_wire::{ActionResultPayload, BrokerToRelaycast}; +async fn a_spawn_task_acked_without_proof_of_receipt_succeeds_with_an_unconfirmed_warning() { + use crate::fleet_wire::BrokerToRelaycast; // Every label a worker can ack with that is not harness acceptance, // including the echo labels of workers that predate it: a body echoed in @@ -936,22 +966,10 @@ async fn a_spawn_task_acked_without_proof_of_receipt_never_reports_success() { panic!("spawn completion must send an action result"); }; assert_eq!(result.invocation_id, "inv-receipt"); - let ActionResultPayload::Error(error) = result.result else { - panic!("{verification:?} is not proof of receipt and must not succeed"); - }; - assert!( - error.error.starts_with("spawn_task_unconfirmed: "), - "{}", - error.error - ); - // The agent is live: name it so the caller can reach it, and say not to - // retry, because a retry duplicates it. - assert!(error.error.contains(worker_name), "{}", error.error); - assert!(error.error.contains("Do not retry"), "{}", error.error); - assert!( - error.error.contains(verification.unwrap_or("none")), - "{}", - error.error + assert_unconfirmed_spawn_success( + &result.result, + worker_name, + verification.unwrap_or("none"), ); // Resolved, so maintenance's readiness deadline can no longer release // the live worker this spawn produced. @@ -1100,14 +1118,7 @@ async fn a_task_verdict_before_proven_readiness_resolves_on_the_proven_ready() { output.output, json!({"spawned": true, "ready": true, "name": worker_name}) ), - (ActionResultPayload::Error(error), false) => { - assert!( - error.error.starts_with("spawn_task_unconfirmed: "), - "{}", - error.error - ); - assert!(error.error.contains(verification), "{}", error.error); - } + (result, false) => assert_unconfirmed_spawn_success(result, worker_name, verification), (other, _) => panic!("{verification}: unexpected result {other:?}"), } assert!(!fixture.runtime.pending_verified_spawns.contains_key(&name)); @@ -1176,8 +1187,7 @@ async fn a_failed_initial_task_releases_the_worker_before_failing_the_spawn() { /// must resolve as unconfirmed and keep the live worker, never release it as a /// failed task (which could kill the run and invite a duplicating retry). #[tokio::test] -async fn an_unproven_acceptance_reports_the_spawn_unconfirmed_and_keeps_the_worker() { - use crate::fleet_wire::ActionResultPayload; +async fn an_unproven_acceptance_succeeds_with_an_unconfirmed_warning_and_keeps_the_worker() { let worker_name = "spawn-task-unproven"; let registry = make_worker_registry_with_worker(worker_name).await; let generation = registry.workers[worker_name].generation; @@ -1200,15 +1210,11 @@ async fn an_unproven_acceptance_reports_the_spawn_unconfirmed_and_keeps_the_work .await; let results = drain_action_results(&mut fixture.fleet_control_rx); assert_eq!(results.len(), 1, "resolved exactly once"); - let ActionResultPayload::Error(error) = &results[0].result else { - panic!("an unproven task must not succeed: {:?}", results[0]); - }; - assert!( - error.error.starts_with("spawn_task_unconfirmed: "), - "{}", - error.error + assert_unconfirmed_spawn_success( + &results[0].result, + name.as_str(), + "harness_acceptance_unproven", ); - assert!(error.error.contains("Do not retry"), "{}", error.error); assert!(!fixture.runtime.pending_verified_spawns.contains_key(&name)); fixture.runtime.handle_maintenance_tick().await; assert!(fixture.runtime.workers.has_worker(&name), "worker kept"); @@ -1218,18 +1224,13 @@ async fn an_unproven_acceptance_reports_the_spawn_unconfirmed_and_keeps_the_work /// returns has its PTY task queued without a binding to this action, so the /// action cannot learn the task's verdict and must not report plain success. #[test] -fn an_already_ready_worker_with_an_unbound_pty_task_is_not_reported_ready() { +fn an_already_ready_worker_with_an_unbound_pty_task_succeeds_with_an_unconfirmed_warning() { use crate::fleet_wire::ActionResultPayload; let name = WorkerName::from("already-ready"); - let ActionResultPayload::Error(error) = - super::fleet::already_ready_spawn_result("inv".into(), &name, true).result - else { - panic!("an unbound PTY task must not report success"); - }; - assert!( - error.error.starts_with("spawn_task_unconfirmed: "), - "{}", - error.error + assert_unconfirmed_spawn_success( + &super::fleet::already_ready_spawn_result("inv".into(), &name, true).result, + name.as_str(), + "unbound", ); assert!(matches!( super::fleet::already_ready_spawn_result("inv".into(), &name, false).result, @@ -5324,6 +5325,209 @@ async fn stale_recovery_progress_does_not_block_the_worker() { cleanup_worker_registry(fixture.runtime.workers).await; } +fn delivery_verified_event_for( + name: &str, + generation: Uuid, + delivery_id: &str, + event_id: &str, + verification: Option<&str>, +) -> WorkerEvent { + let mut payload = json!({ + "delivery_id": delivery_id, + "event_id": event_id, + }); + if let Some(verification) = verification { + payload["verification"] = json!(verification); + } + WorkerEvent::Message { + name: WorkerName::from(name), + generation, + value: json!({ "type": "delivery_verified", "payload": payload }), + } +} + +#[tokio::test] +async fn terminal_failure_without_remaining_work_does_not_block_the_worker() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let delivery_id = DeliveryId::new("del_terminal_failure_only"); + let pending = make_pending_delivery(delivery_id.as_str(), worker_name); + let event_id = pending.delivery.event_id.clone(); + let mut fixture = + worker_event_runtime_fixture(registry, HashMap::from([(delivery_id.clone(), pending)])); + + fixture + .runtime + .handle_worker_event(delivery_lifecycle_worker_event( + worker_name, + generation, + "delivery_failed", + delivery_id.as_str(), + event_id.as_str(), + )) + .await; + + assert!(!fixture + .runtime + .pending_deliveries + .contains_key(&delivery_id)); + assert_eq!( + fixture.runtime.workers.workers[worker_name].state, + AgentWorkState::Working, + "a dead-lettered delivery with no remaining work must not pin blocked_on_send" + ); + let mut kinds = Vec::new(); + while let Ok(frame) = fixture._sdk_out_rx.try_recv() { + if let Some(kind) = frame.payload.get("kind").and_then(Value::as_str) { + kinds.push(kind.to_string()); + } + } + assert!(kinds.iter().any(|kind| kind == "delivery_failed")); + assert!(!kinds.iter().any(|kind| kind == "agent_blocked_on_send")); + cleanup_worker_registry(fixture.runtime.workers).await; +} + +#[tokio::test] +async fn terminal_failure_with_remaining_work_stays_blocked_on_send() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let failed_id = DeliveryId::new("del_terminal_failure_first"); + let waiting_id = DeliveryId::new("del_terminal_failure_waiting"); + let failed = make_pending_delivery(failed_id.as_str(), worker_name); + let waiting = make_pending_delivery(waiting_id.as_str(), worker_name); + let event_id = failed.delivery.event_id.clone(); + let mut fixture = worker_event_runtime_fixture( + registry, + HashMap::from([(failed_id.clone(), failed), (waiting_id.clone(), waiting)]), + ); + + fixture + .runtime + .handle_worker_event(delivery_lifecycle_worker_event( + worker_name, + generation, + "delivery_failed", + failed_id.as_str(), + event_id.as_str(), + )) + .await; + + assert_eq!( + fixture.runtime.workers.workers[worker_name].state, + AgentWorkState::BlockedOnSend + ); + cleanup_worker_registry(fixture.runtime.workers).await; +} + +#[tokio::test] +async fn pty_delivery_verified_requires_harness_acceptance() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let delivery_id = DeliveryId::new("del_legacy_verification"); + let pending = make_pending_delivery(delivery_id.as_str(), worker_name); + let event_id = pending.delivery.event_id.clone(); + let mut fixture = + worker_event_runtime_fixture(registry, HashMap::from([(delivery_id.clone(), pending)])); + + for legacy in [Some("echo"), Some("timeout_fallback"), None] { + fixture + .runtime + .handle_worker_event(delivery_verified_event_for( + worker_name, + generation, + delivery_id.as_str(), + event_id.as_str(), + legacy, + )) + .await; + assert!( + fixture + .runtime + .pending_deliveries + .contains_key(&delivery_id), + "PTY verification {legacy:?} must not clear delivery custody" + ); + } + while let Ok(frame) = fixture._sdk_out_rx.try_recv() { + assert_ne!( + frame.payload.get("kind").and_then(Value::as_str), + Some("message_delivery_confirmed") + ); + } + + fixture + .runtime + .handle_worker_event(delivery_verified_event_for( + worker_name, + generation, + delivery_id.as_str(), + event_id.as_str(), + Some("harness_acceptance"), + )) + .await; + assert!(!fixture + .runtime + .pending_deliveries + .contains_key(&delivery_id)); + cleanup_worker_registry(fixture.runtime.workers).await; +} + +#[tokio::test] +async fn terminal_unconfirmed_delivery_never_blocks_the_worker() { + let worker_name = "worker-a"; + let registry = make_worker_registry_with_worker(worker_name).await; + let generation = registry.workers[worker_name].generation; + let delivery_id = DeliveryId::new("del_unconfirmed_terminal"); + let pending = make_pending_delivery(delivery_id.as_str(), worker_name); + let event_id = pending.delivery.event_id.clone(); + let mut fixture = + worker_event_runtime_fixture(registry, HashMap::from([(delivery_id.clone(), pending)])); + + fixture + .runtime + .handle_worker_event(WorkerEvent::Message { + name: WorkerName::from(worker_name), + generation, + value: json!({"type": "delivery_unconfirmed", "payload": { + "delivery_id": delivery_id.as_str(), + "event_id": event_id.as_str(), + "outcome": "unconfirmed", + "terminal": true, + "reason": crate::broker::delivery_verification::HARNESS_ACCEPTANCE_UNPROVEN, + "attempts": 1, + }}), + }) + .await; + + assert_ne!( + fixture.runtime.workers.workers[worker_name].state, + AgentWorkState::BlockedOnSend, + "Unconfirmed is a delivery, not a blocked send" + ); + let mut kinds = Vec::new(); + while let Ok(frame) = fixture._sdk_out_rx.try_recv() { + if let Some(kind) = frame.payload.get("kind").and_then(Value::as_str) { + kinds.push(kind.to_string()); + } + } + assert!( + kinds.iter().any(|kind| kind == "delivery_unconfirmed"), + "{kinds:?}" + ); + assert!( + !kinds.iter().any(|kind| kind == "agent_blocked_on_send"), + "{kinds:?}" + ); + assert!( + !kinds.iter().any(|kind| kind == "delivery_failed"), + "{kinds:?}" + ); + cleanup_worker_registry(fixture.runtime.workers).await; +} + #[tokio::test] async fn submit_recovery_progress_defers_broker_delivery_retry() { let worker_name = "worker-a"; diff --git a/crates/broker/src/runtime/worker_events.rs b/crates/broker/src/runtime/worker_events.rs index 8e4f4731d..7ccd52000 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -183,6 +183,61 @@ fn record_started_harness_pid( true } +/// Validate a worker's `delivery_verified` frame before it clears custody. +/// +/// PTY workers must name harness acceptance (or a replay of an already +/// completed delivery). Legacy `echo` and `timeout_fallback` frames prove only +/// that bytes reached the terminal, so they never confirm a delivery. Headless +/// workers report process success without a verification field. +fn accepted_delivery_verification(verification: Option<&str>, is_pty: bool) -> Option<&str> { + match verification { + Some("harness_acceptance" | "completed_replay") => verification, + Some("echo" | "timeout_fallback") => None, + Some(_) | None if is_pty => None, + Some(other) => Some(other), + None => Some("worker_confirmation"), + } +} + +/// Worker state after a terminal `delivery_failed`, as `(state, published +/// state, reason)`. The failed delivery is already dead-lettered, so the +/// worker stays blocked only while another delivery for it awaits acceptance; +/// otherwise later output and idle events must be able to move it on. +fn state_after_terminal_delivery_failure( + pending_delivery_count: usize, +) -> (AgentWorkState, &'static str, &'static str) { + if pending_delivery_count > 0 { + (AgentWorkState::BlockedOnSend, "stuck", "blocked_on_send") + } else { + (AgentWorkState::Working, "working", "delivery_failed") + } +} + +#[cfg(test)] +mod terminal_delivery_failure_state_tests { + use super::*; + + #[test] + fn last_failed_delivery_publishes_a_non_stuck_state() { + assert_eq!( + state_after_terminal_delivery_failure(0), + (AgentWorkState::Working, "working", "delivery_failed"), + "a worker with no remaining deliveries must publish its way out of `stuck`" + ); + assert_eq!( + state_after_terminal_delivery_failure(2), + (AgentWorkState::BlockedOnSend, "stuck", "blocked_on_send") + ); + } +} + +/// Whether a worker's `delivery_unconfirmed` frame is the terminal Unconfirmed +/// outcome (delivered without harness evidence) rather than a mid-recovery +/// progress report for a body still awaiting acceptance. +fn is_terminal_unconfirmed(payload: &Value) -> bool { + payload.get("terminal").and_then(Value::as_bool) == Some(true) +} + fn worker_event_is_current(current_generation: Option, event_generation: Uuid) -> bool { current_generation == Some(event_generation) } @@ -871,6 +926,66 @@ impl BrokerRuntime { .get("event_id") .and_then(Value::as_str) .unwrap_or(""); + // A terminal `delivery_unconfirmed` is the + // Unconfirmed outcome: the body was delivered + // (custody already released by the preceding + // delivery_ack) without harness evidence. It is + // never a blocked worker and never `stuck`. + let terminal_unconfirmed = msg_type == "delivery_unconfirmed" + && is_terminal_unconfirmed(payload); + if terminal_unconfirmed { + if let Some(pending) = pending_verified_spawns + .get_mut(&name) + .filter(|pending| pending.matches_task(generation, event_id)) + { + pending.record_task_verification( + false, + Some("harness_acceptance_unproven"), + ); + } + let resolved = pending_verified_spawns + .get(&name) + .is_some_and(|pending| pending.can_report_ready(generation)) + .then(|| pending_verified_spawns.remove(&name)) + .flatten(); + if let Some(pending) = resolved { + let _ = fleet_control_tx + .send(FleetControlCommand::Send( + crate::fleet_wire::BrokerToRelaycast::ActionResult( + pending.completion(&name), + ), + )) + .await; + } + if let Some(handle) = workers.workers.get_mut(&name) { + handle.last_activity_at = Instant::now(); + if handle.state == AgentWorkState::BlockedOnSend { + handle.state = AgentWorkState::Working; + } + } + tracing::info!( + target = "agent_relay::broker", + worker = %name, + delivery_id = %delivery_id, + event_id = %event_id, + "delivery unconfirmed: delivered without harness evidence" + ); + let _ = send_event( + sdk_out_tx, + json!({ + "kind": "delivery_unconfirmed", + "name": name, + "delivery_id": payload.get("delivery_id"), + "event_id": payload.get("event_id"), + "outcome": "unconfirmed", + "terminal": true, + "reason": payload.get("reason"), + "attempts": payload.get("attempts"), + }), + ) + .await; + return; + } let matching_pending = pending_deliveries .get_mut(delivery_id) .filter(|pending| pending.delivery.event_id.as_str() == event_id) @@ -997,26 +1112,21 @@ impl BrokerRuntime { .await; } - // Modern PTY workers use "harness_acceptance": - // editor echo alone is never delivery confirmation. - let verification = label.unwrap_or("worker_confirmation"); let is_pty = workers .workers .get(&name) .is_some_and(|handle| handle.spec.runtime == AgentRuntime::Pty); - let completed_replay = verification == "completed_replay"; - if is_pty - && verification - != crate::broker::delivery_verification::HARNESS_ACCEPTANCE - && !completed_replay - { + let Some(verification) = accepted_delivery_verification( + payload.get("verification").and_then(Value::as_str), + is_pty, + ) else { tracing::warn!( target = "agent_relay::broker", worker = %name, delivery_id = %delivery_id, event_id = %event_id, - verification = %verification, - "ignoring PTY delivery verification without harness acceptance" + verification = ?payload.get("verification"), + "ignoring delivery_verified without harness-acceptance evidence" ); let _ = send_event( sdk_out_tx, @@ -1026,12 +1136,13 @@ impl BrokerRuntime { "delivery_id": delivery_id, "event_id": event_id, "reason": "PTY delivery verification did not prove harness acceptance", - "verification": verification, + "verification": payload.get("verification"), }), ) .await; return; - } + }; + let completed_replay = verification == "completed_replay"; let reason = payload.get("reason").and_then(Value::as_str); tracing::debug!( target = "agent_relay::broker", @@ -1205,15 +1316,13 @@ impl BrokerRuntime { .values() .filter(|candidate| candidate.worker_name == name) .count(); + let (state, published_state, transition_reason) = + state_after_terminal_delivery_failure(pending_delivery_count); if let Some(handle) = workers.workers.get_mut(&name) { handle.last_activity_at = Instant::now(); - handle.state = if pending_delivery_count > 0 { - AgentWorkState::BlockedOnSend - } else { - AgentWorkState::Working - }; + handle.state = state; } - if pending_delivery_count > 0 { + if state == AgentWorkState::BlockedOnSend { let _ = send_broker_event( sdk_out_tx, BrokerEvent::AgentBlockedOnSend { @@ -1223,22 +1332,17 @@ impl BrokerRuntime { }, ) .await; - publish_agent_state_transition( - ws_control_tx, - &name, - "stuck", - Some("blocked_on_send"), - ) - .await; - } else { - publish_agent_state_transition( - ws_control_tx, - &name, - "working", - Some("delivery_failed"), - ) - .await; } + // Always publish: an earlier delivery_unconfirmed + // may have published `stuck`, and an in-memory + // reset alone would leave that status visible. + publish_agent_state_transition( + ws_control_tx, + &name, + published_state, + Some(transition_reason), + ) + .await; let _ = emit_dropped_delivery_failures( sdk_out_tx, dead_letters, diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 39c768e6a..d23ff3e78 100644 --- a/crates/broker/src/wrap.rs +++ b/crates/broker/src/wrap.rs @@ -1,6 +1,10 @@ use std::collections::VecDeque; use std::future::Future; use std::pin::Pin; +use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, +}; use std::time::{Duration, Instant}; use crate::{ @@ -23,9 +27,10 @@ use tokio::{sync::mpsc, time::MissedTickBehavior}; use crate::broker::{ delivery_verification::{ - assess_harness_acceptance, DeliveryOutcome, HarnessAcceptance, PendingActivity, - PendingVerification, ThrottleState, VerificationOutput, ACTIVITY_BUFFER_KEEP_BYTES, - ACTIVITY_BUFFER_MAX_BYTES, ACTIVITY_WINDOW, MAX_VERIFICATION_ATTEMPTS, VERIFICATION_WINDOW, + 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, + MAX_VERIFICATION_ATTEMPTS, VERIFICATION_WINDOW, }, injection_format::{format_injection_for_worker_with_workspace, McpReminderThrottle}, }; @@ -132,21 +137,27 @@ pub(crate) fn injection_submit_followup_delay(cli: &str) -> Option { /// Retry only the submit gesture for a body proven to remain in the live /// composer. Never write the body twice: an accepted first write whose activity /// marker was missed would otherwise create a duplicate turn. +/// +/// `cancel` lets human input withdraw the gesture until it reaches the child: +/// the drainer skips a not-yet-started key and withholds Codex's delayed CR. pub(crate) fn submit_injection_recovery( pty: &PtySession, resolved_cli: &str, completed_attempts: usize, + cancel: Arc, ) -> anyhow::Result<(tokio::sync::oneshot::Receiver>, u64)> { let lower = resolved_cli.to_ascii_lowercase(); if lower.contains("codex") && completed_attempts == 1 { // Field recovery for Codex 0.150/0.160: a plain Enter left an idle // multiline composer parked, while End followed by a distinct CR and // later LF submitted it. First reproduce the non-destructive - // End half. The event loop schedules the delayed CR separately so a - // human write during that delay can cancel it before it reaches the - // composer. The final bounded attempt below sends LF only if the same - // delivery is still visibly parked. - return pty.submit_write_paced_with_output_boundary(b"\x1b[F".to_vec(), Duration::ZERO); + // End+Enter half; the final bounded attempt below sends LF only if the + // same delivery is still visibly parked. + return pty.submit_cancellable_write_with_output_boundary( + b"\x1b[F".to_vec(), + Some((Duration::from_secs(1), b"\r".to_vec())), + cancel, + ); } let key = if completed_attempts >= 2 { @@ -154,24 +165,22 @@ pub(crate) fn submit_injection_recovery( } else { b'\r' }; - pty.submit_write_paced_with_output_boundary(vec![key], Duration::ZERO) + pty.submit_cancellable_write_with_output_boundary(vec![key], None, cancel) } -/// Codex's first parked-composer retry uses a delayed CR after cursor-end. -/// Keeping the delay outside the PTY compound-write queue lets the worker -/// re-check human composer ownership before it submits the follow-up key. -pub(crate) fn injection_recovery_followup_delay( - resolved_cli: &str, - completed_attempts: usize, -) -> Option { - (resolved_cli.to_ascii_lowercase().contains("codex") && completed_attempts == 1) - .then_some(Duration::from_secs(1)) +/// Cancel every queued submit-key recovery because a human now owns the input. +pub(crate) fn cancel_recovery_writes(cancels: &mut Vec>) { + for cancel in cancels.drain(..) { + cancel.store(true, Ordering::Release); + } } -pub(crate) fn submit_injection_recovery_followup( - pty: &PtySession, -) -> anyhow::Result<(tokio::sync::oneshot::Receiver>, u64)> { - pty.submit_write_paced_with_output_boundary(b"\r".to_vec(), Duration::ZERO) +/// Start a cancellable recovery, pruning flags whose writes already settled. +pub(crate) fn new_recovery_cancel(cancels: &mut Vec>) -> Arc { + cancels.retain(|cancel| Arc::strong_count(cancel) > 1); + let cancel = Arc::new(AtomicBool::new(false)); + cancels.push(cancel.clone()); + cancel } /// Warn (without retrying) when a one-shot auto-response keystroke can't be @@ -273,7 +282,6 @@ enum PendingWrapWrite { verification: PendingVerification, output_boundary: u64, human_input_generation: u64, - followup_after_ack: Option, }, } @@ -284,9 +292,6 @@ type PendingWrapWriteAck = ( std::result::Result, ); type PendingWrapWriteAckFuture = Pin + Send>>; -type PendingWrapRecoveryFollowup = (PendingVerification, u64); -type PendingWrapRecoveryFollowupFuture = - Pin + Send>>; async fn await_wrap_write_ack( ack_rx: tokio::sync::oneshot::Receiver>, @@ -310,6 +315,14 @@ fn wrap_injection_timer_allowed(has_pending_write_ack: bool, has_pending_accepta !has_pending_write_ack && !has_pending_acceptance } +/// A lone Enter keystroke submits the composer and takes whatever it holds, +/// including a failed relay draft, so it releases the wrap failed-draft latch. +/// A multiline or bracketed paste also carries CR/LF bytes but adds text to +/// the composer rather than submitting it, so it never releases the latch. +fn human_input_submits_composer(data: &[u8]) -> bool { + !data.is_empty() && data.iter().all(|byte| matches!(byte, b'\r' | b'\n')) +} + // Readiness deferrals must not spend the bounded delivery retry budget. fn prepare_wrap_retry( verification: &mut PendingVerification, @@ -1603,8 +1616,6 @@ pub(crate) async fn run_wrap( let mut pending_wrap_injections: VecDeque = VecDeque::new(); let mut pending_wrap_writes: FuturesUnordered = FuturesUnordered::new(); - let mut pending_wrap_recovery_followups: FuturesUnordered = - FuturesUnordered::new(); let mut mcp_reminder_throttle = McpReminderThrottle::new(); // Echo verification state @@ -1671,6 +1682,12 @@ pub(crate) async fn run_wrap( // is hit (at which point the child is not consuming input anyway). let mut stdin_pending: VecDeque> = VecDeque::new(); let mut human_input_generation = 0u64; + let mut recovery_cancels: Vec> = Vec::new(); + // Expected echo of a delivery whose body was written but never proven + // accepted. It may still be typed in the composer, so no later relay + // message is injected on top of it until the composer is provably idle or + // the human submits it. + let mut wrap_failed_draft: Option = None; let mut stdin_retry_deadline: Option = None; const STDIN_RETRY_INTERVAL: Duration = Duration::from_millis(4); @@ -1687,6 +1704,14 @@ pub(crate) async fn run_wrap( // keystrokes are never dropped or reordered under back-pressure. Some(data) = stdin_rx.recv() => { human_input_generation = human_input_generation.saturating_add(1); + cancel_recovery_writes(&mut recovery_cancels); + if wrap_failed_draft.is_some() && human_input_submits_composer(&data) { + tracing::info!( + target = "agent_relay::worker::wrap", + "wrap: human submitted the composer; releasing the failed-draft latch" + ); + wrap_failed_draft = None; + } if !pending_verifications.is_empty() { for verification in pending_verifications.drain(..) { tracing::warn!( @@ -2245,8 +2270,8 @@ pub(crate) async fn run_wrap( _ = pending_injection_interval.tick(), if wrap_injection_timer_allowed( - !pending_wrap_writes.is_empty() || !pending_wrap_recovery_followups.is_empty(), - !pending_verifications.is_empty(), + !pending_wrap_writes.is_empty(), + !pending_verifications.is_empty() || wrap_failed_draft.is_some(), ) => { // Give backlogged human keystrokes priority onto the PTY FIFO // over a new automated injection — see the auto-responder gate @@ -2357,6 +2382,9 @@ pub(crate) async fn run_wrap( "wrap: human input followed the injection write; automatic verification cancelled" ); throttle.record(DeliveryOutcome::Failed); + if draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &injection) { + wrap_failed_draft = Some(injection.clone()); + } continue; } @@ -2414,7 +2442,6 @@ pub(crate) async fn run_wrap( mut verification, output_boundary, human_input_generation: submitted_generation, - followup_after_ack, } => { if submitted_generation != human_input_generation { tracing::warn!( @@ -2422,6 +2449,9 @@ pub(crate) async fn run_wrap( "wrap: human input followed submit-key recovery; further automatic recovery cancelled" ); throttle.record(DeliveryOutcome::Failed); + if draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &verification.expected_echo) { + wrap_failed_draft = Some(verification.expected_echo.clone()); + } continue; } @@ -2432,16 +2462,12 @@ pub(crate) async fn run_wrap( "wrap: submit-key recovery was not confirmed; body left untouched" ); throttle.record(DeliveryOutcome::Failed); + if draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &verification.expected_echo) { + wrap_failed_draft = Some(verification.expected_echo.clone()); + } continue; } - if let Some(delay) = followup_after_ack { - pending_wrap_recovery_followups.push(Box::pin(async move { - tokio::time::sleep(delay).await; - (verification, submitted_generation) - })); - continue; - } tracing::debug!( delivery_id = %verification.delivery_id, @@ -2464,42 +2490,6 @@ pub(crate) async fn run_wrap( } } - Some((verification, submitted_generation)) = pending_wrap_recovery_followups.next(), - if !pending_wrap_recovery_followups.is_empty() => { - if submitted_generation != human_input_generation { - tracing::warn!( - event_id = %verification.event_id, - "wrap: human input took ownership before delayed submit recovery" - ); - throttle.record(DeliveryOutcome::Failed); - continue; - } - match submit_injection_recovery_followup(&pty) { - Ok((ack_rx, output_boundary)) => { - let pending_write = PendingWrapWrite::Retry { - verification, - output_boundary, - human_input_generation: submitted_generation, - followup_after_ack: None, - }; - pending_wrap_writes.push(Box::pin(async move { - ( - pending_write, - await_wrap_write_ack(ack_rx, WRAP_WRITE_ACK_TIMEOUT).await, - ) - })); - } - Err(error) => { - tracing::error!( - event_id = %verification.event_id, - error = %error, - "wrap: failed to queue delayed submit recovery; body left untouched" - ); - throttle.record(DeliveryOutcome::Failed); - } - } - } - // Verification tick: check for timed-out wrap verifications _ = verification_tick.tick() => { let mut i = 0; @@ -2529,7 +2519,6 @@ pub(crate) async fn run_wrap( && !pv.acceptance_expired() && stdin_pending.is_empty() && pending_wrap_writes.is_empty() - && pending_wrap_recovery_followups.is_empty() && crate::devin::can_inject(&resolved_cli, &pty) => { let completed_attempts = pv.attempts; @@ -2546,16 +2535,13 @@ pub(crate) async fn run_wrap( &pty, &resolved_cli, completed_attempts, + new_recovery_cancel(&mut recovery_cancels), ) { Ok((ack_rx, output_boundary)) => { let pending_write = PendingWrapWrite::Retry { verification: pv, output_boundary, human_input_generation, - followup_after_ack: injection_recovery_followup_delay( - &resolved_cli, - completed_attempts, - ), }; pending_wrap_writes.push(Box::pin(async move { ( @@ -2575,6 +2561,9 @@ pub(crate) async fn run_wrap( "wrap: failed to queue submit-key recovery; body left untouched" ); throttle.record(DeliveryOutcome::Failed); + if draft_visible_at_cursor(&Snapshot::capture(&pty), &resolved_cli, &pv.expected_echo) { + wrap_failed_draft = Some(pv.expected_echo.clone()); + } } } } @@ -2582,15 +2571,13 @@ pub(crate) async fn run_wrap( if !pv.acceptance_expired() && (!stdin_pending.is_empty() || !pending_wrap_writes.is_empty() - || !pending_wrap_recovery_followups.is_empty() || !crate::devin::can_inject(&resolved_cli, &pty)) => { - // An in-flight writer or a temporarily busy - // Devin prompt takes priority. Human input - // already drains these verifications in the - // stdin arm, so this deferral never resets the - // total acceptance lifetime. - pv.injected_at = Instant::now(); + // An in-flight writer or a briefly unready + // Devin prompt takes priority. Deferring spends + // no retry budget and never resets the total + // acceptance lifetime. + prepare_wrap_retry(&mut pv, false, Instant::now()); pending_verifications.push_back(pv); } HarnessAcceptance::Parked => { @@ -2600,14 +2587,16 @@ pub(crate) async fn run_wrap( "wrap: body remained parked after bounded submit-key recovery" ); throttle.record(DeliveryOutcome::Failed); + // Parked is the positive evidence. + wrap_failed_draft = Some(pv.expected_echo.clone()); } HarnessAcceptance::Inconclusive => { - tracing::error!( + tracing::info!( event_id = %pv.event_id, attempts = pv.attempts, - "wrap: harness acceptance could not be proven; body left untouched" + "wrap: delivery unconfirmed; delivered without harness evidence" ); - throttle.record(DeliveryOutcome::Failed); + throttle.record(DeliveryOutcome::Unverified); } } } else { @@ -2615,6 +2604,19 @@ pub(crate) async fn run_wrap( } } + if let Some(expected_echo) = wrap_failed_draft.as_deref() { + if pending_verifications.is_empty() + && pending_wrap_writes.is_empty() + && failed_draft_released(&resolved_cli, expected_echo, &Snapshot::capture(&pty)) + { + tracing::info!( + target = "agent_relay::worker::wrap", + "wrap: failed draft left the composer; resuming injection" + ); + wrap_failed_draft = None; + } + } + if activity_detector.is_some() { let mut i = 0; while i < pending_activities.len() { @@ -2635,7 +2637,6 @@ pub(crate) async fn run_wrap( if stdin_pending.is_empty() && pending_verifications.is_empty() && pending_wrap_writes.is_empty() - && pending_wrap_recovery_followups.is_empty() { pty_auto.try_auto_enter(&pty); } @@ -2703,8 +2704,7 @@ mod tests { use super::{ await_wrap_write_ack, buffer_and_drain_stdin, drain_stdin_buffer, injection_submit_followup_delay, submit_injection_body, submit_injection_recovery, - submit_injection_recovery_followup, wrap_injection_timer_allowed, wrap_write_ack_error, - STDIN_PENDING_MAX_CHUNKS, + wrap_injection_timer_allowed, wrap_write_ack_error, STDIN_PENDING_MAX_CHUNKS, }; use crate::broker::delivery_verification::{PendingVerification, MAX_VERIFICATION_ATTEMPTS}; use crate::ids::{DeliveryId, EventId, MessageTarget}; @@ -2791,30 +2791,15 @@ mod tests { let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} }); tokio::time::sleep(Duration::from_millis(100)).await; - let (first_ack, _) = submit_injection_recovery(&pty, "codex", 1).unwrap(); + let (first_ack, _) = + submit_injection_recovery(&pty, "codex", 1, Default::default()).unwrap(); tokio::time::timeout(Duration::from_secs(3), first_ack) .await - .expect("End recovery timed out") - .expect("drainer exited") - .expect("End recovery failed"); - for _ in 0..50 { - if std::fs::read(&log).is_ok_and(|bytes| bytes.len() == 3) { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - assert_eq!( - std::fs::read(&log).unwrap(), - b"\x1b[F", - "the delayed CR must not share the cursor-positioning write" - ); - let (followup_ack, _) = submit_injection_recovery_followup(&pty).unwrap(); - tokio::time::timeout(Duration::from_secs(3), followup_ack) - .await - .expect("CR recovery timed out") + .expect("End+CR recovery timed out") .expect("drainer exited") - .expect("CR recovery failed"); - let (second_ack, _) = submit_injection_recovery(&pty, "codex", 2).unwrap(); + .expect("End+CR recovery failed"); + let (second_ack, _) = + submit_injection_recovery(&pty, "codex", 2, Default::default()).unwrap(); tokio::time::timeout(Duration::from_secs(3), second_ack) .await .expect("LF recovery timed out") @@ -2968,6 +2953,18 @@ sys.stdout.flush()"#; #[test] fn wrap_injection_timer_waits_for_write_ack_and_harness_acceptance() { + assert!(super::human_input_submits_composer(b"\r")); + assert!(super::human_input_submits_composer(b"\r\n")); + assert!( + !super::human_input_submits_composer(b"ok\n"), + "typed line plus newline is text" + ); + assert!( + !super::human_input_submits_composer(b"\x1b[200~line one\rline two\x1b[201~"), + "a bracketed multiline paste is not a submit" + ); + assert!(!super::human_input_submits_composer(b"")); + assert!(!super::human_input_submits_composer(b"typing")); assert!(wrap_injection_timer_allowed(false, false)); assert!(!wrap_injection_timer_allowed(true, false)); assert!(!wrap_injection_timer_allowed(false, true)); diff --git a/crates/broker/tests/pty_injection_integrity.rs b/crates/broker/tests/pty_injection_integrity.rs index 57d71c380..a8caa171b 100644 --- a/crates/broker/tests/pty_injection_integrity.rs +++ b/crates/broker/tests/pty_injection_integrity.rs @@ -37,6 +37,14 @@ data=b'' while True: chunk=os.read(0,65536) with open(os.environ['TRANSCRIPT'],'ab') as f: f.write(chunk) + if mode == 'opaque': + # Consumes input but never shows it, never repaints: no evidence either way. + continue + if mode == 'stuck': + # Shows typed text in the composer but swallows every submit key. + 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) + continue data+=chunk if mode == 'hostile' and b'\x1b\x1b' in data: os.write(1,b'\x1b[2J\x1b[HTrust prompt, no composer') @@ -53,6 +61,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("RELAY_FAILED_DRAFT_ESCALATION_MS", "2000") .env("TRANSCRIPT", dir.path().join("transcript")) .env("AGENT_RELAY_LOCAL_ONLY", "1") .env( @@ -101,6 +110,21 @@ while True: } } } + /// The next frame whose type is one of `kinds`, skipping everything else. + fn wait_any(&self, kinds: &[&str], secs: u64) -> Value { + let deadline = Instant::now() + Duration::from_secs(secs); + loop { + let line = self + .rx + .recv_timeout(deadline.saturating_duration_since(Instant::now())) + .unwrap_or_else(|_| panic!("timed out waiting for {kinds:?}")); + if let Ok(frame) = serde_json::from_str::(&line) { + if kinds.iter().any(|kind| frame["type"] == *kind) { + return frame; + } + } + } + } fn deliver(&mut self, id: &str, body: &str) { self.send("deliver_relay",json!({"delivery_id":format!("del_{id}"),"event_id":id,"from":"broker","target":"integrity","body":body,"priority":2,"injection_mode":"wait"})); } @@ -187,3 +211,80 @@ fn disappearing_composer_fails_before_body_injection() { ); assert!(!String::from_utf8_lossy(&f.transcript()).contains("must not type")); } + +/// Khaliq's #1959 policy: a harness that gives no evidence either way still +/// receives every delivery, each reported unconfirmed, never failed, and an +/// unproven delivery never latches the next one. +#[test] +fn opaque_harness_keeps_receiving_consecutive_deliveries_as_unconfirmed() { + let mut f = Fixture::new("opaque"); + for id in ["opaque_one", "opaque_two", "opaque_three"] { + let body = format!("OPAQUE-BODY-{id}"); + f.deliver(id, &body); + let frame = f.wait_any( + &[ + "delivery_unconfirmed", + "delivery_failed", + "delivery_verified", + ], + 30, + ); + assert_eq!(frame["type"], "delivery_unconfirmed", "{frame}"); + assert_eq!(frame["payload"]["terminal"], true, "{frame}"); + assert_eq!(frame["payload"]["outcome"], "unconfirmed", "{frame}"); + assert!( + String::from_utf8_lossy(&f.transcript()).contains(&body), + "every delivery is written to an opaque harness" + ); + } +} + +/// A body still visibly parked after bounded recovery latches the composer: +/// the next delivery queues and is never typed over the draft, and after the +/// bounded timeout the worker sends one terminal stuck signal. +#[test] +fn visibly_stuck_draft_latches_queue_and_escalates_without_typing_over_it() { + let mut f = Fixture::new("stuck"); + f.deliver("stuck_first", "STUCK-DRAFT-first-delivery"); + let failed = f.wait_any( + &[ + "delivery_failed", + "delivery_verified", + "delivery_unconfirmed", + ], + 60, + ); + // Mid-recovery progress frames are non-terminal delivery_unconfirmed. + let failed = + if failed["type"] == "delivery_unconfirmed" && failed["payload"]["terminal"] != true { + f.wait_any(&["delivery_failed", "delivery_verified"], 60) + } else { + failed + }; + assert_eq!(failed["type"], "delivery_failed", "{failed}"); + assert!( + failed["payload"]["reason"] + .as_str() + .unwrap_or_default() + .contains("parked"), + "{failed}" + ); + f.deliver("stuck_second", "QUEUED-BEHIND-the-stuck-draft"); + let blocked = f.wait_any(&["agent_blocked_on_send"], 30); + let blocked = if blocked["payload"]["reason"] == "failed_draft_latched" + && blocked["payload"]["terminal"] != true + { + f.wait_any(&["agent_blocked_on_send"], 30) + } else { + blocked + }; + assert_eq!( + blocked["payload"]["reason"], "failed_draft_latched", + "{blocked}" + ); + assert_eq!(blocked["payload"]["terminal"], true, "{blocked}"); + assert!( + !String::from_utf8_lossy(&f.transcript()).contains("QUEUED-BEHIND"), + "the queued delivery must never be typed over the stuck draft" + ); +} diff --git a/crates/relay-pty/src/pty.rs b/crates/relay-pty/src/pty.rs index b5ecced05..19a96efb1 100644 --- a/crates/relay-pty/src/pty.rs +++ b/crates/relay-pty/src/pty.rs @@ -132,10 +132,23 @@ enum WriteMsg { /// Relay still must prevent a terminal reply or human keystroke from /// splicing into that boundary. followup: Option, + /// Optional cancellation shared with the submitter. The drainer checks + /// it before the primary bytes and again after the follow-up delay, so + /// a later owner of the input (a human typing) can stop a queued submit + /// key from reaching the child. A cancelled write acks + /// [`io::ErrorKind::Interrupted`] and leaves the drainer running. + cancel: Option>, ack: oneshot::Sender>, }, } +fn write_cancelled() -> io::Error { + io::Error::new( + io::ErrorKind::Interrupted, + "queued PTY write cancelled before it reached the child", + ) +} + struct FollowupWrite { delay: Duration, bytes: Vec, @@ -657,12 +670,22 @@ fn drain_write_queue( bytes, pace, followup, + cancel, ack, } => { if bytes.is_empty() && followup.as_ref().is_none_or(|part| part.bytes.is_empty()) { let _ = ack.send(Ok(())); continue; } + let is_cancelled = || { + cancel + .as_ref() + .is_some_and(|flag| flag.load(Ordering::Acquire)) + }; + if is_cancelled() { + let _ = ack.send(Err(write_cancelled())); + continue; + } if matches!(bytes.as_slice(), b"\x1b[B" | b"\x1b[A" | b"\r" | b"\n") { tracing::debug!(target: "relay_pty::startup_input", control = ?bytes, followup_length = followup.as_ref().map(|part| part.bytes.len()), @@ -683,6 +706,9 @@ fn drain_write_queue( return Ok(()); } thread::sleep(part.delay); + if is_cancelled() { + return Err(write_cancelled()); + } writer.write_all(&part.bytes).and_then(|_| writer.flush()) }); match result { @@ -693,6 +719,10 @@ fn drain_write_queue( no_pid_alive_checks.store(0, Ordering::Relaxed); let _ = ack.send(Ok(())); } + Err(err) if err.kind() == io::ErrorKind::Interrupted && is_cancelled() => { + // Only the follow-up was withheld; the PTY is healthy. + let _ = ack.send(Err(err)); + } Err(err) => { let _ = ack.send(Err(err)); break; @@ -713,12 +743,14 @@ fn enqueue_user_write( bytes: Vec, pace: Duration, followup: Option, + cancel: Option>, ) -> Result>> { let (ack_tx, ack_rx) = oneshot::channel::>(); match write_tx.try_send(WriteMsg::UserInput { bytes, pace, followup, + cancel, ack: ack_tx, }) { Ok(()) => Ok(ack_rx), @@ -969,6 +1001,7 @@ impl PtySession { bytes: Vec, pace: Duration, followup: Option, + cancel: Option>, ) -> Result<(oneshot::Receiver>, u64)> { #[cfg(unix)] let _order_guard = { @@ -1010,6 +1043,7 @@ impl PtySession { bytes, pace, followup, + cancel, )?; Ok((ack, boundary)) } @@ -1286,6 +1320,7 @@ impl PtySession { bytes, Duration::ZERO, None, + None, ) } @@ -1312,7 +1347,14 @@ impl PtySession { bytes: Vec, pace: Duration, ) -> Result>> { - enqueue_user_write(&self.write_tx, &self.no_pid_alive_checks, bytes, pace, None) + enqueue_user_write( + &self.write_tx, + &self.no_pid_alive_checks, + bytes, + pace, + None, + None, + ) } /// Queue a paced input atomically with the receive-time output watermark. @@ -1326,7 +1368,7 @@ impl PtySession { bytes: Vec, pace: Duration, ) -> Result<(oneshot::Receiver>, u64)> { - self.enqueue_write_with_output_boundary(bytes, pace, None) + self.enqueue_write_with_output_boundary(bytes, pace, None, None) } /// Submit a paced primary payload followed by a delayed second write as @@ -1353,6 +1395,7 @@ impl PtySession { delay: followup_delay, bytes: followup_bytes, }), + None, ) } @@ -1372,6 +1415,26 @@ impl PtySession { delay: followup_delay, bytes: followup_bytes, }), + None, + ) + } + + /// Queue a submit-key write, with an optional delayed follow-up, that the + /// caller can cancel until each part reaches the child. Setting `cancel` + /// before the drainer starts the entry skips it entirely; setting it during + /// the follow-up delay withholds the follow-up. Either way the ack resolves + /// to [`io::ErrorKind::Interrupted`]. + pub fn submit_cancellable_write_with_output_boundary( + &self, + bytes: Vec, + followup: Option<(Duration, Vec)>, + cancel: Arc, + ) -> Result<(oneshot::Receiver>, u64)> { + self.enqueue_write_with_output_boundary( + bytes, + Duration::ZERO, + followup.map(|(delay, bytes)| FollowupWrite { delay, bytes }), + Some(cancel), ) } @@ -2216,6 +2279,7 @@ mod tests { b"same-echo\n".to_vec(), Duration::ZERO, None, + None, ) .unwrap(); submitted_tx.send((boundary, ack)).unwrap(); @@ -2614,6 +2678,7 @@ mod tests { bytes: b"injection-body".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx_1, }) .unwrap(); @@ -2623,6 +2688,7 @@ mod tests { bytes: b"\r".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx_2, }) .unwrap(); @@ -2698,6 +2764,7 @@ mod tests { bytes: b"should-fail\n".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx, }) .expect("queue accepts user input"); @@ -2737,6 +2804,7 @@ mod tests { bytes: b"flush-should-fail\n".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx, }) .expect("queue accepts user input"); @@ -2972,6 +3040,7 @@ mod tests { b"queued\n".to_vec(), Duration::ZERO, None, + None, ) .expect("submit path accepts the write"); // Block until the drainer has dequeued the message and is wedged inside @@ -3154,6 +3223,7 @@ mod tests { bytes: b"go\x1b[A\r".to_vec(), pace: StdDuration::from_millis(1), followup: None, + cancel: None, ack: ack_tx, }) .expect("queue accepts paced user input"); @@ -3211,6 +3281,7 @@ mod tests { delay: StdDuration::from_millis(5), bytes: b"\r".to_vec(), }), + cancel: None, ack: ack_tx, }) .expect("queue accepts the compound write"); @@ -3277,6 +3348,7 @@ mod tests { delay: StdDuration::ZERO, bytes: b"\r".to_vec(), }), + cancel: None, ack: ack_tx, }) .expect("queue accepts the compound write"); @@ -3294,6 +3366,95 @@ mod tests { drainer.join().expect("drainer thread joins cleanly"); } + /// A human taking over the composer during a delayed submit recovery must + /// stop the queued follow-up key, not merely relabel its ack afterwards. + #[tokio::test] + async fn cancelled_followup_is_withheld_and_drainer_keeps_running() { + use std::sync::Mutex as StdMutex; + + struct RecordingWriter { + chunks: Arc>>>, + } + impl std::io::Write for RecordingWriter { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.chunks.lock().unwrap().push(buf.to_vec()); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + let chunks = Arc::new(StdMutex::new(Vec::new())); + let (tx, rx) = std_mpsc::sync_channel::(WRITE_QUEUE_DEPTH); + let writer = RecordingWriter { + chunks: chunks.clone(), + }; + let drainer = std::thread::spawn(move || { + super::drain_write_queue(writer, rx, Arc::new(AtomicU32::new(0))) + }); + + let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let (ack_tx, ack_rx) = oneshot::channel::>(); + tx.send(WriteMsg::UserInput { + bytes: b"\x1b[F".to_vec(), + pace: StdDuration::ZERO, + followup: Some(FollowupWrite { + delay: StdDuration::from_millis(200), + bytes: b"\r".to_vec(), + }), + cancel: Some(cancel.clone()), + ack: ack_tx, + }) + .expect("queue accepts the recovery write"); + tokio::time::sleep(Duration::from_millis(50)).await; + cancel.store(true, Ordering::Release); + + let ack = tokio::time::timeout(Duration::from_secs(1), ack_rx) + .await + .expect("drainer resolves the cancelled write") + .expect("ack sender not dropped"); + assert_eq!( + ack.expect_err("cancelled follow-up must not ack success") + .kind(), + std::io::ErrorKind::Interrupted + ); + + let skipped = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let (skip_tx, skip_rx) = oneshot::channel::>(); + tx.send(WriteMsg::UserInput { + bytes: b"\r".to_vec(), + pace: StdDuration::ZERO, + followup: None, + cancel: Some(skipped), + ack: skip_tx, + }) + .expect("queue accepts a pre-cancelled write"); + assert!(skip_rx.await.unwrap().is_err()); + + let (human_tx, human_rx) = oneshot::channel::>(); + tx.send(WriteMsg::UserInput { + bytes: b"human".to_vec(), + pace: StdDuration::ZERO, + followup: None, + cancel: None, + ack: human_tx, + }) + .expect("queue accepts the human write"); + assert!( + human_rx.await.unwrap().is_ok(), + "drainer survives cancellation" + ); + + drop(tx); + drainer.join().expect("drainer thread joins cleanly"); + assert_eq!( + chunks.lock().unwrap().as_slice(), + [b"\x1b[F".to_vec(), b"human".to_vec()], + "neither the cancelled follow-up CR nor the pre-cancelled CR reaches the child" + ); + } + /// Regression: mise/asdf/rtx expose provider CLIs (codex, claude, gemini) /// as symlinked shims that all point at the version-manager binary. Those /// shims dispatch on `argv[0]` (the shim's basename), so canonicalising diff --git a/crates/relay-pty/src/snapshot.rs b/crates/relay-pty/src/snapshot.rs index a00428674..5d288bce8 100644 --- a/crates/relay-pty/src/snapshot.rs +++ b/crates/relay-pty/src/snapshot.rs @@ -189,6 +189,32 @@ impl Snapshot { out } + /// Plain text from the cursor cell through the bottom of the viewport: the + /// complement of [`Self::to_plain_through_cursor`]. Together they let a + /// caller ask whether specific text sits immediately around the cursor. + pub fn to_plain_from_cursor(&self) -> String { + let cursor_row = self.cursor.0.saturating_sub(1) as usize; + let cursor_col = (self.cursor.1.saturating_sub(1) as usize) + .saturating_add(usize::from(self.input_needs_wrap)); + let mut out = String::new(); + for (row_index, row) in self.cells.iter().enumerate().skip(cursor_row) { + let skip = if row_index == cursor_row { + cursor_col.min(row.len()) + } else { + 0 + }; + let row_start = out.len(); + for cell in row.iter().skip(skip) { + out.push(cell.c); + } + while out.len() > row_start && out.ends_with(' ') { + out.pop(); + } + out.push('\n'); + } + out + } + /// Whether the cursor row contains visible cells at or to the right of /// the cursor. A cursor-bounded composer can look like a bare prompt while /// a draft remains in cells to its right, so callers must not classify it diff --git a/tests/e2e/fleet/nodes/stub-agent.cjs b/tests/e2e/fleet/nodes/stub-agent.cjs index f8631dd37..1f17624e2 100644 --- a/tests/e2e/fleet/nodes/stub-agent.cjs +++ b/tests/e2e/fleet/nodes/stub-agent.cjs @@ -59,6 +59,9 @@ 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❯ '); 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 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 c143f087f..dbe3fcf2f 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'; @@ -424,11 +424,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}`);