diff --git a/.clippy.toml b/.clippy.toml index 7bfabfe..93caca3 100644 --- a/.clippy.toml +++ b/.clippy.toml @@ -20,7 +20,7 @@ too-many-lines-threshold = 150 too-many-arguments-threshold = 7 # MSRV from Cargo.toml. -msrv = "1.94.0" +msrv = "1.98.0" # Warn when passing types larger than this by value (bytes). trivial-copy-size-limit = 32 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6b8fa51..dc4f34e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 - run: cargo check --all-features @@ -25,7 +25,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 - run: cargo test --all-features @@ -35,7 +35,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 with: components: clippy - uses: Swatinem/rust-cache@v2 @@ -47,7 +47,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 with: components: clippy - uses: Swatinem/rust-cache@v2 @@ -59,7 +59,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 with: components: rustfmt - run: cargo fmt --all -- --check @@ -70,7 +70,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 - run: cargo test --doc --all-features @@ -80,7 +80,7 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 - run: cargo doc --no-deps --all-features env: @@ -92,6 +92,6 @@ jobs: - uses: actions/checkout@v7 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 - run: cargo build --examples --all-features diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index cf3da4e..efc5513 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -27,7 +27,7 @@ jobs: exit 1 fi - - uses: dtolnay/rust-toolchain@stable + - uses: dtolnay/rust-toolchain@1.98.0 - uses: Swatinem/rust-cache@v2 diff --git a/CHANGELOG.md b/CHANGELOG.md index a98e5c5..a06ec3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Added +- `LoopMachine::set_context_tokens(tokens)` — a driver-fed context estimate. The driver measures the conversation (`count_context`, preferring the manager's counter) whenever history grows outside a model response — after `accept_input` at run start, so the machine's compaction trigger sees the true size of committed history plus the new input before the run's first model call instead of the zero `accept_input` resets to. Estimate-only: no state transition, no effect once terminal. +- `LoopMachine::compaction_noop(tokens_before, tokens_after)` — a second compaction feed method for passes that changed nothing (no compactor ran, a pre-compact hook vetoed, or the compactor returned the conversation unchanged). Unlike `compaction_result`, it leaves the committed history and the pending buffer untouched and adopts `tokens_after` as the current context size. Both compaction feeds compare the driver's measured pre-pass and post-pass token counts: when nothing was shaved off, the machine terminates the run with `ContextExceeded` instead of trusting an estimate recorded at compaction-request time (which could be stale — set before the last turn's tool results, or zero at run start). Alt drivers servicing `MachineStep::Compact` should pass both measurements from one counter. - `McpToolProvider::with_call_timeout(Duration)` — a per-call budget for every adapted tool's `tools/call` round-trip (default 60s). A call that exceeds it resolves to a *soft* error result naming the tool and the budget, so a wedged MCP server costs one tool result instead of hanging the agent loop indefinitely. Updates already-discovered tools and applies to later refreshes. Pinned by `call_timeout_cuts_a_slow_tool_with_a_soft_error` / `default_timeout_lets_a_quick_tool_through` in `tests/mcp_tool_provider.rs`. - MCP server adapter (`McpServerAdapter`) — serve a loopctl `ToolRegistry` over MCP (stdio), consumable by any MCP client. `McpServerAdapter::new(registry, ctx, name, version)` + `serve_stdio()` implements `ServerHandler` (`list_tools` → `all_schemas`, `call_tool` → `Tool::call`). Served calls are raced against the request's cancellation token: an already-cancelled request resolves to a cancelled result without invoking the tool, and a client cancel (`notifications/cancelled`) or disconnect drops the in-flight tool future and resolves to a cancelled tool-level result — a wedged tool no longer leaks a task per call (tools must be cancellation-safe, the same contract the engine's dispatch path imposes). `list_tools` forwards each tool's `is_read_only` as the MCP `annotations.readOnlyHint` (with `destructiveHint: false`); tools whose input schema does not compile as JSON Schema — malformed keywords, uncompilable regexes, dangling or external `$ref`s (external references are refused, never fetched) — or is not object-typed are omitted from the listing with a warning rather than advertised with a schema strict clients may reject (the `mcp` feature now pulls `jsonschema` for this check); unknown tool names return a `METHOD_NOT_FOUND` protocol error listing the registered names; empty descriptions are omitted rather than sent as `""`. Tool names are forwarded verbatim — the MCP spec recommends `^[a-zA-Z0-9_-]{1,64}$` and conforming names are the embedding application's responsibility (documented in the module docs). Transport-agnostic: `serve(impl IntoTransport)` works for future HTTP/SSE. The module promoted to `mcp/{convert,server}.rs` with symmetric inbound/outbound converters co-located. Example at `examples/mcp_server.rs` (echo + failing tool, Ctrl-C → graceful cancel; includes a piped JSON-RPC acceptance recipe). - MCP transports (stdio + Streamable HTTP/SSE) — `McpClient::stdio(command)` spawns an MCP server as a child process over stdio; `McpClient::http_sse(endpoint)` and `McpClient::http_sse_with_client(endpoint, reqwest::Client)` connect via the Streamable HTTP transport (rmcp handles `Mcp-Session-Id`, JSON-vs-SSE response splitting, and DELETE-on-close). `McpClient::reconnect(&StreamRetryConfig)` re-establishes a dropped connection using the crate's existing backoff strategy (the one retry strategy for the crate, not a second one). New public `CommandSpec` describes what to spawn. A stdio example server ships at `examples/mcp-stdio-server.rs` for transport testing. The `mcp` feature now enables `streaming` (for `StreamRetryConfig`) and `reqwest` (for the HTTP client); `default = []` is unchanged. @@ -20,6 +22,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Changed +- **Breaking:** the minimum supported Rust version is now **1.98** (was 1.94), riding the 0.3.0 minor bump. CI is anchored to `1.98.0` instead of floating `stable`, so toolchain releases no longer change lint verdicts mid-PR. Migration: build with rustc ≥ 1.98 (`rustup update stable`); older toolchains get a clear `cargo` error from the `rust-version` gate. +- `BareLoop` constructors now seed a default `ContextManager` (a `TruncatingCompactor` with window and threshold synced from the session config) when the supplied manager bundle carries none: `new`, `new_with_managers` (only when the bundle has no context manager), and `from_machine`. Behavior change: default-configured loops with `auto_compact` on now actually compact at the threshold (observer-visible via `on_compaction`) instead of growing unbounded; a compaction pass that cannot reduce the history terminates the run with a typed `LoopError::ContextExceeded` rather than silently resetting the estimate and sending over-window conversations. `auto_compact: false` still disables threshold compaction, and hosts installing their own `ContextManager` are unaffected. Pinned by `tests/compaction_noop.rs` (`over_limit_context_is_never_sent_to_the_provider`, `default_loop_compacts_at_the_threshold`, `host_installed_context_manager_is_unaffected`). +- `ConstrainedProfile::apply` attaches a `ContextManager` synced from the loop's session config (replacing whatever the constructor seeded), so the profile's context budgeting is enforced machinery rather than a marketing claim. Pinned by `small_model_profile_compacts_at_the_threshold` in `tests/compaction_noop.rs`. +- `ContextManager::compact_with_reason` (and `compact_manual` / `ensure_context_fits`) now report a successful pass that shrinks neither the message list nor the token count as `EnsureContextResult::NoAction` instead of `Compacted`. Classification and the returned token fields use the **manager's configured counter**, not the compactor's self-report: `tokens_after`/`tokens_saved` on the `Compacted` outcome are normalized to the manager's measurements, and `compact_with_reason`'s overflow check re-counts the result with the same counter (previously it trusted the compactor-reported value). Callers matching `Compacted` to learn "compaction occurred" no longer see no-action passes; `on_compaction` observers and post-compact hooks stay silent for them (the engine already skips both on `NoAction`). Migration: code that treated any `Ok(Compacted(..))` as "the messages may have changed" should use `into_messages()` — the returned list is identical under `NoAction`. +- **Breaking:** `LoopMachine::compaction_result` now takes `(compacted, tokens_before, tokens_after)` — the compacted history plus the driver's measured full-history size ahead of the pass and the compacted size after it, replacing the estimate the machine used to record at compaction-request time. The machine's `last_compaction_tokens` field is gone (its serialized state changes accordingly for checkpoints). Migration: pass the two measurements from the same counter the driver uses for its context estimate; the no-progress guard fails the run when `tokens_after >= tokens_before`. - Provider HTTP clients now set a **read timeout** (maximum gap between response bytes) instead of a total request timeout, and `with_timeout` configures that read timeout. A total HTTP-layer cap aborted every SSE stream longer than the configured duration (default 2 minutes) — pre-empting the `StreamHandler`'s per-event/total-stream deadlines and the engine's turn timeout, which own generation-length budgets. Long healthy streams now run as long as they keep producing bytes; a server silent for the configured gap (default 120s) is still aborted. Behavior change: generations longer than the old total cap no longer fail at the HTTP layer. - **Breaking:** Machine turn indices are now 0-indexed (`CallLLM { turn: 0 }` for the first turn). Previously 1-indexed (`turn: 1`). `AwaitingModel { turn }` and `AwaitingTools { turn }` follow the same convention. Callers matching on these variants in tests or drivers must adjust. @@ -42,6 +49,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Fixed +- The machine's context estimate is no longer zero at run start: the driver feeds `set_context_tokens(count_context(full_history))` right after `accept_input`, so the compaction trigger runs before a run's first model call. Previously a session whose *committed* history exceeded the context window (from prior runs, a host-seeded resume, or a `from_machine` checkpoint restore) sent its first request over-window before any check could run (pinned by `first_request_of_an_over_window_run_is_never_sent`, previously failing at `[192, 217]` against a 200-token window); such a run now compacts first or fails with a typed `LoopError::ContextExceeded` before any request. The same measurement now also runs after tool dispatch: the estimate is refreshed the moment tool results are appended (`set_context_tokens` after `tool_results`), so one turn's tool output — however large — is visible to the compaction trigger before the next request (pinned by `tool_result_growth_alone_crosses_the_threshold`). Landing that exposed a mock divergence: `MockApiClient` now streams tool-call arguments as an `InputJson` delta instead of embedding them in `PartStart`, matching how real providers stream — the accumulator never read PartStart-carried input, so mock-served tool input silently arrived empty. Note the accompanying policy change: a fresh input that itself crosses the threshold (or the 95% emergency line) now triggers compaction before its first request, where previously the check was blind until the first response. +- No-op compaction passes no longer report a hard-coded zero estimate or commit the in-flight run's pending messages. Previously the driver's no-manager, pre-compact-hook-veto, and `NoAction` paths returned `tokens_after = 0`, which blinded the machine's no-progress guard and reset its context estimate — the loop kept calling the provider with an over-window conversation until the turn budget was spent. Worse, feeding the uncompacted history back through `compaction_result` committed the current run's partial messages mid-run, so a later failure leaked the aborted run's prompt, tool calls, and results into committed history forever (`discard_pending` could no longer undo it). These paths now return the measured estimate (`count_context`) through an explicit no-op signal that leaves pending untouched; when compaction genuinely cannot reduce, the run terminates with a typed `LoopError::ContextExceeded`. Pinned by `failed_run_after_noop_compaction_leaves_history_clean` (engine) and `pre_compact_hook_veto_reports_measured_estimate` (hooks) — the hook-veto pass now surfaces the real size instead of zero. +- A no-change compaction pass is no longer reported as `EnsureContextResult::Compacted`: `ContextManager` used to wrap any successful compactor outcome as `Compacted`, so short-but-over-threshold conversations fired `on_compaction` observers (and post-compact hooks) on every triggering turn with zero savings and identical messages, violating both contracts. Same-list/zero-savings outcomes now map to `NoAction`. Pinned by `no_change_pass_is_not_reported_as_compacted` (compact.rs). - The non-streaming fallback forwards the turn's `RequestOptions` and is bounded by the total-stream deadline (`StreamHandler::fallback_non_streaming`): it previously called `create_message` (dropping any configured `response_format`/`tool_constraint`) and raced only the cancel signal, so a hanging fallback request hung the turn. Pinned by `fallback_non_streaming_forwards_request_options`, `fallback_non_streaming_honors_the_total_deadline`, and `completed_fallback_response_racing_the_deadline_is_accepted` (the deadline bounds waiting, not completion — a resolved response is accepted even past expiry, while a hanging one is still cut). - MCP `tools/call` round-trips are bounded by the new per-call timeout (see `McpToolProvider::with_call_timeout` above) — previously a wedged server hung the agent loop with cancellation as the only exit. - Tool-result parts in a `CallTools` turn preserve **model request order** across preresolved (unknown-tool) and dispatched calls. Previously the turn's results were assembled as `[all preresolved, then all dispatched]`, which reordered the parts the model saw relative to the calls it made. Provider-safe in practice (providers match by `tool_call_id`, not position), but order-non-preserving and surprising to hosts that assume positional alignment. Pinned by `test_mixed_known_unknown_tools_preserve_request_order`. diff --git a/Cargo.toml b/Cargo.toml index 61c5091..0cd1f2c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ documentation = "https://docs.rs/loopctl" keywords = ["agent", "framework", "llm", "loop"] categories = ["api-bindings", "development-tools"] readme = "README.md" -rust-version = "1.94" +rust-version = "1.98" [lib] name = "loopctl" diff --git a/examples/mcp-adapter.rs b/examples/mcp-adapter.rs index 7c74d1b..9822d99 100644 --- a/examples/mcp-adapter.rs +++ b/examples/mcp-adapter.rs @@ -44,7 +44,7 @@ impl GreetServer { "hello, world!".to_string() } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for GreetServer {} diff --git a/examples/mcp-stdio-server.rs b/examples/mcp-stdio-server.rs index ae7742a..8c0ed73 100644 --- a/examples/mcp-stdio-server.rs +++ b/examples/mcp-stdio-server.rs @@ -42,7 +42,7 @@ impl StdioServer { "hello from stdio".to_string() } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for StdioServer {} diff --git a/src/compact.rs b/src/compact.rs index dfbacf4..e131161 100644 --- a/src/compact.rs +++ b/src/compact.rs @@ -668,13 +668,27 @@ impl ContextManager { }); } - Ok(EnsureContextResult::Compacted(outcome)) + if outcome.messages.len() == message_count && tokens_after >= tokens_before { + return Ok(EnsureContextResult::NoAction(outcome.messages)); + } + let mut normalized = outcome; + normalized.tokens_after = tokens_after; + normalized.tokens_saved = tokens_before.saturating_sub(tokens_after); + Ok(EnsureContextResult::Compacted(normalized)) } /// Manually trigger compaction regardless of threshold. /// /// Use this when the agent or a tool explicitly requests context - /// reduction. The `reason` is set to [`CompactReason::Manual`]. + /// reduction. The `reason` is set to [`CompactReason::Manual`]. A + /// successful pass is classified with this manager's configured + /// [`TokenCounter`], not the compactor's self-report: when the message + /// count is unchanged and the measured token count did not shrink, the + /// pass is reported as [`EnsureContextResult::NoAction`] — compaction + /// did not occur, so compaction observers and hooks stay silent. On the + /// [`EnsureContextResult::Compacted`] path the outcome's token fields + /// are normalized to the same counter, so telemetry reflects the + /// manager's measurements. /// # Errors /// /// Returns a [`ContextOverflow`] if compaction fails or the result @@ -693,7 +707,15 @@ impl ContextManager { /// Use this when the decision to compact was already made (e.g. by the /// driving state machine) and the `reason` should reach the compactor's /// [`CompactionContext`] so it can vary strategy (e.g. more aggressive - /// summarization for [`CompactReason::Emergency`]). + /// summarization for [`CompactReason::Emergency`]). A successful pass + /// is classified with this manager's configured [`TokenCounter`], not + /// the compactor's self-report: when the message count is unchanged and + /// the measured token count did not shrink, the pass is reported as + /// [`EnsureContextResult::NoAction`] — compaction did not occur, so + /// compaction observers and hooks stay silent. On the + /// [`EnsureContextResult::Compacted`] path the outcome's token fields + /// are normalized to the same counter, so telemetry reflects the + /// manager's measurements. /// /// # Errors /// @@ -735,9 +757,10 @@ impl ContextManager { }); } - if outcome.tokens_after > self.context_window { + let tokens_after = self.estimate_tokens(&outcome.messages); + if tokens_after > self.context_window { return Err(ContextOverflow { - tokens_used: outcome.tokens_after, + tokens_used: tokens_after, context_window: self.context_window, message_count, trigger: CompactReason::Manual, @@ -745,7 +768,13 @@ impl ContextManager { }); } - Ok(EnsureContextResult::Compacted(outcome)) + if outcome.messages.len() == message_count && tokens_after >= tokens_before { + return Ok(EnsureContextResult::NoAction(outcome.messages)); + } + let mut normalized = outcome; + normalized.tokens_after = tokens_after; + normalized.tokens_saved = tokens_before.saturating_sub(tokens_after); + Ok(EnsureContextResult::Compacted(normalized)) } /// Build telemetry for a compaction operation. @@ -1237,4 +1266,25 @@ mod tests { let outcome = CompactionOutcome::compacted(msgs, 100, 500); assert_eq!(outcome.tokens_saved, 0); } + + #[tokio::test] + async fn no_change_pass_is_not_reported_as_compacted() { + use crate::compact::truncating::TruncatingCompactor; + use crate::message::Message; + + let manager = ContextManager::new(std::sync::Arc::new( + TruncatingCompactor::new().with_min_messages(10), + )); + let messages: Vec = (0..3).map(|i| Message::user(format!("msg {i}"))).collect(); + let result = manager + .compact_manual(messages.clone(), 1) + .await + .expect("manual compaction must not error"); + if let EnsureContextResult::Compacted(outcome) = result { + assert!( + outcome.messages.len() < messages.len() || outcome.tokens_saved > 0, + "doc: Compacted means compaction occurred and produced a shorter message list — got identical list with zero savings" + ); + } + } } diff --git a/src/engine/bare.rs b/src/engine/bare.rs index 7647e17..28c97b6 100644 --- a/src/engine/bare.rs +++ b/src/engine/bare.rs @@ -61,7 +61,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use crate::cancel::CancelSignal; -use crate::compact::ContextManager; +use crate::compact::{ContextManager, TruncatingCompactor}; use crate::config::SessionConfig; use crate::engine::core::{ LoopMachine, MachineOutcome, MachinePolicy, MachineState, MachineStep, ModelResponse, @@ -353,8 +353,10 @@ impl BareLoop { /// Create a new `BareLoop` with the given components. /// - /// Initializes an empty conversation history and a - /// fresh [`LoopManagers`]. The cancellation signal starts as non-cancelled. + /// Initializes an empty conversation history and a fresh + /// [`LoopManagers`] seeded with a default [`ContextManager`] synced from + /// the session config (see [`Self::new_with_managers`]). The cancellation + /// signal starts as non-cancelled. /// /// # Parameters /// @@ -381,6 +383,13 @@ impl BareLoop { /// [`LoopManagers`] — for example, to enable loop detection or /// circuit-breaker policies. /// + /// When the supplied bundle carries no [`ContextManager`], a default one + /// (a [`TruncatingCompactor`](crate::compact::TruncatingCompactor)) is + /// installed with its context window and threshold synced from + /// `session_config`, so the session's auto-compaction settings are backed + /// by machinery that can actually reduce the history. A bundle that + /// already carries a context manager is used as-is. + /// /// # Parameters /// /// - `client` — The LLM API client, wrapped in `Arc`. @@ -406,8 +415,12 @@ impl BareLoop { client: Arc, tools: ToolRegistry, session_config: SessionConfig, - managers: LoopManagers, + mut managers: LoopManagers, ) -> Self { + if managers.context_manager().is_none() { + let seeded = Self::default_context_manager(&session_config); + managers.set_context_manager(Arc::new(seeded)); + } Self { client, tools: Arc::new(tools), @@ -572,8 +585,13 @@ impl BareLoop { /// Constructs a [`BareLoop`] whose machine is `machine` — for example to /// resume a serialized run: deserialize the machine, wrap it in a loop with /// the original client/tools, and continue driving it with - /// [`run()`](crate::engine::core::Loop::run). The session/run config - /// is taken from the machine. + /// [`run()`](crate::engine::core::Loop::run). The supplied + /// `session_config` configures the resumed loop; per-run settings come + /// from the [`RunConfig`](crate::engine::RunConfig) the caller passes to + /// each `run()` call — the machine itself stores no configuration. A + /// fresh [`LoopManagers`] is created with a default + /// [`ContextManager`] synced from `session_config` (see + /// [`Self::new_with_managers`]). #[must_use] pub fn from_machine( machine: LoopMachine, @@ -581,12 +599,15 @@ impl BareLoop { client: Arc, tools: ToolRegistry, ) -> Self { + let mut managers = LoopManagers::new(); + let seeded = Self::default_context_manager(&session_config); + managers.set_context_manager(Arc::new(seeded)); Self { client, tools: Arc::new(tools), session: Session::new(session_config), machine, - managers: LoopManagers::new(), + managers, reflector: Arc::new(NoopReflector), recovery: Arc::new(ExponentialBackoffRecovery::new(3)), cancelled: Arc::new(CancelSignal::new()), @@ -599,6 +620,21 @@ impl BareLoop { } } + /// Build the default context manager for a session config. + /// + /// A [`TruncatingCompactor`] behind a [`ContextManager`] whose context + /// window and threshold mirror `session_config`'s. Installed by + /// [`Self::new_with_managers`] and [`Self::from_machine`] when the + /// manager bundle carries no compaction machinery of its own, so the + /// session's auto-compaction trigger is never an alarm without a + /// sprinkler. Hosts that want different behavior install their own + /// manager, which is never overridden. + fn default_context_manager(session_config: &SessionConfig) -> ContextManager { + ContextManager::new(Arc::new(TruncatingCompactor::default())) + .with_context_window(session_config.context_window) + .with_threshold(session_config.compact_threshold) + } + /// Get the tool registry. /// /// Returns a reference to the [`ToolRegistry`] containing all tools @@ -1143,6 +1179,8 @@ impl BareLoop { Role::User, slots.into_iter().flatten().collect(), )]); + let estimate = self.count_context(&self.machine.full_history()); + self.machine.set_context_tokens(estimate); Ok(()) } @@ -1150,7 +1188,14 @@ impl BareLoop { /// /// Runs the configured [`ContextManager`](crate::compact::ContextManager) /// over the machine-owned history (firing `on_compaction` and hooks), then - /// feeds the compacted history back to the machine. + /// feeds the outcome back to the machine: a rewritten history with the + /// driver's measured before/after token sizes through + /// [`LoopMachine::compaction_result`](crate::engine::core::LoopMachine::compaction_result), + /// an unchanged one with the same measurements through + /// [`LoopMachine::compaction_noop`](crate::engine::core::LoopMachine::compaction_noop) + /// so the pending buffer survives. The machine already sits in + /// `AwaitingCompaction` for this reason when the step arrives; the driver + /// only performs the I/O and feeds the result back. /// /// # Errors /// @@ -1161,10 +1206,20 @@ impl BareLoop { reason: crate::compact::types::CompactReason, ) -> Result<(), LoopError> { let turn = self.machine.turns_taken(); - // The machine is already `AwaitingCompaction` for this reason; the - // driver just performs the IO and feeds the result back. - let (compacted, tokens_after) = self.run_compaction(turn, reason).await?; - self.machine.compaction_result(compacted, tokens_after); + let outcome = self.run_compaction(turn, reason).await?; + match outcome.compacted { + Some(compacted) => { + self.machine.compaction_result( + compacted, + outcome.tokens_before, + outcome.tokens_after, + ); + } + None => { + self.machine + .compaction_noop(outcome.tokens_before, outcome.tokens_after); + } + } Ok(()) } } @@ -1189,6 +1244,8 @@ impl crate::engine::core::Loop for BareLoop { self.session.runs.push(Run::new(input, run_config)); self.notify_run_start(); self.machine.accept_input(input); + let estimate = self.count_context(&self.machine.full_history()); + self.machine.set_context_tokens(estimate); loop { let policy = self.machine_policy(); diff --git a/src/engine/bare/compact.rs b/src/engine/bare/compact.rs index 9c94c6d..539fc1c 100644 --- a/src/engine/bare/compact.rs +++ b/src/engine/bare/compact.rs @@ -19,24 +19,61 @@ use crate::capabilities::Hookable; use crate::message::Message; use crate::observer::CompactedContext; +/// The result of servicing a [`MachineStep::Compact`](crate::engine::core::MachineStep::Compact), +/// before it is fed back into the machine. +/// +/// Carries the driver's token measurements of the conversation, both taken +/// with the same counter: ahead of the pass (`tokens_before`) and after it +/// (`tokens_after`). `compacted` decides which machine feed applies: a +/// rewritten history replaces the committed one and clears the pending +/// buffer +/// ([`LoopMachine::compaction_result`](crate::engine::core::LoopMachine::compaction_result)), +/// while an unchanged history must leave both alone +/// ([`LoopMachine::compaction_noop`](crate::engine::core::LoopMachine::compaction_noop)) +/// — committing the in-flight run's partial messages mid-run would make a +/// later failure un-discardable. +pub(super) struct CompactStepOutcome { + /// Measured token size of the full history ahead of the compaction pass. + /// + /// Taken by the driver before the compactor runs, so the machine's + /// no-progress guard compares two real measurements of the same + /// conversation rather than estimates recorded at different times. + pub(super) tokens_before: u64, + + /// Measured token size of the conversation after the pass. + /// + /// The size of the compacted list when `compacted` is `Some`, or of the + /// unchanged history otherwise. Equal to `tokens_before` whenever the + /// pass changed nothing. + pub(super) tokens_after: u64, + + /// The compacted message list, when the pass rewrote the history. + /// + /// `None` when compaction changed nothing — no compactor ran, a + /// pre-compact hook vetoed the pass, or the manager reported no action. + pub(super) compacted: Option>, +} + impl BareLoop { /// Compact the conversation context owned by the driving machine. /// /// Run when the machine requests it via /// [`MachineStep::Compact`](crate::engine::core::MachineStep::Compact). /// Reads the history from [`LoopMachine::history`](crate::engine::core::LoopMachine::history), - /// asks the configured [`ContextManager`](crate::compact::ContextManager) - /// to reduce it, fires + /// measures its token size, asks the configured + /// [`ContextManager`](crate::compact::ContextManager) to reduce it, fires /// [`on_compaction`](crate::observer::LoopObserver::on_compaction) and the - /// post-compact hook when compaction occurred, and returns the compacted - /// messages alongside the post-compaction token estimate. The caller feeds - /// both back via - /// [`LoopMachine::compaction_result`](crate::engine::core::LoopMachine::compaction_result). + /// post-compact hook when compaction occurred, and returns a + /// [`CompactStepOutcome`] — the measured before/after pair plus the + /// compacted list when one was produced — for the driver to feed back + /// into the machine. /// - /// When no `ContextManager` is set the history is returned unchanged with a - /// zero token estimate, so the machine can resume without compaction. When - /// a pre-compact hook aborts compaction the history is likewise returned - /// unchanged. + /// When no `ContextManager` is set, a pre-compact hook aborts the pass, + /// or the manager reports [`EnsureContextResult::NoAction`], the + /// conversation is returned unchanged with measured before/after sizes — + /// never a hard-coded zero — so the machine's no-progress guard compares + /// real measurements and terminates with a typed error when compaction + /// cannot reduce instead of silently looping. /// /// # Errors /// @@ -47,15 +84,24 @@ impl BareLoop { &mut self, turn: usize, reason: crate::compact::types::CompactReason, - ) -> Result<(Vec, u64), LoopError> { + ) -> Result { let history = self.machine.full_history(); + let tokens_before = self.count_context(&history); let Some(ctx_manager) = self.managers.context_manager() else { - return Ok((history, 0)); + return Ok(CompactStepOutcome { + tokens_before, + tokens_after: tokens_before, + compacted: None, + }); }; #[cfg(feature = "hooks")] if self.pre_compact_hook_aborts(&history) { - return Ok((history, 0)); + return Ok(CompactStepOutcome { + tokens_before, + tokens_after: tokens_before, + compacted: None, + }); } #[cfg(feature = "hooks")] @@ -67,10 +113,9 @@ impl BareLoop { match result { Ok(EnsureContextResult::Compacted(outcome)) => { let tokens_after = outcome.tokens_after; - let tokens_saved = outcome.tokens_saved; + let tokens_saved = tokens_before.saturating_sub(tokens_after); #[cfg(feature = "hooks")] let messages_after = outcome.messages.len(); - let tokens_before = tokens_after.saturating_add(tokens_saved); self.managers.observers().on_compaction(&CompactedContext { tokens_before, tokens_after, @@ -84,9 +129,20 @@ impl BareLoop { tokens_saved, compact_start.elapsed(), ); - Ok((outcome.messages, tokens_after)) + Ok(CompactStepOutcome { + tokens_before, + tokens_after, + compacted: Some(outcome.messages), + }) + } + Ok(EnsureContextResult::NoAction(messages)) => { + let tokens_after = self.count_context(&messages); + Ok(CompactStepOutcome { + tokens_before, + tokens_after, + compacted: None, + }) } - Ok(EnsureContextResult::NoAction(messages)) => Ok((messages, 0)), Err(overflow) => Err(LoopError::ContextExceeded { used: overflow.tokens_used, limit: overflow.context_window, diff --git a/src/engine/bare/tests.rs b/src/engine/bare/tests.rs index 5b49267..82c2150 100644 --- a/src/engine/bare/tests.rs +++ b/src/engine/bare/tests.rs @@ -1244,12 +1244,16 @@ async fn compaction_sees_pending_messages() { let config = make_config() .with_context_window(100) - .with_compact_threshold(10); + .with_compact_threshold(20); let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), config); agent.set_context_manager(Arc::new( - crate::compact::ContextManager::new(Arc::new(crate::compact::TruncatingCompactor::new())) - .with_context_window(100) - .with_threshold(10), + crate::compact::ContextManager::new(Arc::new( + crate::compact::TruncatingCompactor::new() + .with_preserve_recent(1) + .with_min_messages(2), + )) + .with_context_window(100) + .with_threshold(20), )); agent @@ -1306,7 +1310,10 @@ async fn context_token_count_includes_model_response_message() { let counter_clone = Arc::clone(&token_ctr); let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), make_config()); - agent.set_token_counter(counter_clone); + agent.set_context_manager(Arc::new( + crate::compact::ContextManager::new(Arc::new(crate::compact::TruncatingCompactor::new())) + .with_token_counter(counter_clone), + )); agent.run("hi", &RunConfig::default()).await.unwrap(); @@ -1364,12 +1371,12 @@ async fn compaction_then_failure_leaves_history_compacted() { let config = make_config() .with_context_window(100) - .with_compact_threshold(10); + .with_compact_threshold(80); let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), config); agent.set_context_manager(Arc::new( crate::compact::ContextManager::new(Arc::new(crate::compact::TruncatingCompactor::new())) .with_context_window(100) - .with_threshold(10), + .with_threshold(80), )); agent.run("first run", &RunConfig::default()).await.unwrap(); @@ -4805,3 +4812,32 @@ fn fluent_with_observer_equivalent_to_register_observer() { "both paths register the same number of observers" ); } + +#[tokio::test] +async fn failed_run_after_noop_compaction_leaves_history_clean() { + let client = MockClient::new("test-model"); + client.add_tool_only_response("call_1", "echo", json!({"message": "hi"})); + + let mut config = make_config(); + config.context_window = 100; + config.compact_threshold = 50; + config.auto_compact = true; + + let mut registry = ToolRegistry::new(); + registry.register(EchoTool); + let mut agent = BareLoop::new(Arc::new(client), registry, config); + + let result = agent.run(&"x".repeat(300), &make_run_config()).await; + match result { + Err(LoopError::ContextExceeded { .. }) => {} + other => panic!( + "mock has no response for the post-compact call; an unshrinkable over-threshold \ + conversation must fail with ContextExceeded instead, got {other:?}" + ), + } + assert!( + agent.conversation().is_empty(), + "discard_pending doc: no messages from the abandoned run may leak into the next run's context; got {} messages", + agent.conversation().len() + ); +} diff --git a/src/engine/core/machine.rs b/src/engine/core/machine.rs index 1f2bb5a..3f1ed0b 100644 --- a/src/engine/core/machine.rs +++ b/src/engine/core/machine.rs @@ -333,7 +333,8 @@ pub struct MachinePolicy { /// machine.tool_results(Vec::new()); /// } /// MachineStep::Compact { .. } => { -/// machine.compaction_result(machine.history().to_vec(), 0); +/// let measured_before = 80; +/// machine.compaction_result(machine.history().to_vec(), measured_before, 60); /// } /// MachineStep::Done(MachineOutcome::Completed { final_text }) => { /// assert_eq!(final_text, "Hi there"); @@ -388,17 +389,6 @@ pub struct LoopMachine { /// policy to decide whether to emit a [`MachineStep::Compact`]. context_tokens: u64, - /// Context tokens at the time of the last compaction request. - /// - /// When the machine emits `Compact`, it records `context_tokens` - /// here. After `compaction_result`, if `tokens_after` has not - /// decreased below this value, the machine transitions to - /// [`MachineOutcome::Failed`] with - /// [`LoopError::ContextExceeded`] instead of requesting another - /// compaction — preventing an infinite compaction cycle when the - /// compactor cannot reduce the context. - last_compaction_tokens: Option, - /// Whether [`Self::cancel`] has been called. /// /// A cancellation request from the driver. Once set, the next @@ -433,7 +423,6 @@ impl LoopMachine { state: MachineState::Start, turns_taken: 0, context_tokens: 0, - last_compaction_tokens: None, cancelled: false, pending_tools: Vec::new(), } @@ -460,6 +449,23 @@ impl LoopMachine { self.pending_tools.clear(); } + /// Feed a driver-measured context estimate into the machine. + /// + /// The driver calls this whenever history grows outside a model response + /// — after [`Self::accept_input`] at run start, so the first + /// [`Self::next_step`] sees the true size of the committed history plus + /// the new input instead of the zero `accept_input` reset to, and after + /// any other append the machine would otherwise not learn about until the + /// next response. Replaces the current estimate with `tokens`; performs + /// no state transition, so the compaction trigger evaluates it on the + /// next [`Self::next_step`]. No effect once the machine is terminal. + pub fn set_context_tokens(&mut self, tokens: u64) { + if self.is_terminal() { + return; + } + self.context_tokens = tokens; + } + /// Return the next step the driver must perform. /// /// `policy` supplies the turn budget and compaction thresholds the machine @@ -468,7 +474,8 @@ impl LoopMachine { /// /// This is pure and idempotent: calling it twice with no intervening feed /// method ([`Self::model_response`], [`Self::tool_results`], - /// [`Self::compaction_result`], [`Self::cancel`]) returns an equal step. + /// [`Self::compaction_result`], [`Self::compaction_noop`], + /// [`Self::cancel`]) returns an equal step. /// Once the machine is terminal, every subsequent call returns /// [`MachineStep::Done`] with the same [`MachineOutcome`]. /// @@ -520,13 +527,11 @@ impl LoopMachine { } if self.is_emergency(policy) { let reason = CompactReason::Emergency; - self.last_compaction_tokens = Some(self.context_tokens); self.state = MachineState::AwaitingCompaction { reason }; return MachineStep::Compact { reason }; } if policy.auto_compact && self.should_compact(policy) { let reason = CompactReason::ThresholdExceeded; - self.last_compaction_tokens = Some(self.context_tokens); self.state = MachineState::AwaitingCompaction { reason }; return MachineStep::Compact { reason }; } @@ -646,35 +651,83 @@ impl LoopMachine { /// Feed compacted history back into the machine. /// - /// The driver calls this after servicing a [`MachineStep::Compact`], passing - /// the compacted history and `tokens_after` — its estimate of that history's - /// token size, which the machine adopts as the current context size so it - /// does not immediately request another compaction. The next - /// [`Self::next_step`] then requests the deferred [`MachineStep::CallLLM`]. - /// Has no effect once the machine is terminal. - pub fn compaction_result(&mut self, compacted: Vec, tokens_after: u64) { + /// The driver calls this after servicing a [`MachineStep::Compact`] that + /// rewrote the history, passing the compacted history plus two measured + /// estimates from the same token counter: `tokens_before` — the size of + /// the full history ahead of the compaction pass — and `tokens_after` — + /// the size of `compacted`. The machine adopts `tokens_after` as the + /// current context size so it does not immediately request another + /// compaction. The next [`Self::next_step`] then requests the deferred + /// [`MachineStep::CallLLM`]. Has no effect once the machine is terminal. + pub fn compaction_result( + &mut self, + compacted: Vec, + tokens_before: u64, + tokens_after: u64, + ) { if self.is_terminal() { return; } - if let Some(before) = self.last_compaction_tokens - && tokens_after >= before - { - self.state = MachineState::Terminal(MachineOutcome::Failed { - error: LoopError::ContextExceeded { - used: tokens_after, - limit: before, - }, - }); - self.last_compaction_tokens = None; + if self.terminate_on_no_progress(tokens_before, tokens_after) { return; } - self.last_compaction_tokens = None; self.history = compacted; self.pending.clear(); self.context_tokens = tokens_after; self.state = MachineState::Start; } + /// Feed an unchanged compaction result back into the machine. + /// + /// The driver calls this after servicing a [`MachineStep::Compact`] that + /// changed nothing — no compactor ran, a pre-compact hook vetoed the pass, + /// or the compactor returned the conversation unchanged. The committed + /// history and the pending buffer are left untouched: feeding the + /// uncompacted conversation through [`Self::compaction_result`] would + /// commit the current run's partial messages mid-run, so a later failure + /// could no longer discard them. + /// + /// `tokens_before` and `tokens_after` are the driver's measured estimates + /// of the conversation ahead of and after the pass (equal in practice, + /// since nothing changed); the machine adopts `tokens_after` as the + /// current context size. The same no-progress guard as + /// [`Self::compaction_result`] applies: when nothing was shaved off, the + /// machine transitions to [`MachineOutcome::Failed`] with + /// [`LoopError::ContextExceeded`] — compaction cannot shrink this + /// conversation, and another model call would exceed the context window. + /// Has no effect once the machine is terminal. + pub fn compaction_noop(&mut self, tokens_before: u64, tokens_after: u64) { + if self.is_terminal() { + return; + } + if self.terminate_on_no_progress(tokens_before, tokens_after) { + return; + } + self.context_tokens = tokens_after; + self.state = MachineState::Start; + } + + /// Fail the run when a compaction feed made no progress. + /// + /// Shared guard behind [`Self::compaction_result`] and + /// [`Self::compaction_noop`]: compares the driver's measured post-pass + /// token count against its measured pre-pass count of the full history. + /// When nothing was shaved off, transitions to [`MachineOutcome::Failed`] + /// with [`LoopError::ContextExceeded`] (preventing an infinite compaction + /// cycle) and returns `true`; returns `false` when the feed may proceed. + fn terminate_on_no_progress(&mut self, tokens_before: u64, tokens_after: u64) -> bool { + if tokens_after < tokens_before { + return false; + } + self.state = MachineState::Terminal(MachineOutcome::Failed { + error: LoopError::ContextExceeded { + used: tokens_after, + limit: tokens_before, + }, + }); + true + } + /// Mark the run as cancelled. /// /// The next [`Self::next_step`] returns [`MachineStep::Done`] with @@ -878,6 +931,10 @@ mod tests { crate::compact::HeuristicTokenCounter.count(&machine.full_history()) } + fn count_vec_tokens(messages: &[Message]) -> u64 { + crate::compact::CompactionOutcome::estimate_tokens(messages) + } + fn same_step(a: &MachineStep, b: &MachineStep) -> bool { serde_json::to_string(a).unwrap_or_default() == serde_json::to_string(b).unwrap_or_default() } @@ -973,7 +1030,9 @@ mod tests { machine.next_step(policy), MachineStep::Compact { .. } )); - machine.compaction_result(vec![Message::user("compacted")], 0); + let compacted = vec![Message::user("compacted")]; + let tokens_after = count_vec_tokens(&compacted); + machine.compaction_result(compacted, count_tokens(&machine), tokens_after); let snapshot = serde_json::to_string(&machine).expect("serialize"); let mut restored: LoopMachine = serde_json::from_str(&snapshot).expect("deserialize"); let a = machine.next_step(policy); @@ -1167,7 +1226,8 @@ mod tests { MachineStep::Compact { .. } )); let compacted = vec![Message::user("compacted-only")]; - machine.compaction_result(compacted.clone(), 0); + let tokens_after = count_vec_tokens(&compacted); + machine.compaction_result(compacted.clone(), count_tokens(&machine), tokens_after); // Compare by serialized form: Message is not PartialEq. let got = serde_json::to_string(&machine.full_history()).expect("serialize history"); let want = serde_json::to_string(&compacted).expect("serialize expected"); @@ -1219,7 +1279,9 @@ mod tests { machine.next_step(policy), MachineStep::Compact { .. } )); - machine.compaction_result(vec![Message::user("compacted")], 0); + let compacted = vec![Message::user("compacted")]; + let tokens_after = count_vec_tokens(&compacted); + machine.compaction_result(compacted, count_tokens(&machine), tokens_after); assert_eq!( machine.history().len(), @@ -1250,7 +1312,12 @@ mod tests { machine.next_step(policy), MachineStep::Compact { .. } )); - machine.compaction_result(vec![Message::user("compacted")], 90); + let tokens_before = count_tokens(&machine); + machine.compaction_result( + vec![Message::user("compacted")], + tokens_before, + tokens_before, + ); match machine.next_step(policy) { MachineStep::Done(MachineOutcome::Failed { @@ -1279,7 +1346,8 @@ mod tests { machine.next_step(policy), MachineStep::Compact { .. } )); - machine.compaction_result(vec![Message::user("compacted")], 30); + let tokens_before = count_tokens(&machine); + machine.compaction_result(vec![Message::user("compacted")], tokens_before, 30); assert!( matches!(machine.next_step(policy), MachineStep::CallLLM { .. }), diff --git a/src/error.rs b/src/error.rs index 2f66796..39e36fd 100644 --- a/src/error.rs +++ b/src/error.rs @@ -203,12 +203,15 @@ pub enum LoopError { /// with `limit` to report utilization to the caller. used: u64, - /// Maximum tokens allowed by the model. + /// The token budget that was exceeded. /// - /// Sourced from - /// [`SessionConfig::context_window`](crate::config::SessionConfig::context_window). - /// When `used` exceeds this after compaction, the run cannot - /// continue on the current model. + /// One of two values, depending on where the error originates: + /// the model's context window (from + /// [`SessionConfig::context_window`](crate::config::SessionConfig::context_window)) + /// when a compaction result still does not fit, or the measured + /// pre-compaction size when the no-progress guard ends a run + /// because compaction could not reduce the conversation. Pair + /// with `used` to report utilization to the caller. limit: u64, }, diff --git a/src/mcp/server.rs b/src/mcp/server.rs index 8cf40a0..5fc0644 100644 --- a/src/mcp/server.rs +++ b/src/mcp/server.rs @@ -42,6 +42,7 @@ //! handshake in the example's module doc exercises the same code path with no //! client at all. +use std::future::Future; use std::sync::Arc; use rmcp::ErrorData; @@ -180,14 +181,16 @@ impl McpServerAdapter { /// /// # Errors /// - /// Returns `rmcp::service::ServerInitializeError` if the handshake fails - /// (e.g. the client sends a malformed `initialize`). + /// Returns a boxed `rmcp::service::ServerInitializeError` if the handshake + /// fails (e.g. the client sends a malformed `initialize`). pub async fn serve_stdio( self, - ) -> Result, rmcp::service::ServerInitializeError> - { + ) -> Result< + rmcp::service::RunningService, + Box, + > { let transport = rmcp::transport::io::stdio(); - self.serve(transport).await + self.serve(transport).await.map_err(Box::new) } } @@ -250,11 +253,13 @@ impl ServerHandler for McpServerAdapter { /// /// Never returns `Err` — listing local schemas cannot fail — but the /// [`ServerHandler`] signature is `Result`, so the error type remains. - async fn list_tools( + /// The listing is built synchronously; the returned future is immediately + /// ready. + fn list_tools( &self, _request: Option, _context: RequestContext, - ) -> Result { + ) -> impl Future> { let tools = self .registry .all_schemas() @@ -267,10 +272,10 @@ impl ServerHandler for McpServerAdapter { convert::tool_schema_to_mcp(schema, is_read_only) }) .collect(); - Ok(rmcp::model::ListToolsResult { + std::future::ready(Ok(rmcp::model::ListToolsResult { tools, ..Default::default() - }) + })) } /// `tools/call` → registry lookup → [`Tool::call`]. diff --git a/src/presets.rs b/src/presets.rs index 4e05793..1cf1471 100644 --- a/src/presets.rs +++ b/src/presets.rs @@ -7,8 +7,9 @@ //! [`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::session_config`], +//! Apply a profile with [`ConstrainedProfile::apply`] (context manager + +//! pipeline + contributor) or compose the individual pieces +//! ([`ConstrainedProfile::session_config`], //! [`ConstrainedProfile::run_config`], //! [`ConstrainedProfile::pipeline_builder`], //! [`ConstrainedProfile::request_options`]) by hand. @@ -54,12 +55,15 @@ const MEMOIZED_TOOLS: &[&str] = &["Read", "Glob", "Grep", "LS"]; /// 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. +/// Bundles context-budget machinery (a context manager that compacts the +/// conversation at the session's configured threshold), a smaller window and +/// fewer turns via [`session_config`](Self::session_config) / +/// [`run_config`](Self::run_config), 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 @@ -137,10 +141,15 @@ impl ConstrainedProfile { RequestOptions::new().with_tool_constraint(ToolConstraint::Strict) } - /// Apply the profile's pipeline and goal-reminder contributor to a - /// [`BareLoop`]. + /// Apply the profile's compaction machinery, pipeline, and goal-reminder + /// contributor to a [`BareLoop`]. /// - /// Sets the small-model middleware stack (via [`Self::pipeline_builder`]) and + /// Installs a [`ContextManager`](crate::compact::ContextManager) around a + /// [`TruncatingCompactor`](crate::compact::TruncatingCompactor) with its + /// window and threshold synced from the loop's session config, replacing + /// whatever the constructor seeded, so the profile's context budgeting is + /// enforced by machinery rather than left to the caller. Also 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 @@ -163,6 +172,12 @@ impl ConstrainedProfile { /// ConstrainedProfile::apply(&mut agent).unwrap(); /// ``` pub fn apply(loop_: &mut BareLoop) -> Result<(), LoopError> { + let manager = crate::compact::ContextManager::new(Arc::new( + crate::compact::TruncatingCompactor::default(), + )) + .with_context_window(loop_.session_config().context_window) + .with_threshold(loop_.session_config().compact_threshold); + loop_.set_context_manager(Arc::new(manager)); loop_.set_pipeline(Self::pipeline_builder())?; loop_.add_contributor(Box::new(GoalReminder::new(GOAL_REMINDER_EVERY_N_TURNS))); Ok(()) diff --git a/src/testing.rs b/src/testing.rs index c5d1422..2ab31c8 100644 --- a/src/testing.rs +++ b/src/testing.rs @@ -607,7 +607,8 @@ impl ApiClient for MockApiClient { /// /// ```text /// MessageStart → PartStart(text) → IndexedDelta(text) → MessagePartStop - /// → [PartStart(tool_use) → MessagePartStop] (if tool_call is set) + /// → [PartStart(tool_use) → IndexedDelta(input_json) + /// → MessagePartStop] (if tool_call is set) /// → MessageDelta(stop_reason, usage) → MessageStop /// ``` /// @@ -674,11 +675,24 @@ impl ApiClient for MockApiClient { events.push(Ok(StreamEvent::PartStop)); - // Tool call content block (if any) + // Tool call content block (if any). Real providers open the tool + // block with an empty input and stream the arguments as input-json + // deltas; the mock mirrors that shape so accumulated tool input is + // identical on the mock and real paths. if let Some(tc) = &response.tool_call { events.push(Ok(StreamEvent::PartStart(PartStart { index: 1, - part: Some(MessagePart::tool_call(&tc.id, &tc.name, tc.input.clone())), + part: Some(MessagePart::tool_call( + &tc.id, + &tc.name, + serde_json::json!({}), + )), + }))); + events.push(Ok(StreamEvent::IndexedDelta(IndexedDelta { + index: 1, + delta: DeltaPart::InputJson { + partial_json: tc.input.to_string(), + }, }))); events.push(Ok(StreamEvent::PartStop)); } @@ -1445,6 +1459,27 @@ mod tests { } }); assert!(has_tool_stop); + + let mut accumulator = crate::stream::StreamAccumulator::new(); + for event in &events { + let Ok(event) = event else { + panic!("mock stream must not carry errors here"); + }; + accumulator + .process(event) + .expect("accumulator accepts mock events"); + } + let assembled = accumulator.build(); + let tool_calls = assembled.tool_call_parts(); + assert_eq!( + tool_calls.len(), + 1, + "one tool call reconstructed from the stream" + ); + let (id, name, input) = tool_calls[0]; + assert_eq!(id, "call_1"); + assert_eq!(name, "echo"); + assert_eq!(input, &json!({"message": "hi"})); } #[tokio::test] diff --git a/tests/compaction_noop.rs b/tests/compaction_noop.rs new file mode 100644 index 0000000..19627fc --- /dev/null +++ b/tests/compaction_noop.rs @@ -0,0 +1,488 @@ +//! Compaction contracts for default-constructed and profile-configured loops. +//! +//! Pins three invariants: no request is ever sent with a conversation whose +//! estimated size exceeds the configured context window, a default-constructed +//! loop has compaction machinery behind its threshold (observer-visible +//! compaction, not silent estimate resets), and the small-model profile keeps +//! the same promise. Also covers the pre-compact-hook veto path (a measured +//! estimate, not a hard-coded zero) and host-installed context managers +//! surviving the constructor's default seeding. +//! +//! Requires the `testing` feature; the hook-veto test also requires `hooks`. + +#![allow( + dead_code, + clippy::pedantic, + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::arithmetic_side_effects, + clippy::indexing_slicing, + clippy::redundant_clone +)] + +#[cfg(feature = "testing")] +mod scenarios { + use std::future::Future; + use std::pin::Pin; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::{Arc, Mutex}; + + use futures::Stream; + use loopctl::api::error::ApiError; + use loopctl::api::{ApiClient, NonStreamingResponse, StreamRequest}; + use loopctl::compact::{ + CompactionContext, CompactionOutcome, ContextCompactor, ContextManager, + HeuristicTokenCounter, TokenCounter, + }; + use loopctl::config::SessionConfig; + use loopctl::engine::core::Loop; + use loopctl::engine::{BareLoop, RunConfig}; + use loopctl::error::LoopError; + use loopctl::message::Message; + use loopctl::observer::{CompactedContext, LoopObserver}; + use loopctl::stream::StreamEvent; + use loopctl::testing::{MockApiClient, MockResponse, MockToolCall}; + use loopctl::tool::{Tool, ToolContext, ToolOutput, ToolRegistry, ToolSchema}; + + #[derive(Clone)] + struct RecordingClient { + inner: MockApiClient, + request_tokens: Arc>>, + } + + impl RecordingClient { + fn wrap(inner: MockApiClient) -> Self { + Self { + inner, + request_tokens: Arc::new(Mutex::new(Vec::new())), + } + } + + fn record(&self, request: &StreamRequest) { + let tokens = HeuristicTokenCounter.count(&request.messages); + self.request_tokens + .lock() + .expect("request log lock") + .push(tokens); + } + + fn served_request_tokens(&self) -> Vec { + self.request_tokens + .lock() + .expect("request log lock") + .clone() + } + } + + impl ApiClient for RecordingClient { + fn model(&self) -> String { + self.inner.model() + } + + fn stream_messages( + &self, + request: &StreamRequest, + ) -> Pin> + Send + 'static>> { + self.record(request); + self.inner.stream_messages(request) + } + + fn create_message( + &self, + request: &StreamRequest, + ) -> Pin> + Send + '_>> + { + self.record(request); + self.inner.create_message(request) + } + } + + struct CompactionCounter { + events: AtomicUsize, + } + + impl LoopObserver for CompactionCounter { + fn name(&self) -> &'static str { + "CompactionCounter" + } + + fn on_compaction(&self, _ctx: &CompactedContext) { + self.events.fetch_add(1, Ordering::SeqCst); + } + } + + struct EchoTool; + + impl Tool for EchoTool { + fn name(&self) -> &'static str { + "echo" + } + + fn description(&self) -> &'static str { + "Returns its input inside a fixed-size result" + } + + fn schema(&self) -> ToolSchema { + ToolSchema { + tool: self.name().to_string(), + description: self.description().to_string(), + input_schema: serde_json::json!({"type": "object"}), + } + } + + fn call( + &self, + input: serde_json::Value, + _ctx: &ToolContext, + ) -> Pin> + Send + '_>> + { + Box::pin(async move { + let fill = input + .get("fill") + .and_then(serde_json::Value::as_u64) + .and_then(|fill| usize::try_from(fill).ok()) + .unwrap_or(200); + let payload = format!("echo: {input} {}", "r".repeat(fill)); + Ok(ToolOutput::text(payload)) + }) + } + } + + /// One scripted assistant turn: `text_chars` of text plus an echo call. + /// + /// The step index varies both the text and the tool input so loop and + /// convergence detection see distinct responses across turns. + fn tool_turn_response(step: usize, text_chars: usize) -> MockResponse { + MockResponse { + text: format!("step {step} {}", "w".repeat(text_chars)), + tool_call: Some(MockToolCall { + id: format!("call_{step}"), + name: "echo".to_string(), + input: serde_json::json!({"step": step}), + }), + stop_reason: "tool_use".to_string(), + } + } + + /// A tool turn whose echo result is `fill` characters, so a single + /// dispatch can grow the history by a controlled amount. + fn tool_turn_response_with_fill(step: usize, text_chars: usize, fill: usize) -> MockResponse { + MockResponse { + text: format!("step {step} {}", "w".repeat(text_chars)), + tool_call: Some(MockToolCall { + id: format!("call_{step}"), + name: "echo".to_string(), + input: serde_json::json!({"step": step, "fill": fill}), + }), + stop_reason: "tool_use".to_string(), + } + } + + fn final_response() -> MockResponse { + MockResponse { + text: "all done".to_string(), + tool_call: None, + stop_reason: "end_turn".to_string(), + } + } + + /// A conversation of `turns` echo turns that grows past a small window, + /// ending with a text-only response so the run completes. + fn growing_conversation_script(turns: usize) -> Vec { + (0..turns) + .map(|step| tool_turn_response(step, 40)) + .chain(std::iter::once(final_response())) + .collect() + } + + fn registry_with_echo() -> ToolRegistry { + let mut registry = ToolRegistry::new(); + registry.register(EchoTool); + registry + } + + #[tokio::test] + async fn over_limit_context_is_never_sent_to_the_provider() { + let script = vec![ + final_response(), + tool_turn_response(0, 40), + tool_turn_response(1, 40), + tool_turn_response(2, 40), + tool_turn_response(3, 40), + tool_turn_response(4, 40), + ]; + let client = RecordingClient::wrap(MockApiClient::new("test-model").with_responses(script)); + + let config = SessionConfig::default() + .with_context_window(200) + .with_compact_threshold(50); + let client_handle = client.clone(); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + + let first = agent.run(&"x".repeat(200), &RunConfig::default()).await; + assert!(first.is_ok(), "the first run stays under the threshold"); + + let _second = agent.run(&"y".repeat(150), &RunConfig::default()).await; + + let served = client_handle.served_request_tokens(); + assert!( + !served.is_empty(), + "the loop must have served at least one request" + ); + for tokens in &served { + assert!( + *tokens <= 200, + "no request may exceed the 200-token window; served estimates {served:?}" + ); + } + } + + #[tokio::test] + async fn first_request_of_an_over_window_run_is_never_sent() { + let script = vec![ + final_response(), + tool_turn_response(0, 40), + tool_turn_response(1, 40), + ]; + let client = RecordingClient::wrap(MockApiClient::new("test-model").with_responses(script)); + + let config = SessionConfig::default() + .with_context_window(200) + .with_compact_threshold(80); + let client_handle = client.clone(); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + + let first = agent.run(&"x".repeat(600), &RunConfig::default()).await; + assert!( + first.is_ok(), + "the first run starts under the threshold and completes: {first:?}" + ); + + let _second = agent.run(&"y".repeat(200), &RunConfig::default()).await; + + let served = client_handle.served_request_tokens(); + for tokens in &served { + assert!( + *tokens <= 200, + "committed history over the window must trigger compaction (or a typed \ + failure) before the run's first request; served estimates {served:?}" + ); + } + } + + #[tokio::test] + async fn tool_result_growth_alone_crosses_the_threshold() { + let script = vec![tool_turn_response_with_fill(0, 20, 8_000), final_response()]; + let client = RecordingClient::wrap(MockApiClient::new("test-model").with_responses(script)); + + let config = SessionConfig::default() + .with_context_window(2_000) + .with_compact_threshold(80); + let client_handle = client.clone(); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + + let result = agent.run("keep echoing", &RunConfig::default()).await; + + match result { + Err(LoopError::ContextExceeded { .. }) => {} + other => panic!( + "tool-result growth alone must trip the compaction check and fail to \ + shrink the three-message history; got {other:?}" + ), + } + let served = client_handle.served_request_tokens(); + assert_eq!( + served.len(), + 1, + "the follow-up request must wait for the compaction check to see the \ + tool-result growth; served {served:?}" + ); + for tokens in &served { + assert!( + *tokens <= 2_000, + "no request may exceed the 2_000-token window; served estimates {served:?}" + ); + } + } + + #[tokio::test] + async fn default_loop_compacts_at_the_threshold() { + let client = RecordingClient::wrap( + MockApiClient::new("test-model").with_responses(growing_conversation_script(12)), + ); + let observer = Arc::new(CompactionCounter { + events: AtomicUsize::new(0), + }); + + let config = SessionConfig::default() + .with_context_window(600) + .with_compact_threshold(80); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + agent.register_observer(Arc::clone(&observer) as Arc); + + let result = agent.run("keep echoing", &RunConfig::default()).await; + + assert!( + result.is_ok(), + "a default-configured loop must survive crossing the threshold: {result:?}" + ); + assert!( + observer.events.load(Ordering::SeqCst) >= 1, + "a default-constructed loop must compact at the threshold (on_compaction fired {} times)", + observer.events.load(Ordering::SeqCst) + ); + } + + #[tokio::test] + async fn small_model_profile_compacts_at_the_threshold() { + let client = RecordingClient::wrap( + MockApiClient::new("test-model").with_responses(growing_conversation_script(12)), + ); + let observer = Arc::new(CompactionCounter { + events: AtomicUsize::new(0), + }); + + let config = + loopctl::presets::ConstrainedProfile::session_config().with_context_window(600); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + loopctl::presets::ConstrainedProfile::apply(&mut agent).expect("profile applies"); + agent.register_observer(Arc::clone(&observer) as Arc); + + let result = agent.run("keep echoing", &RunConfig::default()).await; + + assert!( + result.is_ok(), + "a profile-configured loop must survive crossing the threshold: {result:?}" + ); + assert!( + observer.events.load(Ordering::SeqCst) >= 1, + "the small-model profile must compact at the threshold (on_compaction fired {} times)", + observer.events.load(Ordering::SeqCst) + ); + } + + #[cfg(feature = "hooks")] + #[tokio::test] + async fn pre_compact_hook_veto_reports_measured_estimate() { + use loopctl::hooks::context::{CompactResult, PreCompactContext}; + use loopctl::hooks::{Hook, HookExecutor}; + + struct VetoCompactHook; + + impl Hook for VetoCompactHook { + fn name(&self) -> &str { + "veto_compact" + } + + fn on_pre_compact(&self, _ctx: &PreCompactContext) -> Option { + Some(CompactResult::abort("not now")) + } + } + + let client = RecordingClient::wrap( + MockApiClient::new("test-model").with_responses(growing_conversation_script(12)), + ); + + let config = SessionConfig::default() + .with_context_window(200) + .with_compact_threshold(50); + let client_handle = client.clone(); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + let mut executor = HookExecutor::new(); + executor.register(Arc::new(VetoCompactHook)); + agent.set_hook_executor(Arc::new(executor)); + + let result = agent.run("keep echoing", &RunConfig::default()).await; + + match result { + Err(LoopError::ContextExceeded { used, .. }) => { + assert!( + used > 0, + "the vetoed pass must report a measured estimate, got {used}" + ); + } + other => panic!( + "a vetoed compaction over the threshold must fail with ContextExceeded, got {other:?}" + ), + } + for tokens in client_handle.served_request_tokens() { + assert!( + tokens <= 200, + "the vetoed pass must not keep sending over-window requests (window 200)" + ); + } + } + + #[tokio::test] + async fn host_installed_context_manager_is_unaffected() { + struct ShrinkingCompactor { + ran: Arc, + } + + impl ContextCompactor for ShrinkingCompactor { + fn compact( + &self, + messages: Vec, + _target_tokens: u64, + context: CompactionContext, + ) -> Pin + Send + '_>> { + let ran = Arc::clone(&self.ran); + Box::pin(async move { + ran.store(true, Ordering::SeqCst); + let kept: Vec = messages.last().cloned().into_iter().collect(); + let tokens_after = context.counter.count(&kept); + CompactionOutcome { + tokens_saved: context.tokens_before.saturating_sub(tokens_after), + messages: kept, + tokens_after, + success: true, + error: None, + } + }) + } + } + + let script = growing_conversation_script(12); + let client = RecordingClient::wrap(MockApiClient::new("test-model").with_responses(script)); + + let ran = Arc::new(AtomicBool::new(false)); + let manager = ContextManager::new(Arc::new(ShrinkingCompactor { + ran: Arc::clone(&ran), + })); + + let config = SessionConfig::default() + .with_context_window(200) + .with_compact_threshold(50); + let mut agent = BareLoop::new(Arc::new(client), registry_with_echo(), config); + agent.set_context_manager(Arc::new(manager)); + + let result = agent.run("keep echoing", &RunConfig::default()).await; + + assert!( + result.is_ok(), + "a run with a host-installed manager must complete: {result:?}" + ); + assert!( + ran.load(Ordering::SeqCst), + "the host-installed compactor must serve the compaction, not the constructor default" + ); + let conversation = agent.conversation(); + assert!( + conversation.len() < 10, + "the host compactor's aggressive shape must survive; got {} messages", + conversation.len() + ); + assert!( + !conversation.iter().any(|m| { + m.parts.iter().any(|p| { + matches!( + p, + loopctl::message::MessagePart::Text { text } if text.contains("keep echoing") + ) + }) + }), + "the default truncator always preserves the first message; its absence proves the host compactor served the pass" + ); + } +} diff --git a/tests/mcp_tool_provider.rs b/tests/mcp_tool_provider.rs index acd1801..233ca9a 100644 --- a/tests/mcp_tool_provider.rs +++ b/tests/mcp_tool_provider.rs @@ -80,7 +80,7 @@ impl EchoServer { "ok".to_string() } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for EchoServer {} @@ -187,7 +187,7 @@ impl SoftErrorServer { Ok(CallToolResult::error(vec![ContentBlock::text("boom")])) } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for SoftErrorServer {} @@ -227,7 +227,7 @@ impl EmptyErrorServer { Ok(CallToolResult::error(vec![])) } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for EmptyErrorServer {} @@ -253,7 +253,7 @@ impl ProtocolErrorServer { Err(McpErrorData::invalid_params("not allowed", None)) } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for ProtocolErrorServer {} @@ -341,7 +341,7 @@ impl MultipartServer { ])) } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for MultipartServer {} @@ -415,7 +415,7 @@ impl AudioServer { )])) } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for AudioServer {} @@ -488,7 +488,7 @@ impl CollisionServer { "b".to_string() } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for CollisionServer {} @@ -649,16 +649,16 @@ impl AnnotatedServer { } impl ServerHandler for AnnotatedServer { - async fn list_tools( + fn list_tools( &self, _request: Option, _ctx: rmcp::service::RequestContext, - ) -> Result { - Ok(rmcp::model::ListToolsResult { + ) -> impl Future> { + std::future::ready(Ok(rmcp::model::ListToolsResult { next_cursor: None, tools: vec![annotated_tool(true), plain_tool()], ..Default::default() - }) + })) } } @@ -730,7 +730,7 @@ impl SlowServer { "finally".to_string() } } - +#[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for SlowServer {} diff --git a/tests/mcp_transports.rs b/tests/mcp_transports.rs index 169940a..d1091d6 100644 --- a/tests/mcp_transports.rs +++ b/tests/mcp_transports.rs @@ -157,6 +157,7 @@ async fn reconnect_in_process_client_is_error() { "ok".into() } } + #[allow(clippy::unused_async_trait_impl)] // FIXME(rmcp): drop when tool_handler emits awaits #[tool_handler] impl ServerHandler for S {}