diff --git a/Cargo.lock b/Cargo.lock index d8d74e512..e2979cf30 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5875,7 +5875,7 @@ dependencies = [ [[package]] name = "pebble-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7" +source = "git+https://github.com/lithoscomputer/pebble?rev=430740f1114f859d6d173683f25b5007cf2023ca#430740f1114f859d6d173683f25b5007cf2023ca" dependencies = [ "async-trait", "futures-util", @@ -5892,7 +5892,7 @@ dependencies = [ [[package]] name = "pebble-cli-core" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7" +source = "git+https://github.com/lithoscomputer/pebble?rev=430740f1114f859d6d173683f25b5007cf2023ca#430740f1114f859d6d173683f25b5007cf2023ca" dependencies = [ "anyhow", "async-trait", @@ -5921,7 +5921,7 @@ dependencies = [ [[package]] name = "pebble-coding-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=4c0063327394cd0f1e9fee2c24541b829b93a8f7#4c0063327394cd0f1e9fee2c24541b829b93a8f7" +source = "git+https://github.com/lithoscomputer/pebble?rev=430740f1114f859d6d173683f25b5007cf2023ca#430740f1114f859d6d173683f25b5007cf2023ca" dependencies = [ "async-trait", "futures-util", diff --git a/Cargo.toml b/Cargo.toml index afb360523..9166bbcbb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -123,9 +123,9 @@ sandbox-driver-testing = { git = "https://github.com/lithoscomputer/sandbox-driv # sandbox-driver revision, so move the two pins together. Pebble pins the # same lithos-llm rev as fabro, and its lockfile policy is that every shared # crate resolves to the version lithos-llm locks. -pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7" } -pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7", features = ["mcp", "search-providers"] } -pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "4c0063327394cd0f1e9fee2c24541b829b93a8f7" } +pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "430740f1114f859d6d173683f25b5007cf2023ca" } +pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "430740f1114f859d6d173683f25b5007cf2023ca", features = ["mcp", "search-providers"] } +pebble-cli-core = { git = "https://github.com/lithoscomputer/pebble", rev = "430740f1114f859d6d173683f25b5007cf2023ca" } sentry = { version = "0.35", default-features = false, features = ["backtrace", "contexts", "ureq", "rustls"] } fork = "0.2" exec = "0.3" diff --git a/docs/internal/events.md b/docs/internal/events.md index abc4134ab..c9e9133f4 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -1618,7 +1618,8 @@ Emitted when the agent fails over to a different LLM provider/model. "from_model": "claude-sonnet-4-20250514", "to_provider": "openai", "to_model": "gpt-4o", - "error": "rate limited" + "error": "rate limited", + "continuation": "continue_turn" } } ``` @@ -1630,6 +1631,19 @@ Emitted when the agent fails over to a different LLM provider/model. | `to_provider` | string | Failover provider | | `to_model` | string | Failover model | | `error` | string | Error that triggered failover | +| `continuation` | string? | How the new route carried the prompt on, as pebble reported it: `replay_prompt` (nothing the prompt committed was in the conversation, so the new route was asked the prompt again) or `continue_turn` (the conversation held assistant output or tool results, so the new route continued from there). Absent on events written before pebble reported it and on one-shot prompt stages, which re-send their request themselves | + +### `agent.route.failover.stopped` + +Pebble's `RouteFailoverStopped` event, stored verbatim like every other +pebble event fabro does not mirror. An agent stage with fallback routes +publishes it when a model failure ends the prompt on its current route +anyway: the failure does not qualify for failover (`reason: "ineligible"`) +or every route has been taken (`reason: "exhausted"`). It follows the +`agent.error` that reports the failure; a stage without fallback routes and +a cancelled prompt publish nothing here. The properties are pebble's +envelope (`seq`, `stream_id`, `session_id`, `timestamp`) plus +`event.RouteFailoverStopped` with `route`, `attempt`, `reason`, and `error`. ### Agent events that are never serialized diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 228a529ff..01ee3eba7 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -1498,6 +1498,7 @@ impl Event { to_model = %props.to_model, requested_reasoning_effort = ?props.requested_reasoning_effort, effective_reasoning_effort = ?props.effective_reasoning_effort, + continuation = ?props.continuation, error = %props.error, "LLM provider failover" ); diff --git a/lib/components/fabro-workflow/src/handler/llm/fallback.rs b/lib/components/fabro-workflow/src/handler/llm/fallback.rs index dcdbee61a..fa86dd93b 100644 --- a/lib/components/fabro-workflow/src/handler/llm/fallback.rs +++ b/lib/components/fabro-workflow/src/handler/llm/fallback.rs @@ -14,6 +14,7 @@ use fabro_types::FailoverProps; use lithos_llm::catalog::ProviderId; use lithos_llm::types::ReasoningEffort; use pebble_coding_agent::FallbackRoute; +use pebble_coding_agent::events::FailoverContinuation; use super::controls::EffectiveRequestControls; use crate::event::{Emitter, Event, StageScope}; @@ -116,13 +117,16 @@ impl FallbackPlan { /// /// `from` may be a route that failed during activation without serving /// traffic; `error` says why it was abandoned. Consecutive payloads - /// chain: one's `to` is the next one's `from`. + /// chain: one's `to` is the next one's `from`. `continuation` is how + /// pebble said the new route carried the prompt on; a one-shot stage, + /// which re-sends its request itself, has none to report. pub(crate) fn failover_props( &self, from: &str, to: &str, attempt: u32, error: &str, + continuation: Option, ) -> FailoverProps { let (from_provider, from_model) = split_selector(from); let (to_provider, to_model) = split_selector(to); @@ -141,6 +145,7 @@ impl FallbackPlan { requested_reasoning_effort: self.original.controls.reasoning_effort, effective_reasoning_effort, error: error.to_string(), + continuation: continuation.map(|continuation| continuation.as_str().to_string()), } } } @@ -283,6 +288,7 @@ pub(crate) fn emit_failover( &plan.current().selector(), plan.attempt(), error, + None, ), }, stage_scope, @@ -376,4 +382,63 @@ mod tests { assert!(!plan.has_next()); assert!(!plan.advance()); } + + #[test] + fn failover_props_carry_the_continuation_pebble_reported() { + let policy = + ModelFallbackPolicy::new(BTreeMap::from([("claude-fable-5".to_string(), vec![ + FallbackTarget::new("openai", "gpt-5.6-sol"), + ])])); + let (plan, notices) = fallback_plan( + &enabled_fallback_catalog(), + &policy, + "claude-fable-5", + &builtin::anthropic(), + EffectiveRequestControls { + reasoning_effort: Some(ReasoningEffort::Medium), + speed: None, + }, + ); + assert!(notices.is_empty()); + + let continued = plan.failover_props( + "anthropic/claude-fable-5", + "openai/gpt-5.6-sol", + 1, + "overloaded", + Some(FailoverContinuation::ContinueTurn), + ); + assert_eq!(continued.continuation.as_deref(), Some("continue_turn")); + assert_eq!(continued.original_provider.as_deref(), Some("anthropic")); + assert_eq!(continued.original_model.as_deref(), Some("claude-fable-5")); + assert_eq!(continued.attempt, Some(1)); + assert_eq!(continued.from_provider, "anthropic"); + assert_eq!(continued.from_model, "claude-fable-5"); + assert_eq!(continued.to_provider, "openai"); + assert_eq!(continued.to_model, "gpt-5.6-sol"); + assert_eq!( + continued.requested_reasoning_effort, + Some(ReasoningEffort::Medium) + ); + assert_eq!(continued.error, "overloaded"); + + let replayed = plan.failover_props( + "anthropic/claude-fable-5", + "openai/gpt-5.6-sol", + 1, + "overloaded", + Some(FailoverContinuation::ReplayPrompt), + ); + assert_eq!(replayed.continuation.as_deref(), Some("replay_prompt")); + + // A one-shot stage walks the plan itself and reports no continuation. + let one_shot = plan.failover_props( + "anthropic/claude-fable-5", + "openai/gpt-5.6-sol", + 1, + "overloaded", + None, + ); + assert_eq!(one_shot.continuation, None); + } } diff --git a/lib/components/fabro-workflow/src/handler/llm/pebble.rs b/lib/components/fabro-workflow/src/handler/llm/pebble.rs index 52ba54737..b21e3e60b 100644 --- a/lib/components/fabro-workflow/src/handler/llm/pebble.rs +++ b/lib/components/fabro-workflow/src/handler/llm/pebble.rs @@ -167,7 +167,9 @@ fn classify_agent_error(error: pebble_coding_agent::Error) -> AgentErrorDisposit /// an MCP server's outcome or disconnect are facts the run already has /// events for, so those are mirrored onto the run's own `agent.failover`, /// `agent.mcp.ready`, `agent.mcp.failed`, and `agent.mcp.disconnected` -/// events instead of being stored twice. +/// events instead of being stored twice. A failover that stops short, with +/// the chain exhausted or the error ineligible, has no event of fabro's own +/// and is stored as pebble's `agent.route.failover.stopped`. struct WorkflowEventSink { emitter: Arc, node_id: String, @@ -184,16 +186,31 @@ impl EventSink for WorkflowEventSink { // watchdog. self.emitter.touch(); match &event.event { + // The failed route's accounting (`usage`, `cost_usd_micros`, + // `inference_ms`, `tool_ms`) is not mirrored: the stage's totals + // already include it through the prompt report, and no run event + // of fabro's own carries per-route usage yet. CodingEvent::RouteFailover { from, to, attempt, error, + usage: _, + cost_usd_micros: _, + inference_ms: _, + tool_ms: _, + continuation, } => { self.emitter.emit_scoped( &Event::Failover { stage: self.node_id.clone(), - props: self.plan.failover_props(from, to, *attempt, &error.message), + props: self.plan.failover_props( + from, + to, + *attempt, + &error.message, + Some(*continuation), + ), }, &self.scope, ); diff --git a/lib/components/fabro-workflow/tests/it/pebble_agent.rs b/lib/components/fabro-workflow/tests/it/pebble_agent.rs index e7a359c97..d8d88a466 100644 --- a/lib/components/fabro-workflow/tests/it/pebble_agent.rs +++ b/lib/components/fabro-workflow/tests/it/pebble_agent.rs @@ -44,7 +44,7 @@ use fabro_workflow::test_support::WorkflowRunner; use httpmock::Method::POST; use httpmock::MockServer; use lithos_llm::catalog::ProviderId; -use pebble_coding_agent::events::CodingEvent; +use pebble_coding_agent::events::{CodingEvent, FailoverStop}; use tokio_util::sync::CancellationToken; const MODEL: &str = "mock-model"; @@ -1103,6 +1103,11 @@ async fn failover_continues_the_conversation_without_rerunning_tools() { "got {}", failover.error ); + assert_eq!( + failover.continuation.as_deref(), + Some("continue_turn"), + "the primary committed a tool result, so the backup continued the turn" + ); let tool_completions = coding_events(&stage.events) .into_iter() .filter(|(_, event)| matches!(event, CodingEvent::ToolCallCompleted { .. })) @@ -1117,6 +1122,129 @@ async fn failover_continues_the_conversation_without_rerunning_tools() { ); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn an_exhausted_fallback_chain_stores_the_stopped_failover() { + let stage = Stage::new().await; + let revoked = |then: httpmock::Then, key: &str| { + then.status(401) + .header("content-type", "application/json") + .json_body(serde_json::json!({ + "error": { "message": format!("{key} key revoked"), "type": "invalid_request_error" } + })); + }; + let primary = stage + .server + .mock_async(|when, then| { + when.method(POST).path("/primary/v1/chat/completions"); + revoked(then, "primary"); + }) + .await; + let backup = stage + .server + .mock_async(|when, then| { + when.method(POST).path("/backup/v1/chat/completions"); + revoked(then, "backup"); + }) + .await; + + let overlay = format!( + "{}\n{}", + provider_toml( + "primary", + "primary-model", + &stage.server.url("/primary/v1"), + "openai" + ), + provider_toml( + "backup", + "backup-model", + &stage.server.url("/backup/v1"), + "openai" + ), + ); + let catalog = Arc::new(fabro_llm::test_support::test_catalog_with_overlay(&overlay)); + let primary_provider = ProviderId::new("primary"); + let fallbacks = model_fallback::resolve_model_fallbacks( + &catalog, + &[primary_provider.clone(), ProviderId::new("backup")], + &BTreeMap::from([("primary-model".to_string(), vec![ + "backup/backup-model".parse::().unwrap(), + ])]), + ) + .expect("the fallback chain resolves"); + let backend = PebbleBackend::new_with_catalog( + "primary-model".to_string(), + primary_provider, + fallbacks.policy, + mock_credentials(), + Arc::clone(&stage.hub), + catalog, + ); + + let mut graph = agent_graph("Exhausted", "Say hello"); + let work = graph.nodes.get_mut("work").unwrap(); + work.attrs + .insert("max_retries".to_string(), AttrValue::Integer(0)); + graph.edges.retain(|edge| edge.from != "work"); + let mut fail_edge = Edge::new("work", "exit"); + fail_edge.attrs.insert( + "condition".to_string(), + AttrValue::String("outcome=failed".to_string()), + ); + graph.edges.push(fail_edge); + + let (_, state) = stage + .run(backend, &graph, CancellationToken::new()) + .await + .expect("the fail edge carries the run to exit"); + + assert_eq!(primary.calls_async().await, 1); + assert_eq!(backup.calls_async().await, 1); + assert_eq!( + work_stage(&state) + .completion + .as_ref() + .expect("the work stage completes") + .outcome, + StageOutcome::Failed { + retry_requested: false, + } + ); + + // The move to the backup is fabro's own event; the stop on the backup + // is pebble's, stored under its derived name after the error it reports. + assert_eq!(count(&stage.events, "agent.failover"), 1); + assert_eq!(count(&stage.events, "agent.route.failover.stopped"), 1); + let stopped_at = position(&stage.events, "agent.route.failover.stopped").unwrap(); + assert!(work_stage_event(&stage.events, stopped_at)); + let error_at = position(&stage.events, "agent.error").expect("the model error is stored"); + assert!( + error_at < stopped_at, + "the stop follows the error, got {:?}", + names(&stage.events) + ); + let (route, attempt, reason, error) = coding_events(&stage.events) + .into_iter() + .find_map(|(_, event)| match event { + CodingEvent::RouteFailoverStopped { + route, + attempt, + reason, + error, + } => Some((route, attempt, reason, error)), + _ => None, + }) + .expect("the stopped failover is stored as pebble's event"); + assert_eq!(route, "backup/backup-model"); + assert_eq!(attempt, 1); + assert_eq!(reason, FailoverStop::Exhausted); + assert!( + error.message.contains("backup key revoked"), + "got {}", + error.message + ); +} + // --- Durability // --------------------------------------------------------------- diff --git a/lib/foundation/fabro-types/src/run_event/agent.rs b/lib/foundation/fabro-types/src/run_event/agent.rs index 91504f375..0776172f7 100644 --- a/lib/foundation/fabro-types/src/run_event/agent.rs +++ b/lib/foundation/fabro-types/src/run_event/agent.rs @@ -77,6 +77,7 @@ pub fn coding_event_name(event: &CodingEvent) -> &'static str { CodingEvent::LoopDetected => "agent.loop.detected", CodingEvent::ToolRoundsExhausted { .. } => "agent.tool.rounds.exhausted", CodingEvent::RouteFailover { .. } => "agent.route.failover", + CodingEvent::RouteFailoverStopped { .. } => "agent.route.failover.stopped", CodingEvent::McpServerReady { .. } => "agent.mcp.server.ready", CodingEvent::McpServerFailed { .. } => "agent.mcp.server.failed", CodingEvent::McpServerDisconnected { .. } => "agent.mcp.server.disconnected", @@ -125,6 +126,7 @@ pub const CODING_EVENT_NAMES: &[&str] = &[ "agent.loop.detected", "agent.tool.rounds.exhausted", "agent.route.failover", + "agent.route.failover.stopped", "agent.mcp.server.ready", "agent.mcp.server.failed", "agent.mcp.server.disconnected", @@ -280,7 +282,7 @@ pub struct AgentMcpDisconnectedProps { mod tests { use std::time::{Duration, UNIX_EPOCH}; - use pebble_coding_agent::events::TokenUsage; + use pebble_coding_agent::events::{ErrorData, ErrorKind, FailoverStop, TokenUsage}; use serde_json::json; use super::*; @@ -356,4 +358,16 @@ mod tests { assert!(is_coding_event_name("todo.updated")); assert!(!is_coding_event_name("agent.session.activated")); } + + #[test] + fn a_stopped_failover_has_its_own_name() { + let stopped = CodingEvent::RouteFailoverStopped { + route: "anthropic/claude-fable-5".to_string(), + attempt: 2, + reason: FailoverStop::Exhausted, + error: ErrorData::new(ErrorKind::Llm, "overloaded"), + }; + assert_eq!(coding_event_name(&stopped), "agent.route.failover.stopped"); + assert!(is_coding_event_name("agent.route.failover.stopped")); + } } diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index 19406b57b..85f07ac22 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -268,6 +268,15 @@ pub struct FailoverProps { #[serde(default, skip_serializing_if = "Option::is_none")] pub effective_reasoning_effort: Option, pub error: String, + /// How the new route carried the prompt on, as pebble reported it: + /// `replay_prompt` when nothing the prompt committed was in the + /// conversation and the new route was asked the prompt again, or + /// `continue_turn` when the conversation held assistant output or tool + /// results and the new route continued from there. Absent on events + /// written before pebble reported it, and on one-shot prompt stages, + /// which walk the plan themselves and always re-send the request. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub continuation: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/lib/foundation/fabro-types/src/run_event/mod.rs b/lib/foundation/fabro-types/src/run_event/mod.rs index e4d48f71c..d88a3b07a 100644 --- a/lib/foundation/fabro-types/src/run_event/mod.rs +++ b/lib/foundation/fabro-types/src/run_event/mod.rs @@ -1268,6 +1268,44 @@ mod tests { assert_eq!(props.attempt, None); assert_eq!(props.requested_reasoning_effort, None); assert_eq!(props.effective_reasoning_effort, None); + assert_eq!(props.continuation, None); + } + + #[test] + fn failover_event_round_trips_its_continuation() { + let body = EventBody::Failover(FailoverProps { + original_provider: Some("anthropic".to_string()), + original_model: Some("claude-fable-5".to_string()), + attempt: Some(1), + from_provider: "anthropic".to_string(), + from_model: "claude-fable-5".to_string(), + to_provider: "openai".to_string(), + to_model: "gpt-5.6-sol".to_string(), + requested_reasoning_effort: None, + effective_reasoning_effort: None, + error: "overloaded".to_string(), + continuation: Some("continue_turn".to_string()), + }); + let value = serde_json::to_value(&body).unwrap(); + assert_eq!(value["event"], "agent.failover"); + assert_eq!(value["properties"]["continuation"], "continue_turn"); + let parsed: EventBody = serde_json::from_value(value).unwrap(); + assert_eq!(parsed, body); + + // A one-shot stage, or an event written before pebble reported the + // continuation, omits the field rather than writing `null`. + let EventBody::Failover(mut props) = body else { + unreachable!() + }; + props.continuation = None; + let value = serde_json::to_value(EventBody::Failover(props)).unwrap(); + assert!( + value["properties"] + .as_object() + .unwrap() + .get("continuation") + .is_none() + ); } #[test]