From b5fd907c3d2f14cc7be2f8fa73ad504d11f2fd82 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 11 Sep 2026 18:47:12 -0600 Subject: [PATCH] Take files touched and route failover from pebble Steps 1 and 2 of .ai/plans/pebble-absorbs-embedder-concerns.md, pinning pebble 1a5abe4. Files touched come from `PromptReport`: pebble computes them from every successful write, edit, and patch across the prompt, subagents included, so the event-fed `FileTracking` and the tracking half of `WorkflowEventSink` go. The stage unions the reports of its prompts. Route failover is pebble's. The stage resolves its plan from the catalog as before and hands pebble the remaining routes through `fallback_routes`, each with its controls and the stage's output limit. Pebble keeps the conversation, moves it to the next route, requeues pending steering, and continues the prompt; the stage's plan follows the route the report says the prompt ended on, re-activates the session there, and mirrors pebble's `RouteFailover` as the run's `agent.failover` event with the same payload as before. `prompt_with_failover`, `resume_agent_on_route`, and the route bookkeeping in `LiveAgent` go. One-shot prompt stages still walk the plan themselves. `agent.route.failover` and `agent.tool.rounds.exhausted` join the derived event names; both variants were falling back to `agent.event`. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 24 +- Cargo.toml | 4 +- .../src/handler/llm/fallback.rs | 110 ++++- .../fabro-workflow/src/handler/llm/pebble.rs | 409 ++++++------------ .../fabro-types/src/run_event/agent.rs | 4 + 5 files changed, 230 insertions(+), 321 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 267957002..cbad92433 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2063,7 +2063,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2190,7 +2190,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5299,7 +5299,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5864,7 +5864,7 @@ dependencies = [ [[package]] name = "pebble-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=4db661c44b09048db28d892568382408fbbc6c6e#4db661c44b09048db28d892568382408fbbc6c6e" +source = "git+https://github.com/lithoscomputer/pebble?rev=1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5#1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5" dependencies = [ "async-trait", "futures-util", @@ -5881,7 +5881,7 @@ dependencies = [ [[package]] name = "pebble-coding-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=4db661c44b09048db28d892568382408fbbc6c6e#4db661c44b09048db28d892568382408fbbc6c6e" +source = "git+https://github.com/lithoscomputer/pebble?rev=1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5#1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5" dependencies = [ "async-trait", "futures-util", @@ -6246,7 +6246,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -6723,7 +6723,7 @@ dependencies = [ "errno 0.3.14", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6782,7 +6782,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7426,7 +7426,7 @@ version = "1.4.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" dependencies = [ - "errno 0.2.8", + "errno 0.3.14", "libc", ] @@ -7991,7 +7991,7 @@ dependencies = [ "getrandom 0.4.1", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -8026,7 +8026,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -9093,7 +9093,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 981294dd4..b4ed9beba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -113,8 +113,8 @@ futures-util = "0.3" # the merge commit once it lands. 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 = "4db661c44b09048db28d892568382408fbbc6c6e" } -pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "4db661c44b09048db28d892568382408fbbc6c6e" } +pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5" } +pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "1a5abe4a83b3c13f8ecbb38f761d17ede8e3d7d5" } sandbox-driver = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } sandbox-driver-protocol = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } sandbox-driver-host = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } diff --git a/lib/components/fabro-workflow/src/handler/llm/fallback.rs b/lib/components/fabro-workflow/src/handler/llm/fallback.rs index 4df99e0a9..dcdbee61a 100644 --- a/lib/components/fabro-workflow/src/handler/llm/fallback.rs +++ b/lib/components/fabro-workflow/src/handler/llm/fallback.rs @@ -2,7 +2,10 @@ //! //! The plan belongs to the originally requested model: advancing it never //! activates a target model's own chain. `model_fallback.rs` decides the -//! policy; this module walks it and records each failover as a run event. +//! policy; this module resolves it against the catalog and records each +//! failover as a run event. Agent stages hand the resolved routes to pebble +//! ([`FallbackPlan::pebble_routes`]), which executes them and reports each +//! move as `RouteFailover`; one-shot prompt stages walk the plan themselves. use fabro_graphviz::graph::Node; use fabro_llm::FallbackTarget; @@ -10,6 +13,7 @@ use fabro_llm::lithos_catalog::Catalog; use fabro_types::FailoverProps; use lithos_llm::catalog::ProviderId; use lithos_llm::types::ReasoningEffort; +use pebble_coding_agent::FallbackRoute; use super::controls::EffectiveRequestControls; use crate::event::{Emitter, Event, StageScope}; @@ -71,6 +75,83 @@ impl FallbackPlan { false } } + + /// Moves to the route whose `provider/model` selector is `selector`, the + /// route pebble reports a prompt ended on. Returns whether the position + /// changed; a selector the plan does not know leaves it where it was. + pub(crate) fn advance_to(&mut self, selector: &str) -> bool { + if self.current().selector() == selector { + return false; + } + match self + .remaining + .iter() + .position(|route| route.selector() == selector) + { + Some(index) => { + self.position = index + 1; + true + } + None => false, + } + } + + /// The routes after the current one, as pebble executes them: each with + /// its own controls and the stage's output limit. + pub(crate) fn pebble_routes(&self, max_tokens: Option) -> Vec { + self.remaining + .iter() + .skip(self.position) + .map(|route| { + FallbackRoute::new(route.selector()) + .with_reasoning_effort(route.controls.reasoning_effort) + .with_speed(route.controls.speed) + .with_max_tokens(max_tokens) + }) + .collect() + } + + /// The `agent.failover` payload for a move from `from` to `to`, both + /// `provider/model` selectors, on this plan. + /// + /// `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`. + pub(crate) fn failover_props( + &self, + from: &str, + to: &str, + attempt: u32, + error: &str, + ) -> FailoverProps { + let (from_provider, from_model) = split_selector(from); + let (to_provider, to_model) = split_selector(to); + let effective_reasoning_effort = std::iter::once(&self.original) + .chain(self.remaining.iter()) + .find(|route| route.selector() == to) + .and_then(|route| route.controls.reasoning_effort); + FailoverProps { + original_provider: Some(self.original.target.provider.to_string()), + original_model: Some(self.original.target.model.to_string()), + attempt: Some(attempt), + from_provider, + from_model, + to_provider, + to_model, + requested_reasoning_effort: self.original.controls.reasoning_effort, + effective_reasoning_effort, + error: error.to_string(), + } + } +} + +/// A `provider/model` selector split at its first slash; a selector with no +/// slash is all model. +fn split_selector(selector: &str) -> (String, String) { + match selector.split_once('/') { + Some((provider, model)) => (provider.to_string(), model.to_string()), + None => (String::new(), selector.to_string()), + } } /// Request controls resolved for one fallback target. @@ -186,12 +267,7 @@ pub(crate) fn fallback_plan( } /// Emit `agent.failover` for the plan's most recent -/// [`FallbackPlan::advance`]. -/// -/// `from` is the previously attempted candidate, which may have failed -/// during activation without ever serving traffic; `error` says why it -/// was abandoned. Consecutive events therefore chain — one event's `to` -/// is the next event's `from` — recording every candidate the plan tried. +/// [`FallbackPlan::advance`], on a one-shot stage that walks the plan itself. pub(crate) fn emit_failover( node: &Node, emitter: &Emitter, @@ -199,23 +275,15 @@ pub(crate) fn emit_failover( plan: &FallbackPlan, error: &str, ) { - let from = plan.previous(); - let to = plan.current(); emitter.emit_scoped( &Event::Failover { stage: node.id.clone(), - props: FailoverProps { - original_provider: Some(plan.original.target.provider.to_string()), - original_model: Some(plan.original.target.model.to_string()), - attempt: Some(plan.attempt()), - from_provider: from.target.provider.to_string(), - from_model: from.target.model.to_string(), - to_provider: to.target.provider.to_string(), - to_model: to.target.model.to_string(), - requested_reasoning_effort: plan.original.controls.reasoning_effort, - effective_reasoning_effort: to.controls.reasoning_effort, - error: error.to_string(), - }, + props: plan.failover_props( + &plan.previous().selector(), + &plan.current().selector(), + plan.attempt(), + error, + ), }, stage_scope, ); diff --git a/lib/components/fabro-workflow/src/handler/llm/pebble.rs b/lib/components/fabro-workflow/src/handler/llm/pebble.rs index 63040474c..30732b03a 100644 --- a/lib/components/fabro-workflow/src/handler/llm/pebble.rs +++ b/lib/components/fabro-workflow/src/handler/llm/pebble.rs @@ -4,10 +4,12 @@ //! One agent serves one stage invocation. At `full` fidelity, stages sharing a //! `thread_id` continue one conversation: the agent is exported when a stage //! ends and resumed by the next, which binds its own event scope, hooks, and -//! interviewer. Model failover keeps the conversation as it stands and asks -//! the next route to continue it, so no tool effect repeats. +//! interviewer. Model failover is pebble's: the stage hands it the resolved +//! fallback routes, pebble keeps the conversation as it stands and asks the +//! next route to continue it, and this module mirrors each move as the run's +//! `agent.failover` event. -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeSet, HashMap, HashSet}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -16,7 +18,7 @@ use fabro_graphviz::graph::Node; use fabro_llm::credentials::CredentialProvider; use fabro_llm::lithos_catalog::Catalog; use fabro_llm::types::ResponseFormat; -use fabro_llm::{Client, ClientOptions, ErrorData, Request, Response}; +use fabro_llm::{Client, ClientOptions, Request, Response}; use fabro_mcp::config::McpServerSettings; use fabro_mcp::connection_manager::McpConnectionManager; use fabro_sandbox::{RunSandbox, SecretRedactor}; @@ -34,12 +36,12 @@ use pebble_coding_agent::events::{ Actor, CodingAgentEvent, CodingEvent, EventSink, EventSinkError, }; use pebble_coding_agent::extensions::HumanInputProvider; -use pebble_coding_agent::state::{Message, SessionRecord}; +use pebble_coding_agent::state::Message; use pebble_coding_agent::subagents::SubagentOptions; -use pebble_coding_agent::tools::{RegisteredTool, ToolEnvProvider, canonical_tool_name}; +use pebble_coding_agent::tools::{RegisteredTool, ToolEnvProvider}; use pebble_coding_agent::{ CodingAgent, CodingAgentBuilder, CodingAgentControlHandle, CodingAgentExport, - CodingAgentOptions, CodingInput, InterruptReason, ResumeMode, ShutdownReason, SteeringLease, + CodingAgentOptions, CodingInput, InterruptReason, ShutdownReason, SteeringLease, SteeringMessage, SteeringOutcome, }; use tokio_util::sync::CancellationToken; @@ -106,25 +108,19 @@ struct CachedThread { } /// How the backend reports a failed prompt. +/// +/// A model error reaches this after pebble has followed every fallback route +/// the stage gave it, so it is terminal here whatever its kind. enum AgentErrorDisposition { /// The run's token cancelled the prompt; surface as `Error::Cancelled`. Cancelled, - /// Underlying LLM error eligible for provider failover. - FailoverEligible(ErrorData), /// Terminal error; abort the invocation with this workflow `Error`. Terminal(Error), } -fn classify_agent_error( - error: pebble_coding_agent::Error, - allow_failover: bool, -) -> AgentErrorDisposition { +fn classify_agent_error(error: pebble_coding_agent::Error) -> AgentErrorDisposition { if let Some(llm) = error.llm_source() { - let data = llm.data(); - if allow_failover && llm.failover_eligible() { - return AgentErrorDisposition::FailoverEligible(data); - } - return AgentErrorDisposition::Terminal(Error::from(data)); + return AgentErrorDisposition::Terminal(Error::from(llm.data())); } match error { pebble_coding_agent::Error::Interrupted(InterruptReason::Cancelled) => { @@ -150,6 +146,11 @@ fn classify_agent_error( pebble_coding_agent::Error::EventSink(sink) => AgentErrorDisposition::Terminal(Error::Io( format!("Failed to persist agent events: {sink:#}"), )), + pebble_coding_agent::Error::FallbackRoute { route, source } => { + AgentErrorDisposition::Terminal(Error::Precondition(format!( + "Fallback route {route} could not be started: {source:#}" + ))) + } // `InterruptReason` may grow; a reason this build does not know still // ended the prompt. pebble_coding_agent::Error::Interrupted(_) => AgentErrorDisposition::Terminal( @@ -163,99 +164,16 @@ fn classify_agent_error( // --- Event sink ----------------------------------------------------------- -/// Files a stage's tool calls changed, paired from `ToolCallStarted` -/// arguments and a successful `ToolCallCompleted`. -#[derive(Default)] -struct FileTracking { - /// `tool_call_id` → paths for in-flight write, edit, and patch calls. - pending: HashMap>, - /// Every path successfully written. - touched: HashSet, - /// The most recently written path. - last: Option, -} - -impl FileTracking { - fn snapshot(&self) -> (Vec, Option) { - let mut files: Vec = self.touched.iter().cloned().collect(); - files.sort(); - (files, self.last.clone()) - } -} - -/// The paths a tool call will write, from its arguments. -fn written_paths(tool_name: &str, arguments: &serde_json::Value) -> Vec { - match canonical_tool_name(tool_name) { - "write_file" | "edit_file" => arguments - .get("file_path") - .or_else(|| arguments.get("path")) - .and_then(serde_json::Value::as_str) - .map(|path| vec![path.to_string()]) - .unwrap_or_default(), - "apply_patch" => { - let patch = arguments - .as_str() - .or_else(|| arguments.get("patch").and_then(serde_json::Value::as_str)) - .unwrap_or_default(); - patch_written_paths(patch) - } - _ => Vec::new(), - } -} - -/// The files an `apply_patch` patch creates or changes, in patch order. -fn patch_written_paths(patch: &str) -> Vec { - const MARKERS: [&str; 3] = ["*** Add File: ", "*** Update File: ", "*** Move to: "]; - patch - .lines() - .filter_map(|line| { - MARKERS - .iter() - .find_map(|marker| line.strip_prefix(marker)) - .map(|path| path.trim().to_string()) - }) - .filter(|path| !path.is_empty()) - .collect() -} - -fn track_file_event(event: &CodingEvent, state: &mut FileTracking) { - match event { - CodingEvent::ToolCallStarted { - tool_name, - tool_call_id, - arguments, - } => { - let paths = written_paths(tool_name, arguments); - if !paths.is_empty() { - state.pending.insert(tool_call_id.clone(), paths); - } - } - CodingEvent::ToolCallCompleted { - tool_call_id, - is_error, - .. - } => { - if let Some(paths) = state.pending.remove(tool_call_id) { - if !*is_error { - for path in paths { - state.touched.insert(path.clone()); - state.last = Some(path); - } - } - } - } - _ => {} - } -} - /// Pebble's durable event sink for one stage: every agent event becomes a -/// run event in the run's log before the agent goes on, and the stage's -/// file tracking sees it on the way. +/// run event in the run's log before the agent goes on, and a route failover +/// is mirrored as the run's own `agent.failover` event on the way. struct WorkflowEventSink { - emitter: Arc, - node_id: String, - scope: StageScope, - file_tracking: Arc>, + emitter: Arc, + node_id: String, + scope: StageScope, + /// The stage's resolved plan, for the controls and origin the mirrored + /// failover event names. + plan: FallbackPlan, } #[async_trait] @@ -264,13 +182,21 @@ impl EventSink for WorkflowEventSink { // Every event, including streaming deltas, resets the run's activity // watchdog. self.emitter.touch(); - track_file_event( - &event.event, - &mut self - .file_tracking - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner), - ); + if let CodingEvent::RouteFailover { + from, + to, + attempt, + error, + } = &event.event + { + self.emitter.emit_scoped( + &Event::Failover { + stage: self.node_id.clone(), + props: self.plan.failover_props(from, to, *attempt, &error.message), + }, + &self.scope, + ); + } // Deltas and the prompt's own durability barrier are not run history. if event.event.is_streaming_noise() || matches!(event.event, CodingEvent::ProcessingEnd) { return Ok(()); @@ -396,8 +322,9 @@ impl ActiveControlHandle for PebbleControlHandle { /// One stage invocation's live agent and its accounting. /// -/// Failover replaces the agent while the accumulated usage, cost, and timing -/// keep counting across routes. +/// A stage may run several prompts on one agent (the prompt, output repairs, +/// late steering); the usage, cost, timing, and files of every one of them +/// are summed here, across whatever routes pebble moved through. struct LiveAgent { agent: CodingAgent, handle: Arc, @@ -407,9 +334,32 @@ struct LiveAgent { total_cost: Option, inference_duration: Duration, tool_duration: Duration, + /// Every file the stage's prompts wrote or edited, subagents included. + files_touched: BTreeSet, + /// The most recently written path. + last_file_touched: Option, } impl LiveAgent { + fn new( + agent: CodingAgent, + handle: Arc, + mcp: Option>, + ) -> Self { + Self { + agent, + handle, + lease: None, + mcp, + total_usage: TokenCounts::default(), + total_cost: None, + inference_duration: Duration::ZERO, + tool_duration: Duration::ZERO, + files_touched: BTreeSet::new(), + last_file_touched: None, + } + } + fn record_report(&mut self, report: &pebble_coding_agent::PromptReport) { billing::add_usage(&mut self.total_usage, TokenCounts::from(report.usage)); UsdMicros::accumulate( @@ -422,6 +372,11 @@ impl LiveAgent { .inference_duration .saturating_add(report.timing.inference); self.tool_duration = self.tool_duration.saturating_add(report.timing.tool); + self.files_touched + .extend(report.files_touched.iter().cloned()); + if report.last_file_touched.is_some() { + self.last_file_touched.clone_from(&report.last_file_touched); + } } fn release_lease(&mut self) { @@ -461,7 +416,6 @@ struct StageBindings<'a> { sandbox: &'a Arc, tool_middleware: Option<&'a Arc>, human_input: Option<&'a Arc>, - file_tracking: &'a Arc>, } impl PebbleBackend { @@ -685,16 +639,19 @@ impl PebbleBackend { tools } - /// Bind the stage's services and this route's policy to `builder`. + /// Bind the stage's services, the plan's current route, and the routes + /// left to fail over to, to `builder`. fn bind_builder( &self, mut builder: CodingAgentBuilder, node: &Node, - route: &LlmRoute, + plan: &FallbackPlan, provider: &ProviderContext, bindings: &StageBindings<'_>, mcp: Option<&Arc>, ) -> CodingAgentBuilder { + let route = plan.current(); + let max_tokens = node_max_output_tokens(node).map(i64::from); builder = builder .tools(self.stage_tools(mcp)) .permission_level(PermissionLevel::Full) @@ -704,11 +661,12 @@ impl PebbleBackend { route.controls, bindings.sandbox, )) + .fallback_routes(plan.pebble_routes(max_tokens)) .event_sink(Arc::new(WorkflowEventSink { - emitter: Arc::clone(bindings.emitter), - node_id: bindings.node_id.to_string(), - scope: bindings.stage_scope.clone(), - file_tracking: Arc::clone(bindings.file_tracking), + emitter: Arc::clone(bindings.emitter), + node_id: bindings.node_id.to_string(), + scope: bindings.stage_scope.clone(), + plan: plan.clone(), })) .redactor(Arc::new(SecretRedactor)) .subagents(SubagentOptions::enabled()); @@ -730,11 +688,11 @@ impl PebbleBackend { builder } - /// A new agent on `route`. + /// A new agent on the plan's current route. async fn build_agent( &self, node: &Node, - route: &LlmRoute, + plan: &FallbackPlan, provider: &ProviderContext, bindings: &StageBindings<'_>, mcp: Option<&Arc>, @@ -742,20 +700,20 @@ impl PebbleBackend { let client = self.build_llm_client().await?; let environment: Arc = Arc::clone(bindings.sandbox) as Arc; - let builder = CodingAgent::builder(client, environment).model(route.selector()); - self.bind_builder(builder, node, route, provider, bindings, mcp) + let builder = CodingAgent::builder(client, environment).model(plan.current().selector()); + self.bind_builder(builder, node, plan, provider, bindings, mcp) .build() .await .map_err(|error| Error::handler_with_source("Failed to start agent session", error)) } /// The exported conversation of an earlier stage, continued on the - /// route it was on. + /// route it was on, with the routes it had left. async fn resume_exported_agent( &self, export: CodingAgentExport, node: &Node, - route: &LlmRoute, + plan: &FallbackPlan, provider: &ProviderContext, bindings: &StageBindings<'_>, mcp: Option<&Arc>, @@ -764,39 +722,12 @@ impl PebbleBackend { let environment: Arc = Arc::clone(bindings.sandbox) as Arc; let builder = CodingAgent::resume_from_export(client, environment, export); - self.bind_builder(builder, node, route, provider, bindings, mcp) + self.bind_builder(builder, node, plan, provider, bindings, mcp) .build() .await .map_err(|error| Error::handler_with_source("Failed to resume agent session", error)) } - /// The conversation as it stands, continued on a fallback route. - async fn resume_agent_on_route( - &self, - record: SessionRecord, - node: &Node, - route: &LlmRoute, - provider: &ProviderContext, - bindings: &StageBindings<'_>, - mcp: Option<&Arc>, - ) -> Result { - let client = self.build_llm_client().await?; - let environment: Arc = - Arc::clone(bindings.sandbox) as Arc; - let builder = CodingAgent::resume( - client, - environment, - record, - ResumeMode::UseModel(route.selector()), - ); - self.bind_builder(builder, node, route, provider, bindings, mcp) - .build() - .await - .map_err(|error| { - Error::handler_with_source("Failed to resume agent session on fallback", error) - }) - } - /// Register `live` with the steering hub so steers reach it, and tell /// the run which tools it has. fn activate( @@ -835,13 +766,14 @@ impl PebbleBackend { Ok(()) } - /// Run `input` on `live`, following the fallback plan when the model - /// fails. On success the agent that answered is in `live`. - async fn prompt_with_failover( + /// Run `input` on `live`. Pebble follows the stage's fallback routes + /// itself; the plan here follows the route the prompt ended on, so a + /// later prompt of this stage and a successor on the thread start there, + /// and the run hears which route the session is on now. + async fn prompt_live( &self, live: &mut LiveAgent, input: CodingInput, - node: &Node, fallback_plan: &mut FallbackPlan, stage_id: &StageId, thread_id: Option<&str>, @@ -853,93 +785,17 @@ impl PebbleBackend { .prompt_with_cancellation(input, cancel_token) .await; live.record_report(&report); - let mut last_error = match report.result { - Ok(output) => { - return Ok(output.text.unwrap_or_else(|| live.last_assistant_text())); - } - Err(error) => match classify_agent_error(error, fallback_plan.has_next()) { - AgentErrorDisposition::Cancelled => return Err(Error::Cancelled), - AgentErrorDisposition::Terminal(error) => return Err(error), - AgentErrorDisposition::FailoverEligible(error) => Error::from(error), - }, - }; - - while fallback_plan.advance() { - fallback::emit_failover( - node, - bindings.emitter, - bindings.stage_scope, - fallback_plan, - &last_error.to_string(), - ); - let route = fallback_plan.current().clone(); - let provider = match self.resolve_provider_context( - route.target.model.as_str(), - Some(route.target.provider.as_str()), - ) { - Ok(provider) => provider, - Err(error) => { - last_error = error; - continue; - } - }; - if cancel_token.is_cancelled() { - return Err(Error::Cancelled); - } - - // The record holds the prompt and every committed tool result, so - // the next route continues the conversation as it stands and no - // tool effect repeats. Steering the failed agent still held moves - // with it. - let mut record = live.agent.to_record(); - let pending = live.handle.control.take_pending_input(); - live.discard(ShutdownReason::Error).await; - record.advance_event_cursor(live.agent.committed_event_seq()); - - let mcp = live.mcp.clone(); - let agent = self - .resume_agent_on_route(record, node, &route, &provider, bindings, mcp.as_ref()) - .await; - if cancel_token.is_cancelled() { - return Err(Error::Cancelled); - } - live.agent = match agent { - Ok(agent) => agent, - Err(error) => { - last_error = error; - continue; - } - }; - live.handle = Arc::new(PebbleControlHandle::new(live.agent.control_handle())); - let (steering, follow_ups) = pending.into_parts(); - for message in steering { - live.handle.control.queue_steering(message); - } - for message in follow_ups { - live.handle.control.queue_follow_up(message); - } - self.activate(live, &route, stage_id, thread_id, bindings)?; - - let report = live - .agent - .continue_prompt_with_cancellation(cancel_token) - .await; - live.record_report(&report); - match report.result { - Ok(output) => { - return Ok(output.text.unwrap_or_else(|| live.last_assistant_text())); - } - Err(error) => match classify_agent_error(error, fallback_plan.has_next()) { - AgentErrorDisposition::Cancelled => return Err(Error::Cancelled), - AgentErrorDisposition::Terminal(error) => return Err(error), - AgentErrorDisposition::FailoverEligible(error) => { - last_error = Error::from(error); - } - }, - } + if fallback_plan.advance_to(&report.route) { + live.release_lease(); + self.activate(live, fallback_plan.current(), stage_id, thread_id, bindings)?; + } + match report.result { + Ok(output) => Ok(output.text.unwrap_or_else(|| live.last_assistant_text())), + Err(error) => match classify_agent_error(error) { + AgentErrorDisposition::Cancelled => Err(Error::Cancelled), + AgentErrorDisposition::Terminal(error) => Err(error), + }, } - - Err(last_error) } /// Steers that landed between the answer and the hub's close-the-door @@ -948,7 +804,6 @@ impl PebbleBackend { async fn drain_late_steering( &self, live: &mut LiveAgent, - node: &Node, fallback_plan: &mut FallbackPlan, stage_id: &StageId, thread_id: Option<&str>, @@ -968,10 +823,9 @@ impl PebbleBackend { let (steering, follow_ups) = live.handle.control.take_pending_input().into_parts(); for message in steering.into_iter().chain(follow_ups) { response = self - .prompt_with_failover( + .prompt_live( live, CodingInput::from(message.content().clone()), - node, fallback_plan, stage_id, thread_id, @@ -1227,7 +1081,6 @@ impl CodergenBackend for PebbleBackend { } let stage_scope = StageScope::for_handler(request.context, &node.id); let stage_id = stage_scope.stage_id(); - let file_tracking = Arc::new(Mutex::new(FileTracking::default())); let bindings = StageBindings { node_id: &node.id, stage_scope: &stage_scope, @@ -1235,7 +1088,6 @@ impl CodergenBackend for PebbleBackend { sandbox: request.sandbox, tool_middleware: request.tool_middleware.as_ref(), human_input: request.human_input.as_ref(), - file_tracking: &file_tracking, }; let cached = reuse_key.as_ref().and_then(|key| self.take_thread(key)); @@ -1250,7 +1102,7 @@ impl CodergenBackend for PebbleBackend { .resume_exported_agent( thread.export, node, - &route, + &thread.fallback_plan, &provider, &bindings, thread.mcp.as_ref(), @@ -1276,7 +1128,13 @@ impl CodergenBackend for PebbleBackend { )?; let mcp = self.start_mcp(&bindings, cancel_token).await?; let agent = self - .build_agent(node, &route, &route_provider, &bindings, mcp.as_ref()) + .build_agent( + node, + &fallback_plan, + &route_provider, + &bindings, + mcp.as_ref(), + ) .await?; (agent, fallback_plan, mcp) }; @@ -1294,16 +1152,7 @@ impl CodergenBackend for PebbleBackend { ); let handle = Arc::new(PebbleControlHandle::new(agent.control_handle())); - let mut live = LiveAgent { - agent, - handle, - lease: None, - mcp, - total_usage: TokenCounts::default(), - total_cost: None, - inference_duration: Duration::ZERO, - tool_duration: Duration::ZERO, - }; + let mut live = LiveAgent::new(agent, handle, mcp); let route = fallback_plan.current().clone(); if let Err(error) = self.activate(&mut live, &route, &stage_id, request.thread_id, &bindings) @@ -1314,10 +1163,9 @@ impl CodergenBackend for PebbleBackend { let result = async { let mut response = self - .prompt_with_failover( + .prompt_live( &mut live, CodingInput::text(request.prompt), - node, &mut fallback_plan, &stage_id, request.thread_id, @@ -1330,11 +1178,7 @@ impl CodergenBackend for PebbleBackend { let mut repair_attempts = 0_i64; let mut previous_validation_error = None; loop { - let last_file_touched = file_tracking - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .last - .clone(); + let last_file_touched = live.last_file_touched.clone(); match validate_agent_output_sources( schema, &response, @@ -1358,10 +1202,9 @@ impl CodergenBackend for PebbleBackend { // identical failure mean it ignored the correction. previous_validation_error = Some(error); response = self - .prompt_with_failover( + .prompt_live( &mut live, CodingInput::text(repair_message), - node, &mut fallback_plan, &stage_id, request.thread_id, @@ -1377,7 +1220,6 @@ impl CodergenBackend for PebbleBackend { self.drain_late_steering( &mut live, - node, &mut fallback_plan, &stage_id, request.thread_id, @@ -1429,17 +1271,12 @@ impl CodergenBackend for PebbleBackend { }); } - let (files_touched, last_file_touched) = file_tracking - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .snapshot(); - Ok(CodergenResult::Text { - text: response, - usage: Some(stage_usage), - files_touched, - last_file_touched, - timing: StageTiming::active_only( + text: response, + usage: Some(stage_usage), + files_touched: live.files_touched.into_iter().collect(), + last_file_touched: live.last_file_touched, + timing: StageTiming::active_only( crate::millis_u64(live.inference_duration), crate::millis_u64(live.tool_duration), ), diff --git a/lib/foundation/fabro-types/src/run_event/agent.rs b/lib/foundation/fabro-types/src/run_event/agent.rs index f83168fcc..dc6099eef 100644 --- a/lib/foundation/fabro-types/src/run_event/agent.rs +++ b/lib/foundation/fabro-types/src/run_event/agent.rs @@ -75,6 +75,8 @@ pub fn coding_event_name(event: &CodingEvent) -> &'static str { CodingEvent::Error { .. } => "agent.error", CodingEvent::Warning { .. } => "agent.warning", CodingEvent::LoopDetected => "agent.loop.detected", + CodingEvent::ToolRoundsExhausted { .. } => "agent.tool.rounds.exhausted", + CodingEvent::RouteFailover { .. } => "agent.route.failover", CodingEvent::SteeringInjected { .. } => "agent.steering.injected", CodingEvent::RoundInterrupted { .. } => "agent.round.interrupted", CodingEvent::CompactionStarted { .. } => "agent.compaction.started", @@ -118,6 +120,8 @@ pub const CODING_EVENT_NAMES: &[&str] = &[ "agent.error", "agent.warning", "agent.loop.detected", + "agent.tool.rounds.exhausted", + "agent.route.failover", "agent.steering.injected", "agent.round.interrupted", "agent.compaction.started",