From 14d4ab476872a5b9ded58754bde9f6ccb493a537 Mon Sep 17 00:00:00 2001 From: zengyuanl Date: Tue, 11 Aug 2026 20:05:41 +0000 Subject: [PATCH 1/2] feat(server): advisor_gate stats projection Signed-off-by: zengyuanl --- .../switchyard-server/src/stats/algorithms.rs | 19 ++ .../src/stats/algorithms/advisor_gate.rs | 257 ++++++++++++++++++ 2 files changed, 276 insertions(+) create mode 100644 crates/switchyard-server/src/stats/algorithms/advisor_gate.rs diff --git a/crates/switchyard-server/src/stats/algorithms.rs b/crates/switchyard-server/src/stats/algorithms.rs index 27bfc0ff..f7f0d49c 100644 --- a/crates/switchyard-server/src/stats/algorithms.rs +++ b/crates/switchyard-server/src/stats/algorithms.rs @@ -3,6 +3,7 @@ //! Server-owned projections of algorithm OpenTelemetry metrics. +mod advisor_gate; mod stage_router; use std::collections::HashSet; @@ -10,19 +11,27 @@ use std::collections::HashSet; use prometheus::Registry; use serde::Serialize; +use advisor_gate::{AdvisorGateCumulative, AdvisorGateStatsSnapshot}; use stage_router::{StageRouterCumulative, StageRouterStatsSnapshot}; +const ADVISOR_GATE: &str = "advisor_gate"; const STAGE_ROUTER: &str = "stage_router"; /// Owns algorithm metric baselines behind the generic server stats interface. pub(super) struct AlgorithmStats { registry: Registry, + advisor_gate_baseline: Option, stage_router_baseline: Option, } /// Curated algorithm-specific data included in the JSON stats response. +/// +/// Each block is present only when the deployment contains a route running +/// the matching algorithm; deployments without one omit the key entirely. #[derive(Clone, Debug, Default, PartialEq, Serialize)] pub(crate) struct AlgorithmStatsSnapshot { + #[serde(skip_serializing_if = "Option::is_none")] + pub advisor_gate: Option, #[serde(skip_serializing_if = "Option::is_none")] pub stage_router: Option, } @@ -35,6 +44,9 @@ impl AlgorithmStats { let algorithms: HashSet<_> = algorithms.into_iter().collect(); let families = registry.gather(); Self { + advisor_gate_baseline: algorithms + .contains(ADVISOR_GATE) + .then(|| AdvisorGateCumulative::collect(&families)), stage_router_baseline: algorithms .contains(STAGE_ROUTER) .then(|| StageRouterCumulative::collect(&families)), @@ -45,6 +57,10 @@ impl AlgorithmStats { pub(super) fn snapshot(&self) -> AlgorithmStatsSnapshot { let families = self.registry.gather(); AlgorithmStatsSnapshot { + advisor_gate: self + .advisor_gate_baseline + .as_ref() + .map(|baseline| AdvisorGateCumulative::collect(&families).delta(baseline)), stage_router: self .stage_router_baseline .as_ref() @@ -53,6 +69,9 @@ impl AlgorithmStats { } pub(super) fn reset(&mut self) { + if let Some(baseline) = &mut self.advisor_gate_baseline { + *baseline = AdvisorGateCumulative::collect(&self.registry.gather()); + } if let Some(baseline) = &mut self.stage_router_baseline { *baseline = StageRouterCumulative::collect(&self.registry.gather()); } diff --git a/crates/switchyard-server/src/stats/algorithms/advisor_gate.rs b/crates/switchyard-server/src/stats/algorithms/advisor_gate.rs new file mode 100644 index 00000000..c4ac66d2 --- /dev/null +++ b/crates/switchyard-server/src/stats/algorithms/advisor_gate.rs @@ -0,0 +1,257 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Advisor-gate projection from cumulative Prometheus metric families. + +use std::collections::BTreeMap; + +use prometheus::proto::{Metric, MetricFamily}; +use serde::Serialize; + +const REVIEWS_METRIC: &str = "switchyard_advisor_gate_reviews_total"; +const CONSULT_FAILURES_METRIC: &str = "switchyard_advisor_gate_consult_failures_total"; +const DISCARDED_TURNS_METRIC: &str = "switchyard_advisor_gate_discarded_turns_total"; +const DISCARDED_TOKENS_METRIC: &str = "switchyard_advisor_gate_discarded_tokens_total"; + +#[derive(Clone, Debug, Default)] +pub(super) struct AdvisorGateCumulative { + reviews: BTreeMap, + consult_failures: BTreeMap, + discarded_turns: u64, + discarded_tokens: BTreeMap, +} + +#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)] +struct ReviewKey { + verdict: String, + trigger: String, +} + +/// Human-readable advisor-gate stats derived from its native metrics. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct AdvisorGateStatsSnapshot { + /// Verdicts handed down since the last reset, split by what gated the turn. + pub reviews: BTreeMap, + /// Advisor consults that failed outright, by bounded reason label. + pub consult_failures: BTreeMap, + /// Executor turns (and their tokens) discarded by REDO verdicts; the + /// client never saw them, so terminal usage accounting never priced them. + pub discarded: DiscardedStatsSnapshot, +} + +/// One verdict's counts, split by trigger. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct ReviewStatsSnapshot { + pub total: u64, + pub by_trigger: BTreeMap, +} + +/// REDO-discarded executor turns and their token kinds. +#[derive(Clone, Debug, Default, PartialEq, Serialize)] +pub(crate) struct DiscardedStatsSnapshot { + pub turns: u64, + pub tokens: BTreeMap, +} + +impl AdvisorGateCumulative { + pub(super) fn collect(families: &[MetricFamily]) -> Self { + Self { + reviews: collect_labeled_pairs(families, REVIEWS_METRIC, "verdict", "trigger") + .into_iter() + .map(|((verdict, trigger), count)| (ReviewKey { verdict, trigger }, count)) + .collect(), + consult_failures: collect_labeled(families, CONSULT_FAILURES_METRIC, "reason"), + discarded_turns: collect_total(families, DISCARDED_TURNS_METRIC), + discarded_tokens: collect_labeled(families, DISCARDED_TOKENS_METRIC, "kind"), + } + } + + pub(super) fn delta(&self, baseline: &Self) -> AdvisorGateStatsSnapshot { + let mut reviews: BTreeMap = BTreeMap::new(); + for (key, current) in &self.reviews { + let count = current.saturating_sub(*baseline.reviews.get(key).unwrap_or(&0)); + if count == 0 { + continue; + } + let verdict = reviews.entry(key.verdict.clone()).or_default(); + verdict.total = verdict.total.saturating_add(count); + verdict.by_trigger.insert(key.trigger.clone(), count); + } + AdvisorGateStatsSnapshot { + reviews, + consult_failures: map_delta(&self.consult_failures, &baseline.consult_failures), + discarded: DiscardedStatsSnapshot { + turns: self + .discarded_turns + .saturating_sub(baseline.discarded_turns), + tokens: map_delta(&self.discarded_tokens, &baseline.discarded_tokens), + }, + } + } +} + +fn map_delta( + current: &BTreeMap, + baseline: &BTreeMap, +) -> BTreeMap { + current + .iter() + .filter_map(|(key, value)| { + let delta = value.saturating_sub(*baseline.get(key).unwrap_or(&0)); + (delta > 0).then(|| (key.clone(), delta)) + }) + .collect() +} + +fn collect_labeled( + families: &[MetricFamily], + metric_name: &str, + label_name: &str, +) -> BTreeMap { + let mut counts = BTreeMap::new(); + for metric in metrics(families, metric_name) { + let Some(key) = label(metric, label_name) else { + continue; + }; + if let Some(value) = counter_value(metric) { + let count = counts.entry(key.to_string()).or_insert(0u64); + *count = count.saturating_add(value); + } + } + counts +} + +fn collect_labeled_pairs( + families: &[MetricFamily], + metric_name: &str, + first_label: &str, + second_label: &str, +) -> BTreeMap<(String, String), u64> { + let mut counts = BTreeMap::new(); + for metric in metrics(families, metric_name) { + let (Some(first), Some(second)) = (label(metric, first_label), label(metric, second_label)) + else { + continue; + }; + if let Some(value) = counter_value(metric) { + let count = counts + .entry((first.to_string(), second.to_string())) + .or_insert(0u64); + *count = count.saturating_add(value); + } + } + counts +} + +fn collect_total(families: &[MetricFamily], metric_name: &str) -> u64 { + metrics(families, metric_name) + .filter_map(counter_value) + .fold(0u64, u64::saturating_add) +} + +fn counter_value(metric: &Metric) -> Option { + let counter = metric.get_counter().as_ref()?; + let value = counter.value(); + (value.is_finite() && value > 0.0).then_some(value as u64) +} + +fn metrics<'a>(families: &'a [MetricFamily], name: &'a str) -> impl Iterator { + families + .iter() + .filter(move |family| family.name() == name) + .flat_map(|family| family.get_metric()) +} + +fn label<'a>(metric: &'a Metric, name: &str) -> Option<&'a str> { + metric + .get_label() + .iter() + .find(|label| label.name() == name) + .map(|label| label.value()) +} + +#[cfg(test)] +mod tests { + use opentelemetry::KeyValue; + use opentelemetry::metrics::MeterProvider as _; + use opentelemetry_sdk::metrics::SdkMeterProvider; + use prometheus::Registry; + + use super::*; + use crate::stats::StatsAccumulator; + + #[test] + fn advisor_gate_projection_preserves_reviews_discards_and_reset_baseline() { + let registry = Registry::new(); + let exporter = opentelemetry_prometheus::exporter() + .with_registry(registry.clone()) + .build() + .unwrap_or_else(|error| panic!("failed to build metrics exporter: {error}")); + let provider = SdkMeterProvider::builder().with_reader(exporter).build(); + let meter = provider.meter("switchyard"); + let stats = StatsAccumulator::new(registry, ["advisor_gate"]); + + let reviews = meter.u64_counter("switchyard.advisor_gate.reviews").build(); + reviews.add( + 2, + &[ + KeyValue::new("verdict", "approve"), + KeyValue::new("trigger", "no_tool_call"), + ], + ); + reviews.add( + 1, + &[ + KeyValue::new("verdict", "redo"), + KeyValue::new("trigger", "stall"), + ], + ); + meter + .u64_counter("switchyard.advisor_gate.consult_failures") + .build() + .add(1, &[KeyValue::new("reason", "client_error")]); + meter + .u64_counter("switchyard.advisor_gate.discarded_turns") + .build() + .add(1, &[]); + let tokens = meter + .u64_counter("switchyard.advisor_gate.discarded_tokens") + .build(); + tokens.add(120, &[KeyValue::new("kind", "input")]); + tokens.add(30, &[KeyValue::new("kind", "output")]); + + let snapshot = stats.snapshot(); + let gate = snapshot + .algorithm_stats + .advisor_gate + .unwrap_or_else(|| panic!("advisor-gate stats missing")); + assert_eq!(gate.reviews["approve"].total, 2); + assert_eq!(gate.reviews["approve"].by_trigger["no_tool_call"], 2); + assert_eq!(gate.reviews["redo"].by_trigger["stall"], 1); + assert_eq!(gate.consult_failures["client_error"], 1); + assert_eq!(gate.discarded.turns, 1); + assert_eq!(gate.discarded.tokens["input"], 120); + assert_eq!(gate.discarded.tokens["output"], 30); + + stats.reset(); + assert_eq!( + stats.snapshot().algorithm_stats.advisor_gate, + Some(AdvisorGateStatsSnapshot::default()) + ); + + reviews.add( + 1, + &[ + KeyValue::new("verdict", "unparseable"), + KeyValue::new("trigger", "pattern"), + ], + ); + let after_reset = stats + .snapshot() + .algorithm_stats + .advisor_gate + .unwrap_or_else(|| panic!("advisor-gate stats missing after reset")); + assert_eq!(after_reset.reviews["unparseable"].total, 1); + assert!(!after_reset.reviews.contains_key("approve")); + } +} From 94ea1d1b10d47a965580951244d854bb04340276 Mon Sep 17 00:00:00 2001 From: zengyuanl Date: Tue, 11 Aug 2026 20:11:50 +0000 Subject: [PATCH 2/2] test(server): advisor route end-to-end coverage Signed-off-by: zengyuanl --- crates/switchyard-server/tests/server.rs | 393 +++++++++++++++++++++++ 1 file changed, 393 insertions(+) diff --git a/crates/switchyard-server/tests/server.rs b/crates/switchyard-server/tests/server.rs index a1241c59..7c0b7e21 100644 --- a/crates/switchyard-server/tests/server.rs +++ b/crates/switchyard-server/tests/server.rs @@ -154,6 +154,37 @@ async fn upstream_chat( return Sse::new(stream).into_response(); } + if model == "model/advisor" { + // The review consult carries the serialized transcript in its user + // message, so the original prompt text rides inside it: tests script + // the verdict (or an outage) from the prompt they send. + let haystack = body["messages"].to_string(); + if haystack.contains("advisor-down") { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(json!({"error": {"message": "advisor is unavailable"}})), + ) + .into_response(); + } + let verdict = if haystack.contains("please-redo") { + "REDO run the tests" + } else { + "APPROVE" + }; + return Json(json!({ + "id": "chatcmpl-advisor", + "object": "chat.completion", + "model": model, + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": verdict}, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 40, "completion_tokens": 4, "total_tokens": 44} + })) + .into_response(); + } + let custom_target_schema = body .pointer("/response_format/json_schema/schema/properties/decision/properties/target") .is_some(); @@ -2132,3 +2163,365 @@ async fn request_and_upstream_errors_use_the_inbound_wire_format() -> TestResult ); Ok(()) } + +/// A `type = "advisor"` deployment: gated executor + reviewer on one mock upstream. +fn advisor_state(base_url: &str) -> TestResult { + load_test_config(&format!( + r#" +schema_version = 1 + +[llm_clients.upstream] +format = "openai_chat" +base_url = "{base_url}" + +[targets.executor] +id = "model/executor" +llm_client = "upstream" + +[targets.advisor] +id = "model/advisor" +llm_client = "upstream" + +[routes.gated] +id = "switchyard/advisor" +type = "advisor" +executor_target = "executor" +advisor_target = "advisor" +"#, + )) +} + +fn advisor_chat_body(prompt: &str) -> Value { + json!({ + "model": "switchyard/advisor", + "messages": [{"role": "user", "content": prompt}] + }) +} + +#[tokio::test] +async fn advisor_route_approve_flow_and_stats() -> TestResult { + let upstream = MockUpstream::start().await?; + let app = build_switchyard_router(advisor_state(&upstream.base_url)?); + + let response = send( + &app, + "POST", + "/v1/chat/completions", + Some(advisor_chat_body("hi")), + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); + assert_eq!( + response + .headers + .get("x-model-router-selected-model") + .and_then(|value| value.to_str().ok()), + Some("model/executor") + ); + // Executor turn first, then the review consult. + assert_eq!(upstream.models().await, ["model/executor", "model/advisor"]); + + let stats = send(&app, "GET", "/v1/stats", None).await?.json()?; + assert_eq!(stats["models"]["model/executor"]["calls"], 1); + // The consult lands in the classifier bucket with its usage. + assert_eq!(stats["classifier"]["models"]["model/advisor"]["calls"], 1); + assert_eq!(stats["classifier"]["total_tokens"]["prompt"], 40); + Ok(()) +} + +#[tokio::test] +async fn advisor_route_budget_scoped_by_proxy_header() -> TestResult { + let upstream = MockUpstream::start().await?; + let app = build_switchyard_router(advisor_state(&upstream.base_url)?); + + for (session, expected_consults) in [("eval-a", 1), ("eval-a", 1), ("eval-b", 2)] { + let response = send_with_headers( + &app, + "POST", + "/v1/chat/completions", + Some(advisor_chat_body("hi")), + &[("proxy_x_session_id", session)], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + let consults = upstream + .models() + .await + .iter() + .filter(|model| *model == "model/advisor") + .count(); + assert_eq!(consults, expected_consults, "session {session}"); + } + Ok(()) +} + +#[tokio::test] +async fn advisor_route_streaming_approval_replays_provider_events() -> TestResult { + let upstream = MockUpstream::start().await?; + let app = build_switchyard_router(advisor_state(&upstream.base_url)?); + + let mut body = advisor_chat_body("hi"); + body["stream"] = json!(true); + let response = send(&app, "POST", "/v1/chat/completions", Some(body)).await?; + assert_eq!(response.status, StatusCode::OK); + // The gate buffered the executor stream for the review, then replayed the + // provider events verbatim. + assert_eq!(upstream.models().await, ["model/executor", "model/advisor"]); + let text = response.text()?; + let events: Vec = text + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .filter(|data| *data != "[DONE]") + .map(serde_json::from_str) + .collect::>()?; + assert_eq!(events.len(), 5); + assert_eq!(events[1]["choices"][0]["delta"]["content"], "hello"); + assert_eq!(events[2]["choices"][0]["delta"]["content"], "-partial"); + assert_eq!(events[3]["choices"][0]["delta"]["content"], "-final"); + // Provider-specific usage detail rides through untouched. + assert_eq!( + events[3]["usage"]["prompt_tokens_details"]["cache_creation_tokens"], + 2 + ); + assert_eq!(events[4]["choices"][0]["finish_reason"], "stop"); + assert!(text.trim_end().ends_with("data: [DONE]")); + Ok(()) +} + +#[tokio::test] +async fn advisor_route_routing_log_records_classifier_tier() -> TestResult { + let upstream = MockUpstream::start().await?; + let temp_dir = tempfile::tempdir()?; + let log_path = temp_dir.path().join("routing.jsonl"); + let state = advisor_state(&upstream.base_url)?.with_routing_log(&log_path)?; + let app = build_switchyard_router(state); + + let response = send_with_headers( + &app, + "POST", + "/v1/chat/completions", + Some(advisor_chat_body("hi")), + &[("proxy_x_session_id", "session-1")], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + + let records: Vec = std::fs::read_to_string(&log_path)? + .lines() + .map(serde_json::from_str) + .collect::>()?; + // The consult is appended under the shared judge tier; the served turn is + // the terminal answer row. The discarded-turn row does not exist in v1 — + // its tokens live in the advisor_gate stats block instead. + assert_eq!(records.len(), 2); + let consult = records + .iter() + .find(|record| record["model"] == "model/advisor") + .ok_or("consult row present")?; + assert_eq!(consult["tier"], "classifier"); + assert_eq!(consult["session_id"], "session-1"); + assert_eq!(consult["prompt_tokens"], 40); + Ok(()) +} + +#[tokio::test] +async fn advisor_route_count_tokens_uses_executor() -> TestResult { + let upstream = MockUpstream::start().await?; + let state = load_test_config(&format!( + r#" +schema_version = 1 + +[llm_clients.claude] +format = "anthropic_messages" +base_url = "{base_url}" + +[targets.executor] +id = "model/executor" +llm_client = "claude" + +[targets.advisor] +id = "model/advisor" +llm_client = "claude" + +[routes.gated] +id = "switchyard/advisor" +type = "advisor" +executor_target = "executor" +advisor_target = "advisor" +"#, + base_url = upstream.base_url + ))?; + let app = build_switchyard_router(state); + + let response = send( + &app, + "POST", + "/v1/messages/count_tokens", + Some(json!({ + "model": "switchyard/advisor", + "messages": [{"role": "user", "content": "hi"}] + })), + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["input_tokens"], 7); + // The executor is the route's only completion target, so it backs + // count_tokens; the judge-only advisor never does. + let calls = upstream.calls.lock().await; + assert_eq!(calls.len(), 1); + assert_eq!(calls[0]["model"], "model/executor"); + Ok(()) +} + +/// An advisor deployment whose reviewer client never retries, so a down +/// advisor hits fail-open after a single attempt (the documented deployment +/// posture for the advisor tier). +fn advisor_state_no_retry(base_url: &str) -> TestResult { + load_test_config(&format!( + r#" +schema_version = 1 + +[llm_clients.upstream] +format = "openai_chat" +base_url = "{base_url}" + +[llm_clients.reviewer] +format = "openai_chat" +base_url = "{base_url}" +max_retries = 0 + +[targets.executor] +id = "model/executor" +llm_client = "upstream" + +[targets.advisor] +id = "model/advisor" +llm_client = "reviewer" + +[routes.gated] +id = "switchyard/advisor" +type = "advisor" +executor_target = "executor" +advisor_target = "advisor" +"#, + )) +} + +fn gate_count(stats: &Value, path: &[&str]) -> u64 { + let mut value = &stats["algorithm_stats"]["advisor_gate"]; + for key in path { + value = &value[*key]; + } + value.as_u64().unwrap_or(0) +} + +// REDO mechanics, fail-open, and the /v1/stats advisor_gate projection in one +// sequential test: the OpenTelemetry meter behind algorithm_stats is +// process-global, so this is the only test that emits redo / consult-failure +// metrics and the only one that may assert their exact counts. +#[tokio::test] +async fn advisor_route_redo_fail_open_and_stats_projection() -> TestResult { + let upstream = MockUpstream::start().await?; + let app = build_switchyard_router(advisor_state_no_retry(&upstream.base_url)?); + let before = send(&app, "GET", "/v1/stats", None).await?.json()?; + + // REDO: the gated turn is discarded, the advisor plan is fed back, and + // the executor continues. Each flow gets its own budget scope so the + // second one is still reviewable. + let response = send_with_headers( + &app, + "POST", + "/v1/chat/completions", + Some(advisor_chat_body("please-redo")), + &[("proxy_x_session_id", "redo-flow")], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); + assert_eq!( + upstream.models().await, + ["model/executor", "model/advisor", "model/executor"] + ); + let calls = upstream.calls.lock().await; + let redo_messages = calls[2]["messages"] + .as_array() + .ok_or("redo call has messages")? + .clone(); + drop(calls); + assert_eq!(redo_messages.len(), 3); + assert_eq!(redo_messages[1]["role"], "assistant"); + assert_eq!(redo_messages[1]["content"], "ok"); + assert_eq!(redo_messages[2]["role"], "user"); + let feedback = redo_messages[2]["content"] + .as_str() + .ok_or("feedback is text")?; + assert!(feedback.starts_with("A senior reviewer examined your work")); + assert!(feedback.ends_with("run the tests")); + + // Fail-open: the advisor 503s once (no retries) and the turn still flows. + let response = send_with_headers( + &app, + "POST", + "/v1/chat/completions", + Some(advisor_chat_body("advisor-down")), + &[("proxy_x_session_id", "fail-flow")], + ) + .await?; + assert_eq!(response.status, StatusCode::OK); + assert_eq!(response.json()?["choices"][0]["message"]["content"], "ok"); + assert_eq!( + upstream.models().await, + [ + "model/executor", + "model/advisor", + "model/executor", + "model/executor", + "model/advisor", + ] + ); + + let stats = send(&app, "GET", "/v1/stats", None).await?.json()?; + // State-owned accumulator: three executor answer calls, one failed consult. + assert_eq!(stats["models"]["model/executor"]["calls"], 3); + assert_eq!(stats["classifier"]["total_errors"], 1); + // Projection deltas for the metrics only this test emits. + let redo = gate_count(&stats, &["reviews", "redo", "total"]) + - gate_count(&before, &["reviews", "redo", "total"]); + assert_eq!(redo, 1); + assert_eq!( + gate_count(&stats, &["reviews", "redo", "by_trigger", "no_tool_call"]), + gate_count(&before, &["reviews", "redo", "by_trigger", "no_tool_call"]) + 1 + ); + assert_eq!( + gate_count(&stats, &["discarded", "turns"]), + gate_count(&before, &["discarded", "turns"]) + 1 + ); + // Mock usage: prompt 10 with 7 cached -> 3 non-cached input, 2 output. + assert_eq!( + gate_count(&stats, &["discarded", "tokens", "input"]), + gate_count(&before, &["discarded", "tokens", "input"]) + 3 + ); + assert_eq!( + gate_count(&stats, &["discarded", "tokens", "cached"]), + gate_count(&before, &["discarded", "tokens", "cached"]) + 7 + ); + assert_eq!( + gate_count(&stats, &["discarded", "tokens", "output"]), + gate_count(&before, &["discarded", "tokens", "output"]) + 2 + ); + // The 503 maps to the bounded upstream_5xx reason label. + assert_eq!( + gate_count(&stats, &["consult_failures", "upstream_5xx"]), + gate_count(&before, &["consult_failures", "upstream_5xx"]) + 1 + ); + + // Reset re-baselines the projection: the redo/discard counts this test + // produced disappear from the next snapshot. + let reset = send(&app, "POST", "/v1/stats/reset", None).await?; + assert_eq!(reset.status, StatusCode::OK); + let stats = send(&app, "GET", "/v1/stats", None).await?.json()?; + assert_eq!(gate_count(&stats, &["reviews", "redo", "total"]), 0); + assert_eq!(gate_count(&stats, &["discarded", "turns"]), 0); + Ok(()) +}