diff --git a/crates/broker/src/broker/delivery_verification.rs b/crates/broker/src/broker/delivery_verification.rs index 55c8aa855..97ab93d42 100644 --- a/crates/broker/src/broker/delivery_verification.rs +++ b/crates/broker/src/broker/delivery_verification.rs @@ -185,6 +185,12 @@ impl PendingVerification { if let Some(index) = clean.rfind(&self.expected_echo) { self.activity_buffer .push_str(&clean[index + self.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); @@ -224,6 +230,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() @@ -379,6 +452,17 @@ fn composer_is_idle(snapshot: &Snapshot, cli: &str) -> bool { .unwrap_or(false) } +/// 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. +pub(crate) fn failed_draft_released(cli: &str, expected_echo: &str, snapshot: &Snapshot) -> bool { + expected_tail(expected_echo).is_empty() || composer_is_idle(snapshot, cli) +} + /// Distinguish terminal echo from actual harness acceptance. /// /// Activity is definitive acceptance. A body is considered parked only when @@ -410,7 +494,17 @@ 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() { + // 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) @@ -935,6 +1029,198 @@ mod tests { pty.shutdown().unwrap(); } + #[cfg(unix)] + #[tokio::test] + async fn generic_repaint_cannot_confirm_a_draft_straddling_the_cursor() { + let expected = "Relay message from Lead [evt]: fix idle injection"; + // 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::Inconclusive, + "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::Inconclusive, + "OpenCode transcript words must not confirm a typed-but-unsent body" + ); + 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::Inconclusive, + "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::Inconclusive, + "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_stays_inconclusive() { + // 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::Inconclusive, + "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_stays_inconclusive() { + // 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::Inconclusive, + "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 + .activity_buffer + .push_str("Processing accepted turn"); + + assert_eq!( + assess_harness_acceptance("muse", &verification, &snapshot), + 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(); + } + + #[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"; diff --git a/crates/broker/src/pty_worker.rs b/crates/broker/src/pty_worker.rs index 79c589741..0b97ca822 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, 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}, }; @@ -47,8 +48,8 @@ use crate::util::terminal::{ use crate::util::utf8_stream::Utf8StreamDecoder; use crate::worker::detection::{is_codex_busy_status_line, ActivityDetector}; use crate::wrap::{ - submit_injection_body, submit_injection_recovery, 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; @@ -100,6 +101,54 @@ 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, +} + +/// Record a terminal failure for a delivery whose body may remain in the +/// composer: fence its id against replay for the PTY's lifetime and latch +/// injection until the composer releases the draft. +fn latch_failed_written_delivery( + failed_written_deliveries: &mut HashMap, + failed_composer_latch: &mut Option, + delivery_id: &DeliveryId, + event_id: &EventId, + expected_echo: &str, +) { + failed_written_deliveries.insert(delivery_id.clone(), event_id.clone()); + *failed_composer_latch = Some(FailedComposerLatch { + delivery_id: delivery_id.clone(), + event_id: event_id.clone(), + expected_echo: expected_echo.to_string(), + }); +} + +/// 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() +} + +/// 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 @@ -1047,6 +1096,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`. @@ -1093,6 +1148,9 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { let mut pending_recovery_writes: 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 mut startup_output = String::new(); let mut startup_total_bytes = 0usize; let mut init_request_id: Option = None; @@ -1353,6 +1411,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, @@ -1419,35 +1489,54 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { // child process sees the keystrokes. match frame.payload.get("data").and_then(Value::as_str) { Some(data) => { - 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. - 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); - } // 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, + ); + } + if cancelled_verification { + throttle.record(DeliveryOutcome::Failed); + } + } + match admitted { Ok(ack_rx) => { // Defer the ack until the drainer confirms the // write landed (or failed). The broker holds the @@ -1571,6 +1660,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() { @@ -2085,7 +2187,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(), + 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, @@ -2341,6 +2447,15 @@ 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(), + ); 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" @@ -2355,6 +2470,13 @@ 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(), + ); throttle.record(DeliveryOutcome::Failed); continue; } @@ -2509,17 +2631,10 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { 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(())) => { pv.output_boundary = output_boundary; @@ -2582,13 +2697,38 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { 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, + ); + } } } 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, @@ -2597,6 +2737,13 @@ 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, + ); } Err(_) => { let reason = "submit recovery drainer exited before acknowledging the key"; @@ -2608,6 +2755,13 @@ 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, + ); } } } @@ -2714,7 +2868,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( @@ -2743,6 +2902,13 @@ 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, + ); } } } @@ -2783,6 +2949,17 @@ pub(crate) async fn run_pty_worker(cmd: PtyCommand) -> Result<()> { "attempts": pv.attempts, })).await; throttle.record(DeliveryOutcome::Failed); + // Latch before removing the pending id: the body + // may still be in the composer, so neither a + // replay nor the next queued delivery may be + // typed on top of it. + latch_failed_written_delivery( + &mut failed_written_deliveries, + &mut failed_composer_latch, + &delivery_id, + &event_id, + &pv.expected_echo, + ); pending_worker_delivery_ids.remove(&delivery_id); } } @@ -2791,6 +2968,28 @@ 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; + } + } + if activity_detector.is_some() { let mut i = 0; while i < pending_activities.len() { @@ -3506,9 +3705,13 @@ mod tests { if attempt_started.elapsed() >= VERIFICATION_WINDOW && verification.attempts < verification.max_attempts => { - let (recovery_ack, 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") @@ -3679,7 +3882,8 @@ mod tests { let mut accepted_after = None; for completed_attempts in [1, 2] { let (recovery_ack, boundary) = - submit_injection_recovery(&pty, "codex", completed_attempts).unwrap(); + submit_injection_recovery(&pty, "codex", completed_attempts, Default::default()) + .unwrap(); recovery_ack.await.unwrap().unwrap(); verification.attempts += 1; verification.output_boundary = boundary; @@ -4234,6 +4438,55 @@ mod tests { ); } + #[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", + ); + + 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 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/tests.rs b/crates/broker/src/runtime/tests.rs index b1c4baf68..2d6ab65b1 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -4440,6 +4440,156 @@ async fn stale_recovery_progress_does_not_block_the_worker() { cleanup_worker_registry(fixture.runtime.workers).await; } +fn delivery_verified_worker_event( + 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_worker_event( + 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_worker_event( + 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 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 21983feb9..bbaca3aa3 100644 --- a/crates/broker/src/runtime/worker_events.rs +++ b/crates/broker/src/runtime/worker_events.rs @@ -183,6 +183,54 @@ 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") + ); + } +} + fn worker_event_is_current(current_generation: Option, event_generation: Uuid) -> bool { current_generation == Some(event_generation) } @@ -945,12 +993,24 @@ impl BrokerRuntime { .get("event_id") .and_then(Value::as_str) .unwrap_or(""); - // Modern PTY workers use "harness_acceptance": - // editor echo alone is never delivery confirmation. - let verification = payload - .get("verification") - .and_then(Value::as_str) - .unwrap_or("worker_confirmation"); + let is_pty = workers + .workers + .get(&name) + .is_some_and(|handle| handle.spec.runtime == AgentRuntime::Pty); + 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 = ?payload.get("verification"), + "ignoring delivery_verified without harness-acceptance evidence" + ); + return; + }; let reason = payload.get("reason").and_then(Value::as_str); tracing::debug!( target = "agent_relay::broker", @@ -1064,28 +1124,35 @@ impl BrokerRuntime { ) .await; if let Some(pending) = pending_for_failure { - if let Some(handle) = workers.workers.get_mut(&name) { - handle.last_activity_at = Instant::now(); - handle.state = AgentWorkState::BlockedOnSend; - } let pending_delivery_count = pending_deliveries .values() .filter(|candidate| candidate.worker_name == name) .count(); - let _ = send_broker_event( - sdk_out_tx, - BrokerEvent::AgentBlockedOnSend { - name: name.clone(), - blocked_secs: 0, - pending_delivery_count, - }, - ) - .await; + 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 = state; + } + if state == AgentWorkState::BlockedOnSend { + let _ = send_broker_event( + sdk_out_tx, + BrokerEvent::AgentBlockedOnSend { + name: name.clone(), + blocked_secs: 0, + pending_delivery_count, + }, + ) + .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, - "stuck", - Some("blocked_on_send"), + published_state, + Some(transition_reason), ) .await; let _ = emit_dropped_delivery_failures( diff --git a/crates/broker/src/wrap.rs b/crates/broker/src/wrap.rs index 98ee1fa56..de6ba1a08 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::{ @@ -119,10 +123,14 @@ 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 { @@ -131,11 +139,10 @@ pub(crate) fn submit_injection_recovery( // later LF submitted it. First reproduce the non-destructive // End+Enter half; the final bounded attempt below sends LF only if the // same delivery is still visibly parked. - return pty.submit_write_paced_with_followup_and_output_boundary( + return pty.submit_cancellable_write_with_output_boundary( b"\x1b[F".to_vec(), - Duration::ZERO, - Duration::from_secs(1), - b"\r".to_vec(), + Some((Duration::from_secs(1), b"\r".to_vec())), + cancel, ); } @@ -144,7 +151,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) +} + +/// 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); + } +} + +/// 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 @@ -1566,6 +1588,7 @@ 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(); let mut stdin_retry_deadline: Option = None; const STDIN_RETRY_INTERVAL: Duration = Duration::from_millis(4); @@ -1582,6 +1605,7 @@ 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 !pending_verifications.is_empty() { for verification in pending_verifications.drain(..) { tracing::warn!( @@ -2393,6 +2417,7 @@ 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 { @@ -2629,13 +2654,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+CR recovery timed out") .expect("drainer exited") .expect("End+CR recovery failed"); - let (second_ack, _) = submit_injection_recovery(&pty, "codex", 2).unwrap(); + 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") diff --git a/crates/relay-pty/src/pty.rs b/crates/relay-pty/src/pty.rs index 36e521604..32ebb3480 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, @@ -656,12 +669,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()), @@ -682,6 +705,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 { @@ -692,6 +718,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; @@ -712,12 +742,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), @@ -933,6 +965,7 @@ impl PtySession { bytes: Vec, pace: Duration, followup: Option, + cancel: Option>, ) -> Result<(oneshot::Receiver>, u64)> { #[cfg(unix)] let _order_guard = { @@ -974,6 +1007,7 @@ impl PtySession { bytes, pace, followup, + cancel, )?; Ok((ack, boundary)) } @@ -1221,6 +1255,7 @@ impl PtySession { bytes, Duration::ZERO, None, + None, ) } @@ -1247,7 +1282,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. @@ -1261,7 +1303,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 @@ -1288,6 +1330,7 @@ impl PtySession { delay: followup_delay, bytes: followup_bytes, }), + None, ) } @@ -1307,6 +1350,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), ) } @@ -2023,6 +2086,7 @@ mod tests { b"same-echo\n".to_vec(), Duration::ZERO, None, + None, ) .unwrap(); submitted_tx.send((boundary, ack)).unwrap(); @@ -2421,6 +2485,7 @@ mod tests { bytes: b"injection-body".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx_1, }) .unwrap(); @@ -2430,6 +2495,7 @@ mod tests { bytes: b"\r".to_vec(), pace: Duration::ZERO, followup: None, + cancel: None, ack: ack_tx_2, }) .unwrap(); @@ -2505,6 +2571,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"); @@ -2544,6 +2611,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"); @@ -2779,6 +2847,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 @@ -2961,6 +3030,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"); @@ -3017,6 +3087,7 @@ mod tests { delay: StdDuration::from_millis(5), bytes: b"\r".to_vec(), }), + cancel: None, ack: ack_tx, }) .expect("queue accepts the compound write"); @@ -3083,6 +3154,7 @@ mod tests { delay: StdDuration::ZERO, bytes: b"\r".to_vec(), }), + cancel: None, ack: ack_tx, }) .expect("queue accepts the compound write"); @@ -3100,6 +3172,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