diff --git a/crates/libsy/src/algorithms/advisor_gate.rs b/crates/libsy/src/algorithms/advisor_gate.rs index f74543c19..87c15bd1f 100644 --- a/crates/libsy/src/algorithms/advisor_gate.rs +++ b/crates/libsy/src/algorithms/advisor_gate.rs @@ -10,8 +10,8 @@ //! buffered turn unchanged; `REDO` appends the discarded turn's text and the //! advisor's plan as feedback, then re-invokes the executor so it keeps //! working. Each budget scope (one benchmark evaluation, one session, or the -//! whole instance — see [`budget_scope`]) is reviewed at most `max_reviews` -//! times; afterwards every call is a pure passthrough. +//! whole instance — see [`budget::budget_scope`]) is reviewed at most +//! `max_reviews` times; afterwards every call is a pure passthrough. //! //! This design is a near-superset of solo executor behavior: identical until //! the executor first claims to be done, plus one quality gate that catches @@ -24,34 +24,44 @@ //! turn passes through as an implicit APPROVE — refund the consumed review, //! and count toward a per-scope failure cap that stops consulting a down //! advisor entirely. +//! +//! Structure: [`AdvisorGate`] is a thin orchestrator — the [`signals`] +//! processor folds each event's facts into per-turn state, the [`trigger`] +//! classifier reads them after the executor call, and the [`budget`] ledger +//! holds the only mutable state. -use std::collections::{HashMap, HashSet}; -use std::hash::{DefaultHasher, Hash, Hasher}; use std::sync::Arc; use std::time::Instant; -use parking_lot::Mutex; use switchyard_protocol::{ ContentBlock, InstructionBlock, LlmRequest, Message, ModelId, OutputParams, Request, Role, SamplingParams, }; -use crate::algorithms::util::tool_signals::ToolSignals; use crate::core::algorithm::{Algorithm, Driver, RoutingOutcome}; +use crate::core::processor::{Event, Processor}; use crate::{LibsyError, Result}; +mod budget; +mod signals; mod telemetry; #[cfg(test)] mod tests; mod transcript; +mod trigger; mod turn; +use budget::{ReviewBudget, ScopeKey, budget_scope, stall_key}; +use signals::{GateSignalProcessor, GateSignals}; use telemetry::{ ReviewAudit, emit_discarded_audit, emit_review_audit, record_consult_failure, record_discarded, record_review, }; use transcript::{VERDICT_PATTERN, Verdict, advisor_reply_text, parse_verdict, review_transcript}; -use turn::{GatedTurn, buffer_turn, has_tool_use, reasoning_text, visible_text}; +use trigger::TriggerClassifier; +#[cfg(test)] +use turn::has_tool_use; +use turn::{GatedTurn, buffer_turn, reasoning_text, visible_text}; /// APPROVE/REDO reviewer contract sent as the advisor's system prompt. pub const REVIEWER_SYSTEM_PROMPT: &str = @@ -72,14 +82,6 @@ const REASONING_TAIL_LABEL: &str = /// REDO echo when the discarded turn had neither text nor reasoning; strict /// endpoints (Anthropic) reject empty text blocks, so never echo "". const EMPTY_ECHO_PLACEHOLDER: &str = "(the executor produced no output this turn)"; -/// Failed consults tolerated per scope before the gate stops consulting. -/// Failures refund the review budget — a transient advisor error must not -/// silently exhaust `max_reviews` with zero real reviews — so this separate -/// cap is what bounds per-turn consult latency against a down advisor. -const MAX_FAILED_CONSULTS: u32 = 3; -/// Bounds tracked budget scopes and stall keys; a scope dropped at the bound -/// re-arms like a process restart (rare, harmless). -const MAX_TRACKED_SCOPES: usize = 1_024; /// Benchmark harnesses stamp every request of one evaluation — sub-agents /// included — with this header, so it is the review budget's first-choice /// scope: "reviews for *this* task" survives gateways shared by many tasks. @@ -147,38 +149,6 @@ impl Default for AdvisorGateConfig { } } -/// The trigger with its pattern compiled once at construction. -enum CompiledTrigger { - NoToolCall, - Pattern(regex::Regex), -} - -/// Review budget scope, in precedence order: the benchmark harness header -/// (exact evaluation identity, sub-agents included), then the host-resolved -/// session id, then one instance-wide scope for headerless clients. -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -enum ScopeKey { - Instance, - Client(String), - Session(String), -} - -/// Per-scope review ledger. -#[derive(Default)] -struct ScopeState { - reviews: u32, - failed_consults: u32, - exhaustion_logged: bool, -} - -/// Shared mutable gate state; every access is a short critical section and -/// the lock is never held across an await. -#[derive(Default)] -struct GateState { - scopes: HashMap, - stall_fired: HashSet, -} - /// Advisor review gate: executor turns pass through until the first terminal /// turn, which a stronger advisor reviews once per scope budget (APPROVE /// releases it, REDO feeds the plan back and re-invokes the executor). @@ -186,9 +156,13 @@ pub struct AdvisorGate { executor: ModelId, advisor: ModelId, config: AdvisorGateConfig, - trigger: CompiledTrigger, + /// Folds request- and response-side facts into the per-turn [`GateSignals`]. + signals: GateSignalProcessor, + /// Decides, from the signals, whether the buffered turn warrants review. + trigger: TriggerClassifier, + /// Reserve/refund review ledger and stall latch — the gate's only mutable state. + budget: ReviewBudget, verdict_re: regex::Regex, - state: Mutex, } impl AdvisorGate { @@ -203,112 +177,22 @@ impl AdvisorGate { if config.transcript_max_chars < 256 { return Err(algorithm_error("transcript_max_chars must be at least 256")); } - let trigger = match &config.gate_trigger { - GateTrigger::NoToolCall => CompiledTrigger::NoToolCall, - GateTrigger::Pattern(pattern) => { - if pattern.is_empty() { - return Err(algorithm_error( - "gate_trigger 'pattern' requires a non-empty gate_trigger_pattern", - )); - } - CompiledTrigger::Pattern(regex::Regex::new(pattern).map_err(|error| { - algorithm_error(format!( - "gate_trigger_pattern is not a valid regex: {error}" - )) - })?) - } - }; + let trigger = TriggerClassifier::new(&config)?; let verdict_re = regex::Regex::new(VERDICT_PATTERN).map_err(|error| { algorithm_error(format!("verdict pattern failed to compile: {error}")) })?; + let budget = ReviewBudget::new(config.max_reviews); Ok(Self { executor, advisor, config, + signals: GateSignalProcessor, trigger, + budget, verdict_re, - state: Mutex::new(GateState::default()), }) } - // ── Scope ledger ──────────────────────────────────────────────────────── - - /// Whether the scope's budget or failure cap is spent; logs once per scope. - fn check_exhausted(&self, scope: &ScopeKey) -> bool { - let mut state = self.state.lock(); - let Some(entry) = state.scopes.get_mut(scope) else { - return false; - }; - let exhausted = entry.reviews >= self.config.max_reviews - || entry.failed_consults >= MAX_FAILED_CONSULTS; - if exhausted && !entry.exhaustion_logged { - entry.exhaustion_logged = true; - tracing::info!( - target: "libsy", - scope = ?scope, - "advisor gate: review budget spent; passing through" - ); - } - exhausted - } - - /// Atomically re-checks exhaustion and reserves one review. Reserving - /// before the consult await means concurrent same-scope requests cannot - /// overdraw `max_reviews`; a loser returns its buffered turn unreviewed. - fn try_reserve(&self, scope: &ScopeKey) -> bool { - let mut state = self.state.lock(); - if state.scopes.len() >= MAX_TRACKED_SCOPES && !state.scopes.contains_key(scope) { - let evict = state - .scopes - .keys() - .find(|key| **key != ScopeKey::Instance) - .cloned(); - if let Some(key) = evict { - state.scopes.remove(&key); - } - } - let max_reviews = self.config.max_reviews; - let entry = state.scopes.entry(scope.clone()).or_default(); - if entry.reviews >= max_reviews || entry.failed_consults >= MAX_FAILED_CONSULTS { - return false; - } - entry.reviews += 1; - true - } - - /// Returns a reserved review after a failed consult and counts the - /// failure; applied on fail-open *and* fail-closed paths so the failure - /// cap bounds both. - fn refund_failure(&self, scope: &ScopeKey) { - let mut state = self.state.lock(); - let entry = state.scopes.entry(scope.clone()).or_default(); - entry.reviews = entry.reviews.saturating_sub(1); - entry.failed_consults += 1; - } - - /// Drops a completed session's ledger entry; the instance scope persists. - fn evict_scope(&self, scope: &ScopeKey) { - if *scope == ScopeKey::Instance { - return; - } - self.state.lock().scopes.remove(scope); - } - - fn stall_already_fired(&self, key: u64) -> bool { - self.state.lock().stall_fired.contains(&key) - } - - fn mark_stall_fired(&self, key: u64) { - let mut state = self.state.lock(); - if state.stall_fired.len() >= MAX_TRACKED_SCOPES { - let drop = state.stall_fired.iter().next().copied(); - if let Some(key) = drop { - state.stall_fired.remove(&key); - } - } - state.stall_fired.insert(key); - } - // ── Gate flow ─────────────────────────────────────────────────────────── async fn route_inner( @@ -321,7 +205,7 @@ impl AdvisorGate { // verbatim preserved-body replay, zero buffering. Executor errors // (including ContextWindowExceeded) propagate for the host's // client-visible mapping. - if self.check_exhausted(scope) { + if self.budget.check_exhausted(scope) { return Ok(RoutingOutcome::route_to( self.executor.clone(), Vec::new(), @@ -329,6 +213,19 @@ impl AdvisorGate { )); } + // Request-side signals fold in before the executor runs. + let mut request = request; + let mut signals = GateSignals::default(); + self.signals + .process( + &mut signals, + Event::Request { + request: &mut request, + driver: Some(driver), + }, + ) + .await?; + // Gated phase: generate the turn once, fully buffered, so the gate // can inspect it before the client sees anything. let response = driver @@ -336,39 +233,28 @@ impl AdvisorGate { .await?; let turn = buffer_turn(self.executor.as_str(), response).await?; - // Request-side guard counts come from the shared ToolSignals - // extraction — the same definition of tool-result/turn counting the - // stage router uses. - let signals = ToolSignals::from_request(&request, None); + // Response-side signals fold in after it: the terminal turn never + // appears on a later request, so the trigger runs on this event. + self.signals + .process(&mut signals, Event::ModelResponse(&turn.agg)) + .await?; + + let decision = self.trigger.classify(&signals); // The stall checkpoint fires once per conversation regardless of the - // turn's shape — even a tool-call turn — for executors that grind - // without ever declaring completion. - let stall_key = stall_key(&request); - let stall = self.config.gate_stall_turns > 0 - && !self.stall_already_fired(stall_key) - && signals.assistant_turn_count >= self.config.gate_stall_turns; - let triggered = match &self.trigger { - CompiledTrigger::Pattern(pattern) => { - pattern.is_match(visible_text(&turn.agg).as_deref().unwrap_or("")) - } - CompiledTrigger::NoToolCall => { - !has_tool_use(&turn.agg) - && signals.tool_result_count >= self.config.gate_min_tool_results - } - }; - if !(triggered || stall) { + // turn's shape. Only a stall with no simultaneous trigger latches + // (atomically — one winner per conversation), so a refunded review + // leaves the checkpoint re-armed. + let stall = decision.fired.is_none() + && decision.stalled + && self.budget.try_mark_stall_fired(stall_key(&request)); + if decision.fired.is_none() && !stall { return Ok(RoutingOutcome::answered( self.executor.clone(), request, turn.into_response(), )); } - // A stall consumed by a simultaneous trigger does not latch, so the - // checkpoint can still fire later if this review is refunded. - if stall && !triggered { - self.mark_stall_fired(stall_key); - } - if !self.try_reserve(scope) { + if !self.budget.try_reserve(scope) { return Ok(RoutingOutcome::answered( self.executor.clone(), request, @@ -376,11 +262,7 @@ impl AdvisorGate { )); } - let trigger_label = match (&self.trigger, triggered) { - (CompiledTrigger::Pattern(_), true) => "pattern", - (CompiledTrigger::NoToolCall, true) => "no_tool_call", - _ => "stall", - }; + let trigger_label = decision.fired.unwrap_or("stall"); let review_tail = visible_text(&turn.agg).or_else(|| { reasoning_text(&turn.agg).map(|reasoning| format!("{REASONING_TAIL_LABEL}{reasoning}")) }); @@ -395,7 +277,7 @@ impl AdvisorGate { )), Ok(ConsultOutcome::Redo { plan }) => Ok(self.redo(request, turn, &plan)), Ok(ConsultOutcome::Failed) => { - self.refund_failure(scope); + self.budget.refund_failure(scope); Ok(RoutingOutcome::answered( self.executor.clone(), request, @@ -403,7 +285,7 @@ impl AdvisorGate { )) } Err(error) => { - self.refund_failure(scope); + self.budget.refund_failure(scope); Err(error) } } @@ -593,7 +475,7 @@ impl Algorithm for AdvisorGate { == Some(true); let result = self.route_inner(&driver, request, &scope).await; if session_final { - self.evict_scope(&scope); + self.budget.evict_scope(&scope); } result } @@ -606,44 +488,6 @@ enum ConsultOutcome { Failed, } -// ── Budget scope ──────────────────────────────────────────────────────────── - -/// Resolves the review budget scope: the benchmark harness header wins (it is -/// stamped on every request of one evaluation, sub-agents included), then the -/// host-resolved session id, then one shared instance scope. -fn budget_scope(request: &Request) -> ScopeKey { - let metadata = request.metadata.as_ref(); - if let Some(value) = metadata - .and_then(|metadata| metadata.http_headers.as_ref()) - .and_then(|headers| headers.get(BENCH_SESSION_HEADER)) - .and_then(|value| value.to_str().ok()) - && !value.is_empty() - { - return ScopeKey::Client(value.to_string()); - } - if let Some(id) = metadata.and_then(|metadata| metadata.session_id.as_deref()) - && !id.is_empty() - { - return ScopeKey::Session(id.to_string()); - } - ScopeKey::Instance -} - -/// Latches the stall checkpoint per conversation: hash of the first user -/// message's text, which is constant across a session's turns. -fn stall_key(request: &Request) -> u64 { - let text = request - .llm_request - .messages - .iter() - .find(|message| message.role == Role::User) - .and_then(|message| message.text_content("\n")) - .unwrap_or_default(); - let mut hasher = DefaultHasher::new(); - text.hash(&mut hasher); - hasher.finish() -} - fn algorithm_error(message: impl Into) -> LibsyError { LibsyError::AlgorithmError { message: message.into(), diff --git a/crates/libsy/src/algorithms/advisor_gate/budget.rs b/crates/libsy/src/algorithms/advisor_gate/budget.rs new file mode 100644 index 000000000..6892f3548 --- /dev/null +++ b/crates/libsy/src/algorithms/advisor_gate/budget.rs @@ -0,0 +1,220 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The gate's mutable ledger: the per-scope review budget behind a +//! reserve/refund interface, plus the per-conversation stall latch — the +//! gate's only shared mutable state, behind one lock. + +use std::collections::{HashMap, HashSet}; +use std::hash::{DefaultHasher, Hash, Hasher}; + +use parking_lot::Mutex; +use switchyard_protocol::{Request, Role}; + +use super::BENCH_SESSION_HEADER; + +/// Failed consults tolerated per scope before the gate stops consulting; +/// failures refund the budget, so this cap is what bounds a down advisor. +const MAX_FAILED_CONSULTS: u32 = 3; +/// Bounds tracked budget scopes and stall keys; a scope dropped at the bound +/// re-arms like a process restart (rare, harmless). +const MAX_TRACKED_SCOPES: usize = 1_024; + +/// Review budget scope; resolution and precedence live in [`budget_scope`]. +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +pub(super) enum ScopeKey { + Instance, + Client(String), + Session(String), +} + +/// Per-scope review ledger. +#[derive(Default)] +struct ScopeState { + reviews: u32, + failed_consults: u32, + exhaustion_logged: bool, +} + +/// Shared mutable gate state; locked briefly, never across an await. +#[derive(Default)] +struct GateState { + scopes: HashMap, + stall_fired: HashSet, +} + +/// Per-scope review budget plus the stall checkpoint's per-conversation latch. +pub(super) struct ReviewBudget { + max_reviews: u32, + state: Mutex, +} + +impl ReviewBudget { + /// A fresh ledger allowing `max_reviews` reviews per scope. + pub(super) fn new(max_reviews: u32) -> Self { + Self { + max_reviews, + state: Mutex::new(GateState::default()), + } + } + + /// Whether the scope's budget or failure cap is spent; logs once per scope. + pub(super) fn check_exhausted(&self, scope: &ScopeKey) -> bool { + let mut state = self.state.lock(); + let Some(entry) = state.scopes.get_mut(scope) else { + return false; + }; + let exhausted = + entry.reviews >= self.max_reviews || entry.failed_consults >= MAX_FAILED_CONSULTS; + if exhausted && !entry.exhaustion_logged { + entry.exhaustion_logged = true; + tracing::info!( + target: "libsy", + scope = ?scope, + "advisor gate: review budget spent; passing through" + ); + } + exhausted + } + + /// Atomically re-checks exhaustion and reserves one review, so concurrent + /// same-scope requests cannot overdraw `max_reviews`. + pub(super) fn try_reserve(&self, scope: &ScopeKey) -> bool { + let mut state = self.state.lock(); + if state.scopes.len() >= MAX_TRACKED_SCOPES && !state.scopes.contains_key(scope) { + let evict = state + .scopes + .keys() + .find(|key| **key != ScopeKey::Instance) + .cloned(); + if let Some(key) = evict { + state.scopes.remove(&key); + } + } + let max_reviews = self.max_reviews; + let entry = state.scopes.entry(scope.clone()).or_default(); + if entry.reviews >= max_reviews || entry.failed_consults >= MAX_FAILED_CONSULTS { + return false; + } + entry.reviews += 1; + true + } + + /// Returns a reserved review after a failed consult and counts the + /// failure (fail-open and fail-closed paths alike). + pub(super) fn refund_failure(&self, scope: &ScopeKey) { + let mut state = self.state.lock(); + let entry = state.scopes.entry(scope.clone()).or_default(); + entry.reviews = entry.reviews.saturating_sub(1); + entry.failed_consults += 1; + } + + /// Drops a completed session's ledger entry; the instance scope persists. + pub(super) fn evict_scope(&self, scope: &ScopeKey) { + if *scope == ScopeKey::Instance { + return; + } + self.state.lock().scopes.remove(scope); + } + + /// Atomically latches the stall checkpoint for a conversation key; true + /// only for the call that set the latch. + pub(super) fn try_mark_stall_fired(&self, key: u64) -> bool { + let mut state = self.state.lock(); + if state.stall_fired.contains(&key) { + return false; + } + if state.stall_fired.len() >= MAX_TRACKED_SCOPES { + let drop = state.stall_fired.iter().next().copied(); + if let Some(key) = drop { + state.stall_fired.remove(&key); + } + } + state.stall_fired.insert(key) + } +} + +/// Resolves the review budget scope: the benchmark harness header, then the +/// host-resolved session id, then one shared instance scope. +pub(super) fn budget_scope(request: &Request) -> ScopeKey { + let metadata = request.metadata.as_ref(); + if let Some(value) = metadata + .and_then(|metadata| metadata.http_headers.as_ref()) + .and_then(|headers| headers.get(BENCH_SESSION_HEADER)) + .and_then(|value| value.to_str().ok()) + && !value.is_empty() + { + return ScopeKey::Client(value.to_string()); + } + if let Some(id) = metadata.and_then(|metadata| metadata.session_id.as_deref()) + && !id.is_empty() + { + return ScopeKey::Session(id.to_string()); + } + ScopeKey::Instance +} + +/// Latches the stall checkpoint per conversation: hash of the first user +/// message's text, which is constant across a session's turns. +pub(super) fn stall_key(request: &Request) -> u64 { + let text = request + .llm_request + .messages + .iter() + .find(|message| message.role == Role::User) + .and_then(|message| message.text_content("\n")) + .unwrap_or_default(); + let mut hasher = DefaultHasher::new(); + text.hash(&mut hasher); + hasher.finish() +} + +#[cfg(test)] +mod tests { + use super::*; + + fn scope() -> ScopeKey { + ScopeKey::Session("s1".to_string()) + } + + #[test] + fn reserving_spends_the_budget() { + let budget = ReviewBudget::new(1); + assert!(!budget.check_exhausted(&scope())); + assert!(budget.try_reserve(&scope())); + assert!(budget.check_exhausted(&scope())); + assert!(!budget.try_reserve(&scope())); + } + + #[test] + fn a_refund_reopens_the_budget_until_the_failure_cap() { + let budget = ReviewBudget::new(1); + for _ in 0..MAX_FAILED_CONSULTS { + assert!(budget.try_reserve(&scope())); + budget.refund_failure(&scope()); + } + // Every review was refunded, so it is the failure cap that is spent. + assert!(budget.check_exhausted(&scope())); + assert!(!budget.try_reserve(&scope())); + } + + #[test] + fn eviction_rearms_a_session_scope_but_never_the_instance() { + let budget = ReviewBudget::new(1); + assert!(budget.try_reserve(&scope())); + budget.evict_scope(&scope()); + assert!(budget.try_reserve(&scope())); + + assert!(budget.try_reserve(&ScopeKey::Instance)); + budget.evict_scope(&ScopeKey::Instance); + assert!(!budget.try_reserve(&ScopeKey::Instance)); + } + + #[test] + fn the_stall_latch_admits_one_caller_per_key() { + let budget = ReviewBudget::new(1); + assert!(budget.try_mark_stall_fired(7)); + assert!(!budget.try_mark_stall_fired(7)); + assert!(budget.try_mark_stall_fired(8)); + } +} diff --git a/crates/libsy/src/algorithms/advisor_gate/signals.rs b/crates/libsy/src/algorithms/advisor_gate/signals.rs new file mode 100644 index 000000000..3b44f19bc --- /dev/null +++ b/crates/libsy/src/algorithms/advisor_gate/signals.rs @@ -0,0 +1,164 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Gate signals, split by the event that carries them: conversation counts +//! fold in on [`Event::Request`], the generated turn's shape on +//! [`Event::ModelResponse`], and the trigger classifier reads both. + +use async_trait::async_trait; + +use crate::Result; +use crate::algorithms::util::tool_signals::ToolSignals; +use crate::core::processor::{Event, Processor}; + +use super::turn::{has_tool_use, visible_text}; + +/// Facts the trigger classifier reads, keyed by the event that produced them. +#[derive(Default)] +pub(super) struct GateSignals { + /// Conversation counts from the shared [`ToolSignals`] extraction. + pub(super) conversation: ToolSignals, + /// Shape of the turn the executor just generated. + pub(super) turn: TurnSignals, +} + +/// The generated turn's reviewable shape. +#[derive(Default)] +pub(super) struct TurnSignals { + /// The turn carries tool use (a `ToolUse` stop reason or any tool-call block). + pub(super) has_tool_use: bool, + /// The turn's visible text; `None` when it has no text blocks. + pub(super) visible_text: Option, +} + +/// Fills [`GateSignals`], each event writing its own side. +pub(super) struct GateSignalProcessor; + +#[async_trait] +impl Processor for GateSignalProcessor { + async fn process(&self, state: &mut GateSignals, event: Event<'_>) -> Result<()> { + match event { + Event::Request { request, .. } => { + state.conversation = ToolSignals::from_request(request, None); + } + Event::ModelResponse(agg) => { + state.turn = TurnSignals { + has_tool_use: has_tool_use(agg), + visible_text: visible_text(agg), + }; + } + Event::Decision { .. } => {} + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use switchyard_protocol::{ + AggLlmResponse, ContentBlock, LlmRequest, Message, ModelId, Request, ResponseOutput, Role, + ToolCall, ToolResult, + }; + + fn conversation_request() -> Request { + Request { + llm_request: LlmRequest { + messages: vec![ + Message::text(Role::User, "build X"), + Message::text(Role::Assistant, "working"), + Message { + role: Role::User, + content: vec![ContentBlock::ToolResult(ToolResult { + tool_call_id: "t1".to_string(), + content: vec![ContentBlock::Text { + text: "ok".to_string(), + }], + is_error: None, + })], + }, + ], + ..LlmRequest::default() + }, + raw_request: None, + metadata: None, + } + } + + fn tool_call_agg() -> AggLlmResponse { + AggLlmResponse { + outputs: vec![ResponseOutput { + role: Role::Assistant, + content: vec![ + ContentBlock::Text { + text: "running a tool".to_string(), + }, + ContentBlock::ToolCall(ToolCall { + id: "t1".to_string(), + name: "bash".to_string(), + arguments: serde_json::json!({}), + }), + ], + stop_reason: None, + }], + ..AggLlmResponse::default() + } + } + + #[tokio::test] + async fn each_event_fills_its_own_signal_side() -> Result<()> { + let processor = GateSignalProcessor; + let mut state = GateSignals::default(); + let mut request = conversation_request(); + + processor + .process( + &mut state, + Event::Request { + request: &mut request, + driver: None, + }, + ) + .await?; + assert_eq!(state.conversation.tool_result_count, 1); + assert_eq!(state.conversation.assistant_turn_count, 1); + assert!( + !state.turn.has_tool_use, + "the request never sets turn shape" + ); + + let agg = tool_call_agg(); + processor + .process(&mut state, Event::ModelResponse(&agg)) + .await?; + assert!(state.turn.has_tool_use); + assert_eq!(state.turn.visible_text.as_deref(), Some("running a tool")); + assert_eq!( + state.conversation.tool_result_count, 1, + "the response never rewrites conversation counts" + ); + Ok(()) + } + + #[tokio::test] + async fn the_decision_event_is_a_no_op() -> Result<()> { + let processor = GateSignalProcessor; + let mut state = GateSignals::default(); + let mut request = conversation_request(); + let selected = ModelId::from("executor"); + + processor + .process( + &mut state, + Event::Decision { + request: &mut request, + selected_model_id: &selected, + }, + ) + .await?; + + assert_eq!(state.conversation.tool_result_count, 0); + assert!(state.turn.visible_text.is_none()); + Ok(()) + } +} diff --git a/crates/libsy/src/algorithms/advisor_gate/trigger.rs b/crates/libsy/src/algorithms/advisor_gate/trigger.rs new file mode 100644 index 000000000..bb291a6d2 --- /dev/null +++ b/crates/libsy/src/algorithms/advisor_gate/trigger.rs @@ -0,0 +1,176 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The trigger classifier: decides, from the gate's signals, whether the +//! buffered turn warrants a review. Response-side on purpose — a request-side +//! trigger would never see the terminal turn the gate exists to catch. + +use crate::Result; + +use super::signals::GateSignals; +use super::{AdvisorGateConfig, GateTrigger, algorithm_error}; + +/// The trigger with its pattern compiled once at construction. +enum CompiledTrigger { + NoToolCall, + Pattern(regex::Regex), +} + +/// Classifies buffered turns against the configured trigger and the stall +/// checkpoint threshold. +pub(super) struct TriggerClassifier { + trigger: CompiledTrigger, + /// Tool results required before a `no_tool_call` terminal turn is reviewable. + min_tool_results: u32, + /// Assistant turns at which the stall checkpoint is reached; 0 disables. + stall_turns: u32, +} + +/// What the classifier concluded about one buffered turn. +pub(super) struct TriggerDecision { + /// The trigger that fired, as the telemetry label ("no_tool_call" | + /// "pattern"); `None` when the turn is not terminal. + pub(super) fired: Option<&'static str>, + /// The stall threshold is reached; the per-conversation latch is the ledger's. + pub(super) stalled: bool, +} + +impl TriggerClassifier { + /// Validates and compiles the configured trigger. + pub(super) fn new(config: &AdvisorGateConfig) -> Result { + let trigger = match &config.gate_trigger { + GateTrigger::NoToolCall => CompiledTrigger::NoToolCall, + GateTrigger::Pattern(pattern) => { + if pattern.is_empty() { + return Err(algorithm_error( + "gate_trigger 'pattern' requires a non-empty gate_trigger_pattern", + )); + } + CompiledTrigger::Pattern(regex::Regex::new(pattern).map_err(|error| { + algorithm_error(format!( + "gate_trigger_pattern is not a valid regex: {error}" + )) + })?) + } + }; + Ok(Self { + trigger, + min_tool_results: config.gate_min_tool_results, + stall_turns: config.gate_stall_turns, + }) + } + + /// Classifies the turn: does it warrant a review, and on which trigger? + pub(super) fn classify(&self, signals: &GateSignals) -> TriggerDecision { + let fired = match &self.trigger { + CompiledTrigger::Pattern(pattern) => pattern + .is_match(signals.turn.visible_text.as_deref().unwrap_or("")) + .then_some("pattern"), + CompiledTrigger::NoToolCall => (!signals.turn.has_tool_use + && signals.conversation.tool_result_count >= self.min_tool_results) + .then_some("no_tool_call"), + }; + let stalled = + self.stall_turns > 0 && signals.conversation.assistant_turn_count >= self.stall_turns; + TriggerDecision { fired, stalled } + } +} + +#[cfg(test)] +mod tests { + use super::super::signals::TurnSignals; + use super::*; + use crate::algorithms::util::tool_signals::ToolSignals; + + fn classifier(config: AdvisorGateConfig) -> TriggerClassifier { + TriggerClassifier::new(&config).expect("test config is valid") + } + + fn signals( + has_tool_use: bool, + visible_text: Option<&str>, + tool_results: u32, + assistant_turns: u32, + ) -> GateSignals { + GateSignals { + conversation: ToolSignals { + tool_result_count: tool_results, + assistant_turn_count: assistant_turns, + ..ToolSignals::default() + }, + turn: TurnSignals { + has_tool_use, + visible_text: visible_text.map(str::to_string), + }, + } + } + + #[test] + fn no_tool_call_fires_on_tool_less_turns_past_the_guard() { + let classifier = classifier(AdvisorGateConfig { + gate_min_tool_results: 2, + ..AdvisorGateConfig::default() + }); + // A tool-less turn under the guard stays quiet; past it, it fires. + assert!( + classifier + .classify(&signals(false, None, 1, 0)) + .fired + .is_none() + ); + assert_eq!( + classifier.classify(&signals(false, None, 2, 0)).fired, + Some("no_tool_call") + ); + // Tool use exempts the turn regardless of the guard. + assert!( + classifier + .classify(&signals(true, None, 5, 0)) + .fired + .is_none() + ); + } + + #[test] + fn pattern_reads_text_only_and_ignores_tool_use() { + let classifier = classifier(AdvisorGateConfig { + gate_trigger: GateTrigger::Pattern("task_complete".to_string()), + ..AdvisorGateConfig::default() + }); + assert_eq!( + classifier + .classify(&signals(true, Some("task_complete: true"), 0, 0)) + .fired, + Some("pattern") + ); + assert!( + classifier + .classify(&signals(false, Some("still working"), 0, 0)) + .fired + .is_none() + ); + // No visible text matches against the empty string, not a panic. + assert!( + classifier + .classify(&signals(false, None, 0, 0)) + .fired + .is_none() + ); + } + + #[test] + fn the_stall_threshold_is_reached_at_the_configured_turn_count() { + let classifier = classifier(AdvisorGateConfig { + gate_stall_turns: 3, + ..AdvisorGateConfig::default() + }); + assert!(!classifier.classify(&signals(true, None, 0, 2)).stalled); + assert!(classifier.classify(&signals(true, None, 0, 3)).stalled); + } + + #[test] + fn a_zero_stall_threshold_disables_the_checkpoint() { + let classifier = classifier(AdvisorGateConfig::default()); + assert!(!classifier.classify(&signals(true, None, 0, 100)).stalled); + } +}