mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
feat(buzz-acp): steering as the default mid-turn mention delivery (#1160)
Signed-off-by: Tyler Longwell <tlongwell@block.xyz> Signed-off-by: Will Pfleger <pfleger.will@gmail.com> Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@sprout-oss.stage.blox.sqprod.co> Co-authored-by: Tyler Longwell <tlongwell@block.xyz> Co-authored-by: npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc <f4a42a97e594b77bdbd8ee35191c8b28a94a4cb871d96f32921558275421fb68@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co> Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Co-authored-by: npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw <4a1bfa0013bc6d14a8d600d8bf6392efefbd2a26ac3c96c9b2a106b0d12297ca@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Tyler
co-authored by
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
Tyler Longwell
npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
Will Pfleger
npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
Tyler Longwell
npub17jjz49l9jjmhhk7cac63j8yt9z555n9cw8vk7v5jz4vzw4ppld5qgj57cc
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
Will Pfleger
npub1fgdl5qqnh3k3f2xkqrvt7cujalhm623x4s7fdjdj5yrtp5fzjl9qrjpucw
parent
14fba21e57
commit
e567491a19
+672
-27
@@ -144,6 +144,28 @@ pub struct AcpClient {
|
||||
observer_agent_index: Option<usize>,
|
||||
/// Best-effort context attached to raw ACP wire events.
|
||||
observer_context: ObserverContext,
|
||||
/// Goose-specific: most recently observed `_meta.goose.activeRunId` from
|
||||
/// a `session/update` notification of kind `session_info_update`.
|
||||
///
|
||||
/// Goose emits this whenever it starts or clears an active prompt run
|
||||
/// (`crates/goose/src/acp/server.rs:2277` `send_active_run_update`).
|
||||
/// Required as `expectedRunId` when calling the non-standard
|
||||
/// `_goose/unstable/session/steer` method to inject a message into an
|
||||
/// in-flight turn without cancelling it.
|
||||
///
|
||||
/// `None` until the first `session_info_update` arrives, or after the
|
||||
/// run clears (goose emits `activeRunId: null` at end of turn). Other
|
||||
/// agents will simply never populate this — readers must treat `None`
|
||||
/// as "no active run to steer into" and fall back to cancel+merge.
|
||||
active_run_id: Option<String>,
|
||||
/// Per-turn channel for receiving goose-native non-cancelling steer
|
||||
/// requests from the main loop. Installed by
|
||||
/// [`install_steer_rx`](Self::install_steer_rx) at dispatch and
|
||||
/// consumed (via `take()`) by `session_prompt_with_idle_timeout` so it
|
||||
/// is dropped at scope exit alongside the turn it served. `None`
|
||||
/// outside of a goose-native turn — the read loop's steer arm is
|
||||
/// disabled in that case.
|
||||
steer_rx: Option<tokio::sync::mpsc::Receiver<crate::pool::SteerRequest>>,
|
||||
}
|
||||
|
||||
impl AcpClient {
|
||||
@@ -233,6 +255,8 @@ impl AcpClient {
|
||||
observer: None,
|
||||
observer_agent_index: None,
|
||||
observer_context: ObserverContext::default(),
|
||||
active_run_id: None,
|
||||
steer_rx: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -421,7 +445,7 @@ impl AcpClient {
|
||||
}
|
||||
|
||||
let result = self
|
||||
.read_until_response_with_idle_timeout(id, idle_timeout, hard_deadline)
|
||||
.read_until_response_with_idle_timeout(session_id, id, idle_timeout, hard_deadline)
|
||||
.await;
|
||||
|
||||
// On timeout errors, leave current_hard_deadline set so cancel_with_cleanup
|
||||
@@ -463,6 +487,41 @@ impl AcpClient {
|
||||
self.last_prompt_id.is_some()
|
||||
}
|
||||
|
||||
/// Most recently observed goose `_meta.goose.activeRunId` from a
|
||||
/// `session_info_update`, if any.
|
||||
///
|
||||
/// Goose-only: other agents leave this `None` for the lifetime of the
|
||||
/// client. Read directly by `read_until_response_with_idle_timeout`'s
|
||||
/// steer arm at write time (see [`crate::pool::SteerRequest`] for
|
||||
/// why the read loop owns this); production callers do not need this
|
||||
/// accessor. Kept as `pub` so tests can introspect the field.
|
||||
#[cfg_attr(not(test), allow(dead_code))]
|
||||
pub fn active_run_id(&self) -> Option<&str> {
|
||||
self.active_run_id.as_deref()
|
||||
}
|
||||
|
||||
/// Install a per-turn steer request channel for goose-native
|
||||
/// non-cancelling mid-turn delivery.
|
||||
///
|
||||
/// Called by the dispatch path immediately before
|
||||
/// [`session_prompt_with_idle_timeout`] for all prompt tasks.
|
||||
/// The matching `Sender` is stored in `TaskMeta.steer_tx` for the
|
||||
/// main loop's mode-gate fork to drive.
|
||||
///
|
||||
/// Panics if a receiver is already installed — there is exactly one
|
||||
/// turn per `AcpClient` at a time, and stacking receivers would
|
||||
/// silently misroute steer requests across turns. The previous
|
||||
/// turn's receiver must have been consumed by the read loop and
|
||||
/// dropped at scope exit before the next turn dispatches.
|
||||
pub fn install_steer_rx(&mut self, rx: tokio::sync::mpsc::Receiver<crate::pool::SteerRequest>) {
|
||||
assert!(
|
||||
self.steer_rx.is_none(),
|
||||
"install_steer_rx: previous turn's receiver was not consumed — \
|
||||
stacking receivers would misroute steer requests across turns"
|
||||
);
|
||||
self.steer_rx = Some(rx);
|
||||
}
|
||||
|
||||
/// Cancel a turn cleanly, handling any pending permission request first.
|
||||
///
|
||||
/// Steps:
|
||||
@@ -557,7 +616,12 @@ impl AcpClient {
|
||||
// but ignore cancellation.
|
||||
let cleanup_idle = std::time::Duration::from_secs(30);
|
||||
let result = self
|
||||
.read_until_response_with_idle_timeout(prompt_id, cleanup_idle, hard_deadline)
|
||||
.read_until_response_with_idle_timeout(
|
||||
session_id,
|
||||
prompt_id,
|
||||
cleanup_idle,
|
||||
hard_deadline,
|
||||
)
|
||||
.await?;
|
||||
self.parse_stop_reason(&result)
|
||||
}
|
||||
@@ -786,14 +850,54 @@ impl AcpClient {
|
||||
/// `hard_deadline` is an absolute `Instant` (pre-computed by the caller) so
|
||||
/// that `cancel_with_cleanup` can inherit the remaining budget from the
|
||||
/// original turn rather than starting a fresh timer.
|
||||
/// Read agent messages until the response with `expected_id` arrives, or
|
||||
/// either of two timeouts fires. Returns `Result<value, IdleTimeout |
|
||||
/// HardTimeout | other>`.
|
||||
///
|
||||
/// - `idle_timeout`: silent-agent guard, **reset on every line of valid
|
||||
/// JSON** (and explicitly on `session/update` notifications).
|
||||
/// - `hard_deadline`: absolute wall-clock cap on the whole call, passed
|
||||
/// in so that `cancel_with_cleanup` can inherit the remaining budget
|
||||
/// from the original turn rather than starting a fresh timer.
|
||||
///
|
||||
/// While reading, the loop interleaves goose-native non-cancelling steer
|
||||
/// requests via `tokio::select!`. The select uses `biased` for
|
||||
/// reader-first throughput, with a pre-select deadline check at the top
|
||||
/// of every loop iteration so a continuously-ready reader arm cannot
|
||||
/// starve the hard deadline (Max's review gate). The steer arm is
|
||||
/// guarded by `pending_steer.is_none()` so at most one steer is in
|
||||
/// flight at a time; a successful steer response is routed to the
|
||||
/// caller's oneshot ack instead of being returned as the prompt result.
|
||||
///
|
||||
/// `session_id` is threaded in lexically by callers so the goose-native
|
||||
/// steer arm can complete `sessionId` in the steer JSON-RPC params at
|
||||
/// write time without needing access to outer state. See
|
||||
/// [`crate::pool::SteerRequest`] for why params are built here and not
|
||||
/// in the main loop.
|
||||
async fn read_until_response_with_idle_timeout(
|
||||
&mut self,
|
||||
session_id: &str,
|
||||
expected_id: u64,
|
||||
idle_timeout: std::time::Duration,
|
||||
hard_deadline: tokio::time::Instant,
|
||||
) -> Result<serde_json::Value, AcpError> {
|
||||
use tokio::time::Instant;
|
||||
|
||||
// Take the per-turn steer receiver into a local so it can be
|
||||
// borrowed independently of `self.reader` inside `select!`.
|
||||
// Dropped at scope exit (return paths drain `pending_steer` first
|
||||
// so the ack_tx oneshot is never leaked silently).
|
||||
let mut steer_rx = self.steer_rx.take();
|
||||
|
||||
// Tracks the in-flight steer write: `(request_id, ack_tx)`. While
|
||||
// `Some`, the steer arm is gated off so we don't stack writes,
|
||||
// and a response matching `id` is routed to the ack_tx instead
|
||||
// of being treated as the prompt result. Drained on every return
|
||||
// path with `PromptCompletedNeutral` so callers are never left
|
||||
// hanging.
|
||||
let mut pending_steer: Option<(u64, tokio::sync::oneshot::Sender<crate::pool::SteerAck>)> =
|
||||
None;
|
||||
|
||||
let mut idle_deadline = Instant::now() + idle_timeout;
|
||||
|
||||
loop {
|
||||
@@ -805,23 +909,157 @@ impl AcpClient {
|
||||
} else {
|
||||
hard_deadline
|
||||
};
|
||||
let remaining = next_deadline.saturating_duration_since(Instant::now());
|
||||
|
||||
// Pre-select deadline check — required by Max's review. Under
|
||||
// `biased`, a continuously-ready reader arm wins every poll and
|
||||
// `sleep_until(next_deadline)` is never reached, silently
|
||||
// defeating the hard-deadline guarantee for agents that keep
|
||||
// producing output (see `acp.rs:608` for why the hard deadline
|
||||
// exists). Check the classified deadline here so a steady-
|
||||
// stream agent is still bounded.
|
||||
if Instant::now() >= next_deadline {
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
// Prompt is timing out — release the withheld event via
|
||||
// PromptCompletedNeutral (no fallback signal: there is
|
||||
// no in-flight turn to signal once we return, and
|
||||
// normal dispatch handles redelivery).
|
||||
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
if idle_fires_first {
|
||||
tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity");
|
||||
return Err(AcpError::IdleTimeout(idle_timeout));
|
||||
} else {
|
||||
tracing::warn!("hard turn timeout exceeded");
|
||||
return Err(AcpError::HardTimeout);
|
||||
}
|
||||
}
|
||||
|
||||
// LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the
|
||||
// read level — the buffer never grows beyond the limit.
|
||||
let read_result = tokio::time::timeout(remaining, self.reader.next()).await;
|
||||
let read_result = tokio::select! {
|
||||
biased;
|
||||
read_result = self.reader.next() => Some(read_result),
|
||||
// Steer arm: gated off whenever a steer write is already in
|
||||
// flight so we don't stack two writes against the same
|
||||
// process. The `async { steer_rx.as_mut()?.recv().await }`
|
||||
// wrapper produces `None` when no receiver is installed,
|
||||
// which mismatches the `Some(req)` pattern and disables the
|
||||
// branch for that iteration (no busy loop). Cancel-safe:
|
||||
// `mpsc::Receiver::recv` does not lose messages on drop.
|
||||
Some(req) = async {
|
||||
match steer_rx.as_mut() {
|
||||
Some(rx) => rx.recv().await,
|
||||
None => None,
|
||||
}
|
||||
}, if pending_steer.is_none() => {
|
||||
// Selected: build steer params at write time using the
|
||||
// lexical `session_id` and the freshest `active_run_id`.
|
||||
//
|
||||
// `active_run_id` is updated by `session/update`
|
||||
// notifications inside this very loop; reading it here
|
||||
// (rather than snapshotting at dispatch) guarantees the
|
||||
// value matches what goose's run-id check will compare
|
||||
// against. If it's `None`, no `session/update` has
|
||||
// arrived yet so we cannot form a valid `expectedRunId`
|
||||
// — ack `ExpectedRunIdMissing` and drop the request
|
||||
// without writing anything. The main loop maps this to
|
||||
// the universal cancel+merge `Steer` fallback.
|
||||
match self.active_run_id.clone() {
|
||||
None => {
|
||||
tracing::warn!(
|
||||
"goose-native steer: no active_run_id at write time \
|
||||
(no session/update seen yet) — falling back to cancel+merge"
|
||||
);
|
||||
let _ = req.ack_tx.send(crate::pool::SteerAck::Err(
|
||||
crate::pool::SteerError::ExpectedRunIdMissing,
|
||||
));
|
||||
}
|
||||
Some(run_id) => {
|
||||
let id = self.next_id;
|
||||
self.next_id += 1;
|
||||
let prompt_block_refs: Vec<&str> =
|
||||
req.prompt_blocks.iter().map(String::as_str).collect();
|
||||
let params =
|
||||
build_steer_params(session_id, &run_id, &prompt_block_refs);
|
||||
let msg = serde_json::json!({
|
||||
"jsonrpc": "2.0",
|
||||
"id": id,
|
||||
"method": "_goose/unstable/session/steer",
|
||||
"params": params,
|
||||
});
|
||||
tracing::debug!(
|
||||
target: "acp::wire",
|
||||
"→ {}",
|
||||
serde_json::to_string(&msg).unwrap_or_default()
|
||||
);
|
||||
match self.write_ndjson(&msg).await {
|
||||
Ok(()) => {
|
||||
pending_steer = Some((id, req.ack_tx));
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
"goose-native steer write failed: {e} — releasing withheld event"
|
||||
);
|
||||
let _ = req.ack_tx.send(crate::pool::SteerAck::Err(
|
||||
crate::pool::SteerError::Transport(e.to_string()),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Loop back to the next iteration without consuming a
|
||||
// reader line; we'll wait for either the prompt
|
||||
// response or the steer response next.
|
||||
None
|
||||
}
|
||||
_ = tokio::time::sleep_until(next_deadline) => {
|
||||
// The pre-select check at the top of the next iteration
|
||||
// would catch this anyway, but firing the deadline arm
|
||||
// here makes the wakeup immediate (no extra reader poll
|
||||
// round-trip when stdout is idle).
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
if idle_fires_first {
|
||||
tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity");
|
||||
return Err(AcpError::IdleTimeout(idle_timeout));
|
||||
} else {
|
||||
tracing::warn!("hard turn timeout exceeded");
|
||||
return Err(AcpError::HardTimeout);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Steer arm fired (or the select selected nothing read-side this
|
||||
// iteration): no reader frame to process, loop to re-evaluate
|
||||
// deadlines and arm the next select.
|
||||
let read_result = match read_result {
|
||||
Some(r) => r,
|
||||
None => continue,
|
||||
};
|
||||
|
||||
match read_result {
|
||||
Ok(None) => return Err(AcpError::AgentExited),
|
||||
Ok(Some(Err(LinesCodecError::MaxLineLengthExceeded))) => {
|
||||
None => {
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
return Err(AcpError::AgentExited);
|
||||
}
|
||||
Some(Err(LinesCodecError::MaxLineLengthExceeded)) => {
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
return Err(AcpError::Protocol(
|
||||
"agent stdout line exceeded 10MB limit".into(),
|
||||
));
|
||||
}
|
||||
Ok(Some(Err(e))) => {
|
||||
Some(Err(e)) => {
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
return Err(AcpError::Io(std::io::Error::other(e)));
|
||||
}
|
||||
Ok(Some(Ok(line))) => {
|
||||
Some(Ok(line)) => {
|
||||
let trimmed = line.trim();
|
||||
if trimmed.is_empty() {
|
||||
continue;
|
||||
@@ -852,14 +1090,50 @@ impl AcpClient {
|
||||
// Malformed lines (skipped above) don't count as real agent activity.
|
||||
idle_deadline = Instant::now() + idle_timeout;
|
||||
|
||||
// Check for matching response (has matching id AND no `method`
|
||||
// field — a `method` field means agent-initiated request, not response).
|
||||
// Steer response routing must come BEFORE the prompt
|
||||
// response check: a steer response is a regular
|
||||
// JSON-RPC response (id + result/error, no method),
|
||||
// so the matcher must disambiguate by id. Both checks
|
||||
// share the `no method` guard.
|
||||
if let Some(id) = msg.get("id") {
|
||||
if *id == serde_json::json!(expected_id) && msg.get("method").is_none() {
|
||||
if let Some(error) = msg.get("error") {
|
||||
return Err(AcpError::AgentError(error.to_string()));
|
||||
if msg.get("method").is_none() {
|
||||
if let Some((steer_id, _)) = pending_steer.as_ref() {
|
||||
if *id == serde_json::json!(*steer_id) {
|
||||
// Take the ack_tx out and route the
|
||||
// response. We do not return — keep
|
||||
// reading until the prompt response
|
||||
// arrives.
|
||||
let (_, ack_tx) = pending_steer.take().expect("just checked");
|
||||
let ack = if let Some(error) = msg.get("error") {
|
||||
let code = error
|
||||
.get("code")
|
||||
.and_then(|c| c.as_i64())
|
||||
.unwrap_or(-1);
|
||||
let message = error.to_string();
|
||||
crate::pool::SteerAck::Err(
|
||||
crate::pool::SteerError::AgentError { code, message },
|
||||
)
|
||||
} else {
|
||||
crate::pool::SteerAck::Success
|
||||
};
|
||||
let _ = ack_tx.send(ack);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if *id == serde_json::json!(expected_id) {
|
||||
if let Some(error) = msg.get("error") {
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ = ack_tx
|
||||
.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
return Err(AcpError::AgentError(error.to_string()));
|
||||
}
|
||||
if let Some((_, ack_tx)) = pending_steer.take() {
|
||||
let _ =
|
||||
ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
|
||||
}
|
||||
return Ok(msg["result"].clone());
|
||||
}
|
||||
return Ok(msg["result"].clone());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -897,17 +1171,6 @@ impl AcpClient {
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(_elapsed) => {
|
||||
// Classification was determined before sleeping — not
|
||||
// affected by scheduler jitter between deadline and wakeup.
|
||||
if idle_fires_first {
|
||||
tracing::warn!("idle timeout ({idle_timeout:?}) — no agent activity");
|
||||
return Err(AcpError::IdleTimeout(idle_timeout));
|
||||
} else {
|
||||
tracing::warn!("hard turn timeout exceeded");
|
||||
return Err(AcpError::HardTimeout);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -918,7 +1181,12 @@ impl AcpClient {
|
||||
/// Returns `true` if the update indicates a tool call started, signaling that
|
||||
/// the idle clock should be explicitly reset (the agent will be silent while
|
||||
/// the tool executes).
|
||||
fn handle_session_update(&self, msg: &serde_json::Value) -> bool {
|
||||
///
|
||||
/// Takes `&mut self` (not `&self`) because some updates carry agent state
|
||||
/// the client must observe — notably goose's `session_info_update` with
|
||||
/// `_meta.goose.activeRunId`, which seeds [`active_run_id`](Self::active_run_id)
|
||||
/// so callers can target `_goose/unstable/session/steer` at the correct run.
|
||||
fn handle_session_update(&mut self, msg: &serde_json::Value) -> bool {
|
||||
let update = &msg["params"]["update"];
|
||||
let update_type = update
|
||||
.get("sessionUpdate")
|
||||
@@ -978,6 +1246,42 @@ impl AcpClient {
|
||||
);
|
||||
false
|
||||
}
|
||||
"session_info_update" => {
|
||||
// Goose-only as of writing: `_meta.goose.activeRunId` carries
|
||||
// the id of the currently-active prompt run, or `null` when
|
||||
// the run has cleared. Other agents don't emit this field;
|
||||
// for them `active_run_id` stays `None` and steer callers
|
||||
// will fall back to cancel+merge.
|
||||
//
|
||||
// Per the ACP `SessionInfoUpdate` schema, `_meta` is a field
|
||||
// on the update object itself — nested inside `update`, not
|
||||
// alongside it at the params level. Goose and buzz-agent both
|
||||
// emit it at `params.update._meta.goose.activeRunId`.
|
||||
let meta = msg["params"]["update"]
|
||||
.get("_meta")
|
||||
.and_then(|m| m.get("goose"));
|
||||
if let Some(goose_meta) = meta {
|
||||
match goose_meta.get("activeRunId") {
|
||||
Some(serde_json::Value::String(run_id)) => {
|
||||
tracing::debug!(
|
||||
target: "acp::update",
|
||||
"session_info_update: activeRunId={run_id}"
|
||||
);
|
||||
self.active_run_id = Some(run_id.clone());
|
||||
}
|
||||
Some(serde_json::Value::Null) => {
|
||||
tracing::debug!(
|
||||
target: "acp::update",
|
||||
"session_info_update: activeRunId cleared"
|
||||
);
|
||||
self.active_run_id = None;
|
||||
}
|
||||
// Missing or non-string/null — leave state untouched.
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
false
|
||||
}
|
||||
"keepalive" => false,
|
||||
other => {
|
||||
tracing::debug!(target: "acp::update", "session/update: {other}");
|
||||
@@ -1094,6 +1398,34 @@ fn build_prompt_params(session_id: &str, prompt_blocks: &[&str]) -> serde_json::
|
||||
})
|
||||
}
|
||||
|
||||
/// Build `_goose/unstable/session/steer` params from one or more text
|
||||
/// content blocks plus the freshest `expectedRunId`.
|
||||
///
|
||||
/// Wire shape:
|
||||
/// ```json
|
||||
/// { "sessionId": "...", "expectedRunId": "...", "prompt": [{"type":"text","text":"..."}, ...] }
|
||||
/// ```
|
||||
///
|
||||
/// Called from the read-loop steer arm at write time so `expectedRunId`
|
||||
/// matches goose's *current* run (it advances on each `session/update`).
|
||||
/// See [`crate::pool::SteerRequest`] for why this is the read loop's job
|
||||
/// and not the main loop's.
|
||||
fn build_steer_params(
|
||||
session_id: &str,
|
||||
expected_run_id: &str,
|
||||
prompt_blocks: &[&str],
|
||||
) -> serde_json::Value {
|
||||
let blocks: Vec<serde_json::Value> = prompt_blocks
|
||||
.iter()
|
||||
.map(|text| serde_json::json!({ "type": "text", "text": text }))
|
||||
.collect();
|
||||
serde_json::json!({
|
||||
"sessionId": session_id,
|
||||
"expectedRunId": expected_run_id,
|
||||
"prompt": blocks,
|
||||
})
|
||||
}
|
||||
|
||||
/// Build a JSON-RPC permission response with `outcome: "selected"`.
|
||||
fn permission_response_selected(id: &serde_json::Value, option_id: &str) -> serde_json::Value {
|
||||
serde_json::json!({
|
||||
@@ -1787,6 +2119,7 @@ mod tests {
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30);
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_millis(100),
|
||||
hard_deadline,
|
||||
@@ -1805,6 +2138,7 @@ mod tests {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_secs(60),
|
||||
hard_deadline,
|
||||
@@ -1828,6 +2162,7 @@ mod tests {
|
||||
let start = std::time::Instant::now();
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_millis(200),
|
||||
hard_deadline,
|
||||
@@ -1848,6 +2183,7 @@ mod tests {
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
42,
|
||||
std::time::Duration::from_secs(2),
|
||||
hard_deadline,
|
||||
@@ -1864,6 +2200,7 @@ mod tests {
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_secs(2),
|
||||
hard_deadline,
|
||||
@@ -1891,6 +2228,7 @@ mod tests {
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
0,
|
||||
std::time::Duration::from_secs(3),
|
||||
hard_deadline,
|
||||
@@ -1906,7 +2244,7 @@ mod tests {
|
||||
let idle = std::time::Duration::from_millis(100);
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(999, idle, hard_deadline)
|
||||
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline)
|
||||
.await;
|
||||
assert!(
|
||||
matches!(result, Err(AcpError::IdleTimeout(_))),
|
||||
@@ -1914,6 +2252,61 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Hard-deadline starvation regression (Max's review gate, Eva's required test).
|
||||
///
|
||||
/// When the read-loop became a `tokio::select!` with `biased; reader →
|
||||
/// steer → sleep_until`, a continuously-ready reader arm could win every
|
||||
/// poll and starve the timer arm — silently defeating the hard-deadline
|
||||
/// guarantee. The fix is a pre-select deadline check at the top of every
|
||||
/// loop iteration; this test pins that behavior.
|
||||
///
|
||||
/// Setup: agent emits a **gapless** stream of valid JSON `session/update`
|
||||
/// notifications (no `sleep` between lines) so the reader arm is
|
||||
/// continuously ready. Each line is valid JSON, so it resets the idle
|
||||
/// clock — and we set idle ≫ hard so idle cannot fire first. With
|
||||
/// `biased; reader → steer → sleep_until`, the reader arm would win
|
||||
/// every poll and `sleep_until` would never be reached. Only the
|
||||
/// pre-select deadline check at the top of the loop can stop us.
|
||||
///
|
||||
/// Without the pre-select check, this test hangs against the infinite
|
||||
/// bash subprocess until the test harness's own outer timeout, and the
|
||||
/// returned error would never be `HardTimeout`.
|
||||
#[tokio::test]
|
||||
async fn hard_deadline_fires_under_continuous_valid_json_stream() {
|
||||
// Truly infinite, gapless stream of valid JSON. No `sleep` between
|
||||
// echoes — the reader arm is continuously ready, which is the
|
||||
// exact starvation scenario the pre-select check guards against.
|
||||
// `while :; do echo ...; done` (not a fixed-count `for`) so the
|
||||
// subprocess never naturally exits before the hard deadline,
|
||||
// regardless of how fast the host drains bash output. Without
|
||||
// this, fast hardware drains a bounded loop in < hard_deadline
|
||||
// and the reader hits EOF (`AgentExited`) before the timer fires,
|
||||
// masking whether the pre-select check actually works.
|
||||
let mut client = spawn_script(
|
||||
r#"while :; do echo '{"jsonrpc":"2.0","method":"session/update","params":{"update":{"sessionUpdate":"agent_message_chunk","content":{"text":"x"}}}}'; done"#,
|
||||
)
|
||||
.await;
|
||||
let hard = std::time::Duration::from_millis(300);
|
||||
let hard_deadline = tokio::time::Instant::now() + hard;
|
||||
let idle = std::time::Duration::from_secs(60); // idle ≫ hard
|
||||
let start = std::time::Instant::now();
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout("test", 999, idle, hard_deadline)
|
||||
.await;
|
||||
let elapsed = start.elapsed();
|
||||
assert!(
|
||||
matches!(result, Err(AcpError::HardTimeout)),
|
||||
"expected HardTimeout under gapless valid-JSON stream, got {result:?} (elapsed {elapsed:?})"
|
||||
);
|
||||
// Must fire close to the hard deadline, not late. Without the
|
||||
// pre-select check the reader arm starves sleep_until and elapsed
|
||||
// tracks the bash subprocess lifetime instead.
|
||||
assert!(
|
||||
elapsed < std::time::Duration::from_secs(2),
|
||||
"HardTimeout fired late ({elapsed:?}); reader arm may be starving sleep_until"
|
||||
);
|
||||
}
|
||||
|
||||
/// Same as `agent_request_with_matching_id_not_consumed_as_response` but
|
||||
/// exercises the non-idle `read_until_response` path (via `send_request`).
|
||||
#[tokio::test]
|
||||
@@ -1957,6 +2350,7 @@ mod tests {
|
||||
let start = std::time::Instant::now();
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_millis(100),
|
||||
hard_deadline,
|
||||
@@ -1990,6 +2384,7 @@ mod tests {
|
||||
let start = std::time::Instant::now();
|
||||
let result = client
|
||||
.read_until_response_with_idle_timeout(
|
||||
"test",
|
||||
999,
|
||||
std::time::Duration::from_millis(200),
|
||||
hard_deadline,
|
||||
@@ -2068,4 +2463,254 @@ mod tests {
|
||||
"systemPrompt should NOT be in params when value is None"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Goose-native steer scaffold (PR follow-up to #1160) ──────────────
|
||||
|
||||
/// Helper: spawn an inert `cat` subprocess so we have a real AcpClient
|
||||
/// to drive `handle_session_update` against. `cat` never writes back,
|
||||
/// which is fine — these tests don't read from the agent, they just
|
||||
/// feed JSON into the parser.
|
||||
async fn spawn_inert_client() -> AcpClient {
|
||||
AcpClient::spawn("cat", &[], &[])
|
||||
.await
|
||||
.expect("spawn cat as inert client")
|
||||
}
|
||||
|
||||
/// Build a `session/update` JSON-RPC notification carrying a
|
||||
/// `session_info_update` with the given `_meta.goose.activeRunId` value.
|
||||
/// Pass `None` to omit the `activeRunId` field entirely.
|
||||
///
|
||||
/// `_meta` is nested inside the `update` object (per the ACP
|
||||
/// `SessionInfoUpdate` schema), matching what goose and buzz-agent
|
||||
/// emit on the wire.
|
||||
fn session_info_update_msg(active_run_id: Option<serde_json::Value>) -> serde_json::Value {
|
||||
let mut goose = serde_json::Map::new();
|
||||
if let Some(v) = active_run_id {
|
||||
goose.insert("activeRunId".to_string(), v);
|
||||
}
|
||||
let mut meta = serde_json::Map::new();
|
||||
meta.insert("goose".to_string(), serde_json::Value::Object(goose));
|
||||
serde_json::json!({
|
||||
"jsonrpc": "2.0",
|
||||
"method": "session/update",
|
||||
"params": {
|
||||
"sessionId": "test-session",
|
||||
"update": {
|
||||
"sessionUpdate": "session_info_update",
|
||||
"_meta": serde_json::Value::Object(meta),
|
||||
},
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_run_id_sets_on_string() {
|
||||
let mut client = spawn_inert_client().await;
|
||||
assert!(client.active_run_id().is_none(), "starts as None");
|
||||
|
||||
let msg = session_info_update_msg(Some(serde_json::json!("run-abc-123")));
|
||||
let _ = client.handle_session_update(&msg);
|
||||
|
||||
assert_eq!(client.active_run_id(), Some("run-abc-123"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_run_id_clears_on_null() {
|
||||
let mut client = spawn_inert_client().await;
|
||||
// Set it first
|
||||
let set_msg = session_info_update_msg(Some(serde_json::json!("run-xyz")));
|
||||
let _ = client.handle_session_update(&set_msg);
|
||||
assert_eq!(client.active_run_id(), Some("run-xyz"));
|
||||
|
||||
// Then clear with explicit null
|
||||
let clear_msg = session_info_update_msg(Some(serde_json::Value::Null));
|
||||
let _ = client.handle_session_update(&clear_msg);
|
||||
assert!(
|
||||
client.active_run_id().is_none(),
|
||||
"explicit null must clear active_run_id"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_run_id_untouched_when_missing() {
|
||||
// Field absent entirely — must NOT clear existing state (only an
|
||||
// explicit null clears; missing means "no new info this update").
|
||||
let mut client = spawn_inert_client().await;
|
||||
let set_msg = session_info_update_msg(Some(serde_json::json!("run-stable")));
|
||||
let _ = client.handle_session_update(&set_msg);
|
||||
assert_eq!(client.active_run_id(), Some("run-stable"));
|
||||
|
||||
// session_info_update with no activeRunId field — leave state alone.
|
||||
let missing_msg = session_info_update_msg(None);
|
||||
let _ = client.handle_session_update(&missing_msg);
|
||||
assert_eq!(
|
||||
client.active_run_id(),
|
||||
Some("run-stable"),
|
||||
"missing activeRunId must leave state untouched"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn active_run_id_untouched_on_wrong_type() {
|
||||
// A number or object in activeRunId is malformed — neither set nor clear.
|
||||
let mut client = spawn_inert_client().await;
|
||||
let set_msg = session_info_update_msg(Some(serde_json::json!("run-stable")));
|
||||
let _ = client.handle_session_update(&set_msg);
|
||||
assert_eq!(client.active_run_id(), Some("run-stable"));
|
||||
|
||||
let wrong_type_msg = session_info_update_msg(Some(serde_json::json!(42)));
|
||||
let _ = client.handle_session_update(&wrong_type_msg);
|
||||
assert_eq!(
|
||||
client.active_run_id(),
|
||||
Some("run-stable"),
|
||||
"non-string/non-null activeRunId must leave state untouched"
|
||||
);
|
||||
}
|
||||
|
||||
// ── Goose-native steer arm tests ──────────────────────────────────────
|
||||
//
|
||||
// These exercise the seam between `install_steer_rx` and the read
|
||||
// loop's steer arm, isolated from `AgentPool` / `EventQueue` /
|
||||
// dispatch. They prove the locked Option-X contract at the read-loop
|
||||
// boundary:
|
||||
// 1. With `active_run_id == None`, the steer arm acks
|
||||
// `Err(ExpectedRunIdMissing)` and writes nothing — the main
|
||||
// loop's "Err-before-pending" fallback path is reachable.
|
||||
// 2. With `active_run_id` set, the steer arm writes the JSON-RPC
|
||||
// request with the matching `expectedRunId` and routes the
|
||||
// response to the ack oneshot as `Success`.
|
||||
//
|
||||
// We don't test the full mode-gate fork here — that lives in lib.rs
|
||||
// and is covered by goose e2e (Eva's lane).
|
||||
|
||||
/// Steer with no `active_run_id` set acks `ExpectedRunIdMissing`
|
||||
/// without writing anything. The read loop continues normally and
|
||||
/// eventually hits the idle timeout (which is fine — we just need to
|
||||
/// observe the ack).
|
||||
#[tokio::test]
|
||||
async fn native_steer_with_no_active_run_id_acks_expected_run_id_missing() {
|
||||
// Quiet process: never emits anything, so the read loop has only
|
||||
// the steer arm and the idle timeout to consider.
|
||||
let mut client = spawn_script("sleep 10").await;
|
||||
assert!(
|
||||
client.active_run_id().is_none(),
|
||||
"precondition: active_run_id starts as None"
|
||||
);
|
||||
|
||||
let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::<crate::pool::SteerRequest>(1);
|
||||
client.install_steer_rx(steer_rx);
|
||||
|
||||
// Fire-and-forget: send a SteerRequest from a separate task so
|
||||
// the read loop picks it up via the select! arm.
|
||||
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<crate::pool::SteerAck>();
|
||||
let send_task = tokio::spawn(async move {
|
||||
steer_tx
|
||||
.send(crate::pool::SteerRequest {
|
||||
prompt_blocks: vec!["test steer body".into()],
|
||||
ack_tx,
|
||||
})
|
||||
.await
|
||||
.expect("steer_tx send should succeed");
|
||||
});
|
||||
|
||||
// Drive the read loop with short idle timeout so the test
|
||||
// doesn't hang. The expected_id is intentionally never going to
|
||||
// be matched (the script writes nothing); the read loop will
|
||||
// exit via IdleTimeout shortly after the steer arm fires.
|
||||
let idle = std::time::Duration::from_millis(500);
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
let read_result = client
|
||||
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline)
|
||||
.await;
|
||||
send_task.await.expect("send_task should complete");
|
||||
|
||||
// Read loop exit shape: IdleTimeout (no agent activity).
|
||||
assert!(
|
||||
matches!(read_result, Err(AcpError::IdleTimeout(_))),
|
||||
"expected IdleTimeout once steer was acked + script stayed silent, got {read_result:?}"
|
||||
);
|
||||
|
||||
// Ack must be ExpectedRunIdMissing — the steer arm bailed out
|
||||
// without writing because active_run_id was None at write time.
|
||||
let ack = ack_rx
|
||||
.await
|
||||
.expect("ack oneshot must have received a SteerAck");
|
||||
match ack {
|
||||
crate::pool::SteerAck::Err(crate::pool::SteerError::ExpectedRunIdMissing) => {}
|
||||
other => panic!("expected SteerAck::Err(ExpectedRunIdMissing), got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// Steer with `active_run_id` set writes the JSON-RPC request and
|
||||
/// routes the matching response to the ack oneshot as `Success`.
|
||||
/// Verifies the wire shape (`sessionId` + `expectedRunId` + `prompt`)
|
||||
/// indirectly: the bash script emits a response keyed by the steer
|
||||
/// id (0), and `Success` only fires if the read loop matched that
|
||||
/// id to its `pending_steer` entry.
|
||||
#[tokio::test]
|
||||
async fn native_steer_with_active_run_id_routes_response_to_ack() {
|
||||
// Script: pause briefly so the test task can install the steer
|
||||
// and we can be sure the response doesn't race ahead of the
|
||||
// write — then emit the steer response (id=0 because next_id
|
||||
// starts at 0 and the steer is the first request the read loop
|
||||
// writes), then idle. This is a JSON-RPC success response with
|
||||
// a `stopReason` payload (matching the shape goose uses for
|
||||
// steer responses in fake_llm.rs).
|
||||
let script = "sleep 0.5; \
|
||||
echo '{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{\"stopReason\":\"end_turn\"}}'; \
|
||||
sleep 10";
|
||||
let mut client = spawn_script(script).await;
|
||||
|
||||
// Set active_run_id via a synthesized session_info_update so the
|
||||
// steer arm has a non-None value to read at write time.
|
||||
let update = session_info_update_msg(Some(serde_json::json!("run-42")));
|
||||
let _ = client.handle_session_update(&update);
|
||||
assert_eq!(client.active_run_id(), Some("run-42"));
|
||||
|
||||
let (steer_tx, steer_rx) = tokio::sync::mpsc::channel::<crate::pool::SteerRequest>(1);
|
||||
client.install_steer_rx(steer_rx);
|
||||
|
||||
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<crate::pool::SteerAck>();
|
||||
let send_task = tokio::spawn(async move {
|
||||
steer_tx
|
||||
.send(crate::pool::SteerRequest {
|
||||
prompt_blocks: vec!["test steer body".into()],
|
||||
ack_tx,
|
||||
})
|
||||
.await
|
||||
.expect("steer_tx send should succeed");
|
||||
});
|
||||
|
||||
// Drive the read loop. Expected_id 999 will never be emitted by
|
||||
// the script so the read loop exits via idle timeout after the
|
||||
// steer response is routed to ack.
|
||||
let idle = std::time::Duration::from_secs(2);
|
||||
let hard_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let read_result = client
|
||||
.read_until_response_with_idle_timeout("sess-test", 999, idle, hard_deadline)
|
||||
.await;
|
||||
send_task.await.expect("send_task should complete");
|
||||
|
||||
// Read loop exit: IdleTimeout (no further activity after the
|
||||
// routed steer response). AgentExited would also be a valid
|
||||
// exit if the bash script terminated early; either is fine —
|
||||
// what matters is the ack.
|
||||
assert!(
|
||||
matches!(
|
||||
read_result,
|
||||
Err(AcpError::IdleTimeout(_)) | Err(AcpError::AgentExited)
|
||||
),
|
||||
"expected IdleTimeout or AgentExited after steer ack, got {read_result:?}"
|
||||
);
|
||||
|
||||
// Ack must be Success: the steer response (id=0) was routed to
|
||||
// pending_steer.ack_tx.
|
||||
let ack = ack_rx
|
||||
.await
|
||||
.expect("ack oneshot must have received a SteerAck");
|
||||
match ack {
|
||||
crate::pool::SteerAck::Success => {}
|
||||
other => panic!("expected SteerAck::Success, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -57,8 +57,16 @@ pub enum MultipleEventHandling {
|
||||
/// Queue new events while a turn is in-flight. Deliver after current turn
|
||||
/// completes. Existing behavior — zero code change in this path.
|
||||
Queue,
|
||||
/// Cancel the in-flight turn and re-dispatch a merged prompt that frames
|
||||
/// the new events as a **steering message** — one that arrived while the
|
||||
/// agent was working, to be woven into the in-progress task rather than
|
||||
/// treated as a replacement. Fires for any author the inbound author gate
|
||||
/// admits (owner ∪ allowlist ∪ siblings). This is the default mid-turn
|
||||
/// delivery path. Requires DedupMode::Queue.
|
||||
Steer,
|
||||
/// Cancel the in-flight turn and re-dispatch a merged prompt combining
|
||||
/// the original events with the new ones, for ANY new @mention.
|
||||
/// the original events with the new ones, framed as a **supersede** (the
|
||||
/// new request replaces the old), for ANY new @mention.
|
||||
/// Requires DedupMode::Queue.
|
||||
Interrupt,
|
||||
/// Cancel the in-flight turn only when the new @mention is from the agent
|
||||
@@ -290,12 +298,15 @@ pub struct CliArgs {
|
||||
pub dedup: DedupMode,
|
||||
|
||||
/// How to handle new @mentions while a turn is already in-flight.
|
||||
/// queue: events wait (default). interrupt: cancel+re-prompt on any mention.
|
||||
/// owner-interrupt: cancel only for agent owner's mentions.
|
||||
/// steer (default): cancel+re-prompt, framing the new mention as a message
|
||||
/// that arrived mid-task — the agent keeps working and weaves it in.
|
||||
/// queue: events wait until the current turn completes.
|
||||
/// interrupt: cancel+re-prompt framed as a supersede (new replaces old).
|
||||
/// owner-interrupt: interrupt only for the agent owner's mentions.
|
||||
#[arg(
|
||||
long,
|
||||
env = "BUZZ_ACP_MULTIPLE_EVENT_HANDLING",
|
||||
default_value = "queue",
|
||||
default_value = "steer",
|
||||
value_enum
|
||||
)]
|
||||
pub multiple_event_handling: MultipleEventHandling,
|
||||
@@ -503,6 +514,33 @@ fn validate_allowlist(entries: &[String]) -> Result<HashSet<String>, ConfigError
|
||||
Ok(validated)
|
||||
}
|
||||
|
||||
/// Validate the `--multiple-event-handling` / `--dedup` combination.
|
||||
///
|
||||
/// Every mid-turn cancel mode (`Steer`, `Interrupt`, `OwnerInterrupt`) requires
|
||||
/// `DedupMode::Queue`: `DedupMode::Drop` discards events during the cancel drain
|
||||
/// window, which would produce incomplete merged prompts. `Queue` handling
|
||||
/// imposes no constraint.
|
||||
fn validate_multiple_event_handling(
|
||||
handling: MultipleEventHandling,
|
||||
dedup: DedupMode,
|
||||
) -> Result<(), ConfigError> {
|
||||
let is_cancel_mode = matches!(
|
||||
handling,
|
||||
MultipleEventHandling::Steer
|
||||
| MultipleEventHandling::Interrupt
|
||||
| MultipleEventHandling::OwnerInterrupt
|
||||
);
|
||||
if is_cancel_mode && matches!(dedup, DedupMode::Drop) {
|
||||
return Err(ConfigError::ConfigFile(
|
||||
"--multiple-event-handling=steer (or interrupt/owner-interrupt) requires \
|
||||
--dedup=queue. DedupMode::Drop discards events during the cancel drain window, \
|
||||
producing incomplete merged prompts."
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn normalize_agent_command_identity(command: &str) -> String {
|
||||
let normalized = command.trim().replace('\\', "/");
|
||||
let trimmed = normalized.trim_end_matches('/');
|
||||
@@ -898,18 +936,7 @@ impl Config {
|
||||
}
|
||||
let model = args.model.or(persona_model);
|
||||
|
||||
if matches!(
|
||||
args.multiple_event_handling,
|
||||
MultipleEventHandling::Interrupt | MultipleEventHandling::OwnerInterrupt
|
||||
) && matches!(args.dedup, DedupMode::Drop)
|
||||
{
|
||||
return Err(ConfigError::ConfigFile(
|
||||
"--multiple-event-handling=interrupt (or owner-interrupt) requires --dedup=queue. \
|
||||
DedupMode::Drop discards events during the cancel drain window, \
|
||||
producing incomplete merged prompts."
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
validate_multiple_event_handling(args.multiple_event_handling, args.dedup)?;
|
||||
|
||||
let config = Config {
|
||||
keys,
|
||||
@@ -2336,6 +2363,61 @@ channels = "ALL"
|
||||
assert!(result.is_empty());
|
||||
}
|
||||
|
||||
// ── Multiple-event-handling validation + default ──────────────────────────
|
||||
|
||||
#[test]
|
||||
fn test_multiple_event_handling_default_is_steer() {
|
||||
// Parse a minimal arg set; the default for --multiple-event-handling
|
||||
// must be `steer` (steering is the default mid-turn delivery path).
|
||||
let args = CliArgs::parse_from(["buzz-acp", "--private-key", &"0".repeat(64)]);
|
||||
assert_eq!(args.multiple_event_handling, MultipleEventHandling::Steer);
|
||||
// Dedup default must remain `queue` so steering's requirement is met.
|
||||
assert!(matches!(args.dedup, DedupMode::Queue));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_validate_steer_requires_queue_dedup() {
|
||||
// Steer + Drop is rejected (drain window would drop events).
|
||||
let err = validate_multiple_event_handling(MultipleEventHandling::Steer, DedupMode::Drop)
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
err.to_string().contains("requires"),
|
||||
"expected a dedup-requirement error, got: {err}"
|
||||
);
|
||||
// Steer + Queue is accepted.
|
||||
assert!(
|
||||
validate_multiple_event_handling(MultipleEventHandling::Steer, DedupMode::Queue)
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_validate_queue_handling_allows_any_dedup() {
|
||||
// The non-cancel `Queue` handling imposes no dedup constraint.
|
||||
assert!(
|
||||
validate_multiple_event_handling(MultipleEventHandling::Queue, DedupMode::Drop).is_ok()
|
||||
);
|
||||
assert!(
|
||||
validate_multiple_event_handling(MultipleEventHandling::Queue, DedupMode::Queue)
|
||||
.is_ok()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_validate_interrupt_modes_still_require_queue() {
|
||||
for mode in [
|
||||
MultipleEventHandling::Interrupt,
|
||||
MultipleEventHandling::OwnerInterrupt,
|
||||
] {
|
||||
assert!(
|
||||
validate_multiple_event_handling(mode, DedupMode::Drop).is_err(),
|
||||
"{mode:?} + Drop should be rejected"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Idle timeout constant + guard (PR #935) ───────────────────────────────
|
||||
|
||||
#[test]
|
||||
fn default_idle_timeout_is_900_seconds() {
|
||||
// Lock the constant value so accidental changes are caught.
|
||||
|
||||
+474
-27
@@ -32,7 +32,7 @@ use pool::{
|
||||
AgentPool, ControlSignal, OwnedAgent, PromptContext, PromptOutcome, PromptResult, PromptSource,
|
||||
SessionState,
|
||||
};
|
||||
use queue::{EventQueue, QueuedEvent, ThreadTags};
|
||||
use queue::{CancelReason, EventQueue, QueuedEvent, ThreadTags};
|
||||
use relay::{HarnessRelay, RelayEventPublisher};
|
||||
use tokio::sync::{mpsc, watch};
|
||||
use tracing_subscriber::EnvFilter;
|
||||
@@ -879,7 +879,30 @@ fn any_respawn_in_flight(crash_history: &[SlotCircuit]) -> bool {
|
||||
/// Result of a background respawn task.
|
||||
struct RespawnResult {
|
||||
index: usize,
|
||||
result: Result<(AcpClient, u32)>,
|
||||
/// Tuple: (initialized client, protocol version, supports_goose_steer).
|
||||
/// The third element is always `true` — the supervisor uses
|
||||
/// try-and-tolerate for the steer extension.
|
||||
result: Result<(AcpClient, u32, bool)>,
|
||||
}
|
||||
|
||||
/// Outcome of a non-cancelling steer attempt, forwarded from a per-attempt
|
||||
/// watcher task (which awaits the `SteerRequest.ack_tx` oneshot) back to
|
||||
/// the main loop's `select!`. The main loop drives queue side-effects from
|
||||
/// this — it cannot await the oneshot itself without blocking the relay
|
||||
/// stream.
|
||||
///
|
||||
/// Carries enough identity to operate on the right withheld event in
|
||||
/// `EventQueue::withheld_native_steer`: `channel_id` is the routing key,
|
||||
/// `event_id` is the hex id of the single event the steer carried.
|
||||
struct SteerAckEvent {
|
||||
channel_id: Uuid,
|
||||
event_id: String,
|
||||
/// `Ok` if the read loop sent any of the locked `SteerAck` variants.
|
||||
/// `Err` if the oneshot was dropped without a send — should not happen
|
||||
/// under the current read-loop drains, but if it ever does the main
|
||||
/// loop treats it as `PromptCompletedNeutral` (release withheld, no
|
||||
/// fallback signal) to avoid leaking the withheld event.
|
||||
ack: std::result::Result<pool::SteerAck, tokio::sync::oneshot::error::RecvError>,
|
||||
}
|
||||
|
||||
/// RAII guard that ensures a `RespawnResult` is sent even if the task panics.
|
||||
@@ -903,7 +926,7 @@ impl RespawnGuard {
|
||||
/// Send the result and disarm the guard. Uses `try_send` (sync) so there
|
||||
/// is no await boundary between marking `sent` and actually enqueueing —
|
||||
/// cancellation cannot slip between the two.
|
||||
fn send(mut self, result: Result<(AcpClient, u32)>) {
|
||||
fn send(mut self, result: Result<(AcpClient, u32, bool)>) {
|
||||
// Invariant: try_send succeeds because the channel capacity equals the
|
||||
// slot count, and respawn_in_flight guarantees at most one outstanding
|
||||
// result per slot. If this ever fails, the channel sizing or the
|
||||
@@ -1022,6 +1045,16 @@ async fn tokio_main() -> Result<()> {
|
||||
tracing::info!(agent = i, "agent initialized: {init_result}");
|
||||
let protocol_version =
|
||||
init_result["protocolVersion"].as_u64().unwrap_or(1) as u32;
|
||||
tracing::info!(
|
||||
agent = i,
|
||||
name = init_result
|
||||
.get("agentInfo")
|
||||
.or_else(|| init_result.get("serverInfo"))
|
||||
.and_then(|info| info.get("name"))
|
||||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("unknown"),
|
||||
"agent initialized — non-cancelling steer enabled (try-and-tolerate)"
|
||||
);
|
||||
acp.observe(
|
||||
"agent_initialized",
|
||||
serde_json::json!({
|
||||
@@ -1331,6 +1364,19 @@ async fn tokio_main() -> Result<()> {
|
||||
// JoinSet for respawn tasks so shutdown can abort them.
|
||||
let mut respawn_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();
|
||||
|
||||
// Channel for non-cancelling steer ack watchers to forward outcomes back
|
||||
// to the main loop. Each `pool.send_steer(...) == Ok(())` spawns a
|
||||
// short-lived task that awaits the `SteerRequest.ack_tx` oneshot and
|
||||
// forwards a `SteerAckEvent`. Unbounded because:
|
||||
// 1. The producer count is bounded by in-flight goose turns
|
||||
// (`agents` slots, capacity-1 `steer_tx` each), so the channel
|
||||
// cannot legitimately back up under steady state.
|
||||
// 2. We must never drop a steer outcome — losing an ack would leak a
|
||||
// withheld event in `EventQueue::withheld_native_steer` until
|
||||
// `IN_FLIGHT_DEADLINE_SECS` expires.
|
||||
let (steer_ack_tx, mut steer_ack_rx) = mpsc::unbounded_channel::<SteerAckEvent>();
|
||||
|
||||
// ── Step 7: Shutdown signal ───────────────────────────────────────────────
|
||||
let (shutdown_tx, mut shutdown_rx) = watch::channel(());
|
||||
|
||||
let tx = shutdown_tx.clone();
|
||||
@@ -1402,6 +1448,7 @@ async fn tokio_main() -> Result<()> {
|
||||
enum PoolEvent {
|
||||
Result(Box<PromptResult>),
|
||||
Panic(tokio::task::JoinError),
|
||||
SteerAck(SteerAckEvent),
|
||||
}
|
||||
|
||||
loop {
|
||||
@@ -1448,7 +1495,7 @@ async fn tokio_main() -> Result<()> {
|
||||
while let Ok(rr) = respawn_rx.try_recv() {
|
||||
crash_history[rr.index].respawn_in_flight = false;
|
||||
match rr.result {
|
||||
Ok((acp, protocol_version)) => {
|
||||
Ok((acp, protocol_version, _)) => {
|
||||
let agent = OwnedAgent {
|
||||
index: rr.index,
|
||||
acp,
|
||||
@@ -1496,6 +1543,14 @@ async fn tokio_main() -> Result<()> {
|
||||
Some(Err(e)) = join_set.join_next(), if !join_set.is_empty() => {
|
||||
Some(PoolEvent::Panic(e))
|
||||
}
|
||||
// Goose-native steer ack from a watcher task. Outcomes drive
|
||||
// queue side-effects (drop / release withheld event) and
|
||||
// optionally the cancel+merge fallback signal. See the
|
||||
// `Some(PoolEvent::SteerAck(...))` match arm below for the
|
||||
// locked semantics (Eva + Max + Perci).
|
||||
Some(ack_event) = steer_ack_rx.recv() => {
|
||||
Some(PoolEvent::SteerAck(ack_event))
|
||||
}
|
||||
control_event = async {
|
||||
match relay_observer_control_rx.as_mut() {
|
||||
Some(rx) => rx.recv().await,
|
||||
@@ -1784,6 +1839,18 @@ async fn tokio_main() -> Result<()> {
|
||||
// buzz_event.event (needed for mode gate below).
|
||||
let author_hex = buzz_event.event.pubkey.to_hex();
|
||||
let event_id_hex = buzz_event.event.id.to_hex();
|
||||
// Clone for the non-cancelling steer fork, which
|
||||
// needs the event to render the steer body. The
|
||||
// clone is unconditional because we don't know
|
||||
// yet whether the mode gate will demand a steer
|
||||
// — checking `multiple_event_handling` here
|
||||
// would couple the queueing path to the mode
|
||||
// and break the existing invariant that every
|
||||
// accepted event goes through `queue.push`
|
||||
// first. `nostr::Event::clone` is cheap (Arc-
|
||||
// backed payload) so the cost is negligible.
|
||||
let event_for_steer = buzz_event.event.clone();
|
||||
let prompt_tag_for_steer = prompt_tag.clone();
|
||||
let accepted = queue.push(QueuedEvent {
|
||||
channel_id: buzz_event.channel_id,
|
||||
event: buzz_event.event,
|
||||
@@ -1797,29 +1864,53 @@ async fn tokio_main() -> Result<()> {
|
||||
// cosmetic stale 👀. Acceptable — see ReactionGuard docs.
|
||||
if accepted {
|
||||
let rc = ctx.rest_client.clone();
|
||||
let eid = event_id_hex.clone();
|
||||
tokio::spawn(async move {
|
||||
pool::reaction_add(&rc, &event_id_hex, "👀").await;
|
||||
pool::reaction_add(&rc, &eid, "👀").await;
|
||||
});
|
||||
}
|
||||
// Event is already queued. If mode requires it AND
|
||||
// the channel has an in-flight task, fire cancel.
|
||||
// the channel has an in-flight task, fire cancel —
|
||||
// OR take the non-cancelling (ACP steer) fork for Steer signals.
|
||||
if accepted && queue.is_channel_in_flight(buzz_event.channel_id) {
|
||||
let should_cancel = match config.multiple_event_handling {
|
||||
MultipleEventHandling::Queue => false,
|
||||
MultipleEventHandling::Interrupt => true,
|
||||
MultipleEventHandling::OwnerInterrupt => {
|
||||
match owner_cache.get() {
|
||||
Some(o) => author_hex == *o,
|
||||
None => false,
|
||||
}
|
||||
// Author eligibility (owner ∪ allowlist ∪ siblings)
|
||||
// is already enforced by the inbound author gate
|
||||
// above, so the mid-turn signal fires for every
|
||||
// event that reaches here.
|
||||
let signal = mode_gate_signal(
|
||||
config.multiple_event_handling,
|
||||
&author_hex,
|
||||
owner_cache.get(),
|
||||
);
|
||||
if let Some(signal) = signal {
|
||||
// Try-and-tolerate fork: when the mode
|
||||
// wants a Steer, attempt the non-cancelling
|
||||
// path first for any agent. On accept,
|
||||
// withhold the queued event and spawn an
|
||||
// ack watcher; the main loop's
|
||||
// `PoolEvent::SteerAck` arm decides
|
||||
// success/release/fallback. On reject
|
||||
// (including `-32601 method_not_found`
|
||||
// from agents that don't implement the
|
||||
// extension), fall through to the universal
|
||||
// cancel+merge `Steer` signal so the event
|
||||
// still reaches the agent.
|
||||
let native_attempted = matches!(signal, ControlSignal::Steer)
|
||||
&& try_native_steer(
|
||||
&mut pool,
|
||||
&mut queue,
|
||||
buzz_event.channel_id,
|
||||
event_for_steer,
|
||||
prompt_tag_for_steer,
|
||||
&steer_ack_tx,
|
||||
);
|
||||
if !native_attempted {
|
||||
signal_in_flight_task(
|
||||
&mut pool,
|
||||
buzz_event.channel_id,
|
||||
signal,
|
||||
);
|
||||
}
|
||||
};
|
||||
if should_cancel {
|
||||
signal_in_flight_task(
|
||||
&mut pool,
|
||||
buzz_event.channel_id,
|
||||
ControlSignal::Interrupt,
|
||||
);
|
||||
}
|
||||
}
|
||||
for (channel_id, thread_tags) in
|
||||
@@ -1973,6 +2064,123 @@ async fn tokio_main() -> Result<()> {
|
||||
typing_channels.insert(channel_id, thread_tags);
|
||||
}
|
||||
}
|
||||
Some(PoolEvent::SteerAck(SteerAckEvent {
|
||||
channel_id,
|
||||
event_id,
|
||||
ack,
|
||||
})) => {
|
||||
// Goose-native steer attempt resolved. Locked semantics
|
||||
// (Eva + Max + Perci, unanimous on Option X):
|
||||
//
|
||||
// Success
|
||||
// The agent received the steer via the non-cancelling
|
||||
// path. Drop the withheld event so normal dispatch
|
||||
// never redelivers it.
|
||||
//
|
||||
// Err(_) where the write never landed (Transport /
|
||||
// ExpectedRunIdMissing):
|
||||
// Delivery state of the underlying message is "never
|
||||
// attempted on the wire". Release withheld back to the
|
||||
// queue front AND issue the cancel+merge fallback so
|
||||
// the message still reaches the agent.
|
||||
//
|
||||
// Err(AgentError { code: -32601, .. })
|
||||
// The agent returned method_not_found — it does not
|
||||
// implement the steer extension. Release withheld AND
|
||||
// fire the cancel+merge fallback so the message still
|
||||
// reaches the agent via the universal path.
|
||||
//
|
||||
// Err(AgentError { code: other, .. })
|
||||
// The write landed and the agent returned a JSON-RPC
|
||||
// error at the application level (e.g. wrong run id).
|
||||
// The agent's turn is still running (or just completed).
|
||||
// Release withheld for normal dispatch; do NOT fire the
|
||||
// fallback signal — the agent already saw the steer
|
||||
// attempt. If the turn is still running, normal dispatch
|
||||
// re-delivers when it completes. If the turn already
|
||||
// ended, there is nothing to cancel.
|
||||
//
|
||||
// PromptCompletedNeutral
|
||||
// The read loop wrote the steer (or was preparing to)
|
||||
// but the prompt completed before the response landed.
|
||||
// Delivery state is unknown — but the prompt completing
|
||||
// means there is no in-flight turn to signal anymore.
|
||||
// Release withheld for normal dispatch; do NOT fire
|
||||
// the fallback signal (it would target a turn that
|
||||
// just ended; normal dispatch already handles
|
||||
// redelivery via the released queue entry).
|
||||
//
|
||||
// Err(PromptCompleted)
|
||||
// `SteerError::PromptCompleted` is returned synchronously
|
||||
// by `pool::send_steer` when no task is in flight (handled
|
||||
// in `try_native_steer`'s Err branch, which falls through
|
||||
// to cancel+merge). It is never routed through the ack
|
||||
// channel, so this variant never appears in `SteerAckEvent`.
|
||||
//
|
||||
// Watcher Err (oneshot dropped)
|
||||
// Should not happen — the read loop drains
|
||||
// pending_steer on every return path. If it does,
|
||||
// treat as PromptCompletedNeutral to avoid leaking
|
||||
// the withheld event in `withheld_native_steer`.
|
||||
let (release_withheld, drop_withheld, signal_fallback) = match &ack {
|
||||
Ok(pool::SteerAck::Success) => (false, true, false),
|
||||
// -32601 = method_not_found: agent does not implement the
|
||||
// steer extension. Fire cancel+merge so the message still
|
||||
// reaches the agent.
|
||||
Ok(pool::SteerAck::Err(pool::SteerError::AgentError { code, .. }))
|
||||
if *code == -32601 =>
|
||||
{
|
||||
(true, false, true)
|
||||
}
|
||||
// AgentError: write landed, agent rejected it at the
|
||||
// application level (e.g. wrong run id). Release for
|
||||
// normal dispatch; no fallback signal (the turn is still
|
||||
// running or just ended — either way there is nothing to
|
||||
// cancel).
|
||||
Ok(pool::SteerAck::Err(pool::SteerError::AgentError { .. })) => {
|
||||
(true, false, false)
|
||||
}
|
||||
// Transport / ExpectedRunIdMissing: write never landed.
|
||||
// Release and fire the cancel+merge fallback so the
|
||||
// message still reaches the agent.
|
||||
Ok(pool::SteerAck::Err(_)) => (true, false, true),
|
||||
Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false),
|
||||
Err(_recv_err) => (true, false, false),
|
||||
};
|
||||
tracing::info!(
|
||||
channel = %channel_id,
|
||||
event_id = %event_id,
|
||||
?ack,
|
||||
release_withheld,
|
||||
drop_withheld,
|
||||
signal_fallback,
|
||||
"non-cancelling steer ack received"
|
||||
);
|
||||
if drop_withheld {
|
||||
queue.remove_event(channel_id, &event_id);
|
||||
}
|
||||
if release_withheld {
|
||||
queue.release_native_steer(channel_id, &event_id);
|
||||
}
|
||||
if signal_fallback {
|
||||
// Universal cancel+merge fallback. Note: the
|
||||
// queued event has already been released to the
|
||||
// front of `queues[channel_id]`, so the cancel
|
||||
// will pick it up as part of the merged batch and
|
||||
// re-prompt the agent.
|
||||
signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer);
|
||||
}
|
||||
// After releasing a withheld event, give dispatch a chance
|
||||
// to re-flush. If the prompt is still in flight, the
|
||||
// channel stays `in_flight_channels` and `flush_next`
|
||||
// skips it — but a Steer fallback signal sent above will
|
||||
// tear down the in-flight task; on its completion the
|
||||
// queue drains. We still try here in case the in-flight
|
||||
// task has already returned.
|
||||
for (channel_id, thread_tags) in dispatch_pending(&mut pool, &mut queue, &ctx) {
|
||||
typing_channels.insert(channel_id, thread_tags);
|
||||
}
|
||||
}
|
||||
None => {} // relay/heartbeat/shutdown branches handled inline above
|
||||
}
|
||||
}
|
||||
@@ -2040,7 +2248,7 @@ async fn tokio_main() -> Result<()> {
|
||||
// Drain any respawn results that completed before the abort. Explicitly
|
||||
// shut down returned agents instead of relying on AcpClient::Drop.
|
||||
while let Ok(rr) = respawn_rx.try_recv() {
|
||||
if let Ok((mut acp, _)) = rr.result {
|
||||
if let Ok((mut acp, _, _)) = rr.result {
|
||||
acp.shutdown().await;
|
||||
tracing::debug!(agent = rr.index, "reaped respawned agent on shutdown");
|
||||
}
|
||||
@@ -2101,6 +2309,34 @@ fn is_owner_control_command(
|
||||
&& event_mentions_agent(event, agent_pubkey_hex)
|
||||
}
|
||||
|
||||
// ── signal_in_flight_task ─────────────────────────────────────────────────────
|
||||
|
||||
/// Decide which [`ControlSignal`] (if any) to send to an in-flight turn when a
|
||||
/// new, already-author-gated event arrives for that channel.
|
||||
///
|
||||
/// Returns `None` to leave the in-flight turn untouched (the event waits in the
|
||||
/// queue and is delivered when the turn completes). Author eligibility — owner
|
||||
/// ∪ allowlist ∪ siblings — is enforced upstream by the inbound author gate, so
|
||||
/// `Steer`/`Interrupt` apply to every event that reaches this point; only
|
||||
/// `OwnerInterrupt` re-checks authorship (owner-only) here.
|
||||
///
|
||||
/// `owner` is the resolved owner pubkey hex, if known.
|
||||
fn mode_gate_signal(
|
||||
handling: MultipleEventHandling,
|
||||
author_hex: &str,
|
||||
owner: Option<&str>,
|
||||
) -> Option<ControlSignal> {
|
||||
match handling {
|
||||
MultipleEventHandling::Queue => None,
|
||||
MultipleEventHandling::Steer => Some(ControlSignal::Steer),
|
||||
MultipleEventHandling::Interrupt => Some(ControlSignal::Interrupt),
|
||||
MultipleEventHandling::OwnerInterrupt => match owner {
|
||||
Some(o) if author_hex == o => Some(ControlSignal::Interrupt),
|
||||
_ => None,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a control signal to the in-flight task for `channel_id`.
|
||||
/// Returns `true` if a signal was sent, `false` if no in-flight task was found.
|
||||
fn signal_in_flight_task(
|
||||
@@ -2123,6 +2359,115 @@ fn signal_in_flight_task(
|
||||
false
|
||||
}
|
||||
|
||||
/// Attempt the non-cancelling (ACP) steer for a freshly-queued event.
|
||||
///
|
||||
/// Caller invariants:
|
||||
/// - `event` has already been pushed into `EventQueue::queues[channel_id]`
|
||||
/// via [`EventQueue::push`] — its `event.id` must still be locatable
|
||||
/// there so [`EventQueue::mark_native_steer_pending`] can move it to the
|
||||
/// side table.
|
||||
/// - `multiple_event_handling` resolved to `ControlSignal::Steer`; this
|
||||
/// function is the non-cancelling fork of that signal.
|
||||
///
|
||||
/// Returns `true` if the native attempt was accepted by the read loop
|
||||
/// (capacity-1 mpsc `try_send` succeeded, event withheld synchronously,
|
||||
/// ack watcher spawned). On `true` the caller MUST NOT issue the
|
||||
/// 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.
|
||||
///
|
||||
/// The withheld event is NOT released here on `false` because no withhold
|
||||
/// was established: `mark_native_steer_pending` only runs on `Ok(())`.
|
||||
fn try_native_steer(
|
||||
pool: &mut AgentPool,
|
||||
queue: &mut EventQueue,
|
||||
channel_id: uuid::Uuid,
|
||||
event: nostr::Event,
|
||||
prompt_tag: String,
|
||||
steer_ack_tx: &mpsc::UnboundedSender<SteerAckEvent>,
|
||||
) -> 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();
|
||||
let event_id_hex = event.id.to_hex();
|
||||
let be = queue::BatchEvent {
|
||||
event,
|
||||
prompt_tag: prompt_tag.clone(),
|
||||
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 (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<pool::SteerAck>();
|
||||
let request = pool::SteerRequest {
|
||||
prompt_blocks: vec![body],
|
||||
ack_tx,
|
||||
};
|
||||
|
||||
match pool.send_steer(channel_id, request) {
|
||||
Ok(()) => {
|
||||
// Withhold the queued event synchronously BEFORE spawning
|
||||
// the watcher: this closes the race where `mark_complete`
|
||||
// clears `in_flight_channels` and a stray `flush_next` could
|
||||
// re-deliver the event via normal dispatch. See
|
||||
// `EventQueue::mark_native_steer_pending` docs at queue.rs:606.
|
||||
let withheld = queue.mark_native_steer_pending(channel_id, &event_id_hex);
|
||||
if !withheld {
|
||||
// Race: the event was already drained out of the queue
|
||||
// before we got here (e.g. a concurrent flush picked it
|
||||
// up). The steer is on the wire; if it succeeds the
|
||||
// agent gets it via the native path AND normal
|
||||
// dispatch — duplicate delivery is benign (agent gets
|
||||
// the same message twice). Log so this is visible if it
|
||||
// ever happens in production.
|
||||
tracing::warn!(
|
||||
channel = %channel_id,
|
||||
event_id = %event_id_hex,
|
||||
"native steer accepted by read loop but event was not in queue to withhold \
|
||||
— possible duplicate delivery if steer succeeds"
|
||||
);
|
||||
}
|
||||
let ack_tx_clone = steer_ack_tx.clone();
|
||||
let event_id_for_watcher = event_id_hex.clone();
|
||||
tokio::spawn(async move {
|
||||
let ack = ack_rx.await;
|
||||
let _ = ack_tx_clone.send(SteerAckEvent {
|
||||
channel_id,
|
||||
event_id: event_id_for_watcher,
|
||||
ack,
|
||||
});
|
||||
});
|
||||
true
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::info!(
|
||||
channel = %channel_id,
|
||||
error = ?e,
|
||||
"non-cancelling steer not accepted — falling back to cancel+merge"
|
||||
);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── dispatch_pending ──────────────────────────────────────────────────────────
|
||||
|
||||
/// Flush queued work to available agents.
|
||||
fn dispatch_pending(
|
||||
pool: &mut AgentPool,
|
||||
@@ -2142,7 +2487,7 @@ fn dispatch_pending(
|
||||
.map(|event| queue::parse_thread_tags(&event.event))
|
||||
.unwrap_or_default();
|
||||
let affinity_hit = pool.has_session_for(channel_id);
|
||||
let agent = match pool.try_claim(Some(channel_id)) {
|
||||
let mut agent = match pool.try_claim(Some(channel_id)) {
|
||||
Some(a) => a,
|
||||
None => {
|
||||
let pending = queue.pending_channels();
|
||||
@@ -2163,6 +2508,19 @@ fn dispatch_pending(
|
||||
let ctx_clone = Arc::clone(ctx);
|
||||
let agent_index = agent.index;
|
||||
|
||||
// Goose-native non-cancelling steer seam: snapshot capability before
|
||||
// the agent moves into `run_prompt_task`, and install the per-turn
|
||||
// steer receiver on the read loop so the main loop's mode-gate fork
|
||||
// (see the `if accepted && queue.is_channel_in_flight(...)` block
|
||||
// in the relay event branch of the main `select!` loop) can drive
|
||||
// it via the matching sender stored in `TaskMeta.steer_tx`.
|
||||
// Install the steer channel for every prompt task — the supervisor
|
||||
// uses try-and-tolerate: it attempts the steer for any agent and
|
||||
// treats `-32601 method_not_found` as "fall back to cancel+merge".
|
||||
let (tx, rx) = tokio::sync::mpsc::channel::<pool::SteerRequest>(1);
|
||||
agent.acp.install_steer_rx(rx);
|
||||
let steer_tx = Some(tx);
|
||||
|
||||
// Prompt text is now built inside run_prompt_task (needs async for
|
||||
// context fetching). Pass None for prompt_text; batch carries the data.
|
||||
let (control_tx, control_rx) = tokio::sync::oneshot::channel::<ControlSignal>();
|
||||
@@ -2186,6 +2544,7 @@ fn dispatch_pending(
|
||||
channel_id: Some(channel_id),
|
||||
recoverable_batch,
|
||||
control_tx: Some(control_tx),
|
||||
steer_tx,
|
||||
},
|
||||
);
|
||||
dispatched_channels.push((channel_id, typing_scope));
|
||||
@@ -2229,8 +2588,14 @@ fn handle_prompt_result(
|
||||
if matches!(result.outcome, PromptOutcome::Cancelled) {
|
||||
// Cancel re-prompt: store as cancelled events so flush_next()
|
||||
// merges them into the next FlushBatch.cancelled_events,
|
||||
// enabling the annotated merged-prompt format.
|
||||
queue.requeue_as_cancelled(batch);
|
||||
// enabling the annotated merged-prompt format. The batch's
|
||||
// cancel_reason (set by the pool task per the control signal)
|
||||
// selects steer vs interrupt framing. It is always set on this
|
||||
// path; if somehow unset, fall back to the gentler Steer framing
|
||||
// — consistent with MergeFraming::for_reason(None) and the
|
||||
// system default — rather than telling the agent to supersede.
|
||||
let reason = batch.cancel_reason.unwrap_or(CancelReason::Steer);
|
||||
queue.requeue_as_cancelled(batch, reason);
|
||||
} else {
|
||||
queue.requeue(batch);
|
||||
}
|
||||
@@ -2561,6 +2926,7 @@ fn dispatch_heartbeat(
|
||||
channel_id: None,
|
||||
recoverable_batch: None,
|
||||
control_tx: None,
|
||||
steer_tx: None,
|
||||
},
|
||||
);
|
||||
*heartbeat_in_flight = true;
|
||||
@@ -2643,6 +3009,7 @@ fn spawn_respawn_task(
|
||||
true
|
||||
}
|
||||
|
||||
// ── spawn_and_init ────────────────────────────────────────────────────────────
|
||||
/// Spawn an agent subprocess and run the MCP `initialize` handshake.
|
||||
///
|
||||
/// Takes owned args so it can run in a background `tokio::spawn` task without
|
||||
@@ -2653,7 +3020,7 @@ async fn spawn_and_init(
|
||||
extra_env: &[(String, String)],
|
||||
agent_index: usize,
|
||||
observer: Option<observer::ObserverHandle>,
|
||||
) -> Result<(AcpClient, u32)> {
|
||||
) -> Result<(AcpClient, u32, bool)> {
|
||||
let mut acp = AcpClient::spawn(command, args, extra_env)
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?;
|
||||
@@ -2670,7 +3037,7 @@ async fn spawn_and_init(
|
||||
"initializeResult": init_result,
|
||||
}),
|
||||
);
|
||||
Ok((acp, protocol_version))
|
||||
Ok((acp, protocol_version, true))
|
||||
}
|
||||
Err(e) => {
|
||||
// Explicitly shut down the spawned child to prevent zombie/leak.
|
||||
@@ -2947,6 +3314,46 @@ mod owner_control_command_tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mode_gate_signal_maps_handling_to_control_signal() {
|
||||
let owner = "a".repeat(64);
|
||||
let other = "b".repeat(64);
|
||||
|
||||
// Queue: never signals — events wait for the turn to finish.
|
||||
assert!(mode_gate_signal(MultipleEventHandling::Queue, &owner, Some(&owner)).is_none());
|
||||
|
||||
// Steer: always steers (eligibility already enforced upstream).
|
||||
assert!(matches!(
|
||||
mode_gate_signal(MultipleEventHandling::Steer, &other, Some(&owner)),
|
||||
Some(ControlSignal::Steer)
|
||||
));
|
||||
// Steer even when owner is unknown — gate doesn't re-check authorship.
|
||||
assert!(matches!(
|
||||
mode_gate_signal(MultipleEventHandling::Steer, &other, None),
|
||||
Some(ControlSignal::Steer)
|
||||
));
|
||||
|
||||
// Interrupt: always interrupts for any eligible author.
|
||||
assert!(matches!(
|
||||
mode_gate_signal(MultipleEventHandling::Interrupt, &other, Some(&owner)),
|
||||
Some(ControlSignal::Interrupt)
|
||||
));
|
||||
|
||||
// OwnerInterrupt: interrupts only for the owner.
|
||||
assert!(matches!(
|
||||
mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &owner, Some(&owner)),
|
||||
Some(ControlSignal::Interrupt)
|
||||
));
|
||||
assert!(
|
||||
mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &other, Some(&owner)).is_none(),
|
||||
"owner-interrupt must not fire for a non-owner author"
|
||||
);
|
||||
assert!(
|
||||
mode_gate_signal(MultipleEventHandling::OwnerInterrupt, &owner, None).is_none(),
|
||||
"owner-interrupt must not fire when the owner is unknown"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn signal_in_flight_task_sends_rotate_once() {
|
||||
let mut pool = AgentPool::from_slots(vec![]);
|
||||
@@ -2962,6 +3369,7 @@ mod owner_control_command_tests {
|
||||
channel_id: Some(channel_id),
|
||||
recoverable_batch: None,
|
||||
control_tx: Some(control_tx),
|
||||
steer_tx: None,
|
||||
},
|
||||
);
|
||||
|
||||
@@ -3103,6 +3511,44 @@ mod author_gate_tests {
|
||||
"the owner must always be accepted under Allowlist"
|
||||
);
|
||||
}
|
||||
|
||||
// The default `respond-to` is OwnerOnly. Under steering, "an ineligible
|
||||
// author must NOT steer" is enforced *here* — author_allowed drops the
|
||||
// event before it reaches the mode gate — not in the gate itself. These
|
||||
// pin that invariant against the default mode.
|
||||
#[tokio::test]
|
||||
async fn test_owner_only_rejects_stranger_so_no_steer() {
|
||||
let cache = cache_with_sibling();
|
||||
assert!(
|
||||
!author_allowed(
|
||||
&RespondTo::OwnerOnly,
|
||||
&HashSet::new(),
|
||||
STRANGER,
|
||||
&cache,
|
||||
&dummy_rest_client()
|
||||
)
|
||||
.await,
|
||||
"under the default OwnerOnly, a stranger must be dropped — so it can never reach the mode gate to steer"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_owner_only_admits_owner_and_sibling_to_steer() {
|
||||
let cache = cache_with_sibling();
|
||||
for (who, label) in [(OWNER, "owner"), (SIBLING, "sibling")] {
|
||||
assert!(
|
||||
author_allowed(
|
||||
&RespondTo::OwnerOnly,
|
||||
&HashSet::new(),
|
||||
who,
|
||||
&cache,
|
||||
&dummy_rest_client()
|
||||
)
|
||||
.await,
|
||||
"under default OwnerOnly, the {label} must be admitted so steering can fire"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -3445,6 +3891,7 @@ mod error_outcome_emission_tests {
|
||||
channel_id: None,
|
||||
recoverable_batch: None,
|
||||
control_tx: None,
|
||||
steer_tx: None,
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
+196
-19
@@ -35,8 +35,8 @@ use crate::acp::{
|
||||
use crate::config::{DedupMode, PermissionMode};
|
||||
use crate::observer;
|
||||
use crate::queue::{
|
||||
ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo, PromptProfile,
|
||||
PromptProfileLookup,
|
||||
CancelReason, ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo,
|
||||
PromptProfile, PromptProfileLookup,
|
||||
};
|
||||
use crate::relay::{ChannelInfo, RestClient};
|
||||
|
||||
@@ -52,6 +52,13 @@ pub struct TaskMeta {
|
||||
/// Control signal for the in-flight prompt task.
|
||||
/// `None` for heartbeat tasks (not controllable) and after signal is consumed.
|
||||
pub control_tx: Option<tokio::sync::oneshot::Sender<ControlSignal>>,
|
||||
/// Steer request channel for non-cancelling mid-turn delivery.
|
||||
/// Capacity-1; `try_send` from the main loop fails on `Full`/`Closed`,
|
||||
/// in which case the caller must fall back to the universal
|
||||
/// `ControlSignal::Steer` cancel+merge path. `None` for heartbeat
|
||||
/// tasks only — all prompt tasks install a steer channel regardless
|
||||
/// of the agent's name.
|
||||
pub steer_tx: Option<tokio::sync::mpsc::Sender<SteerRequest>>,
|
||||
}
|
||||
|
||||
/// Agent-level model capabilities. Populated on first session creation.
|
||||
@@ -185,13 +192,127 @@ fn apply_completed_before_control_signal(
|
||||
pub enum ControlSignal {
|
||||
/// Stop the current turn and drop its triggering batch.
|
||||
Cancel,
|
||||
/// Stop the current turn and requeue its triggering batch for a merged re-prompt.
|
||||
/// Stop the current turn and requeue its triggering batch for a merged
|
||||
/// re-prompt framed as a **supersede**: the new request replaces the old.
|
||||
Interrupt,
|
||||
/// Stop the current turn and requeue its triggering batch for a merged
|
||||
/// re-prompt framed as a **steer**: a message arrived while the agent was
|
||||
/// working; it should continue its work and incorporate the message if
|
||||
/// relevant, not treat it as a replacement task. This is the default
|
||||
/// mid-turn delivery path (see [`MultipleEventHandling::Steer`]).
|
||||
Steer,
|
||||
/// Stop the current turn and drop its triggering batch. The session is
|
||||
/// invalidated just like cancel; the next turn creates a fresh session.
|
||||
Rotate,
|
||||
}
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// 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.
|
||||
///
|
||||
/// ## Why the read loop fills params, not the main loop
|
||||
///
|
||||
/// `expectedRunId` is a *moving target*: the read loop updates
|
||||
/// `self.active_run_id` as goose emits `session/update` notifications, and
|
||||
/// the steer is rejected if the supplied id doesn't match the *current* run.
|
||||
/// A snapshot taken at dispatch (or at mode-gate time) can be stale by the
|
||||
/// time the read loop actually writes the steer line. Filling params at
|
||||
/// write time uses the freshest possible run id and is correct-by-
|
||||
/// construction on the one field whose freshness the protocol checks.
|
||||
/// `sessionId` is in lexical scope inside the read loop's caller
|
||||
/// (`session_prompt_blocks_with_idle_timeout`), so no plumbing is required
|
||||
/// for that — only a function parameter pass-through.
|
||||
///
|
||||
/// If `active_run_id` is `None` at write time (no `session/update` seen yet
|
||||
/// — e.g. agents that never emit run-id metadata), the steer cannot form a
|
||||
/// valid `expectedRunId` and the read loop acks
|
||||
/// [`SteerError::ExpectedRunIdMissing`]. The main loop maps this to the
|
||||
/// "Err-before-pending" bucket: no withhold/mark was established at
|
||||
/// `pool::send_steer` time because the request was rejected before any
|
||||
/// write, so the watcher only needs to release nothing and fall back to the
|
||||
/// universal `ControlSignal::Steer` cancel+merge path.
|
||||
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.
|
||||
pub prompt_blocks: Vec<String>,
|
||||
/// Oneshot for the read loop to report the outcome.
|
||||
pub ack_tx: tokio::sync::oneshot::Sender<SteerAck>,
|
||||
}
|
||||
|
||||
/// Why a goose-native steer failed.
|
||||
///
|
||||
/// String and integer fields are intentionally `Debug`-only — read by
|
||||
/// `tracing` macros in the main loop's `PoolEvent::SteerAck` arm via
|
||||
/// `?ack`. The dead-code lint can't see that path because it doesn't
|
||||
/// trace through `Debug` derives, hence the `#[allow]`.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug)]
|
||||
pub enum SteerError {
|
||||
/// The agent returned a JSON-RPC error response to the steer request.
|
||||
///
|
||||
/// `code` is the JSON-RPC error code:
|
||||
/// - `-32601` (`method_not_found`): the agent does not implement the
|
||||
/// steer extension. The main loop should fire the cancel+merge
|
||||
/// fallback so the message still reaches the agent.
|
||||
/// - Any other code: the write landed and the agent rejected it at the
|
||||
/// application level (e.g. wrong run id). Release the withheld event
|
||||
/// for normal dispatch; do NOT fire the fallback — the turn is still
|
||||
/// running or just ended.
|
||||
AgentError { code: i64, message: String },
|
||||
/// Transport-level failure: write error, read EOF, JSON-RPC framing
|
||||
/// violation, etc. The string carries the underlying `AcpError`'s display.
|
||||
Transport(String),
|
||||
/// At steer-write time `AcpClient::active_run_id` was `None`, so the
|
||||
/// read loop couldn't form a valid `expectedRunId`. The read loop drops
|
||||
/// the request without writing anything; the main loop should release
|
||||
/// any withheld event and fall back to the universal cancel+merge
|
||||
/// `ControlSignal::Steer` path. This is in the same "Err-before-pending"
|
||||
/// bucket as `Transport` write failures: no in-process state was
|
||||
/// established, so no in-process cleanup is needed.
|
||||
ExpectedRunIdMissing,
|
||||
/// The read loop never got to dispatch the steer because the prompt
|
||||
/// completed first. Delivery state for the underlying message is
|
||||
/// unknown after prompt completion — the main loop must treat this as
|
||||
/// "release the withheld event so normal dispatch handles it" with no
|
||||
/// claims that the agent did or did not incorporate it.
|
||||
///
|
||||
/// Returned synchronously by `send_steer` when no task is in flight
|
||||
/// for the channel. Never sent through the ack channel — the ack
|
||||
/// watcher is only spawned on `send_steer` success.
|
||||
PromptCompleted,
|
||||
}
|
||||
|
||||
/// Outcome of a goose-native steer, sent from the read loop back to the
|
||||
/// main loop's ack watcher.
|
||||
#[derive(Debug)]
|
||||
pub enum SteerAck {
|
||||
/// The agent returned a successful response to the steer request.
|
||||
/// The main loop must drop the withheld event (`remove_event`) — it
|
||||
/// has been delivered via the non-cancelling path.
|
||||
Success,
|
||||
/// The steer was attempted but failed. Delivery state for the
|
||||
/// underlying message is unknown after prompt completion; the main
|
||||
/// loop must release the withheld event and fall back to the
|
||||
/// universal `Steer` cancel+merge path so the message still reaches
|
||||
/// the agent.
|
||||
Err(SteerError),
|
||||
/// The prompt completed before the read loop selected the steer arm.
|
||||
/// Treated as a benign no-op: release the withheld event for normal
|
||||
/// dispatch. Do not fire the fallback `Steer` signal — there is no
|
||||
/// in-flight turn to signal, and normal dispatch handles delivery.
|
||||
PromptCompletedNeutral,
|
||||
}
|
||||
|
||||
/// Outcome of a prompt task.
|
||||
#[allow(dead_code)]
|
||||
pub enum PromptOutcome {
|
||||
@@ -342,6 +463,46 @@ impl AgentPool {
|
||||
&mut self.task_map
|
||||
}
|
||||
|
||||
/// Try to send a goose-native steer request to the in-flight task for
|
||||
/// `channel_id`.
|
||||
///
|
||||
/// Returns `Ok(())` if the request was accepted by the read loop's
|
||||
/// receiver (capacity-1 mpsc; one slot is the single in-flight steer
|
||||
/// write). Returns `Err(SteerError::Transport(_))` on `Full`/`Closed`
|
||||
/// (already-in-flight write, or read loop torn down). Callers must
|
||||
/// fall back to the universal `ControlSignal::Steer` cancel+merge path
|
||||
/// on `Err`.
|
||||
///
|
||||
/// This does **not** spawn the ack watcher — the caller owns the
|
||||
/// oneshot `ack_tx` inside `SteerRequest` and is responsible for
|
||||
/// awaiting it and applying the locked Success / Err / PromptCompletedNeutral
|
||||
/// semantics. Caller is also responsible for the synchronous
|
||||
/// `queue.mark_native_steer_pending(...)` *before* spawning the
|
||||
/// watcher, to close the result-vs-ack race.
|
||||
///
|
||||
/// Returns `Err(SteerError::PromptCompleted)` if no task is in flight
|
||||
/// for `channel_id` (the prompt completed between the mode-gate check
|
||||
/// and this call, or the channel was never in flight). This is
|
||||
/// semantically a soft no-op — the caller should release any withheld
|
||||
/// event and let normal dispatch handle delivery.
|
||||
pub fn send_steer(
|
||||
&mut self,
|
||||
channel_id: Uuid,
|
||||
request: SteerRequest,
|
||||
) -> Result<(), SteerError> {
|
||||
let meta = self
|
||||
.task_map
|
||||
.values_mut()
|
||||
.find(|m| m.channel_id == Some(channel_id))
|
||||
.ok_or(SteerError::PromptCompleted)?;
|
||||
let tx = meta
|
||||
.steer_tx
|
||||
.as_ref()
|
||||
.ok_or_else(|| SteerError::Transport("steer_tx not installed".into()))?;
|
||||
tx.try_send(request)
|
||||
.map_err(|e| SteerError::Transport(e.to_string()))
|
||||
}
|
||||
|
||||
pub fn result_tx(&self) -> mpsc::UnboundedSender<PromptResult> {
|
||||
self.result_tx.clone()
|
||||
}
|
||||
@@ -1218,10 +1379,8 @@ pub async fn run_prompt_task(
|
||||
Ok(stop_reason) => {
|
||||
log_stop_reason(&source, &stop_reason);
|
||||
agent.state.invalidate(&source);
|
||||
let retry_batch = match control_signal {
|
||||
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
|
||||
ControlSignal::Cancel | ControlSignal::Rotate => None,
|
||||
};
|
||||
let retry_batch =
|
||||
requeue_cancelled_batch(&ctx, control_signal, batch);
|
||||
let _ = result_tx.send(PromptResult {
|
||||
agent,
|
||||
source,
|
||||
@@ -1232,10 +1391,8 @@ pub async fn run_prompt_task(
|
||||
}
|
||||
Err(AcpError::AgentExited) => {
|
||||
agent.state.invalidate_all();
|
||||
let retry_batch = match control_signal {
|
||||
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
|
||||
ControlSignal::Cancel | ControlSignal::Rotate => None,
|
||||
};
|
||||
let retry_batch =
|
||||
requeue_cancelled_batch(&ctx, control_signal, batch);
|
||||
let _ = result_tx.send(PromptResult {
|
||||
agent,
|
||||
source,
|
||||
@@ -1247,10 +1404,8 @@ pub async fn run_prompt_task(
|
||||
Err(AcpError::IdleTimeout(_) | AcpError::HardTimeout) => {
|
||||
// Cancel drain timed out — agent state uncertain.
|
||||
agent.state.invalidate(&source);
|
||||
let retry_batch = match control_signal {
|
||||
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
|
||||
ControlSignal::Cancel | ControlSignal::Rotate => None,
|
||||
};
|
||||
let retry_batch =
|
||||
requeue_cancelled_batch(&ctx, control_signal, batch);
|
||||
let _ = result_tx.send(PromptResult {
|
||||
agent,
|
||||
source,
|
||||
@@ -1261,10 +1416,8 @@ pub async fn run_prompt_task(
|
||||
}
|
||||
Err(e) => {
|
||||
agent.state.invalidate(&source);
|
||||
let retry_batch = match control_signal {
|
||||
ControlSignal::Interrupt => requeue_batch_if_queue(&ctx, batch),
|
||||
ControlSignal::Cancel | ControlSignal::Rotate => None,
|
||||
};
|
||||
let retry_batch =
|
||||
requeue_cancelled_batch(&ctx, control_signal, batch);
|
||||
let _ = result_tx.send(PromptResult {
|
||||
agent,
|
||||
source,
|
||||
@@ -2035,6 +2188,29 @@ fn requeue_batch_if_queue(ctx: &PromptContext, batch: Option<FlushBatch>) -> Opt
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a cancelling [`ControlSignal`] to the [`CancelReason`] that should frame
|
||||
/// the merged re-prompt, then requeue the batch (in `Queue` dedup mode) with
|
||||
/// that reason stamped onto [`FlushBatch::cancel_reason`]. `Cancel`/`Rotate`
|
||||
/// drop the batch entirely. The reason is consumed by the main loop at requeue
|
||||
/// time (`requeue_as_cancelled`) and ultimately by `format_prompt`.
|
||||
#[inline]
|
||||
fn requeue_cancelled_batch(
|
||||
ctx: &PromptContext,
|
||||
signal: ControlSignal,
|
||||
batch: Option<FlushBatch>,
|
||||
) -> Option<FlushBatch> {
|
||||
let reason = match signal {
|
||||
ControlSignal::Steer => CancelReason::Steer,
|
||||
ControlSignal::Interrupt => CancelReason::Interrupt,
|
||||
// Cancel/Rotate discard the batch — no merged re-prompt.
|
||||
ControlSignal::Cancel | ControlSignal::Rotate => return None,
|
||||
};
|
||||
requeue_batch_if_queue(ctx, batch).map(|mut b| {
|
||||
b.cancel_reason = Some(reason);
|
||||
b
|
||||
})
|
||||
}
|
||||
|
||||
/// Log a stop reason at the appropriate tracing level.
|
||||
fn log_stop_reason(source: &PromptSource, stop_reason: &StopReason) {
|
||||
let label = match source {
|
||||
@@ -2761,6 +2937,7 @@ mod tests {
|
||||
received_at: std::time::Instant::now(),
|
||||
}],
|
||||
cancelled_events: vec![],
|
||||
cancel_reason: None,
|
||||
};
|
||||
let context = ConversationContext::Thread {
|
||||
messages: vec![ContextMessage {
|
||||
|
||||
+801
-22
File diff suppressed because it is too large
Load Diff
@@ -1,7 +1,7 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use serde_json::json;
|
||||
use tokio::sync::{watch, Semaphore};
|
||||
use tokio::sync::{mpsc, watch, Semaphore};
|
||||
use tokio::task::JoinSet;
|
||||
|
||||
use crate::builtin;
|
||||
@@ -31,6 +31,11 @@ pub struct RunCtx<'a> {
|
||||
pub skills: &'a [SkillEntry],
|
||||
pub wire: &'a WireSender,
|
||||
pub cancel: &'a mut watch::Receiver<bool>,
|
||||
/// Mid-turn steer queue. Drained at each round boundary (before the next
|
||||
/// LLM call): queued messages are appended to history as user turns so the
|
||||
/// model sees them on its next request, without restarting the turn. Fed by
|
||||
/// the `_goose/unstable/session/steer` handler.
|
||||
pub steer: &'a mut mpsc::UnboundedReceiver<Vec<ContentBlock>>,
|
||||
pub history: &'a mut Vec<HistoryItem>,
|
||||
pub original_task: &'a mut Option<String>,
|
||||
pub handoff_count: &'a mut usize,
|
||||
@@ -78,6 +83,11 @@ impl RunCtx<'_> {
|
||||
if *self.cancel.borrow() {
|
||||
return Ok(StopReason::Cancelled);
|
||||
}
|
||||
// Round boundary: fold in any steer messages queued since the last
|
||||
// round. They land as user turns so the model incorporates them on
|
||||
// its next request — the turn continues, it is not restarted. Drain
|
||||
// non-blocking; an empty queue is the common case.
|
||||
self.drain_steers();
|
||||
match self.maybe_handoff().await {
|
||||
HandoffOutcome::Cancelled => return Ok(StopReason::Cancelled),
|
||||
// Context was just reset — the prior request's token count no
|
||||
@@ -227,6 +237,27 @@ impl RunCtx<'_> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Non-blocking drain of the steer queue. Each queued steer is appended to
|
||||
/// history as a user turn so the model picks it up on its next request. A
|
||||
/// steer whose blocks all fail to render (e.g. unsupported content) is
|
||||
/// skipped rather than aborting the turn — steering is best-effort
|
||||
/// augmentation, not a hard input contract like the initial prompt.
|
||||
fn drain_steers(&mut self) {
|
||||
while let Ok(blocks) = self.steer.try_recv() {
|
||||
match prompt_to_text(blocks) {
|
||||
Ok(text) if !text.trim().is_empty() => {
|
||||
self.history.push(HistoryItem::User(text));
|
||||
}
|
||||
Ok(_) => {
|
||||
tracing::debug!("dropping empty steer message");
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("dropping unrenderable steer message: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Unified tool-call execution. Three phases:
|
||||
/// 1. Preflight (sequential): emit `pending`; unknown tools fail fast
|
||||
/// with a synthetic result. Cancel here fills every still-empty
|
||||
|
||||
@@ -23,10 +23,11 @@ use crate::config::{Config, MAX_SYSTEM_PROMPT_BYTES, PROTOCOL_VERSION};
|
||||
use crate::hints::SkillEntry;
|
||||
use crate::llm::Llm;
|
||||
use crate::mcp::McpRegistry;
|
||||
use crate::types::HistoryItem;
|
||||
use crate::types::{ContentBlock, HistoryItem};
|
||||
use crate::wire::{
|
||||
classify, Inbound, InitializeParams, SessionCancelParams, SessionNewParams,
|
||||
SessionPromptParams, WireMsg, WireSender, INVALID_PARAMS, METHOD_NOT_FOUND, PARSE_ERROR,
|
||||
SessionPromptParams, SessionSteerParams, WireMsg, WireSender, INVALID_PARAMS, METHOD_NOT_FOUND,
|
||||
PARSE_ERROR,
|
||||
};
|
||||
|
||||
struct App {
|
||||
@@ -43,6 +44,16 @@ struct Session {
|
||||
history: Vec<HistoryItem>,
|
||||
cancel_tx: watch::Sender<bool>,
|
||||
busy: bool,
|
||||
/// Run id of the in-flight prompt, set when a prompt starts and cleared
|
||||
/// when it ends. `None` means no active run — a steer request targeting
|
||||
/// this session is rejected. Steer-capable clients learn this value from
|
||||
/// the `params.update._meta.goose.activeRunId` field on `session/update`.
|
||||
active_run_id: Option<String>,
|
||||
/// Sender for mid-turn steer messages. Created fresh per prompt (like
|
||||
/// `cancel_tx`); the running prompt loop holds the matching receiver and
|
||||
/// drains queued steers at round boundaries. `None` when no prompt is in
|
||||
/// flight.
|
||||
steer_tx: Option<mpsc::UnboundedSender<Vec<ContentBlock>>>,
|
||||
original_task: Option<String>,
|
||||
handoff_count: usize,
|
||||
stop_rejections: u32,
|
||||
@@ -194,6 +205,13 @@ async fn handle_request(
|
||||
cancel_session(app, params).await;
|
||||
wire::send(wire_tx, wire::ok(id, Value::Null)).await;
|
||||
}
|
||||
// goose-compatible non-standard extension: inject user input into the
|
||||
// currently active prompt without starting a new one. Mirrors goose's
|
||||
// `_goose/unstable/session/steer` wire contract so a single client-side
|
||||
// delivery path serves both agents.
|
||||
"_goose/unstable/session/steer" => {
|
||||
steer_session(app, id, params, wire_tx).await;
|
||||
}
|
||||
_ => {
|
||||
wire::send(
|
||||
wire_tx,
|
||||
@@ -334,6 +352,8 @@ async fn session_new(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSen
|
||||
history: Vec::new(),
|
||||
cancel_tx,
|
||||
busy: false,
|
||||
active_run_id: None,
|
||||
steer_tx: None,
|
||||
original_task: None,
|
||||
handoff_count: 0,
|
||||
stop_rejections: 0,
|
||||
@@ -362,6 +382,85 @@ async fn cancel_session(app: &Arc<App>, params: Value) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle `_goose/unstable/session/steer`: queue user input into the in-flight
|
||||
/// prompt. Validation mirrors goose's `on_steer_session`:
|
||||
/// - empty prompt → `invalid_params`
|
||||
/// - no active run (no prompt in flight) → `invalid_params`
|
||||
/// - `expectedRunId` mismatch → `invalid_params` (caller is steering a turn
|
||||
/// that already ended or rotated; it must fall back to cancel+merge)
|
||||
///
|
||||
/// On success the message is queued for pickup at the next round boundary and
|
||||
/// we reply `{ runId, messageId }`, then emit a `queuedSteer` session/update so
|
||||
/// the client can correlate the accepted steer with its eventual pickup.
|
||||
async fn steer_session(app: &Arc<App>, id: Value, params: Value, wire_tx: &WireSender) {
|
||||
let p: SessionSteerParams = match decode(params, "_goose/unstable/session/steer") {
|
||||
Ok(p) => p,
|
||||
Err(m) => return reject(wire_tx, id, INVALID_PARAMS, &m).await,
|
||||
};
|
||||
if p.prompt.is_empty() {
|
||||
return reject(
|
||||
wire_tx,
|
||||
id,
|
||||
INVALID_PARAMS,
|
||||
"steer: prompt must not be empty",
|
||||
)
|
||||
.await;
|
||||
}
|
||||
if p.expected_run_id.is_empty() {
|
||||
return reject(
|
||||
wire_tx,
|
||||
id,
|
||||
INVALID_PARAMS,
|
||||
"steer: expectedRunId must not be empty",
|
||||
)
|
||||
.await;
|
||||
}
|
||||
let message_id = format!("steer_{}", session_token().unwrap_or_else(|_| "x".into()));
|
||||
let run_id = {
|
||||
let sessions = app.sessions.lock().await;
|
||||
let Some(s) = sessions.get(&p.session_id) else {
|
||||
return reject(wire_tx, id, INVALID_PARAMS, "steer: unknown session").await;
|
||||
};
|
||||
let Some(active) = s.active_run_id.as_deref() else {
|
||||
return reject(wire_tx, id, INVALID_PARAMS, "steer: no active run to steer").await;
|
||||
};
|
||||
if active != p.expected_run_id {
|
||||
return reject(
|
||||
wire_tx,
|
||||
id,
|
||||
INVALID_PARAMS,
|
||||
&format!(
|
||||
"steer: expected active run id `{}` but found `{active}`",
|
||||
p.expected_run_id
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
// A live run always has a steer_tx; if the channel is gone the run is
|
||||
// tearing down — treat as no active run rather than queue into the void.
|
||||
match &s.steer_tx {
|
||||
Some(tx) if tx.send(p.prompt).is_ok() => active.to_owned(),
|
||||
_ => return reject(wire_tx, id, INVALID_PARAMS, "steer: no active run to steer").await,
|
||||
}
|
||||
};
|
||||
wire::send(
|
||||
wire_tx,
|
||||
wire::ok(id, json!({ "runId": run_id, "messageId": message_id })),
|
||||
)
|
||||
.await;
|
||||
// Best-effort correlation hint for the client; mirrors goose's
|
||||
// `send_queued_steer_update`. Not load-bearing for delivery.
|
||||
wire::send(
|
||||
wire_tx,
|
||||
wire::session_update_with_goose_meta(
|
||||
&p.session_id,
|
||||
json!({ "sessionUpdate": "session_info_update" }),
|
||||
json!({ "queuedSteer": { "messageId": message_id, "runId": run_id } }),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
fn spawn_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender) {
|
||||
tokio::spawn(async move { run_prompt(app, id, params, wire_tx).await });
|
||||
}
|
||||
@@ -383,6 +482,8 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
|
||||
mut last_request_history_bytes,
|
||||
mut cancel_rx,
|
||||
effective_system_prompt,
|
||||
run_id,
|
||||
mut steer_rx,
|
||||
) = match acquire_session(&app, &p.session_id).await {
|
||||
Ok(v) => v,
|
||||
Err(reason) => {
|
||||
@@ -395,6 +496,17 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
|
||||
.await
|
||||
}
|
||||
};
|
||||
// Advertise the active run id so steer-capable clients can target this turn
|
||||
// via `expectedRunId`. Mirrors goose's `send_active_run_update`.
|
||||
wire::send(
|
||||
&wire_tx,
|
||||
wire::session_update_with_goose_meta(
|
||||
&sid,
|
||||
json!({ "sessionUpdate": "session_info_update" }),
|
||||
json!({ "activeRunId": run_id }),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
let mut ctx = RunCtx {
|
||||
cfg: &app.cfg,
|
||||
session_id: &sid,
|
||||
@@ -404,6 +516,7 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
|
||||
skills: &skills,
|
||||
wire: &wire_tx,
|
||||
cancel: &mut cancel_rx,
|
||||
steer: &mut steer_rx,
|
||||
history: &mut history,
|
||||
original_task: &mut original_task,
|
||||
handoff_count: &mut handoff_count,
|
||||
@@ -414,6 +527,9 @@ async fn run_prompt(app: Arc<App>, id: Value, params: Value, wire_tx: WireSender
|
||||
let result = ctx.run(p.prompt).await;
|
||||
if let Some(s) = app.sessions.lock().await.get_mut(&sid) {
|
||||
s.busy = false;
|
||||
// Clear run state so a late steer can't queue into a finished turn.
|
||||
s.active_run_id = None;
|
||||
s.steer_tx = None;
|
||||
s.history = history;
|
||||
s.original_task = original_task;
|
||||
s.handoff_count = handoff_count;
|
||||
@@ -449,6 +565,8 @@ async fn acquire_session(
|
||||
Option<usize>,
|
||||
watch::Receiver<bool>,
|
||||
Arc<str>,
|
||||
String,
|
||||
mpsc::UnboundedReceiver<Vec<ContentBlock>>,
|
||||
),
|
||||
&'static str,
|
||||
> {
|
||||
@@ -463,6 +581,13 @@ async fn acquire_session(
|
||||
// Skills are read-only after session creation; clone the Vec so RunCtx
|
||||
// can hold a reference without holding the sessions lock.
|
||||
let skills = s.skills.clone();
|
||||
// Fresh run id + steer channel for this turn. The run id lets steer-capable
|
||||
// clients target *this* turn (rejecting steers aimed at a turn that already
|
||||
// ended); the channel carries mid-turn injections to the run loop.
|
||||
let run_id = format!("run_{}", session_token().unwrap_or_else(|_| "x".into()));
|
||||
s.active_run_id = Some(run_id.clone());
|
||||
let (steer_tx, steer_rx) = mpsc::unbounded_channel();
|
||||
s.steer_tx = Some(steer_tx);
|
||||
Ok((
|
||||
s.id.clone(),
|
||||
s.mcp.clone(),
|
||||
@@ -475,6 +600,8 @@ async fn acquire_session(
|
||||
s.last_request_history_bytes,
|
||||
rx,
|
||||
Arc::clone(&s.effective_system_prompt),
|
||||
run_id,
|
||||
steer_rx,
|
||||
))
|
||||
}
|
||||
|
||||
|
||||
@@ -66,6 +66,20 @@ pub struct SessionCancelParams {
|
||||
pub session_id: String,
|
||||
}
|
||||
|
||||
/// Params for goose's non-standard `_goose/unstable/session/steer` request:
|
||||
/// inject user input into the *currently active* prompt without starting a new
|
||||
/// one. `expected_run_id` must match the run id buzz-agent advertised via
|
||||
/// `params.update._meta.goose.activeRunId` on a `session/update`, so a steer
|
||||
/// can't race a turn that already ended or hasn't started.
|
||||
#[derive(Debug, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct SessionSteerParams {
|
||||
pub session_id: String,
|
||||
#[serde(default)]
|
||||
pub prompt: Vec<ContentBlock>,
|
||||
pub expected_run_id: String,
|
||||
}
|
||||
|
||||
pub fn classify(msg: &Value) -> Inbound {
|
||||
if !msg.is_object() || msg.get("jsonrpc").and_then(Value::as_str) != Some("2.0") {
|
||||
return Inbound::Invalid {
|
||||
@@ -112,6 +126,25 @@ pub fn session_update(sid: &str, update: Value) -> Value {
|
||||
})
|
||||
}
|
||||
|
||||
/// A `session/update` notification carrying a `update._meta.goose.<key>` field.
|
||||
/// Used to advertise `activeRunId` (so steer-capable clients can target the
|
||||
/// in-flight run) and `queuedSteer` (so they can correlate an accepted steer
|
||||
/// with the chunk that later picks it up) — matching goose's wire layout where
|
||||
/// `_meta` is nested inside the `update` object (per the ACP `SessionInfoUpdate`
|
||||
/// schema), not alongside it at the params level.
|
||||
pub fn session_update_with_goose_meta(sid: &str, update: Value, goose_meta: Value) -> Value {
|
||||
let mut update = update;
|
||||
update["_meta"] = json!({ "goose": goose_meta });
|
||||
json!({
|
||||
"jsonrpc": "2.0",
|
||||
"method": "session/update",
|
||||
"params": {
|
||||
"sessionId": sid,
|
||||
"update": update,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn send(wire: &WireSender, msg: Value) {
|
||||
let _ = wire.send(WireMsg::Notify(msg)).await;
|
||||
}
|
||||
|
||||
@@ -574,3 +574,214 @@ async fn system_prompt_absent_no_canary() {
|
||||
|
||||
h.shutdown().await;
|
||||
}
|
||||
|
||||
// ─── Steering (_goose/unstable/session/steer) ───────────────────────────────
|
||||
|
||||
/// Wait for the `activeRunId` advert buzz-agent emits at prompt start and
|
||||
/// return the run id, so a steer can target the live turn.
|
||||
async fn recv_active_run_id(h: &mut Harness) -> String {
|
||||
let v = h
|
||||
.recv_until(|v| {
|
||||
v.get("method") == Some(&json!("session/update"))
|
||||
&& v["params"]["update"]["_meta"]["goose"]["activeRunId"].is_string()
|
||||
})
|
||||
.await;
|
||||
v["params"]["update"]["_meta"]["goose"]["activeRunId"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.to_owned()
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn steer_folds_into_active_turn_without_cancelling() {
|
||||
// A two-round turn (tool call → text). A steer sent once the run is live
|
||||
// must (a) be accepted with the matching runId, (b) NOT cancel the turn —
|
||||
// it still ends with end_turn — and (c) reach the provider as a user turn.
|
||||
let (url, captures) = spawn_capturing_fake_llm(vec![
|
||||
openai_tool_call("call_steer", "fake__noop", json!({})),
|
||||
openai_text("acknowledged the steer"),
|
||||
])
|
||||
.await;
|
||||
let mut h = Harness::spawn(&url).await;
|
||||
let sid = init_session(&mut h).await;
|
||||
|
||||
let p_id = h
|
||||
.send(
|
||||
"session/prompt",
|
||||
json!({
|
||||
"sessionId": sid,
|
||||
"prompt": [{"type":"text","text":"work on the original task"}],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
// Learn the run id, then steer into it before the turn finishes.
|
||||
let run_id = recv_active_run_id(&mut h).await;
|
||||
let steer_text = "STEER-CANARY: also consider the edge case";
|
||||
let s_id = h
|
||||
.send(
|
||||
"_goose/unstable/session/steer",
|
||||
json!({
|
||||
"sessionId": sid,
|
||||
"expectedRunId": run_id,
|
||||
"prompt": [{"type":"text","text": steer_text}],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
// Steer is accepted and echoes the run id it landed in.
|
||||
let mut steer_ok = false;
|
||||
let mut end_turn = false;
|
||||
for _ in 0..40 {
|
||||
let v = h.recv().await;
|
||||
if v["id"] == json!(s_id) {
|
||||
assert_eq!(
|
||||
v["result"]["runId"],
|
||||
json!(run_id),
|
||||
"steer ran into the live turn"
|
||||
);
|
||||
assert!(
|
||||
v["result"]["messageId"]
|
||||
.as_str()
|
||||
.is_some_and(|m| m.starts_with("steer_")),
|
||||
"steer reply carries a messageId"
|
||||
);
|
||||
steer_ok = true;
|
||||
} else if v["id"] == json!(p_id) {
|
||||
// The turn was NOT cancelled — it completed normally.
|
||||
assert_eq!(v["result"]["stopReason"], "end_turn");
|
||||
end_turn = true;
|
||||
}
|
||||
if steer_ok && end_turn {
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert!(steer_ok, "steer request was not accepted");
|
||||
assert!(end_turn, "turn did not complete with end_turn after steer");
|
||||
|
||||
// The steered text reached the provider as a user message in some round.
|
||||
let reqs = captures.lock().await;
|
||||
let saw_steer = reqs.iter().any(|req| {
|
||||
req["messages"].as_array().is_some_and(|msgs| {
|
||||
msgs.iter().any(|m| {
|
||||
m["role"] == "user"
|
||||
&& m["content"]
|
||||
.as_str()
|
||||
.is_some_and(|c| c.contains(steer_text))
|
||||
})
|
||||
})
|
||||
});
|
||||
assert!(
|
||||
saw_steer,
|
||||
"steered text never reached the provider; captured requests: {reqs:#?}"
|
||||
);
|
||||
drop(reqs);
|
||||
h.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn steer_rejected_when_no_active_run() {
|
||||
// No prompt in flight → no active run → invalid_params.
|
||||
let url = spawn_fake_llm(vec![]).await;
|
||||
let mut h = Harness::spawn(&url).await;
|
||||
let sid = init_session(&mut h).await;
|
||||
|
||||
let s_id = h
|
||||
.send(
|
||||
"_goose/unstable/session/steer",
|
||||
json!({
|
||||
"sessionId": sid,
|
||||
"expectedRunId": "run_does_not_exist",
|
||||
"prompt": [{"type":"text","text":"hello?"}],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
let v = h.recv_until(|v| v["id"] == json!(s_id)).await;
|
||||
assert_eq!(v["error"]["code"], -32602, "expected invalid_params");
|
||||
h.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn steer_rejected_on_run_id_mismatch() {
|
||||
// A live run, but the caller targets a stale/wrong run id → invalid_params,
|
||||
// so the client falls back to cancel+merge instead of injecting blind.
|
||||
let (url, _captures) = spawn_capturing_fake_llm(vec![
|
||||
openai_tool_call("call_x", "fake__noop", json!({})),
|
||||
openai_text("done"),
|
||||
])
|
||||
.await;
|
||||
let mut h = Harness::spawn(&url).await;
|
||||
let sid = init_session(&mut h).await;
|
||||
|
||||
let p_id = h
|
||||
.send(
|
||||
"session/prompt",
|
||||
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
|
||||
)
|
||||
.await;
|
||||
let _live_run = recv_active_run_id(&mut h).await;
|
||||
|
||||
let s_id = h
|
||||
.send(
|
||||
"_goose/unstable/session/steer",
|
||||
json!({
|
||||
"sessionId": sid,
|
||||
"expectedRunId": "run_stale_mismatch",
|
||||
"prompt": [{"type":"text","text":"too late"}],
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let mut saw_reject = false;
|
||||
for _ in 0..40 {
|
||||
let v = h.recv().await;
|
||||
if v["id"] == json!(s_id) {
|
||||
assert_eq!(
|
||||
v["error"]["code"], -32602,
|
||||
"mismatched runId must be rejected"
|
||||
);
|
||||
saw_reject = true;
|
||||
} else if v["id"] == json!(p_id) {
|
||||
// Turn finishes normally regardless of the rejected steer.
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert!(saw_reject, "run-id mismatch was not rejected");
|
||||
h.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn steer_rejected_on_empty_prompt() {
|
||||
let (url, _captures) = spawn_capturing_fake_llm(vec![
|
||||
openai_tool_call("call_x", "fake__noop", json!({})),
|
||||
openai_text("done"),
|
||||
])
|
||||
.await;
|
||||
let mut h = Harness::spawn(&url).await;
|
||||
let sid = init_session(&mut h).await;
|
||||
let p_id = h
|
||||
.send(
|
||||
"session/prompt",
|
||||
json!({"sessionId": sid, "prompt": [{"type":"text","text":"go"}]}),
|
||||
)
|
||||
.await;
|
||||
let run_id = recv_active_run_id(&mut h).await;
|
||||
let s_id = h
|
||||
.send(
|
||||
"_goose/unstable/session/steer",
|
||||
json!({"sessionId": sid, "expectedRunId": run_id, "prompt": []}),
|
||||
)
|
||||
.await;
|
||||
let mut saw_reject = false;
|
||||
for _ in 0..40 {
|
||||
let v = h.recv().await;
|
||||
if v["id"] == json!(s_id) {
|
||||
assert_eq!(v["error"]["code"], -32602, "empty prompt must be rejected");
|
||||
saw_reject = true;
|
||||
} else if v["id"] == json!(p_id) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert!(saw_reject, "empty steer prompt was not rejected");
|
||||
h.shutdown().await;
|
||||
}
|
||||
|
||||
@@ -1654,7 +1654,7 @@ pub fn spawn_agent_child(
|
||||
.unwrap_or(super::types::DEFAULT_AGENT_MAX_TURN_DURATION_SECONDS);
|
||||
command.env("BUZZ_ACP_MAX_TURN_DURATION", max_dur.to_string());
|
||||
command.env("BUZZ_ACP_AGENTS", record.parallelism.to_string());
|
||||
command.env("BUZZ_ACP_MULTIPLE_EVENT_HANDLING", "owner-interrupt");
|
||||
command.env("BUZZ_ACP_MULTIPLE_EVENT_HANDLING", "steer");
|
||||
command.env("BUZZ_ACP_DEDUP", "queue");
|
||||
if let Some(meta) = runtime_meta {
|
||||
for (key, value) in meta.default_env {
|
||||
|
||||
@@ -357,6 +357,36 @@ export function processTranscriptEvent(
|
||||
);
|
||||
}
|
||||
}
|
||||
} else if (
|
||||
event.kind === "acp_write" &&
|
||||
method === "_goose/unstable/session/steer"
|
||||
) {
|
||||
const promptText = extractPromptText(payload);
|
||||
if (promptText) {
|
||||
const parsedPrompt = parsePromptText(promptText);
|
||||
if (parsedPrompt.userText) {
|
||||
upsertMessage(
|
||||
d,
|
||||
`steer:${ch}:${event.turnId ?? event.seq}`,
|
||||
"user",
|
||||
parsedPrompt.userTitle,
|
||||
parsedPrompt.userText,
|
||||
event.timestamp,
|
||||
channelId,
|
||||
parsedPrompt.userPubkey,
|
||||
);
|
||||
}
|
||||
if (parsedPrompt.sections.length > 0) {
|
||||
upsertMetadata(
|
||||
d,
|
||||
`steer-context:${ch}:${event.turnId ?? event.seq}`,
|
||||
"Prompt context",
|
||||
parsedPrompt.sections,
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
}
|
||||
}
|
||||
} else if (event.kind === "acp_read" && method === "session/update") {
|
||||
const params = asRecord(payload.params);
|
||||
const update = asRecord(params.update);
|
||||
@@ -375,15 +405,20 @@ export function processTranscriptEvent(
|
||||
channelId,
|
||||
);
|
||||
} else if (updateType === "user_message_chunk") {
|
||||
upsertMessage(
|
||||
d,
|
||||
`user:${ch}:${messageId ?? turnKey}`,
|
||||
"user",
|
||||
"User",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
// Suppress user_message_chunk echo when a steer already rendered
|
||||
// the user message for this turn (Goose echoes steered content back).
|
||||
const steerKey = `steer:${ch}:${event.turnId ?? event.seq}`;
|
||||
if (!d.itemsById.has(steerKey)) {
|
||||
upsertMessage(
|
||||
d,
|
||||
`user:${ch}:${messageId ?? turnKey}`,
|
||||
"user",
|
||||
"User",
|
||||
extractContentText(update.content),
|
||||
event.timestamp,
|
||||
channelId,
|
||||
);
|
||||
}
|
||||
} else if (updateType === "agent_thought_chunk") {
|
||||
upsertTextItem(
|
||||
d,
|
||||
|
||||
@@ -19,7 +19,9 @@ export function parsePromptText(text: string): {
|
||||
userTitle: string;
|
||||
userPubkey: string | null;
|
||||
} {
|
||||
const sections = parsePromptSections(text);
|
||||
const sections = parsePromptSections(text).filter(
|
||||
(s) => s.body.trim().length > 0,
|
||||
);
|
||||
if (sections.length === 0) {
|
||||
return {
|
||||
sections: [],
|
||||
|
||||
Reference in New Issue
Block a user