diff --git a/crates/libsy/src/algorithms/vgr.rs b/crates/libsy/src/algorithms/vgr.rs index 4186e1ecf..65f39035d 100644 --- a/crates/libsy/src/algorithms/vgr.rs +++ b/crates/libsy/src/algorithms/vgr.rs @@ -11,19 +11,88 @@ #![allow(dead_code)] -use switchyard_protocol::Request; +use std::sync::Arc; + +use switchyard_protocol::{ModelId, Request}; use self::text::ToolRecord; +use crate::Result; +use crate::algorithms::fall_through::FallThrough; +use crate::algorithms::util::affinity::AffinityRouter; +use crate::core::algorithm::{Algorithm, Driver, RoutingOutcome}; +use crate::core::state::State; +mod config; mod decide; mod readout; mod render; mod rungs; +mod runtime; +mod safety; mod text; +pub use config::{ACTIVE_APPROVAL, ServingMode, Targets, VgrConfig}; +pub use safety::{BreakerConfig, KillSwitch}; + #[cfg(test)] mod tests; +/// A verification-gated route between a local and a capable tier. +pub struct Vgr { + route: FallThrough, + local: ModelId, + cloud: ModelId, +} + +impl Vgr { + /// Validates and constructs a verification-gated route. + pub fn new(config: VgrConfig) -> Result { + config.validate()?; + let local = config.targets.local.clone(); + let cloud = config.targets.cloud.clone(); + // Every local turn re-enters verification; an escalation holds until the + // next user turn. + let turn_affinity = Arc::new( + AffinityRouter::new() + .with_release_on_user_turn() + .with_latch_only([cloud.clone()]), + ); + let classifier = Arc::new(runtime::VgrClassifier { + breaker: safety::CircuitBreaker::new(config.breaker), + config, + }); + let route = FallThrough::new_with_state() + .with_name("vgr") + .with_processor(turn_affinity.clone()) + .with_classifier(turn_affinity) + .with_classifier(classifier); + Ok(Self { + route, + local, + cloud, + }) + } +} + +#[async_trait::async_trait] +impl Algorithm for Vgr { + fn name(&self) -> &str { + "vgr" + } + + async fn route(self: Arc, driver: Driver, request: Request) -> Result { + let mut outcome = self.route.execute(driver, request).await?; + let selected = outcome.selected_model_id()?.clone(); + if selected == self.cloud { + // Falling back to local would bypass the decision that selected cloud. + outcome.selected_model_ids.truncate(1); + } else if selected == self.local && outcome.response.is_none() { + outcome.selected_model_ids = vec![self.local.clone(), self.cloud.clone()]; + } + Ok(outcome) + } +} + /// The verification regime a request's capabilities license. #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] enum Branch { diff --git a/crates/libsy/src/algorithms/vgr/config.rs b/crates/libsy/src/algorithms/vgr/config.rs new file mode 100644 index 000000000..d171a3913 --- /dev/null +++ b/crates/libsy/src/algorithms/vgr/config.rs @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::time::Duration; + +use switchyard_protocol::ModelId; + +use super::safety::{BreakerConfig, KillSwitch}; +use crate::{LibsyError, Result}; + +/// Required attestation for live local commits. +pub const ACTIVE_APPROVAL: &str = "prospective-validation-and-canary-approved"; + +/// Authority granted to VGR decisions. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub enum ServingMode { + /// Always serves cloud without producing an attempt. + #[default] + Off, + /// Serves decisions, for isolated measurement. + Evaluate, + /// Decides but always serves cloud. + Shadow, + /// Serves decisions after operator approval. + Active { + /// Operator approval attestation. + approval: String, + }, +} + +/// Targets used by a VGR route. +#[derive(Clone, Debug)] +pub struct Targets { + /// Tier that produces the candidate attempt. + pub local: ModelId, + /// Tier used when the candidate is not licensed. + pub cloud: ModelId, + /// Local verifier, defaulting to `local`. + pub judge: Option, + /// Capable-tier verifier; unset removes the cloud confirmation rungs. + pub cloud_judge: Option, +} + +/// VGR runtime configuration. +#[derive(Clone, Debug)] +pub struct VgrConfig { + /// Completion and verifier targets. + pub targets: Targets, + /// Authority granted to routing decisions. + pub mode: ServingMode, + /// Budget for one turn's attempt and verification. + pub deadline: Duration, + /// Optional live stop controlled by the operator. + pub kill_switch: Option, + /// Local endpoint breaker tuning. + pub breaker: BreakerConfig, + /// Whether a cheap typing call selects a verification regime. + pub task_typing: bool, + /// Whether the local tier accepts image content. + pub local_supports_images: bool, + /// Lets a long agentic run with earlier tool errors commit on a capable-tier + /// confirmation once it ends in this many consecutive clean tool results. + pub confirmed_recovery_min_clean_tail: Option, +} + +impl VgrConfig { + /// Creates an off-by-default route between local and capable tiers. + pub fn new(local: ModelId, cloud: ModelId) -> Self { + Self { + targets: Targets { + local, + cloud, + judge: None, + cloud_judge: None, + }, + mode: ServingMode::Off, + deadline: Duration::from_secs(30), + kill_switch: None, + breaker: BreakerConfig::default(), + task_typing: true, + local_supports_images: false, + confirmed_recovery_min_clean_tail: None, + } + } + + pub(super) fn validate(&self) -> Result<()> { + if self.deadline.is_zero() || self.breaker.threshold == 0 { + return Err(LibsyError::AlgorithmError { + message: "vgr deadline and breaker threshold must be non-zero".into(), + }); + } + if let ServingMode::Active { approval } = &self.mode + && approval != ACTIVE_APPROVAL + { + return Err(LibsyError::AlgorithmError { + message: format!( + "vgr active mode requires approval attestation {ACTIVE_APPROVAL:?}" + ), + }); + } + Ok(()) + } + + pub(super) fn judge(&self) -> &ModelId { + self.targets.judge.as_ref().unwrap_or(&self.targets.local) + } + + pub(super) fn confirmed_min_clean_tail(&self) -> Option { + self.confirmed_recovery_min_clean_tail + .map(|tail| i32::try_from(tail.max(1)).unwrap_or(i32::MAX)) + } +} diff --git a/crates/libsy/src/algorithms/vgr/runtime.rs b/crates/libsy/src/algorithms/vgr/runtime.rs new file mode 100644 index 000000000..a9ace1439 --- /dev/null +++ b/crates/libsy/src/algorithms/vgr/runtime.rs @@ -0,0 +1,420 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The classifier that runs one verification-gated turn. +//! +//! A proposed tool call is judged against the session's trajectory; a final +//! answer is typed, judged by rungs that run cheapest-first and stop once the +//! decision is settled, and then committed or escalated. Every call is bounded +//! by what is left of the turn's deadline, and a call that fails or times out +//! is evidence never gathered, which does not commit. + +use std::time::{Duration, Instant}; + +use async_trait::async_trait; +use switchyard_protocol::{ + AggLlmResponse, ContentBlock, LlmClientError, ModelId, Request, Response, StopReason, +}; + +use super::config::{ServingMode, VgrConfig}; +use super::decide::{self, AgenticRun, Route, Signals, Tri}; +use super::rungs::{self, Question}; +use super::safety::{CircuitBreaker, endpoint_failure, fallback_eligible}; +use super::{Branch, Capabilities, TaskType, derive_capabilities, readout, text}; +use crate::algorithms::util::buffered_response::buffer_response; +use crate::algorithms::util::decisive; +use crate::core::algorithm::Driver; +use crate::core::classifier::{Classification, Classifier}; +use crate::core::state::{State, StateValue}; +use crate::{LibsyError, Result}; + +const TURN_STREAK_KEY: &str = "vgr.turn_verification.streak"; +const TURN_LATCHED_KEY: &str = "vgr.turn_verification.latched"; +/// Consecutive escalation votes that latch a session to the capable tier. +const TURN_CONFIRMATIONS: u32 = 2; +const TURN_ESCALATE_AT: f64 = 0.5; + +type Scored = (Classification, Option); + +pub(super) struct VgrClassifier { + pub(super) config: VgrConfig, + pub(super) breaker: CircuitBreaker, +} + +#[async_trait] +impl Classifier for VgrClassifier { + async fn score( + &self, + state: &mut State, + request: &mut Request, + driver: &Driver, + ) -> Result { + let started = Instant::now(); + let local = self.config.targets.local.clone(); + let short_circuit = if self.config.mode == ServingMode::Off { + Some("mode_off") + } else if self + .config + .kill_switch + .as_ref() + .is_some_and(|switch| switch.is_engaged()) + { + Some("kill_switch") + } else if self.breaker.is_open() { + Some("breaker_open") + } else if turn_latched(state) { + Some("turn_verification_latched") + } else if !self.config.local_supports_images && request_has_image(request) { + Some("local_image_unsupported") + } else { + None + }; + if let Some(reason) = short_circuit { + return Ok(self.escalate(reason, None)); + } + + let Some(budget) = self.remaining(started) else { + return Ok(self.escalate("local_timed_out", None)); + }; + let attempted = tokio::time::timeout(budget, async { + let response = driver + .call_model(request.clone(), vec![local.clone()]) + .await?; + buffer_response(local.as_str(), response).await + }) + .await; + let buffered = match attempted { + Ok(Ok(buffered)) => { + self.breaker.success(); + buffered + } + Ok(Err(error)) => { + if endpoint_failure(&error) { + self.breaker.failure(); + } + if !fallback_eligible(&error) { + return Err(error); + } + return Ok(self.escalate("local_unavailable", None)); + } + Err(_) => { + self.breaker.failure(); + return Ok(self.escalate("local_timed_out", None)); + } + }; + + if rungs::has_tool_call(&buffered.agg) { + if !tool_use_is_complete(&buffered.agg) { + return Ok(self.escalate("malformed_tool_call", None)); + } + if self + .verify_turn(driver, request, &buffered.agg, state, started) + .await + { + return Ok(self.escalate("turn_verification_escalated", None)); + } + if self.config.mode == ServingMode::Shadow { + return Ok(self.escalate("turn_verification_complete", None)); + } + log("turn_verification_complete", None, Route::Local); + return Ok((decisive(&local), Some(buffered.into_response()))); + } + + let attempt = rungs::response_text(&buffered.agg).unwrap_or_default(); + let task_type = self.type_task(driver, request, started).await; + let caps = derive_capabilities(request, &attempt, task_type); + let signals = self.gather(driver, &caps, request, started).await; + let route = decide::decide(&caps, &signals, self.config.confirmed_min_clean_tail()); + if route == Route::Cloud || self.config.mode == ServingMode::Shadow { + return Ok(self.escalate("decided", Some(caps.branch))); + } + log("decided", Some(caps.branch), Route::Local); + Ok((decisive(&local), Some(buffered.into_response()))) + } +} + +impl VgrClassifier { + fn escalate(&self, reason: &'static str, branch: Option) -> Scored { + log(reason, branch, Route::Cloud); + (decisive(&self.config.targets.cloud), None) + } + + /// Judges one proposed tool call; `true` escalates the rest of the session. + async fn verify_turn( + &self, + driver: &Driver, + request: &Request, + proposed: &AggLlmResponse, + state: &mut State, + started: Instant, + ) -> bool { + let mut call = rungs::build_request( + Question::TurnTrajectory, + &rungs::turn_view(request, proposed), + readout::MAX_OUTPUT_TOKENS, + request.metadata.clone(), + ); + readout::request_logprobs(&mut call); + let judge = self.config.judge(); + let probability = match self.call(driver, call, judge, started).await { + None => return false, + Some(Ok(agg)) => readout::p_yes(&agg), + // Verification fails open, unless the local endpoint itself is down. + Some(Err(error)) => { + if *judge == self.config.targets.local && endpoint_failure(&error) { + self.breaker.failure(); + latch_turns(state); + return true; + } + return false; + } + }; + let Some(probability) = probability else { + return false; + }; + if probability < TURN_ESCALATE_AT { + state.extra.remove(TURN_STREAK_KEY); + return false; + } + let streak = match state.extra.get(TURN_STREAK_KEY) { + Some(StateValue::Count(streak)) => streak.saturating_add(1), + _ => 1, + }; + state + .extra + .insert(TURN_STREAK_KEY.into(), StateValue::Count(streak)); + if streak >= TURN_CONFIRMATIONS { + latch_turns(state); + } + streak >= TURN_CONFIRMATIONS + } + + /// Types the request; abstains, selecting the default regime, on any failure. + async fn type_task( + &self, + driver: &Driver, + request: &Request, + started: Instant, + ) -> Option { + if !self.config.task_typing { + return None; + } + let task = text::user_task_text(&text::turns(request).0); + if task.trim().is_empty() { + return None; + } + let call = rungs::build_typing_request(&task, request.metadata.clone()); + let agg = self + .call(driver, call, self.config.judge(), started) + .await? + .ok()?; + rungs::parse_task_type(&agg) + } + + /// Gathers evidence cheapest-first, stopping once the decision commits. + async fn gather( + &self, + driver: &Driver, + caps: &Capabilities, + request: &Request, + started: Instant, + ) -> Signals { + let mut signals = Signals::default(); + let branch = caps.branch; + let confirmed = self.config.confirmed_min_clean_tail(); + let run = (branch == Branch::Agentic).then(|| decide::agentic_run(caps, confirmed)); + let (Some(judged), false) = ( + caps.transcript.as_deref(), + matches!(branch, Branch::Coding | Branch::Unknown) || run == Some(AgenticRun::Vetoed), + ) else { + return signals; + }; + let settled = |signals: &Signals| decide::decide(caps, signals, confirmed) == Route::Local; + let cloud_confirms = run == Some(AgenticRun::ConfirmedRecovery); + let local = self.config.judge(); + let ask = |target, question, max_output_tokens| { + self.ask( + driver, + target, + question, + judged, + max_output_tokens, + request, + started, + ) + }; + + signals.readout = self.readout(driver, judged, request, started).await; + if settled(&signals) { + return signals; + } + if !(cloud_confirms && decide::locally_verified(branch, &signals)) { + signals.deliberation = match ask( + local, + Question::Deliberation, + rungs::LOCAL_DELIBERATION_MAX_OUTPUT_TOKENS, + ) + .await + { + Tri::Yes => Some(1.0), + Tri::No => Some(0.0), + Tri::Unknown => None, + }; + if settled(&signals) { + return signals; + } + } + let Some(cloud) = self.config.targets.cloud_judge.as_ref() else { + return signals; + }; + let cloud_ask = |question| ask(cloud, question, rungs::DELIBERATION_MAX_OUTPUT_TOKENS); + if cloud_confirms { + if decide::locally_verified(branch, &signals) { + signals.cloud_judge = Some(cloud_ask(Question::Evidence).await); + } + return signals; + } + match branch { + Branch::Answer => { + signals.answer_verifier = Some(cloud_ask(Question::Answer).await); + if !settled(&signals) { + signals.evidence_verifier = Some(cloud_ask(Question::Evidence).await); + } + } + Branch::Chat => { + signals.cloud_judge = Some(cloud_ask(Question::Evidence).await); + // Asked only once the first affirms, so a refutation costs one call. + if signals.cloud_judge == Some(Tri::Yes) { + signals.evidence_confirm = Some(cloud_ask(Question::Answer).await); + } + } + _ => {} + } + signals + } + + async fn readout( + &self, + driver: &Driver, + judged: &str, + request: &Request, + started: Instant, + ) -> Option { + let mut call = rungs::build_request( + Question::Evidence, + judged, + readout::MAX_OUTPUT_TOKENS, + request.metadata.clone(), + ); + readout::request_logprobs(&mut call); + let agg = self + .call(driver, call, self.config.judge(), started) + .await? + .ok()?; + readout::p_yes(&agg) + } + + #[allow(clippy::too_many_arguments)] + async fn ask( + &self, + driver: &Driver, + target: &ModelId, + question: Question, + judged: &str, + max_output_tokens: u64, + request: &Request, + started: Instant, + ) -> Tri { + let call = rungs::build_request( + question, + judged, + max_output_tokens, + request.metadata.clone(), + ); + match self.call(driver, call, target, started).await { + Some(Ok(agg)) => rungs::parse_verdict(&agg), + _ => Tri::Unknown, + } + } + + /// One verifier call inside the remaining budget; `None` once it is spent. + async fn call( + &self, + driver: &Driver, + call: Request, + target: &ModelId, + started: Instant, + ) -> Option> { + let budget = self.remaining(started)?; + let outcome = tokio::time::timeout(budget, async { + let response = driver.call_model(call, vec![target.clone()]).await?; + response + .llm_response + .into_agg() + .await + .map_err(|source| LibsyError::client_call(target.clone(), source)) + }) + .await; + Some(outcome.unwrap_or_else(|_| { + Err(LibsyError::client_call( + target.clone(), + LlmClientError::Timeout { + source: Box::new(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "VGR verifier exceeded the decision deadline", + )), + }, + )) + })) + } + + fn remaining(&self, started: Instant) -> Option { + self.config.deadline.checked_sub(started.elapsed()) + } +} + +fn log(reason: &'static str, branch: Option, served: Route) { + tracing::info!(target: "libsy", reason, branch = ?branch, served = ?served, "vgr decision"); +} + +fn turn_latched(state: &State) -> bool { + matches!( + state.extra.get(TURN_LATCHED_KEY), + Some(StateValue::Count(value)) if *value > 0 + ) +} + +fn latch_turns(state: &mut State) { + state + .extra + .insert(TURN_LATCHED_KEY.into(), StateValue::Count(1)); +} + +fn tool_use_is_complete(response: &AggLlmResponse) -> bool { + response.outputs.iter().all(|output| { + output.stop_reason != Some(StopReason::MaxTokens) + && output.content.iter().all(|block| { + !matches!(block, ContentBlock::ToolCall(call) if call.arguments.is_string()) + }) + }) +} + +/// Whether a request carries image content directly or inside a tool result. +fn request_has_image(request: &Request) -> bool { + fn has_image(block: &ContentBlock) -> bool { + match block { + ContentBlock::Image { .. } => true, + ContentBlock::ToolResult(result) => result.content.iter().any(has_image), + _ => false, + } + } + let llm = &request.llm_request; + llm.instructions + .iter() + .flat_map(|instruction| &instruction.content) + .chain(llm.messages.iter().flat_map(|message| &message.content)) + .any(has_image) +} + +#[cfg(test)] +mod tests; diff --git a/crates/libsy/src/algorithms/vgr/runtime/tests.rs b/crates/libsy/src/algorithms/vgr/runtime/tests.rs new file mode 100644 index 000000000..7944d1d5d --- /dev/null +++ b/crates/libsy/src/algorithms/vgr/runtime/tests.rs @@ -0,0 +1,252 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; + +use serde_json::json; +use switchyard_protocol::{ + AggLlmResponse, Category, ContentBlock, FormatId, LlmClientError, LlmResponse, Message, + Metadata, ModelId, Request, Response, ResponseOutput, Role, StopReason, ToolCall, WireFormat, +}; + +use super::super::config::{ACTIVE_APPROVAL, ServingMode, VgrConfig}; +use super::super::tests::{call, request, result}; +use crate::Result; +use crate::algorithms::vgr::Vgr; +use crate::core::algorithm::Algorithm; +use crate::core::testing::{reply, test_drive_with_models}; + +fn route(configure: impl FnOnce(&mut VgrConfig)) -> Result> { + let mut config = VgrConfig::new(ModelId::new("local"), ModelId::new("cloud")); + config.targets.judge = Some(ModelId::new("judge")); + config.targets.cloud_judge = Some(ModelId::new("cloud-judge")); + config.mode = ServingMode::Active { + approval: ACTIVE_APPROVAL.into(), + }; + config.task_typing = false; + configure(&mut config); + Ok(Arc::new(Vgr::new(config)?)) +} + +fn models() -> HashMap> { + HashMap::from([( + Category::Any, + vec![ModelId::new("local"), ModelId::new("cloud")], + )]) +} + +fn session(messages: Vec) -> Request { + let mut request = request(messages); + request.metadata = Some(Metadata { + session_id: Some("session-1".into()), + ..Metadata::default() + }); + request +} + +fn agg(response: AggLlmResponse) -> Response { + Response { + llm_response: LlmResponse::Agg(response), + metadata: None, + upstream_headers: http::HeaderMap::new(), + } +} + +fn tool_call(arguments: serde_json::Value, stop_reason: StopReason) -> Response { + agg(AggLlmResponse { + outputs: vec![ResponseOutput { + role: Role::Assistant, + content: vec![ContentBlock::ToolCall(ToolCall { + id: "next".into(), + name: "bash".into(), + arguments, + })], + url_citations: Vec::new(), + stop_reason: Some(stop_reason), + }], + ..Default::default() + }) +} + +/// A verifier reply whose first token puts probability `p` on "yes". +fn scored(p: f64) -> Response { + let mut response = AggLlmResponse::default(); + response.preservation.responses.insert( + FormatId::from(WireFormat::OpenAiChat), + json!({"choices": [{"logprobs": {"content": [{"top_logprobs": [ + {"token": "yes", "logprob": p.ln()}, + {"token": "no", "logprob": (1.0 - p).ln()} + ]}]}}]}), + ); + agg(response) +} + +#[derive(Default)] +struct Calls(AtomicUsize, AtomicUsize); + +async fn drive( + route: &Arc, + request: Request, + calls: Arc, + local: fn() -> Response, + judge_p: f64, +) -> Result { + let (selected, _) = test_drive_with_models( + Arc::clone(route), + request, + models(), + move |target: ModelId, _| { + let calls = Arc::clone(&calls); + async move { + match target.as_str() { + "local" => calls.0.fetch_add(1, Ordering::Relaxed), + "cloud-judge" => calls.1.fetch_add(1, Ordering::Relaxed), + _ => 0, + }; + Ok(match target.as_str() { + "local" => local(), + "judge" => scored(judge_p), + "cloud-judge" => reply("yes"), + _ => reply("cloud answer"), + }) + } + }, + ) + .await?; + Ok(selected.to_string()) +} + +fn complete_call() -> Response { + tool_call(json!({"command": "ls"}), StopReason::ToolUse) +} + +#[tokio::test] +async fn malformed_tool_calls_escalate_and_complete_ones_are_judged() -> Result<()> { + let route = route(|_| {})?; + let calls = Arc::new(Calls::default()); + let truncated = || tool_call(json!(r#"{"command":"l"#), StopReason::MaxTokens); + let task = || session(vec![Message::text(Role::User, "list files")]); + assert_eq!( + drive(&route, task(), calls.clone(), truncated, 0.0).await?, + "cloud" + ); + assert_eq!( + drive(&route, task(), calls, complete_call, 0.1).await?, + "local" + ); + Ok(()) +} + +#[tokio::test] +async fn a_truncated_local_attempt_is_judged_before_it_escalates() -> Result<()> { + let route = route(|config| config.targets.judge = None)?; + let calls = Arc::new(Calls::default()); + let task = session(vec![Message::text(Role::User, "explain quicksort")]); + let cut_off = || { + let mut response = reply("Quicksort picks a pivot, then recurses because"); + if let LlmResponse::Agg(agg) = &mut response.llm_response { + agg.outputs[0].stop_reason = Some(StopReason::MaxTokens); + } + response + }; + let selected = drive(&route, task, calls.clone(), cut_off, 0.0).await?; + assert_eq!(selected, "cloud"); + // Attempt, readout and deliberation; read as absent, only the attempt runs. + assert_eq!(calls.0.load(Ordering::Relaxed), 3); + Ok(()) +} + +#[tokio::test] +async fn two_consecutive_escalation_votes_latch_the_session() -> Result<()> { + let route = route(|_| {})?; + let calls = Arc::new(Calls::default()); + let task = || session(vec![Message::text(Role::User, "fix the build")]); + assert_eq!( + drive(&route, task(), calls.clone(), complete_call, 0.9).await?, + "local" + ); + assert_eq!( + drive(&route, task(), calls.clone(), complete_call, 0.9).await?, + "cloud" + ); + assert_eq!(calls.0.load(Ordering::Relaxed), 2); + + // Latched past the user turn: no further local attempt is produced. + assert_eq!( + drive(&route, task(), calls.clone(), complete_call, 0.0).await?, + "cloud" + ); + assert_eq!(calls.0.load(Ordering::Relaxed), 2); + Ok(()) +} + +#[tokio::test] +async fn final_answers_commit_only_on_a_verified_clean_record() -> Result<()> { + let answer = || reply("the build passes"); + let run = |tool_output: &str| { + session(vec![ + Message::text(Role::User, "fix the build"), + call("c1"), + result("c1", tool_output), + ]) + }; + let calls = Arc::new(Calls::default()); + let clean = route(|_| {})?; + assert_eq!( + drive(&clean, run("ok"), calls.clone(), answer, 0.3).await?, + "local" + ); + assert_eq!( + drive(&clean, run("ok"), calls.clone(), answer, 0.1).await?, + "cloud" + ); + + // A failure longer ago than the clean tail is forgiven only on the capable judge. + let mut long = vec![Message::text(Role::User, "fix the build")]; + long.extend([call("e"), result("e", r#"{"error": "missing file"}"#)]); + for index in 0..101 { + let id = format!("c{index}"); + long.extend([call(&id), result(&id, "ok")]); + } + assert_eq!( + drive(&clean, session(long.clone()), calls.clone(), answer, 1.0).await?, + "cloud" + ); + assert_eq!(calls.1.load(Ordering::Relaxed), 0); + let confirmed = route(|config| config.confirmed_recovery_min_clean_tail = Some(1))?; + assert_eq!( + drive(&confirmed, session(long), calls.clone(), answer, 1.0).await?, + "local" + ); + assert_eq!(calls.1.load(Ordering::Relaxed), 1); + Ok(()) +} + +#[tokio::test] +async fn an_unavailable_local_tier_escalates_and_other_failures_surface() -> Result<()> { + let route = route(|_| {})?; + for (status, escalates) in [(503, true), (400, false)] { + let outcome = test_drive_with_models( + Arc::clone(&route), + session(vec![Message::text(Role::User, "task")]), + models(), + move |target: ModelId, _| async move { + match target.as_str() { + "local" => Err(LlmClientError::UpstreamHttp { + status: http::StatusCode::from_u16(status).unwrap_or_default(), + body: String::new(), + }), + _ => Ok(reply("cloud answer")), + } + }, + ) + .await; + assert_eq!( + outcome.is_ok_and(|(selected, _)| selected == "cloud"), + escalates + ); + } + Ok(()) +} diff --git a/crates/libsy/src/algorithms/vgr/safety.rs b/crates/libsy/src/algorithms/vgr/safety.rs new file mode 100644 index 000000000..11acbdcb5 --- /dev/null +++ b/crates/libsy/src/algorithms/vgr/safety.rs @@ -0,0 +1,140 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use http::StatusCode; +use parking_lot::Mutex; +use switchyard_protocol::LlmClientError; + +use crate::LibsyError; + +/// Operator-controlled stop for local attempts. +#[derive(Clone, Debug, Default)] +pub struct KillSwitch(Arc); + +impl KillSwitch { + /// Creates a released switch. + pub fn new() -> Self { + Self::default() + } + + /// Stops local attempts for subsequent requests. + pub fn engage(&self) { + self.0.store(true, Ordering::Relaxed); + } + + /// Allows local attempts again. + pub fn release(&self) { + self.0.store(false, Ordering::Relaxed); + } + + /// Reports whether local attempts are stopped. + pub fn is_engaged(&self) -> bool { + self.0.load(Ordering::Relaxed) + } +} + +/// Circuit-breaker configuration. +#[derive(Clone, Copy, Debug)] +pub struct BreakerConfig { + /// Consecutive endpoint failures required to open. + pub threshold: u32, + /// Delay before one half-open trial is admitted. + pub cooldown: Duration, +} + +impl Default for BreakerConfig { + fn default() -> Self { + Self { + threshold: 5, + cooldown: Duration::from_secs(30), + } + } +} + +#[derive(Debug, Default)] +struct BreakerState { + failures: u32, + opened_at: Option, + trial_in_flight: bool, +} + +#[derive(Debug)] +pub(super) struct CircuitBreaker { + config: BreakerConfig, + state: Mutex, +} + +impl CircuitBreaker { + pub(super) fn new(config: BreakerConfig) -> Self { + Self { + config, + state: Mutex::new(BreakerState::default()), + } + } + + pub(super) fn is_open(&self) -> bool { + let mut state = self.state.lock(); + let Some(opened_at) = state.opened_at else { + return false; + }; + if state.trial_in_flight || opened_at.elapsed() < self.config.cooldown { + return true; + } + state.trial_in_flight = true; + false + } + + pub(super) fn success(&self) { + *self.state.lock() = BreakerState::default(); + } + + pub(super) fn failure(&self) { + let mut state = self.state.lock(); + state.failures = state.failures.saturating_add(1); + state.trial_in_flight = false; + if state.failures >= self.config.threshold { + state.opened_at = Some(Instant::now()); + } + } +} + +/// Errors for which retrying on the capable tier is safe. +pub(super) fn fallback_eligible(error: &LibsyError) -> bool { + matches!( + error, + LibsyError::ClientCall { source, .. } if match source { + LlmClientError::ContextWindowExceeded { .. } + | LlmClientError::Transport { .. } + | LlmClientError::Timeout { .. } => true, + LlmClientError::UpstreamHttp { status, .. } => + matches!( + *status, + StatusCode::FORBIDDEN + | StatusCode::REQUEST_TIMEOUT + | StatusCode::TOO_MANY_REQUESTS + ) || status.is_server_error(), + _ => false, + } + ) +} + +/// Only endpoint unavailability contributes to breaker health. +pub(super) fn endpoint_failure(error: &LibsyError) -> bool { + matches!( + error, + LibsyError::ClientCall { source, .. } if match source { + LlmClientError::Transport { .. } | LlmClientError::Timeout { .. } => true, + LlmClientError::UpstreamHttp { status, .. } => matches!( + *status, + StatusCode::BAD_GATEWAY + | StatusCode::SERVICE_UNAVAILABLE + | StatusCode::GATEWAY_TIMEOUT + ), + _ => false, + } + ) +} diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 9d07aeff9..65544f747 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -37,6 +37,9 @@ pub use algorithms::util::escalation::EscalationJudgeConfig; pub use algorithms::util::prompts::append_note; pub use algorithms::util::subagent::{SubagentGate, SubagentOverride}; pub use algorithms::util::tool_signals::{DEFAULT_RECENT_WINDOW, ToolSemantics, ToolSignals}; +pub use algorithms::vgr::{ + ACTIVE_APPROVAL, BreakerConfig, KillSwitch, ServingMode, Targets as VgrTargets, Vgr, VgrConfig, +}; // Stage-router scoring and tier selection — the shared signal-driven routing // core (scorer, picker, and the `StageClassifier`).