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