diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 2d5bbe8eb97..ec365cd4739 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2682,31 +2682,8 @@ async fn tokio_main() -> Result<()> { // pending_steer on every return path. If it does, // treat as PromptCompletedNeutral to avoid leaking // the withheld event in `withheld_native_steer`. - let (release_withheld, drop_withheld, signal_fallback) = match &ack { - Ok(pool::SteerAck::Success) => (false, true, false), - // -32601 = method_not_found: agent does not implement the - // steer extension. Fire cancel+merge so the message still - // reaches the agent. - Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. })) - if *code == -32601 => - { - (true, false, true) - } - // AgentError: write landed, agent rejected it at the - // application level (e.g. wrong run id). Release for - // normal dispatch; no fallback signal (the turn is still - // running or just ended — either way there is nothing to - // cancel). - Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => { - (true, false, false) - } - // Transport / ExpectedRunIdMissing / OutcomeRejected: the - // steer did not land. Release and fire the cancel+merge - // fallback so the message still reaches the agent. - Ok(pool::SteerAck::Err(_)) => (true, false, true), - Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false), - Err(_recv_err) => (true, false, false), - }; + let (release_withheld, drop_withheld, signal_fallback) = + steer_ack_disposition(&ack); tracing::info!( channel = %channel_id, event_id = %event_id, @@ -2729,6 +2706,20 @@ async fn tokio_main() -> Result<()> { // will pick it up as part of the merged batch and // re-prompt the agent. signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer); + // Channel-visible: adapters without `_session/steering` + // used to cancel silently. One line per fallback event. + for content in channel_notices_for_steer_ack(&ack, &agent_label) { + let thread_tags = typing_channels + .get(&channel_id) + .cloned() + .unwrap_or_default(); + spawn_channel_notice( + Some(&ctx.rest_client), + channel_id, + thread_tags, + content, + ); + } } // After releasing a withheld event, give dispatch a chance // to re-flush. If the prompt is still in flight, the @@ -2982,6 +2973,59 @@ fn signal_in_flight_task( false } +/// Map a steer ack to `(release_withheld, drop_withheld, signal_fallback)`. +/// +/// Extracted from the main-loop `SteerAckEvent` arm so tests can pin the +/// Success path (unchanged: drop withheld, no fallback) separately from +/// the cancel+merge fallback that now posts a channel-visible notice. +fn steer_ack_disposition( + ack: &std::result::Result, +) -> (bool, bool, bool) { + match ack { + Ok(pool::SteerAck::Success) => (false, true, false), + // -32601 = method_not_found: agent does not implement the + // steer extension. Fire cancel+merge so the message still + // reaches the agent. + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. })) if *code == -32601 => { + (true, false, true) + } + // AgentError: write landed, agent rejected it at the + // application level (e.g. wrong run id). Release for + // normal dispatch; no fallback signal (the turn is still + // running or just ended — either way there is nothing to + // cancel). + Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => (true, false, false), + // Transport / ExpectedRunIdMissing / OutcomeRejected: the + // steer did not land. Release and fire the cancel+merge + // fallback so the message still reaches the agent. + Ok(pool::SteerAck::Err(_)) => (true, false, true), + Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false), + Err(_recv_err) => (true, false, false), + } +} + +/// Channel-visible line posted when cancel+merge fallback fires. +fn cancel_merge_fallback_notice(agent_label: &str) -> String { + format!("{agent_label}'s in-flight turn was cancelled by an incoming message.") +} + +/// Notices the `SteerAckEvent` handler posts to the channel for this ack. +/// +/// Empty on the Success / non-fallback paths. Exactly one notice when +/// `signal_fallback` is true. Production iterates this same function so a +/// test that counts these strings is counting what would be posted. +fn channel_notices_for_steer_ack( + ack: &std::result::Result, + agent_label: &str, +) -> Vec { + let (_, _, signal_fallback) = steer_ack_disposition(ack); + if signal_fallback { + vec![cancel_merge_fallback_notice(agent_label)] + } else { + Vec::new() + } +} + /// Attempt the non-cancelling (ACP) steer for a freshly-queued event. /// /// Caller invariants: @@ -3235,29 +3279,41 @@ fn is_auth_error(error: &acp::AcpError) -> bool { message.contains("Re-authenticate") || message.contains("API Error: 401") } -/// Spawn a task that posts a user-visible failure notice to the relay. +/// Spawn a task that posts a user-visible notice (kind:9) to the relay. /// -/// Shared by the hard-cap immediate dead-letter path and the retries-exhausted -/// dead-letter path so neither duplicates the tokio::spawn block. -fn spawn_failure_notice( +/// Shared by dead-letter failure notices and the cancel+merge fallback +/// notice so neither duplicates the tokio::spawn block. +fn spawn_channel_notice( rest_client: Option<&relay::RestClient>, - batch: &FlushBatch, + channel_id: Uuid, + thread_tags: ThreadTags, content: String, ) { if let Some(rest) = rest_client { - let thread_tags = batch - .events - .last() - .map(|be| queue::parse_thread_tags(&be.event)) - .unwrap_or_default(); let rest = rest.clone(); - let channel_id = batch.channel_id; tokio::spawn(async move { pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await; }); } } +/// Spawn a task that posts a user-visible failure notice to the relay. +/// +/// Shared by the hard-cap immediate dead-letter path and the retries-exhausted +/// dead-letter path so neither duplicates the tokio::spawn block. +fn spawn_failure_notice( + rest_client: Option<&relay::RestClient>, + batch: &FlushBatch, + content: String, +) { + let thread_tags = batch + .events + .last() + .map(|be| queue::parse_thread_tags(&be.event)) + .unwrap_or_default(); + spawn_channel_notice(rest_client, batch.channel_id, thread_tags, content); +} + #[allow(clippy::too_many_arguments)] fn handle_prompt_result( pool: &mut AgentPool, @@ -5464,7 +5520,7 @@ mod error_outcome_emission_tests { use crate::pool::{ AgentPool, OwnedAgent, PromptOutcome, PromptResult, PromptSource, TimeoutKind, }; - use crate::queue::{BatchEvent, FlushBatch}; + use crate::queue::{BatchEvent, FlushBatch, FormatPromptArgs}; use nostr::{EventBuilder, Keys, Kind}; use std::collections::HashSet; @@ -6374,6 +6430,125 @@ mod error_outcome_emission_tests { ); } + /// Clean cancel+merge (`PromptOutcome::Cancelled`): original in-flight + /// batch is requeued as cancelled and merged with the incoming event + /// already released by the steer-ack handler. `format_prompt` is the + /// string the agent actually receives — both events must appear there, + /// not merely inside the queue internals. + #[tokio::test] + async fn cancelled_prompt_redelivers_original_event_to_agent_prompt() { + let keys = Keys::generate(); + let original_event = EventBuilder::new(Kind::Custom(9), "original in-flight work") + .sign_with_keys(&keys) + .unwrap(); + let new_event = EventBuilder::new(Kind::Custom(9), "incoming steer message") + .sign_with_keys(&keys) + .unwrap(); + assert_ne!(original_event.id, new_event.id); + let channel_id = Uuid::new_v4(); + let batch = FlushBatch { + channel_id, + events: vec![BatchEvent { + event: original_event.clone(), + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: Some(CancelReason::Steer), + }; + + let agent = dummy_agent(0).await; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: None, + turn_id: "test-turn-id".to_string(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(QueuedEvent { + channel_id, + event: new_event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: "test".into(), + }); + let config = test_config(); + let mut heartbeat_in_flight = false; + let removed_channels = HashSet::new(); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let observer = ObserverHandle::in_process(); + let result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".to_string(), + outcome: PromptOutcome::Cancelled, + batch: Some(batch), + produced_message: false, + produced_file: false, + }; + + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + Some(observer.clone()), + None, + None, + "test-agent", + ); + + let requeued = queue.flush_next().expect("cancelled batch must re-flush"); + assert_eq!(requeued.events.len(), 1); + assert_eq!(requeued.events[0].event.id, new_event.id); + assert_eq!(requeued.cancelled_events.len(), 1); + assert_eq!(requeued.cancelled_events[0].event.id, original_event.id); + assert_eq!(requeued.cancel_reason, Some(CancelReason::Steer)); + + let prompt = queue::format_prompt(&requeued, &FormatPromptArgs::default()).join("\n\n"); + assert!( + prompt.contains("original in-flight work"), + "original cancelled prompt must reach the agent: {prompt}" + ); + assert!( + prompt.contains("incoming steer message"), + "incoming message must reach the agent: {prompt}" + ); + + assert_eq!( + pool.live_count(), + 1, + "Cancelled returns a healthy agent to the pool" + ); + assert_eq!(respawn_tasks.len(), 0, "Cancelled must not respawn"); + assert_eq!( + observer + .snapshot() + .iter() + .filter(|e| e.kind == "turn_error") + .count(), + 0, + "Cancelled must stay off the turn_error feed" + ); + } + /// Explicit Stop (`ControlSignal::Cancel`) on cancel-drain expiry drops /// the triggering batch — `requeue_cancelled_batch` returns `None` for /// `Cancel`/`Rotate`. The observer payload must be the SAME fate-neutral @@ -6727,6 +6902,91 @@ mod error_outcome_emission_tests { } } +#[cfg(test)] +mod cancel_merge_fallback_notice_tests { + //! Pins the cancel+merge fallback channel notice (WO #301). + //! + //! `channel_notices_for_steer_ack` is the same function the + //! `SteerAckEvent` handler iterates when posting — counting these + //! strings counts what would be posted to the channel. The Success + //! path stays empty: this WO does not touch non-cancelling steer. + + use super::*; + + fn notice_count(ack: pool::SteerAck) -> usize { + channel_notices_for_steer_ack(&Ok(ack), "Cursor").len() + } + + #[test] + fn cancel_merge_fallback_posts_exactly_one_channel_notice_per_fallback_event() { + assert_eq!( + notice_count(pool::SteerAck::Err( + pool::SteerError::ExpectedRunIdMissing + )), + 1, + "ExpectedRunIdMissing (adapter without _session/steering) must post once" + ); + assert_eq!( + notice_count(pool::SteerAck::Err(pool::SteerError::Transport( + "write failed".into() + ))), + 1 + ); + assert_eq!( + notice_count(pool::SteerAck::Err(pool::SteerError::OutcomeRejected { + outcome: "failed".into() + })), + 1 + ); + assert_eq!( + notice_count(pool::SteerAck::Err(pool::SteerError::AgentError { + code: -32601, + message: "Method not found".into() + })), + 1, + "method_not_found must fall back and post once" + ); + + assert_eq!( + notice_count(pool::SteerAck::Success), + 0, + "native steer Success must stay silent" + ); + assert_eq!(notice_count(pool::SteerAck::PromptCompletedNeutral), 0); + assert_eq!( + notice_count(pool::SteerAck::Err(pool::SteerError::AgentError { + code: -32602, + message: "Invalid params".into() + })), + 0, + "application-level AgentError must not cancel or post" + ); + } + + #[test] + fn cancel_merge_fallback_notice_names_seat_and_incoming_cancel() { + let notices = channel_notices_for_steer_ack( + &Ok(pool::SteerAck::Err( + pool::SteerError::ExpectedRunIdMissing + )), + "Cursor", + ); + assert_eq!(notices.len(), 1); + assert_eq!( + notices[0], + "Cursor's in-flight turn was cancelled by an incoming message." + ); + } + + #[test] + fn steer_ack_success_disposition_is_unchanged() { + let (release, drop, fallback) = steer_ack_disposition(&Ok(pool::SteerAck::Success)); + assert!(!release, "Success must not release the withheld event"); + assert!(drop, "Success must drop the withheld event"); + assert!(!fallback, "Success must not fire cancel+merge"); + } +} + #[cfg(test)] mod observer_payload_trim_tests { use super::*; diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de2024..9e4faad582c 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -2028,6 +2028,60 @@ mod tests { assert!(prompt.contains("actually scope it to v2 only")); } + /// Exact ExpectedRunIdMissing cancel+merge fallback sequence: + /// original flushed (in flight) → incoming withheld for native steer → + /// release (ack: no transport) → requeue_as_cancelled → mark_complete → + /// flush_next → format_prompt (the string the agent actually receives). + #[test] + fn test_cancel_merge_fallback_withhold_release_redelivers_to_prompt() { + let mut q = EventQueue::new(DedupMode::Queue); + let ch = Uuid::new_v4(); + + q.push(make_queued(ch, "original in-flight work")); + let batch = q.flush_next().unwrap(); + assert!(any_in_flight(&q)); + + let incoming = make_queued(ch, "incoming steer message"); + let incoming_id = incoming.event.id.to_hex(); + q.push(incoming); + assert!( + q.mark_native_steer_pending(ch, &incoming_id), + "incoming event must be withheld during the native-steer ack window" + ); + assert!( + q.flush_next().is_none(), + "withheld incoming must not flush while the original turn is in flight" + ); + + q.release_native_steer(ch, &incoming_id); + q.requeue_as_cancelled(batch, CancelReason::Steer); + q.mark_complete(ch); + + let merged = q.flush_next().expect("merged cancel+merge batch"); + assert_eq!(merged.events.len(), 1, "incoming event in regular bucket"); + assert_eq!( + merged.cancelled_events.len(), + 1, + "original event in cancelled bucket" + ); + assert_eq!(merged.cancel_reason, Some(CancelReason::Steer)); + assert!(merged.events[0].event.content.contains("incoming steer message")); + assert!(merged.cancelled_events[0] + .event + .content + .contains("original in-flight work")); + + let prompt = format_prompt(&merged, &FormatPromptArgs::default()).join("\n\n"); + assert!( + prompt.contains("original in-flight work"), + "original cancelled prompt must reach the agent: {prompt}" + ); + assert!( + prompt.contains("incoming steer message"), + "incoming message must reach the agent: {prompt}" + ); + } + #[test] fn test_format_prompt_steer_framing_multi_event() { // Multi-event header path must also branch on reason.