From 57a746235e0aed3f92c1553a584cb58feb5f9d05 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 12 Sep 2026 16:26:32 -0600 Subject: [PATCH 1/3] Pin pebble to main after lithoscomputer/pebble#12 Pebble's RouteFailover event now describes the route that failed and how the new route carried the prompt on. Record the continuation on fabro's agent.failover event as an optional string (replay_prompt or continue_turn); events written before it existed, and one-shot prompt stages that walk the plan themselves, read as absent. The failed route's usage, cost, and timing are not mirrored: the stage's totals already include them through the prompt report, and no fabro run event carries per-route usage yet. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 6 +- Cargo.toml | 6 +- .../fabro-workflow/src/event/events.rs | 1 + .../src/handler/llm/fallback.rs | 67 ++++++++++++++++++- .../fabro-workflow/src/handler/llm/pebble.rs | 21 +++++- .../fabro-types/src/run_event/misc.rs | 9 +++ .../fabro-types/src/run_event/mod.rs | 38 +++++++++++ 7 files changed, 139 insertions(+), 9 deletions(-) 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/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/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] From 252e7e064321b729b6b3ce15da0a5d29fa3165be Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 12 Sep 2026 16:26:32 -0600 Subject: [PATCH 2/3] Name pebble's stopped failover agent.route.failover.stopped Pebble publishes RouteFailoverStopped when a model error ends a prompt on its route although fallback routes were configured: the error is ineligible or the chain is exhausted. Fabro has no event of its own for that case, so the pebble event is stored as it is under a derived name next to agent.route.failover instead of the generic agent.event. Co-Authored-By: Claude Fable 5.1 --- .../fabro-types/src/run_event/agent.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) 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")); + } } From 9550ff08069834d9379c090987ea0d027dc068e3 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 12 Sep 2026 16:26:32 -0600 Subject: [PATCH 3/3] Cover the failover continuation and the stopped failover The existing failover test asserts the backup continued the turn after the primary committed a tool result. A new test exhausts a two-route chain and checks that one agent.failover and one agent.route.failover.stopped are stored on the work stage, the stop after the error it reports, with the exhausted reason and the failing route. The events catalog documents both. Co-Authored-By: Claude Fable 5.1 --- docs/internal/events.md | 16 ++- .../fabro-workflow/tests/it/pebble_agent.rs | 130 +++++++++++++++++- 2 files changed, 144 insertions(+), 2 deletions(-) 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/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 // ---------------------------------------------------------------