From 9db0235fd7fd4fd8167c48eff5ed0b1f8ca0f174 Mon Sep 17 00:00:00 2001 From: acoliver Date: Sat, 5 Sep 2026 14:59:40 -0300 Subject: [PATCH] Add steering input queue drained at the tool-call boundary Downstream consumers drive agent turns through AgentStream, which until now exposed only events and cancel(). A message typed while the agent is working had no way to reach the running turn: the caller had to wait for the current run to finish and start a new one. This adds SteeringQueue, a cheaply cloneable handle over a tokio unbounded mpsc. The caller keeps a clone wherever input arrives and passes another to the run via RunOptions::steering, mirroring how message_history is attached. The stream claims the queue's single receiver for the run and drains it, FIFO and without blocking, at the tool-call boundary: after the step's tool returns join the history and before the next model request. That is the same boundary Claude Code uses, and it is the only point where the model is guaranteed to see fresh user intent without corrupting message alternation. Each drained text becomes its own ModelRequest with a user prompt part and emits AgentStreamEvent::SteeringDelivered { step, text } after the step's ToolExecuted events and before the next RequestStart, so a consumer can persist the user message in transcript order. Nothing is drained before the first model request, and a run that never crosses a tool boundary delivers nothing; leftovers stay queued and a later run from the same queue delivers them, so the queue survives turns rather than silently eating input. Both AgentStream::new and AgentStream::new_with_cancel share the drain through one helper because the two spawned loops are physically distinct bodies and steering must behave identically in each. AgentStreamEvent is not #[non_exhaustive], so the new variant is source-breaking for exhaustive matches; per the workspace convention that means a minor bump to 0.4.0 across crates, recorded in the changelog. --- CHANGELOG.md | 12 + Cargo.lock | 34 +-- Cargo.toml | 32 +-- serdes-ai-agent/src/lib.rs | 4 +- serdes-ai-agent/src/run.rs | 12 + serdes-ai-agent/src/steering.rs | 223 +++++++++++++++ serdes-ai-agent/src/stream.rs | 462 +++++++++++++++++++++++++++++++- 7 files changed, 744 insertions(+), 35 deletions(-) create mode 100644 serdes-ai-agent/src/steering.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index a74fc5f..b4dc90c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.4.0] - 2026-09-05 + +### Added +- Steering input for streaming runs: a message typed while the agent is working can now reach the running turn instead of waiting for it to finish (`serdes-ai-agent`): + - New `SteeringQueue`, cheaply cloneable, constructible standalone, with `steer(text: String)` to enqueue from any task and `pending_len()` to read how many texts are queued but not yet delivered. Internally a tokio unbounded mpsc; the run claims the single receiver for its lifetime. + - `RunOptions::steering(queue)` attaches a queue to a run, mirroring `message_history`. Clone the queue beforehand wherever user input arrives. + - Queued texts are drained FIFO at the tool-call boundary only: after the step's tool returns are appended to the history and before the loop issues the next model request. Each drained text is appended as its own `ModelRequest` carrying a user prompt part, and one `AgentStreamEvent::SteeringDelivered { step, text }` is emitted per text after that step's `ToolExecuted` events and before the next `RequestStart`, so consumers can persist the user message in transcript order. Both `AgentStream::new` and `AgentStream::new_with_cancel` drain. + - Nothing is drained before the first model request, and a run that never crosses a tool-call boundary (text-only completion, error, cancellation) delivers nothing: leftovers stay queued and a later run created from the same queue delivers them at its first tool-call boundary. + +### Breaking +- `AgentStreamEvent` gains `SteeringDelivered { step, text }`. The enum is not `#[non_exhaustive]`, so downstream `match` sites that list every variant without a wildcard need the new arm. Adding a public type and variant is a minor-version bump for this 0.x workspace: `0.4.0` across all crates. + ## [0.3.0] - 2026-08-24 Combined release integrating PRs #51, #52, #53, #54 and #55. Streaming is now a diff --git a/Cargo.lock b/Cargo.lock index 90c70d3..cf1ea09 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3007,7 +3007,7 @@ dependencies = [ [[package]] name = "serdes-ai" -version = "0.3.0" +version = "0.4.0" dependencies = [ "futures", "pretty_assertions", @@ -3034,7 +3034,7 @@ dependencies = [ [[package]] name = "serdes-ai-a2a" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "axum", @@ -3057,7 +3057,7 @@ dependencies = [ [[package]] name = "serdes-ai-agent" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3084,7 +3084,7 @@ dependencies = [ [[package]] name = "serdes-ai-core" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "base64", @@ -3108,7 +3108,7 @@ dependencies = [ [[package]] name = "serdes-ai-embeddings" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3124,7 +3124,7 @@ dependencies = [ [[package]] name = "serdes-ai-evals" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3145,7 +3145,7 @@ dependencies = [ [[package]] name = "serdes-ai-graph" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3161,7 +3161,7 @@ dependencies = [ [[package]] name = "serdes-ai-macros" -version = "0.3.0" +version = "0.4.0" dependencies = [ "darling", "proc-macro2", @@ -3171,7 +3171,7 @@ dependencies = [ [[package]] name = "serdes-ai-mcp" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "base64", @@ -3194,7 +3194,7 @@ dependencies = [ [[package]] name = "serdes-ai-models" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3228,7 +3228,7 @@ dependencies = [ [[package]] name = "serdes-ai-output" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3251,7 +3251,7 @@ dependencies = [ [[package]] name = "serdes-ai-providers" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "base64", @@ -3273,7 +3273,7 @@ dependencies = [ [[package]] name = "serdes-ai-retries" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3289,7 +3289,7 @@ dependencies = [ [[package]] name = "serdes-ai-streaming" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "bytes", @@ -3310,7 +3310,7 @@ dependencies = [ [[package]] name = "serdes-ai-tools" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "async-trait", @@ -3334,7 +3334,7 @@ dependencies = [ [[package]] name = "serdes-ai-toolsets" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "indexmap", @@ -3352,7 +3352,7 @@ dependencies = [ [[package]] name = "serdes-ai-ui" -version = "0.3.0" +version = "0.4.0" dependencies = [ "async-trait", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 9420425..28b5822 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,7 @@ members = [ ] [workspace.package] -version = "0.3.0" +version = "0.4.0" edition = "2021" license = "MIT" authors = ["serdes-ai contributors"] @@ -93,21 +93,21 @@ pretty_assertions = "1.4" regex = "1.12" # Internal crates -serdes-ai-core = { path = "serdes-ai-core", version = "0.3.0" } -serdes-ai-agent = { path = "serdes-ai-agent", version = "0.3.0" } -serdes-ai-models = { path = "serdes-ai-models", version = "0.3.0" } -serdes-ai-providers = { path = "serdes-ai-providers", version = "0.3.0" } -serdes-ai-tools = { path = "serdes-ai-tools", version = "0.3.0" } -serdes-ai-toolsets = { path = "serdes-ai-toolsets", version = "0.3.0" } -serdes-ai-output = { path = "serdes-ai-output", version = "0.3.0" } -serdes-ai-streaming = { path = "serdes-ai-streaming", version = "0.3.0" } -serdes-ai-mcp = { path = "serdes-ai-mcp", version = "0.3.0" } -serdes-ai-embeddings = { path = "serdes-ai-embeddings", version = "0.3.0" } -serdes-ai-retries = { path = "serdes-ai-retries", version = "0.3.0" } -serdes-ai-graph = { path = "serdes-ai-graph", version = "0.3.0" } -serdes-ai-evals = { path = "serdes-ai-evals", version = "0.3.0" } -serdes-ai-macros = { path = "serdes-ai-macros", version = "0.3.0" } -serdes-ai-a2a = { path = "serdes-ai-a2a", version = "0.3.0" } +serdes-ai-core = { path = "serdes-ai-core", version = "0.4.0" } +serdes-ai-agent = { path = "serdes-ai-agent", version = "0.4.0" } +serdes-ai-models = { path = "serdes-ai-models", version = "0.4.0" } +serdes-ai-providers = { path = "serdes-ai-providers", version = "0.4.0" } +serdes-ai-tools = { path = "serdes-ai-tools", version = "0.4.0" } +serdes-ai-toolsets = { path = "serdes-ai-toolsets", version = "0.4.0" } +serdes-ai-output = { path = "serdes-ai-output", version = "0.4.0" } +serdes-ai-streaming = { path = "serdes-ai-streaming", version = "0.4.0" } +serdes-ai-mcp = { path = "serdes-ai-mcp", version = "0.4.0" } +serdes-ai-embeddings = { path = "serdes-ai-embeddings", version = "0.4.0" } +serdes-ai-retries = { path = "serdes-ai-retries", version = "0.4.0" } +serdes-ai-graph = { path = "serdes-ai-graph", version = "0.4.0" } +serdes-ai-evals = { path = "serdes-ai-evals", version = "0.4.0" } +serdes-ai-macros = { path = "serdes-ai-macros", version = "0.4.0" } +serdes-ai-a2a = { path = "serdes-ai-a2a", version = "0.4.0" } [profile.dev] split-debuginfo = "unpacked" diff --git a/serdes-ai-agent/src/lib.rs b/serdes-ai-agent/src/lib.rs index 1bb10e5..5e0e9ad 100644 --- a/serdes-ai-agent/src/lib.rs +++ b/serdes-ai-agent/src/lib.rs @@ -79,6 +79,7 @@ pub mod history; pub mod instructions; pub mod output; pub mod run; +pub mod steering; pub mod stream; // Re-exports @@ -105,6 +106,7 @@ pub use output::{ pub use run::{ AgentRun, AgentRunResult, CompressionStrategy, ContextCompression, RunOptions, StepResult, }; +pub use steering::SteeringQueue; pub use stream::{AgentStream, AgentStreamEvent}; // Re-export CancellationToken for convenience @@ -116,7 +118,7 @@ pub mod prelude { agent, agent_with_deps, Agent, AgentBuilder, AgentRun, AgentRunError, AgentRunResult, AgentStream, AgentStreamEvent, CancellationToken, CompressionStrategy, ContextCompression, EndStrategy, OutputMode, OutputSchema, OutputValidator, RunContext, RunOptions, RunUsage, - StepResult, UsageLimits, + SteeringQueue, StepResult, UsageLimits, }; } diff --git a/serdes-ai-agent/src/run.rs b/serdes-ai-agent/src/run.rs index e07f96d..dd35da5 100644 --- a/serdes-ai-agent/src/run.rs +++ b/serdes-ai-agent/src/run.rs @@ -5,6 +5,7 @@ use crate::agent::{Agent, EndStrategy}; use crate::context::{generate_run_id, RunContext, RunUsage, UsageLimits}; use crate::errors::{AgentRunError, OutputParseError, OutputValidationError}; +use crate::steering::SteeringQueue; use chrono::Utc; use serde_json::Value as JsonValue; use serdes_ai_core::messages::{RetryPromptPart, ToolCallArgs, ToolReturnPart, UserContent}; @@ -54,6 +55,8 @@ pub struct RunOptions { pub model_settings: Option, /// Message history to continue from. pub message_history: Option>, + /// Steering input queue for this run. + pub steering: Option, /// Usage limits for this run. pub usage_limits: Option, /// Custom metadata. @@ -80,6 +83,15 @@ impl RunOptions { self } + /// Attach a steering input queue. + /// + /// Queued texts are delivered to the model at tool-call boundaries only; + /// see [`SteeringQueue`] for the exact delivery rules. + pub fn steering(mut self, queue: SteeringQueue) -> Self { + self.steering = Some(queue); + self + } + /// Set metadata. pub fn metadata(mut self, metadata: JsonValue) -> Self { self.metadata = Some(metadata); diff --git a/serdes-ai-agent/src/steering.rs b/serdes-ai-agent/src/steering.rs new file mode 100644 index 0000000..e614de3 --- /dev/null +++ b/serdes-ai-agent/src/steering.rs @@ -0,0 +1,223 @@ +//! Steering input for running agent streams. +//! +//! A [`SteeringQueue`] lets a caller inject user text into a run that is +//! already in flight, instead of waiting for the current turn to finish. The +//! queued text is delivered to the model at the next tool-call boundary: after +//! the step's tool returns are appended to the history and before the loop +//! issues the next model request. +//! +//! # Delivery rules +//! +//! - Messages are drained FIFO at tool-call boundaries only. +//! - Messages enqueued before the first model request are NOT delivered +//! immediately; they wait for the first tool-call boundary. +//! - A run that ends without ever crossing a tool-call boundary (text-only +//! completion, error, cancellation) delivers nothing. Undelivered texts stay +//! queued for the caller and survive the run: a later run created from the +//! same queue delivers them at its first tool-call boundary. +//! - A run claims the single channel receiver for its lifetime; concurrent +//! runs on the same queue are not supported (the second run gets no +//! receiver and never drains). +//! +//! # Example +//! +//! ```rust +//! use serdes_ai_agent::{RunOptions, SteeringQueue}; +//! +//! let queue = SteeringQueue::new(); +//! let options = RunOptions::new().steering(queue.clone()); +//! +//! // From any task, e.g. a UI input handler, while the agent is working: +//! queue.steer("focus on the error case".to_string()); +//! +//! // Not yet delivered; delivery happens at the next tool-call boundary. +//! assert_eq!(queue.pending_len(), 1); +//! ``` + +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use parking_lot::Mutex; +use tokio::sync::mpsc; + +/// A cheaply cloneable queue of steering texts for a running agent stream. +/// +/// Clone it freely: every clone feeds the same channel. Pass one clone to the +/// run via [`crate::RunOptions::steering`] and keep another wherever user +/// input arrives. +/// +/// Texts are drained by the agent stream at tool-call boundaries only, never +/// before the first model request and never at a text-only end of turn. +/// Undelivered texts stay queued for the caller across runs. +#[derive(Debug, Clone)] +pub struct SteeringQueue { + tx: mpsc::UnboundedSender, + /// Number of texts enqueued but not yet drained by a run. Authoritative + /// for `pending_len` because the channel receiver cannot report a length. + pending: Arc, + /// The single channel receiver, parked here until a run claims it via + /// [`SteeringQueue::take_receiver`] and parked back when that run's + /// receiver is dropped. + rx_slot: Arc>>>, +} + +impl SteeringQueue { + /// Create a new empty steering queue. + pub fn new() -> Self { + let (tx, rx) = mpsc::unbounded_channel(); + Self { + tx, + pending: Arc::new(AtomicUsize::new(0)), + rx_slot: Arc::new(Mutex::new(Some(rx))), + } + } + + /// Enqueue `text` for delivery at the next tool-call boundary. + /// + /// Returns `true` if the text was queued, `false` if the queue is closed + /// (every receiver has been dropped and none can be claimed again), in + /// which case the text is dropped. With the usual run lifecycle the + /// receiver is parked back on the queue when the run ends, so steering + /// stays possible between runs and returns `true`. + pub fn steer(&self, text: String) -> bool { + self.pending.fetch_add(1, Ordering::SeqCst); + if self.tx.send(text).is_err() { + self.pending.fetch_sub(1, Ordering::SeqCst); + return false; + } + true + } + + /// Number of enqueued texts not yet drained by a run. + /// + /// A positive count after a run finished means the texts were never + /// delivered (the run crossed no tool-call boundary); they remain queued. + pub fn pending_len(&self) -> usize { + self.pending.load(Ordering::SeqCst) + } + + /// Claim the single receiver for a run. Returns `None` if another run + /// already holds it (or holds it via a not-yet-dropped receiver). + pub(crate) fn take_receiver(&self) -> Option { + let rx = self.rx_slot.lock().take()?; + Some(SteeringReceiver { + rx: Some(rx), + pending: Arc::clone(&self.pending), + slot: Arc::clone(&self.rx_slot), + }) + } +} + +impl Default for SteeringQueue { + fn default() -> Self { + Self::new() + } +} + +/// The receiving half of a [`SteeringQueue`], held by a running agent stream. +/// +/// Dropping it parks the receiver (and any undrained texts inside it) back on +/// the originating queue, so leftovers survive the run that claimed them. +pub(crate) struct SteeringReceiver { + /// `Some` for the lifetime of this receiver; taken only by [`Drop`]. + rx: Option>, + pending: Arc, + slot: Arc>>>, +} + +impl SteeringReceiver { + /// Drain every currently queued text, FIFO, without waiting. + pub(crate) fn drain(&mut self) -> Vec { + let mut drained = Vec::new(); + if let Some(rx) = self.rx.as_mut() { + while let Ok(text) = rx.try_recv() { + drained.push(text); + } + } + if !drained.is_empty() { + self.pending.fetch_sub(drained.len(), Ordering::SeqCst); + } + drained + } +} + +impl Drop for SteeringReceiver { + fn drop(&mut self) { + // Park the receiver (with any undrained texts) back on the queue so + // leftovers survive this run and a later run can deliver them. + *self.slot.lock() = self.rx.take(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn steer_before_any_run_stays_pending() { + let queue = SteeringQueue::new(); + assert_eq!(queue.pending_len(), 0); + assert!(queue.steer("one".to_string())); + assert_eq!(queue.pending_len(), 1); + } + + #[test] + fn drain_delivers_fifo_and_empties_pending() { + let queue = SteeringQueue::new(); + queue.steer("one".to_string()); + queue.steer("two".to_string()); + queue.steer("three".to_string()); + + let mut rx = queue.take_receiver().expect("receiver not yet claimed"); + assert_eq!(rx.drain(), vec!["one", "two", "three"]); + assert_eq!(queue.pending_len(), 0); + } + + #[test] + fn drain_only_takes_what_is_queued_and_never_blocks() { + let queue = SteeringQueue::new(); + let mut rx = queue.take_receiver().expect("receiver not yet claimed"); + assert!(rx.drain().is_empty()); + assert!(queue.steer("late".to_string())); + assert_eq!(rx.drain(), vec!["late"]); + assert!(rx.drain().is_empty()); + } + + #[test] + fn receiver_returned_on_drop_keeps_undrained_texts() { + let queue = SteeringQueue::new(); + queue.steer("survivor".to_string()); + { + let _rx = queue.take_receiver().expect("receiver not yet claimed"); + // Dropped without draining anything. + } + + assert_eq!(queue.pending_len(), 1); + + let mut rx = queue.take_receiver().expect("receiver returned on drop"); + assert_eq!(rx.drain(), vec!["survivor"]); + assert_eq!(queue.pending_len(), 0); + } + + #[test] + fn take_receiver_is_single_shot_until_returned() { + let queue = SteeringQueue::new(); + let held = queue.take_receiver(); + assert!(held.is_some()); + assert!(queue.take_receiver().is_none()); + // Only once the holder is dropped does the receiver come back. + drop(held); + assert!(queue.take_receiver().is_some()); + } + + #[test] + fn clones_feed_the_same_queue() { + let queue = SteeringQueue::new(); + let clone = queue.clone(); + assert!(clone.steer("via clone".to_string())); + assert_eq!(queue.pending_len(), 1); + + let mut rx = queue.take_receiver().expect("receiver not yet claimed"); + assert_eq!(rx.drain(), vec!["via clone"]); + } +} diff --git a/serdes-ai-agent/src/stream.rs b/serdes-ai-agent/src/stream.rs index bb3822f..913f004 100644 --- a/serdes-ai-agent/src/stream.rs +++ b/serdes-ai-agent/src/stream.rs @@ -7,6 +7,7 @@ use crate::agent::{Agent, RegisteredTool}; use crate::context::{generate_run_id, RunContext, RunUsage}; use crate::errors::AgentRunError; use crate::run::{CompressionStrategy, RunOptions}; +use crate::steering::{SteeringQueue, SteeringReceiver}; use chrono::Utc; use futures::{Stream, StreamExt}; use serdes_ai_core::messages::{ @@ -96,6 +97,17 @@ pub enum AgentStreamEvent { success: bool, error: Option, }, + /// Steering input was delivered to the run at the tool-call boundary. + /// + /// Emitted after the step's `ToolExecuted` events and before the next + /// `RequestStart`, so consumers can persist the steered user message in + /// transcript order. + SteeringDelivered { + /// The step whose tool-call boundary delivered this message. + step: u32, + /// The steered text, also appended to history as a user prompt. + text: String, + }, /// Thinking delta (for reasoning models). ThinkingDelta { text: String }, /// Model response completed. @@ -192,6 +204,32 @@ fn usage_from_stream_complete(event: &StreamCompleteEvent) -> Option, + messages: &mut Vec, + tx: &mpsc::Sender>, + step: u32, +) { + let Some(receiver) = receiver.as_mut() else { + return; + }; + for text in receiver.drain() { + let mut steer_req = ModelRequest::new(); + steer_req.add_user_prompt(text.clone()); + messages.push(steer_req); + let _ = tx + .send(Ok(AgentStreamEvent::SteeringDelivered { step, text })) + .await; + } +} + impl AgentStream { /// Create a new streaming agent run. /// @@ -237,6 +275,10 @@ impl AgentStream { let _metadata = options.metadata.clone(); let compression_config = options.compression.clone(); let run_id_clone = run_id.clone(); + let mut steering_rx = options + .steering + .as_ref() + .and_then(SteeringQueue::take_receiver); debug!(run_id = %run_id, "AgentStream: spawning streaming task"); @@ -866,6 +908,10 @@ impl AgentStream { messages.push(tool_req); } + // Deliver steering queued while the tools ran, before the + // next model request is issued. + deliver_queued_steering(&mut steering_rx, &mut messages, &tx, step).await; + // Continue to let model respond to tool "error" continue; } @@ -973,6 +1019,10 @@ impl AgentStream { let compression_config = options.compression.clone(); let run_id_clone = run_id.clone(); let cancel_token_clone = cancel_token.clone(); + let mut steering_rx = options + .steering + .as_ref() + .and_then(SteeringQueue::take_receiver); debug!(run_id = %run_id, "AgentStream: spawning streaming task with cancellation support"); @@ -1475,6 +1525,10 @@ impl AgentStream { messages.push(tool_req); } + // Deliver steering queued while the tools ran, before the + // next model request is issued. + deliver_queued_steering(&mut steering_rx, &mut messages, &tx, step).await; + continue; } @@ -1572,7 +1626,7 @@ mod tests { use serdes_ai_models::FunctionModel; use std::sync::{ atomic::{AtomicUsize, Ordering}, - Arc, + Arc, Mutex, }; #[test] @@ -3207,4 +3261,410 @@ mod tests { "exactly one terminal RunComplete must fire (no premature termination)" ); } + + // ======================================================================== + // Steering delivery tests: queued texts reach the model at the + // tool-call boundary, FIFO, never before the first request and never + // at a text-only end of turn. + // ======================================================================== + + /// Collect the text of every user prompt across a captured request + /// history, in order. + fn user_texts(requests: &[ModelRequest]) -> Vec { + requests + .iter() + .flat_map(|req| req.user_prompts()) + .filter_map(|part| part.as_text().map(str::to_string)) + .collect() + } + + /// Two-request mock model: the first call returns a tool call, later + /// calls return terminal text. Every incoming request history is recorded + /// for assertions. + fn tool_then_text_model(recorded: &Arc>>>) -> FunctionModel { + let recorded = Arc::clone(recorded); + FunctionModel::with_stream(move |messages: &[ModelRequest], _settings| { + let call = { + let mut log = recorded.lock().expect("request log poisoned"); + log.push(messages.to_vec()); + log.len() - 1 + }; + let events = if call == 0 { + vec![ + Ok(ModelResponseStreamEvent::part_start( + 0, + ModelResponsePart::ToolCall( + ToolCallPart::new("demo_tool", ToolCallArgs::string("{}")) + .with_tool_call_id("call_1"), + ), + )), + Ok(ModelResponseStreamEvent::part_end(0)), + Ok(ModelResponseStreamEvent::StreamComplete( + StreamCompleteEvent::new(FinishReason::ToolCall), + )), + ] + } else { + vec![ + Ok(ModelResponseStreamEvent::part_start( + 0, + ModelResponsePart::Text(TextPart::new("done")), + )), + Ok(ModelResponseStreamEvent::part_end(0)), + Ok(ModelResponseStreamEvent::StreamComplete( + StreamCompleteEvent::new(FinishReason::Stop), + )), + ] + }; + Box::pin(stream::iter(events)) + }) + } + + /// A steer enqueued while the step-0 tool runs is delivered at that + /// step's tool-call boundary: it appears in the SECOND model request + /// (after the tool return), and `SteeringDelivered` is emitted after + /// `ToolExecuted` and before the next `RequestStart`. + #[tokio::test] + async fn test_steering_delivered_at_tool_boundary_in_second_request() { + let recorded = Arc::new(Mutex::new(Vec::>::new())); + let model = tool_then_text_model(&recorded); + let queue = SteeringQueue::new(); + let tool_queue = queue.clone(); + + let agent = agent(model) + .tool_fn( + "demo_tool", + "Demo tool", + move |_ctx, _args: serde_json::Value| { + tool_queue.steer("check the weather in tokyo".to_string()); + Ok(serdes_ai_tools::ToolReturn::text("ok")) + }, + ) + .build(); + + let options = RunOptions::new().steering(queue.clone()); + let mut stream = agent + .run_stream_with_options("trigger tool then finish", (), options) + .await + .expect("stream should start"); + + let mut order: Vec = Vec::new(); + let mut steering_events = Vec::new(); + while let Some(event) = stream.next().await { + match event.expect("stream event should be ok") { + AgentStreamEvent::RequestStart { step } => { + order.push(format!("request_start:{step}")); + } + AgentStreamEvent::ToolExecuted { ref tool_name, .. } + if tool_name == "demo_tool" => + { + order.push("tool_executed".to_string()); + } + AgentStreamEvent::SteeringDelivered { step, text } => { + order.push(format!("steering:{step}")); + steering_events.push((step, text)); + } + AgentStreamEvent::RunComplete { .. } => order.push("run_complete".to_string()), + _ => {} + } + } + + // Delivered exactly once, for the step whose tool call triggered it. + assert_eq!( + steering_events, + vec![(1, "check the weather in tokyo".to_string())] + ); + + // Event order: after the tool execution, before the next RequestStart. + let tool_pos = order + .iter() + .position(|e| e == "tool_executed") + .expect("tool executed event"); + let steering_pos = order + .iter() + .position(|e| e == "steering:1") + .expect("steering event"); + let request2_pos = order + .iter() + .position(|e| e == "request_start:2") + .expect("second RequestStart"); + assert!(tool_pos < steering_pos, "order was: {order:?}"); + assert!(steering_pos < request2_pos, "order was: {order:?}"); + + // The second model request carries the tool return AND the steered + // user prompt; the first request does not see the steer. + let log = recorded.lock().expect("request log poisoned"); + assert_eq!(log.len(), 2, "expected exactly two model requests"); + assert_eq!( + user_texts(&log[0]), + vec!["trigger tool then finish".to_string()] + ); + assert_eq!( + user_texts(&log[1]), + vec![ + "trigger tool then finish".to_string(), + "check the weather in tokyo".to_string(), + ] + ); + assert_eq!( + log[1].iter().flat_map(|req| req.tool_returns()).count(), + 1, + "the tool return precedes the steered prompt in the same request" + ); + // The delivered text left the queue. + assert_eq!(queue.pending_len(), 0); + } + + /// Multiple steers deliver FIFO at one boundary, whether enqueued before + /// the run started or while its tool was executing. Nothing is drained + /// into the FIRST request. + #[tokio::test] + async fn test_steering_multiple_messages_deliver_fifo_at_one_boundary() { + let recorded = Arc::new(Mutex::new(Vec::>::new())); + let model = tool_then_text_model(&recorded); + let queue = SteeringQueue::new(); + // Enqueued before the run: must NOT reach the first model request. + queue.steer("queued before the run".to_string()); + let tool_queue = queue.clone(); + + let agent = agent(model) + .tool_fn( + "demo_tool", + "Demo tool", + move |_ctx, _args: serde_json::Value| { + tool_queue.steer("first mid-run".to_string()); + tool_queue.steer("second mid-run".to_string()); + Ok(serdes_ai_tools::ToolReturn::text("ok")) + }, + ) + .build(); + + let options = RunOptions::new().steering(queue.clone()); + let mut stream = agent + .run_stream_with_options("trigger tool then finish", (), options) + .await + .expect("stream should start"); + + let mut delivered = Vec::new(); + while let Some(event) = stream.next().await { + if let AgentStreamEvent::SteeringDelivered { text, .. } = + event.expect("stream event should be ok") + { + delivered.push(text); + } + } + + assert_eq!( + delivered, + vec![ + "queued before the run".to_string(), + "first mid-run".to_string(), + "second mid-run".to_string(), + ] + ); + + let log = recorded.lock().expect("request log poisoned"); + assert_eq!(log.len(), 2); + assert_eq!( + user_texts(&log[0]), + vec!["trigger tool then finish".to_string()] + ); + assert_eq!( + user_texts(&log[1]), + vec![ + "trigger tool then finish".to_string(), + "queued before the run".to_string(), + "first mid-run".to_string(), + "second mid-run".to_string(), + ] + ); + assert_eq!(queue.pending_len(), 0); + } + + /// A text-only stream response for the plain-question tests. + fn text_only_stream_events( + ) -> Vec> { + vec![ + Ok(ModelResponseStreamEvent::part_start( + 0, + ModelResponsePart::Text(TextPart::new("all done")), + )), + Ok(ModelResponseStreamEvent::part_end(0)), + Ok(ModelResponseStreamEvent::StreamComplete( + StreamCompleteEvent::new(FinishReason::Stop), + )), + ] + } + + /// A run that never crosses a tool boundary delivers nothing: the steered + /// text stays queued for the caller after the run ends. + #[tokio::test] + async fn test_steering_text_only_run_leaves_text_queued() { + let model = FunctionModel::with_stream(|_messages: &[ModelRequest], _settings| { + Box::pin(stream::iter(text_only_stream_events())) + }); + + let queue = SteeringQueue::new(); + queue.steer("never delivered".to_string()); + + let agent = agent(model).build(); + let options = RunOptions::new().steering(queue.clone()); + let mut stream = agent + .run_stream_with_options("plain question", (), options) + .await + .expect("stream should start"); + + let mut saw_steering = false; + let mut saw_run_complete = false; + while let Some(event) = stream.next().await { + match event.expect("stream event should be ok") { + AgentStreamEvent::SteeringDelivered { .. } => saw_steering = true, + AgentStreamEvent::RunComplete { .. } => saw_run_complete = true, + _ => {} + } + } + + assert!(saw_run_complete, "run should complete normally"); + assert!(!saw_steering, "no tool boundary, so nothing is delivered"); + assert_eq!( + queue.pending_len(), + 1, + "the leftover stays queued for the caller" + ); + } + + /// A leftover from a text-only run survives that run and is delivered by + /// a LATER run created from the same queue, at that run's first + /// tool-call boundary. + #[tokio::test] + async fn test_steering_leftover_delivered_by_follow_up_run() { + let queue = SteeringQueue::new(); + queue.steer("carried over".to_string()); + + // Run 1: text-only, leaves the steer queued. + let text_model = FunctionModel::with_stream(|_messages: &[ModelRequest], _settings| { + Box::pin(stream::iter(text_only_stream_events())) + }); + let plain_agent = agent(text_model).build(); + let options = RunOptions::new().steering(queue.clone()); + let mut stream = plain_agent + .run_stream_with_options("plain question", (), options) + .await + .expect("stream should start"); + while let Some(event) = stream.next().await { + event.expect("stream event should be ok"); + } + assert_eq!(queue.pending_len(), 1); + + // Run 2: crosses a tool boundary, delivers the leftover. + let recorded = Arc::new(Mutex::new(Vec::>::new())); + let model = tool_then_text_model(&recorded); + let tool_agent = agent(model) + .tool_fn( + "demo_tool", + "Demo tool", + |_ctx, _args: serde_json::Value| Ok(serdes_ai_tools::ToolReturn::text("ok")), + ) + .build(); + let options = RunOptions::new().steering(queue.clone()); + let mut stream = tool_agent + .run_stream_with_options("trigger tool then finish", (), options) + .await + .expect("stream should start"); + + let mut delivered = Vec::new(); + while let Some(event) = stream.next().await { + if let AgentStreamEvent::SteeringDelivered { text, .. } = + event.expect("stream event should be ok") + { + delivered.push(text); + } + } + + assert_eq!(delivered, vec!["carried over".to_string()]); + let log = recorded.lock().expect("request log poisoned"); + assert_eq!( + user_texts(&log[1]), + vec![ + "trigger tool then finish".to_string(), + "carried over".to_string(), + ] + ); + assert_eq!(queue.pending_len(), 0); + } + + /// The cancellable loop (`new_with_cancel`) drains at the boundary too, + /// with the same event-ordering guarantees. + #[tokio::test] + async fn test_steering_delivered_through_new_with_cancel() { + let recorded = Arc::new(Mutex::new(Vec::>::new())); + let model = tool_then_text_model(&recorded); + let queue = SteeringQueue::new(); + let tool_queue = queue.clone(); + + let agent = agent(model) + .tool_fn( + "demo_tool", + "Demo tool", + move |_ctx, _args: serde_json::Value| { + tool_queue.steer("cancel path steer".to_string()); + Ok(serdes_ai_tools::ToolReturn::text("ok")) + }, + ) + .build(); + + let options = RunOptions::new().steering(queue.clone()); + let token = CancellationToken::new(); + let mut stream = AgentStream::new_with_cancel( + &agent, + "trigger tool then finish".into(), + (), + options, + token, + ) + .await + .expect("stream should start"); + + let mut order: Vec = Vec::new(); + while let Some(event) = stream.next().await { + match event.expect("stream event should be ok") { + AgentStreamEvent::RequestStart { step } => { + order.push(format!("request_start:{step}")); + } + AgentStreamEvent::ToolExecuted { ref tool_name, .. } + if tool_name == "demo_tool" => + { + order.push("tool_executed".to_string()); + } + AgentStreamEvent::SteeringDelivered { .. } => order.push("steering".to_string()), + AgentStreamEvent::RunComplete { .. } => order.push("run_complete".to_string()), + _ => {} + } + } + + let tool_pos = order + .iter() + .position(|e| e == "tool_executed") + .expect("tool executed event"); + let steering_pos = order + .iter() + .position(|e| e == "steering") + .expect("steering event"); + let request2_pos = order + .iter() + .position(|e| e == "request_start:2") + .expect("second RequestStart"); + assert!(tool_pos < steering_pos, "order was: {order:?}"); + assert!(steering_pos < request2_pos, "order was: {order:?}"); + + let log = recorded.lock().expect("request log poisoned"); + assert_eq!(log.len(), 2); + assert_eq!( + user_texts(&log[1]), + vec![ + "trigger tool then finish".to_string(), + "cancel path steer".to_string(), + ] + ); + assert_eq!(queue.pending_len(), 0); + } }