mirror of
https://github.com/block/buzz.git
synced 2026-10-27 12:40:30 +00:00
fix(acp): keep agent-to-agent thread replies in the thread (#8124)
## Summary When an agent @mentions another agent inside a thread, `resolve_reply_anchor` returned `None`, so the receiving agent's `<context>` had no `--reply-to` instruction. A harness that follows the base prompt literally then posted its reply at the channel top level. Agent↔agent turns in a thread now anchor to the triggering event. The reply stays in the thread and can still nest (the reason #1281 left these turns unanchored). Top-level agent↔agent turns and all human-facing turns are unchanged. Native steering keeps the running turn's `<context>`, so a message steered into a turn anchored to its trigger would be answered under that trigger too. The prompt task now records whether its anchor is the trigger, next to its DM classification (`PromptDmClassification` → `PromptRouting`). The steer guard refuses native steering into a channel thread turn whose anchor is the trigger, or not yet recorded. That message takes the cancel+merge path, whose re-prompt carries its own `<context>`. Behavior change: a same-thread follow-up that arrives before the running turn's prompt is formatted (for example during `session/new`) now cancels and merges instead of steering natively, because the anchor is not known yet. Once the prompt is formatted, follow-ups to human-facing thread turns steer natively as before. ### Related issue Fixes #8122 ### Testing - `cargo test -p buzz-acp -- --test-threads=1` (BUZZ_* env cleared): 1046 passed. `cargo clippy -p buzz-acp --all-targets -D warnings` clean. Pre-push workspace `rust-tests` hook passed. - New regressions, each confirmed to fail when its guard is removed: - `human_follow_up_to_agent_thread_turn_cancels_and_merges`: a real dispatched turn with agent profiles served by the relay fixture. A human message in the same thread is not steered natively. - `test_human_merged_into_agent_thread_turn_anchors_to_root`: the merged re-prompt anchors to the thread root. - `channel_same_thread_follow_up_during_prompt_steers_natively` / `..._during_setup_cancels_and_merges`. - Live local (`buzz-relay` + `buzz-acp` at this head + `buzz-pi-acp`; agent B is a CLI identity with a NIP-OA auth tag): - B @mentions A in a human-rooted thread → A's `<context>` says `--reply-to <B's event>`, and the reply nests under B. On `main` the `<context>` had no reply instruction. - B asks A to `sleep 30` then reply; while A works, the human @mentions A in the same thread → `control signal sent to in-flight task` (cancel+merge), the re-prompt says `--reply-to <root>`, and both replies land at the thread root. --------- Signed-off-by: Leo <a3812943bac59c128e311ca05f83bf91113f179bb6fa9ca96c3d8efe15c330e6@buzz.block.builderlab.xyz> Co-authored-by: Leo <a3812943bac59c128e311ca05f83bf91113f179bb6fa9ca96c3d8efe15c330e6@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -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<bool>,
|
||||
@@ -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<pool::SteerRequest>, Option<ControlSignal>) {
|
||||
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 `<context>`.
|
||||
#[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
|
||||
/// `<context>` does not carry, so it takes the cancel+merge path.
|
||||
#[tokio::test]
|
||||
|
||||
@@ -2420,7 +2420,7 @@ pub async fn run_prompt_task(
|
||||
result_tx: mpsc::UnboundedSender<PromptResult>,
|
||||
control_rx: Option<tokio::sync::oneshot::Receiver<ControlSignal>>,
|
||||
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()
|
||||
|
||||
@@ -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 `<context>`
|
||||
/// (`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 `<context>` the agent actually received.
|
||||
/// The prompt task records each fact when it formats the turn's `<context>`.
|
||||
/// 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
|
||||
/// `<context>` the agent actually received.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct PromptDmClassification(Arc<OnceLock<bool>>);
|
||||
pub struct PromptRouting {
|
||||
is_dm: Arc<OnceLock<bool>>,
|
||||
trigger_anchor: Arc<OnceLock<bool>>,
|
||||
}
|
||||
|
||||
impl PromptDmClassification {
|
||||
/// Record the prompt's classification. Later calls are ignored: a turn has
|
||||
/// one `<context>`.
|
||||
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
|
||||
/// `<context>`.
|
||||
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<bool> {
|
||||
self.0.get().copied()
|
||||
/// The recorded DM classification; `None` until it is recorded.
|
||||
pub fn is_dm(&self) -> Option<bool> {
|
||||
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<bool> {
|
||||
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 `<context>`.
|
||||
///
|
||||
/// 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<PromptDmClassification> {
|
||||
/// how it routed replies (see [`PromptRouting`]).
|
||||
pub fn in_flight_prompt_routing(&self, scope: &SessionScope) -> Option<PromptRouting> {
|
||||
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<String> {
|
||||
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 `<context>`.
|
||||
///
|
||||
/// Limits prompt bloat from unusually long descriptions. Multi-line
|
||||
@@ -2327,9 +2387,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
|
||||
// Human-facing turns are anchored so replies stay readable at layer 1:
|
||||
// - in a thread → anchor to the thread ROOT (no depth-2 nesting)
|
||||
// - top-level → anchor to the triggering event (it becomes the root)
|
||||
// Agent↔agent turns get no forced anchor — deep nesting is intentional
|
||||
// there. DMs are always 1:1 with a human, so they always anchor.
|
||||
// `ReplyRoute::accepts_steer` mirrors this DM rule; keep them in sync.
|
||||
// Agent↔agent turns in a thread anchor to the triggering event — deep
|
||||
// nesting is intentional there — and top-level ones get no anchor.
|
||||
// DMs are always 1:1 with a human, so they always anchor.
|
||||
// `ReplyRoute::accepts_steer` mirrors this DM rule, and
|
||||
// `reply_anchor_is_trigger` reports the agent↔agent thread anchor to the
|
||||
// native-steer guard; keep them in sync.
|
||||
let sender_pubkey = last_event.event.pubkey.to_hex();
|
||||
let reply_anchor = if is_dm {
|
||||
thread_tags
|
||||
@@ -4964,11 +5027,12 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_anchor_agent_to_agent_in_thread_is_none() {
|
||||
// Agent pings agent inside a thread → no forced anchor (deep nesting ok).
|
||||
fn test_anchor_agent_to_agent_in_thread_uses_triggering_event() {
|
||||
// Agent pings agent inside a thread → reply to the triggering event:
|
||||
// stays in the thread, nesting allowed.
|
||||
let tags = thread_tags(Some(ROOT_ID), &[AGENT_B_PK]);
|
||||
let anchor = resolve_reply_anchor(AGENT_A_PK, &tags, TRIGGER_ID, Some(&id_lookup()));
|
||||
assert_eq!(anchor, None);
|
||||
assert_eq!(anchor.as_deref(), Some(TRIGGER_ID));
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -5000,7 +5064,69 @@ mod tests {
|
||||
// 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()));
|
||||
assert_eq!(anchor, None);
|
||||
assert_eq!(anchor.as_deref(), Some(TRIGGER_ID));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_human_merged_into_agent_thread_turn_anchors_to_root() {
|
||||
// An agent↔agent turn in a thread anchors to its trigger. When a
|
||||
// human's message in that thread joins the batch (the steer guard's
|
||||
// cancel+merge path), the merged prompt anchors to the root again.
|
||||
let ch = Uuid::new_v4();
|
||||
let root = "a".repeat(64);
|
||||
let agent = Keys::generate();
|
||||
let human = Keys::generate();
|
||||
let other_agent = Keys::generate().public_key().to_hex();
|
||||
let lookup = HashMap::from([
|
||||
(agent.public_key().to_hex(), profile(true)),
|
||||
(other_agent.clone(), profile(true)),
|
||||
]);
|
||||
let in_thread = |keys: &Keys, content: &str| BatchEvent {
|
||||
edit: None,
|
||||
event: EventBuilder::new(Kind::Custom(9), content)
|
||||
.tags([
|
||||
nostr::Tag::parse(["e", root.as_str(), "", "reply"]).unwrap(),
|
||||
nostr::Tag::parse(["p", other_agent.as_str()]).unwrap(),
|
||||
])
|
||||
.sign_with_keys(keys)
|
||||
.unwrap(),
|
||||
prompt_tag: "@mention".into(),
|
||||
received_at: Instant::now(),
|
||||
};
|
||||
let batch = |events| FlushBatch {
|
||||
channel_id: ch,
|
||||
scope: conv(ch),
|
||||
events,
|
||||
cancelled_events: vec![],
|
||||
cancel_reason: None,
|
||||
};
|
||||
let args = FormatPromptArgs {
|
||||
profile_lookup: Some(&lookup),
|
||||
..Default::default()
|
||||
};
|
||||
let agent_event = in_thread(&agent, "agent asks");
|
||||
let agent_event_id = agent_event.event.id.to_hex();
|
||||
|
||||
let agent_only = batch(vec![agent_event.clone()]);
|
||||
assert!(reply_anchor_is_trigger(&agent_only, false, Some(&lookup)));
|
||||
let prompt = format_prompt(&agent_only, &args).join("\n\n");
|
||||
assert!(
|
||||
prompt.contains(&format!("--reply-to {agent_event_id}")),
|
||||
"{prompt}"
|
||||
);
|
||||
|
||||
let merged = batch(vec![agent_event, in_thread(&human, "human asks")]);
|
||||
assert!(!reply_anchor_is_trigger(&merged, false, Some(&lookup)));
|
||||
let prompt = format_prompt(&merged, &args).join("\n\n");
|
||||
assert!(prompt.contains(&format!("--reply-to {root}")), "{prompt}");
|
||||
assert!(
|
||||
!prompt.contains(&format!("--reply-to {agent_event_id}")),
|
||||
"{prompt}"
|
||||
);
|
||||
|
||||
// A DM renders its own anchor rule, which the steer guard handles
|
||||
// separately.
|
||||
assert!(!reply_anchor_is_trigger(&agent_only, true, Some(&lookup)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Reference in New Issue
Block a user