Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0.

### Added

- `LoopMachine::compaction_noop(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 the driver's measured estimate as the current context size, applying the same no-progress guard (terminal `ContextExceeded` when nothing was shaved off). Alt drivers servicing `MachineStep::Compact` should use it for no-op passes.
- `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.
Expand All @@ -20,6 +21,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0.

### Changed

- `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`. 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`.
- 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.
Expand All @@ -42,6 +46,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0.

### Fixed

- 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`.
Expand Down
37 changes: 35 additions & 2 deletions src/compact.rs
Original file line number Diff line number Diff line change
Expand Up @@ -668,13 +668,19 @@ impl ContextManager {
});
}

if outcome.messages.len() == message_count && outcome.tokens_saved == 0 {
return Ok(EnsureContextResult::NoAction(outcome.messages));
}
Ok(EnsureContextResult::Compacted(outcome))
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
}

/// 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 that shrinks neither the message list nor the token
/// count is reported as [`EnsureContextResult::NoAction`] — compaction
/// did not occur, so compaction observers and hooks stay silent.
/// # Errors
///
/// Returns a [`ContextOverflow`] if compaction fails or the result
Expand All @@ -693,7 +699,10 @@ 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
/// that shrinks neither the message list nor the token count is
/// reported as [`EnsureContextResult::NoAction`] — compaction did not
/// occur, so compaction observers and hooks stay silent.
///
/// # Errors
///
Expand Down Expand Up @@ -745,6 +754,9 @@ impl ContextManager {
});
}

if outcome.messages.len() == message_count && outcome.tokens_saved == 0 {
return Ok(EnsureContextResult::NoAction(outcome.messages));
}
Ok(EnsureContextResult::Compacted(outcome))
}

Expand Down Expand Up @@ -1237,4 +1249,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<Message> = (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"
);
}
}
}
65 changes: 54 additions & 11 deletions src/engine/bare.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -353,8 +353,10 @@ impl<C: ApiClient> BareLoop<C> {

/// 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
///
Expand All @@ -381,6 +383,13 @@ impl<C: ApiClient> BareLoop<C> {
/// [`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`.
Expand All @@ -406,8 +415,12 @@ impl<C: ApiClient> BareLoop<C> {
client: Arc<C>,
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),
Expand Down Expand Up @@ -573,20 +586,25 @@ impl<C: ApiClient> BareLoop<C> {
/// 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.
/// is taken from the machine. 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,
session_config: SessionConfig,
client: Arc<C>,
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()),
Expand All @@ -599,6 +617,21 @@ impl<C: ApiClient> BareLoop<C> {
}
}

/// 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
Expand Down Expand Up @@ -1150,7 +1183,13 @@ impl<C: ApiClient> BareLoop<C> {
///
/// 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 through
/// [`LoopMachine::compaction_result`](crate::engine::core::LoopMachine::compaction_result),
/// an unchanged one 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
///
Expand All @@ -1161,10 +1200,14 @@ impl<C: ApiClient> BareLoop<C> {
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);
match self.run_compaction(turn, reason).await? {
compact::CompactStepOutcome::Compacted(compacted, tokens_after) => {
self.machine.compaction_result(compacted, tokens_after);
}
compact::CompactStepOutcome::Unchanged(tokens_after) => {
self.machine.compaction_noop(tokens_after);
}
}
Ok(())
}
}
Expand Down
Loading