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); + } }