diff --git a/AGENTS.md b/AGENTS.md index cc046923e..8960d966e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -158,6 +158,21 @@ Direct Rust bindings for migrated concrete processors/backends are exposed from | Response component | Plain Python/Rust object | `async process(ctx, response) -> ChatResponse` | Post-process (logging, stats) | | `TranslationEngine` | `switchyard_rust.translation` | `async translate(ctx, request, response) -> Any` | Convert to client's wire format | +### Multi-call backends + +`AdvisorLoopBackend` (`switchyard/lib/backends/advisor_loop_backend.py`) pairs the executor with a +stronger advisor that reviews the executor's first no-tool-call turn once per session (APPROVE +returns it; REDO feeds the advisor's plan back and re-invokes the executor). The trigger is +proxy-side, so it fires even for executors that rarely call tools; advisor text goes into the +first user message, never the newest turn, so the upstream cache prefix stays stable across a +session. It is multi-call — one `call(...)` issues several upstream requests before returning one +`ChatResponse`, so "exactly one `LLMBackend` per chain" holds at the chain level only — and it +does its own stats accounting into the classifier bucket, so it must not be wrapped in +`StatsLlmBackend` (which rejects Python-only backends); the route-bundle builder injects the +accumulator through the constructor instead. Executor and advisor targets dispatch independently +on `LlmTarget.format`; `responses` is rejected at `AdvisorConfig` validation. Compose with a +`type: advisor` route (`switchyard/cli/route_bundle.py`) or an `AdvisorPresets` helper. + ## Project Structure ``` @@ -181,6 +196,8 @@ switchyard/ │ │ ├── openai_llm_backend.py # OpenAiPassthroughBackend │ │ ├── openai_native_backend.py # OpenAiNativeBackend │ │ ├── anthropic_native_llm_backend.py # AnthropicNativeBackend +│ │ ├── advisor_loop_backend.py # AdvisorLoopBackend (advisor review gate) +│ │ ├── advisor_config.py # AdvisorConfig (+ advisor_prompts, advisor_presets) │ │ ├── llm_target.py # LlmTarget, BackendFormat │ │ ├── multi_llm_backend.py # MultiLlmBackend helpers │ │ ├── stats_llm_backend.py # StatsLlmBackend @@ -242,7 +259,7 @@ and their transitives never appear in downstream vulnerability scans. ```bash export OPENROUTER_API_KEY="sk-or-..." -# Serve the minimal Python YAML bundle (noop and passthrough only). +# Serve the minimal Python YAML bundle (noop, passthrough, and advisor routes). switchyard serve --routes examples/route.yaml --port 4000 # Launch against the packaged OpenRouter deployment. diff --git a/benchmark/run-baseline.sh b/benchmark/run-baseline.sh index fa0881658..36617a08d 100755 --- a/benchmark/run-baseline.sh +++ b/benchmark/run-baseline.sh @@ -881,9 +881,21 @@ if [[ "\${SERVER_ENABLED}" == "1" ]]; then DOCKER_RUN_ARGS=( -d --rm --name "\${SWITCHYARD_DOCKER_CONTAINER}" - --network "\${SWITCHYARD_DOCKER_NETWORK}" - --network-alias "\${SWITCHYARD_DOCKER_SERVICE_NAME}" - -p "127.0.0.1:$(q "${PORT}"):$(q "${PORT}")" + ) + if [[ "\${SWITCHYARD_DOCKER_NETWORK_MODE:-bridge}" == "host" ]]; then + # Host networking: for upstreams only routable from the host (VPN / + # corp-internal gateways that Docker bridge networks cannot reach). + # Pair with --harbor-server-url http://: so task + # containers reach the server at the host address. + DOCKER_RUN_ARGS+=(--network host) + else + DOCKER_RUN_ARGS+=( + --network "\${SWITCHYARD_DOCKER_NETWORK}" + --network-alias "\${SWITCHYARD_DOCKER_SERVICE_NAME}" + -p "127.0.0.1:$(q "${PORT}"):$(q "${PORT}")" + ) + fi + DOCKER_RUN_ARGS+=( -v "\${REPO_ROOT}:\${REPO_ROOT}:ro" -v "\${RUN_DIR}:\${RUN_DIR}" ) diff --git a/crates/switchyard-components/src/backends/anthropic.rs b/crates/switchyard-components/src/backends/anthropic.rs index 8fb611373..9bb6fa151 100644 --- a/crates/switchyard-components/src/backends/anthropic.rs +++ b/crates/switchyard-components/src/backends/anthropic.rs @@ -80,24 +80,36 @@ impl AnthropicNativeBackend { } fn outbound_body(&self, request: &ChatRequest) -> Result { - let mut body = match request.request_type() { - ChatRequestType::Anthropic => request.body().clone(), - source => { - self.translation - .translate_request( - request_wire_format(source), - WireFormat::AnthropicMessages, - request.body(), - &self.translation_policy, - ) - .map_err(|error| { - SwitchyardError::Backend(format!( - "failed to translate {source:?} request to Anthropic Messages: {error}" - )) - })? - .body - } - }; + // Native Anthropic requests are a TRUE passthrough: forward the client's + // body verbatim, rewriting only the model id (routing) and merging any + // operator-configured `extra_body`. The strips/normalization in the + // translated branch below exist for OpenAI/Responses -> Anthropic bodies, + // which can carry fields/shapes the Anthropic API rejects. Applying them + // to a real Anthropic client (e.g. Claude Code) silently drops valid + // Anthropic features — `context_management` (context auto-compaction), + // signed thinking blocks, mid-conversation system turns, client tool ids + // — which a passthrough must never do. + if matches!(request.request_type(), ChatRequestType::Anthropic) { + let mut body = request.body().clone(); + set_json_model(&mut body, self.target.model.as_str()); + merge_target_extra_body(&mut body, self.target.extra_body.as_ref()); + return Ok(body); + } + let mut body = self + .translation + .translate_request( + request_wire_format(request.request_type()), + WireFormat::AnthropicMessages, + request.body(), + &self.translation_policy, + ) + .map_err(|error| { + SwitchyardError::Backend(format!( + "failed to translate {:?} request to Anthropic Messages: {error}", + request.request_type() + )) + })? + .body; set_json_model(&mut body, self.target.model.as_str()); strip_anthropic_incompatible_fields(&mut body); normalize_anthropic_body(&mut body); diff --git a/crates/switchyard-components/tests/adversarial_native_backends.rs b/crates/switchyard-components/tests/adversarial_native_backends.rs index 3cc581389..f095fe5fe 100644 --- a/crates/switchyard-components/tests/adversarial_native_backends.rs +++ b/crates/switchyard-components/tests/adversarial_native_backends.rs @@ -669,9 +669,9 @@ fn anthropic_backend_is_anthropic_only() -> Result<()> { Ok(()) } -// Non-streaming Anthropic calls should strip incompatible fields and stamp context. +// Non-streaming native Anthropic calls forward the body verbatim and stamp context. #[tokio::test] -async fn anthropic_non_streaming_strips_incompatible_fields_and_records_context() -> Result<()> { +async fn anthropic_native_passthrough_forwards_body_verbatim() -> Result<()> { let server = OneShotServer::json( 200, json!({ @@ -731,16 +731,22 @@ async fn anthropic_non_streaming_strips_incompatible_fields_and_records_context( ); assert_eq!(request.body["model"], "target-claude"); assert_eq!(request.body["messages"][0]["content"], "hello"); - assert!(request.body.get("reasoning_effort").is_none()); - assert!(request.body.get("context_management").is_none()); + // Native Anthropic is a verbatim passthrough: client fields are preserved, + // not stripped (only the model id is rewritten for routing). Stripping these + // belongs to the translated path, not a real /v1/messages request. + assert_eq!(request.body["reasoning_effort"], "high"); + assert_eq!( + request.body["context_management"], + json!({"strategy": "auto"}) + ); assert_eq!(request.body["made_up_beta_field"], json!({"kept": true})); assert_eq!(request.body["extra_body"], json!({"caller": "value"})); Ok(()) } -// Anthropic-native calls should downgrade Opus-4.8-style system turns for legacy targets. +// Native Anthropic passthrough must NOT relocate message-level system/developer turns. #[tokio::test] -async fn anthropic_lifts_message_level_system_roles_before_native_call() -> Result<()> { +async fn anthropic_native_passthrough_does_not_lift_system_messages() -> Result<()> { let server = OneShotServer::json(200, json!({"id": "msg-test", "content": []}))?; let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; let mut ctx = ProxyContext::new(); @@ -767,28 +773,24 @@ async fn anthropic_lifts_message_level_system_roles_before_native_call() -> Resu .await?; let request = server.captured()?; - assert_eq!(request.body["system"], "System rules.\n\nDeveloper rules."); - let messages = request.body["messages"] - .as_array() - .ok_or_else(|| SwitchyardError::Other("messages should be an array".to_string()))?; - let roles = messages - .iter() - .map(|message| { - message - .get("role") - .and_then(Value::as_str) - .unwrap_or("") - }) - .collect::>(); - assert_eq!(roles, vec!["user", "assistant"]); - assert_eq!(messages[0]["content"], "hello"); - assert_eq!(messages[1]["content"], "ready"); + // Verbatim: no top-level system is synthesized, and the system/developer + // turns stay exactly where the client put them. + assert!(request.body.get("system").is_none()); + assert_eq!( + request.body["messages"], + json!([ + {"role": "system", "content": "System rules."}, + {"role": "user", "content": "hello"}, + {"role": "developer", "content": [{"type": "text", "text": "Developer rules."}]}, + {"role": "assistant", "content": "ready"} + ]) + ); Ok(()) } -// Interleaved system turns should preserve encounter order after lifting. +// Interleaved system/developer turns are preserved in place (no lifting) on passthrough. #[tokio::test] -async fn anthropic_lifts_multiple_interleaved_system_messages_in_order() -> Result<()> { +async fn anthropic_native_passthrough_preserves_interleaved_system_messages() -> Result<()> { let server = OneShotServer::json(200, json!({"id": "msg-test", "content": []}))?; let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; let mut ctx = ProxyContext::new(); @@ -813,24 +815,25 @@ async fn anthropic_lifts_multiple_interleaved_system_messages_in_order() -> Resu .await?; let request = server.captured()?; - assert_eq!( - request.body["system"], - "Top-level rules.\n\nFirst lifted system.\n\nSecond lifted system.\n\nDeveloper lifted system." - ); + // Verbatim: top-level system unchanged, all message-level turns preserved in place. + assert_eq!(request.body["system"], "Top-level rules."); assert_eq!( request.body["messages"], json!([ + {"role": "system", "content": "First lifted system."}, {"role": "user", "content": "first user"}, + {"role": "system", "content": "Second lifted system."}, {"role": "assistant", "content": "assistant reply"}, + {"role": "developer", "content": "Developer lifted system."}, {"role": "user", "content": "second user"} ]) ); Ok(()) } -// Existing structured Anthropic system prompts should keep their shape when lifted text is added. +// Structured system prompt and message-level system (incl. non-text blocks) pass through untouched. #[tokio::test] -async fn anthropic_lifts_message_level_system_into_existing_system_blocks() -> Result<()> { +async fn anthropic_native_passthrough_preserves_structured_system_and_messages() -> Result<()> { let server = OneShotServer::json(200, json!({"id": "msg-test", "content": []}))?; let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; let mut ctx = ProxyContext::new(); @@ -858,16 +861,25 @@ async fn anthropic_lifts_message_level_system_into_existing_system_blocks() -> R .await?; let request = server.captured()?; + // Verbatim: structured system kept as-is; the message-level system turn + // (including its image block) is preserved, not downgraded into system. assert_eq!( request.body["system"], - json!([ - {"type": "text", "text": "Existing system."}, - {"type": "text", "text": "Lifted system.\n\nLifted input text."} - ]) + json!([{"type": "text", "text": "Existing system."}]) ); assert_eq!( request.body["messages"], - json!([{"role": "user", "content": "hello"}]) + json!([ + { + "role": "system", + "content": [ + {"type": "text", "text": "Lifted system."}, + {"type": "image", "source": {"type": "url", "url": "https://example.test/a.png"}}, + {"type": "input_text", "text": "Lifted input text."} + ] + }, + {"role": "user", "content": "hello"} + ]) ); Ok(()) } @@ -898,9 +910,9 @@ async fn anthropic_translates_responses_requests_with_default_max_tokens() -> Re Ok(()) } -// Invalid Anthropic tool-use IDs should be sanitized consistently with results. +// Native Anthropic passthrough preserves client tool-use IDs verbatim (no sanitization). #[tokio::test] -async fn anthropic_sanitizes_invalid_tool_use_ids_and_matching_results() -> Result<()> { +async fn anthropic_native_passthrough_preserves_tool_use_ids() -> Result<()> { let server = OneShotServer::json(200, json!({"id": "msg-test", "content": []}))?; let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; let mut ctx = ProxyContext::new(); @@ -936,8 +948,10 @@ async fn anthropic_sanitizes_invalid_tool_use_ids_and_matching_results() -> Resu .await?; let request = server.captured()?; + // Verbatim: the client's tool-use id is forwarded unchanged (sanitization + // belongs to the translated path; a real Anthropic client sends valid ids). let tool_use_id = &request.body["messages"][1]["content"][0]["id"]; - assert_eq!(tool_use_id, "toolu_01_bad_id"); + assert_eq!(tool_use_id, "toolu_01*bad:id"); assert_eq!( &request.body["messages"][2]["content"][0]["tool_use_id"], tool_use_id @@ -945,9 +959,9 @@ async fn anthropic_sanitizes_invalid_tool_use_ids_and_matching_results() -> Resu Ok(()) } -// Unsigned synthetic thinking blocks should be removed before Anthropic replay. +// Native Anthropic passthrough preserves thinking blocks verbatim (no stripping). #[tokio::test] -async fn anthropic_strips_unsigned_thinking_blocks_before_native_call() -> Result<()> { +async fn anthropic_native_passthrough_preserves_thinking_blocks() -> Result<()> { let server = OneShotServer::json(200, json!({"id": "msg-test", "content": []}))?; let backend = AnthropicNativeBackend::new(anthropic_target(server.base_url().to_string())?)?; let mut ctx = ProxyContext::new(); @@ -985,15 +999,22 @@ async fn anthropic_strips_unsigned_thinking_blocks_before_native_call() -> Resul .await?; let request = server.captured()?; + // Verbatim: thinking blocks (signed or not) are preserved; the upstream API + // decides what to accept. Stripping unsigned blocks belongs to the translated + // path, where they are synthetic translation artifacts. assert_eq!( request.body["messages"][0]["content"] .as_array() .ok_or_else(|| SwitchyardError::Other("content should be an array".to_string()))? .len(), - 1 + 2 ); assert_eq!( request.body["messages"][0]["content"][0]["type"], + "thinking" + ); + assert_eq!( + request.body["messages"][0]["content"][1]["type"], "tool_use" ); assert_eq!( @@ -1004,7 +1025,10 @@ async fn anthropic_strips_unsigned_thinking_blocks_before_native_call() -> Resul request.body["messages"][1]["content"][0]["thinking"], "real" ); - assert_eq!(request.body["messages"][2]["content"], ""); + assert_eq!( + request.body["messages"][2]["content"], + json!([{"type": "thinking", "thinking": "only synthetic"}]) + ); Ok(()) } diff --git a/examples/route.yaml b/examples/route.yaml index 5b3ef858e..b9d5713a0 100644 --- a/examples/route.yaml +++ b/examples/route.yaml @@ -10,3 +10,13 @@ routes: smoke-test: type: noop + + # Executor gated by a stronger advisor model: the advisor reviews the + # executor's first no-tool-call turn (APPROVE lets it stop; REDO sends it + # back with a plan). Tiers accept the same fields as passthrough targets. + # advisor-gate: + # type: advisor + # executor: + # model: moonshotai/kimi-k2.6 + # advisor: + # model: anthropic/claude-opus-4.8 diff --git a/switchyard/__init__.py b/switchyard/__init__.py index ae1574d2f..8f7971e20 100644 --- a/switchyard/__init__.py +++ b/switchyard/__init__.py @@ -13,6 +13,9 @@ from typing import TYPE_CHECKING, Any from switchyard.lib.backends import ( + AdvisorConfig, + AdvisorLoopBackend, + AdvisorPresets, AnthropicNativeBackend, OpenAiNativeBackend, ) @@ -121,7 +124,10 @@ def __getattr__(name: str) -> Any: "BackendFormat", "RandomRoutingProcessorConfig", "LlmTarget", - # Deterministic (LLM-classifier) routing usage case + # Advisor review gate (executor gated by a stronger advisor model) + "AdvisorConfig", + "AdvisorLoopBackend", + "AdvisorPresets", # Translation engine "TranslationEngine", # ChatResponse types diff --git a/switchyard/cli/launchers/launcher_runtime.py b/switchyard/cli/launchers/launcher_runtime.py index 4584d1235..a52311978 100644 --- a/switchyard/cli/launchers/launcher_runtime.py +++ b/switchyard/cli/launchers/launcher_runtime.py @@ -132,6 +132,17 @@ def route_bundle_strategy_summary(route_bundle: str, default_model: str) -> str: target = route.get("target") model = target.get("model") if isinstance(target, _Mapping) else target return f"passthrough: model={model or first_key}" + if route_type == "advisor": + tiers = {} + for field in ("executor", "advisor"): + tier = route.get(field) + tiers[field] = ( + tier.get("model") if isinstance(tier, _Mapping) else tier + ) + return ( + f"advisor: executor={tiers['executor']}, " + f"advisor={tiers['advisor']}" + ) except Exception: pass return f"route: {default_model}" diff --git a/switchyard/cli/route_bundle.py b/switchyard/cli/route_bundle.py index 61f210fa8..fb2b4c210 100644 --- a/switchyard/cli/route_bundle.py +++ b/switchyard/cli/route_bundle.py @@ -13,9 +13,14 @@ from pathlib import Path from typing import Any, Protocol, cast +from pydantic import ValidationError + +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.advisor_loop_backend import AdvisorLoopBackend from switchyard.lib.backends.llm_target import LlmTarget, coerce_llm_target from switchyard.lib.backends.multi_llm_backend import build_native_backend from switchyard.lib.backends.stats_llm_backend import StatsLlmBackend +from switchyard.lib.processors.reasoning_effort_normalizer import ReasoningEffortNormalizer from switchyard.lib.processors.stats_request_processor import StatsRequestProcessor from switchyard.lib.processors.stats_response_processor_accumulator import ( StatsResponseProcessor, @@ -31,9 +36,32 @@ _ENV_REF_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") _TOP_LEVEL_KEYS = frozenset({"defaults", "routes"}) _ROUTE_METADATA_KEYS = frozenset({"display_name", "description"}) +#: ``type: advisor`` route keys: the two tiers plus the scalar +#: :class:`AdvisorConfig` fields tunable from YAML. +_ADVISOR_ROUTE_KEYS = frozenset({ + "type", + "executor", + "advisor", + "reviewer_system_prompt", + "redo_feedback_prefix", + "gate_trigger", + "gate_trigger_pattern", + "max_reviews", + "gate_stall_turns", + "gate_min_tool_results", + "advisor_system_prompt", + "seed_plan_advice", + "seed_advice_prefix", + "advisor_max_tokens", + "advisor_temperature", + "transcript_max_chars", + "fail_open", + "enable_stats", +}) | _ROUTE_METADATA_KEYS _ROUTE_KEYS = { "noop": frozenset({"type"}) | _ROUTE_METADATA_KEYS, "passthrough": frozenset({"type", "target"}) | _ROUTE_METADATA_KEYS, + "advisor": _ADVISOR_ROUTE_KEYS, } @@ -116,7 +144,7 @@ def build_route_bundle_table( pre_routing_request_processors: Sequence[Any] = (), extra_response_processors: Sequence[Any] = (), ) -> RouteTable: - """Build a table containing only noop and passthrough routes.""" + """Build a table of noop, passthrough, and advisor routes.""" bundle = _mapping(_expand_env(raw), "route bundle") _reject_unknown_keys(bundle, _TOP_LEVEL_KEYS, "route bundle") defaults = _mapping(bundle.get("defaults", {}), "defaults") @@ -136,7 +164,7 @@ def build_route_bundle_table( if route_type not in _ROUTE_KEYS: raise RouteBundleConfigError( f"route {route_id!r}: unsupported route type {route_type!r}; " - "expected 'noop' or 'passthrough'" + "expected 'noop', 'passthrough', or 'advisor'" ) _reject_unknown_keys(route, _ROUTE_KEYS[route_type], f"route {route_id!r}") @@ -147,6 +175,17 @@ def build_route_bundle_table( pre_routing_request_processors, extra_response_processors, ) + elif route_type == "advisor": + # The advisor backend is Python-only, so it cannot be wrapped in + # StatsLlmBackend (which requires a Rust-native binding); it records + # into the accumulator itself. The normalizer maps Claude Code's + # ``/effort xhigh`` to a value the executor upstream accepts. + runtime = _build_runtime( + _advisor_backend(route_id, route, defaults, stats), + stats, + [*pre_routing_request_processors, ReasoningEffortNormalizer()], + extra_response_processors, + ) else: target = _target(route_id, route.get("target"), defaults) runtime = _build_runtime( @@ -190,6 +229,51 @@ def _target(route_id: str, value: object, defaults: Mapping[str, object]) -> Llm raise RouteBundleConfigError(f"route {route_id!r}: invalid target: {error}") from error +def _advisor_backend( + route_id: str, + route: Mapping[str, object], + defaults: Mapping[str, object], + stats: StatsAccumulator, +) -> AdvisorLoopBackend: + """Build the review-gate advisor backend from a ``type: advisor`` route. + + The ``executor`` / ``advisor`` tiers are ordinary targets (bundle + ``defaults`` apply to each); the remaining route keys are scalar + :class:`AdvisorConfig` fields. Each tier's ``format`` selects its wire + independently — ``anthropic`` (native ``/v1/messages``; the executor + passthrough preserves prompt caching) or ``openai`` (``/chat/completions``; + Qwen/DeepSeek/vLLM/NIM/OpenAI endpoints). ``responses`` targets are + rejected at validation. + """ + config_data: dict[str, object] = { + key: value + for key, value in route.items() + if key != "type" and key not in _ROUTE_METADATA_KEYS + } + for field in ("executor", "advisor"): + config_data[field] = _advisor_tier(route_id, field, route.get(field), defaults) + try: + config = AdvisorConfig.model_validate(config_data) + except ValidationError as error: + raise RouteBundleConfigError(f"route {route_id!r}: {error}") from error + return AdvisorLoopBackend(config, stats_accumulator=stats) + + +def _advisor_tier( + route_id: str, field: str, value: object, defaults: Mapping[str, object] +) -> LlmTarget: + if value is None: + raise RouteBundleConfigError(f"route {route_id!r}: {field} target is required") + if isinstance(value, str): + tier: dict[str, object] = {"model": value} + else: + tier = _mapping(value, f"route {route_id!r} {field}") + try: + return coerce_llm_target({**defaults, **tier}, default_id=field) + except (TypeError, ValueError) as error: + raise RouteBundleConfigError(f"route {route_id!r}: invalid {field}: {error}") from error + + def _expand_env(value: object) -> object: if isinstance(value, dict): return {key: _expand_env(item) for key, item in value.items()} diff --git a/switchyard/cli/switchyard_cli.py b/switchyard/cli/switchyard_cli.py index 3698e4aa9..2af3de6eb 100644 --- a/switchyard/cli/switchyard_cli.py +++ b/switchyard/cli/switchyard_cli.py @@ -36,9 +36,14 @@ def _cmd_serve(args: argparse.Namespace) -> None: if args.routing_log_file: from switchyard.lib.processors.routing_log_response_processor import ( RoutingLogResponseProcessor, + register_routing_log_sink, ) - response_processors.append(RoutingLogResponseProcessor(args.routing_log_file)) + routing_log = RoutingLogResponseProcessor(args.routing_log_file) + response_processors.append(routing_log) + # Multi-call backends (the advisor) emit their proxy-internal usage + # (consults, discarded turns) into the same log via this sink. + register_routing_log_sink(routing_log) table = load_route_bundle_table( args.routes, diff --git a/switchyard/lib/backends/__init__.py b/switchyard/lib/backends/__init__.py index 53c20ae6f..dc6bd9064 100644 --- a/switchyard/lib/backends/__init__.py +++ b/switchyard/lib/backends/__init__.py @@ -7,6 +7,9 @@ ``from switchyard.lib.backends import OpenAiNativeBackend``. """ +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.advisor_loop_backend import AdvisorLoopBackend +from switchyard.lib.backends.advisor_presets import AdvisorPresets from switchyard.lib.backends.backend_format_resolver import ( BackendFormatResolution, BackendFormatResolver, @@ -21,6 +24,9 @@ ) __all__ = [ + "AdvisorConfig", + "AdvisorLoopBackend", + "AdvisorPresets", "AnthropicNativeBackend", "BackendFormatResolution", "BackendFormatResolver", diff --git a/switchyard/lib/backends/advisor_config.py b/switchyard/lib/backends/advisor_config.py new file mode 100644 index 000000000..693681dab --- /dev/null +++ b/switchyard/lib/backends/advisor_config.py @@ -0,0 +1,185 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Config model for the advisor review gate. + +An advisor chain pairs an **executor** (the base model under test) with a +stronger **advisor**. No advisor tool is injected: the executor works the task +with its own tools; when it first produces a no-tool-call turn — a plan, or a +claim of "done" — the backend consults the advisor once to APPROVE or send it +back (REDO) with an optimized plan. See +``switchyard/lib/backends/advisor_loop_backend.py``. + +Both tiers are ordinary targets; each tier's ``format`` selects its wire +independently and tiers mix freely. ``anthropic`` targets are served native +Anthropic-Messages with the body passed through verbatim (the client's prompt +caching survives); ``openai`` targets (Qwen, DeepSeek, vLLM/NIM, OpenAI) are +served OpenAI Chat Completions, likewise verbatim. ``responses`` targets are +rejected (the advisor loop is Chat-shaped). +""" + +from __future__ import annotations + +import re +from typing import Literal, Self + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + ValidationInfo, + field_validator, + model_validator, +) + +from switchyard.lib.backends.advisor_prompts import ( + ADVISOR_SYSTEM_PROMPT, + REDO_FEEDBACK_PREFIX, + REVIEWER_SYSTEM_PROMPT, + SEED_ADVICE_PREFIX, +) +from switchyard.lib.backends.llm_target import BackendFormat, LlmTarget, coerce_llm_target + + +class AdvisorConfig(BaseModel): + """Configuration for the advisor review gate. + + Attributes: + executor: The base model under test. Runs the user-visible chat + completion with the client's own tools. + advisor: The stronger advisor model. Must be at least as capable. + + reviewer_system_prompt: System prompt for the advisor's review call; + instructs the APPROVE / REDO contract. + redo_feedback_prefix: Prepended to the advisor's REDO plan when it is + injected back to the executor as a user turn. Tune per executor + family (e.g. append "continue using tool calls only" for small + OSS executors). + gate_trigger: What fires the once-per-session review. + ``"no_tool_call"`` (default) reviews the executor's first turn + without tool calls — right for function-calling agent harnesses. + ``"pattern"`` reviews the first turn whose text matches + ``gate_trigger_pattern`` — for text-protocol harnesses (e.g. + Terminal-Bench's terminus), where every turn lacks tool calls and + completion is declared with a textual marker instead. + gate_trigger_pattern: Regex searched against the executor turn's text + when ``gate_trigger`` is ``"pattern"`` + (e.g. ``task_complete["\\s>:]*true`` for terminus). + max_reviews: Budget of advisor reviews per budget scope: the caller's + ``proxy_x_session_id`` header when present (one evaluation/task + on benchmark harnesses, sub-agents included), else one scope for + the whole backend instance. The default (1) preserves the + original once-per-task gate; higher values re-review later + trigger turns (e.g. a re-declared completion after a REDO), + making the gate a sequential best-of-(N+1) with the advisor as + judge. Failed advisor consults do not consume the budget. + gate_stall_turns: When > 0, additionally trigger a review (once per + session, consuming review budget) at the first request whose + conversation already carries at least this many assistant turns — + a mid-task checkpoint for executors that grind without ever + declaring completion. 0 disables. + gate_min_tool_results: For the ``no_tool_call`` trigger: only review + a no-tool-call turn when the conversation carries at least this + many tool results — skips reviewing early commentary turns on + chatty harnesses. 0 reviews as before. + + seed_plan_advice: Consult the advisor once at the start of each + session — before the executor's first turn — and inject its + upfront plan into the session's first user message + (``seed_advice_prefix`` + advice). The advice is cached per + session (keyed by the conversation's stable prefix) and + re-injected identically on every turn, so it stays visible for + the whole session while the upstream cache prefix stays stable. + Proxy-triggered, so it fires even on executors/harnesses that + never call tools. The seed consult uses ``advisor_system_prompt``. + Fail-open: a failed seed consult leaves the session unseeded. + seed_advice_prefix: Prepended to the seeded advice when it is + injected into the first user message. + advisor_system_prompt: System prompt for the seed consult's advisor + call; tells the advisor to plan, not act. + + advisor_max_tokens: Cap on the advisor's output per call. + advisor_temperature: Sampling temperature for the advisor call. ``None`` + (default) omits the field — required for Anthropic targets that + reject ``temperature``. + transcript_max_chars: Cap on the serialized transcript handed to the + advisor, so a long agent conversation can't blow its context. + The default (200k chars ≈ 50k tokens) fits comfortably in a + frontier advisor's window; the middle of an over-cap conversation + is dropped (task head + recent tail survive). + fail_open: When ``True`` (default), an advisor-call failure degrades + gracefully — the turn passes through as APPROVE. When ``False``, + the failure surfaces as 5xx. + enable_stats: Record executor success/error + latency into the shared + accumulator and stamp ``ctx.selected_model``. + preset: Optional name of the preset that produced this config. + """ + + model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True) + + executor: LlmTarget + advisor: LlmTarget + + # review gate + reviewer_system_prompt: str = REVIEWER_SYSTEM_PROMPT + redo_feedback_prefix: str = REDO_FEEDBACK_PREFIX + gate_trigger: Literal["no_tool_call", "pattern"] = "no_tool_call" + gate_trigger_pattern: str = "" + max_reviews: int = Field(default=1, ge=1) + gate_stall_turns: int = Field(default=0, ge=0) + gate_min_tool_results: int = Field(default=0, ge=0) + + # seed advice + seed_plan_advice: bool = False + seed_advice_prefix: str = SEED_ADVICE_PREFIX + advisor_system_prompt: str = ADVISOR_SYSTEM_PROMPT + + # shared + advisor_max_tokens: int = Field(default=2048, ge=1) + advisor_temperature: float | None = None + transcript_max_chars: int = Field(default=200_000, ge=256) + fail_open: bool = True + enable_stats: bool = True + preset: str | None = None + + @field_validator("executor", "advisor", mode="before") + @classmethod + def _coerce_target(cls, value: object, info: ValidationInfo) -> LlmTarget: + return coerce_llm_target(value, default_id=info.field_name or "target") + + @field_validator("executor", "advisor") + @classmethod + def _target_model_non_empty(cls, tier: LlmTarget) -> LlmTarget: + if not tier.model: + raise ValueError("target.model must be a non-empty string") + return tier + + @field_validator("executor", "advisor") + @classmethod + def _target_format_supported(cls, tier: LlmTarget, info: ValidationInfo) -> LlmTarget: + if tier.format == BackendFormat.RESPONSES: + raise ValueError( + f"{info.field_name}.format 'responses' is not supported by the advisor " + "backend (the loop is Chat-shaped); use 'openai' or 'anthropic'" + ) + return tier + + @field_validator("gate_trigger_pattern") + @classmethod + def _pattern_compiles(cls, value: str) -> str: + if value: + try: + re.compile(value) + except re.error as exc: + raise ValueError(f"gate_trigger_pattern is not a valid regex: {exc}") from exc + return value + + @model_validator(mode="after") + def _pattern_trigger_requires_pattern(self) -> Self: + if self.gate_trigger == "pattern" and not self.gate_trigger_pattern: + raise ValueError( + "gate_trigger 'pattern' requires a non-empty gate_trigger_pattern" + ) + return self + +__all__ = ["AdvisorConfig"] diff --git a/switchyard/lib/backends/advisor_loop_backend.py b/switchyard/lib/backends/advisor_loop_backend.py new file mode 100644 index 000000000..910e0ea4a --- /dev/null +++ b/switchyard/lib/backends/advisor_loop_backend.py @@ -0,0 +1,1302 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""``LLMBackend`` that gates the executor with a once-per-session advisor review. + +An earlier design offered the executor an ``advisor`` tool it could call +mid-generation. Trace analysis showed that front-loading the advisor's plan +*suppressed the executor's own test-and-iterate loop* — it trusted the plan, +one-shot it, and declared "done" prematurely (e.g. solving a concurrency task in +4 turns vs the 17 the unadvised baseline needed to catch the bug). Net effect +was within noise, with real losses on tasks the baseline solved by iterating. + +This backend instead uses the advisor as a **once-per-session review gate**: + +1. The executor works the task with its **own** tools (no advisor tool injected, + no upfront advice) — its iteration loop is untouched. +2. The first time the executor produces a turn with **no tool calls** — either a + plan it is about to execute, or a claim that the task is complete — the + backend consults the advisor **once** to review the full transcript: + - ``APPROVE`` → the executor's turn is returned unchanged (sound plan / done). + - ``REDO`` → the advisor's optimized plan is fed back as a user turn and the + executor is re-invoked to **keep working** (it produces tool calls again). +3. Subsequent turns in the same session pass through unreviewed + (once-per-session), so the gate can force at most one extra round of work. + +This is a near-superset of solo behavior — identical to the bare executor until +"done", plus one quality gate — so it is downside-protected (≈ baseline if the +advisor always approves) while catching premature convergence. + +The executor's wire is selected by ``config.executor.format``: ``anthropic`` +executors are delegated verbatim to an :class:`AnthropicNativeBackend` +(``/v1/messages`` — the client's ``cache_control`` breakpoints reach the +upstream unchanged, so prompt caching is honored); ``openai`` executors +(Qwen, DeepSeek, vLLM/NIM, OpenAI) are delegated verbatim to an +:class:`OpenAiNativeBackend` (``/chat/completions``). Tool use is read from +the wire's native shape (Anthropic ``stop_reason``/``tool_use`` blocks, or +OpenAI ``tool_calls``/``finish_reason``); the REDO feedback is plain-string +assistant/user turns, valid on both wires, with a config-tunable prefix +(``redo_feedback_prefix``). The advisor tier is likewise format-dispatched +(``_build_advisor_caller``): Anthropic Messages or OpenAI Chat Completions. +Because the gate's trigger is proxy-side (the first no-tool-call turn), it +fires regardless of the executor's own tool-use discipline. + +Chain integration:: + + [RequestProcessor*] → AdvisorLoopBackend → [ResponseProcessor*] → TranslationEngine + +Declares ``supported_request_types`` for the executor's wire so the +TranslationEngine normalizes any inbound format to it once. The outer chain's +``StatsResponseProcessor`` records executor token usage (including cache reads) +from the returned response; this backend additionally records the advisor +review's usage into the classifier bucket and stamps ``ctx.selected_model``. + +Streaming is single-pass: until a session is reviewed, each executor turn is +streamed and buffered while detecting whether it has tool calls; a passed-through +/ approved turn's buffered events are replayed verbatim, so the turn is generated +once. After the review fires, the session is pure passthrough (the upstream +stream is returned directly — true streaming, full caching, zero overhead). +The review budget (``max_reviews``) is keyed by the caller-declared session +identity: the ``proxy_x_session_id`` header (parsed into +``RequestMetadata.session_id`` by the endpoints) when present, else a single +instance-wide scope. Benchmark harnesses stamp that header with the evaluation +id on every request — including sub-agent conversations — so on a gateway +shared by many tasks each task gets its own budget, exactly the "reviews for +*this* task" semantics ``max_reviews`` is meant to have. The content hash of +the conversation prefix (``_session_key``) is NOT used for budgeting — it is +not reliably stable (harnesses compact history, spawn sub-conversations, and +re-render system context) — and only keys the seed-advice cache and the stall +checkpoint. Failed advisor consults do not consume budget; after +``_MAX_FAILED_CONSULTS_PER_SCOPE`` failures a scope stops consulting entirely, +bounding latency against a down advisor. A pod restart mid-run resets the +budget (rare, harmless). +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import re +import sys +import time +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Protocol + +import httpx + +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.llm_target import BackendFormat +from switchyard.lib.backends.multi_llm_backend import ( + build_native_backend, + resolve_llm_target, +) +from switchyard.lib.chat_response.anthropic import AnthropicResponseStream +from switchyard.lib.chat_response.openai_chat import ResponseStream +from switchyard.lib.request_metadata import CTX_REQUEST_METADATA +from switchyard.lib.roles import LLMBackend +from switchyard_rust.core import ( + ChatRequestType, + ChatResponse, + ChatResponseType, + request_type_enum, + request_type_matches, + request_with_type, +) +from switchyard_rust.translation import TranslationEngine + +if TYPE_CHECKING: + from switchyard.lib.backends.llm_target import LlmTarget + from switchyard.lib.proxy_context import ProxyContext + from switchyard.lib.stats_accumulator import StatsAccumulator + from switchyard_rust.core import ChatRequest + +log = logging.getLogger(__name__) + +_ANTHROPIC_VERSION = "2023-06-01" +#: Distinct session keys on one backend instance past which the hashed +#: conversation prefix is assumed unstable (see ``_session_key``). A single +#: agent run legitimately opens a handful of conversations (the main thread plus +#: any sub-agents); dozens means the key is churning per turn. +_SESSION_CHURN_WARN_AT = 12 +#: Budget scope for callers that send no ``proxy_x_session_id`` header. +_INSTANCE_SCOPE = "__instance__" +#: Failed (fail-open) consults tolerated per budget 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. +_MAX_FAILED_CONSULTS_PER_SCOPE = 3 + + +class AdvisorCaller(Protocol): + """Consults the advisor model and returns ``(text, usage)``.""" + + async def advise(self, *, system: str, transcript: str) -> tuple[str, Any]: + ... + + +@dataclass +class _ExecTurn: + """One executor turn, normalized across the buffered streaming / completion paths. + + Token counts let a REDO-discarded turn (which the client never sees, so the + outer stats processor never prices it) be recorded explicitly. + """ + + has_tool_use: bool + content: str | None + latency_ms: float + completion_body: Any | None = None + stream_events: list[Any] | None = None + input_tokens: int = 0 + output_tokens: int = 0 + cached_tokens: int = 0 + #: Reasoning-model internal text (OpenAI ``reasoning_content``); the + #: fallback evidence when a gated turn has no visible content at all. + reasoning_text: str | None = None + + +class AdvisorLoopBackend(LLMBackend): + """Executor backend gated by a once-per-session advisor review (native Anthropic).""" + + def __init__( + self, + config: AdvisorConfig, + *, + stats_accumulator: StatsAccumulator | None = None, + executor_backend: LLMBackend | None = None, + advisor_caller: AdvisorCaller | None = None, + ) -> None: + self._config = config + self._stats = stats_accumulator if config.enable_stats else None + self._translation = TranslationEngine() + # Sessions whose stall checkpoint (gate_stall_turns) already fired. + self._stall_fired: set[str] = set() + # Per-session seed advice cache for seed_plan_advice ("" = unseeded). + self._seed_advice: dict[str, str] = {} + # Session-key churn observability: the hashed conversation prefix keys + # the seed cache and stall checkpoint, so instability there degrades + # those features (it no longer affects the review budget, which keys on + # the caller's session header). Tracked over *every* request so churn + # stays observable even when the gate never fires. + self._sessions_seen: set[str] = set() + self._session_churn_warned = False + # Review budget, keyed by ``_budget_scope``: the caller's + # ``proxy_x_session_id`` header when present (one evaluation/task on + # benchmark harnesses, shared by its sub-agents), else one + # instance-wide scope. On a gateway shared by many tasks this gives + # each task its own ``max_reviews`` budget. Failed consults refund the + # budget and count separately (``_MAX_FAILED_CONSULTS_PER_SCOPE``). + self._reviews_by_scope: dict[str, int] = {} + self._failed_consults_by_scope: dict[str, int] = {} + self._budget_logged_scopes: set[str] = set() + # Resolve format: auto before wire selection; injected fakes must pin + # a concrete format (probing a fake's endpoint makes no sense). + executor_target = ( + config.executor if executor_backend is not None + else resolve_llm_target(config.executor) + ) + self._request_type_name = _gate_request_type(executor_target.format) + self._request_type = request_type_enum(self._request_type_name) + self._is_openai = self._request_type_name == "openai_chat" + # Pre-compiled pattern trigger; None selects the no_tool_call trigger. + self._trigger_pattern = ( + re.compile(config.gate_trigger_pattern) + if config.gate_trigger == "pattern" + else None + ) + # The executor is delegated to verbatim so caching survives + # (cache_control breakpoints on Anthropic; prefix stability on OpenAI). + self._executor_backend = executor_backend or build_native_backend(executor_target) + self._advisor_caller = advisor_caller or _build_advisor_caller(config) + + async def startup(self) -> None: + await self._executor_backend.startup() + + async def shutdown(self) -> None: + await self._executor_backend.shutdown() + + @property + def supported_request_types(self) -> list[ChatRequestType]: + """The executor's native wire; inbound formats are normalized to it.""" + return [self._request_type] + + async def call(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse: + normalized = self._translation.request_to_any_of( + request, self.supported_request_types, + ) + if not request_type_matches(normalized, self._request_type): + raise TypeError( + "AdvisorLoopBackend expected a " + f"{self._request_type_name} request after translation" + ) + + body = dict(normalized.body) + messages: list[dict[str, Any]] = list(body.get("messages") or []) + session = _session_key(body.get("system"), messages) + if session not in self._sessions_seen: + self._sessions_seen.add(session) + log.info( + "AdvisorLoopBackend: new session key (%d distinct seen)", + len(self._sessions_seen), + ) + if ( + not self._session_churn_warned + and len(self._sessions_seen) >= _SESSION_CHURN_WARN_AT + ): + self._session_churn_warned = True + log.warning( + "AdvisorLoopBackend: %d distinct session keys seen on one backend " + "instance; the hashed conversation prefix is unstable for this " + "client, so seed-advice caching and stall checkpoints may misfire " + "(the review budget is unaffected — it keys on the caller's " + "session header)", + len(self._sessions_seen), + ) + + # Seed the session with upfront advisor advice (consulted once at the + # session-opening request, cached, and re-injected identically on every + # later turn so the upstream cache prefix stays stable). + if self._config.seed_plan_advice: + advice = await _seed_advice_for( + self._seed_advice, session, messages, + caller=self._advisor_caller, config=self._config, stats=self._stats, + ctx=ctx, + ) + if advice: + messages = _with_length_line( + messages, self._config.seed_advice_prefix + advice, + ) + body = {**body, "messages": messages} + normalized = request_with_type(self._request_type_name, body) + + # Once the review budget is spent, every turn is pure passthrough — + # return the upstream stream directly (true streaming, caching intact, + # no buffering). + # + # The budget keys on the caller's declared session identity + # (``proxy_x_session_id`` header), NOT on the conversation content hash: + # content hashes are unstable (harnesses compact history, spawn + # sub-conversations, re-render system context — measured on + # Terminal-Bench, one task minted up to 194 keys and drew 107 reviews + # against a configured ``max_reviews`` of 2), and an instance-wide cap + # breaks the other way on a gateway shared by many tasks, where it + # would bound reviews for the whole run instead of per task. The header + # is stamped per evaluation by benchmark harnesses (sub-agents + # included), so it expresses exactly "reviews for *this* task"; callers + # that send no header fall back to one instance-wide scope. + scope = self._budget_scope(ctx) + if self._scope_exhausted(scope): + if scope not in self._budget_logged_scopes: + self._budget_logged_scopes.add(scope) + log.info( + "AdvisorLoopBackend: review budget spent for scope %s " + "(max_reviews=%d, reviews=%d, failed consults=%d); " + "remaining turns pass through", + scope, + self._config.max_reviews, + self._reviews_by_scope.get(scope, 0), + self._failed_consults_by_scope.get(scope, 0), + ) + return await self._passthrough(ctx, normalized) + + # Budget remains: run the executor and inspect its turn. + turn = await self._run_executor(ctx, normalized) + + # Stall checkpoint (once per session): the conversation has grown past + # ``gate_stall_turns`` assistant turns without the main trigger having + # fired — review mid-task regardless of the turn's shape. + stall = ( + self._config.gate_stall_turns > 0 + and session not in self._stall_fired + and sum(1 for m in messages if m.get("role") == "assistant") + >= self._config.gate_stall_turns + ) + + # Trigger check. "no_tool_call" gates the first turn without tool + # calls (function-calling harnesses; ``gate_min_tool_results`` skips + # early commentary turns before any real work exists); "pattern" + # gates the first turn whose text matches the configured marker + # (text-protocol harnesses, e.g. terminus's ``task_complete: true`` + # declaration). + if self._trigger_pattern is not None: + triggered = bool(self._trigger_pattern.search(turn.content or "")) + else: + triggered = not turn.has_tool_use and ( + _count_tool_results(messages) >= self._config.gate_min_tool_results + ) + if not (triggered or stall): + return await self._finish(ctx, turn) + + # Trigger fired: a plan, a "done", the marker, or a stall checkpoint. + # Gate it. Budget is reserved before the consult (so concurrent + # requests in one scope cannot overdraw across the await) and refunded + # if the consult itself failed — a fail-open error is not a review. + if stall and not triggered: + self._stall_fired.add(session) + # A reasoning-only turn (no visible text) still triggers; hand the + # advisor the reasoning as labeled evidence instead of "(no text)". + review_tail = turn.content + if review_tail is None and turn.reasoning_text: + review_tail = ( + "(the executor produced no visible text this turn; its " + "internal reasoning follows)\n" + turn.reasoning_text + ) + self._reviews_by_scope[scope] = self._reviews_by_scope.get(scope, 0) + 1 + verdict, plan, consulted = await self._review(messages, review_tail, ctx) + if not consulted: + self._reviews_by_scope[scope] -= 1 + self._failed_consults_by_scope[scope] = ( + self._failed_consults_by_scope.get(scope, 0) + 1 + ) + if verdict != "REDO": + return await self._finish(ctx, turn) + + # REDO: feed the optimized plan back and re-invoke so the executor keeps + # working instead of stopping. The session is now reviewed, so the redo + # turn (and everything after it) is plain passthrough. + # Plain-string assistant/user turns are valid on both wires, so the + # feedback shape needs no dialect. The prefix is config-tunable + # (``redo_feedback_prefix``) for per-executor-family steering. + # The gated turn is discarded (the client never sees it) — record its + # usage into the classifier bucket and the routing log so the run's + # cost output prices this proxy-internal turn. + await self._record_discarded_turn(ctx, turn) + # The assistant echo prefers visible text, then the model's own + # reasoning (its generated tokens, upstream-only), then "" — an empty + # echo both risks strict-endpoint rejection and gives the executor a + # void to continue from. + redo_messages = [ + *messages, + {"role": "assistant", "content": turn.content or turn.reasoning_text or ""}, + {"role": "user", "content": self._config.redo_feedback_prefix + plan}, + ] + redo_body = {**body, "messages": redo_messages} + redo_request = request_with_type(self._request_type_name, redo_body) + return await self._passthrough(ctx, redo_request) + + def _budget_scope(self, ctx: ProxyContext) -> str: + """Review-budget key: the caller's session header, else instance-wide.""" + metadata = ctx.metadata.get(CTX_REQUEST_METADATA) + session_id = getattr(metadata, "session_id", None) + return f"client:{session_id}" if session_id else _INSTANCE_SCOPE + + def _scope_exhausted(self, scope: str) -> bool: + """True when a scope has no review budget or too many failed consults.""" + return ( + self._reviews_by_scope.get(scope, 0) >= self._config.max_reviews + or self._failed_consults_by_scope.get(scope, 0) + >= _MAX_FAILED_CONSULTS_PER_SCOPE + ) + + # ------------------------------------------------------------------ + # Executor turn + # ------------------------------------------------------------------ + + async def _run_executor(self, ctx: ProxyContext, request: ChatRequest) -> _ExecTurn: + """Call the executor, buffering its response to detect tool use.""" + started = time.monotonic() + try: + response = await self._executor_backend.call(ctx, request) + except Exception: + # Includes ContextWindowExceeded (the chain uses it for evict-and-retry). + if self._stats is not None: + await self._stats.record_error(self._config.executor.model) + raise + + latency_ms = (time.monotonic() - started) * 1000.0 + if response.response_type == ChatResponseType.ANTHROPIC_STREAM: + events, has_tool_use, content, usage = await _consume_anthropic_stream( + response.stream + ) + return _ExecTurn( + has_tool_use=has_tool_use, + content=content, + latency_ms=latency_ms, + stream_events=events, + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + cached_tokens=usage["cached_tokens"], + ) + if response.response_type == ChatResponseType.OPENAI_STREAM: + events, message, usage = await _consume_openai_stream(response.stream) + return _ExecTurn( + has_tool_use=bool(message.get("tool_calls")), + content=message.get("content") or None, + latency_ms=latency_ms, + stream_events=events, + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + cached_tokens=usage["cached_tokens"], + reasoning_text=message.get("reasoning_content") or None, + ) + body = response.to_body() + reasoning_text = None + if self._is_openai: + has_tool_use, content = _openai_completion_tool_use(body) + reasoning_text = _openai_completion_reasoning(body) + else: + has_tool_use, content = _completion_tool_use(body) + usage = _completion_usage(body, is_openai=self._is_openai) + return _ExecTurn( + has_tool_use=has_tool_use, + content=content, + latency_ms=latency_ms, + completion_body=body, + input_tokens=usage["input_tokens"], + output_tokens=usage["output_tokens"], + cached_tokens=usage["cached_tokens"], + reasoning_text=reasoning_text, + ) + + async def _passthrough(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse: + """Call the executor and return its response verbatim (no buffering).""" + started = time.monotonic() + try: + response = await self._executor_backend.call(ctx, request) + except Exception: + if self._stats is not None: + await self._stats.record_error(self._config.executor.model) + raise + await self._stamp(ctx, (time.monotonic() - started) * 1000.0) + return response + + async def _finish(self, ctx: ProxyContext, turn: _ExecTurn) -> ChatResponse: + """Record stats, stamp ctx, and rebuild the buffered turn as a response.""" + await self._stamp(ctx, turn.latency_ms) + if turn.stream_events is not None: + if self._is_openai: + return ChatResponse.openai_stream( + ResponseStream(_replay_events(turn.stream_events)) + ) + return ChatResponse.anthropic_stream( + AnthropicResponseStream(_replay_events(turn.stream_events)) + ) + if self._is_openai: + return ChatResponse.openai_completion(turn.completion_body) + return ChatResponse.anthropic_completion(turn.completion_body) + + async def _stamp(self, ctx: ProxyContext, latency_ms: float) -> None: + ctx.selected_model = self._config.executor.model + ctx.backend_call_latency_ms = latency_ms + if self._stats is not None: + await self._stats.record_success(self._config.executor.model, latency_ms) + + async def _record_discarded_turn(self, ctx: ProxyContext, turn: _ExecTurn) -> None: + """Price a REDO-discarded executor turn into the classifier bucket + routing log.""" + if self._stats is not None: + await self._stats.record_classifier_usage( + model=self._config.executor.model, + prompt_tokens=turn.input_tokens, + completion_tokens=turn.output_tokens, + cached_tokens=turn.cached_tokens, + latency_ms=turn.latency_ms, + ) + _emit_routing_usage( + ctx, + model=self._config.executor.model, + tier="review_gate_discarded", + prompt_tokens=turn.input_tokens, + completion_tokens=turn.output_tokens, + cached_tokens=turn.cached_tokens, + ) + + # ------------------------------------------------------------------ + # Advisor review + # ------------------------------------------------------------------ + + async def _review( + self, + messages: list[dict[str, Any]], + terminal_content: str | None, + ctx: ProxyContext, + ) -> tuple[str, str, bool]: + """Consult the advisor once; return ``(verdict, plan, consulted)``. + + ``verdict`` is ``"APPROVE"`` or ``"REDO"``. On a fail-open advisor error + or an unparseable reply, defaults to ``APPROVE`` (do not disrupt a + possibly-correct turn). ``consulted`` is False when the advisor call + itself failed, so the caller can refund the review budget. + """ + transcript = self._serialize_transcript(messages, terminal_content) + started = time.monotonic() + try: + text, usage = await self._advisor_caller.advise( + system=self._config.reviewer_system_prompt, transcript=transcript, + ) + except Exception as exc: + if not self._config.fail_open: + raise + log.warning("AdvisorLoopBackend: review failed; approving (fail-open): %s", exc) + if self._stats is not None: + await self._stats.record_classifier_error(self._config.advisor.model) + _audit_review(verdict="APPROVE", error=str(exc), usage=None, + latency_ms=(time.monotonic() - started) * 1000.0) + return "APPROVE", "", False + latency_ms = (time.monotonic() - started) * 1000.0 + verdict, plan = _parse_verdict(text) + # Record the advisor review's token usage so the run's own cost output + # accounts for the advisor, not just the executor: into the classifier + # bucket (the advisor review is a secondary-model consult, like the + # escalation judge — its cost rolls into ``cost_estimate.total_cost``) + # AND into the routing log, so per-session stats + # (``/v1/routing/session-stats``) attribute it to the caller's session. + # Recorded even for an unparseable reply — the tokens were spent. + tokens = _advisor_usage(usage) + if self._stats is not None: + await self._stats.record_classifier_usage( + model=self._config.advisor.model, + prompt_tokens=tokens["prompt_tokens"], + completion_tokens=tokens["completion_tokens"], + cached_tokens=tokens["cached_tokens"], + latency_ms=latency_ms, + ) + _emit_routing_usage( + ctx, + model=self._config.advisor.model, + tier="advisor_review", + prompt_tokens=tokens["prompt_tokens"], + completion_tokens=tokens["completion_tokens"], + cached_tokens=tokens["cached_tokens"], + cache_creation_tokens=tokens["cache_creation_tokens"], + ) + if verdict == "": + # No leading APPROVE/REDO in the reply: treat as a failed consult + # (caller refunds the budget) and pass the turn through unchanged. + _audit_review( + verdict="UNPARSEABLE", error=None, usage=usage, + latency_ms=latency_ms, reply_head=text, + ) + return "APPROVE", "", False + _audit_review( + verdict=verdict, error=None, usage=usage, latency_ms=latency_ms, + reply_head=text, + ) + return verdict, plan, True + + def _serialize_transcript( + self, messages: list[dict[str, Any]], terminal_content: str | None, + ) -> str: + """Serialize the conversation + the executor's terminal turn for review. + + Over ``transcript_max_chars``, the MIDDLE is dropped: the head keeps + the task statement, the tail keeps the executor's most recent work — + the part a completeness review is actually about. (Head-only + truncation left the reviewer judging "genuinely done?" without ever + seeing the recent evidence.) + """ + text = json.dumps(messages, default=str, ensure_ascii=False) + cap = self._config.transcript_max_chars + if len(text) > cap: + head = cap // 4 + tail = cap - head + text = ( + text[:head] + + "\n......\n" + + text[-tail:] + ) + tail_turn = terminal_content or "(no text)" + return ( + f"Conversation so far (JSON):\n\n{text}\n\n" + f"The executor's latest turn (a plan, or its claim the task is done):\n{tail_turn}" + ) + + +# ---------------------------------------------------------------------- +# Advisor callers +# ---------------------------------------------------------------------- + + +def _build_advisor_caller(config: AdvisorConfig) -> AdvisorCaller: + """Build the advisor caller for ``config.advisor``, dispatched on its format.""" + from switchyard.lib.backends.llm_target import BackendFormat + from switchyard.lib.backends.multi_llm_backend import resolve_llm_target + + target = resolve_llm_target(config.advisor) + if target.format == BackendFormat.ANTHROPIC: + return _AnthropicAdvisorCaller( + api_key=target.endpoint.api_key, + base_url=target.endpoint.base_url, + model=target.model, + max_tokens=config.advisor_max_tokens, + temperature=config.advisor_temperature, + timeout=target.endpoint.timeout_secs, + ) + if target.format == BackendFormat.OPENAI: + return _OpenAiAdvisorCaller( + target=target, + max_tokens=config.advisor_max_tokens, + temperature=config.advisor_temperature, + ) + raise ValueError( + f"advisor tier does not support format {target.format!r}; " + "use 'openai' or 'anthropic'" + ) + + +class _AnthropicAdvisorCaller: + """Reviews via an Anthropic-Messages advisor (``/v1/messages``, Bearer auth).""" + + def __init__( + self, *, api_key: str | None, base_url: str | None, model: str, + max_tokens: int, temperature: float | None, timeout: float | None, + ) -> None: + self._url = _messages_url(base_url) + self._api_key = api_key + self._model = model + self._max_tokens = max_tokens + self._temperature = temperature + self._timeout = timeout + + async def advise(self, *, system: str, transcript: str) -> tuple[str, Any]: + body: dict[str, Any] = { + "model": self._model, + "system": system, + "messages": [{"role": "user", "content": transcript}], + "max_tokens": self._max_tokens, + } + if self._temperature is not None: + body["temperature"] = self._temperature + headers = { + "Authorization": f"Bearer {self._api_key}", + "anthropic-version": _ANTHROPIC_VERSION, + "Content-Type": "application/json", + } + async with httpx.AsyncClient(timeout=self._timeout) as client: + response = await client.post(self._url, json=body, headers=headers) + response.raise_for_status() + data = response.json() + return _anthropic_text(data), data.get("usage") + + +class _OpenAiAdvisorCaller: + """Consults an OpenAI-Chat advisor (``/chat/completions`` via the SDK). + + Covers OSS advisors (DeepSeek, Qwen on vLLM/NIM) and OpenAI. Built with + ``max_retries=0`` so a slow or down advisor falls through to the backend's + own ``fail_open`` handling at the configured timeout instead of + compounding via SDK exponential backoff (same rationale as the LLM + classifier's client). + """ + + def __init__( + self, *, target: LlmTarget, max_tokens: int, temperature: float | None, + ) -> None: + from switchyard.lib.llm_client import OpenAILLMClient + + self._client = OpenAILLMClient( + api_key=target.endpoint.api_key, + base_url=target.endpoint.base_url, + timeout=target.endpoint.timeout_secs, + max_retries=0, + ) + self._model = target.model + self._max_tokens = max_tokens + self._temperature = temperature + # Forward target-level overrides so gateway auth headers and vLLM + # chat-template hints configured on the route work here too. + self._extra_body = dict(target.extra_body) if target.extra_body else None + self._extra_headers = dict(target.extra_headers) if target.extra_headers else None + + async def advise(self, *, system: str, transcript: str) -> tuple[str, Any]: + kwargs: dict[str, Any] = { + "model": self._model, + "messages": [ + {"role": "system", "content": system}, + {"role": "user", "content": transcript}, + ], + "max_tokens": self._max_tokens, + } + if self._temperature is not None: + kwargs["temperature"] = self._temperature + if self._extra_body is not None: + kwargs["extra_body"] = self._extra_body + if self._extra_headers is not None: + kwargs["extra_headers"] = self._extra_headers + result = await self._client.acompletion(**kwargs) + choices = getattr(result, "choices", None) or [] + content = getattr(getattr(choices[0], "message", None), "content", None) if choices else None + return (content or "").strip(), getattr(result, "usage", None) + + +# ---------------------------------------------------------------------- +# Module-level helpers +# ---------------------------------------------------------------------- + + +async def _consume_anthropic_stream( + stream: Any, +) -> tuple[list[Any], bool, str | None, dict[str, int]]: + """Buffer an Anthropic stream; return (events, has_tool_use, assistant_text, usage).""" + events: list[Any] = [] + has_tool_use = False + text_parts: list[str] = [] + usage = {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} + async for event in stream: + events.append(event) + etype = _ev(event, "type") + if etype == "message_start": + start_usage = _ev(_ev(event, "message"), "usage") or {} + usage["input_tokens"] = int(start_usage.get("input_tokens") or 0) + usage["cached_tokens"] = int(start_usage.get("cache_read_input_tokens") or 0) + elif etype == "content_block_start": + if _ev(_ev(event, "content_block"), "type") == "tool_use": + has_tool_use = True + elif etype == "content_block_delta": + delta = _ev(event, "delta") + if _ev(delta, "type") == "text_delta": + piece = _ev(delta, "text") + if isinstance(piece, str): + text_parts.append(piece) + elif etype == "message_delta": + if _ev(_ev(event, "delta"), "stop_reason") == "tool_use": + has_tool_use = True + delta_usage = _ev(event, "usage") or {} + usage["output_tokens"] = int(delta_usage.get("output_tokens") or 0) + return events, has_tool_use, ("".join(text_parts) or None), usage + + +def _completion_usage(body: Any, *, is_openai: bool) -> dict[str, int]: + """Read ``_ExecTurn`` token counts from a non-streamed completion body.""" + usage = body.get("usage") if isinstance(body, dict) else None + prompt_tokens, completion_tokens = _usage_tokens(usage) + details = usage if isinstance(usage, dict) else {} + if is_openai: + cached = (details.get("prompt_tokens_details") or {}).get("cached_tokens") or 0 + else: + cached = details.get("cache_read_input_tokens") or 0 + return { + "input_tokens": prompt_tokens or 0, + "output_tokens": completion_tokens or 0, + "cached_tokens": int(cached), + } + + +async def _replay_events(events: list[Any]) -> Any: + """Replay buffered stream events verbatim as a fresh async stream.""" + for event in events: + yield event + + +def _completion_tool_use(body: Any) -> tuple[bool, str | None]: + """Read (has_tool_use, assistant_text) from an Anthropic completion body.""" + if not isinstance(body, dict): + return False, None + content = body.get("content") or [] + has_tool_use = body.get("stop_reason") == "tool_use" or any( + isinstance(b, dict) and b.get("type") == "tool_use" for b in content + ) + return has_tool_use, (_blocks_text(content) or None) + + +def _openai_completion_tool_use(body: Any) -> tuple[bool, str | None]: + """Read (has_tool_use, assistant_text) from an OpenAI chat.completion body. + + Detection is by ``tool_calls`` presence with ``finish_reason`` as a + fallback — some OSS servers mislabel tool-call turns as ``stop``. + """ + if not isinstance(body, dict): + return False, None + choices = body.get("choices") or [{}] + choice = choices[0] if isinstance(choices[0], dict) else {} + message = choice.get("message") or {} + has_tool_use = bool(message.get("tool_calls")) or choice.get("finish_reason") == "tool_calls" + return has_tool_use, (message.get("content") or None) + + +def _openai_completion_reasoning(body: Any) -> str | None: + """Read ``reasoning_content`` from an OpenAI chat.completion body, if any.""" + if not isinstance(body, dict): + return None + choices = body.get("choices") or [{}] + choice = choices[0] if isinstance(choices[0], dict) else {} + message = choice.get("message") or {} + return message.get("reasoning_content") or None + + +def _gate_request_type(fmt: BackendFormat) -> str: + """Map a resolved executor format to its ``request_with_type`` discriminator.""" + if fmt == BackendFormat.ANTHROPIC: + return "anthropic" + if fmt == BackendFormat.OPENAI: + return "openai_chat" + if fmt == BackendFormat.RESPONSES: + # Backstop for format: auto resolving to a Responses endpoint; the + # config validator rejects an explicit responses format earlier. + raise ValueError( + "the advisor strategies are Chat-shaped and do not support " + "Responses executors; use format 'openai' or 'anthropic'" + ) + raise ValueError( + f"advisor executor format {fmt!r} must be resolved before constructing " + "the backend (pin format: 'openai' or 'anthropic' when supplying " + "executor_backend)" + ) + + +async def _consume_openai_stream( + stream: Any, +) -> tuple[list[Any], dict[str, Any], dict[str, int]]: + """Buffer an OpenAI Chat stream; reassemble the assistant message and usage. + + Events are the ``chat.completion.chunk`` dicts the native backend's SSE + parser yields (``[DONE]`` is consumed upstream and never appears; the + backend force-injects ``stream_options.include_usage`` so a final usage + chunk normally arrives). ``delta.tool_calls`` fragments merge by ``index``: + non-empty ``id``/``name`` replace, ``arguments`` fragments concatenate. + Shared by both advisor strategies. + """ + events: list[Any] = [] + text_parts: list[str] = [] + reasoning_parts: list[str] = [] + slots: dict[int, dict[str, str]] = {} + usage = {"input_tokens": 0, "output_tokens": 0, "cached_tokens": 0} + async for event in stream: + events.append(event) + chunk_usage = _ev(event, "usage") + if isinstance(chunk_usage, dict): + usage["input_tokens"] = int(chunk_usage.get("prompt_tokens") or 0) + usage["output_tokens"] = int(chunk_usage.get("completion_tokens") or 0) + details = chunk_usage.get("prompt_tokens_details") or {} + usage["cached_tokens"] = int(details.get("cached_tokens") or 0) + choices = _ev(event, "choices") or [] + delta = _ev(choices[0], "delta") if choices else None + if delta is None: + continue + piece = _ev(delta, "content") + if isinstance(piece, str): + text_parts.append(piece) + # Reasoning models (nemotron on vLLM/NIM) can emit turns whose ONLY + # output is reasoning_content; keep it so the review gate has + # something to show the advisor when visible content is empty. + reasoning_piece = _ev(delta, "reasoning_content") + if isinstance(reasoning_piece, str): + reasoning_parts.append(reasoning_piece) + for fragment in _ev(delta, "tool_calls") or []: + index = int(_ev(fragment, "index") or 0) + slot = slots.setdefault(index, {"id": "", "name": "", "arguments": ""}) + fragment_id = _ev(fragment, "id") + if isinstance(fragment_id, str) and fragment_id: + slot["id"] = fragment_id + function = _ev(fragment, "function") or {} + name = _ev(function, "name") + if isinstance(name, str) and name: + slot["name"] = name + arguments = _ev(function, "arguments") + if isinstance(arguments, str): + slot["arguments"] += arguments + + message: dict[str, Any] = { + "role": "assistant", + "content": "".join(text_parts) or None, + } + if reasoning_parts: + message["reasoning_content"] = "".join(reasoning_parts) + if slots: + message["tool_calls"] = [ + { + # A missing id (some OSS servers omit it in deltas) gets a + # synthesized one so the tool result can reference it. + "id": slot["id"] or f"call_switchyard_{index}", + "type": "function", + "function": { + "name": slot["name"], + # Empty arguments become "{}" so strict endpoints accept + # the replayed history. + "arguments": slot["arguments"] or "{}", + }, + } + for index, slot in sorted(slots.items()) + ] + return events, message, usage + + +def _ev(event: Any, key: str) -> Any: + """Read a field from a stream event (dict from Rust, or an SDK object).""" + if event is None: + return None + if isinstance(event, dict): + return event.get(key) + return getattr(event, key, None) + + +def _with_length_line( + messages: list[dict[str, Any]], line: str, +) -> list[dict[str, Any]]: + """Append a line of text to the **first** user message. + + The doc suggests the latest user message, but the client never sees this + injection, so re-injecting into each turn's newest message would shift the + upstream cache prefix every turn. The first user message is constant across + a session, keeping the prefix stable; the advisor still reads the line via + the forwarded transcript. The list branch emits ``{"type": "text", ...}`` + parts, valid on both the Anthropic and OpenAI wires. Used by + ``seed_plan_advice`` for the seeded upfront plan. + """ + msgs = [dict(m) for m in messages] + for msg in msgs: + if msg.get("role") != "user": + continue + content = msg.get("content") + if isinstance(content, list): + msg["content"] = [*content, {"type": "text", "text": line}] + elif isinstance(content, str) or content is None: + msg["content"] = f"{content or ''}\n\n{line}".lstrip() + break + return msgs + + +def _seed_transcript(messages: list[dict[str, Any]], cap: int) -> str: + """Serialize the session-opening messages for the seed consult.""" + text = json.dumps(messages, default=str, ensure_ascii=False) + if len(text) > cap: + text = text[: cap - 16] + "..." + return ( + f"The task the executor is about to start (JSON):\n\n{text}\n\n" + "The executor has not begun yet. Review the task and give your best " + "upfront plan: the approach, the pitfalls to avoid, and the first " + "concrete steps." + ) + + +async def _seed_advice_for( + cache: dict[str, str], + session: str, + messages: list[dict[str, Any]], + *, + caller: AdvisorCaller, + config: AdvisorConfig, + stats: StatsAccumulator | None, + ctx: ProxyContext | None = None, +) -> str: + """Per-session seed advice for ``seed_plan_advice`` (both strategies). + + The advisor is consulted once per session, at a request that opens the + conversation (no assistant turns yet); the advice is cached so every later + turn of the session re-injects the identical text (stable cache prefix). + A session first seen mid-conversation (e.g. after a proxy restart) is + cached as unseeded — injecting new advice mid-session would shift the + upstream prefix. Fail-open: a failed consult caches "" (no retry storm). + """ + cached = cache.get(session) + if cached is not None: + return cached + if any(m.get("role") == "assistant" for m in messages): + cache[session] = "" + return "" + advice = await _fetch_seed_advice( + caller=caller, config=config, messages=messages, stats=stats, ctx=ctx, + ) + cache[session] = advice + return advice + + +async def _fetch_seed_advice( + *, + caller: AdvisorCaller, + config: AdvisorConfig, + messages: list[dict[str, Any]], + stats: StatsAccumulator | None, + ctx: ProxyContext | None = None, +) -> str: + """Consult the advisor for an upfront plan; "" on fail-open failure.""" + transcript = _seed_transcript(messages, config.transcript_max_chars) + started = time.monotonic() + try: + advice, usage = await caller.advise( + system=config.advisor_system_prompt, transcript=transcript, + ) + except Exception as exc: + if not config.fail_open: + raise + log.warning("seed_plan_advice: advisor call failed; proceeding unseeded: %s", exc) + if stats is not None: + await stats.record_classifier_error(config.advisor.model) + _audit_seed(error=str(exc), usage=None, + latency_ms=(time.monotonic() - started) * 1000.0) + return "" + latency_ms = (time.monotonic() - started) * 1000.0 + tokens = _advisor_usage(usage) + if stats is not None: + await stats.record_classifier_usage( + model=config.advisor.model, + prompt_tokens=tokens["prompt_tokens"], + completion_tokens=tokens["completion_tokens"], + cached_tokens=tokens["cached_tokens"], + latency_ms=latency_ms, + ) + if ctx is not None: + _emit_routing_usage( + ctx, + model=config.advisor.model, + tier="advisor_seed", + prompt_tokens=tokens["prompt_tokens"], + completion_tokens=tokens["completion_tokens"], + cached_tokens=tokens["cached_tokens"], + cache_creation_tokens=tokens["cache_creation_tokens"], + ) + _audit_seed(error=None, usage=usage, latency_ms=latency_ms) + return advice.strip() + + +def _audit_seed(*, error: str | None, usage: Any, latency_ms: float) -> None: + """Emit a one-line ``advisor_seed=...`` audit record to stderr.""" + payload: dict[str, Any] = { + "advisor_seed": True, + "error": error, + "latency_ms": round(latency_ms, 1), + } + _merge_audit_tokens(payload, usage) + sys.stderr.write(f"advisor_seed={json.dumps(payload, sort_keys=True)}\n") + sys.stderr.flush() + + +def _count_tool_results(messages: list[dict[str, Any]]) -> int: + """Count tool results across both wires (OpenAI ``role: tool`` messages, + Anthropic ``tool_result`` blocks in user messages).""" + n = 0 + for m in messages: + if m.get("role") == "tool": + n += 1 + elif m.get("role") == "user" and isinstance(m.get("content"), list): + n += sum( + 1 for b in m["content"] + if isinstance(b, dict) and b.get("type") == "tool_result" + ) + return n + + +def _session_key(system: Any, messages: list[dict[str, Any]]) -> str: + """Stable per-session key: hash of the cache-stable system prefix + first user message. + + The system prompt is *not* constant across a session on real agent harnesses: + Claude Code re-renders volatile context (reminders, todo state, environment) + into it on every request. Hashing the whole thing minted a fresh key per turn, + silently resetting the ``max_reviews`` budget — observed as 87 reviews on a + single Terminal-Bench task instead of the configured 2. + + Only the portion up to and including the client's last ``cache_control`` + breakpoint is used: that prefix is stable by construction, because the client + is asserting it is byte-identical across turns for prompt caching. Anything + after the final breakpoint is volatile by definition and must not affect + session identity. Clients that set no breakpoint fall back to the first user + message alone, which is the stable task statement. + """ + parts: list[str] = ["S:" + _cache_stable_system_text(system)] + for m in messages: + if m.get("role") == "user": + parts.append("U:" + _blocks_text(m.get("content"))) + break + return hashlib.sha256("\n".join(parts).encode("utf-8", "ignore")).hexdigest() + + +def _cache_stable_system_text(system: Any) -> str: + """System text through the last ``cache_control`` breakpoint (stable prefix). + + Returns "" when the client marks no breakpoint, so session identity then rests + on the first user message rather than on volatile per-turn system content. + """ + if isinstance(system, str): + # A bare string carries no breakpoint information; it is echoed verbatim + # by clients that do not use structured system blocks. + return system + if not isinstance(system, list): + return "" + last_breakpoint = -1 + for index, block in enumerate(system): + if isinstance(block, dict) and block.get("cache_control"): + last_breakpoint = index + if last_breakpoint < 0: + return "" + return _blocks_text(system[: last_breakpoint + 1]) + + +def _blocks_text(content: Any) -> str: + """Flatten Anthropic content (string, or a list of blocks) to text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + return "\n".join( + b.get("text", "") for b in content + if isinstance(b, dict) and isinstance(b.get("text"), str) + ) + return "" + + +#: Anchored verdict match: markdown/quote wrappers and a "Verdict:"-style +#: label may precede the verdict word, but PROSE may not — an unanchored +#: window turned "I cannot approve this — REDO: run the tests" into APPROVE +#: (first case-insensitive token wins). Tolerated prefixes: whitespace, +#: ``*_#>"'([`` characters, and a short label ending in ``:``. +_VERDICT_RE = re.compile( + r"^[\s*_#>\"'\(\[`]*(?:(?:final\s+)?verdict\s*:\s*[\s*_#>\"'\(\[`]*)?(APPROVE|REDO)\b", + re.IGNORECASE, +) + + +def _parse_verdict(text: str) -> tuple[str, str]: + """Parse the reviewer reply into (verdict, plan). + + Returns ``("", "")`` when no leading verdict is found — the caller treats + that as a failed consult (budget refunded) rather than a silent APPROVE, + so a hedged or malformed reply cannot burn the review budget. + """ + stripped = (text or "").strip() + match = _VERDICT_RE.match(stripped) + if match is None: + return "", "" + if match.group(1).upper() == "APPROVE": + return "APPROVE", "" + plan = stripped[match.end():].lstrip(" *_:\n-").strip() + return "REDO", plan or stripped + + +def _messages_url(base_url: str | None) -> str: + """Resolve the Anthropic Messages URL from a target base URL.""" + base = (base_url or "https://api.anthropic.com").rstrip("/") + if base.endswith("/v1/messages"): + return base + if base.endswith("/v1"): + return f"{base}/messages" + return f"{base}/v1/messages" + + +def _anthropic_text(data: dict[str, Any]) -> str: + """Join the ``text`` content blocks of an Anthropic Messages response.""" + content = data.get("content") or [] + return "".join( + b.get("text", "") for b in content + if isinstance(b, dict) and b.get("type") == "text" + ).strip() + + +def _usage_tokens(usage: Any) -> tuple[int | None, int | None]: + """Read (input, output) token counts from Anthropic- or OpenAI-shaped usage.""" + if usage is None: + return None, None + + def get(*names: str) -> int | None: + for name in names: + value = usage.get(name) if isinstance(usage, dict) else getattr(usage, name, None) + if value is not None: + return int(value) + return None + + return get("input_tokens", "prompt_tokens"), get("output_tokens", "completion_tokens") + + +def _advisor_usage(usage: Any) -> dict[str, int]: + """Token fields for an advisor consult, with cache buckets folded in. + + Anthropic-shaped usage reports cache reads/writes as SIBLINGS of + ``input_tokens`` — and some gateways (NVIDIA Inference Hub's bedrock + routes) auto-cache large prompts server-side even when the caller set no + ``cache_control``, so a consult's real input lands almost entirely in + ``cache_creation_input_tokens`` while ``input_tokens`` reads as ~2. + ``prompt_tokens`` here is the inclusive total, matching the routing-log + processor's accounting; OpenAI ``prompt_tokens`` are already inclusive. + """ + + def get(container: Any, name: str) -> int: + value = ( + container.get(name) if isinstance(container, dict) + else getattr(container, name, None) + ) + return int(value) if value is not None else 0 + + input_tokens, output_tokens = _usage_tokens(usage) + cache_read = get(usage, "cache_read_input_tokens") + cache_creation = get(usage, "cache_creation_input_tokens") + if cache_read or cache_creation: + prompt = (input_tokens or 0) + cache_read + cache_creation + cached = cache_read + else: + prompt = input_tokens or 0 + details = ( + usage.get("prompt_tokens_details") if isinstance(usage, dict) + else getattr(usage, "prompt_tokens_details", None) + ) + cached = get(details, "cached_tokens") if details is not None else 0 + return { + "prompt_tokens": prompt, + "cached_tokens": cached, + "cache_creation_tokens": cache_creation, + "completion_tokens": output_tokens or 0, + } + + +def _audit_review( + *, + verdict: str, + error: str | None, + usage: Any, + latency_ms: float, + reply_head: str | None = None, +) -> None: + """Emit a one-line ``advisor_review=...`` audit record to stderr. + + ``reply_head`` carries the raw reply's first characters so a verdict the + parser read differently than the reviewer intended is visible in the logs + (the parsed verdict alone hides misparses). + """ + payload: dict[str, Any] = { + "advisor_review": True, + "verdict": verdict, + "error": error, + "latency_ms": round(latency_ms, 1), + } + if reply_head is not None: + payload["reply_head"] = reply_head.strip()[:160] + _merge_audit_tokens(payload, usage) + sys.stderr.write(f"advisor_review={json.dumps(payload, sort_keys=True)}\n") + sys.stderr.flush() + + +def _merge_audit_tokens(payload: dict[str, Any], usage: Any) -> None: + """Add cache-inclusive token fields to an audit payload (no-op when None).""" + if usage is None: + return + tokens = _advisor_usage(usage) + payload["prompt_tokens"] = tokens["prompt_tokens"] + payload["completion_tokens"] = tokens["completion_tokens"] + if tokens["cache_creation_tokens"]: + payload["cache_creation_tokens"] = tokens["cache_creation_tokens"] + if tokens["cached_tokens"]: + payload["cached_tokens"] = tokens["cached_tokens"] + + +def _emit_routing_usage( + ctx: ProxyContext, + *, + model: str, + tier: str, + prompt_tokens: int, + completion_tokens: int, + cached_tokens: int = 0, + cache_creation_tokens: int = 0, +) -> None: + """Append a proxy-internal usage record to the routing log, if one is active. + + The routing-log response processor only sees the chain's terminal + response; advisor consults and REDO-discarded executor turns happen inside + the backend and would otherwise be invisible to + ``/v1/routing/session-stats`` — and with it, per-model cost attribution. + """ + from switchyard.lib.processors.routing_log_response_processor import ( + emit_auxiliary_record, + ) + + metadata = ctx.metadata.get(CTX_REQUEST_METADATA) + emit_auxiliary_record( + session_id=getattr(metadata, "session_id", None), + task=getattr(metadata, "task", None), + model=model, + tier=tier, + prompt_tokens=prompt_tokens, + cached_tokens=cached_tokens, + cache_creation_tokens=cache_creation_tokens, + completion_tokens=completion_tokens, + ) + + +__all__ = ["AdvisorCaller", "AdvisorLoopBackend"] diff --git a/switchyard/lib/backends/advisor_presets.py b/switchyard/lib/backends/advisor_presets.py new file mode 100644 index 000000000..c6c8f82be --- /dev/null +++ b/switchyard/lib/backends/advisor_presets.py @@ -0,0 +1,90 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Named :class:`AdvisorConfig` presets keyed by shipping bundle. + +The shipping default :meth:`AdvisorPresets.opus47_exec_opus48_advisor` pairs an +Opus 4.7 executor with an Opus 4.8 advisor, both served native Anthropic from +NVIDIA Inference Hub. The advisor gates the executor with a once-per-session +review: at the executor's first no-tool-call turn it is consulted to APPROVE +or send the executor back (REDO) with an optimized plan. + +Example:: + + from switchyard import AdvisorLoopBackend, AdvisorPresets + + config = AdvisorPresets.opus47_exec_opus48_advisor(api_key=nvidia_api_key) + backend = AdvisorLoopBackend(config) +""" + +from __future__ import annotations + +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.llm_target import BackendFormat, LlmTarget + +# All shipping presets route through NVIDIA Inference Hub's Anthropic Messages +# endpoint by default; callers override with ``base_url=`` for a different gateway. +_INFERENCE_HUB_BASE_URL = "https://inference-api.nvidia.com/v1" + +# Inference Hub model ids. Both tiers (Opus 4.7 executor, Opus 4.8 advisor) are +# served native Anthropic-Messages at ``/v1/messages`` — no OpenAI translation, +# so prompt caching survives. Both are overridable via the preset's +# ``executor_model`` / ``advisor_model``. +_MODEL_OPUS_4_7_EXECUTOR = "aws/anthropic/bedrock-claude-opus-4-7" +_MODEL_OPUS_4_8_ADVISOR = "aws/anthropic/bedrock-claude-opus-4-8" + + +class AdvisorPresets: + """Factory of pre-built :class:`AdvisorConfig` bundles.""" + + @staticmethod + def opus47_exec_opus48_advisor( + *, + api_key: str, + base_url: str = _INFERENCE_HUB_BASE_URL, + timeout_secs: float | None = 600.0, + executor_model: str = _MODEL_OPUS_4_7_EXECUTOR, + advisor_model: str = _MODEL_OPUS_4_8_ADVISOR, + ) -> AdvisorConfig: + """Opus 4.7 executor + Opus 4.8 advisor on NVIDIA Inference Hub. + + Args: + api_key: Inference Hub API key, used for both tiers (one tenancy). + base_url: OpenAI-compatible gateway base URL. + timeout_secs: Per-call timeout for both tiers. Generous by default + because the advisor consult adds an extra round-trip inside a + single client request. + executor_model: Override the executor model id if your tenancy + serves Opus 4.7 under a different string. + advisor_model: Override the advisor model id likewise. + """ + return AdvisorConfig( + executor=LlmTarget( + # Native Anthropic Messages (``/v1/messages``): the request passes + # through verbatim so the client's cache_control breakpoints reach + # the upstream and prompt caching is honored. Inference Hub wants + # Bearer auth (not Anthropic's x-api-key), so suppress x-api-key + # (api_key="") and carry the key in an Authorization header. + id="executor", + model=executor_model, + format=BackendFormat.ANTHROPIC, + api_key="", + base_url=base_url, + timeout_secs=timeout_secs, + extra_headers={"Authorization": f"Bearer {api_key}"}, + ), + advisor=LlmTarget( + # Anthropic Messages format: consulted via ``/v1/messages`` (the + # advisor caller sends Bearer auth directly from ``api_key``). + id="advisor", + model=advisor_model, + format=BackendFormat.ANTHROPIC, + api_key=api_key, + base_url=base_url, + timeout_secs=timeout_secs, + ), + preset="opus47_exec_opus48_advisor", + ) + + +__all__ = ["AdvisorPresets"] diff --git a/switchyard/lib/backends/advisor_prompts.py b/switchyard/lib/backends/advisor_prompts.py new file mode 100644 index 000000000..9499ddc8a --- /dev/null +++ b/switchyard/lib/backends/advisor_prompts.py @@ -0,0 +1,56 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Default prompts for the advisor review gate. + +The advisor is a once-per-session reviewer: it is consulted at the first point +the executor produces a no-tool-call turn — either a plan it is about to +execute, or a claim that the task is done — and returns ``APPROVE`` (let the +executor stop) or ``REDO`` + an optimized plan (send it back to keep working). +This preserves the executor's own test-and-iterate loop (which front-loaded +advice was found to suppress, causing premature convergence) and adds a +single quality gate on top. + +These are defaults; :class:`~switchyard.lib.backends.advisor_config.AdvisorConfig` +exposes each as an overridable field for ablation. +""" + +from __future__ import annotations + +# Tells the advisor model its role for the optional ``seed_plan_advice`` +# consult, so it advises rather than attempting the task itself. +ADVISOR_SYSTEM_PROMPT = """\ +You are a higher-intelligence advisor model consulted mid-task by a faster executor model. You can see the full conversation: the task, every tool call, and every result. You do not act, write code, or call tools — you provide strategic guidance only: a focused plan or a course correction the executor will carry out. Be concrete and brief.\ +""" + +REVIEWER_SYSTEM_PROMPT = """\ +You are a senior reviewer acting as a quality gate for a faster executor model working a coding/agent task. You are given the full transcript: the task, every action the executor took and every result it saw, and its latest message — in which it has either (a) proposed a plan before doing the work, or (b) concluded the task is complete. + +Decide whether to let the executor stop or send it back to keep working. Put your verdict as the FIRST word of your reply: + +- APPROVE — the proposed plan is sound, OR the work is genuinely complete and correct. Reply with exactly: APPROVE +- REDO — the plan has a real flaw, OR the work is incomplete/incorrect: an unhandled edge case, an untested assumption, a subtly wrong approach, missing verification, or a stated requirement not met. Reply: REDO, then a SHORT, concrete, actionable plan naming exactly what is wrong or missing and what to do about it. No generic advice — point at the specific gap. + +Bias toward APPROVE when the work looks correct and complete; the executor has already done its own iteration. Use REDO specifically to catch a premature "done" on a subtly incomplete solution, or a flawed plan before it is executed. A self-claim of success is not proof — check the actual task requirements against what was actually done. +""" + +#: Prepended to the advisor's REDO plan when it is injected back to the executor +#: as a user turn, instructing it to continue rather than stop. +REDO_FEEDBACK_PREFIX = ( + "A senior reviewer examined your work and determined the task is NOT yet " + "complete or correct. Do not stop here — address the following, then keep " + "working until it is genuinely done:\n\n" +) + +#: Prepended to the advisor's upfront plan when ``seed_plan_advice`` injects it +#: into the session's first user message. +SEED_ADVICE_PREFIX = ( + "\n\nA senior advisor reviewed this task before you started and suggests:\n" +) + +__all__ = [ + "ADVISOR_SYSTEM_PROMPT", + "REDO_FEEDBACK_PREFIX", + "REVIEWER_SYSTEM_PROMPT", + "SEED_ADVICE_PREFIX", +] diff --git a/switchyard/lib/processors/reasoning_effort_normalizer.py b/switchyard/lib/processors/reasoning_effort_normalizer.py new file mode 100644 index 000000000..796ff4150 --- /dev/null +++ b/switchyard/lib/processors/reasoning_effort_normalizer.py @@ -0,0 +1,68 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Normalize non-standard ``reasoning_effort`` values before backend dispatch. + +Claude Code's ``/effort`` picker offers an ``xhigh`` level (and the +Codex launcher's model catalog mirrors it). Both upstream paths reject +that value: NVIDIA Inference Hub's LiteLLM passthrough returns HTTP +500 with ``Invalid effort value: xhigh. Must be one of: 'high', +'medium', 'low', 'max'.`` for Azure Anthropic, and +``Unmapped reasoning effort: xhigh`` for Bedrock. + +This processor runs early in the chain and normalizes the request body +so the upstream never sees an unsupported value. Unknown values are +mapped to ``high`` rather than stripped because the user's intent +(``xhigh`` means "as much reasoning as possible") is closer to ``high`` +than to absent. +""" + +from __future__ import annotations + +import logging + +from switchyard.lib.proxy_context import ProxyContext +from switchyard_rust.core import ChatRequest + +log = logging.getLogger(__name__) + +#: OpenAI-compatible values the upstream accepts. ``"max"`` is non-standard +#: but supported by NVIDIA Hub's LiteLLM for reasoning-budget overrides. +_VALID_REASONING_EFFORT = frozenset({"low", "medium", "high", "max"}) + +#: Aliases the upstream rejects → the nearest valid value. +_REASONING_EFFORT_ALIASES = { + "xhigh": "high", +} + + +class ReasoningEffortNormalizer: + """Normalize ``request.body["reasoning_effort"]`` to an upstream-valid value. + + No-op when the field is absent or already a valid value. Maps known + aliases (``xhigh`` → ``high``); for unrecognized values, replaces + with ``high`` and emits a warning so operators can spot + misconfigured client-side enums. + """ + + async def process(self, ctx: ProxyContext, request: ChatRequest) -> ChatRequest: # noqa: ARG002 + body = request.body + if not isinstance(body, dict): + return request + effort = body.get("reasoning_effort") + if not isinstance(effort, str) or effort in _VALID_REASONING_EFFORT: + return request + + mapped = _REASONING_EFFORT_ALIASES.get(effort, "high") + log.warning( + "ReasoningEffortNormalizer: unsupported reasoning_effort=%r; " + "normalizing to %r before dispatch", + effort, + mapped, + ) + body["reasoning_effort"] = mapped + request.replace_body(body) + return request + + +__all__ = ["ReasoningEffortNormalizer"] diff --git a/switchyard/lib/processors/routing_log_response_processor.py b/switchyard/lib/processors/routing_log_response_processor.py index cc87c0183..c667ea241 100644 --- a/switchyard/lib/processors/routing_log_response_processor.py +++ b/switchyard/lib/processors/routing_log_response_processor.py @@ -73,6 +73,10 @@ def _write_record(self, ctx: ProxyContext, served_model: str, response: ChatResp "tier": ctx.selected_target or "", **_usage_tokens(response.body), } + self._append(record) + + def _append(self, record: dict[str, Any]) -> None: + """Append one record as a JSON line; write failures never propagate.""" try: line = json.dumps(record, separators=(",", ":")) with self._lock, self._log_file.open("a", encoding="utf-8") as handle: @@ -152,6 +156,56 @@ def get_endpoint(self) -> Endpoint: return RoutingLogStatsEndpoint(self) +#: Process-global sink for proxy-internal usage records (advisor consults, +#: REDO-discarded executor turns). Multi-call backends issue upstream requests +#: beneath the chain, so this response processor never sees them; the backends +#: emit records here instead. Registered by the serving entry point that owns +#: the routing log; None (the default) makes emission a no-op. +_AUX_SINK: RoutingLogResponseProcessor | None = None + + +def register_routing_log_sink(processor: RoutingLogResponseProcessor | None) -> None: + """Register (or clear) the routing log that receives auxiliary records.""" + global _AUX_SINK + _AUX_SINK = processor + + +def emit_auxiliary_record( + *, + session_id: str | None, + task: str | None, + model: str, + tier: str, + prompt_tokens: int = 0, + cached_tokens: int = 0, + cache_creation_tokens: int = 0, + completion_tokens: int = 0, +) -> None: + """Append a proxy-internal usage record to the registered routing log. + + The record shape matches ``_write_record`` exactly, so + ``snapshot_session`` (and with it ``/v1/routing/session-stats``) + aggregates chain-terminal and proxy-internal usage alike. + """ + sink = _AUX_SINK + if sink is None: + return + sink._append({ + "ts": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z", + "task": task, + "trial_id": None, + "session_id": session_id, + "model": model, + "tier": tier, + "prompt_tokens": prompt_tokens, + "cached_tokens": cached_tokens, + "cache_creation_tokens": cache_creation_tokens, + "completion_tokens": completion_tokens, + "reasoning_tokens": 0, + "total_tokens": prompt_tokens + completion_tokens, + }) + + def _usage_tokens(body: object) -> dict[str, int]: """Six-field token breakdown matching the global routing-stats schema. diff --git a/tests/test_advisor_config.py b/tests/test_advisor_config.py new file mode 100644 index 000000000..a1dd90b89 --- /dev/null +++ b/tests/test_advisor_config.py @@ -0,0 +1,88 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for :class:`AdvisorConfig` validation, presets, and public exports.""" + +from __future__ import annotations + +import pydantic +import pytest + +import switchyard +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.advisor_loop_backend import AdvisorLoopBackend +from switchyard.lib.backends.advisor_presets import AdvisorPresets +from switchyard.lib.backends.llm_target import BackendFormat + + +def _config(**overrides) -> AdvisorConfig: + base: dict = { + "executor": {"model": "exec-model", "base_url": "http://exec.test", "api_key": "k", + "format": "anthropic"}, + "advisor": {"model": "adv-model", "base_url": "http://adv.test", "api_key": "k", + "format": "anthropic"}, + } + base.update(overrides) + return AdvisorConfig(**base) + + +class TestAdvisorConfig: + def test_coerces_dict_targets(self) -> None: + cfg = _config() + assert cfg.executor.model == "exec-model" + assert cfg.advisor.model == "adv-model" + + def test_rejects_empty_target_model(self) -> None: + # The Rust-backed LlmTarget rejects empty model ids during coercion, + # before the config-level non-empty validator can fire. + with pytest.raises(pydantic.ValidationError, match="must not be empty"): + _config(executor={"model": "", "base_url": "http://e", "api_key": "k"}) + + def test_rejects_responses_format_on_either_tier(self) -> None: + for tier in ("executor", "advisor"): + with pytest.raises(pydantic.ValidationError, match="responses"): + _config(**{tier: {"model": "m", "base_url": "http://t", "api_key": "k", + "format": "responses"}}) + + def test_redo_feedback_prefix_is_configurable(self) -> None: + cfg = _config(redo_feedback_prefix="REVIEWER SAYS: ") + assert cfg.redo_feedback_prefix == "REVIEWER SAYS: " + + def test_accepts_mixed_wire_tiers(self) -> None: + mixed = _config(advisor={"model": "deepseek/deepseek-r2", "base_url": "http://adv.test", + "api_key": "k", "format": "openai"}) + assert mixed.advisor.format == BackendFormat.OPENAI + assert mixed.executor.format == BackendFormat.ANTHROPIC + + +class TestOpusPairPreset: + """Pins the validated executor+advisor pairing on the shipping default.""" + + def test_preset_pairs_opus_47_and_48(self) -> None: + cfg = AdvisorPresets.opus47_exec_opus48_advisor(api_key="nvapi-test") + assert cfg.executor.model == "aws/anthropic/bedrock-claude-opus-4-7" + assert cfg.advisor.model == "aws/anthropic/bedrock-claude-opus-4-8" + assert cfg.preset == "opus47_exec_opus48_advisor" + assert cfg.executor.endpoint.base_url == "https://inference-api.nvidia.com/v1" + # Both tiers are native Anthropic-Messages (no OpenAI translation → + # caching survives). The executor suppresses x-api-key and + # authenticates via Bearer. + assert cfg.executor.format == BackendFormat.ANTHROPIC + assert cfg.advisor.format == BackendFormat.ANTHROPIC + assert cfg.executor.endpoint.api_key == "" + assert cfg.executor.extra_headers == {"Authorization": "Bearer nvapi-test"} + + def test_preset_model_overrides(self) -> None: + cfg = AdvisorPresets.opus47_exec_opus48_advisor( + api_key="k", executor_model="custom/exec", advisor_model="custom/adv", + ) + assert cfg.executor.model == "custom/exec" + assert cfg.advisor.model == "custom/adv" + + +def test_public_exports() -> None: + assert switchyard.AdvisorConfig is AdvisorConfig + assert switchyard.AdvisorLoopBackend is AdvisorLoopBackend + assert switchyard.AdvisorPresets is AdvisorPresets + for name in ("AdvisorConfig", "AdvisorLoopBackend", "AdvisorPresets"): + assert name in switchyard.__all__ diff --git a/tests/test_advisor_loop_backend.py b/tests/test_advisor_loop_backend.py new file mode 100644 index 000000000..1920636e4 --- /dev/null +++ b/tests/test_advisor_loop_backend.py @@ -0,0 +1,1018 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for the review-gate :class:`AdvisorLoopBackend` (native Anthropic). + +All loop tests inject a fake executor backend (returns ``ChatResponse``) and a +fake advisor caller — no network. They cover the gate contract: tool-use turns +pass through unreviewed; the first no-tool-use turn is reviewed once; APPROVE +returns it; REDO re-invokes the executor to continue; the review is +once-per-session and the session is pure passthrough afterward; fail-open +approves. Separate tests cover the Anthropic reviewer caller (respx) and the +pure helpers. Both streaming and completion executor responses are exercised. +""" + +import json +from unittest.mock import AsyncMock, MagicMock + +import httpx +import pytest +import respx + +from switchyard.lib.backends.advisor_config import AdvisorConfig +from switchyard.lib.backends.advisor_loop_backend import ( + _MAX_FAILED_CONSULTS_PER_SCOPE, + AdvisorLoopBackend, + _advisor_usage, + _anthropic_text, + _AnthropicAdvisorCaller, + _build_advisor_caller, + _messages_url, + _OpenAiAdvisorCaller, + _parse_verdict, + _session_key, + _usage_tokens, +) +from switchyard.lib.backends.llm_target import coerce_llm_target +from switchyard.lib.chat_response.anthropic import AnthropicResponseStream +from switchyard.lib.chat_response.openai_chat import ResponseStream as OpenAIResponseStream +from switchyard.lib.proxy_context import ProxyContext +from switchyard.lib.request_metadata import CTX_REQUEST_METADATA, RequestMetadata +from switchyard.lib.stats_accumulator import StatsAccumulator +from switchyard_rust.core import ( + ChatRequest, + ChatRequestType, + ChatResponse, + ChatResponseType, + response_type_matches, +) + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + + +def _completion_resp(*, text=None, tool_use=False, model="exec-model") -> ChatResponse: + """An Anthropic completion ChatResponse (text and/or a tool_use block).""" + content: list[dict] = [] + if tool_use: + content.append({"type": "tool_use", "id": "t1", "name": "bash", "input": {}}) + if text is not None: + content.append({"type": "text", "text": text}) + body = { + "id": "msg-x", "type": "message", "role": "assistant", "model": model, + "content": content, + "stop_reason": "tool_use" if tool_use else "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 3}, + } + return ChatResponse.anthropic_completion(body) + + +async def _agen(events): + for event in events: + yield event + + +def _stream_resp(*, text=None, tool_use=False) -> ChatResponse: + """An Anthropic streaming ChatResponse (SSE event dicts).""" + events: list[dict] = [{"type": "message_start", "message": {"usage": {"input_tokens": 10}}}] + if tool_use: + events.append({"type": "content_block_start", "index": 0, + "content_block": {"type": "tool_use", "id": "t1", "name": "bash", "input": {}}}) + events.append({"type": "message_delta", "delta": {"stop_reason": "tool_use"}, + "usage": {"output_tokens": 1}}) + else: + events.append({"type": "content_block_start", "index": 0, + "content_block": {"type": "text", "text": ""}}) + if text: + events.append({"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": text}}) + events.append({"type": "message_delta", "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 3}}) + events.append({"type": "message_stop"}) + return ChatResponse.anthropic_stream(AnthropicResponseStream(_agen(events))) + + +def _exec_backend(*responses) -> MagicMock: + b = MagicMock() + b.call = AsyncMock(side_effect=list(responses)) + b.startup = AsyncMock() + b.shutdown = AsyncMock() + return b + + +def _reviewer(*verdicts: str) -> MagicMock: + """Fake advisor reviewer: ``advise`` yields successive ``(verdict_text, usage)``.""" + c = MagicMock() + c.advise = AsyncMock(side_effect=[(v, None) for v in verdicts]) + return c + + +def _failing_reviewer(exc: Exception) -> MagicMock: + c = MagicMock() + c.advise = AsyncMock(side_effect=exc) + return c + + +def _config(**overrides) -> AdvisorConfig: + base: dict = { + "executor": {"model": "exec-model", "base_url": "http://exec.test", "api_key": "k", + "format": "anthropic"}, + "advisor": {"model": "adv-model", "base_url": "http://adv.test", "api_key": "k", + "format": "anthropic"}, + } + base.update(overrides) + return AdvisorConfig(**base) + + +def _backend(config, executor_backend, advisor_caller) -> AdvisorLoopBackend: + return AdvisorLoopBackend( + config, executor_backend=executor_backend, advisor_caller=advisor_caller, + ) + + +def _request(**overrides) -> ChatRequest: + body: dict = {"model": "incoming", "system": "sys", + "messages": [{"role": "user", "content": "build X"}]} + body.update(overrides) + return ChatRequest.anthropic(body) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# Gate behavior +# --------------------------------------------------------------------------- + + +async def test_tool_use_turn_passes_through_unreviewed() -> None: + """A turn with a tool_use block means the executor is working — no review.""" + exec_b = _exec_backend(_completion_resp(text="reading", tool_use=True)) + adv = _reviewer() + resp = await _backend(_config(), exec_b, adv).call(ProxyContext(), _request()) + assert resp.to_body()["stop_reason"] == "tool_use" + assert exec_b.call.await_count == 1 + assert adv.advise.await_count == 0 # advisor never consulted + + +async def test_terminal_turn_approved_returns_as_is() -> None: + """First no-tool-use turn is reviewed; APPROVE returns it unchanged.""" + exec_b = _exec_backend(_completion_resp(text="done, all good")) + adv = _reviewer("APPROVE") + resp = await _backend(_config(), exec_b, adv).call(ProxyContext(), _request()) + assert _anthropic_text(resp.to_body()) == "done, all good" + assert exec_b.call.await_count == 1 + assert adv.advise.await_count == 1 + + +async def test_review_records_advisor_cost_into_stats() -> None: + """The advisor review's tokens are recorded into the accumulator so the run's + own cost output (routing_stats) includes the advisor, not just the executor.""" + exec_b = _exec_backend(_completion_resp(text="done")) + adv = MagicMock() + adv.advise = AsyncMock(return_value=("APPROVE", {"input_tokens": 500, "output_tokens": 8})) + stats = MagicMock() + stats.record_classifier_usage = AsyncMock() + stats.record_success = AsyncMock() + backend = AdvisorLoopBackend( + _config(), stats_accumulator=stats, executor_backend=exec_b, advisor_caller=adv, + ) + await backend.call(ProxyContext(), _request()) + stats.record_classifier_usage.assert_awaited_once() + kw = stats.record_classifier_usage.await_args.kwargs + assert kw["model"] == "adv-model" + assert kw["prompt_tokens"] == 500 + assert kw["completion_tokens"] == 8 + + +async def test_review_cost_recorded_against_real_accumulator() -> None: + """The mock-based test above cannot see a recording API that no longer exists: + ``MagicMock`` answers any attribute, so it stays green while the real call + raises ``AttributeError``. Drive the real accumulator so a bucket removed + upstream fails here instead of at the first live advisor consult.""" + exec_b = _exec_backend(_completion_resp(text="done")) + adv = MagicMock() + adv.advise = AsyncMock(return_value=("APPROVE", {"input_tokens": 500, "output_tokens": 8})) + stats = StatsAccumulator() + backend = AdvisorLoopBackend( + _config(), stats_accumulator=stats, executor_backend=exec_b, advisor_caller=adv, + ) + await backend.call(ProxyContext(), _request()) + advisor_stats = stats.snapshot_sync()["classifier"]["models"]["adv-model"] + assert advisor_stats["prompt_tokens"] == 500 + assert advisor_stats["completion_tokens"] == 8 + + +async def test_redo_reinvokes_executor_with_feedback() -> None: + """REDO feeds the plan back and re-invokes the executor to keep working.""" + exec_b = _exec_backend( + _completion_resp(text="I think I'm done"), # terminal → review + _completion_resp(text="continuing", tool_use=True), # redo continuation (passthrough) + ) + adv = _reviewer("REDO: you forgot the empty-input case; add a guard and test it") + resp = await _backend(_config(), exec_b, adv).call(ProxyContext(), _request()) + + assert adv.advise.await_count == 1 + assert exec_b.call.await_count == 2 # original + redo re-invocation + redo_request = exec_b.call.await_args_list[1].args[1] + redo_msgs = redo_request.to_body()["messages"] + assert any(m.get("role") == "assistant" and m.get("content") == "I think I'm done" for m in redo_msgs) + assert any(m.get("role") == "user" and "empty-input case" in (m.get("content") or "") for m in redo_msgs) + # the returned response is the continuation (has the real tool call) + assert resp.to_body()["stop_reason"] == "tool_use" + + +async def test_review_is_once_per_session() -> None: + """Two terminal turns in the same session → advisor consulted only once.""" + exec_b = _exec_backend(_completion_resp(text="done1"), _completion_resp(text="done2")) + adv = _reviewer("APPROVE") # only one verdict provided on purpose + backend = _backend(_config(), exec_b, adv) + await backend.call(ProxyContext(), _request()) # review #1 + await backend.call(ProxyContext(), _request()) # same prefix → no review + assert adv.advise.await_count == 1 + assert exec_b.call.await_count == 2 + + +async def test_review_budget_survives_session_key_churn() -> None: + """A changing conversation prefix must not refill the review budget. + + Real harnesses compact history and re-render system context, minting a fresh + session key mid-run. With a purely per-session budget the gate silently + refilled: one Terminal-Bench task drew 107 reviews against max_reviews=2. + The instance-level ceiling is what bounds it. + """ + exec_b = _exec_backend(*[_completion_resp(text=f"done{i}") for i in range(6)]) + adv = _reviewer(*["APPROVE"] * 6) + backend = _backend(_config(max_reviews=2), exec_b, adv) + # each call carries a DIFFERENT first user message -> a different session key + for i in range(6): + await backend.call( + ProxyContext(), + _request(messages=[{"role": "user", "content": f"distinct task {i}"}]), + ) + assert adv.advise.await_count == 2, "instance ceiling must hold across new session keys" + + +async def test_instance_budget_allows_configured_reviews() -> None: + """The instance ceiling must not fire early — max_reviews reviews still happen.""" + exec_b = _exec_backend(*[_completion_resp(text=f"done{i}") for i in range(4)]) + adv = _reviewer(*["APPROVE"] * 4) + backend = _backend(_config(max_reviews=2), exec_b, adv) + for i in range(4): + await backend.call( + ProxyContext(), + _request(messages=[{"role": "user", "content": f"distinct task {i}"}]), + ) + assert adv.advise.await_count == 2 + + +async def test_reviewed_session_passes_through_verbatim() -> None: + """After the gate fires, later turns are pure passthrough (no advisor, returned as-is).""" + exec_b = _exec_backend( + _completion_resp(text="plan"), # review fires here + _completion_resp(text="more", tool_use=True), # passthrough verbatim + ) + adv = _reviewer("APPROVE") + backend = _backend(_config(), exec_b, adv) + await backend.call(ProxyContext(), _request()) + resp = await backend.call(ProxyContext(), _request()) + assert adv.advise.await_count == 1 + assert resp.to_body()["stop_reason"] == "tool_use" + + +async def test_fail_open_approves_on_review_error() -> None: + exec_b = _exec_backend(_completion_resp(text="done")) + adv = _failing_reviewer(RuntimeError("advisor down")) + resp = await _backend(_config(fail_open=True), exec_b, adv).call(ProxyContext(), _request()) + assert _anthropic_text(resp.to_body()) == "done" + assert exec_b.call.await_count == 1 # no redo + + +async def test_fail_closed_propagates_review_error() -> None: + exec_b = _exec_backend(_completion_resp(text="done")) + adv = _failing_reviewer(RuntimeError("advisor down")) + with pytest.raises(RuntimeError, match="advisor down"): + await _backend(_config(fail_open=False), exec_b, adv).call(ProxyContext(), _request()) + + +async def test_streaming_terminal_approved_replays() -> None: + """Streaming no-tool-use turn → review → APPROVE → replay (one generation).""" + exec_b = _exec_backend(_stream_resp(text="done")) + adv = _reviewer("APPROVE") + resp = await _backend(_config(), exec_b, adv).call(ProxyContext(), _request(stream=True)) + assert exec_b.call.await_count == 1 + assert adv.advise.await_count == 1 + assert response_type_matches(resp, ChatResponseType.ANTHROPIC_STREAM) + events = [e async for e in resp.stream] + assert any(isinstance(e, dict) and e.get("type") == "content_block_delta" for e in events) + + +async def test_streaming_tool_use_passes_through() -> None: + """Streaming turn with a tool_use block → pass through, no review.""" + exec_b = _exec_backend(_stream_resp(tool_use=True)) + adv = _reviewer() + resp = await _backend(_config(), exec_b, adv).call(ProxyContext(), _request(stream=True)) + assert adv.advise.await_count == 0 + assert response_type_matches(resp, ChatResponseType.ANTHROPIC_STREAM) + + +# --------------------------------------------------------------------------- +# OpenAI-wire gate (executor format: openai) +# --------------------------------------------------------------------------- + + +def _openai_config(**overrides) -> AdvisorConfig: + base: dict = { + "executor": {"model": "qwen/qwen3-max", "base_url": "http://exec.test", + "api_key": "k", "format": "openai"}, + "advisor": {"model": "adv-model", "base_url": "http://adv.test", "api_key": "k", + "format": "anthropic"}, + } + base.update(overrides) + return AdvisorConfig(**base) + + +def _openai_completion_resp(*, text=None, tool_calls=False) -> ChatResponse: + message: dict = {"role": "assistant", "content": text} + if tool_calls: + message["tool_calls"] = [{"id": "b1", "type": "function", + "function": {"name": "bash", "arguments": "{}"}}] + body = { + "id": "chatcmpl-x", "object": "chat.completion", "model": "exec-model", + "choices": [{"index": 0, "message": message, + "finish_reason": "tool_calls" if tool_calls else "stop"}], + "usage": {"prompt_tokens": 10, "completion_tokens": 3}, + } + return ChatResponse.openai_completion(body) + + +def _openai_request(**overrides) -> ChatRequest: + body: dict = {"model": "incoming", + "messages": [{"role": "system", "content": "sys"}, + {"role": "user", "content": "build X"}]} + body.update(overrides) + return ChatRequest.openai_chat(body) # type: ignore[arg-type] + + +def test_openai_gate_advertises_openai_wire() -> None: + backend = _backend(_openai_config(), _exec_backend(), _reviewer()) + assert backend.supported_request_types == [ChatRequestType.OPENAI_CHAT] + + +async def test_openai_tool_call_turn_passes_through_unreviewed() -> None: + exec_b = _exec_backend(_openai_completion_resp(tool_calls=True)) + adv = _reviewer() + resp = await _backend(_openai_config(), exec_b, adv).call(ProxyContext(), _openai_request()) + assert resp.to_body()["choices"][0]["finish_reason"] == "tool_calls" + assert adv.advise.await_count == 0 + + +async def test_openai_terminal_turn_approved_returns_as_is() -> None: + exec_b = _exec_backend(_openai_completion_resp(text="done, all good")) + adv = _reviewer("APPROVE") + resp = await _backend(_openai_config(), exec_b, adv).call(ProxyContext(), _openai_request()) + assert resp.to_body()["choices"][0]["message"]["content"] == "done, all good" + assert adv.advise.await_count == 1 + + +async def test_openai_redo_uses_configured_prefix_and_wire() -> None: + exec_b = _exec_backend( + _openai_completion_resp(text="I think I'm done"), + _openai_completion_resp(text="continuing", tool_calls=True), + ) + adv = _reviewer("REDO: verify the output file exists") + config = _openai_config(redo_feedback_prefix="REVIEWER SAYS: ") + resp = await _backend(config, exec_b, adv).call(ProxyContext(), _openai_request()) + + assert exec_b.call.await_count == 2 + redo_request = exec_b.call.await_args_list[1].args[1] + assert redo_request.request_type == ChatRequestType.OPENAI_CHAT + redo_msgs = redo_request.to_body()["messages"] + assert redo_msgs[-1]["role"] == "user" + assert redo_msgs[-1]["content"].startswith("REVIEWER SAYS: verify the output file") + assert redo_msgs[-2] == {"role": "assistant", "content": "I think I'm done"} + assert resp.to_body()["choices"][0]["finish_reason"] == "tool_calls" + + +async def test_openai_streaming_terminal_approved_replays() -> None: + events = [ + {"id": "c", "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "done"}, + "finish_reason": None}]}, + {"id": "c", "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + ] + exec_b = _exec_backend( + ChatResponse.openai_stream(OpenAIResponseStream(_agen(list(events)))), + ) + adv = _reviewer("APPROVE") + resp = await _backend(_openai_config(), exec_b, adv).call( + ProxyContext(), _openai_request(stream=True), + ) + assert adv.advise.await_count == 1 + assert response_type_matches(resp, ChatResponseType.OPENAI_STREAM) + replayed = [e async for e in resp.stream] + assert replayed == events + + +# --------------------------------------------------------------------------- +# Pattern trigger (text-protocol harnesses) +# --------------------------------------------------------------------------- + + +def _pattern_config(**overrides) -> AdvisorConfig: + base: dict = { + "gate_trigger": "pattern", + "gate_trigger_pattern": r'task_complete["\s>:]*true', + "executor": {"model": "nvidia/nemotron-3-ultra", "base_url": "http://exec.test", + "api_key": "k", "format": "openai"}, + "advisor": {"model": "adv-model", "base_url": "http://adv.test", "api_key": "k", + "format": "anthropic"}, + } + base.update(overrides) + return AdvisorConfig(**base) + + +async def test_pattern_trigger_ignores_non_matching_turns() -> None: + """Ordinary command turns (no marker) pass through unreviewed, even with + no tool calls — the whole point for text-protocol harnesses.""" + exec_b = _exec_backend(_openai_completion_resp(text='{"commands": ["ls -la"]}')) + adv = _reviewer() + resp = await _backend(_pattern_config(), exec_b, adv).call(ProxyContext(), _openai_request()) + assert adv.advise.await_count == 0 + assert resp.to_body()["choices"][0]["message"]["content"] == '{"commands": ["ls -la"]}' + + +async def test_pattern_trigger_gates_done_marker_and_redos() -> None: + exec_b = _exec_backend( + _openai_completion_resp(text='All checks pass. "task_complete": true'), + _openai_completion_resp(text='{"commands": ["pytest tests/"]}'), + ) + adv = _reviewer("REDO: the output file was never written; create it and re-verify") + config = _pattern_config(redo_feedback_prefix="REVIEWER: not done. ") + resp = await _backend(config, exec_b, adv).call(ProxyContext(), _openai_request()) + + assert adv.advise.await_count == 1 + assert exec_b.call.await_count == 2 + redo_msgs = exec_b.call.await_args_list[1].args[1].to_body()["messages"] + assert redo_msgs[-1]["content"].startswith("REVIEWER: not done. the output file") + # The client receives the continuation turn, not the premature done-claim. + assert "pytest" in resp.to_body()["choices"][0]["message"]["content"] + + +async def test_pattern_trigger_reviews_once_per_session() -> None: + exec_b = _exec_backend( + _openai_completion_resp(text='"task_complete": true'), + _openai_completion_resp(text='true'), + ) + adv = _reviewer("APPROVE") + backend = _backend(_pattern_config(), exec_b, adv) + await backend.call(ProxyContext(), _openai_request()) # gate fires + await backend.call(ProxyContext(), _openai_request()) # passthrough + assert adv.advise.await_count == 1 + + +def test_pattern_trigger_requires_pattern() -> None: + import pydantic + with pytest.raises(pydantic.ValidationError, match="gate_trigger_pattern"): + _pattern_config(gate_trigger_pattern="") + + +def test_invalid_pattern_rejected() -> None: + import pydantic + with pytest.raises(pydantic.ValidationError): + _pattern_config(gate_trigger_pattern="[unclosed") + + +# --------------------------------------------------------------------------- +# Pure helpers + advisor caller +# --------------------------------------------------------------------------- + + +def test_parse_verdict() -> None: + assert _parse_verdict("APPROVE") == ("APPROVE", "") + assert _parse_verdict("approve, looks complete")[0] == "APPROVE" + v, plan = _parse_verdict("REDO: add a guard for empty input and re-run tests") + assert v == "REDO" and "add a guard" in plan + assert _parse_verdict("hmm not sure") == ("", "") # unclear → unparseable + + +def test_session_key_stable_across_turns() -> None: + msgs = [{"role": "user", "content": "the task"}] + later = msgs + [{"role": "assistant", "content": "..."}, {"role": "user", "content": "tool result"}] + assert _session_key("sys", msgs) == _session_key("sys", later) + assert _session_key("sys", msgs) != _session_key("sys", [{"role": "user", "content": "DIFFERENT"}]) + assert _session_key("sys", msgs) != _session_key("OTHER sys", msgs) # system is part of the key + + +def test_session_key_ignores_volatile_system_tail() -> None: + """A harness that re-renders volatile system context must not reset the budget. + + Claude Code rewrites reminders / todo state / environment into the system + prompt on every request. Hashing all of it minted a fresh session key per + turn, silently resetting ``max_reviews`` — observed as 87 advisor reviews on + one Terminal-Bench task instead of the configured 2. Only the client's + cache_control-marked prefix is stable by construction, so only it counts. + """ + msgs = [{"role": "user", "content": "the task"}] + stable = {"type": "text", "text": "you are an agent", "cache_control": {"type": "ephemeral"}} + turn1 = [stable, {"type": "text", "text": "todo: 1 of 5"}] + turn2 = [stable, {"type": "text", "text": "todo: 4 of 5"}] + + assert _session_key(turn1, msgs) == _session_key(turn2, msgs) + + # A genuinely different cached prefix is still a different session. + other = {"type": "text", "text": "you are something else", "cache_control": {"type": "ephemeral"}} + assert _session_key(turn1, msgs) != _session_key([other], msgs) + + # ...and a different task under the same prefix is still a different session. + assert _session_key(turn1, msgs) != _session_key(turn1, [{"role": "user", "content": "other task"}]) + + +def test_session_key_without_breakpoint_falls_back_to_first_user_message() -> None: + """No cache_control breakpoint -> identity rests on the stable task statement.""" + msgs = [{"role": "user", "content": "the task"}] + a = [{"type": "text", "text": "volatile A"}] + b = [{"type": "text", "text": "volatile B"}] + assert _session_key(a, msgs) == _session_key(b, msgs) + assert _session_key(a, msgs) != _session_key(a, [{"role": "user", "content": "other task"}]) + + +def test_build_advisor_caller_is_anthropic() -> None: + assert isinstance(_build_advisor_caller(_config()), _AnthropicAdvisorCaller) + + +def test_build_advisor_caller_dispatches_openai() -> None: + config = _config(advisor={"model": "deepseek/deepseek-r2", "base_url": "http://adv.test", + "api_key": "k", "format": "openai"}) + assert isinstance(_build_advisor_caller(config), _OpenAiAdvisorCaller) + + +@respx.mock +async def test_openai_advisor_hits_chat_completions_endpoint() -> None: + route = respx.post("https://adv.example/v1/chat/completions").mock( + return_value=httpx.Response(200, json={ + "id": "chatcmpl-x", "object": "chat.completion", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": " use a heap "}}], + "usage": {"prompt_tokens": 42, "completion_tokens": 6}, + }) + ) + target = coerce_llm_target({ + "model": "deepseek/deepseek-r2", "base_url": "https://adv.example/v1", + "api_key": "secret-key", "format": "openai", + "extra_headers": {"X-Gateway": "test"}, + "extra_body": {"chat_template_kwargs": {"enable_thinking": False}}, + }, default_id="advisor") + caller = _OpenAiAdvisorCaller(target=target, max_tokens=256, temperature=None) + + text, usage = await caller.advise(system="advise this", transcript="conversation") + + assert text == "use a heap" + prompt_tokens, completion_tokens = _usage_tokens(usage) + assert (prompt_tokens, completion_tokens) == (42, 6) + request = route.calls.last.request + assert request.headers["authorization"] == "Bearer secret-key" + assert request.headers["x-gateway"] == "test" + body = json.loads(request.content) + assert body["messages"][0] == {"role": "system", "content": "advise this"} + assert body["messages"][1] == {"role": "user", "content": "conversation"} + assert body["max_tokens"] == 256 + assert "temperature" not in body + assert body["chat_template_kwargs"] == {"enable_thinking": False} + + +@respx.mock +async def test_anthropic_reviewer_hits_messages_endpoint() -> None: + route = respx.post("https://inference-api.nvidia.com/v1/messages").mock( + return_value=httpx.Response(200, json={ + "content": [{"type": "text", "text": "APPROVE"}], + "usage": {"input_tokens": 10, "output_tokens": 2}, + }) + ) + caller = _AnthropicAdvisorCaller( + api_key="secret-key", base_url="https://inference-api.nvidia.com/v1", + model="aws/anthropic/bedrock-claude-opus-4-8", max_tokens=256, + temperature=None, timeout=5.0, + ) + text, usage = await caller.advise(system="review this", transcript="conversation") + assert text == "APPROVE" + assert usage["input_tokens"] == 10 + request = route.calls.last.request + assert request.headers["authorization"] == "Bearer secret-key" + body = json.loads(request.content) + assert body["system"] == "review this" + assert "temperature" not in body + + +def test_messages_url_resolution() -> None: + base = "https://inference-api.nvidia.com" + assert _messages_url(f"{base}/v1") == f"{base}/v1/messages" + assert _messages_url(f"{base}/v1/messages") == f"{base}/v1/messages" + assert _messages_url(base) == f"{base}/v1/messages" + + +def test_anthropic_text_joins_text_blocks() -> None: + data = {"content": [{"type": "text", "text": "RE"}, {"type": "thinking", "text": "x"}, + {"type": "text", "text": "DO: fix it"}]} + assert _anthropic_text(data) == "REDO: fix it" + + +# --------------------------------------------------------------------------- +# seed_plan_advice +# --------------------------------------------------------------------------- + + +async def test_seed_consults_once_and_injects_into_first_user_message() -> None: + """The session-opening request triggers one seed consult (advisor prompt, + not the reviewer prompt) and the advice lands in the first user message.""" + exec_b = _exec_backend(_completion_resp(text="working", tool_use=True)) + adv = _reviewer("1. plan the schema first") + config = _config(seed_plan_advice=True) + await _backend(config, exec_b, adv).call(ProxyContext(), _request()) + assert adv.advise.await_count == 1 + assert adv.advise.await_args.kwargs["system"] == config.advisor_system_prompt + sent = exec_b.call.await_args.args[1].to_body() + user = sent["messages"][0] + assert user["role"] == "user" + assert config.seed_advice_prefix.strip() in user["content"] + assert "1. plan the schema first" in user["content"] + + +async def test_seed_reinjected_on_later_turns_without_reconsult() -> None: + """Later turns of the session re-inject the cached advice — one consult total.""" + exec_b = _exec_backend( + _completion_resp(text="working", tool_use=True), + _completion_resp(text="still working", tool_use=True), + ) + adv = _reviewer("1. plan") + backend = _backend(_config(seed_plan_advice=True), exec_b, adv) + await backend.call(ProxyContext(), _request()) + await backend.call(ProxyContext(), _request(messages=[ + {"role": "user", "content": "build X"}, + {"role": "assistant", "content": "working"}, + {"role": "user", "content": "output: ok"}, + ])) + assert adv.advise.await_count == 1 + second = exec_b.call.await_args_list[1].args[1].to_body() + assert "1. plan" in second["messages"][0]["content"] + + +async def test_seed_skipped_for_session_first_seen_mid_conversation() -> None: + """A session whose first observed request already has assistant turns is + never seeded (injecting new advice mid-session would shift the prefix).""" + exec_b = _exec_backend(_completion_resp(text="w", tool_use=True)) + adv = _reviewer() + await _backend(_config(seed_plan_advice=True), exec_b, adv).call( + ProxyContext(), + _request(messages=[ + {"role": "user", "content": "build X"}, + {"role": "assistant", "content": "already going"}, + {"role": "user", "content": "go on"}, + ]), + ) + assert adv.advise.await_count == 0 + + +async def test_seed_fail_open_leaves_session_unseeded_without_retry() -> None: + """A failed seed consult proceeds unseeded and is cached — no retry storm.""" + exec_b = _exec_backend( + _completion_resp(text="w", tool_use=True), + _completion_resp(text="w2", tool_use=True), + ) + adv = _failing_reviewer(RuntimeError("advisor down")) + backend = _backend(_config(seed_plan_advice=True), exec_b, adv) + await backend.call(ProxyContext(), _request()) + sent = exec_b.call.await_args_list[0].args[1].to_body() + assert "advisor reviewed" not in json.dumps(sent["messages"]) + await backend.call(ProxyContext(), _request()) + assert adv.advise.await_count == 1 + + +async def test_seed_composes_with_pattern_gate() -> None: + """Seed at session start + pattern gate at the done-claim, in one route: + first consult uses the advisor prompt, second the reviewer prompt, and the + REDO continuation still carries the seeded first user message.""" + exec_b = _exec_backend( + _openai_completion_resp(text='{"task_complete": true}'), + _openai_completion_resp(text="resumed work"), + ) + adv = MagicMock() + adv.advise = AsyncMock(side_effect=[("1. seeded plan", None), ("REDO fix X", None)]) + config = _pattern_config(seed_plan_advice=True) + resp = await _backend(config, exec_b, adv).call(ProxyContext(), _openai_request()) + assert adv.advise.await_count == 2 + assert adv.advise.await_args_list[0].kwargs["system"] == config.advisor_system_prompt + assert adv.advise.await_args_list[1].kwargs["system"] == config.reviewer_system_prompt + body = resp.to_body() + assert body["choices"][0]["message"]["content"] == "resumed work" + redo = exec_b.call.await_args_list[1].args[1].to_body() + assert "1. seeded plan" in redo["messages"][1]["content"] + assert redo["messages"][-1]["content"].startswith(config.redo_feedback_prefix) + + +# --------------------------------------------------------------------------- +# Review budget (max_reviews), stall checkpoint, and min-tool-results guard +# --------------------------------------------------------------------------- + + +async def test_max_reviews_budget_reviews_redeclared_completion() -> None: + """With max_reviews=2, a completion re-declared after a REDO is reviewed + again (sequential best-of-3 with the advisor as judge); the budget then + exhausts and later turns pass through.""" + exec_b = _exec_backend( + _openai_completion_resp(text='{"task_complete": true}'), # req1: reviewed -> REDO + _openai_completion_resp(text="continuing"), # req1: redo continuation + _openai_completion_resp(text='{"task_complete": true}'), # req2: reviewed -> APPROVE + _openai_completion_resp(text='{"task_complete": true}'), # req3: passthrough + ) + adv = MagicMock() + adv.advise = AsyncMock(side_effect=[("REDO not done", None), ("APPROVE", None)]) + config = _pattern_config(max_reviews=2) + backend = _backend(config, exec_b, adv) + later = [{"role": "user", "content": "build X"}, + {"role": "assistant", "content": "working"}, + {"role": "user", "content": "output"}] + await backend.call(ProxyContext(), _openai_request()) + await backend.call(ProxyContext(), _openai_request(messages=list(later))) + await backend.call(ProxyContext(), _openai_request(messages=list(later))) + assert adv.advise.await_count == 2 # budget spent; third declaration unreviewed + + +async def test_default_budget_keeps_once_per_session() -> None: + exec_b = _exec_backend( + _openai_completion_resp(text='{"task_complete": true}'), + _openai_completion_resp(text='{"task_complete": true}'), + ) + adv = _reviewer("APPROVE") + backend = _backend(_pattern_config(), exec_b, adv) + await backend.call(ProxyContext(), _openai_request()) + await backend.call(ProxyContext(), _openai_request()) + assert adv.advise.await_count == 1 + + +async def test_stall_checkpoint_reviews_mid_task_once() -> None: + """gate_stall_turns fires a mid-task review when the conversation has + grown past the threshold without the pattern ever matching — and only + once per session.""" + exec_b = _exec_backend( + _openai_completion_resp(text='{"commands": ["make"]}'), # stall review -> REDO + _openai_completion_resp(text="course-corrected"), # redo continuation + _openai_completion_resp(text='{"commands": ["ls"]}'), # no re-fire (stall spent) + ) + adv = MagicMock() + adv.advise = AsyncMock(side_effect=[("REDO try approach B", None)]) + config = _pattern_config(max_reviews=2, gate_stall_turns=3) + backend = _backend(config, exec_b, adv) + long_history = [{"role": "user", "content": "build X"}] + for i in range(3): + long_history += [{"role": "assistant", "content": f"cmd {i}"}, + {"role": "user", "content": f"out {i}"}] + await backend.call(ProxyContext(), _openai_request(messages=list(long_history))) + assert adv.advise.await_count == 1 + await backend.call(ProxyContext(), _openai_request(messages=list(long_history))) + assert adv.advise.await_count == 1 # stall fired once; budget remains for the marker + + +async def test_stall_disabled_by_default() -> None: + exec_b = _exec_backend(_openai_completion_resp(text='{"commands": ["make"]}')) + adv = _reviewer() + long_history = [{"role": "user", "content": "build X"}] + for i in range(10): + long_history += [{"role": "assistant", "content": f"c{i}"}, + {"role": "user", "content": f"o{i}"}] + await _backend(_pattern_config(), exec_b, adv).call( + ProxyContext(), _openai_request(messages=long_history)) + assert adv.advise.await_count == 0 + + +async def test_min_tool_results_skips_early_commentary() -> None: + """no_tool_call trigger with gate_min_tool_results: a text-only turn + before any tool results is NOT reviewed; one after enough results is.""" + exec_b = _exec_backend( + _completion_resp(text="here is my plan"), # 0 tool results -> no review + _completion_resp(text="all done"), # 2 tool results -> reviewed + ) + adv = _reviewer("APPROVE") + config = _config(gate_min_tool_results=2) + backend = _backend(config, exec_b, adv) + await backend.call(ProxyContext(), _request()) + assert adv.advise.await_count == 0 + worked = [ + {"role": "user", "content": "build X"}, + {"role": "assistant", "content": [ + {"type": "tool_use", "id": "t1", "name": "bash", "input": {}}]}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t1", "content": "ok"}]}, + {"role": "assistant", "content": [ + {"type": "tool_use", "id": "t2", "name": "bash", "input": {}}]}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "t2", "content": "ok"}]}, + ] + await backend.call(ProxyContext(), _request(messages=worked)) + assert adv.advise.await_count == 1 + + +# --------------------------------------------------------------------------- +# Budget scoping (proxy_x_session_id) and consult failures +# --------------------------------------------------------------------------- + + +def _ctx(session_id: str | None = None) -> ProxyContext: + """A ProxyContext optionally carrying the caller's session header identity.""" + ctx = ProxyContext() + if session_id: + ctx.metadata[CTX_REQUEST_METADATA] = RequestMetadata(session_id=session_id) + return ctx + + +async def test_budget_is_per_client_session_scope() -> None: + """Each proxy_x_session_id gets its own max_reviews budget on one instance. + + On a gateway shared by many tasks (benchmark campaign topology), an + instance-wide cap would bound reviews for the whole run; the header scope + is what makes max_reviews mean "reviews for this task". + """ + exec_b = _exec_backend(*[_completion_resp(text=f"done{i}") for i in range(4)]) + adv = _reviewer(*["APPROVE"] * 4) + backend = _backend(_config(max_reviews=1), exec_b, adv) + await backend.call(_ctx("ev_a"), _request()) + await backend.call(_ctx("ev_b"), _request()) # different task -> own budget + assert adv.advise.await_count == 2 + await backend.call(_ctx("ev_a"), _request()) # ev_a spent -> passthrough + assert adv.advise.await_count == 2 + + +async def test_no_session_header_falls_back_to_instance_scope() -> None: + """Callers without the header share one instance-wide budget (old behavior).""" + exec_b = _exec_backend(*[_completion_resp(text=f"done{i}") for i in range(3)]) + adv = _reviewer(*["APPROVE"] * 3) + backend = _backend(_config(max_reviews=1), exec_b, adv) + for i in range(3): + await backend.call( + ProxyContext(), + _request(messages=[{"role": "user", "content": f"distinct task {i}"}]), + ) + assert adv.advise.await_count == 1 + + +async def test_failed_consult_refunds_budget() -> None: + """A fail-open advisor error is not a review — the budget must survive it.""" + exec_b = _exec_backend( + _completion_resp(text="done"), _completion_resp(text="done again"), + ) + adv = MagicMock() + adv.advise = AsyncMock(side_effect=[RuntimeError("blip"), ("APPROVE", None)]) + backend = _backend(_config(max_reviews=1), exec_b, adv) + await backend.call(_ctx("ev_a"), _request()) # consult fails; budget refunded + await backend.call(_ctx("ev_a"), _request()) # the real review still happens + assert adv.advise.await_count == 2 + + +async def test_repeated_consult_failures_stop_attempts() -> None: + """A down advisor stops being consulted after the per-scope failure cap.""" + exec_b = _exec_backend(*[_completion_resp(text=f"d{i}") for i in range(5)]) + adv = _failing_reviewer(RuntimeError("advisor down")) + backend = _backend(_config(max_reviews=1), exec_b, adv) + for _ in range(5): + await backend.call(_ctx("ev_a"), _request()) + assert adv.advise.await_count == _MAX_FAILED_CONSULTS_PER_SCOPE + + +def test_advisor_usage_folds_gateway_cache_buckets() -> None: + """Hub bedrock routes auto-cache server-side: real input lands in cache_creation.""" + hub = {"input_tokens": 2, "output_tokens": 7, + "cache_creation_input_tokens": 3031, "cache_read_input_tokens": 0} + assert _advisor_usage(hub) == { + "prompt_tokens": 3033, "cached_tokens": 0, + "cache_creation_tokens": 3031, "completion_tokens": 7, + } + warm = {"input_tokens": 14, "output_tokens": 4, + "cache_creation_input_tokens": 0, "cache_read_input_tokens": 15403} + tokens = _advisor_usage(warm) + assert tokens["prompt_tokens"] == 15417 + assert tokens["cached_tokens"] == 15403 + openai_shape = {"prompt_tokens": 900, "completion_tokens": 30, + "prompt_tokens_details": {"cached_tokens": 512}} + assert _advisor_usage(openai_shape) == { + "prompt_tokens": 900, "cached_tokens": 512, + "cache_creation_tokens": 0, "completion_tokens": 30, + } + + +def test_parse_verdict_tolerates_wrappers() -> None: + """Markdown/label wrappers parse; prose before the verdict does not.""" + assert _parse_verdict("**REDO** add error handling")[0] == "REDO" + assert _parse_verdict("Verdict: REDO — cover the empty-input case")[0] == "REDO" + verdict, plan = _parse_verdict("redo:\n- fix the regex") + assert verdict == "REDO" + assert "fix the regex" in plan + assert _parse_verdict("**APPROVE**") == ("APPROVE", "") + # Anything without a LEADING verdict is unparseable — not a silent APPROVE, + # and crucially not an inverted one ("cannot approve ... REDO" used to + # parse as APPROVE under the first-match window). + assert _parse_verdict("") == ("", "") + assert _parse_verdict("x" * 100 + " REDO") == ("", "") + assert _parse_verdict("I cannot approve this — REDO: run the tests") == ("", "") + + +async def test_unparseable_reply_refunds_budget() -> None: + """A hedged/malformed reviewer reply must not burn max_reviews.""" + exec_b = _exec_backend( + _completion_resp(text="done"), _completion_resp(text="done again"), + ) + adv = _reviewer("The transcript is truncated; I cannot evaluate this.", "APPROVE") + backend = _backend(_config(max_reviews=1), exec_b, adv) + resp = await backend.call(_ctx("ev_a"), _request()) + assert _anthropic_text(resp.to_body()) == "done" # turn passed through + await backend.call(_ctx("ev_a"), _request()) # real review still happens + assert adv.advise.await_count == 2 + + +async def test_reasoning_only_turn_reviews_reasoning_and_echoes_it_on_redo() -> None: + """A reasoning-only executor turn is reviewable and its redo echo is non-empty.""" + reasoning = "I believe the task is finished because all files were written." + exec_b = _exec_backend( + ChatResponse.openai_completion({ + "id": "chatcmpl-1", "object": "chat.completion", "model": "exec-model", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": None, + "reasoning_content": reasoning}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 3}, + }), + ChatResponse.openai_completion({ + "id": "chatcmpl-2", "object": "chat.completion", "model": "exec-model", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": "resumed"}}], + "usage": {"prompt_tokens": 12, "completion_tokens": 2}, + }), + ) + adv = _reviewer("REDO state your results visibly and verify them") + config = _config(executor={ + "model": "exec-model", "base_url": "http://exec.test", "api_key": "k", + "format": "openai", + }) + backend = _backend(config, exec_b, adv) + await backend.call( + _ctx("ev_a"), + ChatRequest.openai_chat({ # type: ignore[arg-type] + "model": "incoming", + "messages": [{"role": "user", "content": "build X"}], + }), + ) + transcript = adv.advise.await_args.kwargs["transcript"] + assert "internal reasoning follows" in transcript + assert reasoning in transcript + redo_body = exec_b.call.await_args_list[1].args[1].to_body() + assert redo_body["messages"][-2] == {"role": "assistant", "content": reasoning} + assert redo_body["messages"][-1]["content"].endswith( + "state your results visibly and verify them" + ) + + +def test_transcript_truncation_keeps_task_head_and_recent_tail() -> None: + """Over the cap, the middle drops — the task and the newest work survive.""" + backend = _backend(_config(transcript_max_chars=400), _exec_backend(), _reviewer()) + messages = [{"role": "user", "content": "TASK-STATEMENT"}] + messages += [{"role": "assistant", "content": f"turn-{i} " + "x" * 40} for i in range(30)] + messages += [{"role": "assistant", "content": "NEWEST-EVIDENCE"}] + transcript = backend._serialize_transcript(messages, "done") + assert "TASK-STATEMENT" in transcript + assert "NEWEST-EVIDENCE" in transcript + assert "" in transcript + + +async def test_redo_records_advisor_and_discarded_turn_in_routing_log(tmp_path) -> None: + """Advisor consults and the REDO-discarded executor turn reach session-stats.""" + from switchyard.lib.processors.routing_log_response_processor import ( + RoutingLogResponseProcessor, + register_routing_log_sink, + ) + + log_file = tmp_path / "routing.jsonl" + register_routing_log_sink(RoutingLogResponseProcessor(log_file)) + try: + exec_b = _exec_backend( + _completion_resp(text="done"), _completion_resp(text="kept"), + ) + adv = MagicMock() + adv.advise = AsyncMock( + return_value=("REDO fix it", {"input_tokens": 7, "output_tokens": 2}), + ) + backend = _backend(_config(max_reviews=1), exec_b, adv) + await backend.call(_ctx("ev_a"), _request()) + records = [json.loads(line) for line in log_file.read_text().splitlines()] + by_tier = {r["tier"]: r for r in records} + review = by_tier["advisor_review"] + assert review["model"] == "adv-model" + assert review["session_id"] == "ev_a" + assert review["prompt_tokens"] == 7 + assert review["completion_tokens"] == 2 + discarded = by_tier["review_gate_discarded"] + assert discarded["model"] == "exec-model" + assert discarded["session_id"] == "ev_a" + assert discarded["prompt_tokens"] == 10 # _completion_resp usage + assert discarded["completion_tokens"] == 3 + snapshot = RoutingLogResponseProcessor(log_file).snapshot_session("ev_a") + assert snapshot is not None + assert set(snapshot["models"]) == {"adv-model", "exec-model"} + finally: + register_routing_log_sink(None) diff --git a/tests/test_reasoning_effort_normalizer.py b/tests/test_reasoning_effort_normalizer.py new file mode 100644 index 000000000..bb9370898 --- /dev/null +++ b/tests/test_reasoning_effort_normalizer.py @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Unit tests for ``ReasoningEffortNormalizer``.""" + +from __future__ import annotations + +import pytest + +from switchyard.lib.processors.reasoning_effort_normalizer import ( + ReasoningEffortNormalizer, +) +from switchyard.lib.proxy_context import ProxyContext +from switchyard_rust.core import ChatRequest + + +def _make_request(reasoning_effort: str | None) -> ChatRequest: + body: dict = { + "model": "azure/anthropic/claude-opus-4-6", + "messages": [{"role": "user", "content": "hi"}], + } + if reasoning_effort is not None: + body["reasoning_effort"] = reasoning_effort + return ChatRequest.openai_chat(body) + + +@pytest.mark.parametrize("effort", ["low", "medium", "high", "max"]) +async def test_valid_effort_passes_through(effort: str) -> None: + req = _make_request(effort) + out = await ReasoningEffortNormalizer().process(ProxyContext(), req) + assert out.body["reasoning_effort"] == effort + + +async def test_xhigh_alias_maps_to_high() -> None: + req = _make_request("xhigh") + out = await ReasoningEffortNormalizer().process(ProxyContext(), req) + assert out.body["reasoning_effort"] == "high" + + +async def test_unknown_effort_falls_back_to_high() -> None: + req = _make_request("super-mega") + out = await ReasoningEffortNormalizer().process(ProxyContext(), req) + assert out.body["reasoning_effort"] == "high" + + +async def test_absent_effort_is_noop() -> None: + req = _make_request(None) + out = await ReasoningEffortNormalizer().process(ProxyContext(), req) + assert "reasoning_effort" not in out.body diff --git a/tests/test_route_bundle.py b/tests/test_route_bundle.py index d8773c273..39a79fc26 100644 --- a/tests/test_route_bundle.py +++ b/tests/test_route_bundle.py @@ -14,9 +14,13 @@ import switchyard.cli.switchyard_cli as cli from switchyard.cli.launchers.launcher_runtime import route_bundle_strategy_summary from switchyard.cli.route_bundle import RouteBundleConfigError, build_route_bundle_table +from switchyard.lib.backends.advisor_loop_backend import AdvisorLoopBackend +from switchyard.lib.processors.reasoning_effort_normalizer import ReasoningEffortNormalizer from switchyard.lib.route_table import RouteTable +from switchyard.lib.stats_accumulator import StatsAccumulator from switchyard.server.switchyard_app import build_switchyard_app from switchyard_rust.components import StatsLlmBackend +from switchyard_rust.core import ChatRequestType async def test_noop_route_returns_ok_without_an_upstream() -> None: @@ -120,7 +124,7 @@ def test_passthrough_summary_labels_the_model(tmp_path: Path) -> None: ({"routes": {"r": {}}}, "missing string 'type'"), ( {"routes": {"r": {"type": "random"}}}, - "expected 'noop' or 'passthrough'", + "expected 'noop', 'passthrough', or 'advisor'", ), ( {"routes": {"r": {"type": "noop", "target": "unused"}}}, @@ -174,3 +178,136 @@ def fake_serve(args: object, switchyard: object, **kwargs: object) -> None: assert isinstance(captured["switchyard"], RouteTable) assert captured["switchyard"].registered_models() == ["test/noop"] + + +def _advisor_bundle() -> dict[str, object]: + return { + "routes": { + "myrouter/advisor": { + "type": "advisor", + "display_name": "Advisor gate", + "executor": { + "model": "aws/anthropic/bedrock-claude-opus-4-7", + "api_key": "", + "base_url": "https://exec.invalid/v1", + "format": "anthropic", + "extra_headers": {"Authorization": "Bearer sk-exec"}, + }, + "advisor": { + "model": "aws/anthropic/bedrock-claude-opus-4-8", + "api_key": "sk-adv", + "base_url": "https://adv.invalid/v1", + "format": "anthropic", + }, + }, + }, + } + + +class TestAdvisorRouteType: + """``type: advisor`` wires the executor + review-gate advisor chain via YAML.""" + + def test_registers_and_builds_loop_backend(self) -> None: + table = build_route_bundle_table(_advisor_bundle()) + assert table.registered_models() == ["myrouter/advisor"] + assert table.registered_model_entries()[0]["display_name"] == "Advisor gate" + components = table.lookup_switchyard("myrouter/advisor").iter_components() + backend = next(c for c in components if isinstance(c, AdvisorLoopBackend)) + # Anthropic executor tier advertises the Anthropic wire. + assert backend.supported_request_types == [ChatRequestType.ANTHROPIC] + # Claude Code's ``/effort xhigh`` is normalized before the executor. + assert any(isinstance(c, ReasoningEffortNormalizer) for c in components) + + def test_accumulator_injected_into_backend(self) -> None: + # The advisor backend is Python-only, so it cannot be wrapped in + # StatsLlmBackend; the bundle builder hands it the accumulator instead. + stats = StatsAccumulator() + table = build_route_bundle_table(_advisor_bundle(), stats_accumulator=stats) + backend = next( + c for c in table.iter_components() if isinstance(c, AdvisorLoopBackend) + ) + assert backend._stats is stats + assert not any(isinstance(c, StatsLlmBackend) for c in table.iter_components()) + + def test_openai_wire_with_custom_redo_prefix(self) -> None: + bundle = { + "routes": { + "myrouter/advisor-gate-oss": { + "type": "advisor", + "redo_feedback_prefix": "REVIEWER SAYS: ", + "executor": { + "model": "qwen/qwen3-max", + "api_key": "sk-exec", + "base_url": "https://exec.invalid/v1", + "format": "openai", + }, + "advisor": { + "model": "deepseek/deepseek-r2", + "api_key": "sk-adv", + "base_url": "https://adv.invalid/v1", + "format": "openai", + }, + }, + }, + } + table = build_route_bundle_table(bundle) + backend = next( + c for c in table.iter_components() if isinstance(c, AdvisorLoopBackend) + ) + assert backend.supported_request_types == [ChatRequestType.OPENAI_CHAT] + assert backend._config.redo_feedback_prefix == "REVIEWER SAYS: " + + def test_defaults_apply_to_tiers(self) -> None: + table = build_route_bundle_table({ + "defaults": { + "api_key": "sk-shared", + "base_url": "https://shared.invalid/v1", + "format": "openai", + }, + "routes": { + "adv": { + "type": "advisor", + "executor": "qwen/qwen3-max", + "advisor": "deepseek/deepseek-r2", + }, + }, + }) + backend = next( + c for c in table.iter_components() if isinstance(c, AdvisorLoopBackend) + ) + assert backend._config.executor.model == "qwen/qwen3-max" + assert backend._config.executor.endpoint.base_url == "https://shared.invalid/v1" + assert backend._config.advisor.endpoint.api_key == "sk-shared" + + def test_rejects_unknown_route_key(self) -> None: + bundle = _advisor_bundle() + bundle["routes"]["myrouter/advisor"]["bogus_field"] = 1 # type: ignore[index] + with pytest.raises(RouteBundleConfigError, match="bogus_field"): + build_route_bundle_table(bundle) + + def test_missing_tier_rejected(self) -> None: + bundle = _advisor_bundle() + del bundle["routes"]["myrouter/advisor"]["advisor"] # type: ignore[union-attr] + with pytest.raises(RouteBundleConfigError, match="advisor target is required"): + build_route_bundle_table(bundle) + + def test_responses_format_rejected(self) -> None: + bundle = _advisor_bundle() + bundle["routes"]["myrouter/advisor"]["executor"]["format"] = "responses" # type: ignore[index] + with pytest.raises(RouteBundleConfigError, match="responses"): + build_route_bundle_table(bundle) + + def test_summary_labels_the_tiers(self, tmp_path: Path) -> None: + path = tmp_path / "routes.yaml" + path.write_text( + "routes:\n" + " adv:\n" + " type: advisor\n" + " executor:\n" + " model: exec/model\n" + " advisor:\n" + " model: adv/model\n" + ) + assert route_bundle_strategy_summary(str(path), "adv") == ( + "advisor: executor=exec/model, advisor=adv/model" + )