From 212f0dc45916f309a41be13b9e14fc24e3460e0b Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Sat, 1 Aug 2026 23:35:00 +1200 Subject: [PATCH 1/3] feat: streaming dep optional, non streaming path, streaming flag, release profile --- CHANGELOG.md | 35 ++- Cargo.toml | 18 +- README.md | 37 ++- examples/chat.rs | 35 ++- src/capabilities.rs | 4 +- src/engine/bare.rs | 522 +++++++++++++++++++++++++++++++++--- src/engine/bare/compact.rs | 35 +-- src/engine/bare/dispatch.rs | 38 +-- src/engine/bare/emission.rs | 18 +- src/engine/bare/stream.rs | 8 +- src/managers.rs | 23 +- src/observer.rs | 6 +- src/provider.rs | 7 + src/reflection/llm.rs | 84 +++--- src/stream.rs | 4 +- src/stream/rate_limit.rs | 5 +- src/structured.rs | 4 +- 17 files changed, 671 insertions(+), 212 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7adefde..e28be24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,7 +7,40 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ## [Unreleased] -## [0.2.0] - 2026-08-01 +### Added + +- **Non-streaming engine turn path** (`TurnMode`): the engine can now drive each + turn via `ApiClient::create_message` instead of streaming, selected at runtime + with `BareLoop::set_turn_mode` / `with_turn_mode`. The default is + `TurnMode::Streaming` when `streaming` is compiled in, `TurnMode::NonStreaming` + otherwise. Under the non-streaming path no per-delta observer callbacks fire; + the full assistant text still surfaces via `on_response`. +- **`streaming` feature flag**: gates `StreamHandler`, `stream::handler`, + `StreamCapable`, `text_streamer`, and the `on_text_delta` / `on_thinking_delta` + firing sites. With `default = []`, `async-stream` is no longer pulled and the + engine compiles and runs without any streaming machinery. + +### Changed + +- **`default = []` now means non-streaming.** `async-stream` is an optional + dependency enabled by `streaming`; it is no longer pulled into a bare + `cargo add loopctl`. The HTTP provider features (`openai`, `anthropic`, + `gemini`, and anything that chains from them) now imply `streaming`, so the + common `features = ["openai"]` case preserves the previous streaming-by-default + behavior. Migration: users who enabled only `providers` (not a named provider) + and relied on streaming must add `streaming` explicitly. +- **MSRV corrected to 1.94** (was misdocumented as 1.85 in `AGENTS.md`; the + `Cargo.toml` `rust-version` and `.clippy.toml` already required 1.94). +- `AGENTS.md` feature table, `providers` dependency list, core-deps list, and the + `LoopMemory` object-safety note corrected to match the code. + +### Removed + +- The `record_stream_success` / `record_stream_failure` private methods are + renamed to `record_turn_success` / `record_turn_failure` (shared by both turn + paths). The `StreamCapable` trait and its `LoopManagers` impl now require the + `streaming` feature. + ### Added diff --git a/Cargo.toml b/Cargo.toml index 281d74e..5b54692 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ tracing = "0.1" reqwest = { version = "0.13", default-features = false, features = ["json", "stream", "rustls"], optional = true } bytes = { version = "1", optional = true } -async-stream = "0.3" +async-stream = { version = "0.3", optional = true } httpdate = { version = "1", optional = true } jsonschema = { version = "0.49", optional = true } @@ -45,15 +45,18 @@ testing = [] tool_health = [] tool_shield = ["tool_health"] +# Streaming engine path (StreamHandler, per-delta callbacks, async-stream codegen). +streaming = ["dep:async-stream"] + # Providers providers = ["dep:reqwest", "dep:httpdate", "dep:bytes"] -openai = ["providers"] -anthropic = ["providers"] +openai = ["providers", "streaming"] +anthropic = ["providers", "streaming"] ollama = ["providers", "openai"] deepseek = ["providers", "openai"] grok = ["providers", "openai"] xai = ["grok"] -gemini = ["providers"] +gemini = ["providers", "streaming"] zai = ["providers", "anthropic"] grammar = ["providers"] schema_validation = ["dep:jsonschema"] @@ -92,3 +95,10 @@ let_underscore_must_use = "deny" [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] + +[profile.release] +lto = "fat" +codegen-units = 1 +strip = true +panic = "abort" + diff --git a/README.md b/README.md index 01bc4d8..32def47 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ A trait-based framework for building agent loops with pluggable LLM clients, too [![crates.io](https://img.shields.io/crates/v/loopctl.svg)](https://crates.io/crates/loopctl) [![docs.rs](https://docs.rs/loopctl/badge.svg)](https://docs.rs/loopctl) -[![license](https://img.shields.io/badge/license-MIT%2FApache--2.0-blue.svg)](LICENSE-MIT) +[![license](https://img.shields.io/badge/license-MIT%2FApache--2.0-blue.svg)](LICENSE) ## Overview @@ -146,25 +146,42 @@ let agent = BareLoop::new( | `testing` | No | — | Mock clients, tools, and test fixtures | | `tool_health` | No | — | Per-tool health monitoring, circuit breakers, and self-healing routing | | `tool_shield` | No | `tool_health` | Tool permission shielding and access control | -| `providers` | No | — | Base provider support (`reqwest` + `async-stream`); enables `provider` module | -| `openai` | No | `providers` | OpenAI-compatible API client (`provider::openai`) | -| `anthropic` | No | `providers` | Anthropic Claude API client (`provider::anthropic`) | +| `streaming` | No | `async-stream` | Streaming engine path: `StreamHandler` (retry, timeout, fallback), per-delta observer callbacks (`on_text_delta`, `on_thinking_delta`), `text_streamer`. Without it the engine drives each turn via `ApiClient::create_message`. | +| `providers` | No | `reqwest`, `httpdate`, `bytes` | Base HTTP provider support; enables the `provider` module | +| `openai` | No | `providers`, `streaming` | OpenAI-compatible API client (`provider::openai`) | +| `anthropic` | No | `providers`, `streaming` | Anthropic Claude API client (`provider::anthropic`) | | `ollama` | No | `providers`, `openai` | Ollama local model client (OpenAI-compatible) | | `deepseek` | No | `providers`, `openai` | DeepSeek API client (OpenAI-compatible) | | `grok` | No | `providers`, `openai` | Grok (xAI) API client (OpenAI-compatible) | | `xai` | No | `grok` | Alias for `grok` (xAI API client) | -| `gemini` | No | `providers` | Google Gemini API client (`provider::gemini`) | +| `gemini` | No | `providers`, `streaming` | Google Gemini API client (`provider::gemini`) | | `zai` | No | `providers`, `anthropic` | Z.AI API client (Anthropic-compatible) | | `grammar` | No | `providers` | Tool-call grammar providers for grammar-aware samplers (vLLM `guided_json`); enables the `Grammar` mode of `ToolConstraint` | | `schema_validation` | No | — | JSON Schema validation of `Correction::modified_input` in `LlmReflector` (pulls `jsonschema`); when off, validation is skipped | +### Streaming vs non-streaming + +By default (`default = []`) the engine drives each turn with +[`ApiClient::create_message`] — a single request/response with no streaming +machinery, no `async-stream` dependency, and no per-delta callbacks. The full +assistant text still surfaces through +[`on_response`](https://docs.rs/loopctl/latest/loopctl/observer/trait.LoopObserver.html#method.on_response). + +Enable the `streaming` feature (implied by every HTTP provider) to route turns +through [`StreamHandler`](https://docs.rs/loopctl/latest/loopctl/stream/handler/struct.StreamHandler.html) +with retry, timeout, rate-limit detection, and `on_text_delta` / +`on_thinking_delta` callbacks for real-time token display. Switch a constructed +loop explicitly with +[`set_turn_mode`](https://docs.rs/loopctl/latest/loopctl/engine/struct.BareLoop.html#method.set_turn_mode). + ## Architecture -At the center is **BareLoop**, the default agent loop. Each turn it streams a -response from an **ApiClient** (your LLM provider), accumulates the result, and -dispatches any requested tool calls through a **ToolRegistry**. Results are fed -back into the conversation and the cycle repeats until the model ends its turn -or a configured limit is reached. +At the center is **BareLoop**, the default agent loop. Each turn it requests a +response from an **ApiClient** (your LLM provider) — via the streaming path +(`StreamHandler`) when `streaming` is enabled, or via `create_message` +otherwise — then dispatches any requested tool calls through a +**ToolRegistry**. Results are fed back into the conversation and the cycle +repeats until the model ends its turn or a configured limit is reached. Two cross-cutting concerns run alongside the main loop: diff --git a/examples/chat.rs b/examples/chat.rs index 80eccd5..e114ec0 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -333,11 +333,24 @@ async fn run_repl(client: Arc) { let mut agent = BareLoop::new(client, build_tools(), config); agent.register_observer(Arc::new(PrintingObserver)); - // Stream text deltas in real-time. - agent.set_text_streamer(Arc::new(|delta| { - print!("{delta}"); - let _ = std::io::stdout().flush(); - })); + // Pick the engine turn mode at runtime. `NO_STREAM=1` drives each turn via + // the non-streaming `create_message` path (no per-delta callbacks, the + // assembled response is printed after the turn). Otherwise stream text + // deltas live as they arrive. + let no_stream = std::env::var("NO_STREAM").map_or(false, |v| v == "1"); + #[cfg(feature = "streaming")] + if no_stream { + agent.set_turn_mode(loopctl::engine::TurnMode::NonStreaming); + } else { + agent.set_text_streamer(Arc::new(|delta| { + print!("{delta}"); + let _ = std::io::stdout().flush(); + })); + } + #[cfg(not(feature = "streaming"))] + { + let _ = no_stream; + } // Ctrl-C interrupts the in-flight turn (via loopctl's CancelSignal, which // `select!`s against the stream) and ends the session. The token is @@ -374,8 +387,16 @@ async fn run_repl(client: Arc) { match agent.run(input, &run_config).await { Ok(result) => { - // Text was already streamed live. Just print stats. - if result.output.as_deref().map_or(true, |s| s.is_empty()) { + // Under the live-streaming mode the text was already printed + // by the streamer; under non-streaming mode print the assembled + // response now. + if no_stream { + if let Some(text) = result.output.as_deref() + && !text.is_empty() + { + println!(" {text}"); + } + } else if result.output.as_deref().map_or(true, |s| s.is_empty()) { println!(" (empty response)"); } println!( diff --git a/src/capabilities.rs b/src/capabilities.rs index daef93d..af6f336 100644 --- a/src/capabilities.rs +++ b/src/capabilities.rs @@ -13,7 +13,7 @@ //! | [`Detectable`] | Loop and convergence detection | //! | [`FallbackCapable`] | Model fallback / circuit breaker | //! | [`Compactable`] | Context compaction | -//! | [`StreamCapable`] | Resilient LLM streaming | +//! | `StreamCapable` | Resilient LLM streaming *(requires `streaming` feature)* | //! | [`PipelineAware`] | Middleware pipeline dispatch | //! | `Hookable` | Bidirectional lifecycle hooks *(requires `hooks` feature)* | //! | `HealthTrackable` | Per-tool health monitoring *(requires `tool_health` feature)* | @@ -38,6 +38,7 @@ use crate::fallback::FallbackManager; use crate::hooks::HookExecutor; use crate::middleware::ToolPipeline; use crate::observer::ObserverHost; +#[cfg(feature = "streaming")] use crate::stream::handler::StreamHandler; #[cfg(feature = "tool_health")] use crate::tool::health::ToolHealthRegistry; @@ -202,6 +203,7 @@ pub trait RememberCapable { /// resilient streaming. Useful for custom loop implementations /// that need to control streaming behaviour (timeouts, retries, fallback /// to non-streaming mode). +#[cfg(feature = "streaming")] pub trait StreamCapable { /// Returns the stream handler. /// diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 7a76101..346c7ca 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -91,8 +91,9 @@ use crate::reflection::{ ExponentialBackoffRecovery, NoopReflector, RecoveryAction, RecoveryStrategy, ReflectionContext, Reflector, }; +#[cfg(feature = "streaming")] use crate::stream::handler::StreamHandler; -use crate::stream::{StreamAccumulator, StreamEvent, StreamStopReason, Usage}; +use crate::stream::{StreamStopReason, Usage}; use crate::structured::RequestOptions; #[cfg(feature = "tool_health")] use crate::tool::health::ToolHealthRegistry; @@ -102,8 +103,62 @@ mod compact; mod dispatch; mod emission; mod message; +#[cfg(feature = "streaming")] mod stream; +/// How the engine fulfils each LLM turn. +/// +/// `BareLoop` drives every turn by asking the [`ApiClient`] for a response +/// and folding the result into the conversation. Two mechanisms are +/// available, selected per turn from this enum: +/// +/// - `NonStreaming` calls [`ApiClient::create_message`] and receives a single +/// complete [`Message`](crate::message::Message). It compiles and runs with +/// no streaming dependencies, so it is the default under `default = []`. +/// - `Streaming` calls [`ApiClient::stream_messages`] through the resilient +/// `StreamHandler`, emitting per-delta observer callbacks. Requires the +/// `streaming` feature. +/// +/// The default is `Streaming` when `streaming` is compiled in, otherwise +/// `NonStreaming`. Switch modes on a constructed loop with +/// [`set_turn_mode`](BareLoop::set_turn_mode). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum TurnMode { + /// Fulfil each turn via [`ApiClient::create_message`]. + /// + /// No streaming code is exercised; `on_text_delta`, + /// `on_thinking_delta`, and the text streamer never fire. The full + /// assistant text still surfaces through + /// [`on_response`](crate::observer::LoopObserver::on_response). + #[default] + NonStreaming, + + /// Fulfil each turn via [`ApiClient::stream_messages`] wrapped in + /// [`StreamHandler`](crate::stream::handler::StreamHandler). + /// + /// Requires the `streaming` feature. Constructing or selecting this + /// mode without `streaming` enabled yields a + /// [`LoopError::Config`](crate::error::LoopError::Config) at turn time. + #[cfg(feature = "streaming")] + Streaming, +} + +/// Resolve the constructor default for [`TurnMode`]. +/// +/// Streaming when the `streaming` feature is compiled in, non-streaming +/// otherwise. Kept as a free function so both constructors share one +/// definition and the `cfg` lives in exactly one place. +fn default_turn_mode() -> TurnMode { + #[cfg(feature = "streaming")] + { + TurnMode::Streaming + } + #[cfg(not(feature = "streaming"))] + { + TurnMode::NonStreaming + } +} + /// The framework's default agent loop implementation. /// /// `BareLoop` ties together an [`ApiClient`], [`ToolRegistry`], @@ -235,7 +290,9 @@ pub struct BareLoop { /// /// Set via [`set_text_streamer`](BareLoop::set_text_streamer). /// When set, called from `stream_turn` on every `IndexedDelta` with - /// a `Text` payload, enabling real-time token display. + /// a `Text` payload, enabling real-time token display. Only read by + /// the streaming engine path; absent under `default = []`. + #[cfg(feature = "streaming")] #[allow(clippy::type_complexity)] text_streamer: Option>, @@ -256,6 +313,13 @@ pub struct BareLoop { /// [`set_request_options`](BareLoop::set_request_options). request_options: RequestOptions, + /// How each LLM turn is fulfilled (streaming vs non-streaming). + /// + /// Defaults to [`TurnMode::Streaming`] when the `streaming` feature is + /// enabled and [`TurnMode::NonStreaming`] otherwise. Set via + /// [`set_turn_mode`](BareLoop::set_turn_mode). + turn_mode: TurnMode, + /// Token counter for context-size estimates. /// /// Used by the driver to estimate the context size after each model @@ -341,10 +405,12 @@ impl BareLoop { reflector: Arc::new(NoopReflector), recovery: Arc::new(ExponentialBackoffRecovery::new(3)), cancelled: Arc::new(CancelSignal::new()), + #[cfg(feature = "streaming")] text_streamer: None, contributors: Vec::new(), request_options: RequestOptions::default(), token_counter: Arc::new(crate::compact::HeuristicTokenCounter), + turn_mode: default_turn_mode(), } } @@ -486,10 +552,12 @@ impl BareLoop { reflector: Arc::new(NoopReflector), recovery: Arc::new(ExponentialBackoffRecovery::new(3)), cancelled: Arc::new(CancelSignal::new()), + #[cfg(feature = "streaming")] text_streamer: None, contributors: Vec::new(), request_options: RequestOptions::default(), token_counter: Arc::new(crate::compact::HeuristicTokenCounter), + turn_mode: default_turn_mode(), } } @@ -690,6 +758,7 @@ impl BareLoop { self } + #[cfg(feature = "streaming")] /// Set the [`StreamHandler`] for resilient streaming with retries, /// timeouts, and fallback to non-streaming. /// @@ -874,12 +943,14 @@ impl BareLoop { self.managers.register_observer(observer); } + #[cfg(feature = "streaming")] /// Set a real-time text streaming callback. /// /// The callback is invoked for each text delta token as it arrives - /// from the API during [`run`](crate::engine::core::Loop::run). - /// This enables real-time display of the model's output without - /// waiting for the full turn to complete. + /// from the API during [`run`](crate::engine::core::Loop::run) under the + /// streaming turn mode. This enables real-time display of the model's + /// output without waiting for the full turn to complete. Requires the + /// `streaming` feature; no-op under the non-streaming path. /// /// The callback receives a `&str` containing the delta text fragment. /// It must be `Send + Sync` as it may be called from an async context. @@ -966,6 +1037,52 @@ impl BareLoop { self.request_options = options; } + /// Return the active [`TurnMode`]. + /// + /// Reflects what was set via [`set_turn_mode`](Self::set_turn_mode) or + /// the constructor default (`TurnMode::Streaming` when `streaming` is + /// compiled in, [`TurnMode::NonStreaming`] otherwise). + #[must_use] + pub fn turn_mode(&self) -> TurnMode { + self.turn_mode + } + + /// Select how the engine fulfils each LLM turn. + /// + /// Pass [`TurnMode::NonStreaming`] to drive turns through + /// [`ApiClient::create_message`] with no streaming machinery; pass + /// `TurnMode::Streaming` (requires the `streaming` feature) to drive + /// them through `StreamHandler` with per-delta observer callbacks. + /// + /// Must be called before + /// [`run`](crate::engine::core::Loop::run). + /// + /// # Panics (debug only) + /// + /// In debug builds, panics if called after the session has started + /// (i.e., once the machine has advanced past [`MachineState::Start`]). + /// + /// # Example + /// + /// ```rust,ignore + /// use loopctl::engine::{BareLoop, TurnMode}; + /// + /// let mut agent = BareLoop::new(client, registry, config); + /// agent.set_turn_mode(TurnMode::NonStreaming); + /// ``` + pub fn set_turn_mode(&mut self, mode: TurnMode) { + self.debug_assert_idle(); + self.turn_mode = mode; + } + + /// Select the turn mode, consuming `self`. Fluent mirror of + /// [`set_turn_mode`](BareLoop::set_turn_mode). + #[must_use] + pub fn with_turn_mode(mut self, mode: TurnMode) -> Self { + self.set_turn_mode(mode); + self + } + /// Set the reflector, consuming `self`. Fluent mirror of /// [`set_reflector`](BareLoop::set_reflector). #[must_use] @@ -992,6 +1109,7 @@ impl BareLoop { /// Set the stream handler, consuming `self`. Fluent mirror of /// [`set_stream_handler`](BareLoop::set_stream_handler). + #[cfg(feature = "streaming")] #[must_use] pub fn with_stream_handler(mut self, handler: StreamHandler) -> Self { self.set_stream_handler(handler); @@ -1053,6 +1171,7 @@ impl BareLoop { /// Set the real-time text streaming callback, consuming `self`. Fluent /// mirror of [`set_text_streamer`](BareLoop::set_text_streamer). + #[cfg(feature = "streaming")] #[must_use] pub fn with_text_streamer(mut self, f: Arc) -> Self { self.set_text_streamer(f); @@ -1185,6 +1304,74 @@ impl BareLoop { } } + /// Request one assistant response via the non-streaming API and apply + /// post-turn bookkeeping. + /// + /// Builds the same [`StreamRequest`](crate::api::StreamRequest) as the + /// streaming path (machine history plus contributor messages, system + /// prompt, and tool schemas), then calls + /// [`ApiClient::create_message_with_options`] and races it against + /// [`CancelSignal::notified`] so cancellation still wakes the turn. + /// No per-delta observer callbacks fire — the full assistant text + /// surfaces through + /// [`on_response`](crate::observer::LoopObserver::on_response) like any + /// other turn. + /// + /// On success: records the turn with the fallback manager and fires + /// [`on_stream_success`](crate::observer::LoopObserver::on_stream_success). + /// On a real API failure: fires + /// [`on_stream_failure`](crate::observer::LoopObserver::on_stream_failure) + /// (and [`on_fallback`](crate::observer::LoopObserver::on_fallback) if + /// the breaker trips) and returns the error. A clean cancellation is + /// *not* a failure — see the note below. + /// + /// # Errors + /// + /// [`LoopError::Cancelled`] if the cancel signal fires before or during + /// the request; otherwise whatever [`ApiError`] the client returned, + /// mapped to [`LoopError::Api`](LoopError::Api). + /// + /// # Cancellation note + /// + /// When cancel wins the inner `select!` here, it would flow into + /// [`record_turn_failure`](Self::record_turn_failure) — but in practice + /// the outer `biased` `select!` in + /// [`handle_call_llm`](Self::handle_call_llm) wins the same race and + /// drops this future first, so a clean cancel never trips the breaker + /// or fires `on_stream_failure`. Pinned by + /// `cancel_during_non_streaming_turn_does_not_trip_breaker`. + async fn do_create_message( + &mut self, + contributor_messages: Vec, + ) -> Result<(Message, Option, StreamStopReason), LoopError> { + let mut messages = contributor_messages; + messages.extend(self.machine.full_history()); + let request = crate::api::StreamRequest::new(messages) + .with_system(self.session.config.system_prompt.clone()) + .with_tools(self.build_tool_schemas()); + + if self.cancelled.is_cancelled() { + return Err(LoopError::Cancelled); + } + let cancel = Arc::clone(&self.cancelled); + let client = &self.client; + let options = self.request_options.clone(); + let result = tokio::select! { + biased; + () = cancel.notified() => Err(LoopError::Cancelled), + res = client.create_message_with_options(&request, options) => { + res.map_err(|e| LoopError::Api(e.to_string())) + } + }; + match result { + Ok(response) => { + self.record_turn_success(response.usage.as_ref()); + Ok((response.message, response.usage, response.stop_reason)) + } + Err(e) => Err(self.record_turn_failure(e)), + } + } + /// Stream one assistant response from the API and apply post-stream bookkeeping. /// /// On success: records the success with the fallback manager and fires @@ -1195,28 +1382,54 @@ impl BareLoop { /// [`on_stream_failure`](crate::observer::LoopObserver::on_stream_failure), /// sets the terminal state, and returns the error. /// - /// `LoopError::Cancelled` short-circuits the failure bookkeeping: - /// cancellation is a clean termination, so the run loop records a - /// [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome::Cancelled) + /// On a clean cancellation the outer `cancel.notified()` arm of the + /// `biased` `select!` in [`handle_call_llm`](Self::handle_call_llm) + /// wins and drops this future before + /// [`record_turn_failure`](Self::record_turn_failure) runs — so a cancel + /// during the turn records + /// [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) /// without tripping the fallback or firing `on_stream_failure`. /// /// # Errors /// /// Propagates whatever [`stream_turn`](Self::stream_turn) returned. + #[cfg(feature = "streaming")] async fn do_stream( &mut self, contributor_messages: Vec, ) -> Result<(Message, Option, StreamStopReason), LoopError> { match self.stream_turn(contributor_messages).await { Ok((msg, usage, stop)) => { - self.record_stream_success(usage.as_ref()); + self.record_turn_success(usage.as_ref()); Ok((msg, usage, stop)) } - Err(e) => Err(self.record_stream_failure(e)), + Err(e) => Err(self.record_turn_failure(e)), } } - /// Record a successful stream completion. + /// Dispatch one LLM turn according to [`turn_mode`](self.turn_mode). + /// + /// Single entry point for the run loop's `CallLLM` arm so the + /// cancellation `select!` in [`handle_call_llm`](Self::handle_call_llm) + /// races against exactly one future regardless of mode. + /// + /// # Errors + /// + /// Propagates whatever the selected turn path + /// ([`do_stream`](Self::do_stream) or + /// [`do_create_message`](Self::do_create_message)) returns. + async fn do_turn( + &mut self, + contributor_messages: Vec, + ) -> Result<(Message, Option, StreamStopReason), LoopError> { + match self.turn_mode { + #[cfg(feature = "streaming")] + TurnMode::Streaming => self.do_stream(contributor_messages).await, + TurnMode::NonStreaming => self.do_create_message(contributor_messages).await, + } + } + + /// Record a successful LLM turn (streaming or non-streaming). /// /// Fires [`on_stream_success`](crate::observer::LoopObserver::on_stream_success) /// with this turn's token counts and tells the @@ -1224,11 +1437,12 @@ impl BareLoop { /// is healthy (so a transient failure earlier in the session doesn't keep /// the circuit breaker tripped forever). /// - /// Called from [`do_stream`](Self::do_stream) on the `Ok` branch only; - /// has no return value because the caller already holds the successful - /// `(Message, Option, StreamStopReason)` and just needs the - /// side-effects. - fn record_stream_success(&mut self, usage: Option<&Usage>) { + /// Called from [`do_stream`](Self::do_stream) and + /// [`do_create_message`](Self::do_create_message) on the `Ok` branch + /// only; has no return value because the caller already holds the + /// successful `(Message, Option, StreamStopReason)` and just + /// needs the side-effects. + fn record_turn_success(&mut self, usage: Option<&Usage>) { self.managers.fallback().record_success(); let (in_tok, out_tok) = Self::usage_tokens(usage); @@ -1240,12 +1454,12 @@ impl BareLoop { }); } - /// Record a stream failure and return the error to propagate. + /// Record an LLM-turn failure and return the error to propagate. /// /// Distinguishes [`LoopError::RateLimitEscalation`] (which trips the /// model circuit breaker via /// [`record_model_failure`](crate::fallback::FallbackManager::record_model_failure)) - /// from other stream errors (which only count as a generic API failure + /// from other turn errors (which only count as a generic API failure /// via /// [`record_api_failure`](crate::fallback::FallbackManager::record_api_failure)). /// When the breaker trips and a fallback model is configured, fires @@ -1254,18 +1468,25 @@ impl BareLoop { /// Then fires /// [`on_stream_failure`](crate::observer::LoopObserver::on_stream_failure) /// (regardless of breaker outcome) and returns the original error so the - /// caller can propagate it from [`do_stream`](Self::do_stream); the run - /// loop records the terminal + /// caller can propagate it from [`do_stream`](Self::do_stream) or + /// [`do_create_message`](Self::do_create_message); the run loop records + /// the terminal /// [`MachineOutcome::Failed`](crate::engine::core::MachineOutcome) on /// the machine from the returned error. /// - /// `LoopError::Cancelled` is **not** routed through here — cancellation - /// is a clean termination that records - /// [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) - /// without tripping the breaker or firing `on_stream_failure`. The - /// [`run`](crate::engine::core::Loop::run) loop's `CallLLM` arm - /// handles cancellation before reaching [`do_stream`](Self::do_stream). - fn record_stream_failure(&mut self, e: LoopError) -> LoopError { + /// `LoopError::Cancelled` is **not** routed through here in practice. + /// Cancellation during an in-flight turn is caught by the outer + /// `cancel.notified()` arm of the `biased` `select!` in + /// [`handle_call_llm`](Self::handle_call_llm), which resolves the turn + /// by *dropping* the [`do_turn`](Self::do_turn) future — cancelling the + /// in-flight request before this method can run. So a clean cancel + /// records [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) + /// without tripping the breaker or firing `on_stream_failure`. (This + /// method has no `Cancelled` guard of its own; it relies on that outer + /// select. The regression test + /// `cancel_during_non_streaming_turn_does_not_trip_breaker` pins the + /// invariant.) + fn record_turn_failure(&mut self, e: LoopError) -> LoopError { let tripped = if matches!(e, LoopError::RateLimitEscalation { .. }) { self.managers .fallback() @@ -1625,7 +1846,7 @@ impl BareLoop { ); Err(LoopError::Cancelled) } - stream_outcome = self.do_stream(contributor_messages) => { + stream_outcome = self.do_turn(contributor_messages) => { let (msg, usage, stream_stop) = match stream_outcome { Ok(triple) => triple, Err(LoopError::Cancelled) => { @@ -1957,16 +2178,18 @@ mod tests { use super::*; use crate::api::error::ApiError; use crate::engine::core::Loop; + use crate::fallback::FallbackManager; + use crate::observer::LoopObserver; use crate::stream::{ DeltaPart, IndexedDelta, MessageDelta, MessageDeltaPayload, MessageMetadata, MessageStart, - PartStart, Usage, + PartStart, StreamEvent, Usage, }; use crate::tool::ToolRegistry; use crate::tool::{Tool, ToolContext, ToolError, ToolOutput, ToolSchema}; use serde_json::{Value, json}; use std::future::Future; use std::pin::Pin; - use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Mutex; @@ -2243,11 +2466,32 @@ mod tests { dyn Future> + Send + '_, >, > { - Box::pin(async { + let mut guard = crate::error::recover_guard(self.responses.lock()); + let events = guard.pop_front(); + drop(guard); + Box::pin(async move { + let events = events.ok_or_else(|| ApiError::api("No more mock responses"))?; + let mut accumulator = crate::stream::StreamAccumulator::new(); + let mut stop_reason = crate::stream::StreamStopReason::EndTurn; + for event in events { + if let crate::stream::StreamEvent::MessageDelta(delta) = &event + && let Some(reason) = delta + .delta + .stop_reason + .as_deref() + .and_then(crate::stream::StreamStopReason::from_api_str) + { + stop_reason = reason; + } + accumulator + .process(&event) + .map_err(|e| ApiError::api(e.to_string()))?; + } + let usage = accumulator.usage().copied(); Ok(crate::api::NonStreamingResponse { - message: crate::message::Message::assistant(""), - stop_reason: crate::stream::StreamStopReason::EndTurn, - usage: Some(crate::stream::Usage::default()), + message: accumulator.build(), + stop_reason, + usage, }) }) } @@ -2461,6 +2705,156 @@ mod tests { assert_eq!(result.output.as_deref(), Some("Hello! I'm done.")); } + #[test] + fn turn_mode_default_is_nonstreaming_without_feature() { + let client = MockClient::new("test-model"); + let agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); + #[cfg(not(feature = "streaming"))] + assert_eq!(agent.turn_mode(), TurnMode::NonStreaming); + #[cfg(feature = "streaming")] + assert_eq!(agent.turn_mode(), TurnMode::Streaming); + } + + #[tokio::test] + async fn non_streaming_turn_returns_assembled_message() { + let client = MockClient::new("test-model"); + client.add_text_response("assembled via create_message"); + let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); + agent.set_turn_mode(TurnMode::NonStreaming); + let result = agent.run("Hi", &RunConfig::default()).await.unwrap(); + assert_eq!(result.turn_count(), 1); + assert_eq!( + result.output.as_deref(), + Some("assembled via create_message") + ); + } + + #[tokio::test] + async fn non_streaming_turn_runs_tool_call_loop() { + let client = MockClient::new("test-model"); + client.add_tool_then_text("call_1", "echo", json!({"message": "hi"}), "all done"); + let mut registry = ToolRegistry::new(); + registry.register(EchoTool); + let mut agent = BareLoop::new(Arc::new(client), registry, make_config()); + agent.set_turn_mode(TurnMode::NonStreaming); + let result = agent.run("Hi", &RunConfig::default()).await.unwrap(); + assert_eq!(result.turn_count(), 2); + assert_eq!(result.tool_call_count(), 1); + assert_eq!(result.output.as_deref(), Some("all done")); + } + + #[tokio::test] + async fn non_streaming_turn_respects_cancellation() { + let client = MockClient::new("test-model"); + client.add_text_response("never seen"); + let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); + agent.set_turn_mode(TurnMode::NonStreaming); + agent.cancel(); + let result = agent.run("Hi", &RunConfig::default()).await; + assert!(matches!(result, Err(LoopError::Cancelled))); + } + + /// Observer that records whether `on_stream_failure` fired. + struct FailureRecorder { + on_stream_failure_fired: Arc, + } + + impl LoopObserver for FailureRecorder { + fn name(&self) -> &'static str { + "failure-recorder" + } + fn on_stream_failure(&self, _ctx: &StreamFailureContext) { + self.on_stream_failure_fired.store(true, Ordering::SeqCst); + } + } + + /// A client whose `create_message` never completes on its own, so the + /// cancel `select!` arm in `do_create_message` is the only way the turn + /// resolves. Used to exercise mid-turn cancellation. + struct BlockingClient { + started: Arc, + } + + impl ApiClient for BlockingClient { + fn model(&self) -> String { + "blocking".into() + } + fn stream_messages( + &self, + _request: &crate::api::StreamRequest, + ) -> Pin> + Send + 'static>> + { + Box::pin(futures::stream::empty()) + } + fn create_message( + &self, + _request: &crate::api::StreamRequest, + ) -> Pin< + Box< + dyn Future> + Send + '_, + >, + > { + let started = Arc::clone(&self.started); + Box::pin(async move { + started.store(true, Ordering::SeqCst); + // Park forever; only the cancel `select!` arm resolves the turn. + std::future::pending::<()>().await; + Err(ApiError::api("unreachable: cancel must win the select")) + }) + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancel_during_non_streaming_turn_does_not_trip_breaker() { + let client = BlockingClient { + started: Arc::new(AtomicBool::new(false)), + }; + let started = Arc::clone(&client.started); + let on_stream_failure_fired = Arc::new(AtomicBool::new(false)); + let observer = Arc::new(FailureRecorder { + on_stream_failure_fired: Arc::clone(&on_stream_failure_fired), + }); + let managers = LoopManagers::new() + .with_fallback(FallbackManager::default()) + .with_observer(observer); + let mut agent = BareLoop::new_with_managers( + Arc::new(client), + ToolRegistry::new(), + make_config(), + managers, + ); + agent.set_turn_mode(TurnMode::NonStreaming); + + let cancel_signal = Arc::clone(&agent.cancel_signal()); + let run_handle = tokio::spawn(async move { agent.run("Hi", &RunConfig::default()).await }); + + // Wait until create_message is in flight, then cancel. + let mut waits = 0u32; + while !started.load(Ordering::SeqCst) { + waits += 1; + assert!( + waits <= 1000, + "create_message was never entered — test setup is broken" + ); + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + cancel_signal.cancel(); + let run_result = run_handle.await.unwrap(); + + assert!( + started.load(Ordering::SeqCst), + "test only proves anything if create_message was actually entered" + ); + assert!( + matches!(run_result, Err(LoopError::Cancelled)), + "run must return Err(Cancelled): {run_result:?}" + ); + assert!( + !on_stream_failure_fired.load(Ordering::SeqCst), + "a clean cancel must not fire on_stream_failure (it would trip the breaker)" + ); + } + #[test] fn run_config_is_none_before_first_run() { let client = MockClient::new("test-model"); @@ -2695,6 +3089,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn observer_sequence_text_only_turn() { let client = MockClient::new("test-model"); client.add_text_response("Hi there."); @@ -3511,6 +3906,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_text_streamer_fires_on_text_delta() { let client = MockClient::new("test-model"); client.add_text_response("Hello world"); @@ -3533,6 +3929,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_text_streamer_fires_when_stream_handler_configured() { // Regression: when a StreamHandler is attached, the engine must still // fire text_streamer / on_text_delta for each streamed text delta. The @@ -3571,6 +3968,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_text_streamer_ignores_non_text_deltas() { let client = MockClient::new("test-model"); @@ -3628,6 +4026,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_text_delta_fires_per_sse_chunk_in_order() { struct DeltaRecorder { deltas: Arc>>, @@ -3697,6 +4096,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_text_delta_turn_number_matches_surrounding_turn() { struct TurnRecorder { deltas: Arc>>, @@ -3765,6 +4165,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_text_delta_ignores_non_text_deltas() { struct DeltaRecorder { count: Arc, @@ -3830,6 +4231,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_text_delta_fires_without_streamer() { struct DeltaRecorder { deltas: Arc>>, @@ -3865,6 +4267,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_text_delta_and_streamer_coexist() { struct DeltaRecorder { deltas: Arc>>, @@ -4550,6 +4953,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_stream_turn_cancelled_mid_stream() { let (client, tx) = StreamingMockClient::new("test-model"); let model = client.model.clone(); @@ -5196,6 +5600,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn run_cancel_during_streaming_returns_fast() { let (client, tx) = StreamingMockClient::new("test-model"); tx.send(Ok(StreamEvent::MessageStart(MessageStart { @@ -5384,6 +5789,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_rate_limit_escalation_feeds_circuit_breaker() { use crate::fallback::FallbackManager; use crate::managers::LoopManagers; @@ -5647,20 +6053,56 @@ mod tests { fn create_message( &self, - _request: &crate::api::StreamRequest, + request: &crate::api::StreamRequest, ) -> Pin< Box< dyn Future> + Send + '_, >, > { - Box::pin(async { + let messages = request.messages.clone(); + crate::error::recover_guard(self.seen.lock()).push(messages); + let mut guard = crate::error::recover_guard(self.responses.lock()); + let events = guard.pop_front(); + drop(guard); + Box::pin(async move { + let events = events.ok_or_else(|| ApiError::api("No more mock responses"))?; + let mut accumulator = crate::stream::StreamAccumulator::new(); + let mut stop_reason = crate::stream::StreamStopReason::EndTurn; + for event in events { + if let crate::stream::StreamEvent::MessageDelta(delta) = &event + && let Some(reason) = delta + .delta + .stop_reason + .as_deref() + .and_then(crate::stream::StreamStopReason::from_api_str) + { + stop_reason = reason; + } + accumulator + .process(&event) + .map_err(|e| ApiError::api(e.to_string()))?; + } + let usage = accumulator.usage().copied(); Ok(crate::api::NonStreamingResponse { - message: crate::message::Message::assistant(""), - stop_reason: crate::stream::StreamStopReason::EndTurn, - usage: Some(crate::stream::Usage::default()), + message: accumulator.build(), + stop_reason, + usage, }) }) } + + fn create_message_with_options( + &self, + request: &crate::api::StreamRequest, + options: crate::structured::RequestOptions, + ) -> Pin< + Box< + dyn Future> + Send + '_, + >, + > { + crate::error::recover_guard(self.seen_options.lock()).push(options); + self.create_message(request) + } } struct StaticReminder(String); @@ -6085,6 +6527,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_thinking_delta_fires_per_thinking_delta() { struct ThinkingRecorder { deltas: Arc>>, @@ -6166,6 +6609,7 @@ mod tests { } #[tokio::test] + #[cfg(feature = "streaming")] async fn test_on_thinking_delta_independent_of_text_delta() { struct MixedRecorder { text_calls: Arc>, diff --git a/src/engine/bare/compact.rs b/src/engine/bare/compact.rs index 85020c2..f6b3e9a 100644 --- a/src/engine/bare/compact.rs +++ b/src/engine/bare/compact.rs @@ -8,9 +8,9 @@ //! hook events, and returns the compacted messages plus a post-compaction token //! estimate for the driver to feed back into the machine. -use super::{ApiClient, BareLoop, Instant, LoopError}; +use super::{ApiClient, BareLoop, LoopError}; #[cfg(feature = "hooks")] -use super::{CompactTrigger, PostCompactContext, PreCompactContext}; +use super::{CompactTrigger, Instant, PostCompactContext, PreCompactContext}; use crate::compact::EnsureContextResult; use crate::capabilities::Compactable; @@ -53,11 +53,14 @@ impl BareLoop { return Ok((history, 0)); }; + #[cfg(feature = "hooks")] if self.pre_compact_hook_aborts(&history) { return Ok((history, 0)); } + #[cfg(feature = "hooks")] let messages_before = history.len(); + #[cfg(feature = "hooks")] let compact_start = Instant::now(); let result = ctx_manager.compact_with_reason(history, turn, reason).await; @@ -65,6 +68,7 @@ impl BareLoop { Ok(EnsureContextResult::Compacted(outcome)) => { let tokens_after = outcome.tokens_after; let tokens_saved = outcome.tokens_saved; + #[cfg(feature = "hooks")] let messages_after = outcome.messages.len(); let tokens_before = tokens_after.saturating_add(tokens_saved); self.managers.observers().on_compaction(&CompactedContext { @@ -72,6 +76,7 @@ impl BareLoop { tokens_after, tokens_saved, }); + #[cfg(feature = "hooks")] self.notify_post_compact_hook( messages_before, messages_after, @@ -116,16 +121,6 @@ impl BareLoop { executor.check_pre_compact(&ctx).abort } - /// No-op pre-compact hook check for builds without the `hooks` feature. - /// - /// Always returns `false` so [`run_compaction`](Self::run_compaction) - /// proceeds with compaction unconditionally — there are no hooks to - /// consult. - #[cfg(not(feature = "hooks"))] - fn pre_compact_hook_aborts(&self, _history: &[Message]) -> bool { - false - } - /// Notify the post-compact hook that compaction completed. /// /// Builds a [`PostCompactContext`] from the before/after message @@ -158,20 +153,4 @@ impl BareLoop { }; executor.notify_post_compact(&ctx); } - - /// No-op post-compact notification for builds without the `hooks` feature. - /// - /// Does nothing — there are no hooks to notify. Kept so - /// [`run_compaction`](Self::run_compaction) compiles identically - /// with and without the feature. - #[cfg(not(feature = "hooks"))] - fn notify_post_compact_hook( - &self, - _messages_before: usize, - _messages_after: usize, - _tokens_after: u64, - _tokens_saved: u64, - _duration: std::time::Duration, - ) { - } } diff --git a/src/engine/bare/dispatch.rs b/src/engine/bare/dispatch.rs index fc00f98..8932d85 100644 --- a/src/engine/bare/dispatch.rs +++ b/src/engine/bare/dispatch.rs @@ -385,6 +385,7 @@ impl BareLoop { loop { self.notify_tool_pre(turn_idx, &tc); + #[cfg(feature = "hooks")] if let Some(blocked) = self.check_pre_tool_use_hooks(&tc, turn_idx) { self.notify_tool_post(turn_idx, &tc, &blocked); return Ok(blocked); @@ -403,7 +404,9 @@ impl BareLoop { }; self.post_detection(&tc, &tool_result); self.notify_tool_post(turn_idx, &tc, &tool_result); + #[cfg(feature = "hooks")] self.notify_post_tool_use_hooks(&tc, &tool_result, turn_idx); + #[cfg(feature = "tool_health")] self.record_tool_health(tc.tool.as_str(), &tool_result); self.record_tool_memory(&tc, &tool_result).await; @@ -707,19 +710,6 @@ impl BareLoop { } } - /// Stub for [`check_pre_tool_use_hooks`](Self::check_pre_tool_use_hooks) - /// when the `hooks` feature is disabled. - /// - /// Always returns `None` (no hooks to check) so the call proceeds normally. - #[cfg(not(feature = "hooks"))] - fn check_pre_tool_use_hooks( - &self, - _tc: &ToolCall, - _turn_idx: usize, - ) -> Option { - None - } - /// Notify post-tool-use hooks with the execution result. /// /// If a [`HookExecutor`](crate::hooks::HookExecutor) is configured, builds @@ -757,19 +747,6 @@ impl BareLoop { executor.notify_post_tool_use(&ctx); } - /// Stub for [`notify_post_tool_use_hooks`](Self::notify_post_tool_use_hooks) - /// when the `hooks` feature is disabled. - /// - /// No-op: there are no hooks to notify. - #[cfg(not(feature = "hooks"))] - fn notify_post_tool_use_hooks( - &self, - _tc: &ToolCall, - _tool_result: &ToolDispatchResult, - _turn_idx: usize, - ) { - } - /// Record tool execution health (success or failure) in the health registry. /// /// If a [`ToolHealthRegistry`](crate::tool::health::ToolHealthRegistry) is @@ -789,13 +766,6 @@ impl BareLoop { } } - /// Stub for [`record_tool_health`](Self::record_tool_health) when the - /// `tool_health` feature is disabled. - /// - /// No-op: there is no health registry to record into. - #[cfg(not(feature = "tool_health"))] - fn record_tool_health(&self, _tool_name: &str, _tool_result: &ToolDispatchResult) {} - /// Store a successful tool-execution trajectory into the memory backend. /// /// Called after each tool dispatch that did not error. Guards on @@ -928,7 +898,7 @@ impl BareLoop { } } -#[cfg(test)] +#[cfg(all(test, feature = "testing"))] #[allow(clippy::unnecessary_literal_bound)] mod tests { use crate::api::error::ApiError; diff --git a/src/engine/bare/emission.rs b/src/engine/bare/emission.rs index 19fc743..ed6529b 100644 --- a/src/engine/bare/emission.rs +++ b/src/engine/bare/emission.rs @@ -25,6 +25,7 @@ impl BareLoop { self.managers.observers().on_run_start(&RunStartContext { session_id: self.session.id, }); + #[cfg(feature = "hooks")] self.notify_run_start_hook(); } @@ -42,6 +43,7 @@ impl BareLoop { duration: Duration, error: Option<&LoopError>, ) { + #[cfg(feature = "hooks")] self.notify_run_end_hook(result, error, duration); self.managers.observers().on_run_end(&RunEndContext { success: error.is_none(), @@ -101,14 +103,6 @@ impl BareLoop { executor.notify_run_start(&ctx); } - /// No-op run-start hook for builds without the `hooks` feature. - /// - /// Does nothing — there are no hooks to notify. Kept so - /// [`notify_run_start`](Self::notify_run_start) compiles - /// identically with and without the feature. - #[cfg(not(feature = "hooks"))] - fn notify_run_start_hook(&self) {} - /// Fire the `on_run_end` hook when a hook executor is /// configured. /// @@ -134,14 +128,6 @@ impl BareLoop { executor.notify_run_end(&ctx); } - /// No-op run-end hook for builds without the `hooks` feature. - /// - /// Does nothing — there are no hooks to notify. Kept so - /// [`notify_run_end`](Self::notify_run_end) compiles - /// identically with and without the feature. - #[cfg(not(feature = "hooks"))] - fn notify_run_end_hook(&self, _result: &Run, _error: Option<&LoopError>, _duration: Duration) {} - /// Convert a [`Duration`] to milliseconds as a `u64`. /// /// Saturates at `u64::MAX` if the duration exceeds the `u64` range diff --git a/src/engine/bare/stream.rs b/src/engine/bare/stream.rs index 8e7efa1..6404615 100644 --- a/src/engine/bare/stream.rs +++ b/src/engine/bare/stream.rs @@ -8,13 +8,11 @@ //! a handler via [`set_stream_handler()`](BareLoop::set_stream_handler) opts into //! retry, timeout, fallback, and rate-limit handling. -use super::{ - ApiClient, BareLoop, LoopError, Message, Run, StreamAccumulator, StreamEvent, StreamStopReason, - Usage, -}; +use super::{ApiClient, BareLoop, LoopError, Message, Run}; use crate::capabilities::StreamCapable; use crate::observer::{TextDeltaContext, ThinkingDeltaContext}; use crate::stream::handler::{HandlerEvent, StreamHandlerError}; +use crate::stream::{StreamAccumulator, StreamEvent, StreamStopReason, Usage}; use futures::StreamExt; impl BareLoop { @@ -116,7 +114,7 @@ impl BareLoop { && let crate::stream::DeltaPart::Text { text } = &d.delta { if let Some(streamer) = &self.text_streamer { - streamer(text); + streamer(text.as_str()); } self.managers.observers().on_text_delta(&TextDeltaContext { turn: self.current_run().map_or(0, Run::turn_count), diff --git a/src/managers.rs b/src/managers.rs index 923395f..17ccbe4 100644 --- a/src/managers.rs +++ b/src/managers.rs @@ -17,10 +17,10 @@ //! | [`Detectable`] | Detect repetitive loops and convergence | //! | [`FallbackCapable`] | Circuit-breaker fallback to alternate models | //! | [`Compactable`] | Automatic context compaction when tokens exceed | -//! | [`StreamCapable`] | Resilient streaming with retries and timeouts | -//! | [`Hookable`] | Bidirectional hooks that can block actions | +//! | `StreamCapable` | Resilient streaming with retries and timeouts | +//! | `Hookable` | Bidirectional hooks that can block actions | //! | [`PipelineAware`] | Dispatch tools through a middleware pipeline | -//! | [`HealthTrackable`] | Per-tool health tracking with circuit breakers | +//! | `HealthTrackable` | Per-tool health tracking with circuit breakers | //! //! # `LoopManagers` //! @@ -75,6 +75,7 @@ use crate::hooks::HookExecutor; use crate::middleware::ToolPipeline; use crate::observer::{ConvergenceDetectedContext, LoopDetectedContext}; use crate::observer::{LoopObserver, ObserverHost}; +#[cfg(feature = "streaming")] use crate::stream::handler::StreamHandler; #[cfg(feature = "tool_health")] use crate::tool::health::ToolHealthRegistry; @@ -111,7 +112,7 @@ pub use crate::capabilities::*; /// - [`Detectable`] — via the internal [`DetectionManager`] /// - [`FallbackCapable`] — via the internal [`FallbackManager`] /// - [`Compactable`] — via an optional [`ContextManager`] -/// - [`StreamCapable`] — via an optional [`StreamHandler`] +/// - `StreamCapable` — via an optional `StreamHandler` /// - `Hookable` — via an optional `HookExecutor` *(requires `hooks` feature)* /// - [`PipelineAware`] — via an optional [`ToolPipeline`] /// - `HealthTrackable` — via an optional `ToolHealthRegistry` *(requires `tool_health` feature)* @@ -160,7 +161,9 @@ pub struct LoopManagers { /// /// When set, wraps streaming calls with retry logic, timeout enforcement, /// and automatic fallback to non-streaming when the provider drops the - /// connection mid-stream. + /// connection mid-stream. Absent under `default = []`; requires the + /// `streaming` feature. + #[cfg(feature = "streaming")] stream_handler: Option, /// Optional hook executor for bidirectional lifecycle interception. @@ -208,6 +211,7 @@ impl LoopManagers { observer_host: ObserverHost::new(), tool_pipeline: None, context_manager: None, + #[cfg(feature = "streaming")] stream_handler: None, #[cfg(feature = "hooks")] hook_executor: None, @@ -319,6 +323,7 @@ impl LoopManagers { /// .with_stream_handler(handler) /// ``` #[must_use] + #[cfg(feature = "streaming")] pub fn with_stream_handler(mut self, handler: StreamHandler) -> Self { self.stream_handler = Some(handler); self @@ -383,6 +388,7 @@ impl LoopManagers { /// Set the stream handler for resilient streaming. /// /// Non-consuming variant of [`with_stream_handler`](Self::with_stream_handler). + #[cfg(feature = "streaming")] pub fn set_stream_handler(&mut self, handler: StreamHandler) { self.stream_handler = Some(handler); } @@ -560,6 +566,7 @@ impl crate::capabilities::RememberCapable for LoopManagers { } } +#[cfg(feature = "streaming")] impl crate::capabilities::StreamCapable for LoopManagers { fn stream_handler(&self) -> &StreamHandler { self.stream_handler @@ -654,6 +661,7 @@ mod tests { fn _assert_fallback(_: &dyn FallbackCapable) {} fn _assert_pipeline(_: &dyn PipelineAware) {} fn _assert_compactable(_: &dyn Compactable) {} + #[cfg(feature = "streaming")] fn _assert_stream_capable(_: &dyn StreamCapable) {} let managers = LoopManagers::new(); @@ -662,6 +670,7 @@ mod tests { _assert_fallback(&managers); _assert_pipeline(&managers); _assert_compactable(&managers); + #[cfg(feature = "streaming")] _assert_stream_capable(&managers); } @@ -684,6 +693,7 @@ mod tests { } #[test] + #[cfg(feature = "streaming")] fn test_stream_handler_defaults_to_passthrough() { let managers = LoopManagers::new(); let handler = managers.stream_handler(); @@ -720,6 +730,7 @@ mod tests { } #[test] + #[cfg(feature = "streaming")] fn test_set_stream_handler_overrides_passthrough() { use crate::stream::handler::StreamHandler; @@ -800,6 +811,7 @@ mod tests { fn accepts_detectable(_: &impl Detectable) {} fn accepts_fallback(_: &impl FallbackCapable) {} fn accepts_compactable(_: &impl Compactable) {} + #[cfg(feature = "streaming")] fn accepts_stream_capable(_: &impl StreamCapable) {} fn accepts_pipeline(_: &impl PipelineAware) {} fn accepts_multi_bound(_: &(impl Observable + Detectable + FallbackCapable)) {} @@ -809,6 +821,7 @@ mod tests { accepts_detectable(&managers); accepts_fallback(&managers); accepts_compactable(&managers); + #[cfg(feature = "streaming")] accepts_stream_capable(&managers); accepts_pipeline(&managers); accepts_multi_bound(&managers); diff --git a/src/observer.rs b/src/observer.rs index 3af642c..c0f9cf5 100644 --- a/src/observer.rs +++ b/src/observer.rs @@ -115,7 +115,7 @@ pub trait LoopObserver: Send + Sync { /// arrival order, to reconstruct the per-turn text. /// /// This is the per-token counterpart to the raw - /// [`text_streamer`](crate::engine::BareLoop::set_text_streamer) callback, + /// `text_streamer` callback, /// delivered through the observer system so multiple observers each receive /// every chunk. The streamer remains available for simple single-consumer /// use. @@ -127,7 +127,7 @@ pub trait LoopObserver: Send + Sync { /// /// # Retry caveat (handler path) /// - /// When a [`StreamHandler`](crate::stream::handler::StreamHandler) is + /// When a `StreamHandler` is /// configured, this callback fires for every event of every attempt — /// including partial events from a failed attempt that got cut off /// mid-stream. An observer that concatenates `delta` across calls will, @@ -164,7 +164,7 @@ pub trait LoopObserver: Send + Sync { /// as an empty `delta` (render a placeholder, not the empty string). /// /// Inherits the same retry caveat as [`on_text_delta`](Self::on_text_delta): - /// under a configured [`StreamHandler`](crate::stream::handler::StreamHandler), + /// under a configured `StreamHandler`, /// partial events from a failed attempt fire here too. Buffer until /// [`on_turn_end`](Self::on_turn_end) if you need only committed reasoning. fn on_thinking_delta(&self, _ctx: &ThinkingDeltaContext) {} diff --git a/src/provider.rs b/src/provider.rs index b3617a1..fd2b6ea 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -55,11 +55,13 @@ //! let result = agent.run("Hello!", &RunConfig::default()).await?; //! ``` +#[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] use crate::api::error::ApiError; #[cfg(any(feature = "anthropic", feature = "gemini"))] use crate::message::{MessagePart, Role}; #[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] use futures::StreamExt; +#[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] use std::time::Duration; // SSE line-framing shared by every streaming provider. Each provider keeps @@ -125,6 +127,7 @@ pub(super) async fn read_bounded_body(resp: reqwest::Response) -> Result Self { Self { @@ -206,6 +210,7 @@ impl Default for HttpClientConfig { } } +#[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] impl HttpClientConfig { /// Set the total request timeout. /// @@ -318,12 +323,14 @@ impl HttpClientConfig { /// for pool and TCP knobs, no-opping when `None`. /// /// Used internally by [`HttpClientConfig::build`]. +#[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] trait ClientBuilderExt: Sized { fn maybe_pool_max_idle_per_host(self, val: Option) -> Self; fn maybe_pool_idle_timeout(self, val: Option) -> Self; fn maybe_tcp_keepalive(self, val: Option) -> Self; } +#[cfg(any(feature = "openai", feature = "anthropic", feature = "gemini"))] impl ClientBuilderExt for reqwest::ClientBuilder { fn maybe_pool_max_idle_per_host(self, val: Option) -> Self { match val { diff --git a/src/reflection/llm.rs b/src/reflection/llm.rs index 898e0b7..bf23976 100644 --- a/src/reflection/llm.rs +++ b/src/reflection/llm.rs @@ -159,6 +159,7 @@ impl Reflector for LlmReflector { .await .map_err(|e| ReflectionError::Internal(format!("{e}")))?; + #[cfg(feature = "schema_validation")] validate_modified_input(&analysis, schema_value.as_ref())?; Ok(analysis) }) @@ -205,58 +206,41 @@ fn build_user_message( /// Validate the model's suggested `modified_input` against the failing /// tool's input schema. /// -/// Returns `Ok(())` when there is nothing to validate (no correction, no -/// `modified_input`, or no schema supplied) — these are all legitimate -/// "skip validation" cases. When a schema is supplied and validation is -/// enabled, returns `Ok(())` on a match or -/// [`ReflectionError::Internal`](crate::reflection::ReflectionError::Internal) -/// on a mismatch. +/// Returns `Ok(())` when there is nothing to check (no correction, no +/// `modified_input`, or no schema supplied), or +/// [`ReflectionError::Internal`] when a supplied schema does not conform. /// -/// # Feature gating -/// -/// The schema check itself only runs when the `schema_validation` feature -/// is enabled. Without it, this function is always `Ok(())` once the -/// early-return skips have passed — the analysis is returned unchanged. -/// Callers who want validation must enable the feature. +/// The whole function (and its single call site) is gated behind the +/// `schema_validation` feature: without it there is no validation to do, +/// so neither the function nor the call exist. This keeps the no-feature +/// build free of dead no-op stubs. /// /// # Errors /// -/// See the body — returns `ReflectionError::Internal` only under -/// `schema_validation` + supplied schema + non-conforming `modified_input`. +/// Returns [`ReflectionError::Internal`] only when a schema is supplied +/// and `modified_input` does not conform to it. +#[cfg(feature = "schema_validation")] fn validate_modified_input( analysis: &FailureAnalysis, tool_schema: Option<&serde_json::Value>, ) -> Result<(), ReflectionError> { - // Without the schema_validation feature, validation never runs — bail - // out early so we don't bind `modified_input` / `schema` only to drop - // them on the floor. The signature is unchanged; the early return - // keeps the function a no-op under the default feature set. - #[cfg(not(feature = "schema_validation"))] - { - let _ = (analysis, tool_schema); + let Some(correction) = &analysis.correction else { return Ok(()); - } - - #[cfg(feature = "schema_validation")] - { - let Some(correction) = &analysis.correction else { - return Ok(()); - }; - let Some(modified_input) = &correction.modified_input else { - return Ok(()); - }; - let Some(schema) = tool_schema else { - return Ok(()); - }; - - if !jsonschema::is_valid(schema, modified_input) { - return Err(ReflectionError::Internal( - "corrected input does not match the tool's schema".to_string(), - )); - } + }; + let Some(modified_input) = &correction.modified_input else { + return Ok(()); + }; + let Some(schema) = tool_schema else { + return Ok(()); + }; - Ok(()) + if !jsonschema::is_valid(schema, modified_input) { + return Err(ReflectionError::Internal( + "corrected input does not match the tool's schema".to_string(), + )); } + + Ok(()) } #[cfg(test)] @@ -718,6 +702,7 @@ mod tests { } #[test] + #[cfg(feature = "schema_validation")] fn validate_modified_input_noop_without_correction() { let analysis = FailureAnalysis { is_recoverable: false, @@ -731,6 +716,7 @@ mod tests { } #[test] + #[cfg(feature = "schema_validation")] fn validate_modified_input_noop_without_modified_input() { let analysis = FailureAnalysis { is_recoverable: true, @@ -749,6 +735,7 @@ mod tests { } #[test] + #[cfg(feature = "schema_validation")] fn validate_modified_input_skips_when_no_schema() { // Even with a modified_input present, no schema → Ok (the engine // passes None when the tool isn't in the registry). @@ -769,9 +756,8 @@ mod tests { } #[test] - fn validate_modified_input_passes_without_feature() { - // Without `schema_validation`, even a deliberately-mismatched - // modified_input must return Ok (validation is gated off). + #[cfg(feature = "schema_validation")] + fn validate_modified_input_rejects_mismatched_schema() { let analysis = FailureAnalysis { is_recoverable: true, root_cause: "x".to_string(), @@ -779,7 +765,6 @@ mod tests { correction: Some(Correction { correction_type: CorrectionType::InputFix, description: "fix".to_string(), - // Mismatched shape — schema requires a number, input is a string. modified_input: Some(serde_json::json!({"wrong": "shape"})), alternative_tool: None, guidance: None, @@ -793,19 +778,10 @@ mod tests { "additionalProperties": false }); let result = validate_modified_input(&analysis, Some(&schema)); - // Under `schema_validation` this fails; without it, it's Ok. Pin - // the no-feature behavior here; the under-feature behavior is - // covered by llm_reflector_validates_modified_input_fail. - #[cfg(feature = "schema_validation")] assert!( matches!(result, Err(ReflectionError::Internal(_))), "with schema_validation the mismatch should fail: {result:?}" ); - #[cfg(not(feature = "schema_validation"))] - assert!( - result.is_ok(), - "without schema_validation validation is skipped: {result:?}" - ); } #[test] diff --git a/src/stream.rs b/src/stream.rs index 8a71948..6e483c4 100644 --- a/src/stream.rs +++ b/src/stream.rs @@ -33,7 +33,7 @@ //! //! # Sub-modules //! -//! - **[`handler`]** — [`handler::StreamHandler`] with retry, timeout, +//! - **`handler`** — `handler::StreamHandler` with retry, timeout, //! and fallback for resilient streaming. //! //! # Quick Start @@ -58,9 +58,11 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use std::fmt; +#[cfg(feature = "streaming")] pub mod handler; pub mod rate_limit; +#[cfg(feature = "streaming")] pub use handler::{DetectedRateLimit, RateLimitConfig, RateLimitKind}; pub use rate_limit::{RateLimiter, TokenBucket}; diff --git a/src/stream/rate_limit.rs b/src/stream/rate_limit.rs index 2f4465a..564e86b 100644 --- a/src/stream/rate_limit.rs +++ b/src/stream/rate_limit.rs @@ -6,7 +6,7 @@ //! independent budgets. //! //! This is the proactive complement to the reactive 429 handling in -//! [`handler`](super::handler): the bucket gates a request *before* it fires, +//! `stream::handler`: the bucket gates a request *before* it fires, //! smoothing bursty multi-turn loops so most provider-imposed rate limits are //! never hit. //! @@ -198,7 +198,8 @@ fn elapsed_refill(state: &mut BucketState, at: Instant, capacity: f64, refill_pe /// [`acquire`](Self::acquire) then returns `Ok(())` immediately and never /// allocates a bucket. /// -/// This is the type [`StreamHandler`](super::handler::StreamHandler) holds. +/// This is the type `StreamHandler` holds when the `streaming` feature is +/// enabled. #[derive(Debug)] pub struct RateLimiter { /// Per-provider token buckets, keyed by base URL. diff --git a/src/structured.rs b/src/structured.rs index 44892cf..b2528df 100644 --- a/src/structured.rs +++ b/src/structured.rs @@ -211,8 +211,8 @@ impl ResponseFormat { /// tightening; Anthropic / Gemini tightened `input_schema` / `parameters`). /// /// The enum is `#[non_exhaustive]`: future variants may be added -/// non-breakingly, and the [`Grammar`](Self::Grammar) variant is only -/// present under the `grammar` feature. +/// non-breakingly, and the `Grammar` variant is only present under the +/// `grammar` feature. #[derive(Debug, Clone, Default)] #[non_exhaustive] pub enum ToolConstraint { From a0d4eaf4470134faf2a78867bd75de625ce6644f Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Sun, 2 Aug 2026 20:47:29 +1200 Subject: [PATCH 2/3] fix: cancel no longer trips fallback breaker --- src/engine/bare.rs | 414 +++++++++++++++++++++++---------------------- 1 file changed, 208 insertions(+), 206 deletions(-) diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 346c7ca..6a3d51d 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -119,10 +119,12 @@ mod stream; /// `StreamHandler`, emitting per-delta observer callbacks. Requires the /// `streaming` feature. /// -/// The default is `Streaming` when `streaming` is compiled in, otherwise -/// `NonStreaming`. Switch modes on a constructed loop with -/// [`set_turn_mode`](BareLoop::set_turn_mode). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +/// The constructor default is feature-dependent: `Streaming` when `streaming` +/// is compiled in, otherwise `NonStreaming` (see [`BareLoop::turn_mode`]). It +/// is intentionally *not* a `Default` impl on this enum, because a single +/// fixed `Default` could not express that feature-dependent choice. Switch +/// modes on a constructed loop with [`set_turn_mode`](BareLoop::set_turn_mode). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TurnMode { /// Fulfil each turn via [`ApiClient::create_message`]. /// @@ -130,15 +132,13 @@ pub enum TurnMode { /// `on_thinking_delta`, and the text streamer never fire. The full /// assistant text still surfaces through /// [`on_response`](crate::observer::LoopObserver::on_response). - #[default] NonStreaming, /// Fulfil each turn via [`ApiClient::stream_messages`] wrapped in /// [`StreamHandler`](crate::stream::handler::StreamHandler). /// - /// Requires the `streaming` feature. Constructing or selecting this - /// mode without `streaming` enabled yields a - /// [`LoopError::Config`](crate::error::LoopError::Config) at turn time. + /// Requires the `streaming` feature: this variant only exists when the + /// feature is enabled, so it cannot be constructed or selected without it. #[cfg(feature = "streaming")] Streaming, } @@ -1322,8 +1322,7 @@ impl BareLoop { /// On a real API failure: fires /// [`on_stream_failure`](crate::observer::LoopObserver::on_stream_failure) /// (and [`on_fallback`](crate::observer::LoopObserver::on_fallback) if - /// the breaker trips) and returns the error. A clean cancellation is - /// *not* a failure — see the note below. + /// the breaker trips) and returns the error. /// /// # Errors /// @@ -1331,15 +1330,13 @@ impl BareLoop { /// the request; otherwise whatever [`ApiError`] the client returned, /// mapped to [`LoopError::Api`](LoopError::Api). /// - /// # Cancellation note - /// - /// When cancel wins the inner `select!` here, it would flow into - /// [`record_turn_failure`](Self::record_turn_failure) — but in practice - /// the outer `biased` `select!` in - /// [`handle_call_llm`](Self::handle_call_llm) wins the same race and - /// drops this future first, so a clean cancel never trips the breaker - /// or fires `on_stream_failure`. Pinned by - /// `cancel_during_non_streaming_turn_does_not_trip_breaker`. + /// The already-cancelled case is guarded once in + /// [`do_turn`](Self::do_turn), so this method is only reached on a live + /// run. A `Cancelled` that nonetheless reaches + /// [`record_turn_failure`](Self::record_turn_failure) (cancel winning + /// the `select!` mid-request) is harmless — that method guards + /// `Cancelled` locally and returns it without tripping the breaker or + /// firing `on_stream_failure`. async fn do_create_message( &mut self, contributor_messages: Vec, @@ -1350,9 +1347,6 @@ impl BareLoop { .with_system(self.session.config.system_prompt.clone()) .with_tools(self.build_tool_schemas()); - if self.cancelled.is_cancelled() { - return Err(LoopError::Cancelled); - } let cancel = Arc::clone(&self.cancelled); let client = &self.client; let options = self.request_options.clone(); @@ -1380,14 +1374,9 @@ impl BareLoop { /// [`on_fallback`](crate::observer::LoopObserver::on_fallback) if the /// circuit breaker trips), fires /// [`on_stream_failure`](crate::observer::LoopObserver::on_stream_failure), - /// sets the terminal state, and returns the error. - /// - /// On a clean cancellation the outer `cancel.notified()` arm of the - /// `biased` `select!` in [`handle_call_llm`](Self::handle_call_llm) - /// wins and drops this future before - /// [`record_turn_failure`](Self::record_turn_failure) runs — so a cancel - /// during the turn records - /// [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) + /// sets the terminal state, and returns the error. A `Cancelled` result + /// is guarded by [`record_turn_failure`](Self::record_turn_failure) and + /// records [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) /// without tripping the fallback or firing `on_stream_failure`. /// /// # Errors @@ -1409,19 +1398,23 @@ impl BareLoop { /// Dispatch one LLM turn according to [`turn_mode`](self.turn_mode). /// - /// Single entry point for the run loop's `CallLLM` arm so the - /// cancellation `select!` in [`handle_call_llm`](Self::handle_call_llm) - /// races against exactly one future regardless of mode. + /// Single entry point for the run loop's `CallLLM` arm. Guards the + /// already-cancelled case once here so neither turn path polls its + /// provider future (and its call-time side effects) on a dead run. /// /// # Errors /// - /// Propagates whatever the selected turn path + /// [`LoopError::Cancelled`] if the run is already cancelled before the + /// turn starts; otherwise whatever the selected turn path /// ([`do_stream`](Self::do_stream) or /// [`do_create_message`](Self::do_create_message)) returns. async fn do_turn( &mut self, contributor_messages: Vec, ) -> Result<(Message, Option, StreamStopReason), LoopError> { + if self.cancelled.is_cancelled() { + return Err(LoopError::Cancelled); + } match self.turn_mode { #[cfg(feature = "streaming")] TurnMode::Streaming => self.do_stream(contributor_messages).await, @@ -1474,19 +1467,18 @@ impl BareLoop { /// [`MachineOutcome::Failed`](crate::engine::core::MachineOutcome) on /// the machine from the returned error. /// - /// `LoopError::Cancelled` is **not** routed through here in practice. - /// Cancellation during an in-flight turn is caught by the outer - /// `cancel.notified()` arm of the `biased` `select!` in - /// [`handle_call_llm`](Self::handle_call_llm), which resolves the turn - /// by *dropping* the [`do_turn`](Self::do_turn) future — cancelling the - /// in-flight request before this method can run. So a clean cancel - /// records [`MachineOutcome::Cancelled`](crate::engine::core::MachineOutcome) - /// without tripping the breaker or firing `on_stream_failure`. (This - /// method has no `Cancelled` guard of its own; it relies on that outer - /// select. The regression test - /// `cancel_during_non_streaming_turn_does_not_trip_breaker` pins the - /// invariant.) + /// [`LoopError::Cancelled`] short-circuits: a clean cancellation is not a + /// failure, so it returns early without touching the breaker or firing + /// `on_stream_failure`. This makes the cancel-vs-failure distinction + /// *local* to this method (mirroring [`set_error_state`](Self::set_error_state)), + /// so correctness no longer depends on which `select!` arm wins a cancel + /// race — the outer cancel `select!` in [`handle_call_llm`](Self::handle_call_llm) + /// is pure fast-path delivery, not a load-bearing correctness guard. + /// Pinned by `cancel_during_non_streaming_turn_does_not_trip_breaker`. fn record_turn_failure(&mut self, e: LoopError) -> LoopError { + if matches!(e, LoopError::Cancelled) { + return e; + } let tripped = if matches!(e, LoopError::RateLimitEscalation { .. }) { self.managers .fallback() @@ -1831,11 +1823,10 @@ impl BareLoop { } } - let cancel = Arc::clone(&self.cancelled); - tokio::select! { - biased; - () = cancel.notified() => { - self.machine.cancel(); + let stream_outcome = self.do_turn(contributor_messages).await; + let (msg, usage, stream_stop) = match stream_outcome { + Ok(triple) => triple, + Err(LoopError::Cancelled) => { self.notify_turn_end( current_turn, false, @@ -1844,93 +1835,77 @@ impl BareLoop { 0, 0, ); - Err(LoopError::Cancelled) + return Err(LoopError::Cancelled); } - stream_outcome = self.do_turn(contributor_messages) => { - let (msg, usage, stream_stop) = match stream_outcome { - Ok(triple) => triple, - Err(LoopError::Cancelled) => { - self.notify_turn_end( - current_turn, - false, - Some("cancelled".into()), - turn_start.elapsed(), - 0, - 0, - ); - return Err(LoopError::Cancelled); - } - Err(e) => return Err(e), - }; - - let text = msg.text_content(); - let (turn_in, turn_out) = Self::usage_tokens(usage.as_ref()); - let pattern = self.managers.detection().record_response(&text); - self.notify_response(current_turn, &text, usage); + Err(e) => return Err(e), + }; - if let Some(e) = self.apply_loop_detection(current_turn, &pattern) { - return Err(e); - } + let text = msg.text_content(); + let (turn_in, turn_out) = Self::usage_tokens(usage.as_ref()); + let pattern = self.managers.detection().record_response(&text); + self.notify_response(current_turn, &text, usage); - let tool_calls: Vec = msg - .tool_call_parts() - .into_iter() - .map(|(id, tool, input)| ToolCall { - id: id.to_string(), - tool: tool.to_string(), - input: input.clone(), - }) - .collect(); - let stop_reason = match stream_stop { - StreamStopReason::ToolCall => StopReason::ToolCall, - StreamStopReason::MaxTokens => StopReason::MaxTokens, - StreamStopReason::StopSequence => StopReason::StopSequence, - StreamStopReason::EndTurn => { - if tool_calls.is_empty() { - StopReason::EndTurn - } else { - StopReason::ToolCall - } - } - }; - let model_response = ModelResponse { - message: msg, - input_tokens: turn_in, - output_tokens: turn_out, - stop_reason, - available_tools: self.tools.tool_names(), - }; - let mut context_history = self.machine.full_history(); - context_history.push(model_response.message.clone()); - let context_tokens = self.token_counter.count(&context_history); - self.machine.model_response(model_response, context_tokens); - - let turn_index = current_turn; - let is_empty = tool_calls.is_empty(); - if let Some(run) = self.current_run_mut() { - run.turns.push(crate::engine::core::Turn { - turn: turn_index, - input: turn_input, - output: text, - tool_calls, - input_tokens: turn_in, - output_tokens: turn_out, - }); - } + if let Some(e) = self.apply_loop_detection(current_turn, &pattern) { + return Err(e); + } - if is_empty { - self.notify_turn_end( - current_turn, - true, - None, - turn_start.elapsed(), - turn_in, - turn_out, - ); + let tool_calls: Vec = msg + .tool_call_parts() + .into_iter() + .map(|(id, tool, input)| ToolCall { + id: id.to_string(), + tool: tool.to_string(), + input: input.clone(), + }) + .collect(); + let stop_reason = match stream_stop { + StreamStopReason::ToolCall => StopReason::ToolCall, + StreamStopReason::MaxTokens => StopReason::MaxTokens, + StreamStopReason::StopSequence => StopReason::StopSequence, + StreamStopReason::EndTurn => { + if tool_calls.is_empty() { + StopReason::EndTurn + } else { + StopReason::ToolCall } - Ok(()) } + }; + let model_response = ModelResponse { + message: msg, + input_tokens: turn_in, + output_tokens: turn_out, + stop_reason, + available_tools: self.tools.tool_names(), + }; + let mut context_history = self.machine.full_history(); + context_history.push(model_response.message.clone()); + let context_tokens = self.token_counter.count(&context_history); + self.machine.model_response(model_response, context_tokens); + + let turn_index = current_turn; + let is_empty = tool_calls.is_empty(); + if let Some(run) = self.current_run_mut() { + run.turns.push(crate::engine::core::Turn { + turn: turn_index, + input: turn_input, + output: text, + tool_calls, + input_tokens: turn_in, + output_tokens: turn_out, + }); + } + + if is_empty { + self.notify_turn_end( + current_turn, + true, + None, + turn_start.elapsed(), + turn_in, + turn_out, + ); } + Ok(()) } /// Handle a tool-dispatch request from the machine. @@ -1972,33 +1947,12 @@ impl BareLoop { self.notify_tool_calls_received(current_turn, &tool_calls); - let cancel = Arc::clone(&self.cancelled); - let dispatch = async { - self.dispatch_and_record(&dispatch_calls, current_turn, turn_start, turn_in, turn_out) - .await - }; - - let mut parts: Vec = tokio::select! { - biased; - () = cancel.notified() => { - self.machine.cancel(); - self.notify_turn_end( - current_turn, - false, - Some("cancelled".into()), - turn_start.elapsed(), - 0, - 0, - ); - return Err(LoopError::Cancelled); - } - result = dispatch => match result { - Ok(parts) => parts, - Err(e) => { - self.set_error_state(&e); - return Err(e); - } - }, + let mut parts: Vec = match self + .dispatch_and_record(&dispatch_calls, current_turn, turn_start, turn_in, turn_out) + .await + { + Ok(parts) => parts, + Err(e) => return Err(e), }; preresolved_parts.append(&mut parts); @@ -2182,7 +2136,7 @@ mod tests { use crate::observer::LoopObserver; use crate::stream::{ DeltaPart, IndexedDelta, MessageDelta, MessageDeltaPayload, MessageMetadata, MessageStart, - PartStart, StreamEvent, Usage, + PartStart, StreamAccumulator, StreamEvent, Usage, }; use crate::tool::ToolRegistry; use crate::tool::{Tool, ToolContext, ToolError, ToolOutput, ToolSchema}; @@ -2193,6 +2147,38 @@ mod tests { use std::sync::Mutex; + /// Fold queued [`StreamEvent`]s into a [`NonStreamingResponse`]. + /// + /// Shared by `MockClient` and `RecordingClient` `create_message` impls so + /// the non-streaming path sees the same assembled message, stop reason, + /// and usage the streaming path would have produced. + fn assemble_response( + events: Vec, + ) -> Result { + let mut accumulator = StreamAccumulator::new(); + let mut stop_reason = crate::stream::StreamStopReason::EndTurn; + for event in events { + if let crate::stream::StreamEvent::MessageDelta(delta) = &event + && let Some(reason) = delta + .delta + .stop_reason + .as_deref() + .and_then(crate::stream::StreamStopReason::from_api_str) + { + stop_reason = reason; + } + accumulator + .process(&event) + .map_err(|e| ApiError::api(e.to_string()))?; + } + let usage = accumulator.usage().copied(); + Ok(crate::api::NonStreamingResponse { + message: accumulator.build(), + stop_reason, + usage, + }) + } + #[derive(Clone)] struct MockClient { responses: Arc>>>, @@ -2471,28 +2457,7 @@ mod tests { drop(guard); Box::pin(async move { let events = events.ok_or_else(|| ApiError::api("No more mock responses"))?; - let mut accumulator = crate::stream::StreamAccumulator::new(); - let mut stop_reason = crate::stream::StreamStopReason::EndTurn; - for event in events { - if let crate::stream::StreamEvent::MessageDelta(delta) = &event - && let Some(reason) = delta - .delta - .stop_reason - .as_deref() - .and_then(crate::stream::StreamStopReason::from_api_str) - { - stop_reason = reason; - } - accumulator - .process(&event) - .map_err(|e| ApiError::api(e.to_string()))?; - } - let usage = accumulator.usage().copied(); - Ok(crate::api::NonStreamingResponse { - message: accumulator.build(), - stop_reason, - usage, - }) + assemble_response(events) }) } } @@ -2706,7 +2671,7 @@ mod tests { } #[test] - fn turn_mode_default_is_nonstreaming_without_feature() { + fn turn_mode_default_follows_streaming_feature() { let client = MockClient::new("test-model"); let agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); #[cfg(not(feature = "streaming"))] @@ -2784,7 +2749,12 @@ mod tests { _request: &crate::api::StreamRequest, ) -> Pin> + Send + 'static>> { - Box::pin(futures::stream::empty()) + let started = Arc::clone(&self.started); + Box::pin(futures::stream::once(async move { + started.store(true, Ordering::SeqCst); + std::future::pending::<()>().await; + Ok(StreamEvent::MessageStop) + })) } fn create_message( &self, @@ -2797,7 +2767,6 @@ mod tests { let started = Arc::clone(&self.started); Box::pin(async move { started.store(true, Ordering::SeqCst); - // Park forever; only the cancel `select!` arm resolves the turn. std::future::pending::<()>().await; Err(ApiError::api("unreachable: cancel must win the select")) }) @@ -2855,6 +2824,60 @@ mod tests { ); } + /// Streaming-path twin of the test above: a clean cancel during a + /// streaming turn must not fire `on_stream_failure`. Proves the + /// `record_turn_failure` Cancelled guard holds for both turn modes. + #[cfg(feature = "streaming")] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancel_during_streaming_turn_does_not_trip_breaker() { + let client = BlockingClient { + started: Arc::new(AtomicBool::new(false)), + }; + let started = Arc::clone(&client.started); + let on_stream_failure_fired = Arc::new(AtomicBool::new(false)); + let observer = Arc::new(FailureRecorder { + on_stream_failure_fired: Arc::clone(&on_stream_failure_fired), + }); + let managers = LoopManagers::new() + .with_fallback(FallbackManager::default()) + .with_observer(observer); + let mut agent = BareLoop::new_with_managers( + Arc::new(client), + ToolRegistry::new(), + make_config(), + managers, + ); + // turn_mode defaults to Streaming when the feature is on. + + let cancel_signal = Arc::clone(&agent.cancel_signal()); + let run_handle = tokio::spawn(async move { agent.run("Hi", &RunConfig::default()).await }); + + let mut waits = 0u32; + while !started.load(Ordering::SeqCst) { + waits += 1; + assert!( + waits <= 1000, + "stream_messages was never entered — test setup is broken" + ); + tokio::time::sleep(std::time::Duration::from_millis(2)).await; + } + cancel_signal.cancel(); + let run_result = run_handle.await.unwrap(); + + assert!( + started.load(Ordering::SeqCst), + "test only proves anything if stream_messages was actually entered" + ); + assert!( + matches!(run_result, Err(LoopError::Cancelled)), + "run must return Err(Cancelled): {run_result:?}" + ); + assert!( + !on_stream_failure_fired.load(Ordering::SeqCst), + "a clean cancel must not fire on_stream_failure (it would trip the breaker)" + ); + } + #[test] fn run_config_is_none_before_first_run() { let client = MockClient::new("test-model"); @@ -6066,28 +6089,7 @@ mod tests { drop(guard); Box::pin(async move { let events = events.ok_or_else(|| ApiError::api("No more mock responses"))?; - let mut accumulator = crate::stream::StreamAccumulator::new(); - let mut stop_reason = crate::stream::StreamStopReason::EndTurn; - for event in events { - if let crate::stream::StreamEvent::MessageDelta(delta) = &event - && let Some(reason) = delta - .delta - .stop_reason - .as_deref() - .and_then(crate::stream::StreamStopReason::from_api_str) - { - stop_reason = reason; - } - accumulator - .process(&event) - .map_err(|e| ApiError::api(e.to_string()))?; - } - let usage = accumulator.usage().copied(); - Ok(crate::api::NonStreamingResponse { - message: accumulator.build(), - stop_reason, - usage, - }) + assemble_response(events) }) } From 65d133a0bd4b67f69c1b83cfc3147fe62d4b08b7 Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Sun, 2 Aug 2026 20:48:58 +1200 Subject: [PATCH 3/3] chore: 0.2.0 prep --- CHANGELOG.md | 237 ++++++++++++++--------------------------------- Cargo.toml | 3 +- README.md | 36 ++++--- examples/chat.rs | 16 ++-- src/provider.rs | 2 +- 5 files changed, 101 insertions(+), 193 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e28be24..cfc2981 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,191 +7,87 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ## [Unreleased] -### Added - -- **Non-streaming engine turn path** (`TurnMode`): the engine can now drive each - turn via `ApiClient::create_message` instead of streaming, selected at runtime - with `BareLoop::set_turn_mode` / `with_turn_mode`. The default is - `TurnMode::Streaming` when `streaming` is compiled in, `TurnMode::NonStreaming` - otherwise. Under the non-streaming path no per-delta observer callbacks fire; - the full assistant text still surfaces via `on_response`. -- **`streaming` feature flag**: gates `StreamHandler`, `stream::handler`, - `StreamCapable`, `text_streamer`, and the `on_text_delta` / `on_thinking_delta` - firing sites. With `default = []`, `async-stream` is no longer pulled and the - engine compiles and runs without any streaming machinery. - -### Changed - -- **`default = []` now means non-streaming.** `async-stream` is an optional - dependency enabled by `streaming`; it is no longer pulled into a bare - `cargo add loopctl`. The HTTP provider features (`openai`, `anthropic`, - `gemini`, and anything that chains from them) now imply `streaming`, so the - common `features = ["openai"]` case preserves the previous streaming-by-default - behavior. Migration: users who enabled only `providers` (not a named provider) - and relied on streaming must add `streaming` explicitly. -- **MSRV corrected to 1.94** (was misdocumented as 1.85 in `AGENTS.md`; the - `Cargo.toml` `rust-version` and `.clippy.toml` already required 1.94). -- `AGENTS.md` feature table, `providers` dependency list, core-deps list, and the - `LoopMemory` object-safety note corrected to match the code. - -### Removed - -- The `record_stream_success` / `record_stream_failure` private methods are - renamed to `record_turn_success` / `record_turn_failure` (shared by both turn - paths). The `StreamCapable` trait and its `LoopManagers` impl now require the - `streaming` feature. - +## [0.2.0] - 2026-08-02 ### Added -- **Sans-IO state machine** (`engine::machine::LoopMachine`): serializable, - owns every agent-loop decision (turn counting, max-turn enforcement, - tool-call validity, compaction trigger, history, cancellation). Exposed via - `BareLoop::machine()` / `into_machine()` / `from_machine()` for inspection - and serialize-and-resume. `BareLoop::run()` now drives the machine internally. -- **Session/Run/Turn lifetime model** (`engine::core`): one `Session` spans the - process, one `Run` per `run()` call, one `Turn` per loop iteration. Session - derives per-session totals; `Run` carries per-run turns, tokens, and error. - Construction splits into `SessionConfig` (session-scoped) and `RunConfig` - (per-run budgets). -- **Reasoning-model support** (`DeltaPart::Thinking` + `on_thinking_delta`): - reasoning tokens (Claude extended-thinking, OpenAI o-series, Gemini 2.5+) are - routed as their own stream kind. Stream-only — not accumulated into `Message`. -- **Structured output** (`structured` module): `StructuredOutput` trait, - `ResponseFormat`, `request_structured::()`. All three providers override - `stream_messages_with_options` / `create_message_with_options` to inject the - schema natively. -- **Tool constraints** (`ToolConstraint` enum): `Strict` tightens tool schemas - via the provider's native strict mode; `Grammar` compiles schemas into a - grammar for vLLM-style samplers (`grammar` feature). -- **Tool reflection** (`LlmReflector`): asks the model to classify failed tool - calls and suggest corrections via `request_structured`. -- **Parallel tool dispatch** (`ParallelDispatchConfig`): independent, - concurrency-safe calls within a single turn run concurrently. Sequential by - default. -- **Stream resilience** (`StreamHandler`): retries, timeouts, rate-limit - backoff, and non-streaming fallback. Configurable via - `with_timeout_config` / `with_retry_config` / `with_rate_limit_config`. - `HandlerEvent` enum provides real-time observability during streaming. -- **Client-side rate limiting** (`TokenBucket` / `RateLimiter`): proactive - per-provider token-bucket. One bucket per `base_url`. -- **Context contributors** (`ContextContributor` trait): turn-boundary hook for - injecting messages before each model call. -- **Agent memory** (`LoopMemory` trait, now wired): stores tool-call - trajectories, retrieves relevant entries before each turn, consolidates on - successful runs. Object-safe; configure via `BareLoop::set_memory`. -- **Display hints** (`DisplayHint` on `ToolOutput`): advisory rendering hints - (Text, Diff, Json, Code, Suppress, Markdown) for presentation layers. -- **Middleware**: `VerifyMiddleware` (post-execution verification), - `MemoizingMiddleware` (tool-call result caching with path-aware invalidation). -- **Presets** (`ConstrainedProfile`, `FrontierProfile`, `GoalReminder`): named - runtime profiles for small-model-tuned and frontier configurations. -- **`StreamRequest`**: bundles `(messages, system, tools)` into one parameter - for all `ApiClient` methods. -- **`Role::System`** variant: framework-injected system context. Providers map - to their native representation. -- **Pluggable `TokenCounter`**: `HeuristicTokenCounter` (4 chars/token) default; - swap in a real tokenizer. Synced bidirectionally with `ContextManager`. -- **`OpenAiClientBuilder::with_stream_usage(bool)`**: controls - `stream_options.include_usage`. `ollama()` disables it automatically. -- **`with_tcp_nodelay(bool)`** on all three provider builders + `HttpClientConfig`. -- **HTTP connection-pool injection**: shared `reqwest::Client`, pool knobs - (`pool_max_idle_per_host`, `pool_idle_timeout`, `tcp_keepalive`). -- **Fluent `with_*()` builders** on `LoopConfig`, `BareLoop`, and all provider - builders. `CancelSignal::reset()` for multi-run agents. -- `LoopError` now derives `Serialize`, `Deserialize`, `PartialEq`, `Eq`. +- `TurnMode` enum: engine runs non-streaming (`create_message`) or streaming (`StreamHandler`), selectable at runtime via `set_turn_mode`. Default is feature-dependent. +- `streaming` feature: gates `StreamHandler`, per-delta callbacks, `async-stream`. With `default = []`, no streaming code is compiled. +- Sans-IO `LoopMachine`: serializable, owns all loop decisions. Exposed via `machine()` / `into_machine()` / `from_machine()`. +- Session/Run/Turn lifetime model: `SessionConfig` (session-scoped) + `RunConfig` (per-run budgets). +- Reasoning-model support: `DeltaPart::Thinking` + `on_thinking_delta`. +- Structured output: `StructuredOutput` trait, `ResponseFormat`, `request_structured::()`. +- Tool constraints (`ToolConstraint`): `Strict` and `Grammar` modes. +- Tool reflection (`LlmReflector`): model classifies failed tool calls and suggests corrections. +- Parallel tool dispatch (`ParallelDispatchConfig`). +- `StreamHandler`: retry, timeout, rate-limit backoff, non-streaming fallback. +- Client-side rate limiting: `TokenBucket` / `RateLimiter`, one bucket per `base_url`. +- `ContextContributor` trait: turn-boundary message injection. +- `LoopMemory` trait wired: stores trajectories, retrieves before turns, consolidates on success. +- `DisplayHint` on `ToolOutput`: rendering hints (Text, Diff, Json, Code, Suppress, Markdown). +- Middleware: `VerifyMiddleware`, `MemoizingMiddleware`. +- Presets: `ConstrainedProfile`, `FrontierProfile`, `GoalReminder`. +- `StreamRequest`: bundles `(messages, system, tools)` for all `ApiClient` methods. +- `Role::System` variant. +- Pluggable `TokenCounter` (`HeuristicTokenCounter` default). +- `OpenAiClientBuilder::with_stream_usage(bool)`. +- `with_tcp_nodelay(bool)` on all provider builders. +- HTTP connection-pool injection: shared `reqwest::Client`, pool knobs. +- Fluent `with_*()` builders on `BareLoop` and provider builders. +- `LoopError` derives `Serialize`, `Deserialize`, `PartialEq`, `Eq`. ### Changed -- **Breaking (`create_message` returns typed `NonStreamingResponse`):** no longer - returns raw `serde_json::Value`. The struct carries `message: Message`, - `stop_reason: StreamStopReason`, `usage: Option`. Migration: read fields - from the struct. -- **Breaking (`extract_structured` takes `&Message`):** default implementation - derives from the message; per-provider overrides removed. Migration: change - parameter type or delete override. -- **Breaking (`run()` signature):** `run(&mut self, input: &str, &RunConfig)` - returns `Result`. `Loop::initialize` / `config()` removed. -- **Breaking (session→run lifecycle rename):** `on_session_start` / `on_session_end` - → `on_run_start` / `on_run_end`. Fire on every `run()` call. Migration: rename - methods and context types. -- **Breaking (`LoopConfig` removed):** replaced by `SessionConfig` + `RunConfig`. - The `model` field lives on `ApiClient`. -- **Breaking (compaction thresholds → percentages):** `f64` fractions → `u8` - percentages (0–100). `0.80 → 80`. -- **Breaking (builder renames):** all consuming builders uniformly `with_`-prefixed. - `Option` builders take `Option` (no `_opt` suffix). Migration: add prefix, - wrap literals in `Some(...)`. -- **Breaking (`StreamHandler::stream_turn`):** returns `impl Stream>` instead of a future. -- **Breaking (`StreamRequest` parameter):** all `ApiClient` streaming/creation - methods take `&StreamRequest` instead of positional params. -- **Breaking (`DeltaPart` non-exhaustive):** add `_ =>` arm to downstream matches. -- **Breaking (`Role::System` variant):** add `System =>` arm to downstream matches. -- **Breaking (`ToolOutput` / `ToolDispatchResult` / `ToolPostContext` - non-exhaustive):** use named constructors or `..Default::default()`. -- **Breaking (`Reflector::analyze`):** gains `tool_schema: Option<&ToolSchema>` - parameter. -- **Breaking (`MessagePart::ToolResult` gains `name` field):** - `tool_result()` gains a `name` argument (second param). `#[serde(default)]` - allows deserializing older data. -- **Breaking (`NonStreamingResponse.usage` is `Option`):** symmetric with - streaming. `None` = provider omitted usage. -- **OpenAI streaming usage:** now sets `stream_options.include_usage` and captures - token counts from the final chunk. All three providers report usage on both paths. - All-zero usage collapses to `None`. -- **OpenAI malformed arguments:** non-empty `function.arguments` that fail JSON - parse now return an `ApiError` instead of silently defaulting to `{}`. -- **Gemini tool-call ids:** `functionCall.id` is now parsed (Gemini 3) and echoed - in `functionResponse.id`. `functionResponse` sends both `name` and `id`. -- **SSE invalid-UTF-8:** `take_line` surfaces invalid UTF-8 as a protocol error - instead of silent `U+FFFD` replacement. -- **`StreamStopReason::from_api_str`** accepts `"tool_use"` as alias for - `"tool_call"` (Anthropic). -- **`set_token_counter`** now propagates to the `ContextManager` if one is set, - regardless of setter order. -- **Memory injection** uses `Role::User` (not `Role::System`) with explicit - "reference only" delimitation. -- **Context token estimate** includes the model response message before counting. -- MSRV bumped to 1.94 (let-chain syntax). Removed `parking_lot` dependency. +- **Breaking:** `default = []` no longer pulls `async-stream`; streaming is opt-in via `streaming`. HTTP providers imply it, so `features = ["openai"]` is unchanged. Migration: add `streaming` if you used `providers` alone. +- **Breaking:** `TurnMode` no longer implements `Default`. Migration: use `turn_mode()` / `set_turn_mode()`. +- **Breaking:** `create_message` returns typed `NonStreamingResponse` instead of `serde_json::Value`. +- **Breaking:** `extract_structured` takes `&Message`; per-provider overrides removed. +- **Breaking:** `run()` is `run(&mut self, &str, &RunConfig) -> Result`. `Loop::initialize` / `config()` removed. +- **Breaking:** `on_session_start`/`on_session_end` renamed to `on_run_start`/`on_run_end`. +- **Breaking:** `LoopConfig` removed; split into `SessionConfig` + `RunConfig`. +- **Breaking:** compaction thresholds are `u8` percentages (0–100) instead of `f64`. +- **Breaking:** builders uniformly `with_`-prefixed; `Option` builders take `Option`. +- **Breaking:** `StreamHandler::stream_turn` returns `impl Stream>`. +- **Breaking:** `ApiClient` methods take `&StreamRequest` instead of positional params. +- **Breaking:** `DeltaPart`, `Role`, `ToolOutput`, `ToolDispatchResult`, `ToolPostContext` are `#[non_exhaustive]`. +- **Breaking:** `Reflector::analyze` gains `tool_schema: Option<&ToolSchema>`. +- **Breaking:** `MessagePart::ToolResult` gains `name` field (serde-defaulted for old data). +- **Breaking:** `NonStreamingResponse.usage` is `Option`. +- Cancellation no longer trips the fallback breaker (`record_turn_failure` guards `Cancelled`). +- OpenAI streaming sets `stream_options.include_usage`; all providers report usage on both paths. +- OpenAI malformed `function.arguments` returns `ApiError` instead of defaulting to `{}`. +- Gemini parses `functionCall.id` (Gemini 3) and echoes it in responses. +- SSE invalid UTF-8 surfaced as protocol error instead of `U+FFFD` replacement. +- `StreamStopReason::from_api_str` accepts `"tool_use"` (Anthropic alias). +- MSRV bumped to 1.94. ### Removed -- `StreamHandler::with_config(timeout, retry)` — use `with_timeout_config` / - `with_retry_config`. -- `FallbackManager::record_api_failure` / `record_model_failure` — merged into - `record_failure(FailureKind)`. -- `Loop::process_turn`, `BareLoop::run_turn_body` — replaced by machine-driven - `run()`. -- `StreamTurnResult` — engine assembles result from event stream. -- `StreamHandler::with_request_options` — use `BareLoop::set_request_options`. -- `StreamHandlerError::RateLimitEscalation.prior` field (never read). +- `record_stream_success` / `record_stream_failure` (renamed to `record_turn_*`). +- `StreamCapable` trait now requires the `streaming` feature. +- `StreamHandler::with_config(timeout, retry)` — use `with_timeout_config` / `with_retry_config`. +- `FallbackManager::record_api_failure` / `record_model_failure` — merged into `record_failure(FailureKind)`. +- `Loop::process_turn`, `BareLoop::run_turn_body` — replaced by machine-driven `run()`. +- `StreamTurnResult`, `StreamHandler::with_request_options`, `StreamHandlerError::RateLimitEscalation.prior`. +- `parking_lot` dependency. ### Fixed -- OpenAI streaming dropped multi-chunk tool-call argument fragments after the - first; re-opened tool-call parts on every chunk carrying `function`. -- `StreamAccumulator` dropped parallel tool calls whose arguments arrived - interleaved. -- `BareLoop` was permanently dead after a single cancellation (`CancelSignal` - not re-armed). Now resets in `finalize()`. -- Tool results were split across multiple user messages instead of merged into - one per turn. -- `StreamHandler` accepted invalid timeout/retry configs (validation never - called); `jitter_factor` was validated but never applied. -- Empty-stream fast-fail: zero-event streams could hang for ~20 min before - failing. -- `ToolHealthRegistry::is_tool_available` consumed the HalfOpen recovery probe - as a side effect of a read. -- Anthropic provider hardcoded text-block index to 0, ignoring server index. -- Per-run manager reset wiped session-scoped state between runs. +- Release profile no longer uses `panic = "abort"` (it disabled `catch_unwind` tool-panic isolation). +- OpenAI streaming dropped multi-chunk tool-call argument fragments. +- `StreamAccumulator` dropped parallel tool calls with interleaved arguments. +- `BareLoop` was dead after one cancellation (`CancelSignal` not re-armed; now resets in `finalize()`). +- Tool results split across multiple user messages instead of merged per turn. +- `StreamHandler` accepted invalid timeout/retry configs; `jitter_factor` was validated but never applied. +- Zero-event streams could hang ~20 min before failing. +- `ToolHealthRegistry::is_tool_available` consumed the HalfOpen probe as a read side effect. +- Anthropic provider hardcoded text-block index to 0. +- Per-run manager reset wiped session-scoped state. ### Security -- Auto-commit hook's `git add -A` on empty file list staged the entire working - tree. Now refuses with an error. -- Non-streaming response body size guard (10 MB) fired after full - materialization. Now pre-checks `Content-Length` and caps streaming reads. +- Auto-commit hook's `git add -A` on empty file list staged the whole working tree; now refuses. +- Response body size guard (10 MB) now pre-checks `Content-Length` instead of firing after full materialization. ## [0.1.0] - 2025-07-01 @@ -214,4 +110,5 @@ Initial crates.io release. - Built-in testing utilities for writing LLM loop tests - Example CLIs: hello, REPL, echo tool, and multi-provider chat +[0.2.0]: https://github.com/dch-labs/loopctl/releases/tag/v0.2.0 [0.1.0]: https://github.com/dch-labs/loopctl/releases/tag/v0.1.0 diff --git a/Cargo.toml b/Cargo.toml index 5b54692..c4e7a24 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "loopctl" -version = "0.1.0" +version = "0.2.0" edition = "2024" license = "MIT OR Apache-2.0" description = "A trait-based framework for building agent loops with pluggable LLM clients, tools, and memory" @@ -100,5 +100,4 @@ rustdoc-args = ["--cfg", "docsrs"] lto = "fat" codegen-units = 1 strip = true -panic = "abort" diff --git a/README.md b/README.md index 32def47..8a7197b 100644 --- a/README.md +++ b/README.md @@ -161,25 +161,35 @@ let agent = BareLoop::new( ### Streaming vs non-streaming -By default (`default = []`) the engine drives each turn with -[`ApiClient::create_message`] — a single request/response with no streaming -machinery, no `async-stream` dependency, and no per-delta callbacks. The full -assistant text still surfaces through -[`on_response`](https://docs.rs/loopctl/latest/loopctl/observer/trait.LoopObserver.html#method.on_response). - -Enable the `streaming` feature (implied by every HTTP provider) to route turns -through [`StreamHandler`](https://docs.rs/loopctl/latest/loopctl/stream/handler/struct.StreamHandler.html) -with retry, timeout, rate-limit detection, and `on_text_delta` / -`on_thinking_delta` callbacks for real-time token display. Switch a constructed -loop explicitly with +The engine selects a turn mode at runtime via +[`TurnMode`](https://docs.rs/loopctl/latest/loopctl/engine/enum.TurnMode.html) +(`NonStreaming` or `Streaming`), set with [`set_turn_mode`](https://docs.rs/loopctl/latest/loopctl/engine/struct.BareLoop.html#method.set_turn_mode). +The two modes are independent of *whether* the `streaming` feature is compiled +in, though the feature gates what `Streaming` can do: + +- [`TurnMode::NonStreaming`](https://docs.rs/loopctl/latest/loopctl/engine/enum.TurnMode.html#variant.NonStreaming) + drives each turn via [`ApiClient::create_message`] — a single request/response + with no per-delta callbacks. The full assistant text surfaces through + [`on_response`](https://docs.rs/loopctl/latest/loopctl/observer/trait.LoopObserver.html#method.on_response). + Always available, even under `default = []`. + +- [`TurnMode::Streaming`](https://docs.rs/loopctl/latest/loopctl/engine/enum.TurnMode.html#variant.Streaming) + routes turns through [`StreamHandler`](https://docs.rs/loopctl/latest/loopctl/stream/handler/struct.StreamHandler.html) + with retry, timeout, rate-limit detection, and `on_text_delta` / + `on_thinking_delta` callbacks for real-time token display. Requires the + `streaming` feature (implied by every HTTP provider). + +The constructor default is `Streaming` when the `streaming` feature is enabled +and `NonStreaming` otherwise — but a constructed loop can switch to either mode +at runtime regardless of the default. ## Architecture At the center is **BareLoop**, the default agent loop. Each turn it requests a response from an **ApiClient** (your LLM provider) — via the streaming path -(`StreamHandler`) when `streaming` is enabled, or via `create_message` -otherwise — then dispatches any requested tool calls through a +(`StreamHandler`) under `TurnMode::Streaming`, or via `create_message` under +`TurnMode::NonStreaming` — then dispatches any requested tool calls through a **ToolRegistry**. Results are fed back into the conversation and the cycle repeats until the model ends its turn or a configured limit is reached. diff --git a/examples/chat.rs b/examples/chat.rs index e114ec0..d20354d 100644 --- a/examples/chat.rs +++ b/examples/chat.rs @@ -335,22 +335,24 @@ async fn run_repl(client: Arc) { // Pick the engine turn mode at runtime. `NO_STREAM=1` drives each turn via // the non-streaming `create_message` path (no per-delta callbacks, the - // assembled response is printed after the turn). Otherwise stream text - // deltas live as they arrive. - let no_stream = std::env::var("NO_STREAM").map_or(false, |v| v == "1"); + // assembled response is printed after the turn). Without the `streaming` + // feature the engine is non-streaming regardless, and `no_stream` stays + // true so the result-printing logic below displays each response. + let no_stream = if cfg!(not(feature = "streaming")) { + true + } else { + std::env::var("NO_STREAM").map_or(false, |v| v == "1") + }; #[cfg(feature = "streaming")] if no_stream { agent.set_turn_mode(loopctl::engine::TurnMode::NonStreaming); } else { + agent.set_turn_mode(loopctl::engine::TurnMode::Streaming); agent.set_text_streamer(Arc::new(|delta| { print!("{delta}"); let _ = std::io::stdout().flush(); })); } - #[cfg(not(feature = "streaming"))] - { - let _ = no_stream; - } // Ctrl-C interrupts the in-flight turn (via loopctl's CancelSignal, which // `select!`s against the stream) and ends the session. The token is diff --git a/src/provider.rs b/src/provider.rs index fd2b6ea..27ecce1 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -127,7 +127,6 @@ pub(super) async fn read_bounded_body(resp: reqwest::Response) -> Result Result