diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c4218e..3cea39c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Added +- `ConstrainedProfile`, `FrontierProfile`, and `GoalReminder` (`presets` + module): a named small-model-tuned runtime profile and its frontier opt-out + counterpart. `ConstrainedProfile::apply(&mut loop_)` wires the small-model + middleware stack (verify with `NoopVerifier`, memoize with + `NoopPathExtractor`, output cap) and registers a `GoalReminder` contributor; + `loop_config()` returns a tighter `LoopConfig` (120k window, 100 turns) and + `request_options()` returns `tool_constraint: Strict`. Compose the pieces + individually via `pipeline_builder()` / `loop_config()` / `request_options()`. +- `NoopVerifier` (`middleware::verify`): a `Verifier` that always passes, + co-located with the `Verifier` trait. Default verifier for + `ConstrainedProfile`; swap in a real build/lint step when available. +- `NoopPathExtractor` (`middleware::memoize`): a `PathExtractor` that extracts + no paths, disabling path-based cache invalidation (TTL-only caching). + Default extractor for `ConstrainedProfile`. +- `BareLoop::set_request_options(opts)` builder: set the per-turn + `RequestOptions` (carrying `tool_constraint`) applied to every provider call. + Default is `RequestOptions::default()` (no constraint), reproducing prior + behavior. +- `StreamHandler::with_request_options(opts)` builder: the same option applied + to the handler's stream-open call. - `ContextContributor` trait and `ContributorContext<'a>` (`engine::contributor` module): a write-side hook at the turn boundary. Implementors return an optional `Message` that the loop appends to the conversation before the next @@ -107,6 +127,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Changed +- The engine now calls `ApiClient::stream_messages_with_options` instead of + `stream_messages` on every turn (both the inline-streaming path in + `engine::bare::stream` and the `StreamHandler` path), passing the loop's + `RequestOptions`. This is additive — the default `RequestOptions::default()` + has no `response_format` and `tool_constraint: None`, which reproduces the + prior behavior exactly. Custom `ApiClient` impls that do not override + `stream_messages_with_options` inherit the trait default (which delegates to + `stream_messages` when `response_format` is `None`), so they continue to + work; a `tool_constraint` other than `None` set via `set_request_options` + only takes effect on clients that override `_with_options` (the built-in + OpenAI, Anthropic, and Gemini clients do). - **Breaking:** `message::Role` gains a `System` variant. Every exhaustive `match` on `Role` in downstream code must add a `System =>` arm (or a `_ =>` wildcard). Migration: add `Role::System => /* your mapping */` to each match, diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 15e3964..5fe7738 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -87,6 +87,7 @@ use crate::reflection::{ use crate::runtime::LoopRuntime; use crate::stream::handler::StreamHandler; use crate::stream::{StreamAccumulator, StreamEvent, StreamStopReason, Usage}; +use crate::structured::RequestOptions; #[cfg(feature = "tool_health")] use crate::tool::health::ToolHealthRegistry; use crate::tool::{PermissionCheck, ToolContext, ToolDispatchResult, ToolRegistry, ToolSchema}; @@ -215,6 +216,16 @@ pub struct BareLoop { cancelled: Arc, /// Current lifecycle state, exposed via the [`Loop`](crate::engine::loop_core::Loop) trait. + /// + /// Drives the engine's state machine + /// (`Idle → Processing → WaitingForTool → Processing → … → Completed` / + /// `Failed` / `Cancelled`, plus `Compacting`/`Reflecting` side-states). + /// Set throughout the turn loop and read by [`finalize`](crate::engine::loop_core::Loop::finalize) + /// to decide success vs. failure and to populate the + /// [`SessionResult`](crate::engine::loop_core::SessionResult) fields; also + /// returned from [`state`](crate::engine::loop_core::Loop::state) for + /// outside observers. The [`Idle`](LoopState::Idle) variant additionally + /// gates configuration setters via [`debug_assert_idle`](Self::debug_assert_idle). state: LoopState, /// Session-level accumulator for turn counts, token usage, and tool calls. @@ -224,6 +235,13 @@ pub struct BareLoop { budget: SessionResult, /// Session start time, set by [`initialize`](crate::engine::loop_core::Loop::initialize). + /// + /// `None` only before the first `initialize()` call; `Some` thereafter. + /// Read in [`finalize`](crate::engine::loop_core::Loop::finalize) to + /// compute [`SessionResult::total_duration`](crate::engine::loop_core::SessionResult) + /// via `elapsed()`. Captured once per session (not per turn) so the + /// reported duration is the wall-clock lifetime of the whole session, + /// not a single turn. session_start: Option, /// Optional callback invoked for each text delta during streaming. @@ -242,6 +260,14 @@ pub struct BareLoop { /// conversation in registration order. Register via /// [`add_contributor`](BareLoop::add_contributor). contributors: Vec>, + + /// Per-turn `RequestOptions` applied to every provider call. + /// + /// Carries `tool_constraint` (strict/grammar-constrained tool-call + /// decoding). Default is [`RequestOptions::default`], which reproduces + /// the prior (unconstrained) behavior. Set via + /// [`set_request_options`](BareLoop::set_request_options). + request_options: RequestOptions, } // ================================================== @@ -293,6 +319,7 @@ impl BareLoop { session_start: None, text_streamer: None, contributors: Vec::new(), + request_options: RequestOptions::default(), } } @@ -344,6 +371,7 @@ impl BareLoop { session_start: None, text_streamer: None, contributors: Vec::new(), + request_options: RequestOptions::default(), } } @@ -739,6 +767,36 @@ impl BareLoop { self.contributors.push(contributor); } + /// Set the per-turn [`RequestOptions`] applied to every provider call. + /// + /// Carries [`tool_constraint`](crate::structured::ToolConstraint) — set to + /// [`ToolConstraint::Strict`](crate::structured::ToolConstraint::Strict) + /// for strict tool-call decoding (small-model reliability), or leave at the + /// default ([`RequestOptions::default`]) for unconstrained behavior. + /// + /// Must be called before + /// [`run`](crate::engine::loop_core::Loop::run). + /// + /// # Panics (debug only) + /// + /// In debug builds, panics if called after the session has started + /// (i.e., when [`state`](LoopState) is not [`Idle`](LoopState::Idle)). + /// + /// # Example + /// + /// ```rust,ignore + /// use loopctl::structured::{RequestOptions, ToolConstraint}; + /// + /// let mut agent = BareLoop::new(client, registry, config); + /// agent.set_request_options( + /// RequestOptions::new().tool_constraint(ToolConstraint::Strict), + /// ); + /// ``` + pub fn set_request_options(&mut self, options: RequestOptions) { + self.debug_assert_idle(); + self.request_options = options; + } + /// Begin a model switch operation. /// /// Returns a [`ModelSwitch`] builder that lets you optionally update @@ -3831,6 +3889,7 @@ mod tests { struct RecordingClient { responses: Arc>>>, seen: Arc>>>, + seen_options: Arc>>, model_name: Arc>, } @@ -3839,6 +3898,7 @@ mod tests { Self { responses: Arc::new(Mutex::new(Vec::new())), seen: Arc::new(Mutex::new(Vec::new())), + seen_options: Arc::new(Mutex::new(Vec::new())), model_name: Arc::new(Mutex::new(model.to_string())), } } @@ -3944,6 +4004,13 @@ mod tests { fn call_count(&self) -> usize { crate::error::recover_guard(self.seen.lock()).len() } + + fn first_options(&self) -> crate::structured::RequestOptions { + crate::error::recover_guard(self.seen_options.lock()) + .first() + .expect("at least one stream_messages_with_options call") + .clone() + } } impl ApiClient for RecordingClient { @@ -3978,6 +4045,27 @@ mod tests { } } + fn stream_messages_with_options( + &self, + messages: Vec, + _system: Option, + _tools: Option>, + options: crate::structured::RequestOptions, + ) -> Pin> + Send + 'static>> + { + crate::error::recover_guard(self.seen.lock()).push(messages); + crate::error::recover_guard(self.seen_options.lock()).push(options); + let mut guard = crate::error::recover_guard(self.responses.lock()); + if let Some(events) = guard.pop_front() { + let events: Vec> = + events.into_iter().map(Ok).collect(); + Box::pin(futures::stream::iter(events)) + } else { + let err = ApiError::api("No more mock responses"); + Box::pin(futures::stream::iter(vec![Err(err)])) + } + } + fn create_message( &self, _messages: Vec, @@ -4249,4 +4337,116 @@ mod tests { fn _suppress_recording_client_dead_code(c: &RecordingClient) { let _ = c.call_count(); } + + // ========================================================== + // RequestOptions engine plumbing + // ========================================================== + + #[tokio::test] + async fn test_request_options_default_is_unconstrained() { + // A fresh BareLoop has default RequestOptions — the engine reproduces + // v0.1.0 behavior (no tool_constraint). + let client = RecordingClient::new("test-model"); + client.add_text_response("done"); + let config = contributor_config(); + let mut agent = BareLoop::new(Arc::new(client.clone()), ToolRegistry::new(), config); + // No set_request_options call — default path. + agent.run("Hi").await.unwrap(); + + let opts = client.first_options(); + assert!( + matches!( + opts.tool_constraint, + crate::structured::ToolConstraint::None + ), + "default request options must be unconstrained" + ); + } + + #[tokio::test] + async fn test_request_options_strict_reaches_provider() { + // The critical end-to-end proof: a tool_constraint: Strict set on the + // loop reaches the provider's stream_messages_with_options call. + let client = RecordingClient::new("test-model"); + client.add_text_response("done"); + let config = contributor_config(); + let mut agent = BareLoop::new(Arc::new(client.clone()), ToolRegistry::new(), config); + agent.set_request_options( + crate::structured::RequestOptions::new() + .tool_constraint(crate::structured::ToolConstraint::Strict), + ); + agent.run("Hi").await.unwrap(); + + let opts = client.first_options(); + assert!( + matches!( + opts.tool_constraint, + crate::structured::ToolConstraint::Strict + ), + "Strict set on the loop must reach the provider" + ); + } + + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "configuration setters must be called before run()")] + fn test_set_request_options_panics_after_session_start() { + let client = MockClient::new("test-model"); + let config = contributor_config(); + let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), config); + let cfg = LoopConfig { + max_turns: 10, + ..Default::default() + }; + { + let fut = agent.initialize(&cfg); + let mut fut = std::pin::pin!(fut); + futures::executor::block_on(fut.as_mut()).unwrap(); + } + agent.set_request_options(crate::structured::RequestOptions::default()); + } + + // ========================================================== + // ConstrainedProfile::apply + // ========================================================== + + #[tokio::test] + async fn test_constrained_apply_wires_pipeline_and_contributor() { + // Apply() sets the small-model pipeline and registers a GoalReminder. To prove + // the contributor wiring without driving 5 turns (each turn ends on + // end_turn, so reaching turn 5 needs a long tool-call chain), we add + // a cadence-1 GoalReminder on top: it fires on turn 1, so a single + // tool-then-text session (2 turns) is enough. + let mut registry = ToolRegistry::new(); + registry.register(EchoTool); + + let client = RecordingClient::new("test-model"); + client.add_tool_then_text("t1", "echo", json!({"message": "x"}), "done"); + + let mut agent = BareLoop::new(Arc::new(client.clone()), registry, contributor_config()); + // apply() wires the pipeline + a cadence-5 GoalReminder. + crate::presets::ConstrainedProfile::apply(&mut agent).unwrap(); + // Add a cadence-1 reminder so it fires this session. + agent.add_contributor(Box::new(crate::presets::GoalReminder::new(1))); + + let result = agent.run("ship the demo goal").await.unwrap(); + // Tool-call turn + end_turn = 2 turns. + assert!(result.total_turns >= 1); + + // The contributor fired: a Role::System message carrying the first + // user message text reached the provider on some turn's outbound + // conversation. Scan all recorded calls (the reminder fires on turn 1, + // not turn 0). + let all_seen = crate::error::recover_guard(client.seen.lock()).clone(); + let has_reminder = all_seen.iter().flatten().any(|m| { + m.role == Role::System + && m.parts.iter().any( + |p| matches!(p, MessagePart::Text { text } if text.contains("ship the demo goal")), + ) + }); + assert!( + has_reminder, + "GoalReminder (cadence 1) should have injected the goal text as a System message" + ); + } } diff --git a/src/engine/bare/stream.rs b/src/engine/bare/stream.rs index f0c4106..0285252 100644 --- a/src/engine/bare/stream.rs +++ b/src/engine/bare/stream.rs @@ -64,9 +64,12 @@ impl BareLoop { // cost of serialising the messages into an HTTP request body. For // very long sessions (>200 turns with large tool outputs), consider // enabling auto-compaction to bound the history size. - let mut stream = - self.client - .stream_messages(self.conversation.clone(), system, tool_schemas); + let mut stream = self.client.stream_messages_with_options( + self.conversation.clone(), + system, + tool_schemas, + self.request_options.clone(), + ); let mut accumulator = StreamAccumulator::new(); let mut stop_reason = StreamStopReason::EndTurn; loop { diff --git a/src/lib.rs b/src/lib.rs index 092c053..0a8b0f7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -78,6 +78,7 @@ pub mod memory; pub mod message; pub mod middleware; pub mod observer; +pub mod presets; #[cfg(feature = "providers")] pub mod provider; pub mod reflection; diff --git a/src/middleware.rs b/src/middleware.rs index ccf80d7..8dd8bd7 100644 --- a/src/middleware.rs +++ b/src/middleware.rs @@ -65,13 +65,13 @@ use std::sync::Arc; pub use crate::tool::ToolDispatchResult; -pub use memoize::{MemoizingMiddleware, PathExtractor}; +pub use memoize::{MemoizingMiddleware, NoopPathExtractor, PathExtractor}; pub use output_limit::OutputLimitMiddleware; pub use permission::{AskResolverFn, PermissionCheckFn, PermissionMiddleware}; pub use timeout::{TimeoutConfig, TimeoutMiddleware}; pub use tool_call::ToolCallMiddleware; pub use unknown_tool::UnknownToolMiddleware; -pub use verify::{Verifier, VerifyMiddleware, VerifyResult}; +pub use verify::{NoopVerifier, Verifier, VerifyMiddleware, VerifyResult}; // ================================================== // Dispatch context diff --git a/src/middleware/memoize.rs b/src/middleware/memoize.rs index 0697b95..8f6b577 100644 --- a/src/middleware/memoize.rs +++ b/src/middleware/memoize.rs @@ -86,6 +86,36 @@ pub trait PathExtractor: Send + Sync { fn paths(&self, tool_name: &str, input: &Value) -> Vec; } +/// A [`PathExtractor`] that extracts no paths. +/// +/// Disables path-based cache invalidation entirely: cached entries are evicted +/// only by TTL (`ttl_turns`), never by write-class calls. Suitable when the +/// caller cannot (or does not want to) associate tool inputs with filesystem +/// paths — e.g. for a tool whose output is deterministic and path-independent. +/// +/// Zero-sized; cheap to construct and `Arc`-share. +/// +/// # Example +/// +/// ```rust,ignore +/// use std::sync::Arc; +/// use loopctl::middleware::{MemoizingMiddleware, NoopPathExtractor}; +/// +/// let mw = MemoizingMiddleware::new( +/// vec!["Read".into()], +/// vec!["Write".into()], +/// Arc::new(NoopPathExtractor), +/// 5, // TTL in turns — the only invalidation path with this extractor +/// ); +/// ``` +pub struct NoopPathExtractor; + +impl PathExtractor for NoopPathExtractor { + fn paths(&self, _tool_name: &str, _input: &Value) -> Vec { + Vec::new() + } +} + // =================================================== // Cache types (private) // =================================================== @@ -1112,4 +1142,48 @@ mod tests { r3.output ); } + + #[test] + fn noop_path_extractor_returns_empty() { + let extractor = NoopPathExtractor; + // Arbitrary tool/input combinations all yield no paths. + assert!( + extractor + .paths("Read", &serde_json::json!({"path": "/etc/hosts"})) + .is_empty() + ); + assert!( + extractor + .paths("Grep", &serde_json::json!({"pattern": "x"})) + .is_empty() + ); + assert!(extractor.paths("Write", &serde_json::json!({})).is_empty()); + } + + #[tokio::test] + async fn noop_path_extractor_disables_path_invalidation() { + let mw = make_middleware(Arc::new(NoopPathExtractor), 5); + let read_content = ToolContent::from_string("file body"); + let (pipeline, _calls) = pipeline(mw, read_content.clone(), false); + + // Step 1: Read(foo.rs) — populates the cache. + let mut ctx = ctx_for("Read", serde_json::json!({"path": "foo.rs"}), 1); + let r1 = pipeline.dispatch(&mut ctx).await; + assert!(!r1.output.to_string().contains("[cached]")); + + // Step 2: Write(foo.rs) — would normally invalidate, but the + // no-op extractor contributes no paths, so the cache survives. + let mut ctx = ctx_for("Write", serde_json::json!({"path": "foo.rs"}), 1); + let wr = pipeline.dispatch(&mut ctx).await; + assert!(!wr.is_error, "write should succeed"); + + // Step 3: Read(foo.rs) again — still cached (NOT invalidated). + let mut ctx = ctx_for("Read", serde_json::json!({"path": "foo.rs"}), 2); + let r3 = pipeline.dispatch(&mut ctx).await; + assert!( + r3.output.to_string().contains("[cached]"), + "NoopPathExtractor disables path invalidation; read still cached: {}", + r3.output + ); + } } diff --git a/src/middleware/verify.rs b/src/middleware/verify.rs index 0402a9d..84753a3 100644 --- a/src/middleware/verify.rs +++ b/src/middleware/verify.rs @@ -108,6 +108,44 @@ pub struct VerifyResult { pub diagnostics: String, } +/// A [`Verifier`] that always passes with empty diagnostics. +/// +/// Useful as the default verifier when wiring [`VerifyMiddleware`] into a +/// pipeline that has no real build/lint step to run — the middleware still +/// registers and appends a `[verify] passed: ` block, but no actual +/// check happens. Swap in a real verifier (`cargo check`, `tsc`) when one is +/// available. +/// +/// Zero-sized; cheap to construct and `Arc`-share. +/// +/// # Example +/// +/// ```rust,ignore +/// use std::sync::Arc; +/// use loopctl::middleware::{NoopVerifier, Verifier, VerifyMiddleware}; +/// +/// let pipeline = ToolPipeline::builder() +/// .with(VerifyMiddleware::new(Arc::new(NoopVerifier), vec!["Write".into()])) +/// .core(registry) +/// .build()?; +/// ``` +pub struct NoopVerifier; + +impl Verifier for NoopVerifier { + fn verify<'a>( + &'a self, + _ctx: &'a ToolContext, + _tool_name: &'a str, + ) -> Pin + Send + 'a>> { + Box::pin(async move { + VerifyResult { + passed: true, + diagnostics: String::new(), + } + }) + } +} + // =================================================== // VerifyMiddleware // =================================================== @@ -147,7 +185,7 @@ pub struct VerifyResult { pub struct VerifyMiddleware { /// The verifier this middleware invokes after write-class tool calls. /// - /// Held as a shared `Arc` handle (D2) so a single + /// Held as a shared `Arc` handle so a single /// verifier instance can back multiple middleware layers if a future /// composition wants that — e.g. one `cargo check` verifier reused /// across parallel pipelines. The handle is cloned cheaply per @@ -569,4 +607,35 @@ mod tests { assert!(debug.contains("VerifyMiddleware")); assert!(debug.contains("write_tools")); } + + #[tokio::test] + async fn noop_verifier_always_passes() { + let verifier = NoopVerifier; + let ctx = ToolContext::default(); + let result = verifier.verify(&ctx, "Write").await; + assert!(result.passed, "NoopVerifier must always pass"); + assert!( + result.diagnostics.is_empty(), + "NoopVerifier must produce empty diagnostics" + ); + } + + #[tokio::test] + async fn noop_verifier_in_verify_middleware_appends_passed_block() { + let mw = VerifyMiddleware::new(Arc::new(NoopVerifier), write_tools()); + let pipeline = pipeline_with(mw, ToolContent::from_string("wrote 42 bytes"), false); + + let mut ctx = ctx_for("Write"); + let result = pipeline.dispatch(&mut ctx).await; + + let rendered = result.output.to_string(); + assert!( + rendered.contains("[verify]"), + "NoopVerifier still triggers the middleware's append: got {rendered:?}" + ); + assert!( + rendered.contains("[verify] passed:"), + "appended block reports passed status: got {rendered:?}" + ); + } } diff --git a/src/presets.rs b/src/presets.rs new file mode 100644 index 0000000..80df88a --- /dev/null +++ b/src/presets.rs @@ -0,0 +1,414 @@ +//! Named runtime profiles for `BareLoop`. +//! +//! A [`ConstrainedProfile`] bundles the small-model reliability machinery — +//! verify-on-write, tool-call memoization, output truncation, goal +//! re-injection, and strict tool-call decoding — behind one type, so a +//! consumer gets "small-model-ready by default" without per-feature wiring. +//! [`FrontierProfile`] is the named opt-out: v0.1.0-style defaults with none +//! of the small-model machinery installed. +//! +//! Apply a profile with [`ConstrainedProfile::apply`] (pipeline + contributor) +//! or compose the individual pieces ([`ConstrainedProfile::loop_config`], +//! [`ConstrainedProfile::pipeline_builder`], +//! [`ConstrainedProfile::request_options`]) by hand. +//! +//! # Example +//! +//! ```rust,ignore +//! use loopctl::engine::BareLoop; +//! use loopctl::presets::ConstrainedProfile; +//! use loopctl::tool::ToolRegistry; +//! use std::sync::Arc; +//! +//! // `client` is any ApiClient impl; `registry` is your tool set. +//! let mut agent = BareLoop::new(client, registry, ConstrainedProfile::loop_config()); +//! agent.set_request_options(ConstrainedProfile::request_options()); +//! ConstrainedProfile::apply(&mut agent).unwrap(); +//! ``` + +use std::sync::Arc; + +use crate::config::LoopConfig; +use crate::engine::{BareLoop, ContextContributor, ContributorContext}; +use crate::error::LoopError; +use crate::message::{Message, MessagePart, Role}; +use crate::middleware::{ + MemoizingMiddleware, NoopPathExtractor, NoopVerifier, OutputLimitMiddleware, ToolPipeline, + VerifyMiddleware, +}; +use crate::structured::{RequestOptions, ToolConstraint}; + +/// Default per-tool-output cap (characters) applied by `OutputLimitMiddleware`. +const OUTPUT_CAP_CHARS: usize = 16_384; +/// Default cache TTL (turns) for the preset's `MemoizingMiddleware`. +const MEMOIZE_TTL_TURNS: u32 = 5; +/// Default reminder cadence (turns) for [`GoalReminder`]. +const GOAL_REMINDER_EVERY_N_TURNS: usize = 5; +/// Default write-class tool names the preset wires into its middleware. Advisory. +const WRITE_TOOLS: &[&str] = &["Write", "Edit", "MultiEdit"]; +/// Default memoized tool names the preset wires into its middleware. Advisory. +const MEMOIZED_TOOLS: &[&str] = &["Read", "Glob", "Grep", "LS"]; + +// =================================================== +// ConstrainedProfile +// =================================================== + +/// The small-model-tuned runtime profile. +/// +/// Bundles aggressive context budgeting (smaller window, fewer turns), +/// verify-on-write ([`VerifyMiddleware`] with [`NoopVerifier`]), +/// tool-call memoization ([`MemoizingMiddleware`] with [`NoopPathExtractor`]), +/// output truncation ([`OutputLimitMiddleware`]), goal re-injection +/// ([`GoalReminder`]), and strict tool-call decoding +/// ([`ToolConstraint::Strict`]) into one coherent profile. +/// +/// This is the harder-problem profile: it assumes the model drifts off-goal, +/// repeats tool calls, ships broken edits, and emits malformed tool +/// arguments, and installs machinery that catches each. Frontier models +/// tolerate the same defaults fine; use [`FrontierProfile`] to opt out. +/// +/// The no-op verifier and path extractor are wired by default so the profile +/// is functional out of the box — but no actual verification or path-based +/// cache invalidation happens until you swap in real impls. The verify +/// middleware still registers and the cache still works by TTL; replacing +/// [`NoopVerifier`] with `cargo check` / `tsc` and [`NoopPathExtractor`] with +/// a path-aware extractor is the intended upgrade path. +pub struct ConstrainedProfile; + +impl ConstrainedProfile { + /// A [`LoopConfig`] tuned for a small model. + /// + /// Sets a smaller context window and fewer max turns than the v0.1.0 + /// defaults — small models degrade faster as context fills, so the + /// budget is tighter. Compaction threshold is unchanged (tightness comes + /// from the window, not from compacting earlier). + /// + /// Values: `context_window = 120_000`, `max_turns = 100`, + /// `max_tokens = 16_384`, `compact_threshold = 0.80`, `auto_compact = true`. + #[must_use] + pub fn loop_config() -> LoopConfig { + LoopConfig::default() + .with_context_window(120_000) + .with_max_turns(100) + } + + /// A [`ToolPipeline`] builder pre-loaded with the small-model middleware stack. + /// + /// Middleware order (execution order, outermost last): verify → memoize + /// → output-limit. Verify's appended diagnostics flow through memoize + /// (cached alongside the result) and then through the output cap + /// (truncated if the combined output exceeds the cap). + /// + /// No `.core()` is set — pass the result to + /// [`BareLoop::set_pipeline`], which attaches the tool registry. + #[must_use] + pub fn pipeline_builder() -> crate::middleware::ToolPipelineBuilder { + ToolPipeline::builder() + .with(VerifyMiddleware::new( + Arc::new(NoopVerifier), + WRITE_TOOLS.iter().map(|s| (*s).to_string()).collect(), + )) + .with(MemoizingMiddleware::new( + MEMOIZED_TOOLS.iter().map(|s| (*s).to_string()).collect(), + WRITE_TOOLS.iter().map(|s| (*s).to_string()).collect(), + Arc::new(NoopPathExtractor), + MEMOIZE_TTL_TURNS, + )) + .with(OutputLimitMiddleware::new(OUTPUT_CAP_CHARS)) + } + + /// [`RequestOptions`] requesting strict tool-call decoding. + /// + /// Apply via [`BareLoop::set_request_options`] so the constraint reaches + /// the provider on every turn. + #[must_use] + pub fn request_options() -> RequestOptions { + RequestOptions::new().tool_constraint(ToolConstraint::Strict) + } + + /// Apply the profile's pipeline and goal-reminder contributor to a + /// [`BareLoop`]. + /// + /// Sets the small-model middleware stack (via [`Self::pipeline_builder`]) and + /// registers a [`GoalReminder`] firing every 5 turns. Does **not** set + /// the loop's config or request options — those are set separately at + /// construction (`BareLoop::new`) and via + /// [`BareLoop::set_request_options`]. + /// + /// # Errors + /// + /// Returns [`LoopError`] if [`BareLoop::set_pipeline`] fails. + /// + /// # Example + /// + /// ```rust,ignore + /// use loopctl::engine::BareLoop; + /// use loopctl::presets::ConstrainedProfile; + /// use loopctl::tool::ToolRegistry; + /// use std::sync::Arc; + /// + /// // `client` is any ApiClient impl. + /// let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), ConstrainedProfile::loop_config()); + /// ConstrainedProfile::apply(&mut agent).unwrap(); + /// ``` + pub fn apply(loop_: &mut BareLoop) -> Result<(), LoopError> { + loop_.set_pipeline(Self::pipeline_builder())?; + loop_.add_contributor(Box::new(GoalReminder::new(GOAL_REMINDER_EVERY_N_TURNS))); + Ok(()) + } +} + +// =================================================== +// FrontierProfile +// =================================================== + +/// The frontier-opt-out profile. +/// +/// Produces default [`LoopConfig`], no small-model +/// middleware, no tool-call constraint. Named (rather than just "don't use +/// [`ConstrainedProfile`]") so the opt-out is explicit and discoverable. +pub struct FrontierProfile; + +impl FrontierProfile { + /// A default [`LoopConfig`] — no small-model tightening. + /// + /// Literally [`LoopConfig::default`]: the full v0.1.0 context budget + /// (200k window, 200 turns), default compaction threshold, no tightening. + /// The opt-out counterpart to + /// [`ConstrainedProfile::loop_config`](ConstrainedProfile::loop_config); + /// the two are swap-in replacements at the call site. + #[must_use] + pub fn loop_config() -> LoopConfig { + LoopConfig::default() + } + + /// An empty middleware pipeline — no small-model middleware installed. + /// + /// Returns a bare [`ToolPipeline::builder()`] with nothing wired; pass it + /// to [`BareLoop::set_pipeline`] to get the plain v0.1.0 tool-dispatch + /// path (no verify-on-write, no memoization, no output cap). The opt-out + /// counterpart to + /// [`ConstrainedProfile::pipeline_builder`](ConstrainedProfile::pipeline_builder). + #[must_use] + pub fn pipeline_builder() -> crate::middleware::ToolPipelineBuilder { + ToolPipeline::builder() + } + + /// Default [`RequestOptions`] — no tool-call constraint. + /// + /// Literally [`RequestOptions::default`] (`tool_constraint: None`), so + /// tool calls are unconstrained. The opt-out counterpart to + /// [`ConstrainedProfile::request_options`](ConstrainedProfile::request_options) + /// (which sets `Strict`); apply via [`BareLoop::set_request_options`]. + #[must_use] + pub fn request_options() -> RequestOptions { + RequestOptions::default() + } +} + +// =================================================== +// GoalReminder +// =================================================== + +/// A [`ContextContributor`] that re-injects the first user message as a +/// [`Role::System`] reminder every `n` turns. +/// +/// Small models drift off-goal after several turns of tool calling. +/// Re-emitting the original request periodically keeps the model anchored to +/// what the user actually asked for. The reminder text is the **first** +/// [`Role::User`] message in the conversation, taken verbatim — the first +/// thing the user asked is a reasonable goal proxy and avoids parsing user +/// prose. +/// +/// Construct with [`GoalReminder::new`], or get a default-cadence one via +/// [`ConstrainedProfile::apply`]. +pub struct GoalReminder { + /// Reminder cadence in turns. The reminder fires when + /// `turn > 0 && turn % every_n_turns == 0`. + every_n_turns: usize, +} + +impl GoalReminder { + /// Create a goal reminder that fires every `every_n_turns` turns. + /// + /// `every_n_turns` of 0 or 1 disables the cadence guard; with 0 the + /// reminder never fires (turn 0 is always skipped), with 1 it fires on + /// every turn after the first. Sensible values are 3–10. + #[must_use] + pub fn new(every_n_turns: usize) -> Self { + Self { every_n_turns } + } +} + +impl ContextContributor for GoalReminder { + fn contribute(&self, ctx: &ContributorContext<'_>) -> Option { + // Skip turn 0 — the user's message is fresh context, no reminder yet. + if ctx.turn == 0 || self.every_n_turns == 0 { + return None; + } + + if !ctx.turn.is_multiple_of(self.every_n_turns) { + return None; + } + + // Find the first user message and re-emit its text verbatim. + let first_user_text = ctx.conversation.iter().find_map(|m| { + if !matches!(m.role, Role::User) { + return None; + } + let texts: Vec<&str> = m + .parts + .iter() + .filter_map(|p| match p { + MessagePart::Text { text } => Some(text.as_str()), + _ => None, + }) + .collect(); + if texts.is_empty() { + None + } else { + Some(texts.join("\n")) + } + })?; + Some(Message::new( + Role::System, + vec![MessagePart::text(first_user_text)], + )) + } +} + +#[cfg(test)] +#[allow(clippy::float_cmp)] +mod tests { + use super::*; + + #[test] + fn test_constrained_loop_config_values() { + let cfg = ConstrainedProfile::loop_config(); + assert_eq!(cfg.context_window, 120_000); + assert_eq!(cfg.max_turns, 100); + assert_eq!(cfg.max_tokens, 16_384); + assert_eq!(cfg.compact_threshold, 0.80); + assert!(cfg.auto_compact); + } + + #[test] + fn test_constrained_request_options_strict() { + let opts = ConstrainedProfile::request_options(); + assert!(matches!(opts.tool_constraint, ToolConstraint::Strict)); + assert!(opts.response_format.is_none()); + } + + #[test] + fn test_frontier_loop_config_matches_default() { + let preset = FrontierProfile::loop_config(); + let default = LoopConfig::default(); + assert_eq!(preset.context_window, default.context_window); + assert_eq!(preset.max_turns, default.max_turns); + assert_eq!(preset.max_tokens, default.max_tokens); + assert_eq!(preset.compact_threshold, default.compact_threshold); + assert_eq!(preset.auto_compact, default.auto_compact); + } + + #[test] + fn test_frontier_request_options_default() { + let opts = FrontierProfile::request_options(); + assert!(matches!(opts.tool_constraint, ToolConstraint::None)); + assert!(opts.response_format.is_none()); + } + + fn ctx_at(turn: usize, conv: &[Message]) -> ContributorContext<'_> { + ContributorContext::new(turn, conv) + } + + #[test] + fn test_goal_reminder_skips_turn_zero() { + let reminder = GoalReminder::new(5); + let conv = [Message::user("ship the demo")]; + let ctx = ctx_at(0, &conv); + assert!( + reminder.contribute(&ctx).is_none(), + "turn 0 must be skipped" + ); + } + + #[test] + fn test_goal_reminder_never_fires_when_n_is_zero() { + let reminder = GoalReminder::new(0); + let conv = [Message::user("ship the demo")]; + // Even on a multiple-of-0-ish turn, n=0 disables firing. + let ctx = ctx_at(5, &conv); + assert!( + reminder.contribute(&ctx).is_none(), + "n=0 disables the reminder" + ); + } + + #[test] + fn test_goal_reminder_fires_every_n_turns() { + let reminder = GoalReminder::new(5); + let conv = [Message::user("goal")]; + for turn in [1, 2, 3, 4, 6, 7, 8, 9, 11] { + let ctx = ctx_at(turn, &conv); + assert!( + reminder.contribute(&ctx).is_none(), + "turn {turn} must not fire (cadence 5)" + ); + } + for turn in [5, 10, 15, 20] { + let ctx = ctx_at(turn, &conv); + assert!( + reminder.contribute(&ctx).is_some(), + "turn {turn} must fire (cadence 5)" + ); + } + } + + #[test] + fn test_goal_reminder_injects_first_user_message_verbatim() { + let reminder = GoalReminder::new(5); + let conv = [ + Message::assistant("hi"), // not a user message + Message::user("ship the demo"), // first user message + Message::user("a later unrelated request"), // must NOT be picked + ]; + let ctx = ctx_at(5, &conv); + let msg = reminder.contribute(&ctx).expect("turn 5 fires"); + assert_eq!(msg.role, Role::System); + // The reminder text is the FIRST user message, verbatim. + match &msg.parts[0] { + MessagePart::Text { text } => assert_eq!(text, "ship the demo"), + other => panic!("expected Text part, got {other:?}"), + } + } + + #[test] + fn test_goal_reminder_returns_none_with_no_user_message() { + let reminder = GoalReminder::new(5); + let conv = [Message::assistant("no user here")]; + let ctx = ctx_at(5, &conv); + assert!( + reminder.contribute(&ctx).is_none(), + "no user message → no reminder, even on a firing turn" + ); + } + + #[test] + fn test_goal_reminder_handles_multi_part_first_user_message() { + // A first user message with multiple text parts: they are joined + // with newlines into a single reminder. + let reminder = GoalReminder::new(1); + let conv = [Message::new( + Role::User, + vec![MessagePart::text("line one"), MessagePart::text("line two")], + )]; + let ctx = ctx_at(1, &conv); + let msg = reminder.contribute(&ctx).expect("turn 1 fires (cadence 1)"); + match &msg.parts[0] { + MessagePart::Text { text } => { + assert_eq!(text, "line one\nline two"); + } + other => panic!("expected Text part, got {other:?}"), + } + } +} diff --git a/src/stream/handler.rs b/src/stream/handler.rs index 6168468..f810a31 100644 --- a/src/stream/handler.rs +++ b/src/stream/handler.rs @@ -359,8 +359,12 @@ pub struct RateLimitConfig { /// [`default_delay`](Self::default_delay) is always used. pub respect_retry_after: bool, - /// Backoff used when the server gives no `Retry-After`, or when - /// [`respect_retry_after`](Self::respect_retry_after) is `false`. + /// Backoff used when the server gives no `Retry-After`. + /// + /// Also used unconditionally when + /// [`respect_retry_after`](Self::respect_retry_after) is `false`. Defaults + /// to 5s — long enough to let a transient burst clear, short enough that a + /// missing header doesn't stall the agent. pub default_delay: Duration, /// Upper bound on any single rate-limit backoff. @@ -390,9 +394,10 @@ pub struct RateLimitConfig { /// Hard cap on rate-limit retries for a single turn. /// - /// Once this many retries have been exhausted, the turn fails. Distinct - /// from [`fallback_after_retries`](Self::fallback_after_retries), which - /// controls the *escalation* threshold, not the hard stop. + /// Once this many retries have been exhausted, the turn fails outright. + /// Distinct from [`fallback_after_retries`](Self::fallback_after_retries), + /// which controls the *escalation* threshold to a fallback model, not the + /// hard stop after which the turn gives up entirely. pub max_retries: u32, } @@ -464,12 +469,27 @@ impl RateLimitConfig { } /// Which kind of rate-limit / overload response was detected. +/// +/// Set by [`DetectedRateLimit::detect`] when classifying an [`ApiError`]; the +/// distinction matters because the two responses come from different failure +/// modes (a hard per-account quota vs. a transient capacity signal) even +/// though both honour `Retry-After`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum RateLimitKind { /// HTTP 429 Too Many Requests. + /// + /// The canonical rate-limit signal: the caller has exceeded a per-account + /// or per-key quota. The server typically sends a `Retry-After` hint; + /// [`backoff`](RateLimitConfig::backoff) honours it when + /// [`respect_retry_after`](RateLimitConfig::respect_retry_after) is set. RateLimited, - /// HTTP 503 Service Unavailable / 529 Overloaded — a rate-limit-adjacent - /// transient that honours `Retry-After`. + + /// HTTP 503 Service Unavailable / 529 Overloaded. + /// + /// A rate-limit-adjacent transient: the provider is overloaded rather than + /// enforcing a quota. Treated the same as [`RateLimited`](Self::RateLimited) + /// for backoff purposes (it honours `Retry-After`), but surfaced as a + /// distinct kind so callers can log or route it differently. Overloaded, } @@ -481,10 +501,26 @@ pub enum RateLimitKind { #[derive(Debug, Clone)] pub struct DetectedRateLimit { /// The detected rate-limit class. + /// + /// Either [`RateLimitKind::RateLimited`] (HTTP 429) or + /// [`RateLimitKind::Overloaded`] (HTTP 503/529). Determines nothing on its + /// own — both kinds honour `Retry-After` — but lets the caller distinguish + /// a quota hit from a transient capacity signal. pub kind: RateLimitKind, - /// The server-advised delay, `None` if absent or malformed. + + /// The server-advised delay, parsed from the `Retry-After` header. + /// + /// `None` when the header was absent or could not be parsed as a number of + /// seconds or an HTTP-date. When `None`, the caller falls back to + /// [`RateLimitConfig::default_delay`]. pub retry_after: Option, - /// The original error message, preserved for logging / fallback context. + + /// The original error message, preserved verbatim. + /// + /// Kept so the caller can log the provider's wording, include it in a + /// fallback-model prompt, or surface it to the user without losing the + /// diagnostic detail that [`detect`](Self::detect) collapsed into + /// [`kind`](Self::kind). pub message: String, } @@ -580,23 +616,77 @@ fn clamp_delay_to_deadline(delay: Duration, deadline: Option) -> Durati delay.min(remaining) } -/// Outcome of a rate-limit retry decision (see [`StreamHandler::rate_limit_retry`]). +/// Outcome of a rate-limit retry decision. +/// +/// Returned by [`StreamHandler::rate_limit_retry`] for each detected rate +/// limit on the current model. The variants form a three-step escalation +/// ladder: retry in place while the count is low, escalate to the model +/// circuit breaker once it crosses +/// [`fallback_after_retries`](RateLimitConfig::fallback_after_retries), and +/// give up entirely once it crosses +/// [`max_retries`](RateLimitConfig::max_retries). #[derive(Debug)] enum RateLimitRetry { /// Escalate to the model circuit breaker. + /// + /// Returned once the per-model retry count exceeds + /// [`fallback_after_retries`](RateLimitConfig::fallback_after_retries). + /// The caller trips the breaker, which routes subsequent turns to a + /// fallback model if one is configured; if not, the escalation has nowhere + /// to go and the turn fails. Escalate { - /// Number of rate-limit retries honored before escalating. + /// Number of rate-limit retries honored before this escalation. + /// + /// Always strictly greater than + /// [`fallback_after_retries`](RateLimitConfig::fallback_after_retries) + /// — the count that triggered the escalation, incremented before the + /// decision is made. Surfaced for logging and for the + /// [`RateLimitEscalation`](crate::error::LoopError::RateLimitEscalation) + /// error payload. attempts: u32, - /// Last server-advised hint, after clamping. + + /// The server-advised delay from the triggering response. + /// + /// The raw `Retry-After` from [`DetectedRateLimit`] (`None` if the + /// header was absent). Carried unmodified — clamping to + /// [`max_delay`](RateLimitConfig::max_delay) happens in + /// [`backoff`](RateLimitConfig::backoff) on the retry path, not here. + /// Preserved so the escalation consumer can log or forward the + /// provider's hint. retry_after: Option, }, + /// Give up on the current model without escalating. + /// + /// Returned once the per-model retry count exceeds + /// [`max_retries`](RateLimitConfig::max_retries) — the hard stop after + /// which retrying the same model is pointless. Distinct from + /// [`Escalate`](Self::Escalate): escalation hands off to the circuit + /// breaker (and a fallback model), `HardStop` fails the turn outright. HardStop, - /// Sleep `delay` then retry the current model. + + /// Sleep for `delay`, then retry the current model. + /// + /// Returned while the retry count is below both + /// [`fallback_after_retries`](RateLimitConfig::fallback_after_retries) and + /// [`max_retries`](RateLimitConfig::max_retries). The delay is the + /// [`backoff`](RateLimitConfig::backoff) for the detected response, + /// further clamped to the remaining time before the turn's + /// `total_stream_timeout` deadline so a large `Retry-After` cannot overrun + /// the turn budget. Retry(Duration), } /// Pull the carried [`StreamOutcome`] out of a [`StreamHandlerError`], if any. +/// +/// Only [`InitFailed`](StreamHandlerError::InitFailed) and +/// [`StreamFailed`](StreamHandlerError::StreamFailed) carry one (the outcome +/// that was in progress when the error was raised); every other variant maps +/// to `None`. The caller uses the recovered outcome to route the failure into +/// the correct retry budget — a [`RateLimited`](StreamOutcome::RateLimited) +/// outcome draws on [`RateLimitConfig`], distinct from the transport-retry +/// budget, so a rate-limit storm cannot exhaust transport retries (nor vice +/// versa). fn carried_outcome(error: &StreamHandlerError) -> Option { match error { StreamHandlerError::InitFailed(o) | StreamHandlerError::StreamFailed(o) => { @@ -624,10 +714,29 @@ async fn sleep_cancellable( } /// Result of polling the stream once inside [`StreamHandler::next_event`]. +/// +/// Produced by the `tokio::select!` that races the stream against the +/// per-event timeout. Only two outcomes materialize here: the stream produced +/// an item ([`Next`](Self::Next)), or the per-event timeout fired first +/// ([`TimedOut`](Self::TimedOut)). Cancellation and the total-stream deadline +/// are also raced in the same `select!`, but they return directly as +/// [`StreamHandlerError::Cancelled`] / `StreamFailed` and so do not need a +/// variant here. enum EventPoll { - /// The stream produced an item (`None` means the stream ended). + /// The stream produced an item before the per-event timeout. + /// + /// Delegates the three sub-cases to the caller: `Some(Ok(event))` is + /// yielded to the accumulator, `Some(Err(api_error))` becomes an API-error + /// outcome, and `None` means the stream ended cleanly (turn completes). Next(Option>), + /// The per-event timeout fired before any item arrived. + /// + /// Increments the consecutive-timeout counter; once it reaches + /// [`max_consecutive_timeouts`](StreamTimeoutConfig::max_consecutive_timeouts), + /// the caller escalates to a [`StreamFailed`](StreamHandlerError::StreamFailed) + /// event-timeout outcome. A lower threshold (`min(2, max_consecutive_timeouts)`) + /// applies when no events have been received yet (empty-stream fast-fail). TimedOut, } @@ -638,15 +747,43 @@ enum EventPoll { /// populate [`StreamOutcome`] fields when it short-circuits. struct EventDiagnostics { /// Events processed so far this turn. + /// + /// Surfaced on the [`StreamOutcome::TotalTimeout`] and + /// [`StreamOutcome::RateLimited`] outcomes so the caller can tell a + /// mid-stream failure (some events got through) from an immediate one + /// (nothing arrived). Not used by [`event_timeout`](Self::event_timeout), + /// which reports consecutive-timeout count instead. events_processed: u64, + /// When the stream started, for elapsed-duration outcomes. + /// + /// Read via [`Instant::elapsed`] when building + /// [`StreamOutcome::TotalTimeout`]'s `duration` field. Captured once at + /// the top of [`process_events`](StreamHandler::process_events) rather + /// than per event so the reported duration is the full stream lifetime, + /// not the time since the most recent event. stream_start: Instant, + /// Whether partial content has been accumulated. + /// + /// Recomputed each loop iteration (the whole [`EventDiagnostics`] is + /// rebuilt per iteration in [`process_events`](StreamHandler::process_events)) + /// from the accumulator's current part count: `true` once at least one + /// usable event has been received. Flows into the `has_partial_data` flag + /// on [`StreamOutcome::TotalTimeout`], [`StreamOutcome::EventTimeout`], + /// and [`StreamOutcome::RateLimited`], letting a downstream consumer + /// decide whether to salvage the partial output or discard it. has_partial_data: bool, } impl EventDiagnostics { /// Build the [`StreamOutcome::TotalTimeout`] for this point in the stream. + /// + /// Snapshots the current diagnostic state — partial-data flag, events + /// processed so far, and elapsed time since [`stream_start`](Self::stream_start) + /// — into a `TotalTimeout` outcome. Used by [`process_events`](StreamHandler::process_events) + /// when the turn's `total_stream_timeout` deadline fires (both at the + /// top-of-loop check and inside the per-event `select!`). fn total_timeout(&self) -> StreamOutcome { StreamOutcome::TotalTimeout { has_partial_data: self.has_partial_data, @@ -656,6 +793,12 @@ impl EventDiagnostics { } /// Build the [`StreamOutcome::EventTimeout`] for this point in the stream. + /// + /// Carries the partial-data flag plus the caller-supplied + /// `consecutive_timeouts` count (this method does not track the counter + /// itself — `process_events` owns it and passes the current value in). + /// Used once the per-event timeout crosses + /// [`max_consecutive_timeouts`](StreamTimeoutConfig::max_consecutive_timeouts). fn event_timeout(&self, consecutive_timeouts: u32) -> StreamOutcome { StreamOutcome::EventTimeout { has_partial_data: self.has_partial_data, @@ -663,8 +806,15 @@ impl EventDiagnostics { } } - /// Map a stream API error to the matching outcome: rate-limit when detected, - /// otherwise a generic init failure. + /// Map a stream API error to the matching [`StreamHandlerError`]. + /// + /// Two branches: if [`DetectedRateLimit::detect`] classifies the error as + /// a 429/503/529, builds a [`StreamOutcome::RateLimited`] carrying the + /// parsed `Retry-After` and current progress; otherwise wraps it as a + /// generic [`StreamOutcome::InitFailed`] with `attempts: 1` (this is the + /// per-event error path, not the init-retry path, so the attempt counter + /// isn't meaningful here). Used by `process_events` when the stream yields + /// an `Err` event. fn api_error_outcome(&self, error: &crate::api::error::ApiError) -> StreamHandlerError { if let Some(detail) = DetectedRateLimit::detect(error) { return StreamHandlerError::StreamFailed(StreamOutcome::RateLimited { @@ -947,11 +1097,14 @@ impl std::error::Error for StreamHandlerError {} // StreamProgress // =================================================== -/// Progress data emitted during long-running streams. +/// A snapshot of stream progress for external reporting. /// -/// Passed to the progress callback at regular intervals -/// ([`progress_interval`](StreamTimeoutConfig::progress_interval)) -/// to report stream health. +/// Plain data struct carrying the two progress signals a consumer is likely +/// to want (elapsed time and events processed). `StreamHandler` does not +/// itself emit `StreamProgress` — it has no built-in progress callback. The +/// struct is shipped so a downstream consumer that drives its own progress +/// reporting (metrics observer, TUI heartbeat, deadline watcher) has a +/// shared shape to read or fill. /// /// # Example /// @@ -968,8 +1121,17 @@ impl std::error::Error for StreamHandlerError {} #[derive(Debug, Clone)] pub struct StreamProgress { /// Time elapsed since the stream started. + /// + /// Wall-clock duration from stream open to the snapshot point. Useful for + /// heartbeat-style reporting (“still streaming after Ns”) and for + /// deadline-aware consumers that compare it against their own budget. pub elapsed: Duration, + /// Number of SSE events processed so far. + /// + /// Count of stream events successfully accumulated up to the snapshot + /// point. A flat or slow-growing count is the early signal of a stalled + /// stream before a timeout fires. pub events_processed: u64, } @@ -979,17 +1141,35 @@ pub struct StreamProgress { /// Holds configuration for the streaming resilience layer. /// -/// `StreamHandler` stores [`StreamTimeoutConfig`] and [`StreamRetryConfig`] -/// for the three-phase streaming lifecycle: +/// `StreamHandler` wraps an [`ApiClient`]'s streaming path with timeout, +/// retry, and rate-limit handling. It owns four independent budgets that +/// together make a stream turn robust: +/// +/// 1. **Timeouts** ([`StreamTimeoutConfig`]) — initial-event, per-event, and +/// total-stream deadlines, plus the consecutive-timeout escalation +/// threshold and the optional non-streaming fallback. /// -/// 1. **Initialize** — Opens a stream, waits for the first event with a -/// timeout, retries with exponential backoff on failure. +/// 2. **Transport retries** ([`StreamRetryConfig`]) — exponential backoff for +/// stream-initialization failures (connection drops, transport errors), +/// distinct from rate-limit retries. /// -/// 2. **Process** — Consumes events with per-event and total timeouts. -/// Emits progress callbacks at regular intervals. +/// 3. **Rate-limit handling** ([`RateLimitConfig`]) — `Retry-After`-aware +/// backoff for 429/503/529 responses, with its own retry budget, an +/// escalation threshold to the model circuit breaker, and a hard stop. +/// Kept independent from transport retries so a rate-limit storm cannot +/// exhaust the transport budget (nor vice versa). /// -/// 3. **Recover** — If streaming fails, falls back to -/// [`ApiClient::create_message`] for a non-streaming response. +/// 4. **Proactive throttling** (optional +/// [`RateLimiter`](crate::stream::rate_limit::RateLimiter)) — a per-provider +/// token bucket that gates each stream attempt *before* it fires, sleeping +/// up to `max_wait` rather than risking a 429. +/// +/// On exhaustion, the handler escalates: rate-limit retries trip the model +/// circuit breaker (route to a fallback model), transport retries fall back +/// to [`ApiClient::create_message`] when +/// [`fallback_to_non_streaming`](StreamTimeoutConfig::fallback_to_non_streaming) +/// is set, and a turn that can't recover fails with a typed +/// [`StreamHandlerError`]. /// /// # Example /// @@ -1010,13 +1190,46 @@ pub struct StreamProgress { /// ``` pub struct StreamHandler { /// Timeout configuration for all phases. + /// + /// Drives the initial-event / per-event / total-stream deadlines, the + /// consecutive-timeout escalation threshold, and the + /// non-streaming-fallback toggle. Read on every turn in `stream_turn` and + /// on each event poll. timeout_config: StreamTimeoutConfig, - /// Retry configuration for stream initialization. + + /// Retry configuration for stream-initialization failures. + /// + /// Exponential backoff applied to transport-level failures (connection + /// drops, TLS errors, etc.) before any event arrives. Read in the + /// init-retry loop; distinct from `rate_limit_config`, which has its own + /// budget. retry_config: StreamRetryConfig, + /// Rate-limit detection + backoff policy. + /// + /// Governs reactive handling of server-returned 429/503/529 responses + /// (honoured `Retry-After`, default delay, cap, escalation threshold to + /// the model circuit breaker, hard-stop ceiling). Read by the + /// `rate_limit_retry` decision on each detected rate limit. rate_limit_config: RateLimitConfig, + /// Optional proactive per-provider rate limiter (token bucket). + /// + /// When set, each stream attempt is gated by `gate_on_rate_limit` *before* + /// firing — sleeping up to the limiter's `max_wait` for a token rather + /// than risking a 429. `None` (the default) means reactive-only handling: + /// no pre-throttling, server 429s still handled via `rate_limit_config`. rate_limiter: Option>, + + /// Per-turn request options applied to every stream-open call. + /// + /// Carries [`tool_constraint`](crate::structured::ToolConstraint) for + /// constrained tool-call decoding. Default is + /// [`RequestOptions::default`](crate::structured::RequestOptions::default) + /// (no constraint), reproducing the prior unconstrained behavior. + /// Passed to [`ApiClient::stream_messages_with_options`] at the stream-open + /// call site. Set via [`with_request_options`](StreamHandler::with_request_options). + request_options: crate::structured::RequestOptions, } impl fmt::Debug for StreamHandler { @@ -1026,6 +1239,7 @@ impl fmt::Debug for StreamHandler { .field("retry_config", &self.retry_config) .field("rate_limit_config", &self.rate_limit_config) .field("rate_limiter", &self.rate_limiter) + .field("request_options", &self.request_options) .finish() } } @@ -1059,6 +1273,7 @@ impl StreamHandler { retry_config: StreamRetryConfig::default(), rate_limit_config: RateLimitConfig::default(), rate_limiter: None, + request_options: crate::structured::RequestOptions::default(), } } @@ -1088,13 +1303,35 @@ impl StreamHandler { self } + /// Set the per-turn [`RequestOptions`](crate::structured::RequestOptions) + /// applied to every stream-open call. Consuming builder. + /// + /// Default is [`RequestOptions::default`](crate::structured::RequestOptions::default) + /// (no constraint), which reproduces prior behavior. Set + /// [`tool_constraint`](crate::structured::ToolConstraint) to + /// [`Strict`](crate::structured::ToolConstraint::Strict) for constrained + /// tool-call decoding. + #[must_use] + pub fn with_request_options(mut self, options: crate::structured::RequestOptions) -> Self { + self.request_options = options; + self + } + /// Returns a reference to the timeout configuration. + /// + /// Read-only access to the [`StreamTimeoutConfig`] stored on the handler. + /// Mutate via [`with_config`](StreamHandler::with_config) (which replaces + /// both timeout and retry together); there is no per-field setter. #[must_use] pub fn timeout_config(&self) -> &StreamTimeoutConfig { &self.timeout_config } /// Returns a reference to the retry configuration. + /// + /// Read-only access to the [`StreamRetryConfig`] stored on the handler. + /// Mutate via [`with_config`](StreamHandler::with_config) (which replaces + /// both retry and timeout together); there is no per-field setter. #[must_use] pub fn retry_config(&self) -> &StreamRetryConfig { &self.retry_config @@ -1120,6 +1357,10 @@ impl StreamHandler { } /// Returns a reference to the rate-limit configuration. + /// + /// Read-only access to the [`RateLimitConfig`] stored on the handler. + /// Mutate via + /// [`with_rate_limit_config`](StreamHandler::with_rate_limit_config). #[must_use] pub fn rate_limit_config(&self) -> &RateLimitConfig { &self.rate_limit_config @@ -1160,11 +1401,17 @@ impl StreamHandler { /// Primary entry point for resilient streaming. It /// orchestrates the full lifecycle: /// - /// 1. Opens a stream via [`ApiClient::stream_messages`]. + /// 1. Opens a stream via [`ApiClient::stream_messages_with_options`] + /// (gated by the proactive [`RateLimiter`](crate::stream::rate_limit::RateLimiter), + /// if attached). /// 2. Processes events with per-event and total timeouts. - /// 3. On transient errors, retries with exponential backoff. - /// 4. If all retries fail and `fallback_to_non_streaming` is enabled, - /// falls back to [`ApiClient::create_message`]. + /// 3. On transient transport errors, retries with exponential backoff; + /// on 429/503/529 responses, backs off under the rate-limit budget + /// and escalates to the model circuit breaker once that budget is + /// exhausted. + /// 4. If all retries fail and + /// [`fallback_to_non_streaming`](StreamTimeoutConfig::fallback_to_non_streaming) + /// is enabled, falls back to [`ApiClient::create_message`]. /// /// The returned [`StreamTurnResult`] carries the accumulated message, /// token usage, stop reason, and timing regardless of which path @@ -1326,8 +1573,13 @@ impl StreamHandler { /// Attempt a single streaming pass. /// - /// Opens a stream via [`ApiClient::stream_messages`] and processes - /// all events with timeout and cancellation support. + /// Gates on the proactive [`RateLimiter`](crate::stream::rate_limit::RateLimiter) + /// (if attached), opens a stream via + /// [`ApiClient::stream_messages_with_options`] carrying the handler's + /// [`RequestOptions`](crate::structured::RequestOptions), and processes + /// all events with timeout and cancellation support. Called inside the + /// retry loop in [`stream_turn`](StreamHandler::stream_turn), so a retried + /// 429 re-gates on the rate limiter. /// /// # Errors /// @@ -1343,7 +1595,12 @@ impl StreamHandler { ) -> Result { self.gate_on_rate_limit(client, cancel, total_deadline) .await?; - let stream = client.stream_messages(conversation, system, tool_schemas); + let stream = client.stream_messages_with_options( + conversation, + system, + tool_schemas, + self.request_options.clone(), + ); self.process_events(stream, cancel, total_deadline).await } @@ -1553,6 +1810,18 @@ impl StreamHandler { } /// Whether the total-stream deadline has already passed. + /// + /// Polled between events at the top of [`next_event`](Self::next_event)'s + /// loop, before the per-event `select!` commits to another wait. This + /// catches a deadline that elapsed while the loop was processing the + /// previous event (or building diagnostics) — the + /// [`total_deadline_future`](Self::total_deadline_future) `select!` arm + /// only fires *during* a wait, so without this check a long event handler + /// could overshoot the deadline by up to one event's processing time. + /// + /// `None` means no total-stream deadline is configured (the turn is + /// bounded only by the per-event timeout) and the function returns + /// `false` for every poll. fn deadline_exceeded(total_deadline: Option) -> bool { match total_deadline { Some(deadline) => Instant::now() >= deadline, @@ -1692,14 +1961,46 @@ impl StreamHandler { #[non_exhaustive] pub struct StreamTurnResult { /// The fully accumulated assistant message. + /// + /// On the streaming path, built by the accumulator from every received + /// event — so it carries the full part structure (text blocks, tool + /// calls, images). On the non-streaming fallback path, built from just + /// the first text part of the JSON response, so non-text parts are lost + /// when [`from_fallback`](Self::from_fallback) is `true`. pub message: Message, + /// Token counts for this turn, if reported by the provider. + /// + /// Populated on the streaming path from the provider's final usage event. + /// `None` on the non-streaming fallback path (the JSON extraction does + /// not parse usage), and `None` on either path when the provider did not + /// report usage. pub usage: Option, + /// Why the model stopped generating. + /// + /// [`EndTurn`](StreamStopReason::EndTurn), + /// [`ToolCall`](StreamStopReason::ToolCall), etc. — parsed from the + /// provider's stop signal. On the non-streaming fallback path, defaults to + /// [`EndTurn`](StreamStopReason::EndTurn) when the response carries no + /// parseable stop reason. pub stop_reason: StreamStopReason, + /// Whether the result came from a non-streaming fallback. + /// + /// `false` on the normal streaming path; `true` when streaming exhausted + /// its retries and [`fallback_to_non_streaming`](StreamTimeoutConfig::fallback_to_non_streaming) + /// routed the turn through [`ApiClient::create_message`]. Callers can use + /// this to downgrade trust in the result (the fallback path loses non-text + /// parts and usage) or surface a warning to the user. pub from_fallback: bool, + /// Wall-clock time spent on this turn. + /// + /// Measured from the turn's stream-open (or fallback dispatch) to the + /// completed result, inclusive of any retries and backoff sleeps along + /// the way — so it reflects the *true* latency the turn incurred, not + /// just the final successful attempt's duration. pub elapsed: Duration, }