feat(python): expose escalation mode to Python

crates/switchyard-py/src/libsy_bindings.rs

@@ -11,9 +11,9 @@ use pyo3::prelude::*;
 use serde_json::{Value, json};
 use switchyard_libsy::{
     Algorithm, ClassifierContractConfig, CustomClassifierConfig, CustomClassifierPolicy,
-    HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig, LlmFallback, LlmTarget,
-    LlmTargetSet, LlmTaskClassifier, Noop, PickerMode, Random, StageRouter, StageRouterConfig,
-    TaskClassifierConfig,
+    EscalationJudgeConfig, HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig,
+    LlmFallback, LlmTarget, LlmTargetSet, LlmTaskClassifier, Noop, PickerMode, Random, StageRouter,
+    StageRouterConfig, TaskClassifierConfig,
 };
 use switchyard_protocol::{
     AggLlmResponse, Context, Decision, LlmClientError, LlmResponse, Metadata, Request, Response,
@@ -411,6 +411,61 @@ fn custom_classifier_algorithm(
     Ok(PyAlgorithm::new(Arc::new(algorithm)))
 }
 
+/// Construct escalation routing: each efficient answer is judged, and a confirmed streak
+/// of escalate verdicts moves the session to the capable target and latches it there.
+///
+/// Sessions are identified by request metadata (e.g. a `session-id` header), which
+/// `confirmations >= 2` requires since the streak is retained per session.
+#[pyfunction(name = "escalation")]
+#[pyo3(signature = (
+    judge_target,
+    efficient_target,
+    capable_target,
+    *,
+    prompt=None,
+    confirmations=None,
+    recent_turn_window=None,
+    window_message_chars=None,
+    max_output_tokens=4096
+))]
+#[allow(clippy::too_many_arguments)]
+fn escalation_algorithm(
+    py: Python<'_>,
+    judge_target: Py<PyLlmTarget>,
+    efficient_target: Py<PyLlmTarget>,
+    capable_target: Py<PyLlmTarget>,
+    prompt: Option<String>,
+    confirmations: Option<u32>,
+    recent_turn_window: Option<usize>,
+    window_message_chars: Option<usize>,
+    max_output_tokens: u64,
+) -> PyResult<PyAlgorithm> {
+    let mut contract = ClassifierContractConfig::default();
+    if let Some(prompt) = prompt {
+        contract = contract.with_prompt(prompt);
+    }
+    let mut config = EscalationJudgeConfig::default();
+    if let Some(value) = confirmations {
+        config.confirmations = value;
+    }
+    if let Some(value) = recent_turn_window {
+        config.recent_turn_window = value;
+    }
+    if let Some(value) = window_message_chars {
+        config.window_message_chars = value;
+    }
+    let algorithm = LlmTaskClassifier::new(LlmClassifierConfig::Escalation {
+        judge_target: judge_target.bind(py).try_borrow()?.clone_core(py),
+        efficient_target: efficient_target.bind(py).try_borrow()?.clone_core(py),
+        capable_target: capable_target.bind(py).try_borrow()?.clone_core(py),
+        contract,
+        config,
+        max_output_tokens,
+    })
+    .map_err(|error| PyValueError::new_err(error.to_string()))?;
+    Ok(PyAlgorithm::new(Arc::new(algorithm)))
+}
+
 /// Construct signal-driven stage routing with an optional LLM classifier fallback.
 #[pyfunction(name = "stage_router")]
 #[pyo3(signature = (
@@ -512,7 +567,11 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
         llm_task_classifier_algorithm,
         &libsy_module
     )?)?;
-    libsy_module.add_function(wrap_pyfunction!(custom_classifier_algorithm, &libsy_module)?)?;
+    libsy_module.add_function(wrap_pyfunction!(
+        custom_classifier_algorithm,
+        &libsy_module
+    )?)?;
+    libsy_module.add_function(wrap_pyfunction!(escalation_algorithm, &libsy_module)?)?;
     libsy_module.add_function(wrap_pyfunction!(stage_router_algorithm, &libsy_module)?)?;
     libsy_module.add("LibsyError", module.getattr("LibsyError")?)?;
     module.add_submodule(&libsy_module)?;

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "maturin"
 [project]
 # Gumloop fork: dist renamed (mirrors gumloop-celery); import names are unchanged.
 name = "gumloop-nemo-switchyard"
-version = "0.2.0+gumloop.0.1.1"
+version = "0.2.0+gumloop.0.2.0"
 description = "Typed, composable LLM routing with request/response translation and multi-backend orchestration"
 readme = "README.md"
 license = "Apache-2.0"

switchyard_rust/libsy.py

@@ -19,6 +19,7 @@
         "LlmTarget",
         "TaskClassifierConfig",
         "custom_classifier",
+        "escalation",
         "llm_task_classifier",
         "noop",
         "random",
@@ -118,6 +119,18 @@ def custom_classifier(
         config: CustomClassifierConfig,
     ) -> Algorithm: ...
 
+    def escalation(
+        judge_target: LlmTarget,
+        efficient_target: LlmTarget,
+        capable_target: LlmTarget,
+        *,
+        prompt: str | None = None,
+        confirmations: int | None = None,
+        recent_turn_window: int | None = None,
+        window_message_chars: int | None = None,
+        max_output_tokens: int = 4096,
+    ) -> Algorithm: ...
+
     def stage_router(
         capable_target: LlmTarget,
         efficient_target: LlmTarget,

tests/test_libsy_escalation.py

@@ -0,0 +1,146 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for the escalation-mode Python binding.
+
+Imports go through ``switchyard_rust.libsy`` — the bindings-only surface that must
+stay importable without the ``lib`` extra's provider SDKs.
+"""
+
+import json
+from typing import Any
+
+import pytest
+
+from switchyard_rust.libsy import LlmTarget, escalation
+
+
+def request_body(text: str = "keep going") -> dict[str, Any]:
+    return {
+        "model": "auto",
+        "messages": [
+            {"role": "user", "content": [{"type": "text", "text": text}]}
+        ],
+    }
+
+
+class EchoClient:
+    def __init__(self, model: str) -> None:
+        self.model = model
+        self.calls: list[dict[str, Any]] = []
+
+    async def call(self, request: dict[str, Any]) -> dict[str, Any]:
+        self.calls.append(request)
+        return {
+            "model": self.model,
+            "outputs": [
+                {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": self.model}],
+                    "stop_reason": "end_turn",
+                }
+            ],
+        }
+
+
+class VerdictJudge(EchoClient):
+    def __init__(self, verdicts: list[bool]) -> None:
+        super().__init__("judge")
+        self.verdicts = verdicts
+
+    async def call(self, request: dict[str, Any]) -> dict[str, Any]:
+        self.calls.append(request)
+        escalate = self.verdicts.pop(0)
+        text = json.dumps({"escalate": escalate, "reason": "test verdict"})
+        return {
+            "model": self.model,
+            "outputs": [
+                {
+                    "role": "assistant",
+                    "content": [{"type": "text", "text": text}],
+                    "stop_reason": "end_turn",
+                }
+            ],
+        }
+
+
+def build(judge, efficient, capable, confirmations=1):
+    return escalation(
+        LlmTarget("judge", judge),
+        LlmTarget("efficient", efficient),
+        LlmTarget("capable", capable),
+        confirmations=confirmations,
+    )
+
+
+async def test_decline_returns_the_efficient_answer_without_a_second_call() -> None:
+    efficient, capable = EchoClient("efficient"), EchoClient("capable")
+    algorithm = build(VerdictJudge([False]), efficient, capable)
+
+    _, response = await algorithm.run(request_body())
+
+    assert response["model"] == "efficient"
+    assert len(efficient.calls) == 1
+    assert capable.calls == []
+
+
+async def test_confirmed_escalation_moves_to_the_capable_target() -> None:
+    efficient, capable = EchoClient("efficient"), EchoClient("capable")
+    algorithm = build(VerdictJudge([True]), efficient, capable, confirmations=1)
+
+    decisions, response = await algorithm.run(request_body())
+
+    assert response["model"] == "capable"
+    assert len(efficient.calls) == 1  # judged answer was produced first
+    assert len(capable.calls) == 1
+    assert decisions[-1]["selected_model"] == "capable"
+
+
+async def test_streak_below_confirmations_stays_efficient_then_latches() -> None:
+    """With confirmations=2 the first escalate verdict stays efficient; the second,
+    in the same session, latches capable."""
+    efficient, capable = EchoClient("efficient"), EchoClient("capable")
+    judge = VerdictJudge([True, True])
+    algorithm = build(judge, efficient, capable, confirmations=2)
+    headers = {"session-id": "run-42"}
+
+    _, first = await algorithm.run(request_body("step one"), headers=headers)
+    _, second = await algorithm.run(request_body("step two"), headers=headers)
+
+    assert first["model"] == "efficient"
+    assert second["model"] == "capable"
+    # A latched session stops consulting the judge entirely.
+    _, third = await algorithm.run(request_body("step three"), headers=headers)
+    assert third["model"] == "capable"
+    assert len(judge.calls) == 2
+
+
+async def test_streak_does_not_cross_sessions() -> None:
+    efficient, capable = EchoClient("efficient"), EchoClient("capable")
+    judge = VerdictJudge([True, True])
+    algorithm = build(judge, efficient, capable, confirmations=2)
+
+    _, first = await algorithm.run(request_body(), headers={"session-id": "run-a"})
+    _, second = await algorithm.run(request_body(), headers={"session-id": "run-b"})
+
+    assert first["model"] == "efficient"
+    assert second["model"] == "efficient"
+
+
+async def test_judge_failure_stays_on_the_efficient_answer() -> None:
+    class FailingJudge:
+        async def call(self, request: dict[str, Any]) -> dict[str, Any]:
+            raise RuntimeError("judge unavailable")
+
+    efficient, capable = EchoClient("efficient"), EchoClient("capable")
+    algorithm = build(FailingJudge(), efficient, capable)
+
+    _, response = await algorithm.run(request_body())
+
+    assert response["model"] == "efficient"
+    assert capable.calls == []
+
+
+def test_zero_confirmations_is_rejected() -> None:
+    with pytest.raises(ValueError, match="confirmations"):
+        build(EchoClient("judge"), EchoClient("efficient"), EchoClient("capable"), confirmations=0)