From d1391783590a550a3f7d8cdf9f7d5311d23ae771 Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Tue, 4 Aug 2026 09:52:45 +1200 Subject: [PATCH 1/3] fix: docs consistency --- CHANGELOG.md | 3 +++ src/api.rs | 4 ++-- src/config.rs | 26 +++++++++++++++++--------- src/engine/bare.rs | 6 +++--- 4 files changed, 25 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 849eba9..17fe599 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Fixed +- 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)". +- Reconciled the planning docs (ROADMAP, CONTEXT, ARCHITECTURE, README, DEPENDENCIES, DCH-DESIGN, the v0.2.0 release file) to the shipped 0.2.0 reality: status Planned→Shipped, `compact_threshold` u16→u8, `Loop::process_turn` soft-deprecated→removed, `LoopRuntime`/`LoopConfig`/`SessionResult`/`run_session` → their shipped replacements (`managers`/`SessionConfig`+`RunConfig`/`Run`+`Session`/`run`), MSRV 1.85→1.94, doctest count 303→286. Added a staleness banner to `LOOPCTL-DESIGN.md`. - 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 diff --git a/src/api.rs b/src/api.rs index dd8a052..e6ca52e 100644 --- a/src/api.rs +++ b/src/api.rs @@ -182,7 +182,7 @@ pub struct NonStreamingResponse { /// /// fn stream_messages( /// &self, -/// request: StreamRequest, +/// request: &StreamRequest, /// ) -> Pin> + Send + 'static>> { /// // Clone data from &self, then build and return a stream /// let model = self.model.clone(); @@ -193,7 +193,7 @@ pub struct NonStreamingResponse { /// /// fn create_message( /// &self, -/// request: StreamRequest, +/// request: &StreamRequest, /// ) -> Pin> + Send + '_>> { /// // Non-streaming fallback /// todo!() diff --git a/src/config.rs b/src/config.rs index a3832f3..095dd96 100644 --- a/src/config.rs +++ b/src/config.rs @@ -156,17 +156,25 @@ pub enum ParallelMode { /// docs for the ordering invariant. Choose this for read-heavy, /// multi-call turns where latency is the sum of independent operations. /// - /// # Side-effect divergence from Sequential + /// # Side-effects (same granularity in both modes) /// - /// In Sequential mode, loop detection and observer events fire on - /// **every** retry attempt — a tool that fails twice then succeeds - /// produces 3 detection operations and 3 observer pairs. + /// Detection, observer, hook, and health side-effects all fire on **every** + /// retry attempt in both modes. A tool that fails twice then succeeds + /// produces 3 detection operations, 3 observer PRE+POST pairs, and 3 health + /// recordings regardless of `ParallelMode`. All four side-effect targets are + /// thread-safe ([`DetectionManager`](crate::detection::DetectionManager) + /// and the observer registry use + /// `Mutex`/immutable-`Vec` interiors; + /// [`ToolHealthRegistry`](crate::tool::health::ToolHealthRegistry) uses + /// atomic counters), so concurrent retry attempts in Parallel mode dispatch + /// side-effects safely without serialization. /// - /// In Parallel mode, detection and observer side-effects fire **once** - /// on the final result only (they are not thread-safe). Health - /// tracking fires on **every** attempt in both modes (it uses atomic - /// counters). Intermediate retries during parallel dispatch are - /// invisible to loop detection and observers but visible to health. + /// The only retry-related difference between modes is **interleaving**, not + /// granularity: in Sequential the classic `[pre A, post A, pre B, post B]` + /// order is strict, while in Parallel the PRE/POST events for independent + /// calls in the same wave interleave as those calls progress concurrently. + /// Observers that pair `on_tool_pre`/`on_tool_post` should key on + /// `tool_call_id` (carried in both contexts), not on arrival order. Parallel, } diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 1767ca2..d8f8454 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -246,8 +246,8 @@ pub struct BareLoop { /// [`compaction_result`](LoopMachine::compaction_result), and /// [`inject`](LoopMachine::inject). It is (re)created at the top of every /// [`run()`](crate::engine::core::Loop::run) call from the run config - /// and user prompt; before that it holds an empty placeholder so the struct - /// is always valid. + /// and user prompt; before that it holds an empty machine (no history, no + /// pending messages) so the struct is always valid. machine: LoopMachine, /// Framework managers bundle — holds all cross-cutting infrastructure. @@ -524,7 +524,7 @@ impl BareLoop { /// accumulated history, turns taken, or the machine's internal state). The /// machine is (re)created at the top of every /// [`run()`](crate::engine::core::Loop::run) call; before the first run - /// it holds an empty placeholder. + /// it holds an empty machine (no history, no pending messages). #[must_use] pub fn machine(&self) -> &LoopMachine { &self.machine From 3eab7febc203cf51f039eee59232861264c59b80 Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Tue, 4 Aug 2026 09:53:07 +1200 Subject: [PATCH 2/3] chore: parallel dispatch test --- src/engine/bare/dispatch.rs | 118 ++++++++++++++++++++++++++++++++++++ 1 file changed, 118 insertions(+) diff --git a/src/engine/bare/dispatch.rs b/src/engine/bare/dispatch.rs index 5e73e6e..cc08acb 100644 --- a/src/engine/bare/dispatch.rs +++ b/src/engine/bare/dispatch.rs @@ -1483,6 +1483,124 @@ mod tests { ); } + #[tokio::test] + async fn parallel_retried_call_fires_side_effects_per_attempt() { + // Pins the documented contract (config.rs `ParallelMode`): detection, + // observer, hook, and health side-effects fire on EVERY retry attempt + // in BOTH modes. A retried parallel call must therefore emit multiple + // observer PRE+POST pairs, not one. Guards against a future change + // re-introducing per-mode gating that the contract explicitly disclaims + // (all side-effect targets are Send + Sync). + use crate::observer::{LoopObserver, ToolPostContext, ToolPreContext}; + use crate::reflection::{ + FailureAnalysis, FailureSeverity, RecoveryAction, RecoveryStrategy, + }; + use std::sync::atomic::{AtomicU32, Ordering}; + + struct AlwaysRecoverable; + impl crate::reflection::Reflector for AlwaysRecoverable { + fn analyze( + &self, + error: &str, + tool_name: &str, + _tool_input: &Value, + _tool_schema: Option<&crate::tool::ToolSchema>, + _context: &crate::reflection::ReflectionContext, + ) -> Pin< + Box< + dyn Future> + + Send + + '_, + >, + > { + let error = error.to_string(); + let tool_name = tool_name.to_string(); + Box::pin(async move { + Ok(FailureAnalysis { + is_recoverable: true, + root_cause: error, + severity: FailureSeverity::Medium, + correction: None, + context: format!("tool: {tool_name}"), + }) + }) + } + } + + // Retry the first two attempts, then give up with a soft error so the + // call terminates. Each attempt is a full dispatch with PRE+POST. + struct RetryTwice; + impl RecoveryStrategy for RetryTwice { + fn decide( + &self, + _analysis: &FailureAnalysis, + attempt: u32, + _max_attempts: u32, + ) -> Pin + Send + '_>> { + Box::pin(async move { + if attempt < 2 { + RecoveryAction::Retry { + delay: std::time::Duration::ZERO, + } + } else { + RecoveryAction::Skip("giving up".into()) + } + }) + } + } + + struct CountingObserver { + pre: Arc, + post: Arc, + } + impl LoopObserver for CountingObserver { + fn name(&self) -> &'static str { + "counting" + } + fn on_tool_pre(&self, _ctx: &ToolPreContext) { + self.pre.fetch_add(1, Ordering::Relaxed); + } + fn on_tool_post(&self, _ctx: &ToolPostContext) { + self.post.fetch_add(1, Ordering::Relaxed); + } + } + + // Tool that always errors; recovery drives 3 attempts (2 retries + 1 skip). + let error_tool = crate::tool::FnTool::new( + "error_tool".into(), + "Always errors".into(), + Value::Object(serde_json::Map::new()), + |_, _| Box::pin(async { Err(ToolError::Execution("boom".to_string())) }), + ); + + let pre_count = Arc::new(AtomicU32::new(0)); + let post_count = Arc::new(AtomicU32::new(0)); + + let mut registry = ToolRegistry::new(); + registry.register(error_tool); + let mut bare = make_parallel_loop(registry); + bare.set_reflector(Arc::new(AlwaysRecoverable)); + bare.set_recovery_strategy(Arc::new(RetryTwice)); + bare.register_observer(Arc::new(CountingObserver { + pre: Arc::clone(&pre_count), + post: Arc::clone(&post_count), + })); + + let calls = vec![make_call("1", "error_tool", Value::Null)]; + let _ = bare.dispatch_tools(&calls, 0).await.ok(); + + let pres = pre_count.load(Ordering::Relaxed); + let posts = post_count.load(Ordering::Relaxed); + assert!( + pres >= 2 && posts >= 2, + "parallel retried call must fire side-effects per attempt; got pre={pres} post={posts}" + ); + assert_eq!( + pres, posts, + "every PRE must have a matching POST (pairing invariant)" + ); + } + #[tokio::test] async fn execute_tool_call_runs_recovery_on_failure() { use crate::reflection::{ From 29ff20bf8ed8d286da02a7c959eb366b0fe69ee9 Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Tue, 4 Aug 2026 10:17:59 +1200 Subject: [PATCH 3/3] refactor: dedup recovery test helper, tighten parallel assertion --- src/engine/bare/dispatch.rs | 149 +++++++++++------------------------- 1 file changed, 46 insertions(+), 103 deletions(-) diff --git a/src/engine/bare/dispatch.rs b/src/engine/bare/dispatch.rs index cc08acb..a44330a 100644 --- a/src/engine/bare/dispatch.rs +++ b/src/engine/bare/dispatch.rs @@ -905,6 +905,7 @@ mod tests { use crate::engine::core::ToolCall; use crate::engine::{Run, RunConfig}; use crate::message::ToolContent; + use crate::reflection::{FailureAnalysis, FailureSeverity}; use crate::tool::{ Tool, ToolContext, ToolError, ToolOutput, ToolSchema, registry::ToolRegistry, }; @@ -918,6 +919,37 @@ mod tests { use super::*; + /// Reflector that marks every failure recoverable, used by the recovery tests. + struct AlwaysRecoverable; + impl crate::reflection::Reflector for AlwaysRecoverable { + fn analyze( + &self, + error: &str, + tool_name: &str, + _tool_input: &Value, + _tool_schema: Option<&crate::tool::ToolSchema>, + _context: &crate::reflection::ReflectionContext, + ) -> Pin< + Box< + dyn Future> + + Send + + '_, + >, + > { + let error = error.to_string(); + let tool_name = tool_name.to_string(); + Box::pin(async move { + Ok(FailureAnalysis { + is_recoverable: true, + root_cause: error, + severity: FailureSeverity::Medium, + correction: None, + context: format!("tool: {tool_name}"), + }) + }) + } + } + #[test] fn truncate_to_short_string_unchanged() { assert_eq!(truncate_to("hello", 10), "hello"); @@ -1398,39 +1430,7 @@ mod tests { #[tokio::test] async fn recovery_backoff_cancelled_promptly() { - use crate::reflection::{ - FailureAnalysis, FailureSeverity, RecoveryAction, RecoveryStrategy, - }; - - struct AlwaysRecoverable; - impl crate::reflection::Reflector for AlwaysRecoverable { - fn analyze( - &self, - error: &str, - tool_name: &str, - _tool_input: &Value, - _tool_schema: Option<&crate::tool::ToolSchema>, - _context: &crate::reflection::ReflectionContext, - ) -> Pin< - Box< - dyn Future> - + Send - + '_, - >, - > { - let error = error.to_string(); - let tool_name = tool_name.to_string(); - Box::pin(async move { - Ok(FailureAnalysis { - is_recoverable: true, - root_cause: error, - severity: FailureSeverity::Medium, - correction: None, - context: format!("tool: {tool_name}"), - }) - }) - } - } + use crate::reflection::{FailureAnalysis, RecoveryAction, RecoveryStrategy}; struct SlowRetry; impl RecoveryStrategy for SlowRetry { @@ -1492,41 +1492,9 @@ mod tests { // re-introducing per-mode gating that the contract explicitly disclaims // (all side-effect targets are Send + Sync). use crate::observer::{LoopObserver, ToolPostContext, ToolPreContext}; - use crate::reflection::{ - FailureAnalysis, FailureSeverity, RecoveryAction, RecoveryStrategy, - }; + use crate::reflection::{FailureAnalysis, RecoveryAction, RecoveryStrategy}; use std::sync::atomic::{AtomicU32, Ordering}; - struct AlwaysRecoverable; - impl crate::reflection::Reflector for AlwaysRecoverable { - fn analyze( - &self, - error: &str, - tool_name: &str, - _tool_input: &Value, - _tool_schema: Option<&crate::tool::ToolSchema>, - _context: &crate::reflection::ReflectionContext, - ) -> Pin< - Box< - dyn Future> - + Send - + '_, - >, - > { - let error = error.to_string(); - let tool_name = tool_name.to_string(); - Box::pin(async move { - Ok(FailureAnalysis { - is_recoverable: true, - root_cause: error, - severity: FailureSeverity::Medium, - correction: None, - context: format!("tool: {tool_name}"), - }) - }) - } - } - // Retry the first two attempts, then give up with a soft error so the // call terminates. Each attempt is a full dispatch with PRE+POST. struct RetryTwice; @@ -1587,13 +1555,20 @@ mod tests { })); let calls = vec![make_call("1", "error_tool", Value::Null)]; - let _ = bare.dispatch_tools(&calls, 0).await.ok(); + let _ = bare + .dispatch_tools(&calls, 0) + .await + .expect("dispatch should not hard-error"); let pres = pre_count.load(Ordering::Relaxed); let posts = post_count.load(Ordering::Relaxed); - assert!( - pres >= 2 && posts >= 2, - "parallel retried call must fire side-effects per attempt; got pre={pres} post={posts}" + assert_eq!( + pres, 3, + "RetryTwice does 2 retries + 1 final skip = 3 attempts = 3 PRE events; got {pres}" + ); + assert_eq!( + posts, 3, + "matching 3 POST events for the 3 attempts; got {posts}" ); assert_eq!( pres, posts, @@ -1603,41 +1578,9 @@ mod tests { #[tokio::test] async fn execute_tool_call_runs_recovery_on_failure() { - use crate::reflection::{ - FailureAnalysis, FailureSeverity, RecoveryAction, RecoveryStrategy, - }; + use crate::reflection::{FailureAnalysis, RecoveryAction, RecoveryStrategy}; use std::sync::atomic::{AtomicU32, Ordering}; - struct AlwaysRecoverable; - impl crate::reflection::Reflector for AlwaysRecoverable { - fn analyze( - &self, - error: &str, - tool_name: &str, - _tool_input: &Value, - _tool_schema: Option<&crate::tool::ToolSchema>, - _context: &crate::reflection::ReflectionContext, - ) -> Pin< - Box< - dyn Future> - + Send - + '_, - >, - > { - let error = error.to_string(); - let tool_name = tool_name.to_string(); - Box::pin(async move { - Ok(FailureAnalysis { - is_recoverable: true, - root_cause: error, - severity: FailureSeverity::Medium, - correction: None, - context: format!("tool: {tool_name}"), - }) - }) - } - } - struct CountingRetry { calls: Arc, }