diff --git a/Cargo.lock b/Cargo.lock index e53ec7fa..7bbd5dbe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -195,6 +195,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "errno" version = "0.3.14" @@ -239,6 +245,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -369,6 +381,25 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "h2" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "839c0e8a181239723652be9062bb56ca5bf5f64011f73b623f6f4fc59086a228" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -454,6 +485,7 @@ dependencies = [ "bytes", "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", @@ -630,6 +662,16 @@ dependencies = [ "icu_properties", ] +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown 0.17.1", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -937,6 +979,7 @@ dependencies = [ "bytes", "futures-core", "futures-util", + "h2", "http", "http-body", "http-body-util", diff --git a/Cargo.toml b/Cargo.toml index 81b375f0..088c121a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,6 +48,13 @@ reqwest = { version = "0.12", default-features = false, features = [ "json", "rustls-tls", "stream", + # `http2` lets TLS endpoints negotiate HTTP/2 via ALPN, which is what + # enables the transport-liveness PING keepalives the provider clients + # configure (`http2_keep_alive_*`): a dead peer fails the in-flight call in + # under a minute even when zero application bytes are flowing — e.g. a + # provider stream that goes silent during long hidden reasoning. Plaintext + # endpoints (local Ollama/LM Studio) have no ALPN and stay on HTTP/1.1. + "http2", ] } # Optional embedded SQLite checkpointer backend (`graph::checkpoint::sqlite`). diff --git a/src/harness/agent_loop/model_call.rs b/src/harness/agent_loop/model_call.rs index 01272a82..7ee34c34 100644 --- a/src/harness/agent_loop/model_call.rs +++ b/src/harness/agent_loop/model_call.rs @@ -6,6 +6,16 @@ //! Split out of `agent_loop/mod.rs`; see that module's doc comment for //! the full loop lifecycle, limits, and backoff design. +/// Timeout-message label for a call bounded by the run's remaining wall-clock +/// budget: the run is out of time, not (necessarily) this call wedged. +pub(super) const RUN_BOUND_LABEL: &str = "remaining wall-clock budget"; + +/// Timeout-message label for a model call bounded by the per-call ceiling +/// ([`RunLimits::max_model_call_ms`][crate::harness::limits::RunLimits::max_model_call_ms]): +/// this one call ran past the time any single call is allowed, with run time +/// still left. +pub(super) const PER_CALL_BOUND_LABEL: &str = "per-model-call ceiling"; + use super::*; use crate::harness::cache::{ CachePolicy, CacheSkipReason, apply_prompt_cache_breakpoints, scoped_cache_key, @@ -325,14 +335,16 @@ impl AgentHarness { if ctx.cancellation.is_cancelled() { return Err(TinyAgentsError::Cancelled); } - // Bound this individual provider call by the run's *remaining* - // wall-clock budget so a hung or slow model call is interrupted + // Bound this individual provider call by the tighter of the + // run's *remaining* wall-clock budget and the per-model-call + // ceiling, so a hung or slow model call is interrupted // mid-flight, not merely detected by the between-call deadline // check. reqwest/futures are cancel-safe, so dropping the future - // on elapse cancels the underlying request. When neither the run - // config nor the harness policy configures a timeout the call is - // awaited unbounded. - let remaining = self.call_budget(ctx); + // on elapse cancels the underlying request. Recomputed here, + // inside the attempt loop, so every retry attempt gets a fresh + // per-call window. When neither the run config nor the harness + // policy configures any timeout the call is awaited unbounded. + let (remaining, bound) = self.model_call_budget(ctx); let attempt_result = if streaming { let fut = self.invoke_model_streaming_once( state, @@ -342,7 +354,8 @@ impl AgentHarness { call_id, &mut deltas_emitted, ); - Self::with_call_budget(remaining, run_id.as_str(), "model call", fut).await + Self::with_call_budget(remaining, run_id.as_str(), "model call", bound, fut) + .await } else { // Race the wall-clock-bounded unary call against cooperative // cancellation, mirroring the streaming path: a cancel @@ -354,8 +367,13 @@ impl AgentHarness { // still short-circuits before the request is ever issued. let cancellation = ctx.cancellation.clone(); let fut = model.invoke(state, request.clone()); - let budgeted = - Self::with_call_budget(remaining, run_id.as_str(), "model call", fut); + let budgeted = Self::with_call_budget( + remaining, + run_id.as_str(), + "model call", + bound, + fut, + ); tokio::select! { biased; _ = cancellation.cancelled() => Err(TinyAgentsError::Cancelled), @@ -510,6 +528,13 @@ impl AgentHarness { /// (`None`). Honoring the policy source lets a sub-agent whose child /// [`RunConfig`] carries no per-run timeout still be bounded by its /// harness's policy-level wall-clock cap. + /// + /// This is the budget for **tool calls**, which are bounded only by the + /// run's remaining time (a sub-agent delegation is a tool call wrapping an + /// entire child run). Model calls go through + /// [`model_call_budget`](Self::model_call_budget), which additionally + /// clamps to the per-call ceiling + /// [`RunLimits::max_model_call_ms`][crate::harness::limits::RunLimits::max_model_call_ms]. pub(super) fn call_budget(&self, ctx: &RunContext) -> Option { let config_budget = ctx.remaining_wall_clock(); let policy_budget = self.policy.limits.max_wall_clock_ms.map(|ms| { @@ -525,6 +550,35 @@ impl AgentHarness { } } + /// Computes the wall-clock budget for the next individual **model** call: + /// the tighter of the run's remaining budget ([`call_budget`](Self::call_budget)) + /// and the per-call ceiling + /// [`RunLimits::max_model_call_ms`][crate::harness::limits::RunLimits::max_model_call_ms]. + /// + /// Computed afresh for every call *and every retry attempt*, so each + /// attempt gets its own full per-call window rather than inheriting an + /// earlier attempt's consumption — while still never overshooting the run + /// deadline. Returns the budget together with the label of whichever bound + /// is in force, so the timeout error names the ceiling that actually fired + /// (a per-call ceiling means "this one call wedged"; the run's remainder + /// means "the run is out of time") — field triage needs to tell them apart. + pub(super) fn model_call_budget( + &self, + ctx: &RunContext, + ) -> (Option, &'static str) { + let run_budget = self.call_budget(ctx); + let per_call = self + .policy + .limits + .max_model_call_ms + .map(Duration::from_millis); + match (run_budget, per_call) { + (Some(run), Some(cap)) if cap < run => (Some(cap), PER_CALL_BOUND_LABEL), + (None, Some(cap)) => (Some(cap), PER_CALL_BOUND_LABEL), + (run, _) => (run, RUN_BOUND_LABEL), + } + } + /// Awaits a single call future (model or tool), optionally bounded by /// `budget`. /// @@ -534,14 +588,18 @@ impl AgentHarness { /// [`TinyAgentsError::Timeout`] is returned. When `budget` is `None` (no /// run timeout configured) the future is awaited without a bound. /// - /// `budget` is the run's *remaining* wall-clock budget at the time the call - /// is issued, so each successive call gets a tighter bound as the deadline - /// approaches. `what` names the kind of call in the timeout message (e.g. - /// `"model call"`, `"tool call"`). + /// `budget` is either the run's *remaining* wall-clock budget at the time + /// the call is issued (each successive call gets a tighter bound as the + /// deadline approaches) or, for model calls, the per-call ceiling when that + /// is tighter. `what` names the kind of call in the timeout message (e.g. + /// `"model call"`, `"tool call"`); `bound` names which budget source is in + /// force ([`RUN_BOUND_LABEL`] / [`PER_CALL_BOUND_LABEL`]) so the message + /// says which ceiling fired. pub(super) async fn with_call_budget( budget: Option, run_id: &str, what: &str, + bound: &str, fut: F, ) -> Result where @@ -551,8 +609,7 @@ impl AgentHarness { Some(budget) => match tokio::time::timeout(budget, fut).await { Ok(result) => result, Err(_) => Err(TinyAgentsError::Timeout(format!( - "{what} for run `{run_id}` exceeded its remaining wall-clock budget \ - ({} ms)", + "{what} for run `{run_id}` exceeded its {bound} ({} ms)", budget.as_millis() ))), }, diff --git a/src/harness/agent_loop/test.rs b/src/harness/agent_loop/test.rs index 9f9f4a64..446b000e 100644 --- a/src/harness/agent_loop/test.rs +++ b/src/harness/agent_loop/test.rs @@ -2495,6 +2495,137 @@ async fn slow_tool_call_is_timed_out_by_remaining_budget() { assert!(matches!(err, TinyAgentsError::Timeout(_)), "got {err:?}"); } +#[tokio::test] +async fn per_model_call_ceiling_times_out_a_slow_call_with_run_time_left() { + use std::time::Duration; + + use crate::harness::testkit::SlowModel; + + // The run has plenty of wall clock left (60s), but the per-model-call + // ceiling (20ms) is tighter than the model's 200ms sleep, so the ceiling + // interrupts the call — and the error must name the ceiling, not the run's + // remaining budget, so triage can tell a wedged call from an exhausted run. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "slow", + Arc::new(SlowModel::new(Duration::from_millis(200), "too late")), + ); + harness.with_policy(RunPolicy { + limits: RunLimits::default().with_max_model_call_ms(Some(20)), + ..RunPolicy::default() + }); + + let config = RunConfig::new("per-call-cap-run").with_timeout_ms(60_000); + let err = harness + .invoke(&(), (), config, vec![Message::user("hi")]) + .await + .expect_err("a call slower than the per-call ceiling must time out"); + + match &err { + TinyAgentsError::Timeout(msg) => { + assert!(msg.contains("per-model-call ceiling"), "{msg}"); + } + other => panic!("expected Timeout, got {other:?}"), + } +} + +#[tokio::test] +async fn per_model_call_ceiling_bounds_calls_without_any_run_deadline() { + use std::time::Duration; + + use crate::harness::testkit::SlowModel; + + // With no run timeout and no policy wall clock, a model call used to be + // awaited unbounded. The per-call ceiling alone must bound it. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "slow", + Arc::new(SlowModel::new(Duration::from_millis(200), "too late")), + ); + harness.with_policy(RunPolicy { + limits: RunLimits::default().with_max_model_call_ms(Some(20)), + ..RunPolicy::default() + }); + + let err = harness + .invoke_default(&(), vec![Message::user("hi")]) + .await + .expect_err("the ceiling alone must bound an otherwise-unbounded call"); + + match &err { + TinyAgentsError::Timeout(msg) => { + assert!(msg.contains("per-model-call ceiling"), "{msg}"); + } + other => panic!("expected Timeout, got {other:?}"), + } +} + +#[tokio::test] +async fn run_remainder_bounds_a_model_call_when_tighter_than_the_ceiling() { + use std::time::Duration; + + use crate::harness::testkit::SlowModel; + + // A generous ceiling (10s) never extends a call past the run's own + // deadline (20ms): the tighter source wins, and the error blames the + // remaining wall-clock budget. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "slow", + Arc::new(SlowModel::new(Duration::from_millis(200), "too late")), + ); + harness.with_policy(RunPolicy { + limits: RunLimits::default().with_max_model_call_ms(Some(10_000)), + ..RunPolicy::default() + }); + + let config = RunConfig::new("remainder-tighter-run").with_timeout_ms(20); + let err = harness + .invoke(&(), (), config, vec![Message::user("hi")]) + .await + .expect_err("the run deadline must still bound the call"); + + match &err { + TinyAgentsError::Timeout(msg) => { + assert!(msg.contains("remaining wall-clock budget"), "{msg}"); + } + other => panic!("expected Timeout, got {other:?}"), + } +} + +#[tokio::test] +async fn per_model_call_ceiling_does_not_bound_tool_calls() { + use std::time::Duration; + + // A 20ms per-model-call ceiling with a 100ms tool: tool calls keep the + // remaining-only budget (a sub-agent delegation is a tool call wrapping an + // entire child run), so the run completes despite the tool outliving the + // model-call ceiling many times over. + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model( + "mock", + Arc::new(MockModel::with_responses(vec![ + tool_call_response("call-1", "slow", json!({})), + text_response("done", 0, 0), + ])), + ); + harness.register_tool(Arc::new(SlowTool { + delay: Duration::from_millis(100), + })); + harness.with_policy(RunPolicy { + limits: RunLimits::default().with_max_model_call_ms(Some(20)), + ..RunPolicy::default() + }); + + let config = RunConfig::new("tool-uncapped-run").with_timeout_ms(60_000); + let run = harness + .invoke(&(), (), config, vec![Message::user("go")]) + .await + .expect("the tool call must not inherit the per-model-call ceiling"); + + assert_eq!(run.text(), Some("done".to_string())); +} + #[tokio::test] async fn inherited_tool_timeout_is_enforced_without_a_run_deadline() { use std::time::Duration; diff --git a/src/harness/agent_loop/tools.rs b/src/harness/agent_loop/tools.rs index 28bf9b0c..ef083ed8 100644 --- a/src/harness/agent_loop/tools.rs +++ b/src/harness/agent_loop/tools.rs @@ -577,7 +577,14 @@ impl AgentHarness { let outcome = fut.await.map(|wrapped| wrapped.into_result()); apply_tool_error_policy(&error_policy, &policy_call, outcome) }; - let outcome = Self::with_call_budget(run_budget, &run_id, "tool call", guarded).await; + let outcome = Self::with_call_budget( + run_budget, + &run_id, + "tool call", + super::model_call::RUN_BOUND_LABEL, + guarded, + ) + .await; let result = match outcome { Ok(result) => result, Err(err) => { @@ -700,7 +707,14 @@ impl AgentHarness { // failure, inside the run-budget wrapper that stays fatal. let guarded = async move { apply_tool_error_policy(&error_policy, &policy_call, fut.await) }; - Self::with_call_budget(run_budget, &run_id, "tool call", guarded).await + Self::with_call_budget( + run_budget, + &run_id, + "tool call", + super::model_call::RUN_BOUND_LABEL, + guarded, + ) + .await }); } diff --git a/src/harness/limits/mod.rs b/src/harness/limits/mod.rs index 9d42bd4f..0b3b87ad 100644 --- a/src/harness/limits/mod.rs +++ b/src/harness/limits/mod.rs @@ -49,6 +49,13 @@ impl RunLimits { self } + /// Sets a per-model-call wall-clock ceiling in milliseconds. `None` + /// removes the ceiling. See [`RunLimits::max_model_call_ms`]. + pub fn with_max_model_call_ms(mut self, ms: Option) -> Self { + self.max_model_call_ms = ms; + self + } + /// Sets the per-call retry cap (a retry *count*, not counting the first /// attempt). See [`RunLimits::max_retries_per_call`]. pub fn with_max_retries_per_call(mut self, n: usize) -> Self { diff --git a/src/harness/limits/types.rs b/src/harness/limits/types.rs index b4cd37d4..3846f350 100644 --- a/src/harness/limits/types.rs +++ b/src/harness/limits/types.rs @@ -27,6 +27,32 @@ pub struct RunLimits { pub max_tool_calls: usize, /// Maximum elapsed wall-clock time in milliseconds. `None` means no limit. pub max_wall_clock_ms: Option, + /// Maximum wall-clock time in milliseconds for a **single model call**, + /// applied afresh to every call (and every retry attempt). `None` means no + /// per-call ceiling. + /// + /// This bounds an individual hung or wedged provider call independently of + /// [`max_wall_clock_ms`](Self::max_wall_clock_ms), which measures the whole + /// run. Without it the only per-call bound is the run's *remaining* + /// wall-clock budget, which conflates two different guards: hang detection + /// (a wedged call should die fast) and runaway-run bounding (a run should + /// not live forever). With only the run deadline, a host must choose one + /// number for both — a generous run ceiling means a hung call can hold the + /// run for that whole ceiling, and a tight one kills late calls in long, + /// *productive* runs even though every earlier call succeeded. + /// + /// The effective budget for a model call is the tighter of this ceiling and + /// the run's remaining wall-clock budget, so a per-call cap can never + /// extend a run past its deadline. Deliberately **not** applied to tool + /// calls: a sub-agent delegation is a tool call wrapping an entire child + /// run and must not inherit a model-call-sized cap; tools carry their own + /// [`ToolTimeoutSettings`][crate::harness::tool::ToolTimeoutSettings] + /// deadlines and remain bounded by the run's remaining budget. + /// + /// Size it generously: a hidden-reasoning model call can be legitimately + /// silent for minutes, so this is a backstop for calls that will never + /// return, not a latency target. + pub max_model_call_ms: Option, /// Maximum number of retry *attempts* (not counting the first try) /// permitted for an individual model call. Reconciled with /// [`crate::harness::retry::RetryPolicy::max_attempts`] by the agent loop @@ -150,6 +176,7 @@ impl Default for RunLimits { max_model_calls: 25, max_tool_calls: 50, max_wall_clock_ms: None, + max_model_call_ms: None, max_retries_per_call: 3, max_depth: Self::DEFAULT_MAX_DEPTH, behavior: LimitBehavior::Error, diff --git a/src/harness/providers/openai/mod.rs b/src/harness/providers/openai/mod.rs index 94a30e05..aa826c9e 100644 --- a/src/harness/providers/openai/mod.rs +++ b/src/harness/providers/openai/mod.rs @@ -82,6 +82,29 @@ const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 30; /// Default overall timeout applied to unary calls when the request does not set /// [`ModelRequest::timeout_ms`]. Streaming calls get no overall cap by default. const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 600; +/// HTTP/2 PING keepalive interval and ack timeout for the default provider +/// client. Streaming calls deliberately carry no overall request timeout (a +/// total cap would truncate a legitimately long stream — e.g. a reasoning +/// model that is app-silent for minutes), so transport-level PINGs are what +/// distinguishes "thinking" from "dead": a peer that stops acking fails the +/// in-flight call in roughly `interval + timeout` (~1 min) even with zero +/// application bytes flowing. Only applies where TLS ALPN negotiated h2; +/// plaintext HTTP/1.1 endpoints (local Ollama/LM Studio) are unaffected. +const DEFAULT_KEEP_ALIVE_SECS: u64 = 30; + +/// Builds the default `reqwest` client for provider transports: connect +/// timeout plus HTTP/2 PING keepalives (see [`DEFAULT_KEEP_ALIVE_SECS`]). +/// Hosts that need different transport policy inject their own client via +/// `with_client`, which opts out of all of this. +pub(crate) fn default_provider_client() -> reqwest::Client { + reqwest::Client::builder() + .connect_timeout(std::time::Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS)) + .http2_keep_alive_interval(std::time::Duration::from_secs(DEFAULT_KEEP_ALIVE_SECS)) + .http2_keep_alive_timeout(std::time::Duration::from_secs(DEFAULT_KEEP_ALIVE_SECS)) + .http2_keep_alive_while_idle(true) + .build() + .expect("default reqwest client builds") +} mod convert; mod local; diff --git a/src/harness/providers/openai/transport.rs b/src/harness/providers/openai/transport.rs index 00ef30d3..4b05e0fa 100644 --- a/src/harness/providers/openai/transport.rs +++ b/src/harness/providers/openai/transport.rs @@ -432,10 +432,7 @@ impl OpenAiModel { /// (`gpt-4.1-mini`), and the default base URL (`https://api.openai.com/v1`). pub fn new(api_key: impl Into) -> Self { Self { - client: reqwest::Client::builder() - .connect_timeout(Duration::from_secs(DEFAULT_CONNECT_TIMEOUT_SECS)) - .build() - .expect("default reqwest client builds"), + client: default_provider_client(), caller_owned_client: false, api_key: api_key.into(), auth: AuthStyle::Bearer,