From 5e601691f2df2f75aae2483ed784c900aea8a2f1 Mon Sep 17 00:00:00 2001 From: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Thu, 13 Aug 2026 15:50:27 -0400 Subject: [PATCH] fix(acp): close edit steering lifecycle gaps Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> (cherry picked from commit 5e03c2c6fe91b2b0873633e12cf8aa17301d8e00) Signed-off-by: Larry <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-acp/src/pool.rs | 113 +++++++++++++++++++++++++++++++----- 1 file changed, 97 insertions(+), 16 deletions(-) diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 7850bdd32..35002988d 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -738,9 +738,11 @@ impl AgentPool { && meta.membership_generation == Some(membership_generation) }) .filter(|meta| { - meta.steer_tx - .as_ref() - .is_some_and(|tx| !tx.is_closed() && tx.capacity() > 0) + meta.control_tx.is_some() + && meta + .steer_tx + .as_ref() + .is_some_and(|tx| !tx.is_closed() && tx.capacity() > 0) }) .map(|meta| meta.turn_id.clone()) } @@ -793,6 +795,11 @@ impl AgentPool { && m.membership_generation == Some(membership_generation) }) .ok_or(SteerError::PromptCompleted)?; + if meta.control_tx.is_none() { + return Err(SteerError::Transport( + "turn cancellation already committed".into(), + )); + } let tx = meta .steer_tx .as_ref() @@ -1652,6 +1659,14 @@ pub async fn run_prompt_task( .unwrap_or_default(); let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone()); + // Establish the edit target as the failure-routing floor before any + // fallible session setup. Once the original event is resolved below this + // is refined to its visible thread, but session/new and initial-message + // failures must never fall back to the invisible edit event or channel root. + if let Some(ref mut batch) = batch { + initialize_edit_failure_routing(batch); + } + // // Core memory is delivered inside the system prompt the harness already // builds (system role for protocol >= 2, the `[System]` user-message @@ -2139,18 +2154,10 @@ pub async fn run_prompt_task( Some(event) => resolve_edit_routing(&event.event, &ctx.rest_client).await, None => None, }; - b.failure_thread_tags = b.events.last().and_then(|event| { - crate::queue::edit_target_id(&event.event).map(|target_event_id| { - resolved_edit - .as_ref() - .map(crate::queue::ResolvedEdit::reply_thread_tags) - .unwrap_or_else(|| crate::queue::ThreadTags { - root_event_id: Some(target_event_id.clone()), - parent_event_id: Some(target_event_id), - mentioned_pubkeys: Vec::new(), - }) - }) - }); + b.failure_thread_tags = resolved_edit + .as_ref() + .map(crate::queue::ResolvedEdit::reply_thread_tags) + .or_else(|| b.failure_thread_tags.clone()); let conversation_context = if ctx.context_message_limit > 0 { match resolved_edit @@ -3121,6 +3128,18 @@ fn conversation_context_delta( } } +/// Establish failure routing for an edit batch without network access. +/// Resolution may later refine this target fallback to the original thread. +fn initialize_edit_failure_routing(batch: &mut FlushBatch) { + batch.failure_thread_tags = batch.events.last().and_then(|event| { + crate::queue::edit_target_id(&event.event).map(|target_event_id| crate::queue::ThreadTags { + root_event_id: Some(target_event_id.clone()), + parent_event_id: Some(target_event_id), + mentioned_pubkeys: Vec::new(), + }) + }); +} + /// Resolve a kind:40003 edit through its original event for reply routing. /// Failure deliberately falls back to the edit target id in `format_prompt`. pub(crate) async fn resolve_edit_routing( @@ -4668,6 +4687,7 @@ mod tests { let mut pool = AgentPool::from_slots(vec![None]); let task_id = pool.join_set.spawn(async {}).id(); let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1); + let (control_tx, _control_rx) = tokio::sync::oneshot::channel(); pool.task_map_mut().insert( task_id, TaskMeta { @@ -4676,7 +4696,7 @@ mod tests { membership_generation: Some(old_generation), turn_id: "pre-removal-turn".into(), recoverable_batch: None, - control_tx: None, + control_tx: Some(control_tx), steer_tx: Some(steer_tx), successful_steer_deliveries: HashSet::new(), }, @@ -4697,6 +4717,67 @@ mod tests { assert!(steer_rx.try_recv().is_err()); } + #[tokio::test] + async fn native_steer_rejects_task_after_cancellation_is_committed() { + let channel_id = Uuid::new_v4(); + let generation = 2; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1); + pool.task_map_mut().insert( + task_id, + TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + membership_generation: Some(generation), + turn_id: "cancelling-turn".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: Some(steer_tx), + successful_steer_deliveries: HashSet::new(), + }, + ); + + assert!(!pool.native_steer_available(channel_id, generation)); + let (ack_tx, _ack_rx) = tokio::sync::oneshot::channel(); + let request = SteerRequest { + prompt_blocks: vec!["late edit".into()], + ack_tx, + }; + assert!(matches!( + pool.send_steer(channel_id, generation, request), + Err(SteerError::Transport(message)) if message.contains("cancellation") + )); + assert!(steer_rx.try_recv().is_err()); + } + + #[test] + fn edit_failure_routing_is_initialized_before_resolution() { + let target = "ab".repeat(32); + let keys = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(40003), "edited mention") + .tags([Tag::parse(["e", target.as_str()]).unwrap()]) + .sign_with_keys(&keys) + .unwrap(); + let mut batch = FlushBatch { + channel_id: Uuid::new_v4(), + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + failure_thread_tags: None, + }; + + initialize_edit_failure_routing(&mut batch); + + let routing = batch.failure_thread_tags.expect("edit target fallback"); + assert_eq!(routing.root_event_id.as_deref(), Some(target.as_str())); + assert_eq!(routing.parent_event_id.as_deref(), Some(target.as_str())); + } + #[test] fn public_session_forwards_channel_origin_to_mcp() { let channel_id = Uuid::new_v4();