diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index d2161a45e5..dd7bb3a27e 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -4883,9 +4883,9 @@ fn dispatch_pending( .state .set_scope_owner_generation(scope.clone(), owner_generation); - // The prompt task records how it classified the channel, for the + // The prompt task records how it routed replies, for the // native-steer guard (`EventQueue::in_flight_accepts_steer`). - let prompt_dm = queue.in_flight_prompt_dm(&scope).unwrap_or_default(); + let prompt_routing = queue.in_flight_prompt_routing(&scope).unwrap_or_default(); let abort_handle = pool.join_set.spawn(async move { pool::run_prompt_task( agent, @@ -4895,7 +4895,7 @@ fn dispatch_pending( result_tx, Some(control_rx), task_turn_id, - prompt_dm, + prompt_routing, ) .await; }); @@ -10055,8 +10055,8 @@ mod edit_native_steer_tests { /// Drive `incoming` through the listener's steer decision while a turn /// for `running_event` is in flight in the same conversation session /// (the channel policy, or any DM). `prompt_is_dm` is how the running - /// turn's prompt classified the channel; `None` means the prompt has not - /// been formatted yet. + /// turn's prompt classified the channel, recorded with a thread-shared + /// reply anchor; `None` means the prompt has not been formatted yet. fn steer_into_running_turn( is_dm: bool, prompt_is_dm: Option, @@ -10084,10 +10084,11 @@ mod edit_native_steer_tests { queue.flush_next().expect("running turn"); assert!(queue.is_scope_in_flight(&scope)); if let Some(prompt_is_dm) = prompt_is_dm { - queue - .in_flight_prompt_dm(&scope) - .expect("in-flight turn records its prompt classification") - .record(prompt_is_dm); + let routing = queue + .in_flight_prompt_routing(&scope) + .expect("in-flight turn records its prompt routing"); + routing.record_dm(prompt_is_dm); + routing.record_trigger_anchor(false); } let mut pool = AgentPool::from_slots(vec![]); @@ -10224,6 +10225,7 @@ mod edit_native_steer_tests { hold_at: HoldAt, running_event: nostr::Event, incoming: nostr::Event, + profiles: &[nostr::Event], ) -> (Option, Option) { let channel_id = Uuid::new_v4(); let is_dm = channel_type == "dm"; @@ -10277,23 +10279,32 @@ mod edit_native_steer_tests { .insert(scope.clone(), "live-session".into()); } // The relay has no newer metadata, project, or history: every query - // returns no events, so the turn keeps the startup channel metadata. + // except a profile lookup returns no events, so the turn keeps the + // startup channel metadata. let relay = tokio::net::TcpListener::bind("127.0.0.1:0") .await .expect("bind relay fixture"); let base_url = format!("http://{}", relay.local_addr().unwrap()); + let profiles = Arc::new(serde_json::to_string(profiles).expect("profiles json")); let relay = tokio::spawn(async move { use tokio::io::{AsyncReadExt, AsyncWriteExt}; loop { let (mut socket, _) = relay.accept().await.expect("accept relay query"); + let profiles = Arc::clone(&profiles); tokio::spawn(async move { let mut request = [0; 16384]; - let _ = socket.read(&mut request).await; - let _ = socket - .write_all( - b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n[]", - ) - .await; + let read = socket.read(&mut request).await.unwrap_or(0); + let body = + if String::from_utf8_lossy(&request[..read]).contains("\"kinds\":[0]") { + profiles.as_str() + } else { + "[]" + }; + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + let _ = socket.write_all(response.as_bytes()).await; }); } }); @@ -10372,9 +10383,15 @@ mod edit_native_steer_tests { /// top-level follow-up must be steered natively. #[tokio::test] async fn dispatched_dm_turn_records_classification_for_native_steer() { - let (steer, control) = - steer_into_dispatched_turn("dm", None, HoldAt::Prompt, message(None), message(None)) - .await; + let (steer, control) = steer_into_dispatched_turn( + "dm", + None, + HoldAt::Prompt, + message(None), + message(None), + &[], + ) + .await; assert!(steer.is_some(), "DM follow-up is sent as a native steer"); assert_eq!( @@ -10394,6 +10411,7 @@ mod edit_native_steer_tests { HoldAt::SessionNew, message(None), message(None), + &[], ) .await; @@ -10415,6 +10433,7 @@ mod edit_native_steer_tests { HoldAt::SessionNew, message(None), message(None), + &[], ) .await; @@ -10425,12 +10444,11 @@ mod edit_native_steer_tests { assert_eq!(control, Some(ControlSignal::Steer)); } - /// A known channel's prompt is never rendered as a DM, so a same-thread - /// follow-up is steered natively before the turn's classification is - /// recorded, as before DM steering existed: here the prompt task is - /// still waiting on `session/new` ahead of an `initial_message`. + /// Until a thread turn's prompt is formatted, its reply anchor is + /// unknown: an agent↔agent turn anchors to its own trigger. A same-thread + /// follow-up during `session/new` therefore takes the cancel+merge path. #[tokio::test] - async fn channel_same_thread_follow_up_during_setup_steers_natively() { + async fn channel_same_thread_follow_up_during_setup_cancels_and_merges() { let root = "ab".repeat(32); let (steer, control) = steer_into_dispatched_turn( "stream", @@ -10438,6 +10456,30 @@ mod edit_native_steer_tests { HoldAt::SessionNew, message(Some(&root)), message(Some(&root)), + &[], + ) + .await; + + assert!( + steer.is_none(), + "no native steer before the anchor is known" + ); + assert_eq!(control, Some(ControlSignal::Steer)); + } + + /// A human-facing thread turn anchors to the thread root, which every + /// same-thread message shares, so a follow-up is steered natively once + /// the prompt has recorded that anchor. + #[tokio::test] + async fn channel_same_thread_follow_up_during_prompt_steers_natively() { + let root = "ab".repeat(32); + let (steer, control) = steer_into_dispatched_turn( + "stream", + None, + HoldAt::Prompt, + message(Some(&root)), + message(Some(&root)), + &[], ) .await; @@ -10448,6 +10490,45 @@ mod edit_native_steer_tests { ); } + /// An agent↔agent thread turn replies under its own trigger. A human's + /// message in that thread must reply at the thread root, so it takes the + /// cancel+merge path, whose re-prompt carries its own ``. + #[tokio::test] + async fn human_follow_up_to_agent_thread_turn_cancels_and_merges() { + use nostr::{EventBuilder, Keys, Kind, Tag}; + let root = "ab".repeat(32); + let agent = Keys::generate(); + let other_agent = Keys::generate(); + let agent_profile = |keys: &Keys| { + EventBuilder::new(Kind::Metadata, "{}") + .tags([Tag::parse(["auth", &"cd".repeat(32), "", &"ef".repeat(64)]).unwrap()]) + .sign_with_keys(keys) + .unwrap() + }; + let running = EventBuilder::new( + Kind::Custom(buzz_core::kind::KIND_STREAM_MESSAGE as u16), + "agent asks", + ) + .tags([ + Tag::parse(["e", root.as_str(), "", "reply"]).unwrap(), + Tag::parse(["p", other_agent.public_key().to_hex().as_str()]).unwrap(), + ]) + .sign_with_keys(&agent) + .unwrap(); + let (steer, control) = steer_into_dispatched_turn( + "stream", + None, + HoldAt::Prompt, + running, + message(Some(&root)), + &[agent_profile(&agent), agent_profile(&other_agent)], + ) + .await; + + assert!(steer.is_none(), "no native steer into an agent-only turn"); + assert_eq!(control, Some(ControlSignal::Steer)); + } + /// A DM thread reply needs a `--reply-to` the running top-level turn's /// `` does not carry, so it takes the cancel+merge path. #[tokio::test] diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 992d894d59..7a7e3229e7 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -2420,7 +2420,7 @@ pub async fn run_prompt_task( result_tx: mpsc::UnboundedSender, control_rx: Option>, turn_id: String, - prompt_dm: crate::queue::PromptDmClassification, + prompt_routing: crate::queue::PromptRouting, ) { // Is this a channel prompt or a heartbeat? let source = match &batch { @@ -2543,7 +2543,7 @@ pub async fn run_prompt_task( let initial_message_pending = ctx.initial_message.is_some() && matches!(&source, PromptSource::Channel(scope) if !agent.state.sessions.contains_key(scope)); if !initial_message_pending { - prompt_dm.record(prompt_is_dm); + prompt_routing.record_dm(prompt_is_dm); } // @@ -3003,7 +3003,7 @@ pub async fn run_prompt_task( } } // Any `initial_message` setup turn is done; see `prompt_is_dm`. - prompt_dm.record(prompt_is_dm); + prompt_routing.record_dm(prompt_is_dm); // When the batch is a single slash-command message (e.g. "@Eva /goal …"), // `slash_command` holds the bare command. It is sent as the FIRST prompt @@ -3104,6 +3104,11 @@ pub async fn run_prompt_task( let profile_lookup = fetch_prompt_profile_lookup(b, conversation_context.as_ref(), &ctx.rest_client).await; + prompt_routing.record_trigger_anchor(crate::queue::reply_anchor_is_trigger( + b, + is_dm, + profile_lookup.as_ref(), + )); let known_names: Vec<&str> = profile_lookup .iter() diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 3f2ea34dbb..cc5ae876c1 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -201,35 +201,51 @@ impl ReplyRoute { } } -/// Whether the running turn's prompt rendered its channel as a DM. +/// How the running turn's prompt routes replies, as the steer guard needs to +/// know. /// -/// The prompt task records this when it formats the turn's `` -/// (`format_prompt`'s `is_dm`). The steer guard reads the recorded value -/// instead of classifying the incoming event again: channel metadata can -/// become resolvable after the turn starts, and the guard must apply the rule -/// that matches the `` the agent actually received. +/// The prompt task records each fact when it formats the turn's ``. +/// The steer guard reads the recorded values instead of deriving them from +/// the incoming event again: channel metadata can become resolvable after the +/// turn starts, and the guard must apply the rule that matches the +/// `` the agent actually received. #[derive(Debug, Clone, Default)] -pub struct PromptDmClassification(Arc>); +pub struct PromptRouting { + is_dm: Arc>, + trigger_anchor: Arc>, +} -impl PromptDmClassification { - /// Record the prompt's classification. Later calls are ignored: a turn has - /// one ``. - pub fn record(&self, is_dm: bool) { - let _ = self.0.set(is_dm); +impl PromptRouting { + /// Record whether the prompt rendered the channel as a DM + /// (`format_prompt`'s `is_dm`). Later calls are ignored: a turn has one + /// ``. + pub fn record_dm(&self, is_dm: bool) { + let _ = self.is_dm.set(is_dm); } - /// The recorded classification; `None` before the prompt is formatted. - pub fn get(&self) -> Option { - self.0.get().copied() + /// The recorded DM classification; `None` until it is recorded. + pub fn is_dm(&self) -> Option { + self.is_dm.get().copied() + } + + /// Record whether the prompt anchored replies to its triggering message + /// rather than to the thread (see [`reply_anchor_is_trigger`]). Later + /// calls are ignored. + pub fn record_trigger_anchor(&self, trigger_anchor: bool) { + let _ = self.trigger_anchor.set(trigger_anchor); + } + + /// The recorded anchor kind; `None` before the prompt is formatted. + pub fn trigger_anchor(&self) -> Option { + self.trigger_anchor.get().copied() } } -/// Reply destination of an in-flight turn, and how its prompt classified the -/// channel. +/// Reply destination of an in-flight turn, and how its prompt routed replies. #[derive(Debug, Clone)] struct InFlightReplyRoute { route: ReplyRoute, - prompt_is_dm: PromptDmClassification, + prompt: PromptRouting, } /// Why a batch's prior turn was cancelled — controls how `format_prompt` @@ -1028,12 +1044,18 @@ impl EventQueue { /// not be steered natively; the cancel+merge path re-dispatches it with /// its own full ``. /// - /// The rule follows how the running turn's prompt classified the channel - /// (see [`PromptDmClassification`]). Until the prompt task records that, - /// `channel_is_dm` is the listener's own classification, where unresolved - /// metadata counts as a DM. A channel the listener knows is not a DM also - /// renders as one in the prompt, so the channel rule applies at once. - /// Otherwise the prompt may render either way, and the answer is `false`. + /// The rule follows how the running turn's prompt routed replies (see + /// [`PromptRouting`]). Until the prompt task records its DM + /// classification, `channel_is_dm` is the listener's own classification, + /// where unresolved metadata counts as a DM. A channel the listener knows + /// is not a DM also renders as one in the prompt, so the channel rule + /// applies at once. Otherwise the prompt may render either way, and the + /// answer is `false`. + /// + /// A channel turn inside a thread may anchor replies to its own trigger + /// (see [`reply_anchor_is_trigger`]), a destination no other message + /// shares. Such a turn accepts no native steer, and neither does one whose + /// prompt has not yet recorded its anchor. pub fn in_flight_accepts_steer( &self, scope: &SessionScope, @@ -1043,19 +1065,26 @@ impl EventQueue { let Some(running) = self.in_flight_reply_routes.get(scope) else { return false; }; - match running.prompt_is_dm.get() { - Some(is_dm) => running.route.accepts_steer(incoming, is_dm), - None if !channel_is_dm => running.route.accepts_steer(incoming, false), - None => false, + let is_dm = match running.prompt.is_dm() { + Some(is_dm) => is_dm, + None if !channel_is_dm => false, + None => return false, + }; + if !is_dm + && running.route.root_event_id.is_some() + && running.prompt.trigger_anchor() != Some(false) + { + return false; } + running.route.accepts_steer(incoming, is_dm) } /// The cell in which the prompt task for `scope`'s in-flight turn records - /// its DM classification (see [`PromptDmClassification`]). - pub fn in_flight_prompt_dm(&self, scope: &SessionScope) -> Option { + /// how it routed replies (see [`PromptRouting`]). + pub fn in_flight_prompt_routing(&self, scope: &SessionScope) -> Option { self.in_flight_reply_routes .get(scope) - .map(|running| running.prompt_is_dm.clone()) + .map(|running| running.prompt.clone()) } fn record_in_flight_reply_route(&mut self, scope: &SessionScope, events: &[BatchEvent]) { @@ -1065,7 +1094,7 @@ impl EventQueue { scope.clone(), InFlightReplyRoute { route: last.reply_route(), - prompt_is_dm: PromptDmClassification::default(), + prompt: PromptRouting::default(), }, ); } @@ -1725,12 +1754,13 @@ fn turn_is_human_facing( /// Resolve the `--reply-to` anchor for a non-DM turn. /// -/// Returns `Some(id)` only for human-facing turns (see [`turn_is_human_facing`]): -/// - in a thread → the thread ROOT, keeping the reply flat at layer 1 +/// Human-facing turns (see [`turn_is_human_facing`]) stay flat: +/// - in a thread → the thread ROOT, keeping the reply at layer 1 /// - top-level → the triggering event id, which becomes the new thread root /// -/// Returns `None` for agent↔agent turns, leaving the agent free to nest deeply -/// (intentional for agent coordination). +/// Agent↔agent turns may nest (intentional for agent coordination): +/// - in a thread → the triggering event id, so the reply stays in the thread +/// - top-level → `None`, leaving the destination to the agent fn resolve_reply_anchor( sender_pubkey: &str, thread_tags: &ThreadTags, @@ -1738,7 +1768,10 @@ fn resolve_reply_anchor( profile_lookup: Option<&PromptProfileLookup>, ) -> Option { if !turn_is_human_facing(sender_pubkey, thread_tags, profile_lookup) { - return None; + return thread_tags + .root_event_id + .is_some() + .then(|| triggering_event_id.to_string()); } Some( thread_tags @@ -1748,6 +1781,33 @@ fn resolve_reply_anchor( ) } +/// Whether `format_prompt` anchors `batch`'s ordinary replies to its +/// triggering message instead of the thread every same-thread message +/// shares: an agent↔agent turn inside a channel thread (see +/// [`resolve_reply_anchor`]). The prompt task records this for the +/// native-steer guard (see [`EventQueue::in_flight_accepts_steer`]). +pub fn reply_anchor_is_trigger( + batch: &FlushBatch, + is_dm: bool, + profile_lookup: Option<&PromptProfileLookup>, +) -> bool { + let Some(last_event) = batch.events.last() else { + return false; + }; + let thread_tags = last_event.routing_thread_tags(); + let routing_event_id = last_event.routing_event_id(); + !is_dm + && thread_tags.root_event_id.is_some() + && resolve_reply_anchor( + &last_event.event.pubkey.to_hex(), + &thread_tags, + &routing_event_id, + profile_lookup, + ) + .as_deref() + == Some(routing_event_id.as_str()) +} + /// Maximum length (in characters) of a channel description rendered into ``. /// /// Limits prompt bloat from unusually long descriptions. Multi-line @@ -2327,9 +2387,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec