diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 0f899dfd4d..8460372aba 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -1617,7 +1617,9 @@ impl AcpClient { "steer accepted as {STEER_OUTCOME_STARTED_NEW_TURN}: \ awaited turn had ended — hard deadline not renewed" ); - crate::pool::SteerAck::Success + crate::pool::SteerAck::Success { + session_id: session_id.to_owned(), + } } Some(_) => { let renew_now = Instant::now(); @@ -1629,7 +1631,9 @@ impl AcpClient { "steer success: renewed hard deadline ({max_duration:?} from now)" ); } - crate::pool::SteerAck::Success + crate::pool::SteerAck::Success { + session_id: session_id.to_owned(), + } } None => { // Report the raw string when @@ -3799,7 +3803,7 @@ mod tests { .await .expect("ack oneshot must have received a SteerAck"); match ack { - crate::pool::SteerAck::Success => {} + crate::pool::SteerAck::Success { .. } => {} other => panic!("expected SteerAck::Success, got {other:?}"), } } @@ -3860,7 +3864,7 @@ mod tests { .await .expect("ack oneshot must have received a SteerAck"); match ack { - crate::pool::SteerAck::Success => {} + crate::pool::SteerAck::Success { .. } => {} other => panic!("expected SteerAck::Success, got {other:?}"), } } @@ -4044,7 +4048,7 @@ mod tests { "_session/steering must not carry expectedRunId; wrote: {written}" ); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "injected outcome must ack Success, got {ack:?}" ); } @@ -4076,7 +4080,7 @@ mod tests { // no `outcome`) — the OutcomeRejected guard applies only to // `_session/steering`. assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "goose success result must ack Success, got {ack:?}" ); } @@ -4181,7 +4185,7 @@ mod tests { assert_eq!(result.unwrap()["done"], serde_json::json!(true)); let ack = ack_rx.await.expect("ack must be received"); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "injected must ack Success, got {ack:?}" ); } @@ -4238,7 +4242,7 @@ mod tests { // rather than released — hence Success, not an Err. let ack = ack_rx.await.expect("ack must be received"); assert!( - matches!(ack, crate::pool::SteerAck::Success), + matches!(ack, crate::pool::SteerAck::Success { .. }), "startedNewTurn is a delivery success, got {ack:?}" ); } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index b69a1f453a..fa348eeb3c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -2821,7 +2821,7 @@ async fn tokio_main() -> Result<()> { // 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), + 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. @@ -2854,8 +2854,19 @@ async fn tokio_main() -> Result<()> { signal_fallback, "non-cancelling steer ack received" ); - if matches!(ack, Ok(pool::SteerAck::Success)) { + if let Ok(pool::SteerAck::Success { session_id }) = &ack { queue.extend_in_flight_deadline(channel_id, config.max_turn_duration_secs); + if !pool.record_successful_steer( + channel_id, + event_id.clone(), + session_id.clone(), + ) { + tracing::warn!( + channel = %channel_id, + event_id = %event_id, + "successful steer lost its in-flight delivery ledger" + ); + } } if drop_withheld { queue.remove_event(channel_id, &event_id); @@ -3318,6 +3329,7 @@ fn dispatch_pending( recoverable_batch, control_tx: Some(control_tx), steer_tx, + successful_steer_deliveries: HashSet::new(), }, ); dispatched_channels.push((channel_id, typing_scope)); @@ -3399,9 +3411,30 @@ fn handle_prompt_result( ) -> LoopAction { let before = pool.task_map().len(); let agent_index = result.agent.index; + let successful_steer_deliveries = pool + .task_map() + .values() + .find(|meta| meta.agent_index == agent_index) + .map(|meta| meta.successful_steer_deliveries.clone()) + .unwrap_or_default(); pool.task_map_mut() .retain(|_, meta| meta.agent_index != agent_index); debug_assert_eq!(before, pool.task_map().len() + 1); + if let PromptSource::Channel(channel_id) = &result.source { + // The task may have invalidated this session before returning. Never + // resurrect delivery state for a dead session; its replacement must + // receive fresh standing context and history. + if let Some(live_session_id) = result.agent.state.sessions.get(channel_id).cloned() { + let event_ids = successful_steer_deliveries + .into_iter() + .filter(|delivery| delivery.session_id == live_session_id) + .map(|delivery| delivery.event_id); + result + .agent + .state + .mark_channel_delivery_success(*channel_id, false, event_ids); + } + } // The hard-timeout death_message (below) must describe the batch's // *actual* fate, not just the `recently_active` eligibility flag — a @@ -3932,6 +3965,7 @@ fn dispatch_heartbeat( recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); *heartbeat_in_flight = true; @@ -4574,17 +4608,23 @@ mod heartbeat_base_prompt_tests { // heartbeat user message, composed as `[Base]\n{bp}\n\n{prompt}`. This is // the second half of the round-2 regression (the first being initial_message). + fn heartbeat_standing() -> queue::StandingContext<'static> { + queue::StandingContext { + base_prompt: Some("you are a helpful agent"), + ..Default::default() + } + } + #[test] fn test_heartbeat_legacy_agent_gets_base_prepended() { // protocol_version 1 + Some(base_prompt): heartbeat prompt is prefixed // with the [Base] section exactly as the legacy session/new path would. let prompt = "[System: Heartbeat]\nrun feed get"; - let composed = pool::prepend_base_for_legacy(1, Some("you are a helpful agent"), prompt); + let composed = pool::prepend_standing_for_legacy(1, &heartbeat_standing(), prompt); assert_eq!( composed, "[Base]\nyou are a helpful agent\n\n[System: Heartbeat]\nrun feed get" ); - assert!(composed.starts_with("[Base]\nyou are a helpful agent\n\n")); } #[test] @@ -4592,7 +4632,7 @@ mod heartbeat_base_prompt_tests { // protocol_version 2 gets base_prompt via session/new; the heartbeat // prompt is sent verbatim. let prompt = "[System: Heartbeat]\nrun feed get"; - let composed = pool::prepend_base_for_legacy(2, Some("you are a helpful agent"), prompt); + let composed = pool::prepend_standing_for_legacy(2, &heartbeat_standing(), prompt); assert_eq!(composed, prompt); } } @@ -4703,6 +4743,7 @@ mod owner_control_command_tests { recoverable_batch: None, control_tx: Some(control_tx), steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); @@ -6473,6 +6514,263 @@ mod error_outcome_emission_tests { } } + #[tokio::test] + async fn successful_native_steer_is_transferred_to_live_session_delivery_state() { + let channel_id = Uuid::new_v4(); + let steer_event_id = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let mut agent = dummy_agent(0).await; + agent + .state + .sessions + .insert(channel_id, "live-session".into()); + agent + .state + .deliveries + .insert(channel_id, Default::default()); + + 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: Some(channel_id), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::from([ + crate::pool::SuccessfulSteerDelivery { + event_id: steer_event_id.into(), + session_id: "live-session".into(), + }, + ]), + }, + ); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + 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 result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + + let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); + assert!(returned.state.deliveries[&channel_id] + .delivered_event_ids + .contains(steer_event_id)); + } + + #[tokio::test] + async fn in_flight_stale_native_steer_ack_cannot_update_replacement_session() { + let channel_id = Uuid::new_v4(); + let mut agent = dummy_agent(0).await; + agent + .state + .sessions + .insert(channel_id, "replacement-session".into()); + agent + .state + .deliveries + .insert(channel_id, Default::default()); + + 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: Some(channel_id), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::from([ + crate::pool::SuccessfulSteerDelivery { + event_id: "stale-event".into(), + session_id: "old-session".into(), + }, + ]), + }, + ); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + 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 result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + + let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); + assert!(returned.state.deliveries[&channel_id] + .delivered_event_ids + .is_empty()); + } + + #[tokio::test] + async fn successful_native_steer_ack_after_task_return_updates_matching_live_session() { + let channel_id = Uuid::new_v4(); + let steer_event_id = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + let mut agent = dummy_agent(0).await; + agent + .state + .sessions + .insert(channel_id, "live-session".into()); + agent + .state + .deliveries + .insert(channel_id, Default::default()); + let mut pool = AgentPool::from_slots(vec![Some(agent)]); + + assert!(pool.record_successful_steer( + channel_id, + steer_event_id.into(), + "live-session".into(), + )); + let returned = pool.agents_mut()[0].as_ref().expect("idle returned agent"); + assert!(returned.state.deliveries[&channel_id] + .delivered_event_ids + .contains(steer_event_id)); + } + + #[tokio::test] + async fn late_native_steer_ack_cannot_update_replacement_session() { + let channel_id = Uuid::new_v4(); + let mut agent = dummy_agent(0).await; + agent + .state + .sessions + .insert(channel_id, "replacement-session".into()); + agent + .state + .deliveries + .insert(channel_id, Default::default()); + let mut pool = AgentPool::from_slots(vec![Some(agent)]); + + assert!(!pool.record_successful_steer( + channel_id, + "stale-event".into(), + "old-session".into(), + )); + let returned = pool.agents_mut()[0].as_ref().expect("replacement agent"); + assert!(returned.state.deliveries[&channel_id] + .delivered_event_ids + .is_empty()); + } + + #[tokio::test] + async fn invalidated_session_does_not_resurrect_successful_steer_delivery_state() { + let channel_id = Uuid::new_v4(); + let agent = dummy_agent(0).await; + // No live session: simulates the prompt task invalidating before return. + 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: Some(channel_id), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::from([ + crate::pool::SuccessfulSteerDelivery { + event_id: "stale-event".into(), + session_id: "invalidated-session".into(), + }, + ]), + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + 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 result = PromptResult { + agent, + source: PromptSource::Channel(channel_id), + turn_id: "test-turn-id".into(), + outcome: PromptOutcome::Ok(crate::acp::StopReason::EndTurn), + batch: None, + }; + + handle_prompt_result( + &mut pool, + &mut queue, + &config, + result, + &mut heartbeat_in_flight, + &removed_channels, + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + + let returned = pool.agents_mut()[0].as_ref().expect("returned agent"); + assert!(!returned.state.deliveries.contains_key(&channel_id)); + } + /// Drive one error outcome through `handle_prompt_result` and return how /// many `turn_error` events it emitted to the observer feed. async fn turn_errors_emitted_for(outcome: PromptOutcome) -> usize { @@ -6493,6 +6791,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); @@ -6569,6 +6868,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); started_rx.await.unwrap(); @@ -6661,6 +6961,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6752,6 +7053,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6857,6 +7159,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -6933,6 +7236,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -7027,6 +7331,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let config = test_config(); @@ -7143,6 +7448,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -7282,6 +7588,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -7470,6 +7777,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); @@ -7555,6 +7863,7 @@ mod error_outcome_emission_tests { recoverable_batch: None, control_tx: None, steer_tx: None, + successful_steer_deliveries: HashSet::new(), }, ); let mut queue = EventQueue::new(config::DedupMode::Queue); diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index fd18bda98d..33bd5507fb 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -50,6 +50,12 @@ const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60); // a recoverable copy in TaskMeta for panic recovery in Queue mode. /// Metadata stored per in-flight task for panic recovery. +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub struct SuccessfulSteerDelivery { + pub event_id: String, + pub session_id: String, +} + pub struct TaskMeta { pub agent_index: usize, pub channel_id: Option, @@ -67,6 +73,10 @@ pub struct TaskMeta { /// tasks only — all prompt tasks install a steer channel regardless /// of the agent's name. pub steer_tx: Option>, + /// Successful non-cancelling steers acknowledged while this task owned the + /// live session. The session ID prevents a late ack from contaminating a + /// replacement session after task return. + pub successful_steer_deliveries: HashSet, } /// Agent-level model capabilities. Populated on first session creation. @@ -80,7 +90,17 @@ pub struct AgentModelCapabilities { pub available_models_raw: Option, } -/// Per-channel session IDs and turn counters. +/// Successful deliveries associated with one live channel session. +#[derive(Default)] +pub struct ChannelDeliveryState { + /// Whether a legacy user message has successfully carried standing context. + pub standing_context_sent: bool, + /// Buzz event IDs already delivered to this ACP session, either as trigger + /// events or conversation context. + pub delivered_event_ids: HashSet, +} + +/// Per-channel session IDs, turn counters, and delivery state. /// /// Separated from `OwnedAgent` so the state machine is testable without /// spawning a real agent subprocess. @@ -94,6 +114,8 @@ pub struct SessionState { pub turn_counts: HashMap, /// Turn counter for the heartbeat session. pub heartbeat_turn_count: u32, + /// Whether the live heartbeat session has successfully received `[Base]`. + pub heartbeat_standing_context_sent: bool, /// channel_id → rendered NIP-AE core prompt section, populated once at /// session creation per Tyler's spec (no mid-session refresh). pub core_sections: HashMap, @@ -104,6 +126,9 @@ pub struct SessionState { /// fetch fails — all fail open. Cleared on session invalidation alongside /// `core_sections` so the next session picks up any canvas change. pub canvas_sections: HashMap, + /// Per-channel successful-delivery state. Created with the ACP session and + /// cleared atomically with every invalidation path. + pub deliveries: HashMap, } impl SessionState { @@ -116,6 +141,7 @@ impl SessionState { PromptSource::Heartbeat => { self.heartbeat_session = None; self.heartbeat_turn_count = 0; + self.heartbeat_standing_context_sent = false; } } } @@ -126,6 +152,7 @@ impl SessionState { self.turn_counts.remove(channel_id); self.core_sections.remove(channel_id); self.canvas_sections.remove(channel_id); + self.deliveries.remove(channel_id); self.sessions.remove(channel_id).is_some() } @@ -135,8 +162,21 @@ impl SessionState { self.turn_counts.clear(); self.heartbeat_session = None; self.heartbeat_turn_count = 0; + self.heartbeat_standing_context_sent = false; self.core_sections.clear(); self.canvas_sections.clear(); + self.deliveries.clear(); + } + + pub(crate) fn mark_channel_delivery_success( + &mut self, + channel_id: Uuid, + standing_context_sent: bool, + event_ids: impl IntoIterator, + ) { + let delivery = self.deliveries.entry(channel_id).or_default(); + delivery.standing_context_sent |= standing_context_sent; + delivery.delivered_event_ids.extend(event_ids); } #[cfg(test)] @@ -145,6 +185,7 @@ impl SessionState { || self.turn_counts.contains_key(channel_id) || self.core_sections.contains_key(channel_id) || self.canvas_sections.contains_key(channel_id) + || self.deliveries.contains_key(channel_id) } } @@ -408,7 +449,7 @@ pub enum SteerAck { /// The agent returned a successful response to the steer request. /// The main loop must drop the withheld event (`remove_event`) — it /// has been delivered via the non-cancelling path. - Success, + Success { session_id: String }, /// The steer was attempted but failed. Delivery state for the /// underlying message is unknown after prompt completion; the main /// loop must release the withheld event and fall back to the @@ -696,6 +737,40 @@ impl AgentPool { .map_err(|e| SteerError::Transport(e.to_string())) } + /// Durably associate a successful steer with the exact ACP session that + /// accepted it. Acks may arrive before or after the prompt result: while + /// the task is in flight we stage the delivery in `TaskMeta`; after return + /// we write directly to the idle agent's matching live-session ledger. + pub fn record_successful_steer( + &mut self, + channel_id: Uuid, + event_id: String, + session_id: String, + ) -> bool { + if let Some(meta) = self + .task_map + .values_mut() + .find(|meta| meta.channel_id == Some(channel_id)) + { + meta.successful_steer_deliveries + .insert(SuccessfulSteerDelivery { + event_id, + session_id, + }); + return true; + } + + let Some(agent) = self.agents.iter_mut().flatten().find(|agent| { + agent.state.sessions.get(&channel_id).map(String::as_str) == Some(session_id.as_str()) + }) else { + return false; + }; + agent + .state + .mark_channel_delivery_success(channel_id, false, [event_id]); + true + } + pub fn result_tx(&self) -> mpsc::UnboundedSender { self.result_tx.clone() } @@ -1205,42 +1280,30 @@ async fn apply_permission_mode( Ok(()) } -/// Prepend the `[Base]` section to a user-message body for legacy agents. +/// Prepend a legacy agent's standing context to a user-message body. /// -/// Legacy agents (`protocol_version < 2`) don't receive `base_prompt` via the -/// system role in `session/new`, so it must ride along in the user message. -/// Agents with `protocol_version >= 2`, or any agent without a `base_prompt`, -/// get `body` unchanged. The gate lives here so the heartbeat and -/// initial-message dispatch paths can't drift apart again. -pub(crate) fn prepend_base_for_legacy( - protocol_version: u32, - base_prompt: Option<&str>, - body: &str, -) -> String { - match base_prompt { - Some(bp) if protocol_version < 2 => { - format!("{}\n\n{body}", crate::queue::base_section(bp)) - } - _ => body.to_string(), - } -} - -/// Prepend the `[Channel Canvas]` section to the legacy initial-message body. +/// Legacy agents (`protocol_version < 2`) don't receive standing context via +/// the system role in `session/new`, so it must ride along in the user message +/// — in the session's *first* one, and never again. Agents with +/// `protocol_version >= 2`, or an empty [`StandingContext`], get `body` +/// unchanged. Both legacy dispatch paths (initial message, heartbeat) go +/// through this one gate so they can't drift apart again. /// -/// Protocol-v2 agents already receive the canvas in `systemPrompt`; only -/// legacy (protocol_version < 2) agents need it injected here so it arrives -/// before the first prompt — the same "every turn" semantics as per-turn core. -/// Heartbeats never have an initial_message, so the caller is responsible for -/// not passing a canvas when `source` is `Heartbeat`. -pub(crate) fn prepend_canvas_for_legacy( +/// A heartbeat passes base only: it has no channel, so there is no core or +/// canvas to carry, and it has never been given the persona. +pub(crate) fn prepend_standing_for_legacy( protocol_version: u32, - agent_canvas: Option<&str>, + standing: &crate::queue::StandingContext<'_>, body: &str, ) -> String { - match agent_canvas { - Some(canvas) if protocol_version < 2 => format!("{canvas}\n\n{body}"), - _ => body.to_string(), + if protocol_version >= 2 { + return body.to_string(); + } + let sections = standing.sections(); + if sections.is_empty() { + return body.to_string(); } + format!("{}\n\n{body}", sections.join("\n\n")) } /// Frame the `session/new` `systemPrompt` so each present prompt carries its own @@ -1619,6 +1682,10 @@ pub async fn run_prompt_task( "created session {sid} for channel {cid}" ); agent.state.sessions.insert(*cid, sid.clone()); + agent + .state + .deliveries + .insert(*cid, ChannelDeliveryState::default()); // Seed a zero usage baseline: buzz-acp spawned this session // so prior usage is zero by definition — first turn is reliable. agent.acp.notify_session_spawned(&sid); @@ -1718,6 +1785,32 @@ pub async fn run_prompt_task( }), ); + // Standing context is fixed for the life of a session. Agents with + // systemPrompt support already hold it from session/new; legacy agents + // receive it in the session's first user message and never again. + // + // `is_new_session` comes from the session registry, which is cleared + // whenever a session is invalidated — so the replacement session re-delivers + // rather than leaving the agent unbriefed. + let standing = crate::queue::StandingContext { + base_prompt: ctx.base_prompt, + system_prompt: ctx.system_prompt.as_deref(), + team_instructions: ctx.team_instructions.as_deref(), + agent_core: agent_core.as_deref(), + agent_canvas: agent_canvas.as_deref(), + }; + // Delivery state is committed only after ACP confirms success. Existing + // sessions created before this field existed fail safe by behaving as + // undelivered once, rather than silently omitting standing context. + let mut standing_context_sent = match &source { + PromptSource::Channel(cid) => agent + .state + .deliveries + .get(cid) + .is_some_and(|delivery| delivery.standing_context_sent), + PromptSource::Heartbeat => agent.state.heartbeat_standing_context_sent, + }; + if is_new_session { if let (PromptSource::Channel(cid), Some(ref initial_msg)) = (&source, &ctx.initial_message) { @@ -1725,30 +1818,15 @@ pub async fn run_prompt_task( target: "pool::session", "sending initial_message to session {session_id} for channel {cid}" ); - // For agents with systemPrompt support (protocol_version >= 2), - // base_prompt is delivered via the system role in session/new. - // Legacy agents receive it via [Base] in the user message instead. - // Canvas is also injected here for legacy agents: protocol-v2 agents - // already have it in systemPrompt; legacy agents need it before the - // first prompt, matching the "every turn" per-turn delivery semantics. - let init_msg = prepend_base_for_legacy( + let init_msg = prepend_standing_for_legacy( if agent.has_system_prompt_support() { 2 } else { 1 }, - ctx.base_prompt, + &standing, initial_msg, ); - let init_msg = prepend_canvas_for_legacy( - if agent.has_system_prompt_support() { - 2 - } else { - 1 - }, - agent_canvas.as_deref(), - &init_msg, - ); let init_result = agent .acp .session_prompt_with_idle_timeout( @@ -1765,6 +1843,12 @@ pub async fn run_prompt_task( target: "pool::session", "initial_message complete for channel {cid}: {stop_reason:?}" ); + // The legacy agent has its standing context now; the turn + // prompt below must not repeat it. Every other arm returns. + standing_context_sent = true; + if !agent.has_system_prompt_support() { + agent.state.mark_channel_delivery_success(*cid, true, []); + } } Err(AcpError::AgentExited) => { agent.state.invalidate_all(); @@ -1866,18 +1950,31 @@ pub async fn run_prompt_task( // (`prompt[0].text.startsWith("/")`) fires; the wrapped Buzz context // follows as a second block. let mut slash_command: Option = None; + // Event IDs represented by this prompt. Commit only after ACP reports a + // successful turn; failed/cancelled prompts must be retryable without loss. + let mut pending_delivered_event_ids = HashSet::new(); let prompt_sections: Vec = if let Some(text) = prompt_text { // Heartbeats create their session before this point, so a Goose method-not-found // probe has already selected the correct framing for this process. - let text = prepend_base_for_legacy( - if agent.has_system_prompt_support() { - 2 - } else { - 1 - }, - ctx.base_prompt, - &text, - ); + // + // Only the first heartbeat of a session carries `[Base]`; later ticks + // reuse the same session, so the agent already has it. + let text = if standing_context_sent { + text + } else { + prepend_standing_for_legacy( + if agent.has_system_prompt_support() { + 2 + } else { + 1 + }, + &crate::queue::StandingContext { + base_prompt: ctx.base_prompt, + ..Default::default() + }, + &text, + ) + }; vec![text] } else if let Some(ref b) = batch { // Build prompt from batch with context enrichment. @@ -1889,6 +1986,31 @@ pub async fn run_prompt_task( } else { None }; + let rendered_batch_ids: HashSet = b + .events + .iter() + .chain(b.cancelled_events.iter()) + .map(|event| event.event.id.to_hex()) + .collect(); + let delivered_ids = agent + .state + .deliveries + .get(&b.channel_id) + .map(|delivery| &delivery.delivered_event_ids) + .cloned() + .unwrap_or_default(); + let conversation_context_had_delivered_events = + conversation_context.as_ref().is_some_and(|context| { + conversation_context_event_ids(Some(context)) + .iter() + .any(|event_id| delivered_ids.contains(event_id)) + }); + let conversation_context = + conversation_context_delta(conversation_context, &delivered_ids, &rendered_batch_ids); + pending_delivered_event_ids.extend(rendered_batch_ids); + pending_delivered_event_ids.extend(conversation_context_event_ids( + conversation_context.as_ref(), + )); let profile_lookup = fetch_prompt_profile_lookup(b, conversation_context.as_ref(), &ctx.rest_client).await; @@ -1912,15 +2034,17 @@ pub async fn run_prompt_task( crate::queue::format_prompt( b, &crate::queue::FormatPromptArgs { - agent_core: agent_core.as_deref(), + agent_core: standing.agent_core, channel_info: channel_info.as_ref(), conversation_context: conversation_context.as_ref(), + conversation_context_had_delivered_events, profile_lookup: profile_lookup.as_ref(), has_system_prompt_support: agent.has_system_prompt_support(), - base_prompt: ctx.base_prompt, - system_prompt: ctx.system_prompt.as_deref(), - team_instructions: ctx.team_instructions.as_deref(), - agent_canvas: agent_canvas.as_deref(), + base_prompt: standing.base_prompt, + system_prompt: standing.system_prompt, + team_instructions: standing.team_instructions, + agent_canvas: standing.agent_canvas, + standing_context_sent, }, ) } else { @@ -1960,6 +2084,28 @@ pub async fn run_prompt_task( .collect(), None => prompt_sections.iter().map(String::as_str).collect(), }; + let prompt_bytes: usize = prompt_blocks.iter().map(|block| block.len()).sum(); + let has_standing_context = match &source { + PromptSource::Channel(_) => !standing.sections().is_empty(), + PromptSource::Heartbeat => ctx.base_prompt.is_some(), + }; + let standing_context_included = + !agent.has_system_prompt_support() && !standing_context_sent && has_standing_context; + tracing::info!( + target: "pool::prompt", + prompt_bytes, + standing_context_included, + delivered_event_delta = pending_delivered_event_ids.len(), + "prompt context delivery" + ); + agent.acp.observe( + "prompt_context_delivery", + serde_json::json!({ + "promptBytes": prompt_bytes, + "standingContextIncluded": standing_context_included, + "eventDeltaCount": pending_delivered_event_ids.len(), + }), + ); // Turn start, labelled exactly as `log_stop_reason` labels the end, so a // log reads as start/stop pairs. Purely observational: an unpaired start is @@ -2110,6 +2256,14 @@ pub async fn run_prompt_task( "control signal arrived but turn already completed — treating as success" ); } + if let PromptSource::Channel(cid) = &source { + let standing_sent = !agent.has_system_prompt_support(); + agent.state.mark_channel_delivery_success( + *cid, + standing_sent, + pending_delivered_event_ids.iter().cloned(), + ); + } apply_completed_before_control_signal( &mut agent.state, &source, @@ -2144,6 +2298,17 @@ pub async fn run_prompt_task( Ok(stop_reason) => { log_stop_reason(&source, &stop_reason); + if let PromptSource::Channel(cid) = &source { + let standing_sent = !agent.has_system_prompt_support(); + agent.state.mark_channel_delivery_success( + *cid, + standing_sent, + pending_delivered_event_ids.iter().cloned(), + ); + } else if !agent.has_system_prompt_support() { + agent.state.heartbeat_standing_context_sent = true; + } + let should_rotate = matches!( stop_reason, StopReason::MaxTokens | StopReason::MaxTurnRequests @@ -2638,6 +2803,67 @@ pub(crate) fn render_canvas_section(event_id: &str, timestamp: &str, channel_uui ) } +fn conversation_context_event_ids(context: Option<&ConversationContext>) -> HashSet { + match context { + Some(ConversationContext::Thread { messages, .. }) + | Some(ConversationContext::Dm { messages, .. }) => messages + .iter() + .filter(|message| !message.event_id.is_empty()) + .map(|message| message.event_id.clone()) + .collect(), + None => HashSet::new(), + } +} + +/// Remove events already delivered to this live ACP session. Triggering events +/// are also excluded because they are rendered separately in `[Event]`. +/// IDs are compared in Buzz's canonical 64-character lowercase hex form: relay +/// context JSON supplies the same form emitted by `EventId::to_hex()`. A +/// non-canonical or missing ID deliberately fails open and may be re-sent. +fn conversation_context_delta( + context: Option, + delivered: &HashSet, + triggering: &HashSet, +) -> Option { + let filter = |messages: Vec| { + messages + .into_iter() + .filter(|message| { + message.event_id.is_empty() + || (!delivered.contains(&message.event_id) + && !triggering.contains(&message.event_id)) + }) + .collect::>() + }; + + match context? { + ConversationContext::Thread { + messages, + total, + truncated, + } => { + let messages = filter(messages); + (!messages.is_empty()).then_some(ConversationContext::Thread { + messages, + total, + truncated, + }) + } + ConversationContext::Dm { + messages, + total, + truncated, + } => { + let messages = filter(messages); + (!messages.is_empty()).then_some(ConversationContext::Dm { + messages, + total, + truncated, + }) + } + } +} + /// Fetch conversation context (thread or DM) for a batch before prompting. /// /// Returns `None` if: @@ -3150,7 +3376,14 @@ fn json_to_context_message(obj: &serde_json::Value) -> Option { }) .unwrap_or_else(|| "unknown".to_string()); + let event_id = obj + .get("id") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(); + Some(ContextMessage { + event_id, pubkey: pubkey.to_string(), timestamp, content: content.to_string(), @@ -4083,23 +4316,46 @@ mod tests { // a legacy agent WITH a base_prompt must get [Base] prepended to the user // message. This is the exact regression that shipped in the round-2 bug. + fn base_only(base_prompt: Option<&str>) -> crate::queue::StandingContext<'_> { + crate::queue::StandingContext { + base_prompt, + ..Default::default() + } + } + #[test] fn test_initial_message_legacy_agent_gets_base_prepended() { // protocol_version 1 + Some(base_prompt): [Base] rides along in the // user message, composed as `[Base]\n{bp}\n\n{initial_msg}`. - let composed = prepend_base_for_legacy(1, Some("you are a helpful agent"), "hello channel"); + let composed = prepend_standing_for_legacy( + 1, + &base_only(Some("you are a helpful agent")), + "hello channel", + ); assert_eq!(composed, "[Base]\nyou are a helpful agent\n\nhello channel"); - assert!(composed.starts_with("[Base]\nyou are a helpful agent\n\n")); } #[test] fn test_initial_message_modern_agent_omits_base() { // protocol_version 2 receives base_prompt via session/new, so the user // message is left untouched even when a base_prompt is present. - let composed = prepend_base_for_legacy(2, Some("you are a helpful agent"), "hello channel"); + let composed = prepend_standing_for_legacy( + 2, + &base_only(Some("you are a helpful agent")), + "hello channel", + ); assert_eq!(composed, "hello channel"); } + #[test] + fn test_heartbeat_standing_block_is_base_only() { + // A heartbeat has no channel, so core and canvas are absent by + // construction — and it has never carried the persona. Pin that the + // shared helper does not start handing heartbeats [System]. + let composed = prepend_standing_for_legacy(1, &base_only(Some("be helpful")), "tick"); + assert_eq!(composed, "[Base]\nbe helpful\n\ntick"); + } + #[test] fn goose_uses_system_prompt_only_after_custom_method_succeeds() { assert!(!has_system_prompt_support(2, "goose", None)); @@ -4154,82 +4410,75 @@ mod tests { #[test] fn test_initial_message_legacy_agent_without_base_is_unchanged() { // No base_prompt configured: nothing to prepend regardless of version. - let composed = prepend_base_for_legacy(1, None, "hello channel"); + let composed = prepend_standing_for_legacy(1, &base_only(None), "hello channel"); assert_eq!(composed, "hello channel"); } - // ── prepend_canvas_for_legacy ───────────────────────────────────────────── + // ── prepend_standing_for_legacy ─────────────────────────────────────────── + + fn full_standing() -> crate::queue::StandingContext<'static> { + crate::queue::StandingContext { + base_prompt: Some("be helpful"), + system_prompt: Some("you are Eva"), + team_instructions: Some("ship small"), + agent_core: Some("[Agent Memory — core]\nremember this"), + agent_canvas: Some("[Channel Canvas]\ncanvas content"), + } + } #[test] - fn test_initial_message_legacy_agent_gets_canvas_prepended() { - // Legacy agents (protocol_version < 2) receive the canvas section before - // the initial-message body so it arrives before the first prompt. - let canvas = "[Channel Canvas]\nCanvas revision (event ID): abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234\nLast modified: 2024-01-15T10:30:00Z\nFetch current content with: buzz canvas get --channel 00f1ccaf-1506-4dd7-9a0e-fa67e9e486ae"; - let composed = prepend_canvas_for_legacy(1, Some(canvas), "do the thing"); - assert!( - composed.starts_with("[Channel Canvas]"), - "canvas must precede the body" - ); - assert!( - composed.ends_with("do the thing"), - "body must follow the canvas" - ); + fn test_initial_message_legacy_agent_gets_whole_standing_block() { + // The initial message is the legacy agent's first contact, so it must + // carry every standing section — not just [Base] and the canvas, which + // left the agent acting on its first turn with no persona and no memory. + let composed = prepend_standing_for_legacy(1, &full_standing(), "do the thing"); + let positions: Vec = [ + "[Base]", + "[System]", + "[Team Instructions]", + "[Agent Memory — core]", + "[Channel Canvas]", + "do the thing", + ] + .iter() + .map(|needle| { + composed + .find(needle) + .unwrap_or_else(|| panic!("missing {needle} in: {composed}")) + }) + .collect(); assert!( - composed.contains("\n\ndo the thing"), - "canvas and body separated by blank line" + positions.windows(2).all(|w| w[0] < w[1]), + "sections must match the per-turn order, body last; got: {composed}" ); } #[test] - fn test_initial_message_modern_agent_omits_canvas_from_body() { - // Protocol-v2 agents receive canvas in systemPrompt; it must NOT be - // duplicated in the initial-message user turn. - let canvas = "[Channel Canvas]\nsome section"; - let composed = prepend_canvas_for_legacy(2, Some(canvas), "do the thing"); + fn test_initial_message_standing_order_matches_per_turn_order() { + // Both legacy paths render through StandingContext, so the initial + // message and a first-turn prompt agree section-for-section. + let standing = full_standing(); + let composed = prepend_standing_for_legacy(1, &standing, "do the thing"); assert_eq!( - composed, "do the thing", - "modern agent initial message must not contain canvas" - ); - assert!( - !composed.contains("[Channel Canvas]"), - "canvas must be absent from modern agent initial message" + composed, + format!("{}\n\ndo the thing", standing.sections().join("\n\n")) ); } #[test] - fn test_initial_message_legacy_agent_no_canvas_is_unchanged() { - // No canvas present: body passes through unmodified. - let composed = prepend_canvas_for_legacy(1, None, "do the thing"); + fn test_initial_message_modern_agent_omits_standing_block() { + // Protocol-v2 agents hold all of this from session/new; repeating it in + // the initial-message user turn would double-render every section. + let composed = prepend_standing_for_legacy(2, &full_standing(), "do the thing"); assert_eq!(composed, "do the thing"); } #[test] - fn test_initial_message_legacy_canvas_and_base_compose_correctly() { - // Verify the full composition order when both base and canvas are present: - // [Base] → canvas section → initial-message body. - let canvas = "[Channel Canvas]\ncanvas content"; - let base_composed = prepend_base_for_legacy(1, Some("be helpful"), "do the thing"); - let full = prepend_canvas_for_legacy(1, Some(canvas), &base_composed); - assert!( - full.starts_with("[Channel Canvas]"), - "canvas must be first in composed message" - ); - assert!( - full.contains("[Base]"), - "base must be present in composed message" - ); - assert!( - full.ends_with("do the thing"), - "body must be last in composed message" - ); - // Order: canvas → base → body - let canvas_pos = full.find("[Channel Canvas]").unwrap(); - let base_pos = full.find("[Base]").unwrap(); - let body_pos = full.find("do the thing").unwrap(); - assert!( - canvas_pos < base_pos && base_pos < body_pos, - "order must be: canvas → base → body" - ); + fn test_initial_message_legacy_agent_without_standing_is_unchanged() { + // Nothing configured: body passes through with no stray blank lines. + let composed = + prepend_standing_for_legacy(1, &crate::queue::StandingContext::default(), "do it"); + assert_eq!(composed, "do it"); } // Pin the session/new systemPrompt framing: each present prompt carries its @@ -5154,6 +5403,7 @@ mod tests { }; let context = ConversationContext::Thread { messages: vec![ContextMessage { + event_id: String::new(), pubkey: "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb".into(), timestamp: "2026-03-25T05:51:25Z".into(), content: "follow up".into(), @@ -5227,6 +5477,635 @@ mod tests { assert!(parse_kind0_profile_lookup(json!({})).is_none()); } + fn context_message(event_id: &str, content: &str) -> ContextMessage { + ContextMessage { + event_id: event_id.to_string(), + pubkey: "author".into(), + timestamp: "2026-08-09T00:00:00Z".into(), + content: content.into(), + } + } + + #[tokio::test] + async fn run_prompt_task_commits_standing_context_only_after_acp_success() { + let capture = std::env::temp_dir().join(format!( + "buzz-acp-standing-lifecycle-{}.ndjson", + Uuid::new_v4() + )); + let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''"); + let script = format!( + r#"count=0 +while IFS= read -r line; do + printf '%s\n' "$line" >> '{quoted_capture}' + count=$((count + 1)) + if [ "$count" -eq 1 ]; then + printf '%s\n' '{{"jsonrpc":"2.0","id":0,"error":{{"code":-32000,"message":"retry me"}}}}' + else + printf '%s\n' "{{\"jsonrpc\":\"2.0\",\"id\":$((count - 1)),\"result\":{{\"stopReason\":\"end_turn\"}}}}" + fi +done"# + ); + let acp = AcpClient::spawn("bash", &["-c".to_string(), script], &[], false) + .await + .expect("spawn lifecycle ACP script"); + let mut agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "legacy-test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + agent.state.heartbeat_session = Some("live-session".into()); + + let mut ctx = make_prompt_context_no_owner(); + ctx.base_prompt = Some("standing-once"); + let ctx = Arc::new(ctx); + let (result_tx, mut result_rx) = mpsc::unbounded_channel(); + + for turn in 1..=3 { + run_prompt_task( + agent, + None, + Some(format!("heartbeat-{turn}")), + Arc::clone(&ctx), + result_tx.clone(), + None, + format!("turn-{turn}"), + ) + .await; + let result = result_rx.recv().await.expect("prompt result"); + match turn { + 1 => assert!(matches!(result.outcome, PromptOutcome::Error(_))), + _ => assert!(matches!( + result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )), + } + assert_eq!( + result.agent.state.heartbeat_standing_context_sent, + turn >= 2, + "failed first delivery must not commit; first success must commit" + ); + agent = result.agent; + } + agent.acp.shutdown().await; + + let requests: Vec = std::fs::read_to_string(&capture) + .expect("read captured ACP requests") + .lines() + .map(|line| serde_json::from_str(line).expect("captured request is JSON")) + .collect(); + std::fs::remove_file(&capture).expect("remove ACP capture"); + assert_eq!(requests.len(), 3); + let prompt_text = |index: usize| { + requests[index]["params"]["prompt"][0]["text"] + .as_str() + .expect("text prompt") + }; + assert_eq!(prompt_text(0), "[Base]\nstanding-once\n\nheartbeat-1"); + assert_eq!( + prompt_text(1), + "[Base]\nstanding-once\n\nheartbeat-2", + "retry after ACP failure must resend standing context" + ); + assert_eq!( + prompt_text(2), + "heartbeat-3", + "turn after ACP success must omit standing context" + ); + } + + #[tokio::test] + async fn channel_prompt_commits_delivery_state_only_after_acp_success() { + let capture = std::env::temp_dir().join(format!( + "buzz-acp-channel-delivery-lifecycle-{}.ndjson", + Uuid::new_v4() + )); + let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''"); + let script = format!( + r#"count=0 +while IFS= read -r line; do + printf '%s\n' "$line" >> '{quoted_capture}' + count=$((count + 1)) + if [ "$count" -eq 1 ]; then + printf '%s\n' '{{"jsonrpc":"2.0","id":0,"error":{{"code":-32000,"message":"retry me"}}}}' + else + printf '%s\n' "{{\"jsonrpc\":\"2.0\",\"id\":$((count - 1)),\"result\":{{\"stopReason\":\"end_turn\"}}}}" + fi +done"# + ); + let acp = AcpClient::spawn("bash", &["-c".to_string(), script], &[], false) + .await + .expect("spawn channel lifecycle ACP script"); + let channel_id = Uuid::new_v4(); + let mut agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "legacy-test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + agent + .state + .sessions + .insert(channel_id, "live-session".into()); + agent + .state + .deliveries + .insert(channel_id, ChannelDeliveryState::default()); + + let mut ctx = make_prompt_context_no_owner(); + ctx.base_prompt = Some("standing-once"); + let ctx = Arc::new(ctx); + let (result_tx, mut result_rx) = mpsc::unbounded_channel(); + + for turn in 1..=3 { + let event = EventBuilder::new(Kind::Custom(9), format!("channel-{turn}")) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let event_id = event.id.to_hex(); + let batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + run_prompt_task( + agent, + Some(batch), + None, + Arc::clone(&ctx), + result_tx.clone(), + None, + format!("turn-{turn}"), + ) + .await; + let result = result_rx.recv().await.expect("prompt result"); + match turn { + 1 => assert!(matches!(result.outcome, PromptOutcome::Error(_))), + _ => assert!(matches!( + result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )), + } + let delivery = &result.agent.state.deliveries[&channel_id]; + assert_eq!( + delivery.standing_context_sent, + turn >= 2, + "failed channel delivery must not commit; first success must commit" + ); + assert_eq!( + delivery.delivered_event_ids.contains(&event_id), + turn >= 2, + "channel event IDs must commit only after ACP success" + ); + agent = result.agent; + } + agent.acp.shutdown().await; + + let requests: Vec = std::fs::read_to_string(&capture) + .expect("read captured ACP requests") + .lines() + .map(|line| serde_json::from_str(line).expect("captured request is JSON")) + .collect(); + std::fs::remove_file(&capture).expect("remove ACP capture"); + let prompt_text = |index: usize| { + requests[index]["params"]["prompt"][0]["text"] + .as_str() + .expect("text prompt") + }; + assert!(prompt_text(0).contains("[Base]\nstanding-once")); + assert!( + prompt_text(1).contains("[Base]\nstanding-once"), + "retry after channel ACP failure must resend standing context" + ); + assert!( + !prompt_text(2).contains("[Base]\nstanding-once"), + "turn after channel ACP success must omit standing context" + ); + } + + #[tokio::test] + async fn merged_cancel_prompt_commits_and_deduplicates_all_rendered_event_ids() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let channel_id = Uuid::new_v4(); + let keys = Keys::generate(); + let carry_over = EventBuilder::new(Kind::Custom(9), "merged carry-over sentinel") + .sign_with_keys(&keys) + .unwrap(); + let carry_over_id = carry_over.id.to_hex(); + let new_event = EventBuilder::new(Kind::Custom(9), "merged new-event sentinel") + .sign_with_keys(&keys) + .unwrap(); + let new_event_id = new_event.id.to_hex(); + let next_event = EventBuilder::new(Kind::Custom(9), "ordinary next-turn sentinel") + .sign_with_keys(&keys) + .unwrap(); + let merged_batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event: new_event.clone(), + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![crate::queue::BatchEvent { + event: carry_over.clone(), + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancel_reason: Some(crate::queue::CancelReason::Steer), + }; + let next_batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event: next_event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + + // Return both merged events as DM history. They must be excluded from + // the merged prompt's context and, after success, from the next turn. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind context server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let response_body = serde_json::to_string(&vec![carry_over, new_event]).unwrap(); + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut request = vec![0; 16 * 1024]; + let _ = socket.read(&mut request).await; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), response_body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let capture = std::env::temp_dir().join(format!( + "buzz-acp-merged-delivery-wire-{}.ndjson", + Uuid::new_v4() + )); + let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''"); + let script = format!( + r#"count=0 +while IFS= read -r line; do + printf '%s\n' "$line" >> '{quoted_capture}' + printf '%s\n' "{{\"jsonrpc\":\"2.0\",\"id\":$count,\"result\":{{\"stopReason\":\"end_turn\"}}}}" + count=$((count + 1)) +done"# + ); + let acp = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .expect("spawn wire-capture ACP"); + let mut agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "legacy-test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + agent + .state + .sessions + .insert(channel_id, "live-session".into()); + agent + .state + .deliveries + .insert(channel_id, ChannelDeliveryState::default()); + + let mut ctx = make_prompt_context_no_owner(); + ctx.context_message_limit = 10; + ctx.rest_client.base_url = base_url.clone(); + ctx.channel_info = ChannelInfoResolver::new( + HashMap::from([( + channel_id, + crate::relay::ChannelInfo { + name: "test-dm".into(), + channel_type: "dm".into(), + }, + )]), + RestClient { + http: reqwest::Client::new(), + base_url, + keys: ctx.agent_keys.clone(), + auth_tag_json: None, + }, + ); + let ctx = Arc::new(ctx); + let (result_tx, mut result_rx) = mpsc::unbounded_channel(); + + for (turn_id, batch) in [("merged-turn", merged_batch), ("next-turn", next_batch)] { + run_prompt_task( + agent, + Some(batch), + None, + Arc::clone(&ctx), + result_tx.clone(), + None, + turn_id.into(), + ) + .await; + let result = result_rx.recv().await.expect("prompt result"); + assert!(matches!( + result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )); + agent = result.agent; + } + let delivery = &agent.state.deliveries[&channel_id]; + assert!(delivery.delivered_event_ids.contains(&carry_over_id)); + assert!(delivery.delivered_event_ids.contains(&new_event_id)); + agent.acp.shutdown().await; + server.abort(); + + let requests: Vec = std::fs::read_to_string(&capture) + .expect("read captured prompts") + .lines() + .map(|line| serde_json::from_str(line).expect("captured prompt JSON")) + .collect(); + std::fs::remove_file(&capture).expect("remove prompt capture"); + assert_eq!(requests.len(), 2); + let wire = |index: usize| { + requests[index]["params"]["prompt"] + .as_array() + .expect("prompt blocks") + .iter() + .filter_map(|block| block["text"].as_str()) + .collect::>() + .join("\n") + }; + let merged_wire = wire(0); + assert_eq!(merged_wire.matches("merged carry-over sentinel").count(), 1); + assert_eq!(merged_wire.matches("merged new-event sentinel").count(), 1); + let next_wire = wire(1); + assert!(next_wire.contains("ordinary next-turn sentinel")); + assert!(!next_wire.contains("merged carry-over sentinel")); + assert!(!next_wire.contains("merged new-event sentinel")); + assert!(!next_wire.contains(&carry_over_id)); + assert!(!next_wire.contains(&new_event_id)); + } + + #[tokio::test] + async fn late_successful_steer_ack_excludes_event_from_next_channel_wire_prompt() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let channel_id = Uuid::new_v4(); + let keys = Keys::generate(); + let steered_event = EventBuilder::new(Kind::Custom(9), "steered context must not replay") + .sign_with_keys(&keys) + .unwrap(); + let steered_event_id = steered_event.id.to_hex(); + let trigger = EventBuilder::new(Kind::Custom(9), "ordinary next turn") + .sign_with_keys(&keys) + .unwrap(); + let batch = FlushBatch { + channel_id, + events: vec![crate::queue::BatchEvent { + event: trigger, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + + // The local REST bridge returns the already-delivered steer as DM + // history. Profile/reaction requests may also arrive; the same valid + // event array is harmless for those best-effort paths. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind context server"); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let response_body = serde_json::to_string(&vec![steered_event]).unwrap(); + let server = tokio::spawn(async move { + while let Ok((mut socket, _)) = listener.accept().await { + let mut request = vec![0; 16 * 1024]; + let _ = socket.read(&mut request).await; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response_body.len(), response_body + ); + let _ = socket.write_all(response.as_bytes()).await; + } + }); + + let capture = std::env::temp_dir().join(format!( + "buzz-acp-late-steer-wire-{}.ndjson", + Uuid::new_v4() + )); + let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''"); + let script = format!( + r#"IFS= read -r line +printf '%s\n' "$line" > '{quoted_capture}' +printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"# + ); + let acp = AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .expect("spawn wire-capture ACP"); + let mut agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + agent_name: "legacy-test-agent".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + agent + .state + .sessions + .insert(channel_id, "live-session".into()); + agent + .state + .deliveries + .insert(channel_id, ChannelDeliveryState::default()); + + // Model the adversarial ordering: the task result has already retired + // its TaskMeta and returned the agent before the successful ack arrives. + let mut pool = AgentPool::from_slots(vec![Some(agent)]); + assert!(pool.record_successful_steer( + channel_id, + steered_event_id.clone(), + "live-session".into(), + )); + let agent = pool + .try_claim(Some(channel_id)) + .expect("claim returned agent"); + + let mut ctx = make_prompt_context_no_owner(); + ctx.context_message_limit = 10; + ctx.rest_client.base_url = base_url.clone(); + ctx.channel_info = ChannelInfoResolver::new( + HashMap::from([( + channel_id, + crate::relay::ChannelInfo { + name: "test-dm".into(), + channel_type: "dm".into(), + }, + )]), + RestClient { + http: reqwest::Client::new(), + base_url, + keys: ctx.agent_keys.clone(), + auth_tag_json: None, + }, + ); + let (result_tx, mut result_rx) = mpsc::unbounded_channel(); + run_prompt_task( + agent, + Some(batch), + None, + Arc::new(ctx), + result_tx, + None, + "next-turn".into(), + ) + .await; + let mut result = result_rx.recv().await.expect("next prompt result"); + assert!(matches!( + result.outcome, + PromptOutcome::Ok(StopReason::EndTurn) + )); + result.agent.acp.shutdown().await; + server.abort(); + + let request: serde_json::Value = + serde_json::from_str(&std::fs::read_to_string(&capture).expect("read captured prompt")) + .expect("captured prompt JSON"); + std::fs::remove_file(&capture).expect("remove prompt capture"); + let wire = request["params"]["prompt"] + .as_array() + .expect("prompt blocks") + .iter() + .filter_map(|block| block["text"].as_str()) + .collect::>() + .join("\n"); + assert!(wire.contains("ordinary next turn")); + assert!(!wire.contains("steered context must not replay")); + assert!(!wire.contains(&steered_event_id)); + } + + #[test] + fn delivery_state_commits_only_when_explicitly_marked_successful() { + let channel = Uuid::new_v4(); + let mut state = SessionState::default(); + state + .deliveries + .insert(channel, ChannelDeliveryState::default()); + + // Building or attempting a prompt does not mutate delivery state. + let delivery = state.deliveries.get(&channel).unwrap(); + assert!(!delivery.standing_context_sent); + assert!(delivery.delivered_event_ids.is_empty()); + + state.mark_channel_delivery_success( + channel, + true, + ["trigger".to_string(), "context".to_string()], + ); + let delivery = state.deliveries.get(&channel).unwrap(); + assert!(delivery.standing_context_sent); + assert_eq!(delivery.delivered_event_ids.len(), 2); + } + + #[test] + fn delivery_state_is_cleared_on_rotation_and_restarts_empty() { + let channel = Uuid::new_v4(); + let mut state = SessionState::default(); + state.sessions.insert(channel, "old-session".into()); + state.mark_channel_delivery_success(channel, true, ["old-event".to_string()]); + + assert!(state.invalidate_channel(&channel)); + assert!(!state.deliveries.contains_key(&channel)); + + state.sessions.insert(channel, "new-session".into()); + state + .deliveries + .insert(channel, ChannelDeliveryState::default()); + let delivery = state.deliveries.get(&channel).unwrap(); + assert!(!delivery.standing_context_sent); + assert!(delivery.delivered_event_ids.is_empty()); + } + + #[test] + fn conversation_context_delta_omits_delivered_and_triggering_events() { + let delivered = HashSet::from(["old".to_string()]); + let triggering = HashSet::from(["trigger".to_string()]); + let context = ConversationContext::Thread { + messages: vec![ + context_message("old", "already sent"), + context_message("trigger", "rendered as trigger"), + context_message("new", "new context"), + ], + total: 3, + truncated: false, + }; + + let delta = conversation_context_delta(Some(context), &delivered, &triggering) + .expect("new context remains"); + match delta { + ConversationContext::Thread { + messages, + total, + truncated, + } => { + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].event_id, "new"); + assert_eq!(total, 3); + assert!(!truncated); + } + _ => panic!("expected thread context"), + } + } + + #[test] + fn conversation_context_delta_returns_none_when_no_new_events_remain() { + let delivered = HashSet::from(["old".to_string()]); + let context = ConversationContext::Dm { + messages: vec![context_message("old", "already sent")], + total: 1, + truncated: false, + }; + + assert!(conversation_context_delta(Some(context), &delivered, &HashSet::new()).is_none()); + } + + #[test] + fn conversation_context_delta_preserves_unidentified_legacy_messages() { + let context = ConversationContext::Dm { + messages: vec![context_message("", "cannot safely deduplicate")], + total: 1, + truncated: false, + }; + + assert!( + conversation_context_delta(Some(context), &HashSet::new(), &HashSet::new()).is_some() + ); + } + #[test] fn test_json_to_context_message_missing_pubkey_uses_default() { let obj = json!({ "content": "hello" }); @@ -5279,8 +6158,23 @@ mod tests { s.turn_counts.insert(ch_b, 3); s.core_sections.insert(ch_a, "core-a".into()); s.core_sections.insert(ch_b, "core-b".into()); + s.deliveries.insert( + ch_a, + ChannelDeliveryState { + standing_context_sent: true, + delivered_event_ids: HashSet::from(["event-a".into()]), + }, + ); + s.deliveries.insert( + ch_b, + ChannelDeliveryState { + standing_context_sent: true, + delivered_event_ids: HashSet::from(["event-b".into()]), + }, + ); s.heartbeat_session = Some("sess-hb".into()); s.heartbeat_turn_count = 7; + s.heartbeat_standing_context_sent = true; (s, ch_a, ch_b) } @@ -5346,6 +6240,7 @@ mod tests { assert!(s.heartbeat_session.is_none()); assert_eq!(s.heartbeat_turn_count, 0); + assert!(!s.heartbeat_standing_context_sent); // channels untouched assert_eq!(s.sessions.len(), 2); assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5); @@ -5364,6 +6259,7 @@ mod tests { assert!(s.core_sections.is_empty()); assert!(s.heartbeat_session.is_none()); assert_eq!(s.heartbeat_turn_count, 0); + assert!(!s.heartbeat_standing_context_sent); } #[test] diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 5c960de202..3bf1962242 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -990,6 +990,9 @@ pub enum ConversationContext { /// A single message in a conversation context section. #[derive(Debug, Clone)] pub struct ContextMessage { + /// Nostr event ID. Legacy REST fixtures may omit it, in which case it is + /// empty and cannot participate in delivery deduplication. + pub event_id: String, pub pubkey: String, pub timestamp: String, pub content: String, @@ -1241,6 +1244,7 @@ fn format_context_hints( thread_tags: &ThreadTags, is_dm: bool, has_conversation_context: bool, + conversation_context_had_delivered_events: bool, reply_anchor: Option<&str>, ) -> String { let channel_display = match channel_info { @@ -1258,6 +1262,10 @@ fn format_context_hints( "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." } else if has_conversation_context { "Conversation context included below. Use `buzz messages get --channel ` for full history if truncated." + } else if conversation_context_had_delivered_events && is_reply { + "Earlier thread context was already delivered in this session. Use `buzz messages thread --channel --event ` to re-read the reply chain." + } else if conversation_context_had_delivered_events { + "Earlier conversation context was already delivered in this session. Use `buzz messages get --channel ` to re-read it." } else if is_reply { "Use `buzz messages thread --channel --event ` to fetch the reply chain." } else { @@ -1285,6 +1293,8 @@ fn format_context_hints( } else if let Some(ref root) = thread_tags.root_event_id { let ctx_hint = if has_conversation_context { "Thread context included below. Use `buzz messages thread --channel --event ` for full history if truncated." + } else if conversation_context_had_delivered_events { + "Earlier thread context was already delivered in this session. Use `buzz messages thread --channel --event ` to re-read it." } else { "Use `buzz messages thread --channel --event ` to fetch thread context." }; @@ -1359,6 +1369,9 @@ pub struct FormatPromptArgs<'a> { pub agent_core: Option<&'a str>, pub channel_info: Option<&'a PromptChannelInfo>, pub conversation_context: Option<&'a ConversationContext>, + /// True when delivery-delta filtering removed at least one event that this + /// live session had already received. Trigger-only context does not set it. + pub conversation_context_had_delivered_events: bool, pub profile_lookup: Option<&'a PromptProfileLookup>, /// When true, base_prompt and system_prompt are delivered via the system /// role (session/new) and omitted from the user message. When false @@ -1374,9 +1387,62 @@ pub struct FormatPromptArgs<'a> { /// /// For modern agents (protocol_version >= 2) the section is delivered via /// the system role in session/new; omit here to avoid duplication. - /// For legacy agents it rides in the user message on every turn of the - /// session, alongside `[Base]`/`[System]`/`[Agent Memory — core]`. pub agent_canvas: Option<&'a str>, + /// Set once this session's standing context has already been delivered — + /// see [`StandingContext`]. Only meaningful for legacy agents; modern + /// agents are gated by `has_system_prompt_support` regardless. + /// + /// Defaults to `false` so a caller that never sets it behaves as if this + /// were the session's first message. + pub standing_context_sent: bool, +} + +/// The prompt sections that do not change for the life of a session: base +/// prompt, persona, team instructions, core memory, and channel canvas. +/// +/// Protocol-v2 agents receive all of this through the system role at +/// `session/new`, once. Legacy agents (`protocol_version < 2`) have no system +/// role, so it has to ride in a user message — but only in the session's +/// *first* one. Re-sending it every turn makes the standing framing the newest +/// and most-repeated text in the window, outweighing the conversation it exists +/// to frame, and evicting real channel history that much sooner. +/// +/// Both legacy dispatch paths (initial message, batch flush) render through +/// this one type so their section set and ordering cannot drift apart. +#[derive(Default)] +pub(crate) struct StandingContext<'a> { + pub base_prompt: Option<&'a str>, + pub system_prompt: Option<&'a str>, + pub team_instructions: Option<&'a str>, + pub agent_core: Option<&'a str>, + pub agent_canvas: Option<&'a str>, +} + +impl StandingContext<'_> { + /// Render the sections in the order legacy agents have always seen them. + pub(crate) fn sections(&self) -> Vec { + let mut sections = Vec::with_capacity(5); + if let Some(bp) = self.base_prompt { + sections.push(base_section(bp)); + } + if let Some(sp) = self.system_prompt { + sections.push(format!("[System]\n{sp}")); + } + if let Some(team) = self + .team_instructions + .map(str::trim) + .filter(|value| !value.is_empty()) + { + sections.push(format!("[Team Instructions]\n{team}")); + } + if let Some(core) = self.agent_core { + sections.push(core.to_string()); + } + if let Some(canvas) = self.agent_canvas { + sections.push(canvas.to_string()); + } + sections + } } /// Format the `[Base]` section for the base prompt. @@ -1391,12 +1457,12 @@ pub(crate) fn base_section(base_prompt: &str) -> String { /// Format a [`FlushBatch`] into the per-section prompt blocks for the agent. /// /// Produces a stable prompt with these sections (in order): -/// 0. `[Base]` — base prompt (only for legacy agents without systemPrompt support) -/// 1. `[System]` — system prompt (only for legacy agents without systemPrompt support) -/// 2. `[Agent Memory — core]` — if agent core memory is set -/// 3. `[Context]` — scope, channel name, and contextual hints for the agent -/// 4. `[Thread Context]` or `[Conversation Context]` — if fetched -/// 5. `[Event]` / `[Buzz events]` — the triggering event(s) +/// 0. [`StandingContext`] — `[Base]`, `[System]`, `[Team Instructions]`, +/// `[Agent Memory — core]`, `[Channel Canvas]`. Legacy agents only, and only +/// on the session's first message (see `standing_context_sent`) +/// 1. `[Context]` — scope, channel name, and contextual hints for the agent +/// 2. `[Thread Context]` or `[Conversation Context]` — if fetched +/// 3. `[Event]` / `[Buzz events]` — the triggering event(s) /// /// Each section is returned as its own block rather than one joined string so /// the observer frame's size trimmer (`fit_observer_event_to_budget`) elides @@ -1428,38 +1494,22 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec = Vec::with_capacity(7); - // For legacy agents (protocol_version < 2), inject base_prompt and - // system_prompt as user-message sections. Modern agents receive these - // via the system role in session/new. - if !args.has_system_prompt_support { - if let Some(bp) = args.base_prompt { - sections.push(base_section(bp)); - } - if let Some(sp) = args.system_prompt { - sections.push(format!("[System]\n{sp}")); - } - if let Some(team) = args - .team_instructions - .map(str::trim) - .filter(|value| !value.is_empty()) - { - sections.push(format!("[Team Instructions]\n{team}")); - } - } - - // NIP-AE agent core memory (rendered by `engram_fetch::build_core_section`). - // For modern agents (protocol_version >= 2), core is delivered via the - // system role in session/new, so it is omitted here to avoid duplication. - // Legacy agents have no system role, so core rides in the user message - // alongside `[Base]`/`[System]`. - if !args.has_system_prompt_support { - if let Some(core) = args.agent_core { - sections.push(core.to_string()); - } - // Channel canvas metadata — same delivery semantics as core for legacy agents. - if let Some(canvas) = args.agent_canvas { - sections.push(canvas.to_string()); - } + // Standing context — base prompt, persona, team instructions, core memory + // and canvas. Modern agents received all of it via the system role in + // session/new. Legacy agents get it here, in the session's first message + // only; `standing_context_sent` means an earlier message in this session + // already carried it. + if !args.has_system_prompt_support && !args.standing_context_sent { + sections.extend( + StandingContext { + base_prompt: args.base_prompt, + system_prompt: args.system_prompt, + team_instructions: args.team_instructions, + agent_core: args.agent_core, + agent_canvas: args.agent_canvas, + } + .sections(), + ); } // 2. Context hints (with a human-aware reply anchor). @@ -1489,6 +1539,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec