fix: harden detection layer and rate limit refill clock

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0.
 
 ### Fixed
 
+- Detection-layer long-tail correctness (six items): `check_file_reads` now normalizes the query under the recorded op's real tool name (was: empty string) and matches by bidirectional path containment (was: exact equality); `LoopDetector::new` clamps `window_size == 0` to 1 with a warning; `find_best_match` rejects zero-score candidates even at threshold 0.0 and breaks ties lexicographically; `ToolShield::with_thresholds` swaps inverted warn/block pairs with a warning. All non-breaking.
 - `SessionConfig::compact_threshold` now clamps into its documented `0..=100` range on the validating construction paths (`Default`, `with_compact_threshold`, and `Deserialize`). Previously only the `with_compact_threshold` builder clamped; deserialized configs could carry an out-of-range value (e.g. `200` from disk), which the compaction subsystem would then interpret as "never compact." A single canonical clamp method plus a field-level serde deserialize helper now enforce the range silently. `Default` was already in range (80); its clamp call is defensive. Direct public struct-literal construction (`SessionConfig { compact_threshold: 200, .. }`) still bypasses normalization — the field is `pub`, and callers using that form are responsible for honoring the documented range. Non-breaking (silent clamp; no signature changes).
 - Corrected the `ParallelMode::Parallel` doc, which falsely claimed detection/observer side-effects "are not thread-safe" and fire "once on the final result only" in parallel mode. They are thread-safe (`DetectionManager` and the observer registry use `Mutex`/immutable-`Vec` interiors; `ToolHealthRegistry` uses atomics) and fire on every retry attempt in both modes, exactly as the code already does. No behavior change; the code matched the corrected doc all along.
 - Fixed code-level doc contradictions: the `ApiClient` trait example showed `request: StreamRequest` (by-value) instead of `&StreamRequest` (matches the real trait), and `BareLoop::machine` was described as an "empty placeholder" rather than the real "empty machine (no history, no pending messages)".

src/detection/loop_detector.rs

@@ -1098,7 +1098,13 @@ impl LoopDetector {
     /// let config = LoopDetectorConfig { window_size: 100, ..Default::default() };
     /// let detector = LoopDetector::new(config, Arc::new(NoOpToolSignature));
     /// ```
-    pub fn new(config: LoopDetectorConfig, signature: Arc<dyn ToolSignature>) -> Self {
+    pub fn new(mut config: LoopDetectorConfig, signature: Arc<dyn ToolSignature>) -> Self {
+        if config.window_size == 0 {
+            tracing::warn!(
+                "LoopDetectorConfig.window_size was 0; clamping to 1 (the smallest sensible window)"
+            );
+            config.window_size = 1;
+        }
         Self {
             operations: Mutex::new(VecDeque::with_capacity(config.window_size)),
             config,
@@ -1580,13 +1586,16 @@ impl LoopDetector {
         };
 
         let sig = &self.signature;
-        let normalized_input = sig.normalize_param_for_comparison("", file_path);
         let read_count = ops
             .iter()
             .filter(|o| {
-                sig.is_file_read_tool(&o.tool)
-                    && sig.normalize_param_for_comparison(&o.tool, &o.primary_param)
-                        == normalized_input
+                if !sig.is_file_read_tool(&o.tool) {
+                    return false;
+                }
+                let normalized_op = sig.normalize_param_for_comparison(&o.tool, &o.primary_param);
+                let normalized_query = sig.normalize_param_for_comparison(&o.tool, file_path);
+                normalized_op.contains(normalized_query.as_str())
+                    || normalized_query.contains(normalized_op.as_str())
             })
             .count();
 
@@ -2539,4 +2548,84 @@ mod tests {
             "Should contain tool signature suggestion: {msg}"
         );
     }
+
+    /// Signature that flags the `Read` tool as a file-read tool and prefixes
+    /// the primary parameter with the tool name during comparison, so tests
+    /// can verify which tool name `check_file_reads` passes to
+    /// `normalize_param_for_comparison`.
+    struct ToolNameNormalizingSig;
+
+    impl ToolSignature for ToolNameNormalizingSig {
+        fn is_file_read_tool(&self, tool: &str) -> bool {
+            tool == "read"
+        }
+
+        fn normalize_param_for_comparison(&self, tool: &str, param: &str) -> String {
+            format!("{tool}:{param}")
+        }
+    }
+
+    #[test]
+    fn check_file_reads_matches_containment_both_directions() {
+        // Use the shared TestToolSignature which treats "Read" as a file-read
+        // tool and leaves params unchanged under normalization. Lower the
+        // threshold so a single recorded read trips the check.
+        let config = LoopDetectorConfig {
+            max_same_file_reads: 1,
+            ..Default::default()
+        };
+        let mk = || LoopDetector::new(config.clone(), Arc::new(TestToolSignature));
+
+        // Forward direction: op path "/a/b/c" contains query "/a/b".
+        let detector = mk();
+        detector.record(Operation::new("Read", "/a/b/c"));
+        assert!(
+            detector.check_file_reads("/a/b"),
+            "op path containing the query path should match"
+        );
+
+        // Reverse direction: query "/a/b/c" contains op path "/a".
+        let detector2 = mk();
+        detector2.record(Operation::new("Read", "/a"));
+        assert!(
+            detector2.check_file_reads("/a/b/c"),
+            "query path containing the op path should match"
+        );
+    }
+
+    #[test]
+    fn check_file_reads_uses_real_tool_name() {
+        // ToolNameNormalizingSig prepends the tool name during normalization,
+        // so the match only succeeds if check_file_reads passes the recorded
+        // op's tool name ("read") rather than an empty string.
+        let detector = LoopDetector::new(
+            LoopDetectorConfig {
+                max_same_file_reads: 1,
+                ..Default::default()
+            },
+            Arc::new(ToolNameNormalizingSig),
+        );
+
+        detector.record(Operation::new("read", "/file"));
+        // normalize_param_for_comparison("read", "/file") == "read:/file"
+        // and the query is normalized with the same tool name, so the
+        // containment check only holds if both use "read".
+        assert!(
+            detector.check_file_reads("/file"),
+            "check_file_reads must normalize under the recorded op's tool name"
+        );
+    }
+
+    #[test]
+    fn window_size_zero_clamps_to_one() {
+        let config = LoopDetectorConfig {
+            window_size: 0,
+            ..Default::default()
+        };
+        let detector = LoopDetector::new(config, Arc::new(TestToolSignature));
+        assert_eq!(
+            detector.config.window_size, 1,
+            "a window_size of 0 must be clamped to 1"
+        );
+    }
 }

src/middleware/unknown_tool.rs

@@ -148,9 +148,11 @@ impl UnknownToolMiddleware {
 
     /// Find the best matching tool name from a list, given a threshold.
     ///
-    /// Returns the name with the highest similarity score that meets or
-    /// exceeds `threshold`. Returns `None` when no candidate scores high
-    /// enough.
+    /// Returns the name with the highest similarity score that exceeds zero
+    /// and meets or exceeds `threshold`. Ties (equal scores) are broken
+    /// lexicographically — the smaller name wins — so the result is
+    /// deterministic regardless of candidate order. Returns `None` when no
+    /// candidate scores high enough.
     fn find_best_match_inner<'a>(
         requested: &str,
         available: &[&'a str],
@@ -159,10 +161,18 @@ impl UnknownToolMiddleware {
         let mut best: Option<(&'a str, f64)> = None;
         for &name in available {
             let score = Self::similarity(requested, name);
-            if score >= threshold {
+            if score > 0.0 && score >= threshold {
                 match best {
-                    Some((_, best_score)) if score <= best_score => {}
-                    _ => best = Some((name, score)),
+                    None => best = Some((name, score)),
+                    Some((_, best_score)) if score > best_score => {
+                        best = Some((name, score));
+                    }
+                    Some((best_name, best_score))
+                        if (score - best_score).abs() < f64::EPSILON && name < best_name =>
+                    {
+                        best = Some((name, score));
+                    }
+                    _ => {}
                 }
             }
         }
@@ -385,6 +395,39 @@ mod tests {
         );
     }
 
+    #[test]
+    fn find_best_match_threshold_zero_rejects_zero_score() {
+        let available = ["bbb"];
+        assert!(
+            UnknownToolMiddleware::find_best_match_inner("xyz", &available, 0.0).is_none(),
+            "a zero-similarity candidate must be rejected even at threshold 0.0"
+        );
+    }
+
+    #[test]
+    fn find_best_match_ties_break_lexicographically() {
+        let forward = ["toll", "toil"];
+        let (suggestion, score) =
+            UnknownToolMiddleware::find_best_match_inner("tool", &forward, 0.0).unwrap();
+        assert_eq!(
+            suggestion, "toil",
+            "tie should break to the lexicographically smaller name"
+        );
+        assert!(
+            (score - 0.8).abs() < 1e-6,
+            "expected tied score 0.8, got {score}"
+        );
+
+        // Reversed input order must yield the same winner.
+        let reversed = ["toil", "toll"];
+        let (suggestion_rev, _) =
+            UnknownToolMiddleware::find_best_match_inner("tool", &reversed, 0.0).unwrap();
+        assert_eq!(
+            suggestion_rev, "toil",
+            "tie-break must be order-independent"
+        );
+    }
+
     use crate::message::ToolContent;
     use crate::tool::ToolDispatchResult;
     use std::time::Duration;

src/stream/rate_limit.rs

@@ -1,9 +1,8 @@
 //! Proactive client-side rate limiting (token bucket).
 //!
-//! A [`TokenBucket`] is a continuous-fill bucket: it allows a burst equal to its
-//! capacity, then refills at `capacity / 60` tokens per second. [`RateLimiter`]
-//! holds one bucket per provider identity (`base_url`) so distinct providers get
-//! independent budgets.
+//! A [`TokenBucket`] allows a burst equal to its capacity, then refills at
+//! `capacity / 60` tokens per second. [`RateLimiter`] holds one bucket per
+//! provider identity (`base_url`) so distinct providers get independent budgets.
 //!
 //! This is the proactive complement to the reactive 429 handling in
 //! `stream::handler`: the bucket gates a request *before* it fires,
@@ -143,18 +142,9 @@ impl TokenBucket {
         self.take_at(Instant::now())
     }
 
-    /// Tokens available at a known instant (after a lazy refill). Non-consuming.
+    /// Tokens available at a known instant (after a lazy refill).
     ///
-    /// # Example
-    ///
-    /// ```rust
-    /// use loopctl::stream::rate_limit::TokenBucket;
-    ///
-    /// let bucket = TokenBucket::new(10);
-    /// // A fresh bucket is full.
-    /// assert!((bucket.available() - 10.0).abs() < 0.5);
-    /// ```
-    #[must_use]
+    /// Advances `last_refill` to `at`, banking the elapsed refill.
     pub fn available_at(&self, at: Instant) -> f64 {
         let mut state = self
             .state
@@ -163,31 +153,42 @@ impl TokenBucket {
         elapsed_refill(&mut state, at, self.capacity, self.refill_per_sec)
     }
 
-    /// Tokens available right now (after a lazy refill). Non-consuming.
-    ///
-    /// Thin wrapper around [`available_at`](Self::available_at) that
-    /// plugs in `Instant::now()`. Useful for observability — reporting
-    /// the current budget without consuming a token.
-    #[must_use]
+    /// Tokens available right now (after a lazy refill).
     pub fn available(&self) -> f64 {
         self.available_at(Instant::now())
     }
 }
 
 /// Apply the elapsed-time refill to `state` in place and return the new token count.
 ///
-/// Caps at `capacity` so a long idle does not overflow. Leaves `last_refill` at
-/// `at` so the next caller only accounts for the gap since this call.
+/// Caps at `capacity` so a long idle does not overflow. When the bucket fills
+/// before `at` (the elapsed time exceeded what was needed to reach capacity),
+/// `last_refill` advances only to the fill point — not to `at`. This prevents
+/// a caller that passes a far-future instant from freezing the bucket: the
+/// excess time beyond capacity is irrelevant, so the clock ignores it.
 fn elapsed_refill(state: &mut BucketState, at: Instant, capacity: f64, refill_per_sec: f64) -> f64 {
     let elapsed = at.saturating_duration_since(state.last_refill);
     if elapsed.is_zero() {
         return state.tokens;
     }
+    if refill_per_sec <= 0.0 {
+        return state.tokens;
+    }
     let added = elapsed.as_secs_f64() * refill_per_sec;
-    let topped = (state.tokens + added).min(capacity);
-    state.tokens = topped;
-    state.last_refill = at;
-    topped
+    let raw = state.tokens + added;
+    if raw >= capacity {
+        let needed = capacity - state.tokens;
+        let secs_to_fill = needed / refill_per_sec;
+        state.last_refill = state
+            .last_refill
+            .checked_add(Duration::from_secs_f64(secs_to_fill))
+            .unwrap_or(at);
+        state.tokens = capacity;
+    } else {
+        state.tokens = raw;
+        state.last_refill = at;
+    }
+    state.tokens
 }
 
 /// One [`TokenBucket`] per distinct provider identity (`base_url`).
@@ -385,4 +386,78 @@ mod tests {
         let result = bucket.take();
         assert!(result.is_err(), "empty bucket should return Err");
     }
+
+    #[test]
+    fn future_instant_breaks_rate_limit() {
+        let bucket = TokenBucket::new(10);
+        let now = Instant::now();
+        let one_hour_later = now + Duration::from_hours(1);
+        let one_min_later = now + Duration::from_mins(1);
+
+        // Drain all 10 tokens at t=now.
+        for _ in 0..10 {
+            bucket.take_at(now).expect("burst tokens");
+        }
+
+        // Poison: take_at with a future instant refills and sets
+        // last_refill = one_hour_later. The bucket now thinks time is
+        // 1 hour ahead of reality.
+        let _poison = bucket.take_at(one_hour_later);
+
+        // Drain any remaining tokens at the future instant.
+        for _ in 0..9 {
+            let _drain = bucket.take_at(one_hour_later);
+        }
+
+        // Now 1 minute has passed in real time. Normally 1 min of refill
+        // would restore tokens. After the fix, take_at should succeed
+        // because the bucket must not allow a future instant to freeze
+        // the refill clock.
+        let result = bucket.take_at(one_min_later);
+        assert!(
+            result.is_ok(),
+            "take_at must succeed — 1 min of refill should restore a token; the future-instant call must not freeze the bucket"
+        );
+    }
+
+    #[test]
+    fn past_instant_does_not_refund_or_corrupt() {
+        // A take_at called with an instant earlier than last_refill must not
+        // fabricate tokens from negative elapsed time, and must not rewind the
+        // clock. The probe uses a small forward step so the partial refill is
+        // stable (a far-future step would re-fill on every call).
+        let bucket = TokenBucket::new(5);
+        let t0 = Instant::now();
+        // 6s ≈ 0.5 token at capacity 5 (refill 5/60 per second).
+        let t1 = t0 + Duration::from_secs(6);
+
+        // Drain at t0.
+        for _ in 0..5 {
+            bucket.take_at(t0).expect("drain while full");
+        }
+        assert!(bucket.take_at(t0).is_err(), "drained at t0");
+
+        // Advance to t1: 6s of refill ≈ 0.5 token — still under one, take fails.
+        // last_refill is now t1, tokens ≈ 0.5.
+        assert!(
+            bucket.take_at(t1).is_err(),
+            "partial refill under one token"
+        );
+
+        // Probe: take at a past instant (t0 < last_refill = t1).
+        // Correct: saturating_duration_since returns ZERO → early return, no
+        // token granted, last_refill untouched.
+        assert!(
+            bucket.take_at(t0).is_err(),
+            "past-instant take must not refund tokens from negative elapsed time"
+        );
+
+        // Integrity: take_at(t1) again must observe the same 0.5-token state.
+        // A bug that rewound last_refill to t0 would make this see a fresh 6s
+        // gap (0.5 + 0.5 = 1.0 token) and succeed.
+        assert!(
+            bucket.take_at(t1).is_err(),
+            "last_refill must not rewind — bucket state unchanged by the past-instant probe"
+        );
+    }
 }