diff --git a/Justfile b/Justfile index fe5d7bf2858..96456d7953c 100644 --- a/Justfile +++ b/Justfile @@ -340,6 +340,21 @@ test-unit: # `cargo test --workspace`; without this step a manifest edit that # diverges Rust from the corpus ships green. cargo nextest run -p buzz-agent --lib + # buzz-acp harness: ACP prompt formatting, the event queue, and the pool + # lifecycle state machine. Pure in-process tests, no infra. Enumerated + # explicitly because nothing in CI runs `cargo test --workspace`; without + # this step the mid-turn steer prompt's reply-anchor invariants compile + # but never execute. + # + # The `BUZZ_*` scrub is load-bearing, not hygiene — see + # `run_buzz_acp_unit_tests` in scripts/run-tests.sh for why. Subshell so + # the steps above keep their environment. + ( + while IFS= read -r var; do + unset "$var" + done < <(env | sed -n 's/^\(BUZZ_[A-Za-z0-9_]*\)=.*/\1/p') + cargo nextest run -p buzz-acp + ) else ./scripts/run-tests.sh unit fi diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 146214197a8..b599c43298b 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -273,7 +273,8 @@ async fn author_allowed( } } -/// Resolve whether `channel_id` is a DM, for the inbound author gate. +/// Resolve whether `channel_id` is a DM for the inbound author gate — the +/// fail-closed projection of [`classify_dm`]. /// /// Resolution order: /// 1. Startup discovery metadata (`startup_info`) — covers channels known at @@ -290,14 +291,74 @@ pub(crate) async fn is_dm_channel( channel_id: Uuid, channel_info: &pool::ChannelInfoResolver, ) -> bool { + classify_dm(channel_id, channel_info).await.gates_as_dm() +} + +/// How one inbound event's channel resolved, from one metadata lookup. +/// +/// Closed rather than a pair of booleans: the authorization gate has an answer +/// for every state and native steering does not, so no pair of flags can say +/// "no answer". Callers take a named projection instead of choosing a field. +/// +/// Resolve once per event: `ChannelInfoResolver` does not cache the unresolved +/// case, so a second call pays a second lazy REST fetch on the main loop, for +/// exactly the channels that already failed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DmClassification { + Dm, + NonDm, + /// Channel metadata did not resolve for this event. + Unresolved, +} + +impl DmClassification { + /// Authorization reading — fails **closed**. An unresolved channel counts as + /// a DM so `respond_to` modes that admit non-owner authors cannot be + /// exercised inside a channel we failed to classify. + pub(crate) fn gates_as_dm(self) -> bool { + matches!(self, Self::Dm | Self::Unresolved) + } + + /// Native-steer reading — declines rather than guesses. `None` means the + /// native path must not run for this event. + pub(crate) fn native_steer_scope(self) -> Option { + match self { + Self::Dm => Some(NativeSteerScope::Dm), + Self::NonDm => Some(NativeSteerScope::NonDm), + Self::Unresolved => None, + } + } +} + +/// The DM reading native steering takes: resolved states only. +/// +/// There is no conservative default — see `queue::resolve_reply_anchor` for why +/// neither anchor rule is safe for a channel nobody could classify, and +/// `queue::format_native_steer_prompt` for the anchoring this path does apply. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum NativeSteerScope { + Dm, + NonDm, +} + +/// Resolve `channel_id`'s type once, for the two projections that need it. +/// +/// See [`DmClassification`] for why unresolved metadata still gates but carries +/// no native scope, and [`is_dm_channel`] for the resolution order. +pub(crate) async fn classify_dm( + channel_id: Uuid, + channel_info: &pool::ChannelInfoResolver, +) -> DmClassification { match channel_info.resolve(channel_id).await { - Some(info) => info.channel_type == "dm", + Some(info) if info.channel_type == "dm" => DmClassification::Dm, + Some(_) => DmClassification::NonDm, None => { tracing::warn!( channel_id = %channel_id, - "channel type unresolved — treating as DM for author gate (fail closed)" + "channel type unresolved — treating as DM for author gate (fail \ + closed); native steering declines, event falls back to cancel+merge" ); - true + DmClassification::Unresolved } } } @@ -2865,18 +2926,18 @@ async fn tokio_main() -> Result<()> { // launched by the same human). Allowlist adds the // explicit pubkey list on top, for external people; // it never revokes same-owner team bots. + // + // DM hardening: resolve the channel type once, here, + // for both the gate below and the native-steer fork. + // See `DmClassification` for the two readings. + let dm = classify_dm(buzz_event.channel_id, &ctx.channel_info).await; { let author = buzz_event.event.pubkey.to_hex(); - // DM hardening: resolve channel type (fail-closed - // to DM) so allowlist/anyone modes cannot be - // exercised by non-owner authors inside DMs. - let is_dm = - is_dm_channel(buzz_event.channel_id, &ctx.channel_info).await; let allowed = author_allowed( &config.respond_to, &config.respond_to_allowlist, &author, - is_dm, + dm.gates_as_dm(), &owner_cache, &ctx.rest_client, ) @@ -2886,7 +2947,7 @@ async fn tokio_main() -> Result<()> { channel_id = %buzz_event.channel_id, author = %buzz_event.event.pubkey.to_hex(), mode = %config.respond_to, - is_dm, + is_dm = dm.gates_as_dm(), "inbound author gate — dropping event" ); continue; @@ -2968,6 +3029,7 @@ async fn tokio_main() -> Result<()> { buzz_event.channel_id, event_for_steer, prompt_tag_for_steer, + dm, &steer_ack_tx, ); if !native_attempted { @@ -3608,6 +3670,22 @@ fn signal_in_flight_task( false } +/// Build a native steer's prompt blocks from an already-resolved scope. +/// +/// The decline on unresolved metadata happens in [`try_native_steer`], not here: +/// [`NativeSteerScope`] has no unresolved state left to reject. +fn native_steer_prompt_blocks( + channel_id: uuid::Uuid, + be: &queue::BatchEvent, + scope: NativeSteerScope, +) -> Vec { + vec![queue::format_native_steer_prompt( + channel_id, + be, + matches!(scope, NativeSteerScope::Dm), + )] +} + /// Attempt the non-cancelling (ACP) steer for a freshly-queued event. /// /// Caller invariants: @@ -3624,11 +3702,11 @@ fn signal_in_flight_task( /// universal cancel+merge `ControlSignal::Steer` fallback — the watcher /// will issue it from the ack arm if the native attempt fails. /// -/// Returns `false` if `pool.send_steer` failed (no in-flight task, -/// `steer_tx` already full from a prior in-flight steer, or read loop -/// torn down). The caller MUST fall through to -/// `signal_in_flight_task(channel_id, ControlSignal::Steer)` so the -/// event still reaches the agent via the universal path. +/// Returns `false` if the channel is [`DmClassification::Unresolved`], or if +/// `pool.send_steer` failed (no in-flight task, `steer_tx` already full from a +/// prior in-flight steer, or read loop torn down). The caller MUST fall through +/// to `signal_in_flight_task(channel_id, ControlSignal::Steer)` so the event +/// still reaches the agent via the universal path. /// /// The withheld event is NOT released here on `false` because no withhold /// was established: `mark_native_steer_pending` only runs on `Ok(())`. @@ -3638,34 +3716,32 @@ fn try_native_steer( channel_id: uuid::Uuid, event: nostr::Event, prompt_tag: String, + dm: DmClassification, steer_ack_tx: &mpsc::UnboundedSender, ) -> bool { - // Build the steer body: framing strings come from - // `queue::native_steer_framing()` (Eva's drift-proof requirement — - // native and cancel+merge fallback share these so the agent gets the - // same orientation regardless of transport). The single event block - // is rendered by `queue::format_event_block`, the same function - // `queue::format_prompt` uses internally for `[Buzz event: …]` - // sections, so the rendering also cannot drift. - // - // Passing `None` for `channel_info` / `profile_lookup` is intentional: - // native steer is a *delta* into a live turn — the agent already saw - // channel context and the actor's profile in the original prompt, - // duplicating it here would defeat the point of non-cancelling - // steering (which is to inject only what's new). - let (header, closing) = queue::native_steer_framing(); + // Decline before the prompt, the send and the withhold, so the event stays + // queued for the caller's cancel+merge fallback, which resolves the channel + // again at flush time. See `queue::resolve_reply_anchor` for why no anchor + // rule is safe without metadata. + let Some(scope) = dm.native_steer_scope() else { + tracing::debug!( + channel = %channel_id, + "channel metadata unresolved — declining native steer for cancel+merge" + ); + return false; + }; + let event_id_hex = event.id.to_hex(); let be = queue::BatchEvent { event, - prompt_tag: prompt_tag.clone(), + prompt_tag, received_at: std::time::Instant::now(), }; - let event_block = queue::format_event_block(channel_id, None, &be, None); - let body = format!("{header}\n\n[Buzz event: {prompt_tag}]\n{event_block}\n\n{closing}"); + let prompt_blocks = native_steer_prompt_blocks(channel_id, &be, scope); let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::(); let request = pool::SteerRequest { - prompt_blocks: vec![body], + prompt_blocks, ack_tx, }; @@ -5606,6 +5682,250 @@ mod author_gate_tests { ); } + /// The two readings part company only on unresolved metadata. A classifier + /// that failed closed on both would pass every `is_dm_channel` test above, + /// since those only ever observe the gate reading. + #[tokio::test] + async fn test_classify_dm_splits_only_when_metadata_is_unresolved() { + let dm_id = Uuid::new_v4(); + let stream_id = Uuid::new_v4(); + let startup = HashMap::from([ + ( + dm_id, + relay::ChannelInfo { + name: "dm".into(), + channel_type: "dm".into(), + description: None, + }, + ), + ( + stream_id, + relay::ChannelInfo { + name: "stream".into(), + channel_type: "stream".into(), + description: None, + }, + ), + ]); + let resolver = resolver(startup); + + let dm = classify_dm(dm_id, &resolver).await; + assert_eq!(dm, DmClassification::Dm); + assert!( + dm.gates_as_dm() && dm.native_steer_scope() == Some(NativeSteerScope::Dm), + "a resolved DM gates as a DM and carries the DM native scope" + ); + + let stream = classify_dm(stream_id, &resolver).await; + assert_eq!(stream, DmClassification::NonDm); + assert!( + !stream.gates_as_dm() && stream.native_steer_scope() == Some(NativeSteerScope::NonDm), + "a resolved stream gates open and steers with the channel scope" + ); + + let unresolved = classify_dm(Uuid::new_v4(), &resolver).await; + assert_eq!(unresolved, DmClassification::Unresolved); + assert!( + unresolved.gates_as_dm(), + "author gate fails closed: an unclassified channel must not admit non-owner authors" + ); + assert_eq!( + unresolved.native_steer_scope(), + None, + "native steering must decline an unclassified channel, not map it to a resolved scope" + ); + } + + /// The composite, through the production seam: `DmClassification` in at + /// `try_native_steer`, rendered scope and anchor out on the wire. No arm + /// supplies the intermediate [`NativeSteerScope`] by hand, so a regression in + /// that translation — where the cross-thread defect lived — fails here. + /// + /// The resolved arms are the control for the unresolved one: they steer on + /// this fixture, so its `false` is a decline and not a pool that could never + /// have steered. + #[tokio::test] + async fn test_native_steer_scope_and_anchor_through_the_classification_seam() { + /// `(accepted, prompt_sent_on_the_wire, event_still_flushable, triggering_id)` + async fn attempt( + dm: DmClassification, + tags: Vec, + ) -> (bool, Option, bool, String) { + let ch = Uuid::new_v4(); + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@bot steer") + .tags(tags) + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let triggering_id = event.id.to_hex(); + + let mut queue = EventQueue::new(config::DedupMode::Queue); + queue.push(QueuedEvent { + channel_id: ch, + event: event.clone(), + received_at: std::time::Instant::now(), + prompt_tag: "@mention".into(), + }); + + let mut pool = AgentPool::from_slots(vec![None]); + let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1); + 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(ch), + turn_id: "test-turn-id".into(), + recoverable_batch: None, + control_tx: None, + steer_tx: Some(steer_tx), + successful_steer_deliveries: Default::default(), + }, + ); + + let (ack_tx, _ack_rx) = mpsc::unbounded_channel(); + let accepted = try_native_steer( + &mut pool, + &mut queue, + ch, + event, + "@mention".into(), + dm, + &ack_tx, + ); + let sent = steer_rx + .try_recv() + .ok() + .map(|request| request.prompt_blocks.join("\n\n")); + (accepted, sent, queue.has_flushable_work(), triggering_id) + } + + let root = "d".repeat(64); + let threaded = || vec![nostr::Tag::parse(["e", &root, "", "reply"]).expect("e tag")]; + + // Resolved DM: anchors to the message being answered, not the DM root. + let (accepted, sent, flushable, triggering_id) = + attempt(DmClassification::Dm, threaded()).await; + assert!(accepted, "a resolved DM steers natively on this fixture"); + assert!(!flushable, "an accepted steer withholds the queued event"); + let sent = sent.expect("resolved DM puts a request on the wire"); + assert!(sent.contains("Scope: dm"), "{sent}"); + assert!( + sent.contains(&format!("--reply-to {triggering_id}")), + "a DM reply anchors to the triggering event: {sent}" + ); + + // Resolved non-DM, same thread: anchors to the root so replies stay flat. + let (accepted, sent, flushable, _) = attempt(DmClassification::NonDm, threaded()).await; + assert!( + accepted, + "a resolved non-DM steers natively on this fixture" + ); + assert!(!flushable, "an accepted steer withholds the queued event"); + let sent = sent.expect("resolved non-DM puts a request on the wire"); + assert!( + sent.contains("Scope: thread") && !sent.contains("Scope: dm"), + "{sent}" + ); + assert!( + sent.contains(&format!("--reply-to {root}")), + "a threaded non-DM reply anchors to the thread root: {sent}" + ); + + // Unresolved: nothing goes out and the event stays queued for + // cancel+merge, which resolves again at flush time. + let (accepted, sent, flushable, _) = + attempt(DmClassification::Unresolved, threaded()).await; + assert!( + !accepted, + "unresolved metadata must decline the native steer" + ); + assert!( + sent.is_none(), + "the decline must land before the steer is sent: {sent:?}" + ); + assert!( + flushable, + "a declined event must stay queued for the cancel+merge fallback" + ); + } + + /// Both resolved states carry a native scope and they differ; `Unresolved` + /// carries none. Fails if `Unresolved` is ever mapped onto either. + #[test] + fn test_native_steer_scope_is_resolved_only() { + assert_eq!( + DmClassification::Dm.native_steer_scope(), + Some(NativeSteerScope::Dm) + ); + assert_eq!( + DmClassification::NonDm.native_steer_scope(), + Some(NativeSteerScope::NonDm) + ); + assert_eq!(DmClassification::Unresolved.native_steer_scope(), None); + } + + /// Scope and anchor inside the formatter, given a scope. Supplies + /// [`NativeSteerScope`] by hand, so it cannot see the classification that + /// produced it — the seam test above carries that half. + #[test] + fn test_resolved_scopes_render_their_own_scope_and_anchor() { + let ch = Uuid::new_v4(); + let root = "d".repeat(64); + let threaded = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@bot steer") + .tags([nostr::Tag::parse(["e", &root, "", "reply"]).expect("e tag")]) + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let triggering_id = threaded.id.to_hex(); + let be = queue::BatchEvent { + event: threaded, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }; + + let dm = native_steer_prompt_blocks(ch, &be, NativeSteerScope::Dm).join("\n\n"); + assert!( + dm.contains("Scope: dm"), + "resolved DM renders DM scope: {dm}" + ); + assert!( + dm.contains(&format!("--reply-to {triggering_id}")), + "a DM reply anchors to the message being answered: {dm}" + ); + + // Same event, non-DM: a thread stays flat, anchored to its root, where + // the DM arm above anchored to the message being answered. + let threaded_channel = + native_steer_prompt_blocks(ch, &be, NativeSteerScope::NonDm).join("\n\n"); + assert!( + threaded_channel.contains("Scope: thread") && !threaded_channel.contains("Scope: dm"), + "resolved non-DM in a thread renders thread scope: {threaded_channel}" + ); + assert!( + threaded_channel.contains(&format!("--reply-to {root}")), + "a threaded channel reply anchors to the thread root: {threaded_channel}" + ); + + let top_level_event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@bot steer") + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let top_level_id = top_level_event.id.to_hex(); + let top_level_be = queue::BatchEvent { + event: top_level_event, + prompt_tag: "@mention".into(), + received_at: std::time::Instant::now(), + }; + let channel = + native_steer_prompt_blocks(ch, &top_level_be, NativeSteerScope::NonDm).join("\n\n"); + assert!( + channel.contains("Scope: channel") && !channel.contains("Scope: dm"), + "resolved non-DM at top level renders channel scope: {channel}" + ); + assert!( + channel.contains(&format!("--reply-to {top_level_id}")), + "a top-level channel reply anchors to the triggering event: {channel}" + ); + } + async fn lazy_resolver_with_response( response: serde_json::Value, ) -> ( diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 38749577398..c98785aee17 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -373,17 +373,16 @@ pub enum ControlSignal { }, } -/// Goose-native non-cancelling steer request, sent from the main loop to an -/// in-flight prompt task's read loop via a capacity-1 mpsc channel. +/// Non-cancelling steer request, sent from the main loop to an in-flight prompt +/// task's read loop via a capacity-1 mpsc channel. /// /// The read loop owns the `AcpClient`'s reader/writer for the duration of the /// turn, so we cannot drive a steer write from the main thread directly. The -/// main loop carries the steer prompt body (already framed by -/// `queue::native_steer_framing()` + `queue::format_event_block`); the read -/// loop completes `sessionId` (lexical) and `expectedRunId` -/// (`AcpClient::active_run_id` at write time) when it actually emits the -/// JSON-RPC request. The main loop awaits a `SteerAck` on the `ack_tx` -/// oneshot. +/// main loop carries the steer prompt body (already built by +/// `queue::format_native_steer_prompt`); the read loop completes `sessionId` +/// (lexical) and `expectedRunId` (`AcpClient::active_run_id` at write time) when +/// it actually emits the JSON-RPC request. The main loop awaits a `SteerAck` on +/// the `ack_tx` oneshot. /// /// ## Why the read loop fills params, not the main loop /// @@ -412,8 +411,9 @@ pub enum ControlSignal { pub struct SteerRequest { /// Prompt body text blocks. Each entry becomes one `text` content /// block in `params.prompt`. Built by the main loop via - /// `queue::native_steer_framing()` + `queue::format_event_block` so - /// the wording cannot drift from the cancel+merge fallback path. + /// `queue::format_native_steer_prompt` — see that function for what a steer + /// body carries, what it omits, and where it diverges from the cancel+merge + /// fallback. pub prompt_blocks: Vec, /// Oneshot for the read loop to report the outcome. pub ack_tx: tokio::sync::oneshot::Sender, diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs index 60866518bad..a4583c70c5b 100644 --- a/crates/buzz-acp/src/queue.rs +++ b/crates/buzz-acp/src/queue.rs @@ -1095,11 +1095,13 @@ fn format_prompt_actor(pubkey: &str, profile_lookup: Option<&PromptProfileLookup /// Includes: event_id, channel (name + UUID), kind, sender (hex + npub), /// time, content, all tags (never stripped), and parsed structural fields. /// -/// Reused by the goose-native steer path (lib.rs mode-gate) to render the -/// single withheld event for delivery via `_goose/unstable/session/steer`, -/// without paying for the batch-level context blocks the in-flight turn -/// already has. -pub(crate) fn format_event_block( +/// Reused by [`format_native_steer_prompt`] to render the single withheld event. +/// +/// Keep this private. Reachable from outside this module, it lets a caller +/// assemble a steer body from the event alone, with no routing context — the +/// defect [`format_native_steer_prompt`] exists to prevent. Widening it is how +/// that defect comes back. +fn format_event_block( channel_id: Uuid, channel_info: Option<&PromptChannelInfo>, be: &BatchEvent, @@ -1224,20 +1226,34 @@ fn turn_is_human_facing( thread_tags.mentioned_pubkeys.iter().any(|pk| !is_agent(pk)) } -/// Resolve the `--reply-to` anchor for a non-DM turn. +/// Resolve the `--reply-to` anchor for a 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 -/// - top-level → the triggering event id, which becomes the new thread root +/// - DM in a thread → the triggering event, not the root: a DM thread's root +/// is its opening message, not the message being answered +/// - DM at top level → no anchor; there is no thread to stay in +/// - human-facing ([`turn_is_human_facing`]) in a thread → the thread ROOT, +/// keeping the reply flat at layer 1 +/// - human-facing top-level → the triggering event, which becomes the root +/// - agent↔agent → `None`; deep nesting is intentional there /// -/// Returns `None` for agent↔agent turns, leaving the agent free to nest deeply -/// (intentional for agent coordination). +/// `is_dm` must be definitive. Both branches are wrong for a channel nobody +/// could classify, and the author gate's `is_dm_channel` is no help — it calls an +/// *unresolved* channel a DM to fail closed. So the native path declines without +/// metadata; the fallback retries at flush time and, failing that, +/// `format_prompt`'s `unwrap_or(false)` hands this function a guess anyway. fn resolve_reply_anchor( sender_pubkey: &str, thread_tags: &ThreadTags, triggering_event_id: &str, + is_dm: bool, profile_lookup: Option<&PromptProfileLookup>, ) -> Option { + if is_dm { + return thread_tags + .root_event_id + .is_some() + .then(|| triggering_event_id.to_string()); + } if !turn_is_human_facing(sender_pubkey, thread_tags, profile_lookup) { return None; } @@ -1587,27 +1603,15 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec (&'static str, &'static str) { +/// Framing parity is the claim here, not behavioural parity — anchoring differs +/// by design, see [`resolve_reply_anchor`]. +/// +/// Keep this private, with [`format_event_block`]: the two together are the +/// hand-assembled steer body that shipped without a reply destination. +fn native_steer_framing() -> (&'static str, &'static str) { let framing = MergeFraming::for_reason(Some(CancelReason::Steer)); (framing.new_header_single, framing.closing_note) } +/// Format the complete prompt body for a native (non-cancelling) steer of a +/// single withheld event. +/// +/// A *delta* into a live turn: standing context, channel name and description, +/// profile labels and conversation history are omitted, since the turn usually +/// holds them and the hints tell the agent to fetch what it lacks. Routing +/// context is exempt — a steer can arrive from a different thread, or scope, than +/// the one the turn is working in. +/// +/// No profile lookup: producing one is an async relay query, and this path runs +/// synchronously on the main event loop. [`turn_is_human_facing`] therefore reads +/// every identity as human, so every non-DM native steer is anchored, including +/// the agent↔agent ones the cancel+merge fallback leaves free to nest — +/// deliberate, since losing a human's reply destination is the worse failure. +/// +/// `is_dm` must be definitive channel metadata, not the author gate's +/// fail-closed classification — see [`resolve_reply_anchor`]. +pub(crate) fn format_native_steer_prompt(channel_id: Uuid, be: &BatchEvent, is_dm: bool) -> String { + let thread_tags = parse_thread_tags(&be.event); + let context = format_context_hints( + channel_id, + None, // channel_info: name and description are not re-rendered + &thread_tags, + is_dm, + // No conversation context attached, and none claimed as delivered: a + // steer can cross into a thread this turn has not seen. + false, + false, + resolve_reply_anchor( + &be.event.pubkey.to_hex(), + &thread_tags, + &be.event.id.to_hex(), + is_dm, + None, + ) + .as_deref(), + ); + let (header, closing) = native_steer_framing(); + let event_block = format_event_block(channel_id, None, be, None); + format!( + "{context}\n\n{header}\n\n[Buzz event: {}]\n{event_block}\n\n{closing}", + be.prompt_tag + ) +} + #[cfg(test)] mod tests { use super::*; @@ -2250,6 +2299,249 @@ mod tests { assert!(!prompt.contains("supersedes")); } + // ── Native steer prompt body ───────────────────────────────────────────── + + /// A NIP-10 reply `e` tag rooting an event at `root`. + fn reply_e_tag(root: &str) -> Vec> { + vec![vec![ + "e".into(), + root.to_string(), + "".into(), + "reply".into(), + ]] + } + + fn steer_batch_event(event: Event) -> BatchEvent { + BatchEvent { + event, + prompt_tag: "@mention".into(), + received_at: Instant::now(), + } + } + + /// The regression this path exists to prevent: a native steer from thread B, + /// arriving while the turn works in thread A, must name thread B as the reply + /// destination — the same answer the cancel+merge fallback gives for the same + /// event (`test_steer_cross_thread_reply_targets_steering_message`). Before + /// the fix the native body was framing + event block only, carrying no + /// `[Context]` and no anchor, so thread A was the only live destination in + /// the agent's context. + #[test] + fn test_native_steer_cross_thread_matches_fallback_anchor() { + let ch = Uuid::new_v4(); + let thread_a = "a".repeat(64); + let thread_b = "b".repeat(64); + + let original = + make_event_with_tags("@bot keep working on thread A", reply_e_tag(&thread_a)); + let steering = make_event_with_tags("@bot note from thread B", reply_e_tag(&thread_b)); + let steering_id = steering.id.to_hex(); + + let native = format_native_steer_prompt(ch, &steer_batch_event(steering.clone()), false); + let fallback = format_prompt( + &FlushBatch { + channel_id: ch, + events: vec![steer_batch_event(steering)], + cancelled_events: vec![steer_batch_event(original)], + cancel_reason: Some(CancelReason::Steer), + }, + &FormatPromptArgs::default(), + ) + .join("\n\n"); + + for (transport, prompt) in [("native", &native), ("fallback", &fallback)] { + assert!( + prompt.contains("[Context]"), + "{transport} must state routing context: {prompt}" + ); + assert!( + prompt.contains(&format!("Thread root: {thread_b}")), + "{transport} must scope to the steering thread: {prompt}" + ); + assert!( + prompt.contains(&format!("--reply-to {thread_b}")), + "{transport} must anchor to the steering thread root: {prompt}" + ); + assert!( + !prompt.contains(&format!("--reply-to {thread_a}")), + "{transport} must not anchor to the turn's original thread: {prompt}" + ); + } + + // Native keeps its delta shape: shared steer framing and the event + // itself, without the fallback's original-request section. + assert!(native.contains("[New message — arrived while you were working]")); + assert!(native.contains("[Buzz event: @mention]")); + assert!(native.contains(&format!("Event ID: {steering_id}"))); + assert!(native.contains("Continue your in-progress work")); + assert!(!native.contains("[What you were working on]")); + } + + /// The anchor asymmetry from [`format_native_steer_prompt`], pinned. Same + /// batch as the parity test above; the only variable is the profile lookup. + #[test] + fn test_native_steer_anchors_the_agent_turn_the_fallback_leaves_free() { + let ch = Uuid::new_v4(); + let thread_a = "a".repeat(64); + let thread_b = "b".repeat(64); + + let original = + make_event_with_tags("@bot keep working on thread A", reply_e_tag(&thread_a)); + let steering = make_event_with_tags("@bot note from thread B", reply_e_tag(&thread_b)); + // Keyed on the steering event: `format_prompt` resolves the anchor from + // the batch's last event, not the cancelled one. + let agents_only = HashMap::from([(steering.pubkey.to_hex(), profile(true))]); + + let native = format_native_steer_prompt(ch, &steer_batch_event(steering.clone()), false); + let fallback = format_prompt( + &FlushBatch { + channel_id: ch, + events: vec![steer_batch_event(steering)], + cancelled_events: vec![steer_batch_event(original)], + cancel_reason: Some(CancelReason::Steer), + }, + &FormatPromptArgs { + profile_lookup: Some(&agents_only), + ..FormatPromptArgs::default() + }, + ) + .join("\n\n"); + + assert!( + native.contains(&format!("--reply-to {thread_b}")), + "native has no profile lookup, so it anchors every non-DM steer: {native}" + ); + assert!( + !fallback.contains("--reply-to"), + "fallback sees an agent-only turn and leaves it free to nest: {fallback}" + ); + } + + /// A steer nested below its thread root must state both ancestry lines and + /// still anchor at the root. Fixtures with one `reply` tag make root and + /// parent the same id, so the `Parent:` line goes unrendered. + #[test] + fn test_native_steer_nested_reply_states_root_and_parent() { + let ch = Uuid::new_v4(); + let root_b = "b".repeat(64); + let parent_c = "c".repeat(64); + + let steering = make_event_with_tags( + "@bot note from below the root of thread B", + vec![ + vec!["e".into(), root_b.clone(), String::new(), "root".into()], + vec!["e".into(), parent_c.clone(), String::new(), "reply".into()], + ], + ); + + let prompt = format_native_steer_prompt(ch, &steer_batch_event(steering), false); + + assert!( + prompt.contains(&format!("Thread root: {root_b}")), + "{prompt}" + ); + assert!(prompt.contains(&format!("Parent: {parent_c}")), "{prompt}"); + assert!(prompt.contains(&format!("--reply-to {root_b}")), "{prompt}"); + assert!( + !prompt.contains(&format!("--reply-to {parent_c}")), + "a nested steer stays flat at layer 1, anchored to the root: {prompt}" + ); + } + + #[test] + fn test_native_steer_top_level_opens_thread_at_steering_event() { + let ch = Uuid::new_v4(); + let steering = make_event_with_tags("@bot new subject entirely", vec![]); + let steering_id = steering.id.to_hex(); + + let prompt = format_native_steer_prompt(ch, &steer_batch_event(steering), false); + + assert!(prompt.contains("Scope: channel"), "{prompt}"); + assert!( + prompt.contains("This is a new top-level message"), + "{prompt}" + ); + assert!( + prompt.contains(&format!("--reply-to {steering_id}")), + "{prompt}" + ); + } + + /// DM replies anchor to the steering event, not the DM's opening message — + /// the rule `format_prompt` already applies (`resolve_reply_anchor`). + #[test] + fn test_native_steer_dm_reply_anchors_to_steering_event() { + let ch = Uuid::new_v4(); + let dm_root = "d".repeat(64); + let steering = make_event_with_tags("one more thing", reply_e_tag(&dm_root)); + let steering_id = steering.id.to_hex(); + + let prompt = format_native_steer_prompt(ch, &steer_batch_event(steering), true); + + assert!(prompt.contains("Scope: dm"), "{prompt}"); + assert!( + prompt.contains(&format!("--reply-to {steering_id}")), + "{prompt}" + ); + assert!( + !prompt.contains(&format!("--reply-to {dm_root}")), + "a DM reply anchors to the message being answered, not the DM root: {prompt}" + ); + } + + #[test] + fn test_native_steer_top_level_dm_forces_no_anchor() { + let ch = Uuid::new_v4(); + let steering = make_event_with_tags("hey", vec![]); + + let prompt = format_native_steer_prompt(ch, &steer_batch_event(steering), true); + + assert!(prompt.contains("Scope: dm"), "{prompt}"); + assert!( + !prompt.contains("--reply-to"), + "a top-level DM gets no forced anchor: {prompt}" + ); + } + + /// Routing context is in; the enrichment a full dispatch adds stays out — + /// pins the delta decision so native steer cannot grow into a full prompt. + #[test] + fn test_native_steer_omits_the_enrichment_a_full_dispatch_adds() { + let ch = Uuid::new_v4(); + let steering = make_event_with_tags("@bot note", reply_e_tag(&"b".repeat(64))); + + let prompt = format_native_steer_prompt(ch, &steer_batch_event(steering), false); + + for absent in [ + "[Base]", + "[Agent Instructions]", + "[Team Instructions]", + "[Agent Memory — core]", + "[Channel Canvas]", + "[Thread Context]", + "[Conversation Context]", + ] { + assert!( + !prompt.contains(absent), + "native steer must not repeat {absent}: {prompt}" + ); + } + // Scope to the `[Context]` block: the event block renders its own + // unenriched `Channel:` line, so an unscoped assertion passes regardless. + let (context, _) = prompt + .split_once("\n\n[New message") + .expect("native steer body opens with [Context], then the steer header"); + assert!(context.starts_with("[Context]"), "{context}"); + assert!( + context.contains(&format!("Channel: {ch}")), + "channel must render as a bare UUID, unenriched: {context}" + ); + assert!( + !context.contains("Description:"), + "channel description belongs to the original prompt, not the delta: {context}" + ); + } + // ── Test 9b: requeue preserves events ──────────────────────────────────── #[test] @@ -3580,7 +3872,7 @@ mod tests { fn test_anchor_human_in_thread_uses_root() { // Human asks inside a thread → anchor to the thread ROOT (flat at L1). let tags = thread_tags(Some(ROOT_ID), &[AGENT_A_PK]); - let anchor = resolve_reply_anchor(HUMAN_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(HUMAN_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor.as_deref(), Some(ROOT_ID)); } @@ -3588,7 +3880,7 @@ mod tests { fn test_anchor_human_top_level_uses_triggering_event() { // Human top-level mention (no thread tags) → triggering event is root. let tags = thread_tags(None, &[AGENT_A_PK]); - let anchor = resolve_reply_anchor(HUMAN_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(HUMAN_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor.as_deref(), Some(TRIGGER_ID)); } @@ -3596,14 +3888,14 @@ mod tests { fn test_anchor_agent_to_agent_in_thread_is_none() { // Agent pings agent inside a thread → no forced anchor (deep nesting ok). let tags = thread_tags(Some(ROOT_ID), &[AGENT_B_PK]); - let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor, None); } #[test] fn test_anchor_agent_to_agent_top_level_is_none() { let tags = thread_tags(None, &[AGENT_B_PK]); - let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor, None); } @@ -3611,7 +3903,7 @@ mod tests { fn test_anchor_agent_sender_but_human_tagged_flattens() { // Agent-authored, but a human is tagged → human-facing → anchor to root. let tags = thread_tags(Some(ROOT_ID), &[AGENT_B_PK, HUMAN_PK]); - let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor.as_deref(), Some(ROOT_ID)); } @@ -3619,7 +3911,7 @@ mod tests { fn test_anchor_unknown_identity_treated_as_human() { // No profile lookup → fail open (treat as human so visibility is kept). let tags = thread_tags(Some(ROOT_ID), &[]); - let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, None); + let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, false, None); assert_eq!(anchor.as_deref(), Some(ROOT_ID)); } @@ -3628,7 +3920,7 @@ mod tests { // Raw p-tag presence must NOT flatten when every tagged pubkey is an // agent — this is the regression Pinky flagged. let tags = thread_tags(Some(ROOT_ID), &[AGENT_A_PK, AGENT_B_PK]); - let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, Some(&id_lookup())); + let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, false, Some(&id_lookup())); assert_eq!(anchor, None); } diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 9dca8c82c37..dd297633dc4 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -120,6 +120,40 @@ run_unit_tests() { # `just test-unit` — the two lists must stay in step. run_test_step "buzz-agent unit tests" \ cargo test -p buzz-agent --lib -- --nocapture + + # buzz-acp harness: ACP prompt formatting, the event queue, and the pool + # lifecycle state machine. No infra. Mirrors the nextest path in + # `just test-unit` — the two lists must stay in step. + run_test_step "buzz-acp unit tests" run_buzz_acp_unit_tests +} + +# buzz-acp's tests, with the `BUZZ_*` family cleared. `just test-unit` scrubs the +# same family before its nextest path. +# +# `config.rs` asserts clap defaults through `CliArgs::parse_from`, which reads +# `#[arg(env)]` unconditionally, so those tests assume a fixed environment — and +# a buzz-acp-hosted agent sets the very variables the package under test reads. +# The whole family goes rather than a list of names: a list only approximates a +# no-Buzz-config environment, and it drifts as tests are added. Today +# `BUZZ_ACP_LAZY_POOL` and `BUZZ_ACP_MULTIPLE_EVENT_HANDLING` break default +# assertions and `BUZZ_ACP_ALLOWED_RESPOND_TO` breaks a test whose premise is that +# the option is unset. Subshell so the other steps keep their environment. +# shellcheck disable=SC2329 # invoked indirectly, via run_test_step "$@" +run_buzz_acp_unit_tests() { + ( + while IFS= read -r var; do + unset "$var" + done < <(env | sed -n 's/^\(BUZZ_[A-Za-z0-9_]*\)=.*/\1/p') + # `run_test_step` invokes this from an `if`, which suppresses `errexit` for + # the whole call, so a failed `unset` — a readonly variable — would leak and + # still report the step as passed. Assert the outcome, not each `unset`. + leaked="$(env | sed -n 's/^\(BUZZ_[A-Za-z0-9_]*\)=.*/\1/p' | tr '\n' ' ')" + if [ -n "${leaked}" ]; then + error "BUZZ_* survived the scrub: ${leaked}" + exit 1 + fi + cargo test -p buzz-acp -- --nocapture + ) } # ---- DB / integration tests (infra required) --------------------------------