diff --git a/CHANGELOG.md b/CHANGELOG.md index cfc2981..849eba9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ## [Unreleased] +### Fixed + +- Restored the no-`#[allow(clippy::*)]` lint contract. Fixed: a private `TextStreamer` type alias, lossless integer-to-float casts (centralized in an internal `numeric` module), `PartLane`/`TerminalStage` lane enums replacing bool fields, and stale-allow deletions. No public API change. + ## [0.2.0] - 2026-08-02 ### Added diff --git a/src/compact/types.rs b/src/compact/types.rs index 6a02b27..d034912 100644 --- a/src/compact/types.rs +++ b/src/compact/types.rs @@ -378,8 +378,7 @@ impl ContextOverflow { if self.context_window == 0 { return f64::INFINITY; } - f64::from(u32::try_from(self.tokens_used).unwrap_or(u32::MAX)) - / f64::from(u32::try_from(self.context_window).unwrap_or(u32::MAX)) + crate::numeric::unit_ratio(self.tokens_used, self.context_window) } } diff --git a/src/detection/convergence.rs b/src/detection/convergence.rs index 749995a..338d30d 100644 --- a/src/detection/convergence.rs +++ b/src/detection/convergence.rs @@ -796,7 +796,6 @@ impl ConvergenceDetector { /// intersection size to union size. /// /// Returns `0.0` if either input is empty. - #[allow(clippy::cast_precision_loss)] #[must_use] pub fn compute_similarity(a: &str, b: &str) -> f32 { if a.is_empty() || b.is_empty() { @@ -817,11 +816,7 @@ impl ConvergenceDetector { let intersection = a_words.intersection(&b_words).count(); let union = a_words.union(&b_words).count(); - if union == 0 { - return 0.0; - } - - intersection as f32 / union as f32 + crate::numeric::unit_ratio(intersection, union) } fn normalize_text(text: &str) -> String { @@ -855,6 +850,14 @@ impl Default for ConvergenceDetector { mod tests { use super::*; + #[test] + fn compute_similarity_result_unchanged_after_cast_fix() { + let identical = ConvergenceDetector::compute_similarity("abc", "abc"); + let disjoint = ConvergenceDetector::compute_similarity("abc", "xyz"); + assert!((identical - 1.0).abs() < f32::EPSILON); + assert!(disjoint.abs() < f32::EPSILON); + } + #[test] fn test_convergence_detection() { let config = ConvergenceConfig { diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 6a3d51d..1767ca2 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -106,6 +106,18 @@ mod message; #[cfg(feature = "streaming")] mod stream; +/// Shared callback invoked once per text delta during streaming. +/// +/// A clonable, thread-safe closure stored in [`BareLoop`] via +/// [`set_text_streamer`](BareLoop::set_text_streamer) and invoked from the +/// streaming engine path on every [`IndexedDelta`](crate::stream::IndexedDelta) +/// whose payload is [`Text`](crate::stream::DeltaPart::Text). The bounds +/// mirror the requirements of that path: `Send + Sync` because the engine may +/// dispatch deltas from an async task, and `Arc` so the same callback can be +/// shared across the engine and any observer without copying the closure. +#[cfg(feature = "streaming")] +type TextStreamer = Arc; + /// How the engine fulfils each LLM turn. /// /// `BareLoop` drives every turn by asking the [`ApiClient`] for a response @@ -293,8 +305,7 @@ pub struct BareLoop { /// 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>, + text_streamer: Option, /// Turn-boundary context contributors. /// @@ -971,7 +982,7 @@ impl BareLoop { /// buf.lock().unwrap_or_else(|e| e.into_inner()).push_str(delta); /// })); /// ``` - pub fn set_text_streamer(&mut self, f: Arc) { + pub fn set_text_streamer(&mut self, f: TextStreamer) { self.debug_assert_idle(); self.text_streamer = Some(f); } @@ -1173,7 +1184,7 @@ impl BareLoop { /// mirror of [`set_text_streamer`](BareLoop::set_text_streamer). #[cfg(feature = "streaming")] #[must_use] - pub fn with_text_streamer(mut self, f: Arc) -> Self { + pub fn with_text_streamer(mut self, f: TextStreamer) -> Self { self.set_text_streamer(f); self } @@ -2147,6 +2158,15 @@ mod tests { use std::sync::Mutex; + #[cfg(feature = "streaming")] + #[test] + fn text_streamer_alias_compiles_unchanged() { + let client = MockClient::new("test-model"); + let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); + agent.set_text_streamer(Arc::new(|_| ())); + assert!(agent.text_streamer.is_some()); + } + /// Fold queued [`StreamEvent`]s into a [`NonStreamingResponse`]. /// /// Shared by `MockClient` and `RecordingClient` `create_message` impls so diff --git a/src/engine/bare/dispatch.rs b/src/engine/bare/dispatch.rs index 8932d85..5e73e6e 100644 --- a/src/engine/bare/dispatch.rs +++ b/src/engine/bare/dispatch.rs @@ -899,7 +899,6 @@ impl BareLoop { } #[cfg(all(test, feature = "testing"))] -#[allow(clippy::unnecessary_literal_bound)] mod tests { use crate::api::error::ApiError; use crate::config::SessionConfig; @@ -990,10 +989,10 @@ mod tests { struct PanicTool; impl Tool for PanicTool { - fn name(&self) -> &str { + fn name(&self) -> &'static str { "panic_tool" } - fn description(&self) -> &str { + fn description(&self) -> &'static str { "Panics on call" } fn schema(&self) -> ToolSchema { diff --git a/src/lib.rs b/src/lib.rs index e682758..35b6a11 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -78,6 +78,7 @@ pub mod managers; pub mod memory; pub mod message; pub mod middleware; +pub(crate) mod numeric; pub mod observer; pub mod presets; #[cfg(feature = "providers")] diff --git a/src/memory/builtin.rs b/src/memory/builtin.rs index f6f29f1..521b694 100644 --- a/src/memory/builtin.rs +++ b/src/memory/builtin.rs @@ -254,9 +254,9 @@ impl LoopMemory for InMemoryStore { .filter(|w| memory_lower.contains(*w)) .count(); let base_score = entry.relevance; - #[allow(clippy::cast_precision_loss)] + let denom = query_words.len().max(1); let query_bonus = if word_matches > 0 { - word_matches as f32 / query_words.len().max(1) as f32 + crate::numeric::unit_ratio(word_matches, denom) } else { 0.0 }; @@ -508,6 +508,26 @@ mod tests { assert!((results[2].relevance - 0.1).abs() < 1e-6); } + #[tokio::test] + async fn retrieve_ranks_more_word_matches_higher_at_equal_relevance() { + let store = InMemoryStore::new(); + + let mut one_match = MemoryEntry::new(MemoryCategory::Fact, "alpha only"); + one_match.relevance = 0.5; + let mut many_matches = MemoryEntry::new(MemoryCategory::Fact, "alpha beta gamma"); + many_matches.relevance = 0.5; + + store.store(one_match).await.unwrap(); + store.store(many_matches).await.unwrap(); + + let results = store.retrieve("alpha beta gamma", 2).await.unwrap(); + assert_eq!(results.len(), 2); + assert!( + results[0].memory.contains("alpha beta gamma"), + "the entry matching more query words must rank higher at equal relevance" + ); + } + #[tokio::test] async fn loop_memory_round_trips() { use crate::memory::{LoopMemory, MemoryCategory, MemoryEntry}; diff --git a/src/middleware/tool_call.rs b/src/middleware/tool_call.rs index 3c22eb1..8ce88bd 100644 --- a/src/middleware/tool_call.rs +++ b/src/middleware/tool_call.rs @@ -106,7 +106,6 @@ impl ToolCallMiddleware { } #[cfg(test)] -#[allow(clippy::unnecessary_literal_bound)] mod tests { use super::*; use crate::cancel::CancelSignal; @@ -130,10 +129,10 @@ mod tests { } impl Tool for PanickingTool { - fn name(&self) -> &str { + fn name(&self) -> &'static str { "panic_tool" } - fn description(&self) -> &str { + fn description(&self) -> &'static str { "A tool that panics" } fn schema(&self) -> ToolSchema { diff --git a/src/numeric.rs b/src/numeric.rs new file mode 100644 index 0000000..8e0552a --- /dev/null +++ b/src/numeric.rs @@ -0,0 +1,398 @@ +//! Ratios of integer counts as floats, computed without tripping +//! `cast_precision_loss`. +//! +//! Clippy's `cast_precision_loss` lint — denied via the crate's pedantic +//! config — rejects direct `usize as f32`, `usize as f64`, and `f64 as f32` +//! casts, because a float mantissa cannot hold a full 64-bit integer. The only +//! fully lint-clean paths from a count to a float go through an integer that +//! fits the mantissa: `u32` for `f64` (lossless), `u16` for `f32` (lossless). +//! +//! This module wraps those paths behind a single concept — the ratio of two +//! counts — so the narrowing, shared scaling, and zero-denominator handling +//! live in one place instead of being inlined (and silently drifted) per call +//! site. The generic [`unit_ratio`] is the entry point; pass two integer counts +//! of the same type and Rust infers the family (`u64` → `f64`, `usize` → `f32`) +//! from the argument types via the [`RatioInt`] implementations. + +/// Integer family usable as the operands of a [`unit_ratio`] computation. +/// +/// Implemented for the wide unsigned integer that holds the input counts +/// (`u64` for `f64`, `usize` for `f32`). The associated [`Narrow`](Self::Narrow) +/// type is the matching integer that losslessly converts to the family's float, +/// and [`Float`](Self::Float) is the float itself. +/// +/// The associated constants and functions exist because the integer operations +/// the ratio needs — `checked_div` and `saturating_add` — are *inherent* +/// methods on each integer rather than trait methods, so no standard-library +/// generic bound can reach them. The same applies to float `/`, which is kept +/// behind [`float_div`](Self::float_div): a generic `T::Float / T::Float` would +/// trip the crate's `arithmetic_side_effects` lint because clippy cannot prove +/// the operands are floats. Burying the concrete division inside the impl keeps +/// the lint satisfied while letting the generic [`unit_ratio`] stay uniform. +pub(crate) trait RatioInt: Copy + PartialOrd + Sized { + /// Narrow unsigned integer that converts losslessly into [`Float`](Self::Float). + /// + /// The widest integer that fits the float's mantissa — `u32` for `f64`, + /// `u16` for `f32`. Counts are scaled into this type's range before the + /// final float conversion so the conversion never trips + /// `cast_precision_loss`. + type Narrow: Copy; + + /// Float type the ratio is returned as. + /// + /// Bounded by `Default` (so the generic can produce `0.0` without a + /// literal) and `From` (the lossless conversion this family + /// revolves around). + type Float: Copy + Default + From; + + /// Representation of zero on the wide integer. + /// + /// Used both for the zero-denominator guard and to detect a scaled + /// numerator that has floored to zero (see [`unit_ratio`]). + const ZERO: Self; + + /// Representation of one on the wide integer. + /// + /// The minimum value a clamped scaled numerator is raised to, and the + /// `+1` added to the scale so a count equal to `NARROW_MAX` is left + /// unscaled. + const ONE: Self; + + /// `Self::MAX` — the saturating fallback when an integer op would overflow. + /// + /// Used inside the scale computation when the divisor itself overflows; + /// in practice the inputs are far smaller, so this is purely defensive. + const MAX: Self; + + /// [`Narrow`](Self::Narrow)::`MAX`. + /// + /// Used as the saturating fallback when a scaled count still does not fit + /// — a defensive guard, since the shared scaling is sized to prevent this + /// in practice. + const NARROW_MAX: Self::Narrow; + + /// One past [`Narrow`](Self::Narrow)::`MAX` — the divisor threshold. + /// + /// This is the smallest divisor that guarantees a scaled count fits in + /// `Narrow`. Using exactly `NARROW_MAX + 1` (rather than `NARROW_MAX`) is + /// what keeps a count equal to `NARROW_MAX` on the unscaled path: + /// `NARROW_MAX / (NARROW_MAX + 1) == 0`, so the computed scale is `1` and + /// the count survives intact. The boundary test in this module pins this + /// invariant, which was the site of an off-by-one bug during development. + const RANGE: Self; + + /// Maximum of two values. + /// + /// Used to pick the larger of the numerator and denominator when sizing the + /// shared scale, so a count that fits never forces the other operand to be + /// scaled away. + #[must_use] + fn max(self, other: Self) -> Self; + + /// Checked division. + /// + /// Returns `None` on divide-by-zero so the caller can supply a fallback + /// rather than panicking — required by the no-panic policy. + fn checked_div(self, rhs: Self) -> Option; + + /// Saturating addition. + /// + /// Used only on the `+1` term when computing the scale, so the result can + /// never overflow; saturating is the lint-clean spelling. + #[must_use] + fn saturating_add(self, rhs: Self) -> Self; + + /// Fallible narrowing to [`Narrow`](Self::Narrow). + /// + /// Returns `None` if the value exceeds `NARROW_MAX`; the caller falls back + /// to [`NARROW_MAX`](Self::NARROW_MAX). In practice the shared scaling has + /// already brought the value into range, so the fallback is defensive. + fn try_into_narrow(self) -> Option; + + /// Divide two floats to form the ratio. + /// + /// A trait method rather than a generic `T::Float / T::Float` solely to + /// satisfy the crate's `arithmetic_side_effects` lint — see the trait-level + /// docs. + fn float_div(numerator: Self::Float, denominator: Self::Float) -> Self::Float; +} + +/// [`RatioInt`] for `f64` ratios: counts in `u64`, narrowed via `u32`. +/// +/// `u32` fits losslessly into `f64`'s 52-bit mantissa, so this is the highest +/// family the crate needs. Counts above `u32::MAX` are scaled before +/// narrowing — see [`unit_ratio`]. +impl RatioInt for u64 { + type Narrow = u32; + type Float = f64; + const ZERO: Self = 0; + const ONE: Self = 1; + const MAX: Self = u64::MAX; + const NARROW_MAX: Self::Narrow = u32::MAX; + const RANGE: Self = u32::MAX as u64 + 1; + + fn max(self, other: Self) -> Self { + core::cmp::max(self, other) + } + + fn checked_div(self, rhs: Self) -> Option { + u64::checked_div(self, rhs) + } + + fn saturating_add(self, rhs: Self) -> Self { + u64::saturating_add(self, rhs) + } + + fn try_into_narrow(self) -> Option { + u32::try_from(self).ok() + } + + fn float_div(numerator: Self::Float, denominator: Self::Float) -> Self::Float { + numerator / denominator + } +} + +/// [`RatioInt`] for `f32` ratios: counts in `usize`, narrowed via `u16`. +/// +/// `u16` fits losslessly into `f32`'s 24-bit mantissa, so this is the family +/// for similarity and match-fraction scores that return `f32`. Counts above +/// `u16::MAX` are scaled before narrowing — see [`unit_ratio`]. +impl RatioInt for usize { + type Narrow = u16; + type Float = f32; + const ZERO: Self = 0; + const ONE: Self = 1; + const MAX: Self = usize::MAX; + const NARROW_MAX: Self::Narrow = u16::MAX; + const RANGE: Self = u16::MAX as usize + 1; + + fn max(self, other: Self) -> Self { + core::cmp::max(self, other) + } + + fn checked_div(self, rhs: Self) -> Option { + usize::checked_div(self, rhs) + } + + fn saturating_add(self, rhs: Self) -> Self { + usize::saturating_add(self, rhs) + } + + fn try_into_narrow(self) -> Option { + u16::try_from(self).ok() + } + + fn float_div(numerator: Self::Float, denominator: Self::Float) -> Self::Float { + numerator / denominator + } +} + +/// Ratio of two counts as a float, computed without tripping +/// `cast_precision_loss`. +/// +/// Both counts are first scaled by a shared divisor and then each is narrowed +/// through the family's [`Narrow`](RatioInt::Narrow) integer (the widest +/// integer that losslessly converts to the family's [`Float`](RatioInt::Float)). +/// Scaling *both* operands by the same factor preserves the quotient, so even +/// when both counts exceed `Narrow::MAX` the ratio is exact rather than +/// collapsing to `1.0` via independent saturation. +/// +/// The divisor is sized off the larger of the two counts: `count / RANGE + 1`, +/// where `RANGE` is one past `Narrow::MAX`. The `+1` keeps a count equal to +/// `Narrow::MAX` on the unscaled path (`NARROW_MAX / (NARROW_MAX + 1) == 0`, +/// so the divisor is `1`); the boundary test in this module pins that +/// invariant, which was the site of an off-by-one bug during development. +/// +/// Two further edge cases are handled explicitly: +/// +/// - **Zero denominator** returns `0.0` (rather than `NaN` or `inf`), matching +/// what every current caller wants for an empty set or window. +/// - **Scaled operand that floors to `0`** is clamped up to `1` — applied +/// symmetrically to numerator and denominator. For the numerator this keeps +/// a small positive value over a large one (e.g. `1 / 70_000`) from +/// collapsing to `0.0`; for the denominator it keeps a large numerator over a +/// small one (e.g. `70_000 / 1`) from producing `inf`. Either way, a positive +/// denominator guarantees a finite result. +/// +/// The result is not clamped to `[0, 1]`: when the numerator exceeds the +/// denominator it rises above `1.0` (e.g. context-window overflow reports a +/// utilization greater than one). +/// Divide `value` by `scale`, keeping any positive input strictly positive. +/// +/// The integer division floors, so a small positive `value` divided by a large +/// `scale` can drop to zero — collapsing a genuine operand of the ratio. This +/// clamps such a floored result back up to one so that a positive numerator or +/// denominator never vanishes (which would otherwise yield `0.0` or `inf` +/// respectively). Used symmetrically on both operands of [`unit_ratio`]. +fn scale_operand(value: T, scale: T) -> T { + let scaled = T::checked_div(value, scale).unwrap_or(value); + if value > T::ZERO && scaled == T::ZERO { + T::ONE + } else { + scaled + } +} + +#[must_use] +pub(crate) fn unit_ratio(numerator: T, denominator: T) -> T::Float { + if denominator == T::ZERO { + return T::Float::default(); + } + let scale = T::saturating_add( + T::checked_div(T::max(denominator, numerator), T::RANGE).unwrap_or(T::MAX), + T::ONE, + ); + let numerator_scaled = scale_operand(numerator, scale); + let denominator_scaled = scale_operand(denominator, scale); + let numerator_narrow = numerator_scaled.try_into_narrow().unwrap_or(T::NARROW_MAX); + let denominator_narrow = denominator_scaled + .try_into_narrow() + .unwrap_or(T::NARROW_MAX); + T::float_div( + T::Float::from(numerator_narrow), + T::Float::from(denominator_narrow), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unit_ratio_f64_normal_range() { + assert!((unit_ratio::(1, 4) - 0.25).abs() < 0.001); + assert!((unit_ratio::(3, 3) - 1.0).abs() < f64::EPSILON); + assert!(unit_ratio::(0, 3).abs() < f64::EPSILON); + } + + #[test] + fn unit_ratio_f64_is_zero_for_zero_denominator() { + assert!(unit_ratio::(5, 0).abs() < f64::EPSILON); + } + + #[test] + fn unit_ratio_f64_exceeds_one_for_overflow() { + let ratio = unit_ratio::(150, 100); + assert!( + (ratio - 1.5).abs() < 0.001, + "overflow must exceed 1.0, got {ratio}" + ); + } + + #[test] + fn unit_ratio_f64_large_partial_overlap_does_not_collapse_to_one() { + let over = u64::from(u32::MAX) + 1; + let ratio = unit_ratio::(over * 7 / 10, over); + assert!( + (ratio - 0.7).abs() < 0.01, + "partial overlap above u32::MAX must not collapse to 1.0, got {ratio}" + ); + } + + #[test] + fn unit_ratio_f64_small_positive_numerator_stays_positive_at_large_scale() { + // numerator=1 over a denominator above u32::MAX triggers shared + // scaling; the integer division 1/scale would floor to 0, collapsing a + // genuine match to 0.0. The result must stay strictly positive. + let over = u64::from(u32::MAX) + 1; + let ratio = unit_ratio::(1, over); + assert!( + ratio > 0.0, + "positive numerator must yield positive ratio, got {ratio}" + ); + } + + #[test] + fn unit_ratio_f64_large_overflow_exceeds_one() { + let over = u64::from(u32::MAX) + 1; + let ratio = unit_ratio::(over + over / 2, over); + assert!( + ratio > 1.0, + "overflow above u32::MAX must stay above 1.0, got {ratio}" + ); + } + + #[test] + fn unit_ratio_f32_identity_and_disjoint_in_normal_range() { + assert!((unit_ratio::(3, 3) - 1.0).abs() < f32::EPSILON); + assert!(unit_ratio::(0, 3).abs() < f32::EPSILON); + } + + #[test] + fn unit_ratio_f32_preserves_fraction_in_normal_range() { + assert!((unit_ratio::(1, 4) - 0.25).abs() < 0.01); + assert!((unit_ratio::(2, 3) - 0.66).abs() < 0.01); + } + + #[test] + fn unit_ratio_f32_is_zero_for_zero_denominator() { + assert!(unit_ratio::(5, 0).abs() < f32::EPSILON); + } + + #[test] + fn unit_ratio_f32_large_partial_overlap_does_not_collapse_to_one() { + let ratio = unit_ratio::(70_000, 100_000); + assert!((ratio - 0.7_f32).abs() < 0.01, "expected ~0.7, got {ratio}"); + assert!( + (ratio - 1.0_f32).abs() > 0.01, + "partial overlap must not collapse to 1.0" + ); + } + + #[test] + fn unit_ratio_f32_small_positive_numerator_stays_positive_at_large_scale() { + // numerator=1 over a denominator above u16::MAX triggers shared + // scaling; the integer division 1/2 would floor to 0, collapsing a + // genuine match to 0.0. The result must stay strictly positive. + let ratio = unit_ratio::(1, 70_000); + assert!( + ratio > 0.0, + "positive numerator must yield positive ratio, got {ratio}" + ); + } + + #[test] + fn unit_ratio_at_narrow_threshold_does_not_overscale() { + // A count equal to Narrow::MAX must stay on the unscaled path: the + // divisor is count / RANGE + 1, and RANGE == NARROW_MAX + 1 keeps it + // at 1. A near-identical pair (differing by 1) must therefore yield a + // result just under 1.0, not collapse to exactly 1.0 (which a wrong + // RANGE == NARROW_MAX would cause by scaling both operands to equal + // floored values). + let near_max = u16::MAX as usize; + let ratio = unit_ratio::(near_max.saturating_sub(1), near_max); + assert!( + ratio < 1.0, + "near-identical counts must not collapse to 1.0 at the threshold, got {ratio}" + ); + assert!( + (ratio - 0.999_97_f32).abs() < 0.001, + "expected ~0.99997, got {ratio}" + ); + } + + #[test] + fn unit_ratio_f32_small_positive_denominator_stays_finite_at_large_scale() { + // A large numerator (above u16::MAX) with a small nonzero denominator + // triggers shared scaling; the integer division den/scale would floor + // to 0, producing inf (or NaN) without a denominator clamp. The result + // must stay finite and reflect the true large ratio. + let ratio = unit_ratio::(70_000, 1); + assert!( + ratio.is_finite(), + "positive denominator must yield a finite ratio, got {ratio}" + ); + assert!(ratio > 1.0, "70_000/1 must exceed 1.0, got {ratio}"); + } + + #[test] + fn unit_ratio_f64_small_positive_denominator_stays_finite_at_large_scale() { + let over = u64::from(u32::MAX) + 1; + let ratio = unit_ratio::(over, 1); + assert!( + ratio.is_finite(), + "positive denominator must yield a finite ratio, got {ratio}" + ); + assert!(ratio > 1.0, "over/1 must exceed 1.0, got {ratio}"); + } +} diff --git a/src/presets.rs b/src/presets.rs index f587821..479e5d5 100644 --- a/src/presets.rs +++ b/src/presets.rs @@ -289,7 +289,6 @@ impl ContextContributor for GoalReminder { } #[cfg(test)] -#[allow(clippy::float_cmp)] mod tests { use super::*; diff --git a/src/provider/anthropic.rs b/src/provider/anthropic.rs index 3545da7..fdfa31e 100644 --- a/src/provider/anthropic.rs +++ b/src/provider/anthropic.rs @@ -883,7 +883,6 @@ impl SseReader { /// - Emitting [`PartStop`] when parts finish. /// - Emitting the final [`MessageDelta`] with stop reason and usage. #[derive(Default)] -#[allow(clippy::struct_excessive_bools)] struct StreamEmitter { /// Whether [`MessageStart`] has been emitted for the current stream. /// diff --git a/src/provider/gemini.rs b/src/provider/gemini.rs index 36605e4..8cd8e30 100644 --- a/src/provider/gemini.rs +++ b/src/provider/gemini.rs @@ -809,6 +809,75 @@ impl SseReader { } } +/// Whether a content lane (text or thinking) currently has an open part. +/// +/// Gemini interleaves regular text parts and reasoning (`thought: true`) +/// parts within a single chunk's `parts[]` array. [`StreamEmitter`] opens a +/// part for each lane with a [`PartStart`](StreamEvent::PartStart) on its +/// first non-empty fragment and closes it with a +/// [`PartStop`](StreamEvent::PartStop) when the lane switches or the stream +/// finishes. The two lanes are mutually exclusive — switching emits a +/// `PartStop` for the active lane first — so this enum tracks each lane's +/// open/closed state without a bare `bool`. +#[derive(Default)] +enum PartLane { + /// No part is open for this lane. + /// + /// The default state before the stream delivers any content for the + /// lane, and the state it returns to once a part has been closed. + #[default] + Closed, + + /// A part is open and accumulating deltas. + /// + /// Set when the first non-empty fragment opens the lane; cleared when + /// [`extract_finish_reason`](StreamEmitter::extract_finish_reason) or a + /// lane switch emits the matching + /// [`PartStop`](StreamEvent::PartStop). + Open, +} + +/// How far the stream has progressed through its terminal sequence. +/// +/// A Gemini stream has two independent terminal transitions: +/// [`extract_finish_reason`](StreamEmitter::extract_finish_reason) processes a +/// `finishReason` chunk (emitting a [`MessageDelta`](StreamEvent::MessageDelta) +/// and closing any open parts), and [`finish`](StreamEmitter::finish) appends +/// the synthetic [`MessageStop`](StreamEvent::MessageStop). Either may occur +/// first — a `finishReason` chunk may arrive before stream end, and `finish` +/// is always called at stream end — and each must fire exactly once. Because +/// the two are independent, this enum encodes all four combinations rather +/// than a single linear progression, advancing to [`Terminal`](Self::Terminal) +/// only once both are complete. +#[derive(Default)] +enum TerminalStage { + /// Neither transition has run yet. + /// + /// The default state at the start of a stream. + #[default] + Pending, + + /// [`finish`](StreamEmitter::finish) has emitted the synthetic + /// [`MessageStop`](StreamEvent::MessageStop). + /// + /// Subsequent `finish` calls are no-ops. + StopEmitted, + + /// A `finishReason` chunk has been fully processed. + /// + /// Subsequent `finishReason` chunks (e.g. from proxies that re-emit) + /// are no-ops. + FinishReasonSeen, + + /// Both transitions are complete. + /// + /// The only state from which no further terminal work is possible; + /// reached from either [`StopEmitted`](Self::StopEmitted) or + /// [`FinishReasonSeen`](Self::FinishReasonSeen) once the other + /// transition fires. + Terminal, +} + /// Stateful translator that converts Gemini SSE chunks into /// [`StreamEvent`]s. /// @@ -819,7 +888,6 @@ impl SseReader { /// with regular parts in the same `parts[]` array, flagged by a /// `thought: true` boolean on each thought part. #[derive(Default)] -#[allow(clippy::struct_excessive_bools)] struct StreamEmitter { /// Whether [`StreamEvent::MessageStart`] has been emitted for the /// current stream. @@ -836,7 +904,7 @@ struct StreamEmitter { /// first non-empty text fragment and tracks the open state so /// [`extract_finish_reason`](Self::extract_finish_reason) emits exactly /// one [`StreamEvent::PartStop`] to close it. - text_part_open: bool, + text: PartLane, /// Whether the reasoning (thinking) content part is currently open. /// @@ -844,11 +912,8 @@ struct StreamEmitter { /// opens a thinking part on the first non-empty thought fragment and /// closes it in [`extract_finish_reason`](Self::extract_finish_reason), /// symmetric to the text lane. - thinking_part_open: bool, + thinking: PartLane, - /// Whether a tool-call part is currently open. - /// - /// Each `functionCall` part emits a [`PartStart`]. Without tracking, /// Next tool-call index for multiple function calls in one response. /// /// Gemini can emit several `functionCall` parts in a single chunk. @@ -856,19 +921,15 @@ struct StreamEmitter { /// accumulator can distinguish them. next_tool_index: usize, - /// Whether the terminal stop signal has been processed. - /// - /// Set by [`finish`](Self::finish) when it appends the synthetic - /// [`StreamEvent::MessageStop`]. Guards against emitting a second - /// `MessageStop` if `finish` is called again after the stream ends. - message_stop_emitted: bool, - - /// Whether `finishReason` has already been processed. + /// How far the stream has progressed through its terminal sequence. /// - /// Guards against a second `finishReason` chunk (e.g. from proxies that - /// re-emit). Distinct from `message_stop_emitted` — the finish-reason - /// processing and the terminal `MessageStop` synthesis are independent. - finish_reason_processed: bool, + /// Tracks two independent transitions — `finishReason` processing (by + /// [`extract_finish_reason`](Self::extract_finish_reason)) and the + /// synthetic [`StreamEvent::MessageStop`] emission (by + /// [`finish`](Self::finish)) — as a single field. The two are + /// independent: a `finishReason` chunk may arrive before or after + /// `MessageStop`, and each must fire exactly once. + terminal: TerminalStage, /// Buffered [`StreamEvent`]s waiting to be yielded to the consumer. /// @@ -941,12 +1002,12 @@ impl StreamEmitter { .unwrap_or(false); if is_thought { - if self.text_part_open { - self.text_part_open = false; + if matches!(self.text, PartLane::Open) { + self.text = PartLane::Closed; self.push(StreamEvent::PartStop); } - if !self.thinking_part_open { - self.thinking_part_open = true; + if matches!(self.thinking, PartLane::Closed) { + self.thinking = PartLane::Open; self.push(StreamEvent::PartStart(PartStart { index: THINKING_PART_INDEX, part: None, @@ -959,12 +1020,12 @@ impl StreamEmitter { }, })); } else { - if self.thinking_part_open { - self.thinking_part_open = false; + if matches!(self.thinking, PartLane::Open) { + self.thinking = PartLane::Closed; self.push(StreamEvent::PartStop); } - if !self.text_part_open { - self.text_part_open = true; + if matches!(self.text, PartLane::Closed) { + self.text = PartLane::Open; self.push(StreamEvent::PartStart(PartStart { index: TEXT_PART_INDEX, part: Some(MessagePart::text("")), @@ -1003,12 +1064,12 @@ impl StreamEmitter { let args = func_call.pointer("/args").cloned().unwrap_or(Value::Null); let args_str = serde_json::to_string(&args).unwrap_or_default(); - if self.thinking_part_open { - self.thinking_part_open = false; + if matches!(self.thinking, PartLane::Open) { + self.thinking = PartLane::Closed; self.push(StreamEvent::PartStop); } - if self.text_part_open { - self.text_part_open = false; + if matches!(self.text, PartLane::Open) { + self.text = PartLane::Closed; self.push(StreamEvent::PartStop); } @@ -1039,9 +1100,12 @@ impl StreamEmitter { /// [`PartStop`](StreamEvent::PartStop), then emits a /// [`MessageDelta`](StreamEvent::MessageDelta) carrying the mapped /// [`StreamStopReason`]. No-op on a second `finishReason` chunk (the - /// `finish_reason_processed` guard defends against proxies/gateways that re-emit). + /// `terminal` stage guards against proxies/gateways that re-emit). fn extract_finish_reason(&mut self, json: &Value) { - if self.finish_reason_processed { + if matches!( + self.terminal, + TerminalStage::FinishReasonSeen | TerminalStage::Terminal + ) { return; } let Some(reason) = json @@ -1050,19 +1114,22 @@ impl StreamEmitter { else { return; }; - self.finish_reason_processed = true; + self.terminal = match self.terminal { + TerminalStage::StopEmitted => TerminalStage::Terminal, + _ => TerminalStage::FinishReasonSeen, + }; let stop = match reason { "MAX_TOKENS" => StreamStopReason::MaxTokens, _ => StreamStopReason::EndTurn, }; - if self.thinking_part_open { - self.thinking_part_open = false; + if matches!(self.thinking, PartLane::Open) { + self.thinking = PartLane::Closed; self.push(StreamEvent::PartStop); } - if self.text_part_open { - self.text_part_open = false; + if matches!(self.text, PartLane::Open) { + self.text = PartLane::Closed; self.push(StreamEvent::PartStop); } @@ -1113,11 +1180,18 @@ impl StreamEmitter { /// /// Drains any remaining pending events and appends the stop event. /// Safe to call exactly once at the end of the stream; subsequent calls - /// return an empty vec (the `message_stop_emitted` flag guards against double-stop). + /// return an empty vec (the `terminal` stage guards against double-stop). fn finish(&mut self) -> Vec { let mut out = self.drain(); - if self.started && !self.message_stop_emitted { - self.message_stop_emitted = true; + let stop_pending = matches!( + self.terminal, + TerminalStage::Pending | TerminalStage::FinishReasonSeen + ); + if self.started && stop_pending { + self.terminal = match self.terminal { + TerminalStage::FinishReasonSeen => TerminalStage::Terminal, + _ => TerminalStage::StopEmitted, + }; out.push(StreamEvent::MessageStop); } out @@ -1139,6 +1213,19 @@ mod tests { use super::*; use crate::message::{Message, MessagePart, Role, ToolContent}; + #[test] + fn gemini_terminal_stage_starts_pending() { + let em = StreamEmitter::default(); + assert!( + matches!(em.terminal, TerminalStage::Pending), + "terminal stage must start pending" + ); + assert!( + matches!(em.text, PartLane::Closed) && matches!(em.thinking, PartLane::Closed), + "both content lanes must start closed" + ); + } + #[test] fn request_body_user_text() { let msgs = vec![Message::user("hello")]; @@ -1853,6 +1940,10 @@ mod tests { })); let events = em.drain(); assert!(events.iter().any(|e| matches!(e, StreamEvent::PartStop))); + assert!( + matches!(em.text, PartLane::Closed), + "extract_finish_reason must close the text lane in state, not just emit PartStop" + ); let md = events .iter() .find(|e| matches!(e, StreamEvent::MessageDelta(_))); @@ -2045,7 +2136,7 @@ mod tests { fn emitter_finish_emits_message_stop_if_needed() { let mut em = StreamEmitter::default(); em.started = true; - em.message_stop_emitted = false; + em.terminal = TerminalStage::Pending; let events = em.finish(); assert!(events.iter().any(|e| matches!(e, StreamEvent::MessageStop))); @@ -2055,7 +2146,7 @@ mod tests { fn emitter_finish_noop_if_already_stopped() { let mut em = StreamEmitter::default(); em.started = true; - em.message_stop_emitted = true; + em.terminal = TerminalStage::StopEmitted; let events = em.finish(); assert!(events.is_empty()); @@ -2063,7 +2154,7 @@ mod tests { #[test] fn emitter_finish_reason_does_not_suppress_message_stop() { - // Regression: extract_finish_reason sets finish_reason_processed, + // Regression: extract_finish_reason advances the terminal stage, // but finish() must still emit MessageStop. Previously both used the // same `finished` flag, causing finish() to skip MessageStop after // a finishReason chunk. @@ -2073,7 +2164,7 @@ mod tests { "candidates": [{"finishReason": "STOP"}] })); em.drain(); - assert!(em.finish_reason_processed); + assert!(matches!(em.terminal, TerminalStage::FinishReasonSeen)); let events = em.finish(); assert!( events.iter().any(|e| matches!(e, StreamEvent::MessageStop)), @@ -2081,6 +2172,47 @@ mod tests { ); } + #[test] + fn emitter_finish_reason_after_finish_advances_to_terminal() { + // A finishReason arriving after finish() still emits its own + // MessageDelta, but a *second* such chunk must be a no-op once the + // terminal stage reaches Terminal. + let mut em = StreamEmitter::default(); + em.started = true; + let stop_events = em.finish(); + assert!( + stop_events + .iter() + .any(|e| matches!(e, StreamEvent::MessageStop)), + "first finish must emit MessageStop" + ); + assert!(matches!(em.terminal, TerminalStage::StopEmitted)); + + em.process_chunk(&serde_json::json!({ + "candidates": [{"finishReason": "STOP"}] + })); + assert!( + matches!(em.terminal, TerminalStage::Terminal), + "late finishReason after finish must advance to Terminal" + ); + let late_events = em.drain(); + assert!( + late_events + .iter() + .any(|e| matches!(e, StreamEvent::MessageDelta(_))), + "the first late finishReason still emits its MessageDelta" + ); + + em.process_chunk(&serde_json::json!({ + "candidates": [{"finishReason": "STOP"}] + })); + let duplicate_events = em.drain(); + assert!( + duplicate_events.is_empty(), + "a second late finishReason must be a no-op once Terminal is reached" + ); + } + #[test] fn sse_reader_take_line_extracts_newline() { let mut reader = SseReader { diff --git a/src/provider/openai.rs b/src/provider/openai.rs index 7bb9a47..4f1e655 100644 --- a/src/provider/openai.rs +++ b/src/provider/openai.rs @@ -1152,6 +1152,32 @@ struct OpenAiToolCallFunction { name: Option, } +/// Whether a content lane (text or thinking) currently has an open part. +/// +/// [`StreamEmitter`] separates assistant text and model reasoning into two +/// independent lanes, each opened with a [`PartStart`](StreamEvent::PartStart) +/// on its first non-empty fragment and closed with a +/// [`PartStop`](StreamEvent::PartStop) when the lane switches or the stream +/// finishes. Only one lane may be open at a time — the emitter emits a +/// `PartStop` for the active lane before opening the other — so this enum +/// tracks the open/closed state of each lane without a bare `bool`. +#[derive(Default)] +enum PartLane { + /// No part is open for this lane. + /// + /// The default state before the stream delivers any content for the + /// lane, and the state it returns to once a part has been closed. + #[default] + Closed, + + /// A part is open and accumulating deltas. + /// + /// Set when the first non-empty fragment opens the lane; cleared when + /// [`process_finish`](StreamEmitter::process_finish) emits the matching + /// [`PartStop`](StreamEvent::PartStop). + Open, +} + /// Stateful translator that converts a sequence of [`OpenAiChunk`]s /// into [`StreamEvent`]s. /// @@ -1164,7 +1190,6 @@ struct OpenAiToolCallFunction { /// Splitting this out from the `try_stream!` macro body makes the /// translation logic testable without a live network connection. #[derive(Default)] -#[allow(clippy::struct_excessive_bools)] struct StreamEmitter { /// Whether [`StreamEvent::MessageStart`] has been emitted for the /// current stream. @@ -1181,7 +1206,7 @@ struct StreamEmitter { /// [`StreamEvent::PartStart`] on the first non-empty fragment and /// tracks the open state so [`process_finish`](Self::process_finish) /// emits exactly one [`StreamEvent::PartStop`] to close it. - text_part_open: bool, + text: PartLane, /// Whether the reasoning (thinking) content part is currently open. /// @@ -1189,7 +1214,7 @@ struct StreamEmitter { /// alias `delta.reasoning`). The emitter opens a thinking part on the /// first non-empty fragment and closes it in `process_finish`, symmetric /// to the text lane. - thinking_part_open: bool, + thinking: PartLane, /// Tool-call indices that have already had their /// [`StreamEvent::PartStart`] emitted. @@ -1305,12 +1330,12 @@ impl StreamEmitter { if let Some(text) = &delta.content && !text.is_empty() { - if self.thinking_part_open { - self.thinking_part_open = false; + if matches!(self.thinking, PartLane::Open) { + self.thinking = PartLane::Closed; self.push(StreamEvent::PartStop); } - if !self.text_part_open { - self.text_part_open = true; + if matches!(self.text, PartLane::Closed) { + self.text = PartLane::Open; self.push(StreamEvent::PartStart(PartStart { index: TEXT_PART_INDEX, part: Some(MessagePart::text("")), @@ -1325,12 +1350,12 @@ impl StreamEmitter { if let Some(reasoning) = &delta.reasoning_content && !reasoning.is_empty() { - if self.text_part_open { - self.text_part_open = false; + if matches!(self.text, PartLane::Open) { + self.text = PartLane::Closed; self.push(StreamEvent::PartStop); } - if !self.thinking_part_open { - self.thinking_part_open = true; + if matches!(self.thinking, PartLane::Closed) { + self.thinking = PartLane::Open; self.push(StreamEvent::PartStart(PartStart { index: THINKING_PART_INDEX, part: None, @@ -1411,11 +1436,13 @@ impl StreamEmitter { } self.finished = true; - if self.text_part_open { + if matches!(self.text, PartLane::Open) { + self.text = PartLane::Closed; self.push(StreamEvent::PartStop); } - if self.thinking_part_open { + if matches!(self.thinking, PartLane::Open) { + self.thinking = PartLane::Closed; self.push(StreamEvent::PartStop); } @@ -1491,6 +1518,15 @@ mod tests { use crate::message::{Message, MessagePart, Role, ToolContent}; use crate::tool::ToolSchema; + #[test] + fn openai_emitter_part_lane_default_closed() { + let em = StreamEmitter::default(); + assert!( + matches!(em.text, PartLane::Closed) && matches!(em.thinking, PartLane::Closed), + "both content lanes must start closed" + ); + } + #[test] fn request_body_includes_system_message_first() { let msgs = vec![Message::user("hello")]; @@ -2199,6 +2235,42 @@ mod tests { ); } + #[test] + fn emitter_finish_closes_lanes_so_late_delta_reopens() { + let mut em = StreamEmitter::default(); + + let text_chunk = OpenAiChunk::parse( + r#"{"id":"c1","model":"gpt-4o","choices":[{"delta":{"content":"hi"},"finish_reason":null}]}"#, + ) + .unwrap(); + em.process_chunk(&text_chunk); + em.drain(); + + let finish = OpenAiChunk::parse( + r#"{"id":"c1","model":"gpt-4o","choices":[{"delta":null,"finish_reason":"stop"}]}"#, + ) + .unwrap(); + em.process_chunk(&finish); + em.drain(); + assert!( + matches!(em.text, PartLane::Closed), + "process_finish must close the text lane" + ); + + let late_delta = OpenAiChunk::parse( + r#"{"id":"c1","model":"gpt-4o","choices":[{"delta":{"content":"more"},"finish_reason":null}]}"#, + ) + .unwrap(); + em.process_chunk(&late_delta); + let events = em.drain(); + assert!( + events + .iter() + .any(|e| matches!(e, StreamEvent::PartStart(_))), + "a delta after finish must re-open the text lane with PartStart" + ); + } + #[test] fn emitter_finish_with_tool_calls_stop_reason() { let mut em = StreamEmitter::default(); @@ -3026,7 +3098,10 @@ mod tests { ) }); assert!(!has_text, "reasoning-only chunk must not emit Text deltas"); - assert!(!em.text_part_open, "reasoning must not set text_part_open"); + assert!( + matches!(em.text, PartLane::Closed), + "reasoning must not open the text lane" + ); } #[test] diff --git a/src/tool/health.rs b/src/tool/health.rs index 876e5fe..b5a93bf 100644 --- a/src/tool/health.rs +++ b/src/tool/health.rs @@ -288,6 +288,12 @@ impl ToolStats { /// /// Returns 1.0 when no calls have been recorded (new tools start /// optimistic). + /// + /// The ratio is computed as `(successes * EWMA_SCALE) / total / EWMA_SCALE` + /// rather than `successes / total` directly so the intermediate result + /// keeps six digits of integer precision before the final narrowing to + /// `f64` — small success rates over very large call counts would otherwise + /// floor to zero in pure integer arithmetic. #[must_use] pub fn success_rate(&self) -> f64 { let total = self.total_calls.load(Ordering::Relaxed); @@ -295,15 +301,11 @@ impl ToolStats { return 1.0; } let successes = self.success_count.load(Ordering::Relaxed); - // Compute `successes * EWMA_SCALE / total` entirely in integers. - // `successes <= total`, so the result is in `[0, EWMA_SCALE]` (i.e. fits in u32). let rate = successes .saturating_mul(EWMA_SCALE) .checked_div(total) .unwrap_or(0); - // Result is in `[0, EWMA_SCALE]` (≤ 1_000_000), so `u32` is safe. - f64::from(u32::try_from(rate).unwrap_or(u32::MAX)) - / f64::from(u32::try_from(EWMA_SCALE).unwrap_or(u32::MAX)) + crate::numeric::unit_ratio(rate, EWMA_SCALE) } /// Composite health score (0.0–1.0) blending success rate with EWMA. @@ -318,9 +320,7 @@ impl ToolStats { #[must_use] pub fn health_score(&self) -> f64 { let v = self.ewma_success.load(Ordering::Relaxed).min(EWMA_SCALE); - // `v` is clamped to `EWMA_SCALE` (1_000_000), so `u32` is safe. - let ewma = f64::from(u32::try_from(v).unwrap_or(u32::MAX)) - / f64::from(u32::try_from(EWMA_SCALE).unwrap_or(u32::MAX)); + let ewma = crate::numeric::unit_ratio(v, EWMA_SCALE); 0.3 * self.success_rate() + 0.7 * ewma }