switchyard_libsy/algorithms/advisor_gate.rs
1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Executor gated by a once-per-session advisor review.
5//!
6//! The executor answers every client-visible turn. Turns with tool calls pass
7//! through unreviewed; the first *terminal* turn — no tool calls (or a text
8//! match under the `pattern` trigger) — is buffered and shown to a stronger
9//! advisor model together with the full transcript. `APPROVE` releases the
10//! buffered turn unchanged; `REDO` appends the discarded turn's text and the
11//! advisor's plan as feedback, then re-invokes the executor so it keeps
12//! working. Each budget scope (one benchmark evaluation, one session, or the
13//! whole instance — see [`budget::budget_scope`]) is reviewed at most
14//! `max_reviews` times; afterwards every call is a pure passthrough.
15//!
16//! This design is a near-superset of solo executor behavior: identical until
17//! the executor first claims to be done, plus one quality gate that catches
18//! premature convergence. Front-loading advice was measured to suppress the
19//! executor's own test-and-iterate loop, so no advice is injected up front.
20//!
21//! Failure posture: executor errors always propagate (including
22//! `ContextWindowExceeded`, which hosts map to a client-visible 400 so agent
23//! harnesses can compact). Advisor errors honor `fail_open` — the buffered
24//! turn passes through as an implicit APPROVE — refund the consumed review,
25//! and count toward a per-scope failure cap that stops consulting a down
26//! advisor entirely.
27//!
28//! Structure: [`AdvisorGate`] is a thin orchestrator — the [`signals`]
29//! processor folds each event's facts into per-turn state, the [`trigger`]
30//! classifier reads them after the executor call, and the [`budget`] ledger
31//! holds the only mutable state.
32
33use std::sync::Arc;
34use std::time::Instant;
35
36use switchyard_protocol::{
37 Category, ContentBlock, InstructionBlock, LlmRequest, Message, ModelId, OutputParams, Request,
38 Role, SamplingParams,
39};
40
41use crate::core::algorithm::{Algorithm, Driver, RoutingOutcome};
42use crate::core::processor::{Event, Processor};
43use crate::{LibsyError, Result};
44
45mod budget;
46mod signals;
47mod telemetry;
48#[cfg(test)]
49mod tests;
50mod transcript;
51mod trigger;
52mod turn;
53
54use super::util::buffered_response::{BufferedResponse, buffer_response};
55use budget::{ReviewBudget, ScopeKey, budget_scope, stall_key};
56use signals::{GateSignalProcessor, GateSignals};
57use telemetry::{
58 ReviewAudit, emit_discarded_audit, emit_review_audit, record_consult_failure, record_discarded,
59 record_review,
60};
61use transcript::{VERDICT_PATTERN, Verdict, advisor_reply_text, parse_verdict, review_transcript};
62use trigger::TriggerClassifier;
63#[cfg(test)]
64use turn::has_tool_use;
65use turn::{reasoning_text, visible_text};
66
67/// APPROVE/REDO reviewer contract sent as the advisor's system prompt.
68pub const REVIEWER_SYSTEM_PROMPT: &str =
69 include_str!("../prompts/advisor-gate/reviewer-system-prompt.md");
70
71/// Prepended to the advisor's REDO plan when it is fed back as a user turn,
72/// instructing the executor to continue rather than stop.
73pub const REDO_FEEDBACK_PREFIX: &str = concat!(
74 include_str!("../prompts/advisor-gate/redo-feedback-prefix.md"),
75 "\n"
76);
77
78/// Labels the executor's internal reasoning when a turn has no visible text,
79/// so the advisor still has evidence to review (reasoning models on vLLM/NIM
80/// can emit turns whose only output is reasoning).
81const REASONING_TAIL_LABEL: &str =
82 "(the executor produced no visible text this turn; its internal reasoning follows)\n";
83/// REDO echo when the discarded turn had neither text nor reasoning; strict
84/// endpoints (Anthropic) reject empty text blocks, so never echo "".
85const EMPTY_ECHO_PLACEHOLDER: &str = "(the executor produced no output this turn)";
86/// Benchmark harnesses stamp every request of one evaluation — sub-agents
87/// included — with this header, so it is the review budget's first-choice
88/// scope: "reviews for *this* task" survives gateways shared by many tasks.
89const BENCH_SESSION_HEADER: &str = "proxy_x_session_id";
90
91/// How the gate decides a buffered executor turn is terminal.
92#[derive(Clone, Debug, PartialEq)]
93pub enum GateTrigger {
94 /// First turn without tool calls (subject to `gate_min_tool_results`).
95 NoToolCall,
96 /// First turn whose visible text matches this regex (searched, not anchored) —
97 /// for text-protocol harnesses where every turn lacks tool calls and
98 /// completion is declared with a textual marker instead.
99 Pattern(String),
100}
101
102/// Gate knobs; defaults mirror the benchmarked Python advisor configuration.
103#[derive(Clone, Debug)]
104pub struct AdvisorGateConfig {
105 /// System prompt for the advisor's review call; states the APPROVE/REDO contract.
106 pub reviewer_system_prompt: String,
107 /// Prepended to the advisor's REDO plan when fed back to the executor.
108 pub redo_feedback_prefix: String,
109 /// What fires the review.
110 pub gate_trigger: GateTrigger,
111 /// Reviews allowed per budget scope. 1 keeps the original once-per-task
112 /// gate; higher values re-review later terminal turns, making the gate a
113 /// sequential best-of-(N+1) with the advisor as judge.
114 pub max_reviews: u32,
115 /// When > 0, additionally review (once per conversation, consuming budget)
116 /// the first request already carrying at least this many assistant turns —
117 /// a mid-task checkpoint for executors that grind without declaring
118 /// completion. 0 disables.
119 pub gate_stall_turns: u32,
120 /// For the `no_tool_call` trigger: only review once the conversation
121 /// carries at least this many tool results, skipping early commentary
122 /// turns on chatty harnesses. 0 reviews from the first terminal turn.
123 pub gate_min_tool_results: u32,
124 /// Cap on the advisor's output per consult.
125 pub advisor_max_tokens: u64,
126 /// Sampling temperature for the consult; `None` omits the field on the wire.
127 pub advisor_temperature: Option<f64>,
128 /// Cap on the serialized transcript handed to the advisor; the middle of
129 /// an over-cap conversation is dropped (task head + recent tail survive).
130 pub transcript_max_chars: usize,
131 /// When true (default), an advisor failure degrades to APPROVE; when
132 /// false, it propagates as the turn's error.
133 pub fail_open: bool,
134}
135
136impl Default for AdvisorGateConfig {
137 fn default() -> Self {
138 Self {
139 reviewer_system_prompt: REVIEWER_SYSTEM_PROMPT.to_string(),
140 redo_feedback_prefix: REDO_FEEDBACK_PREFIX.to_string(),
141 gate_trigger: GateTrigger::NoToolCall,
142 max_reviews: 1,
143 gate_stall_turns: 0,
144 gate_min_tool_results: 0,
145 advisor_max_tokens: 2048,
146 advisor_temperature: None,
147 transcript_max_chars: 200_000,
148 fail_open: true,
149 }
150 }
151}
152
153/// Advisor review gate: executor turns pass through until the first terminal
154/// turn, which a stronger advisor reviews once per scope budget (APPROVE
155/// releases it, REDO feeds the plan back and re-invokes the executor).
156pub struct AdvisorGate {
157 config: AdvisorGateConfig,
158 /// Folds request- and response-side facts into the per-turn [`GateSignals`].
159 signals: GateSignalProcessor,
160 /// Decides, from the signals, whether the buffered turn warrants review.
161 trigger: TriggerClassifier,
162 /// Reserve/refund review ledger and stall latch — the gate's only mutable state.
163 budget: ReviewBudget,
164 verdict_re: regex::Regex,
165}
166
167impl AdvisorGate {
168 /// Validates the config. Models are supplied when each request runs.
169 pub fn new(config: AdvisorGateConfig) -> Result<Self> {
170 if config.max_reviews < 1 {
171 return Err(algorithm_error("max_reviews must be at least 1"));
172 }
173 if config.advisor_max_tokens < 1 {
174 return Err(algorithm_error("advisor_max_tokens must be at least 1"));
175 }
176 if config.transcript_max_chars < 256 {
177 return Err(algorithm_error("transcript_max_chars must be at least 256"));
178 }
179 let trigger = TriggerClassifier::new(&config)?;
180 let verdict_re = regex::Regex::new(VERDICT_PATTERN).map_err(|error| {
181 algorithm_error(format!("verdict pattern failed to compile: {error}"))
182 })?;
183 let budget = ReviewBudget::new(config.max_reviews);
184 Ok(Self {
185 config,
186 signals: GateSignalProcessor,
187 trigger,
188 budget,
189 verdict_re,
190 })
191 }
192
193 // ── Gate flow ───────────────────────────────────────────────────────────
194
195 async fn route_inner(
196 &self,
197 driver: &Driver,
198 request: Request,
199 scope: &ScopeKey,
200 ) -> Result<RoutingOutcome> {
201 let executor_models = driver.models_for(&Category::Efficient).to_vec();
202 let executor = executor_models
203 .first()
204 .ok_or_else(|| LibsyError::AlgorithmError {
205 message: "no models available for category Efficient".to_string(),
206 })?;
207
208 // Spent budget (or failure cap): pure passthrough — live stream,
209 // verbatim preserved-body replay, zero buffering. Executor errors
210 // (including ContextWindowExceeded) propagate for the host's
211 // client-visible mapping.
212 if self.budget.check_exhausted(scope) {
213 return Ok(RoutingOutcome::route_to(
214 executor.clone(),
215 executor_models[1..].to_vec(),
216 request,
217 ));
218 }
219
220 // Request-side signals fold in before the executor runs.
221 let mut request = request;
222 let mut signals = GateSignals::default();
223 self.signals
224 .process(
225 &mut signals,
226 Event::Request {
227 request: &mut request,
228 driver,
229 },
230 )
231 .await?;
232
233 // Gated phase: generate the turn once, fully buffered, so the gate
234 // can inspect it before the client sees anything.
235 let response = driver
236 .call_model(request.clone(), executor_models.clone())
237 .await?;
238 let served_executor = response
239 .served_model()
240 .cloned()
241 .unwrap_or_else(|| executor.clone());
242 let turn = buffer_response(served_executor.as_str(), response).await?;
243
244 // Response-side signals fold in after it: the terminal turn never
245 // appears on a later request, so the trigger runs on this event.
246 self.signals
247 .process(&mut signals, Event::ModelResponse(&turn.agg))
248 .await?;
249
250 let decision = self.trigger.classify(&signals);
251 // The stall checkpoint fires once per conversation regardless of the
252 // turn's shape. Only a stall with no simultaneous trigger latches
253 // (atomically — one winner per conversation). The latch is provisional
254 // until a review completes: a refunded consult or a spent budget
255 // re-arms it so a later eligible turn is reviewed instead of silently
256 // passing through.
257 let stall_key = stall_key(&request);
258 let stall = decision.fired.is_none()
259 && decision.stalled
260 && self.budget.try_mark_stall_fired(stall_key);
261 if decision.fired.is_none() && !stall {
262 return Ok(RoutingOutcome::answered(
263 served_executor.clone(),
264 request,
265 turn.into_response(),
266 ));
267 }
268 if !self.budget.try_reserve(scope) {
269 if stall {
270 self.budget.clear_stall_fired(stall_key);
271 }
272 return Ok(RoutingOutcome::answered(
273 served_executor.clone(),
274 request,
275 turn.into_response(),
276 ));
277 }
278
279 let trigger_label = decision.fired.unwrap_or("stall");
280 let review_tail = visible_text(&turn.agg).or_else(|| {
281 reasoning_text(&turn.agg).map(|reasoning| format!("{REASONING_TAIL_LABEL}{reasoning}"))
282 });
283 match self
284 .consult(driver, &request, review_tail.as_deref(), trigger_label)
285 .await
286 {
287 Ok(ConsultOutcome::Approve) => {
288 driver.set_evidence(serde_json::json!({
289 "source": "advisor",
290 "verdict": "approve",
291 "trigger": trigger_label,
292 }));
293 Ok(RoutingOutcome::answered(
294 served_executor.clone(),
295 request,
296 turn.into_response(),
297 ))
298 }
299 Ok(ConsultOutcome::Redo { plan }) => {
300 driver.set_evidence(serde_json::json!({
301 "source": "advisor",
302 "verdict": "redo",
303 "trigger": trigger_label,
304 }));
305 Ok(self.redo(
306 executor,
307 &executor_models[1..],
308 &served_executor,
309 request,
310 turn,
311 &plan,
312 ))
313 }
314 Ok(ConsultOutcome::Failed { reason }) => {
315 self.budget.refund_failure(scope);
316 if stall {
317 self.budget.clear_stall_fired(stall_key);
318 }
319 driver.set_evidence(serde_json::json!({
320 "source": "advisor",
321 "verdict": "fail_open",
322 "trigger": trigger_label,
323 "reason_code": reason,
324 }));
325 Ok(RoutingOutcome::answered(
326 served_executor,
327 request,
328 turn.into_response(),
329 ))
330 }
331 Err(error) => {
332 self.budget.refund_failure(scope);
333 if stall {
334 self.budget.clear_stall_fired(stall_key);
335 }
336 Err(error)
337 }
338 }
339 }
340
341 /// REDO: the client never sees the gated turn. Its text (or reasoning) is
342 /// echoed as an assistant message, the advisor's plan follows as user
343 /// feedback, and the executor continues as a pure passthrough call.
344 fn redo(
345 &self,
346 executor: &ModelId,
347 executor_fallbacks: &[ModelId],
348 served_executor: &ModelId,
349 request: Request,
350 turn: BufferedResponse,
351 plan: &str,
352 ) -> RoutingOutcome {
353 record_discarded(&turn.agg.usage);
354 emit_discarded_audit(served_executor.as_str(), &turn.agg.usage);
355 let echo = visible_text(&turn.agg)
356 .or_else(|| reasoning_text(&turn.agg))
357 .unwrap_or_else(|| EMPTY_ECHO_PLACEHOLDER.to_string());
358 let mut redo = request;
359 redo.llm_request
360 .messages
361 .push(Message::text(Role::Assistant, echo));
362 redo.llm_request.messages.push(Message::text(
363 Role::User,
364 format!("{}{}", self.config.redo_feedback_prefix, plan),
365 ));
366 // Mandatory after any message mutation: codecs otherwise replay the
367 // preserved pre-surgery body verbatim and the feedback never reaches
368 // the executor.
369 crate::algorithms::util::prompts::drop_exact_replay(&mut redo);
370 RoutingOutcome::route_to(executor.clone(), executor_fallbacks.to_vec(), redo)
371 }
372
373 /// Consults the advisor over the buffered transcript and parses the
374 /// verdict. `Ok(Failed)` covers fail-open errors and unparseable replies
375 /// (the caller refunds); fail-closed errors return `Err`.
376 async fn consult(
377 &self,
378 driver: &Driver,
379 base: &Request,
380 review_tail: Option<&str>,
381 trigger: &'static str,
382 ) -> Result<ConsultOutcome> {
383 // The advisor reviews the FULL transcript: system/developer content is
384 // normalized out of `messages` into `instructions`, so prepend it back
385 // as leading messages (identical {role, content} shape) — the task
386 // constraints the verdict must check against usually live there.
387 let transcript_messages: Vec<Message> = base
388 .llm_request
389 .instructions
390 .iter()
391 .map(|block| Message {
392 role: block.role,
393 content: block.content.clone(),
394 })
395 .chain(base.llm_request.messages.iter().cloned())
396 .collect();
397 let transcript = review_transcript(
398 &transcript_messages,
399 review_tail,
400 self.config.transcript_max_chars,
401 );
402 let consult_request = self.build_consult_request(base, transcript);
403 let started = Instant::now();
404 // An unresolvable advisor is treated like any other consult failure, so
405 // fail_open still returns the buffered executor turn to the client.
406 let reply = match driver.first_model_for(&Category::Judge) {
407 Ok(advisor) => {
408 let advisor = advisor.clone();
409 let advisor_models = driver.models_for(&Category::Judge).to_vec();
410 match driver.call_model(consult_request, advisor_models).await {
411 Ok(response) => {
412 let served_advisor = response
413 .served_model()
414 .cloned()
415 .unwrap_or_else(|| advisor.clone());
416 response
417 .llm_response
418 .into_agg()
419 .await
420 .map_err(|source| LibsyError::client_call(served_advisor, source))
421 }
422 Err(error) => Err(error),
423 }
424 }
425 Err(error) => Err(error),
426 };
427 let latency_ms = started.elapsed().as_secs_f64() * 1000.0;
428 let agg = match reply {
429 Ok(agg) => agg,
430 Err(error) => {
431 let reason = crate::algorithms::util::llm_judge::libsy_error_reason(&error);
432 record_consult_failure(reason);
433 if !self.config.fail_open {
434 // Surface as an algorithm failure (5xx), never as the
435 // advisor's own client error: a typed ContextWindowExceeded
436 // from the consult would otherwise reach the client as 400
437 // context_length_exceeded and trigger compaction of a
438 // healthy conversation.
439 return Err(algorithm_error(format!(
440 "advisor consult failed (fail_open = false): {error}"
441 )));
442 }
443 tracing::warn!(
444 target: "libsy",
445 error = %error,
446 "advisor gate: consult failed; passing the turn through (fail open)"
447 );
448 emit_review_audit(ReviewAudit {
449 verdict: "APPROVE",
450 error: Some(error.to_string()),
451 latency_ms,
452 reply_head: None,
453 usage: None,
454 });
455 return Ok(ConsultOutcome::Failed { reason });
456 }
457 };
458 let reply_text = advisor_reply_text(&agg);
459 let reply_head: String = reply_text.chars().take(160).collect();
460 match parse_verdict(&self.verdict_re, &reply_text) {
461 Some(Verdict::Approve) => {
462 record_review("approve", trigger);
463 emit_review_audit(ReviewAudit {
464 verdict: "APPROVE",
465 error: None,
466 latency_ms,
467 reply_head: Some(reply_head),
468 usage: Some(&agg.usage),
469 });
470 Ok(ConsultOutcome::Approve)
471 }
472 Some(Verdict::Redo { plan }) => {
473 record_review("redo", trigger);
474 emit_review_audit(ReviewAudit {
475 verdict: "REDO",
476 error: None,
477 latency_ms,
478 reply_head: Some(reply_head),
479 usage: Some(&agg.usage),
480 });
481 Ok(ConsultOutcome::Redo { plan })
482 }
483 None => {
484 // The advisor spent real tokens on a reply the gate cannot
485 // act on; the observer already recorded them. Refunded by
486 // the caller so a flaky advisor cannot burn the budget.
487 record_review("unparseable", trigger);
488 emit_review_audit(ReviewAudit {
489 verdict: "UNPARSEABLE",
490 error: None,
491 latency_ms,
492 reply_head: Some(reply_head),
493 usage: Some(&agg.usage),
494 });
495 Ok(ConsultOutcome::Failed {
496 reason: "parse_error",
497 })
498 }
499 }
500 }
501
502 /// A fresh, buffered, tool-free request carrying the reviewer contract and
503 /// the serialized transcript; metadata is kept for session correlation.
504 fn build_consult_request(&self, base: &Request, transcript: String) -> Request {
505 Request {
506 llm_request: LlmRequest {
507 model: base.llm_request.model.clone(),
508 instructions: vec![InstructionBlock {
509 role: Role::System,
510 content: vec![ContentBlock::Text {
511 text: self.config.reviewer_system_prompt.clone(),
512 }],
513 }],
514 messages: vec![Message::text(Role::User, transcript)],
515 sampling: SamplingParams {
516 temperature: self.config.advisor_temperature,
517 ..SamplingParams::default()
518 },
519 output: OutputParams {
520 max_output_tokens: Some(self.config.advisor_max_tokens),
521 response_format: None,
522 },
523 ..LlmRequest::default()
524 },
525 raw_request: None,
526 metadata: base.metadata.clone(),
527 }
528 }
529}
530
531#[async_trait::async_trait]
532impl Algorithm for AdvisorGate {
533 fn name(&self) -> &str {
534 "advisor_gate"
535 }
536
537 async fn route(self: Arc<Self>, driver: Driver, request: Request) -> Result<RoutingOutcome> {
538 let scope = budget_scope(&request);
539 let session_final = request
540 .metadata
541 .as_ref()
542 .and_then(|metadata| metadata.session_final)
543 == Some(true);
544 let result = self.route_inner(&driver, request, &scope).await;
545 if session_final {
546 self.budget.evict_scope(&scope);
547 }
548 result
549 }
550}
551
552/// Outcome of one consult; `Failed` = fail-open error or unparseable reply.
553enum ConsultOutcome {
554 Approve,
555 Redo { plan: String },
556 Failed { reason: &'static str },
557}
558
559fn algorithm_error(message: impl Into<String>) -> LibsyError {
560 LibsyError::AlgorithmError {
561 message: message.into(),
562 }
563}