diff --git a/CHANGELOG.md b/CHANGELOG.md index dd2f468..43e84d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- Provider failover chain, core (#209): `failover::FailoverProvider` retries a step on the next provider of an ordered chain when the active one is unavailable (connection failure, timeout, dropped stream, HTTP 5xx or 429); request errors (401/403/404/400, malformed responses) fail loud and never fail over. The answering provider stays active and the primary is retried after a cooldown. Each step records the route under `tole_route` in its usage row (durable, auditable); rows holding only tole's own keys are not counted as provider steps. The env/config wiring follows separately. - Child-agent handoff schema and verify gate (#383): `tole_core::handoff` defines the typed handoff document (role `explore` / `implement` / `verify` / `fix`, artifact references by path and rev, verification evidence), strict validation of the model-authored input (workspace-relative paths, option-safe refs, bounded fields), the close gate (an implement/fix child without passing verification is `Unverified` / `VerifyFailed`, never closed silently) and the prompt for the follow-up `fix` child. `agent_start` takes `role` and `consumes` (artifact references the child reads itself), a role child gets a `handoff_submit` tool, and `agent_poll` reports the gate and a ready-made `fix_prompt` (the parent decides whether to start the fix child; nothing is spawned automatically). Design and anti-goals (no DAG engine) in `docs/child-agents-handoff.md`. - Sibling change notices, core (#380): `sibling_watch` tracks which files each child agent reads and writes in one shared append-only log (owner-only, bounded) and tells a child, once per stale path, when a sibling changed a file it had read or written; the `Tracked` tool decorator records `read_file` / `write_file` / `edit_file` / `delete_file` on success. Interrupts gain a `source`: a notice is delivered at the next safe point as an attributed entry (`source: notice`) and never withdraws output or supersedes a final answer. Wiring into `tole run` children follows separately. - Sibling change notices (#380): `sibling_watch` tracks which files each child agent reads and writes in one shared append-only log (owner-only, bounded) and tells a child, once per stale path, when a sibling changed a file it had read or written; the `Tracked` tool decorator records `read_file` / `write_file` / `edit_file` / `delete_file` on success. Interrupts gain a `source`: a notice is delivered at the next safe point as an attributed entry (`source: notice`) and never withdraws output or supersedes a final answer. Children spawned by `agent_start` get `TOLE_AGENT_ID` and a shared `TOLE_AGENT_TOUCH_LOG`; in a child, `tole run` wraps its file tools and runs the watcher, so a sibling's change reaches the model at its next safe point. diff --git a/crates/tole-cli/src/usage_report.rs b/crates/tole-cli/src/usage_report.rs index 2ca7053..1420666 100644 --- a/crates/tole-cli/src/usage_report.rs +++ b/crates/tole-cli/src/usage_report.rs @@ -55,15 +55,18 @@ pub struct UsageReport { pub history_growth_per_step: Option, } -/// True when the ledger row carries a provider-reported usage object, -/// i.e. it is NOT a wire-only row (`{"tole_wire": ...}` and nothing -/// else, stored when a provider reports request sizes but no usage). -/// The single definition of "provider step" for the mission step budget, -/// `fact/mission.steps` and the report, so wire-only rows never change -/// step accounting (#211). +/// True when the ledger row carries a provider-reported usage object, i.e. +/// it is NOT a row holding only tole's own namespaced keys (`tole_wire`, +/// stored when a provider reports request sizes but no usage, #211; +/// `tole_route`, the failover route, #209). The single definition of +/// "provider step" for the mission step budget, `fact/mission.steps` and +/// the report, so such rows never change step accounting. pub fn reports_provider_usage(rec: &UsageRecord) -> bool { match rec.usage.as_object() { - Some(o) => !(o.len() == 1 && o.contains_key("tole_wire")), + // Only tole's own namespaced keys (`tole_wire`, `tole_route`, ...): + // the provider reported nothing for this step. + // An empty object still counts (the pre-#211 contract). + Some(o) => o.is_empty() || o.keys().any(|k| !k.starts_with("tole_")), None => true, } } @@ -675,6 +678,18 @@ mod tests { assert_eq!(t[0].input_tokens, Some(4)); } + #[test] + fn rows_with_only_tole_keys_are_not_provider_steps() { + let route_only = rec(json!({"tole_route": {"provider": "reserve"}})); + let wire_and_route = + rec(json!({"tole_wire": wire(1, 2, 3), "tole_route": {"provider": "p"}})); + let real = rec(json!({"prompt_tokens": 4, "tole_route": {"provider": "p"}})); + assert!(!reports_provider_usage(&route_only)); + assert!(!reports_provider_usage(&wire_and_route)); + assert!(reports_provider_usage(&real)); + assert_eq!(provider_steps(&[route_only, wire_and_route, real]), 1); + } + #[test] fn unknown_stays_unknown_never_zero() { let entries = vec![user(1), assistant(2), user(3), assistant(4)]; diff --git a/crates/tole-core/src/failover.rs b/crates/tole-core/src/failover.rs new file mode 100644 index 0000000..21b84c9 --- /dev/null +++ b/crates/tole-core/src/failover.rs @@ -0,0 +1,354 @@ +//! Provider failover chain (#209). +//! +//! `FailoverProvider` wraps an ordered chain of providers (primary first). +//! When the active provider fails with an error that means "this provider +//! is unavailable" — connection failure, timeout, a dropped stream, HTTP +//! 5xx or 429 — the same step is retried on the next provider in the chain. +//! Errors that mean "this request is wrong" (401/403/404/400, a malformed +//! response) fail loud and never fail over: a reserve would only hide a +//! misconfiguration. +//! +//! The provider that answered stays active (no flapping, no repeated slow +//! failures on a dead primary); after `retry_primary_after` the primary is +//! tried again. Each answered step reports `last_route()` — the provider +//! that answered and, when it took a failover, where from and why — which +//! the turn loop stores under `tole_route` in the step's usage row, so a +//! switch is durable and auditable. + +use crate::entry::Entry; +use crate::provider::{Provider, ProviderError, ProviderOutput}; +use serde_json::{json, Value}; +use std::time::{Duration, Instant}; + +/// Default time on a reserve before the primary is tried again. +pub const DEFAULT_RETRY_PRIMARY_AFTER: Duration = Duration::from_secs(300); + +/// Longest error text kept per failed provider in the route record. +const MAX_ERROR_CHARS: usize = 200; + +/// Does this provider error mean "provider unavailable" (fail over) rather +/// than "request wrong" (fail loud)? +pub fn is_failover_error(message: &str) -> bool { + let m = message.to_ascii_lowercase(); + if let Some(rest) = m.strip_prefix("http status: ") { + let code: u16 = rest + .chars() + .take_while(char::is_ascii_digit) + .collect::() + .parse() + .unwrap_or(0); + return code == 429 || (500..600).contains(&code); + } + if m.starts_with("malformed") { + return false; + } + [ + "timeout", + "timed out", + "io:", + "connection", + "host not found", + "dns", + "stream read", + "reset", + "broken pipe", + "unexpected eof", + ] + .iter() + .any(|k| m.contains(k)) +} + +/// One provider in the chain, with the label recorded in `tole_route`. +pub struct Link { + pub label: String, + pub provider: Box, +} + +pub struct FailoverProvider { + chain: Vec, + active: usize, + left_primary_at: Option, + retry_primary_after: Duration, + last_answered: Option, + /// `[(label, error)]` of the providers that failed before the last + /// answer of the current step. + last_failed: Vec<(String, String)>, +} + +impl FailoverProvider { + /// `chain[0]` is the primary. Panics on an empty chain (a host bug). + pub fn new(chain: Vec) -> Self { + assert!(!chain.is_empty(), "a failover chain needs a provider"); + Self { + chain, + active: 0, + left_primary_at: None, + retry_primary_after: DEFAULT_RETRY_PRIMARY_AFTER, + last_answered: None, + last_failed: Vec::new(), + } + } + + pub fn with_retry_primary_after(mut self, after: Duration) -> Self { + self.retry_primary_after = after; + self + } + + fn answered(&self) -> Option<&dyn Provider> { + self.last_answered.map(|i| self.chain[i].provider.as_ref()) + } +} + +fn short(s: &str) -> String { + let mut out: String = s.chars().take(MAX_ERROR_CHARS).collect(); + if s.chars().count() > MAX_ERROR_CHARS { + out.push('…'); + } + out +} + +impl Provider for FailoverProvider { + fn complete(&mut self, transcript: &[Entry]) -> Result { + // Back to the primary once the cooldown has passed. + if self.active > 0 + && self + .left_primary_at + .is_some_and(|t| t.elapsed() >= self.retry_primary_after) + { + self.active = 0; + self.left_primary_at = None; + } + self.last_answered = None; + self.last_failed.clear(); + let start = self.active; + for i in start..self.chain.len() { + match self.chain[i].provider.complete(transcript) { + Ok(out) => { + if i != self.active { + if self.active == 0 { + self.left_primary_at = Some(Instant::now()); + } + self.active = i; + } + self.last_answered = Some(i); + return Ok(out); + } + Err(e) if i + 1 < self.chain.len() && is_failover_error(&e.0) => { + self.last_failed + .push((self.chain[i].label.clone(), short(&e.0))); + } + Err(e) => { + if self.last_failed.is_empty() { + return Err(e); + } + // Keep the last error first: the turn loop's transient + // retry classification reads the start of the message. + let earlier: Vec = self + .last_failed + .iter() + .map(|(l, m)| format!("{l}: {m}")) + .collect(); + return Err(ProviderError(format!( + "{} [after failover from {}]", + e.0, + earlier.join("; ") + ))); + } + } + } + unreachable!("the loop returns on the last provider") + } + + fn last_usage(&self) -> Option { + self.answered().and_then(Provider::last_usage) + } + + fn last_reasoning(&self) -> Option { + self.answered().and_then(Provider::last_reasoning) + } + + fn last_wire_stats(&self) -> Option { + self.answered().and_then(Provider::last_wire_stats) + } + + fn last_route(&self) -> Option { + let i = self.last_answered?; + let mut route = json!({ "provider": self.chain[i].label }); + if !self.last_failed.is_empty() { + route["failover"] = json!({ + "from": self.last_failed[0].0, + "errors": self + .last_failed + .iter() + .map(|(l, m)| json!({ "provider": l, "error": m })) + .collect::>(), + }); + } + Some(route) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::Cell; + use std::rc::Rc; + + /// Scripted provider that counts its calls. + struct Scripted { + replies: Vec>, + calls: Rc>, + } + + impl Provider for Scripted { + fn complete(&mut self, _t: &[Entry]) -> Result { + let n = self.calls.get(); + self.calls.set(n + 1); + match self + .replies + .get(n) + .copied() + .unwrap_or(Err("script exhausted")) + { + Ok(text) => Ok(ProviderOutput::Final { text: text.into() }), + Err(e) => Err(ProviderError(e.into())), + } + } + fn last_usage(&self) -> Option { + Some(json!({ "prompt_tokens": self.calls.get() })) + } + } + + fn link( + label: &str, + replies: Vec>, + ) -> (Link, Rc>) { + let calls = Rc::new(Cell::new(0)); + ( + Link { + label: label.into(), + provider: Box::new(Scripted { + replies, + calls: Rc::clone(&calls), + }), + }, + calls, + ) + } + + fn text(o: ProviderOutput) -> String { + match o { + ProviderOutput::Final { text } => text, + other => panic!("{other:?}"), + } + } + + #[test] + fn classification() { + for yes in [ + "http status: 503 Service Unavailable — upstream down", + "http status: 500 Internal Server Error — x", + "http status: 429 Too Many Requests — slow down", + "io: Connection refused (os error 61)", + "timeout: global", + "host not found", + "stream read: connection reset by peer", + ] { + assert!(is_failover_error(yes), "{yes}"); + } + for no in [ + "http status: 401 Unauthorized — bad key", + "http status: 403 Forbidden — x", + "http status: 404 Not Found — no such model", + "http status: 400 Bad Request — context too long", + "malformed response: no choices[0].message", + "malformed tool_call: missing function.name", + ] { + assert!(!is_failover_error(no), "{no}"); + } + } + + #[test] + fn a_down_primary_fails_over_and_the_route_is_reported() { + let (p, pc) = link( + "primary", + vec![Err("http status: 503 Service Unavailable — down")], + ); + let (r, rc) = link("reserve", vec![Ok("from reserve"), Ok("again")]); + let mut f = FailoverProvider::new(vec![p, r]); + assert_eq!(text(f.complete(&[]).unwrap()), "from reserve"); + let route = f.last_route().unwrap(); + assert_eq!(route["provider"], "reserve"); + assert_eq!(route["failover"]["from"], "primary"); + assert!(route["failover"]["errors"][0]["error"] + .as_str() + .unwrap() + .contains("503")); + assert_eq!( + f.last_usage().unwrap()["prompt_tokens"], + 1, + "usage of the answering provider" + ); + // Sticky: the next step goes straight to the reserve. + assert_eq!(text(f.complete(&[]).unwrap()), "again"); + assert_eq!((pc.get(), rc.get()), (1, 2)); + assert!(f.last_route().unwrap().get("failover").is_none()); + } + + #[test] + fn request_errors_fail_loud_and_never_reach_the_reserve() { + let (p, _) = link( + "primary", + vec![Err("http status: 401 Unauthorized — bad key")], + ); + let (r, rc) = link("reserve", vec![Ok("unused")]); + let mut f = FailoverProvider::new(vec![p, r]); + let e = f.complete(&[]).unwrap_err(); + assert!(e.0.starts_with("http status: 401"), "{}", e.0); + assert_eq!(rc.get(), 0); + assert!(f.last_route().is_none()); + } + + #[test] + fn the_primary_is_retried_after_the_cooldown() { + let (p, pc) = link( + "primary", + vec![ + Err("io: Connection refused (os error 61)"), + Ok("primary is back"), + ], + ); + let (r, _) = link("reserve", vec![Ok("reserve")]); + let mut f = FailoverProvider::new(vec![p, r]).with_retry_primary_after(Duration::ZERO); + assert_eq!(text(f.complete(&[]).unwrap()), "reserve"); + assert_eq!(text(f.complete(&[]).unwrap()), "primary is back"); + assert_eq!(pc.get(), 2); + assert_eq!(f.last_route().unwrap()["provider"], "primary"); + } + + #[test] + fn when_every_provider_fails_the_last_error_leads() { + let (p, _) = link("primary", vec![Err("http status: 502 Bad Gateway — a")]); + let (r, _) = link("reserve", vec![Err("timeout: global")]); + let mut f = FailoverProvider::new(vec![p, r]); + let e = f.complete(&[]).unwrap_err().0; + assert!(e.starts_with("timeout: global"), "{e}"); + assert!( + e.contains("after failover from primary: http status: 502"), + "{e}" + ); + } + + #[test] + fn a_single_provider_chain_is_transparent() { + let (p, _) = link( + "primary", + vec![Err("http status: 503 Service Unavailable — x")], + ); + let mut f = FailoverProvider::new(vec![p]); + assert_eq!( + f.complete(&[]).unwrap_err().0, + "http status: 503 Service Unavailable — x" + ); + } +} diff --git a/crates/tole-core/src/lib.rs b/crates/tole-core/src/lib.rs index 0d76f90..fc1cfa4 100644 --- a/crates/tole-core/src/lib.rs +++ b/crates/tole-core/src/lib.rs @@ -33,6 +33,7 @@ pub mod cancel; #[cfg(feature = "shell-tools")] pub mod cora_search; pub mod entry; +pub mod failover; pub mod file_tools; // Tool-call authorization gate (#303): crate-internal, no public API. pub(crate) mod gate; diff --git a/crates/tole-core/src/provider.rs b/crates/tole-core/src/provider.rs index 85a77d5..cbb57cd 100644 --- a/crates/tole-core/src/provider.rs +++ b/crates/tole-core/src/provider.rs @@ -78,4 +78,12 @@ pub trait Provider { fn last_wire_stats(&self) -> Option { None } + + /// Which provider answered the LAST `complete()` call, and whether it + /// took a failover to get there (#209, `failover::FailoverProvider`). + /// Stored by the turn loop under the namespaced `tole_route` key of the + /// step's usage record. Default None: a single provider reports no route. + fn last_route(&self) -> Option { + None + } } diff --git a/crates/tole-core/src/turn.rs b/crates/tole-core/src/turn.rs index eb6a610..8d9448e 100644 --- a/crates/tole-core/src/turn.rs +++ b/crates/tole-core/src/turn.rs @@ -536,7 +536,7 @@ fn drive( // reported wire stats but no usage still leaves a record holding // only `tole_wire` (readers sum `prompt_tokens` etc. with a 0 // default). Neither reported: no record, as before. - let usage = match (p.last_usage(), p.last_wire_stats()) { + let mut usage = match (p.last_usage(), p.last_wire_stats()) { (Some(mut u), Some(w)) => { if let Some(obj) = u.as_object_mut() { obj.insert("tole_wire".to_string(), w); @@ -547,6 +547,13 @@ fn drive( (None, Some(w)) => Some(json!({ "tole_wire": w })), (None, None) => None, }; + // Issue #209: the provider that answered (and any failover) rides + // along under the namespaced `tole_route` key — a switch is durable. + if let Some(route) = p.last_route() { + if let Some(obj) = usage.get_or_insert_with(|| json!({})).as_object_mut() { + obj.insert("tole_route".to_string(), route); + } + } if let Some(u) = usage { if let Some(last) = s.entries().last() { s.commit(Commit::new().usage(UsageRecord {