jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 04:30:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): steer mid-turn DM follow-ups natively again (#8058)

Browse Source 
🤖
## Summary
Some agents can take a new message into a turn that is already running.
This is called "native steering": the agent keeps working and folds the
new message in, instead of stopping and starting over.

#6132 made this safer in channels: a message is steered in only when it
belongs where the running turn will reply. In a channel, each top-level
message gets its own reply thread, so a second top-level message must
not be steered into a turn that replies in the first one's thread.

That rule also ran in DMs, where it does not fit. In a DM, a top-level
message is answered at the top of the conversation, not in a new thread.
So every top-level DM message replies in the same place. But the rule
treated each one as its own thread. As a result, **every DM follow-up
sent while the agent was working cancelled and restarted the turn**,
instead of being steered in as it was before #6132.

Now:
- **DMs:** a follow-up is steered natively when it replies in the same
place as the running turn: both top-level, or both in the same thread. A
thread reply during a top-level DM turn (or the reverse) still takes the
cancel-and-restart path, because the running turn would reply in the
wrong place.
- **Channels:** unchanged from #6132.

## Details
- The queue now records each running turn's reply route: its thread root
(or none) and its reply thread. One function,
`ReplyRoute::accepts_steer`, decides whether a new message may be
steered in. It compares thread roots in DMs and reply threads in
channels.
- The DM rule mirrors how the prompt is built: a top-level DM's
`<context>` has no `--reply-to` target, and a threaded DM's target is in
its thread.
- The steer decision must agree with how the running turn's prompt was
written. The prompt task records whether it writes the prompt as a DM,
and the steer decision reads that recorded value. So if channel metadata
was missing when the turn started (the prompt then uses the normal
channel form) and loads later, the DM rule is not applied to that turn.
- The prompt task records this value as soon as channel metadata
resolves, before session setup. A follow-up sent during a slow
`session/new` is steered in, as on main. When an `initial_message` setup
prompt runs first, the value is recorded after that setup turn, because
the setup prompt reads the same steer mailbox.
- Before the value is recorded, the steer decision uses the listener's
own channel classification. A channel the listener knows is not a DM is
always written in channel form, so the channel rule applies at once,
exactly as on main. A DM, or a channel whose type is not resolved, takes
the cancel-and-restart path until the value is recorded.

### Related issue
Follow-up to #6132 (merged). Found in review after merge. No separate
issue.

### Testing
- New tests drive DM messages through the listener's real steer decision
with a running turn. A second top-level DM message is steered natively
with no cancel. A DM thread reply during a top-level DM turn is not
steered and takes cancel-and-restart.
- Direct tests of the steer rule cover channel and DM cases: same
thread, different thread, top-level after top-level, and top-level mixed
with thread replies.
- With the DM rule disabled, the DM rule test and the top-level DM
listener test fail.
- Listener tests also cover a DM turn whose prompt was written as a
channel (top-level follow-up is not steered) and a DM follow-up that
arrives before the value is recorded (not steered).
- Production-path tests dispatch a real turn through `dispatch_pending`
and `run_prompt_task` to an agent that holds at `session/new` or
`session/prompt`, then send a follow-up through the real listener. A
top-level DM follow-up during the prompt or during `session/new` is
steered with no cancel. A DM follow-up before an `initial_message` setup
turn is not steered. A same-thread channel follow-up during setup is
steered with no cancel.
- Mutation checks: each fails its test. They are: not recording the
value in `run_prompt_task`, passing a detached value from dispatch, not
recording it before session setup, recording it before an
`initial_message` turn, and removing the known-channel rule for the
unrecorded case.
- `cargo test -p buzz-acp` passes on the rebased head, with the
timing-sensitive `acp::tests` run alone (`--test-threads=1`; this PR
does not change them, and they fail only under full parallel load on
this laptop). `cargo clippy -p buzz-acp --tests -D warnings` and `just
desktop-tauri-clippy` are clean.

---------

Signed-off-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
This commit is contained in:
Logan Johnson
2026-10-05 14:43:21 -04:00
committed by GitHub
parent fd885b5530
commit 0a070dfd3f
3 changed files with 575 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

399
crates/buzz-acp/src/lib.rs
View File

@@ -635,6 +635,9 @@ struct QueuedNormalListenerEvent {
/// The admitted event as a native steer would render it, including an
/// edit's resolved original-message routing.
steer_event: queue::BatchEvent,
/// The listener's DM classification of the channel (unresolved counts as
/// a DM); see [`EventQueue::in_flight_accepts_steer`].
channel_is_dm: bool,
}
impl QueuedNormalListenerEvent {
@@ -664,13 +667,18 @@ impl QueuedNormalListenerEvent {
return;
};
// A native steer keeps the running turn's `<context>`, so it may only
// carry a message that replies in the same thread. Under the channel
// policy one session spans threads; a message for another thread takes
// the cancel+merge path, whose re-prompt carries its own `<context>`.
let same_reply_thread = queue.in_flight_reply_thread(&self.scope)
== Some(self.steer_event.reply_thread().as_str());
// carry a message that replies in the same place. A channel-policy or
// DM session spans several destinations; a message for another one
// takes the cancel+merge path, whose re-prompt carries its own
// `<context>`. The DM rule applies only when the running turn's
// prompt itself was rendered as a DM.
let same_reply_route = queue.in_flight_accepts_steer(
&self.scope,
&self.steer_event.reply_route(),
self.channel_is_dm,
);
let native_attempted = matches!(signal, ControlSignal::Steer)
&& same_reply_thread
&& same_reply_route
&& try_native_steer(
pool,
queue,
@@ -714,6 +722,7 @@ impl NormalListenerIngress {
self,
queue: &mut EventQueue,
session_scope: scope::SessionScope,
channel_is_dm: bool,
) -> QueuedNormalListenerEvent {
let Self {
buzz_event,
@@ -744,6 +753,7 @@ impl NormalListenerIngress {
effective_author,
reaction_target_id,
steer_event,
channel_is_dm,
}
}
}
@@ -3521,14 +3531,13 @@ async fn run_harness(
// channel-keyed routing. Telemetry only for now —
// queue/pool partitioning by scope lands in a
// follow-up (see ticket outline steps 2–4).
let session_scope = ingress.session_scope(
config.session_policy,
is_dm_channel(
ingress.buzz_event.channel_id,
&ctx.channel_info,
)
.await,
);
let channel_is_dm = is_dm_channel(
ingress.buzz_event.channel_id,
&ctx.channel_info,
)
.await;
let session_scope =
ingress.session_scope(config.session_policy, channel_is_dm);
tracing::debug!(
channel_id = %session_scope.channel_id(),
scope = %session_scope.telemetry_label(),
@@ -3537,7 +3546,7 @@ async fn run_harness(
policy = %config.session_policy,
"admitted event — resolved session scope"
);
let queued = ingress.push(&mut queue, session_scope);
let queued = ingress.push(&mut queue, session_scope, channel_is_dm);
// 👀 — immediate "seen" reaction, only if the event
// was actually queued (not dropped by DedupMode::Drop).
// Fire-and-forget: on rare fast-failure paths the
@@ -4350,8 +4359,9 @@ fn try_native_steer(
// 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).
// The caller steers natively only a message in the running turn's reply
// thread, so the turn's own `<context>` still routes the reply. An edit's
// The caller steers natively only a message that replies where the
// running turn replies (`ReplyRoute::accepts_steer`), so the turn's own
// `<context>` still routes the reply. An edit's
// block names its original (`Edit of:`) and the original's thread root.
let event_id_hex = be.event.id.to_hex();
let body = native_steer_body(channel_id, &be);
@@ -4873,6 +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
// native-steer guard (`EventQueue::in_flight_accepts_steer`).
let prompt_dm = queue.in_flight_prompt_dm(&scope).unwrap_or_default();
let abort_handle = pool.join_set.spawn(async move {
pool::run_prompt_task(
agent,
@@ -4882,6 +4895,7 @@ fn dispatch_pending(
result_tx,
Some(control_rx),
task_turn_id,
prompt_dm,
)
.await;
});
@@ -5603,6 +5617,7 @@ fn dispatch_heartbeat(
result_tx,
None,
task_turn_id,
Default::default(),
)
.await;
});
@@ -10016,7 +10031,7 @@ mod edit_native_steer_tests {
};
let scope = ingress.session_scope(scope::SessionPolicy::Channel, false);
let mut queue = EventQueue::new(config::DedupMode::Queue);
let queued = ingress.push(&mut queue, scope);
let queued = ingress.push(&mut queue, scope, false);
assert_eq!(queued.steer_event.edit, Some(resolved));
assert_eq!(queued.reaction_target_id, original.id.to_hex());
}
@@ -10028,6 +10043,26 @@ mod edit_native_steer_tests {
fn steer_edit_into_running_turn(
running_event: nostr::Event,
original: &nostr::Event,
) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
let edit = edit_event(&original.id.to_hex(), &[]);
let resolved = queue::ResolvedEdit {
target_event_id: original.id.to_hex(),
target_thread_tags: queue::parse_thread_tags(original),
};
steer_into_running_turn(false, Some(false), running_event, edit, Some(resolved))
}
/// 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.
fn steer_into_running_turn(
is_dm: bool,
prompt_is_dm: Option<bool>,
running_event: nostr::Event,
incoming: nostr::Event,
incoming_edit: Option<queue::ResolvedEdit>,
) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
let channel_id = Uuid::new_v4();
let ingress =
@@ -10044,10 +10079,16 @@ mod edit_native_steer_tests {
let mut queue = EventQueue::new(config::DedupMode::Queue);
let running = ingress(running_event, None);
let scope = running.session_scope(scope::SessionPolicy::Channel, false);
running.push(&mut queue, scope.clone());
let scope = running.session_scope(scope::SessionPolicy::Channel, is_dm);
running.push(&mut queue, scope.clone(), is_dm);
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 mut pool = AgentPool::from_slots(vec![]);
let (control_tx, mut control_rx) = tokio::sync::oneshot::channel();
@@ -10067,20 +10108,15 @@ mod edit_native_steer_tests {
},
);
let edit = edit_event(&original.id.to_hex(), &[]);
let resolved = queue::ResolvedEdit {
target_event_id: original.id.to_hex(),
target_thread_tags: queue::parse_thread_tags(original),
};
let edit_ingress = ingress(edit, Some(resolved));
let incoming_ingress = ingress(incoming, incoming_edit);
assert_eq!(
edit_ingress.session_scope(scope::SessionPolicy::Channel, false),
incoming_ingress.session_scope(scope::SessionPolicy::Channel, is_dm),
scope,
"channel policy: one session spans every thread"
"one conversation session spans every thread"
);
let (ack_tx, _ack_rx) = mpsc::unbounded_channel();
edit_ingress
.push(&mut queue, scope.clone())
incoming_ingress
.push(&mut queue, scope.clone(), is_dm)
.steer_or_interrupt(
MultipleEventHandling::Steer,
None,
@@ -10151,6 +10187,309 @@ mod edit_native_steer_tests {
assert!(steer.is_none(), "no native steer across top-level threads");
assert_eq!(control, Some(ControlSignal::Steer));
}
/// A top-level DM message's `<context>` names no reply target, so a
/// second top-level DM message replies in the same place and is steered
/// natively into the running turn rather than cancelling it.
#[tokio::test]
async fn dm_top_level_follow_up_steers_natively() {
let (steer, control) =
steer_into_running_turn(true, Some(true), message(None), message(None), None);
assert!(steer.is_some(), "DM follow-up is sent as a native steer");
assert_eq!(
control, None,
"native steer must not cancel the running turn"
);
}
/// Where a dispatched turn is held when the follow-up arrives: the agent
/// never answers the first request with this method.
enum HoldAt {
/// Session setup, before the turn's own prompt is formatted.
SessionNew,
/// The turn's own prompt is running.
Prompt,
}
/// Dispatch a turn for `running_event` through the production
/// `dispatch_pending` → `run_prompt_task` path to an agent that holds at
/// `hold_at`, then drive `incoming` through the real listener admission.
/// `channel_type` is the channel's metadata (`"dm"` or `"stream"`), known
/// to the listener and the prompt alike. Returns the native steer request,
/// if any, and the control signal sent to the running turn, if any.
async fn steer_into_dispatched_turn(
channel_type: &str,
initial_message: Option<&str>,
hold_at: HoldAt,
running_event: nostr::Event,
incoming: nostr::Event,
) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
let channel_id = Uuid::new_v4();
let is_dm = channel_type == "dm";
let capture = std::env::temp_dir().join(format!(
"buzz-acp-dispatched-steer-{}.ndjson",
Uuid::new_v4()
));
let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''");
let held_method = match hold_at {
HoldAt::SessionNew => "session/new",
HoldAt::Prompt => "session/prompt",
};
// Record each request and never answer: the turn stays in flight at
// its first request, which is `held_method`.
let script = format!(
r#"while IFS= read -r line; do printf '%s\n' "$line" >> '{quoted_capture}'; done"#
);
let acp = acp::AcpClient::spawn("bash", &["-c".into(), script], &[], false)
.await
.expect("spawn holding ACP");
let ingress = |event: nostr::Event| NormalListenerIngress {
buzz_event: relay::BuzzEvent {
connection_generation: 0,
channel_id,
event,
},
effective_author: "author".into(),
prompt_tag: "@mention".into(),
edit: None,
};
let running = ingress(running_event);
let scope = running.session_scope(scope::SessionPolicy::Channel, is_dm);
let mut agent = pool::OwnedAgent {
index: 0,
acp,
state: pool::SessionState::default(),
model_capabilities: None,
desired_model: None,
model_overridden: false,
desired_model_request_id: None,
desired_model_pending_ack: false,
startup_effort: None,
agent_name: "dispatched-steer-test-agent".into(),
goose_system_prompt_supported: None,
protocol_version: 1,
};
if matches!(hold_at, HoldAt::Prompt) {
agent
.state
.sessions
.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.
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 relay = tokio::spawn(async move {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
loop {
let (mut socket, _) = relay.accept().await.expect("accept relay query");
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 mut ctx = pool::tests::make_prompt_context_no_owner();
ctx.rest_client.base_url = base_url;
ctx.initial_message = initial_message.map(str::to_string);
ctx.channel_info = pool::ChannelInfoResolver::new(
HashMap::from([(
channel_id,
relay::ChannelInfo {
name: "test-channel".into(),
channel_type: channel_type.into(),
description: None,
},
)]),
ctx.rest_client.clone(),
);
let ctx = Arc::new(ctx);
let mut pool = AgentPool::from_slots(vec![Some(agent)]);
let mut queue = EventQueue::new(config::DedupMode::Queue);
running.push(&mut queue, scope.clone(), is_dm);
let mut last_activity = tokio::time::Instant::now();
let dispatched = dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, None);
assert_eq!(dispatched.len(), 1, "the running turn is dispatched");
// The turn has passed every step before `held_method` once that
// request reaches the agent.
let held_request = format!("\"{held_method}\"");
tokio::time::timeout(Duration::from_secs(10), async {
while !std::fs::read_to_string(&capture)
.unwrap_or_default()
.contains(&held_request)
{
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("dispatched turn reaches its held request");
// Observe the listener's decision through the running task's own
// steer and control channels.
let (control_tx, mut control_rx) = tokio::sync::oneshot::channel();
let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1);
let meta = pool
.task_map_mut()
.values_mut()
.find(|meta| meta.scope.as_ref() == Some(&scope))
.expect("running task");
meta.control_tx = Some(control_tx);
meta.steer_tx = Some(steer_tx);
let (ack_tx, _ack_rx) = mpsc::unbounded_channel();
ingress(incoming)
.push(&mut queue, scope.clone(), is_dm)
.steer_or_interrupt(
MultipleEventHandling::Steer,
None,
&mut pool,
&mut queue,
&ack_tx,
);
let steer = steer_rx.try_recv().ok();
let control = control_rx.try_recv().ok();
pool.join_set.abort_all();
while pool.join_set.join_next().await.is_some() {}
relay.abort();
let _ = std::fs::remove_file(&capture);
(steer, control)
}
/// The DM steer rule depends on the running prompt's own classification,
/// which travels from the queue turn through `dispatch_pending` into
/// `run_prompt_task`, which records it. Drive that production handoff
/// with DM channel metadata, then the real listener admission: a
/// 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;
assert!(steer.is_some(), "DM follow-up is sent as a native steer");
assert_eq!(
control, None,
"native steer must not cancel the running turn"
);
}
/// The prompt task records its classification as soon as channel
/// metadata resolves, so a top-level DM follow-up during a slow
/// `session/new` is steered natively too.
#[tokio::test]
async fn dm_follow_up_during_session_setup_steers_natively() {
let (steer, control) = steer_into_dispatched_turn(
"dm",
None,
HoldAt::SessionNew,
message(None),
message(None),
)
.await;
assert!(steer.is_some(), "DM follow-up is sent as a native steer");
assert_eq!(
control, None,
"native steer must not cancel the running turn"
);
}
/// An `initial_message` setup prompt reads the same steer mailbox, so a
/// DM follow-up must not be admitted for native steering before that
/// setup turn finishes; it takes the cancel+merge path.
#[tokio::test]
async fn dm_follow_up_before_initial_message_cancels_and_merges() {
let (steer, control) = steer_into_dispatched_turn(
"dm",
Some("set up the session"),
HoldAt::SessionNew,
message(None),
message(None),
)
.await;
assert!(
steer.is_none(),
"no native steer into the initial_message setup turn"
);
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`.
#[tokio::test]
async fn channel_same_thread_follow_up_during_setup_steers_natively() {
let root = "ab".repeat(32);
let (steer, control) = steer_into_dispatched_turn(
"stream",
Some("set up the session"),
HoldAt::SessionNew,
message(Some(&root)),
message(Some(&root)),
)
.await;
assert!(steer.is_some(), "same-thread follow-up is a native steer");
assert_eq!(
control, None,
"native steer must not cancel the running turn"
);
}
/// 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]
async fn dm_thread_reply_during_top_level_turn_cancels_and_merges() {
let root = "ab".repeat(32);
let (steer, control) =
steer_into_running_turn(true, Some(true), message(None), message(Some(&root)), None);
assert!(
steer.is_none(),
"no native steer into a different DM thread"
);
assert_eq!(control, Some(ControlSignal::Steer));
}
/// A DM turn whose prompt was rendered while channel metadata was
/// unavailable got channel-style `<context>`: its replies go to a thread
/// rooted at its trigger. The DM rule must not apply once metadata
/// resolves, so a top-level follow-up takes the cancel+merge path.
#[tokio::test]
async fn dm_follow_up_into_channel_formatted_turn_cancels_and_merges() {
let (steer, control) =
steer_into_running_turn(true, Some(false), message(None), message(None), None);
assert!(
steer.is_none(),
"steer guard follows the running prompt's classification"
);
assert_eq!(control, Some(ControlSignal::Steer));
}
/// Before the prompt task records its classification, nothing proves
/// which `<context>` a turn in a DM (or a channel of unresolved type)
/// will carry, so the follow-up takes the cancel+merge path.
#[tokio::test]
async fn dm_follow_up_before_prompt_classification_cancels_and_merges() {
let (steer, control) =
steer_into_running_turn(true, None, message(None), message(None), None);
assert!(steer.is_none(), "no native steer without a recorded prompt");
assert_eq!(control, Some(ControlSignal::Steer));
}
}
#[cfg(test)]

36
crates/buzz-acp/src/pool.rs
View File

@@ -2411,6 +2411,7 @@ fn send_prompt_result(
///
/// The agent is ALWAYS returned — even on panic the `JoinSet` detects the
/// abort and the caller uses `task_map` to recover the agent index.
#[allow(clippy::too_many_arguments)]
pub async fn run_prompt_task(
mut agent: OwnedAgent,
batch: Option<FlushBatch>,
@@ -2419,6 +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,
) {
// Is this a channel prompt or a heartbeat?
let source = match &batch {
@@ -2528,6 +2530,21 @@ pub async fn run_prompt_task(
},
PromptSource::Heartbeat => None,
};
// Whether this turn's prompt renders the channel as a DM. `format_prompt`
// derives the same value from `resolved_channel_info`, which is fixed
// from here on, so the main loop's native-steer guard can rely on it now
// rather than after the session setup below. A follow-up steered before
// this turn's own prompt starts waits in the steer mailbox, so it must
// not be admitted while an `initial_message` setup prompt (which reads the
// same mailbox) is still to come; that case records after the setup turn.
let prompt_is_dm = resolved_channel_info
.as_ref()
.is_some_and(|info| info.channel_type == "dm");
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);
}
//
// Core memory is delivered inside the system prompt the harness already
@@ -2985,6 +3002,8 @@ pub async fn run_prompt_task(
}
}
}
// Any `initial_message` setup turn is done; see `prompt_is_dm`.
prompt_dm.record(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
@@ -3027,10 +3046,7 @@ pub async fn run_prompt_task(
// reuse that exact typed result for prompt formatting.
let channel_info = resolved_channel_info.clone();
let is_dm = channel_info
.as_ref()
.map(|info| info.channel_type == "dm")
.unwrap_or(false);
let is_dm = prompt_is_dm;
let context_target = resolve_context_target(b, is_dm);
let hydrated_thread_root = match &context_target {
ContextTarget::Thread(root) => Some(root),
@@ -5660,7 +5676,7 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec<String>)
}
#[cfg(test)]
mod tests {
pub(crate) mod tests {
use super::*;
use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
use serde_json::json;
@@ -7134,6 +7150,7 @@ mod tests {
result_tx,
None,
"observer-test-turn".into(),
Default::default(),
));
let liveness = tokio::time::timeout(Duration::from_secs(5), async {
request_rx.await.expect("channel context lookup started");
@@ -7227,6 +7244,7 @@ done"#
result_tx.clone(),
None,
format!("turn-{turn}"),
Default::default(),
)
.await;
let result = result_rx.recv().await.expect("prompt result");
@@ -7350,6 +7368,7 @@ done"#
result_tx.clone(),
None,
format!("turn-{turn}"),
Default::default(),
)
.await;
let result = result_rx.recv().await.expect("prompt result");
@@ -7563,6 +7582,7 @@ done"#
result_tx.clone(),
None,
"first-turn".into(),
Default::default(),
)
.await;
let first_result = result_rx.recv().await.expect("first prompt result");
@@ -7582,6 +7602,7 @@ done"#
result_tx,
None,
"follow-up-turn".into(),
Default::default(),
)
.await;
let mut result = result_rx.recv().await.expect("prompt result");
@@ -7752,6 +7773,7 @@ done"#
result_tx.clone(),
None,
turn_id.into(),
Default::default(),
)
.await;
let result = result_rx.recv().await.expect("prompt result");
@@ -7916,6 +7938,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
result_tx,
None,
"next-turn".into(),
Default::default(),
)
.await;
let mut result = result_rx.recv().await.expect("next prompt result");
@@ -10402,7 +10425,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
);
}
pub(super) fn make_prompt_context_no_owner() -> PromptContext {
pub(crate) fn make_prompt_context_no_owner() -> PromptContext {
let agent_keys = nostr::Keys::generate();
make_prompt_context_impl(&agent_keys, None)
}
@@ -11140,6 +11163,7 @@ done"#
result_tx,
None,
"indeterminate-project-turn".into(),
Default::default(),
)
.await;

198
crates/buzz-acp/src/queue.rs
View File

@@ -15,6 +15,7 @@
use nostr::{Event, ToBech32};
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, OnceLock};
use std::time::{Duration, Instant};
use uuid::Uuid;
@@ -154,6 +155,81 @@ impl BatchEvent {
pub fn reply_thread(&self) -> String {
reply_thread(&self.event, self.edit.as_ref())
}
/// See [`ReplyRoute`].
pub fn reply_route(&self) -> ReplyRoute {
ReplyRoute {
root_event_id: self
.routing_thread_tags()
.root_event_id
.map(|root| root.to_ascii_lowercase()),
thread: self.reply_thread(),
}
}
}
/// Where a turn's `<context>` sends ordinary replies, as far as the
/// native-steer guard needs to know.
///
/// A native steer adds a message to a running turn without a new `<context>`,
/// so the turn keeps replying where its own `<context>` points. A message may
/// be steered natively only when that destination is also right for it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplyRoute {
/// Routed thread root, lowercase; `None` for a top-level message.
root_event_id: Option<String>,
/// See [`reply_thread`].
thread: String,
}
impl ReplyRoute {
/// Whether a turn whose `<context>` was rendered for `self` also replies
/// correctly to `incoming`.
///
/// In a channel, a top-level message is answered in a new thread rooted at
/// that message, so two messages share a destination only when they share
/// a [`reply_thread`]. In a DM, a top-level message's `<context>` names no
/// reply target (see `format_prompt`): every top-level DM message replies
/// at the top of the same conversation. DM messages therefore share a
/// destination when they share a thread root, including none.
pub fn accepts_steer(&self, incoming: &ReplyRoute, is_dm: bool) -> bool {
if is_dm {
self.root_event_id == incoming.root_event_id
} else {
self.thread == incoming.thread
}
}
}
/// Whether the running turn's prompt rendered its channel as a DM.
///
/// 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.
#[derive(Debug, Clone, Default)]
pub struct PromptDmClassification(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);
}
/// The recorded classification; `None` before the prompt is formatted.
pub fn get(&self) -> Option<bool> {
self.0.get().copied()
}
}
/// Reply destination of an in-flight turn, and how its prompt classified the
/// channel.
#[derive(Debug, Clone)]
struct InFlightReplyRoute {
route: ReplyRoute,
prompt_is_dm: PromptDmClassification,
}
/// Why a batch's prior turn was cancelled — controls how `format_prompt`
@@ -241,9 +317,9 @@ pub struct EventQueue {
in_flight_deadlines: HashMap<SessionScope, Instant>,
/// Number of events in each in-flight batch (for expiry logging).
in_flight_batch_sizes: HashMap<SessionScope, usize>,
/// Reply thread of each in-flight turn: the [`reply_thread`] of the batch
/// Reply route of each in-flight turn: the [`ReplyRoute`] of the batch
/// event whose `<context>` routes the turn's replies (its last event).
in_flight_reply_threads: HashMap<SessionScope, String>,
in_flight_reply_routes: HashMap<SessionScope, InFlightReplyRoute>,
retry_after: HashMap<SessionScope, Instant>,
/// Per-scope retry attempt counter for exponential backoff / dead-lettering.
retry_counts: HashMap<SessionScope, u32>,
@@ -285,7 +361,7 @@ impl EventQueue {
in_flight_scopes: HashSet::new(),
in_flight_deadlines: HashMap::new(),
in_flight_batch_sizes: HashMap::new(),
in_flight_reply_threads: HashMap::new(),
in_flight_reply_routes: HashMap::new(),
retry_after: HashMap::new(),
retry_counts: HashMap::new(),
dedup_mode,
@@ -433,7 +509,7 @@ impl EventQueue {
);
self.in_flight_scopes.remove(&scope);
self.in_flight_deadlines.remove(&scope);
self.in_flight_reply_threads.remove(&scope);
self.in_flight_reply_routes.remove(&scope);
// Recover any withheld goose-native steer events for the expired
// scope back to the queue front so normal dispatch delivers
// them. Unlike the in-flight batch above (already delivered to a
@@ -477,7 +553,7 @@ impl EventQueue {
.insert(scope.clone(), now + self.in_flight_deadline);
self.in_flight_batch_sizes
.insert(scope.clone(), cancelled.len());
self.record_in_flight_reply_thread(&scope, &cancelled);
self.record_in_flight_reply_route(&scope, &cancelled);
return Some(FlushBatch {
channel_id: scope.channel_id(),
scope,
@@ -515,7 +591,7 @@ impl EventQueue {
.insert(scope.clone(), now + self.in_flight_deadline);
self.in_flight_batch_sizes
.insert(scope.clone(), events.len());
self.record_in_flight_reply_thread(&scope, &events);
self.record_in_flight_reply_route(&scope, &events);
// Merge any cancelled events stored by requeue_as_cancelled().
let cancelled_events = self.cancelled_batches.remove(&scope).unwrap_or_default();
@@ -568,7 +644,7 @@ impl EventQueue {
self.in_flight_scopes.remove(&scope);
self.in_flight_deadlines.remove(&scope);
self.in_flight_batch_sizes.remove(&scope);
self.in_flight_reply_threads.remove(&scope);
self.in_flight_reply_routes.remove(&scope);
let now = Instant::now();
match self.retry_after.get(&scope) {
// Active throttle → scope was requeued; keep retry_counts intact.
@@ -787,7 +863,7 @@ impl EventQueue {
);
self.in_flight_scopes.remove(&scope);
self.in_flight_deadlines.remove(&scope);
self.in_flight_reply_threads.remove(&scope);
self.in_flight_reply_routes.remove(&scope);
// Symmetric with the flush_next expiry block: recover withheld
// goose-native steer events for the expired scope so they are
// not permanently orphaned in the side table.
@@ -943,25 +1019,58 @@ impl EventQueue {
self.in_flight_scopes.contains(&scope.into_scope())
}
/// The reply thread of the turn in flight for `scope`, if any.
/// Whether a message with reply route `incoming` may be steered natively
/// into the turn in flight for `scope`.
///
/// A native steer adds a message to a running turn without a new
/// `<context>`, so the turn keeps replying where its own `<context>`
/// points. A message whose [`reply_thread`] differs (possible under the
/// channel session policy) must not be steered natively; the cancel+merge
/// path re-dispatches it with its own full `<context>`.
pub fn in_flight_reply_thread(&self, scope: &SessionScope) -> Option<&str> {
self.in_flight_reply_threads.get(scope).map(String::as_str)
/// A message whose route the running turn does not accept (see
/// [`ReplyRoute::accepts_steer`]; possible whenever one session spans
/// several reply destinations: the channel session policy, or a DM) must
/// 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`.
pub fn in_flight_accepts_steer(
&self,
scope: &SessionScope,
incoming: &ReplyRoute,
channel_is_dm: bool,
) -> bool {
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,
}
}
fn record_in_flight_reply_thread(&mut self, scope: &SessionScope, events: &[BatchEvent]) {
/// 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> {
self.in_flight_reply_routes
.get(scope)
.map(|running| running.prompt_is_dm.clone())
}
fn record_in_flight_reply_route(&mut self, scope: &SessionScope, events: &[BatchEvent]) {
match events.last() {
Some(last) => {
self.in_flight_reply_threads
.insert(scope.clone(), last.reply_thread());
self.in_flight_reply_routes.insert(
scope.clone(),
InFlightReplyRoute {
route: last.reply_route(),
prompt_is_dm: PromptDmClassification::default(),
},
);
}
None => {
self.in_flight_reply_threads.remove(scope);
self.in_flight_reply_routes.remove(scope);
}
}
}
@@ -1216,8 +1325,8 @@ pub(crate) fn reaction_target_id(event: &Event) -> String {
/// the routed event itself when it is top-level (a reply opens a thread
/// rooted there). Lowercase, so equivalent hex spellings compare equal.
///
/// This is the thread-session key, and it decides whether a message may be
/// steered natively into a running turn (see [`EventQueue::in_flight_reply_thread`]).
/// This is the thread-session key. Outside DMs it also decides whether a
/// message may be steered natively into a running turn (see [`ReplyRoute`]).
pub(crate) fn reply_thread(event: &Event, edit: Option<&ResolvedEdit>) -> String {
routing_thread_tags(event, edit)
.root_event_id
@@ -2220,6 +2329,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
// - 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.
let sender_pubkey = last_event.event.pubkey.to_hex();
let reply_anchor = if is_dm {
thread_tags
@@ -2396,6 +2506,50 @@ pub(crate) fn native_steer_framing() -> (&'static str, &'static str) {
(framing.new_tag, framing.closing_note)
}
#[cfg(test)]
mod reply_route_tests {
use super::ReplyRoute;
fn route(root: Option<&str>, thread: &str) -> ReplyRoute {
ReplyRoute {
root_event_id: root.map(str::to_string),
thread: thread.to_string(),
}
}
/// A top-level message (no root) whose replies open thread `id`.
fn top_level(id: &str) -> ReplyRoute {
route(None, id)
}
/// A reply in the thread rooted at `root`.
fn in_thread(root: &str) -> ReplyRoute {
route(Some(root), root)
}
#[test]
fn channel_steers_only_within_one_reply_thread() {
assert!(in_thread("a").accepts_steer(&in_thread("a"), false));
assert!(!in_thread("a").accepts_steer(&in_thread("b"), false));
// Replies to a top-level message open a thread rooted at it.
assert!(top_level("a").accepts_steer(&in_thread("a"), false));
assert!(in_thread("a").accepts_steer(&top_level("a"), false));
assert!(!top_level("a").accepts_steer(&top_level("b"), false));
assert!(!top_level("a").accepts_steer(&in_thread("b"), false));
}
#[test]
fn dm_steers_within_one_thread_or_the_top_level() {
// Every top-level DM message replies at the top of the conversation.
assert!(top_level("a").accepts_steer(&top_level("b"), true));
assert!(in_thread("a").accepts_steer(&in_thread("a"), true));
assert!(!in_thread("a").accepts_steer(&in_thread("b"), true));
// A top-level DM turn has no `--reply-to`; a threaded one does.
assert!(!top_level("a").accepts_steer(&in_thread("a"), true));
assert!(!in_thread("a").accepts_steer(&top_level("b"), true));
}
}
#[cfg(test)]
mod tests {
use super::*;