feat(routing): sub-agent override as a libsy classifier, restored on the serve path

.agents/skills/switchyard-lib-core/SKILL.md

@@ -39,7 +39,8 @@ right validation set. If the change is driven by a launcher need, also read
 | An OpenAI-compatible provider target such as NVIDIA Inference Hub or OpenRouter | Use the existing OpenAI-compatible backend/profile with `base_url`, `api_key`, and model id wiring. Add a new backend only when the provider has a real wire-format, auth, retry, or health contract that cannot fit that path. |
 | Direct Rust component bindings | Add concrete PyO3 classes under `crates/switchyard-py/src/component_bindings/`, keep config bindings near the component binding that consumes them, and expose them lazily from `switchyard_rust/components.py`. Do not keep growing `core_bindings.rs` or `switchyard_rust/core.py` with concrete component classes. |
 | Route YAML / model dispatch | Use `switchyard/cli/route_bundle.py` and `switchyard/lib/route_table_builders.py`. They build `RouteTable` entries from profile-backed runtimes and keep launchers plus `switchyard serve --routing-profiles` on one path. |
-| Route sub-agent traffic to a fixed worker target | Set `subagent_target: <target-id>` in any profile's common envelope (consumed like `type` in Rust `SerializedProfileConfig`, `crates/switchyard-components-v2/src/config/parsing.rs`). The Python loader (`switchyard/lib/profiles/loader.py`) wraps the built profile in `SubagentOverrideProfile`; detection is the Rust-bound `is_subagent_request(headers)` from `switchyard_rust.profiles`, a thin wrapper over `Metadata::from_headers` + `Metadata::is_subagent_work` in `crates/protocol/src/metadata.rs` (the canonical lineage fact and work-vs-maintenance policy). On the libsy `Algorithm` path, wrap with the `SubagentOverride` combinator (`crates/libsy/src/algorithms/subagent_override.rs`; Python: `switchyard.libsy.algorithms.subagent_override`). Do not re-implement header sniffing inside individual profiles or algorithms. |
+| Route sub-agent traffic to a fixed worker target | Add the `SubagentOverride` classifier (`crates/libsy/src/algorithms/util/subagent.rs`) to a `FallThrough` cascade: it scores one fixed worker target for delegated work and abstains otherwise, so later classifiers serve ordinary traffic. Detection is `Metadata::from_headers` + `Metadata::is_subagent_work` in `crates/protocol/src/metadata.rs` — the canonical lineage fact and work-vs-maintenance policy. Do not re-implement header sniffing inside individual profiles or algorithms. Rust-only today; there is no profile-envelope key and no Python binding for it. |
+| Keep a sub-agent on whatever target it was first routed to | Register `AffinityRouter` (`crates/libsy/src/algorithms/util/affinity.rs`) with `FallThrough::with_component`, which wires its `Processor` and `Classifier` roles from one `Arc` — registering the two separately is easy to half-wire, and a missing processor leaves the classifier permanently empty. Order it before the classifier that seeds the decision. Opt-in; nothing enables it by default. Distinct from the `session_affinity` / Redis pin store in the row below, which is the `switchyard-core` session cache, not a libsy classifier. |
 | Shared/persistent session-affinity pins across workers or pod churn | Configure the latency route with `session_affinity: true` + `affinity_store: redis` + `affinity_store_url` (optional `affinity_store_ttl_seconds`, `affinity_key_prefix`); the escalation_router route takes the same `affinity_store*` keys (no `session_affinity` flag — its latch is always on; default prefix `swyd:esc:`). `SessionAffinity` keeps the Rust `SessionCache` as L1 and reads/writes through the `AffinityPinStore` L2 (`switchyard/lib/redis_pin_store.py`), fail-open behind a 0.1s socket timeout and a 3-failure/10s-cooldown circuit breaker (`switchyard_affinity_l2_breaker_open` gauge). Requires the `switchyard[affinity-redis]` extra. |
 | Stats / telemetry | Reuse `StatsRequestProcessor`, `StatsResponseProcessor`, `StatsLlmBackend`, and `StatsAccumulator`. A profile config should thread one accumulator through all three when stats are enabled. Do not write a parallel collector. |
 | A fixed-path endpoint contributed by per-route components | Set `Endpoint.register_once = True`; `build_switchyard_app(...)` mounts the first instance while still running every component's lifecycle. Leave the default `False` for configurable endpoint classes that may mount distinct instances. |

benchmark/run_cost.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Per-model cost breakdown for a benchmark run's routing log.
+
+Aggregates ``routing_requests.jsonl`` by served model and prices each bucket with
+:func:`switchyard.lib.cost_estimator.estimate_model_cost`, which splits input into
+base / cache-read / cache-write tiers. That split matters: agentic runs are
+cache-dominated (90%+ of prompt tokens are cache reads is normal), so a flat
+input rate overstates cost several-fold.
+
+On a run that routes sub-agent work to a different tier, the served-model column
+doubles as the parent/child attribution key — the routing log records no
+``agent_id``, so which model answered is the only role signal available.
+
+Models absent from the price table contribute $0 and are listed separately, so an
+unpriced model reads as a gap rather than as a free one.
+
+Usage:
+    uv run python benchmark/run_cost.py --run benchmark/tb_runs/<run-name>
+"""
+
+from __future__ import annotations
+
+import argparse
+import collections
+import json
+from pathlib import Path
+
+from switchyard.lib.cost_estimator import MODEL_PRICING, estimate_model_cost
+
+_TOKEN_FIELDS = (
+    "prompt_tokens",
+    "completion_tokens",
+    "cached_tokens",
+    "cache_creation_tokens",
+)
+
+
+def _aggregate(routing_log: Path) -> dict[str, collections.Counter[str]]:
+    """Sum request counts and token fields per served model."""
+    totals: dict[str, collections.Counter[str]] = collections.defaultdict(collections.Counter)
+    for line in routing_log.read_text().splitlines():
+        if not line.strip():
+            continue
+        record = json.loads(line)
+        bucket = totals[record.get("model") or "<unknown>"]
+        bucket["reqs"] += 1
+        for field in _TOKEN_FIELDS:
+            bucket[field] += record.get(field) or 0
+    return totals
+
+
+def _mean_agg_score(run: Path) -> tuple[float, int] | None:
+    """Mean rubric ``agg_score`` across the run's scored tasks, if any."""
+    scores = [
+        json.loads(path.read_text())["agg_score"]
+        for path in run.glob("jobs/*/task-*/verifier/evaluation_results.json")
+    ]
+    return (sum(scores) / len(scores), len(scores)) if scores else None
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run", required=True, type=Path, help="Run directory under tb_runs/")
+    args = parser.parse_args()
+
+    totals = _aggregate(args.run / "routing_requests.jsonl")
+    if not totals:
+        print("no routing records found")
+        return
+
+    unpriced = [model for model in totals if model not in MODEL_PRICING]
+    total_cost = 0.0
+    header = f"{'model':46s} {'reqs':>5s} {'in':>11s} {'out':>9s} {'cached':>11s} {'USD':>9s}"
+    print(header)
+    for model, bucket in sorted(totals.items(), key=lambda item: -item[1]["reqs"]):
+        cost = estimate_model_cost(
+            model,
+            bucket["prompt_tokens"],
+            bucket["completion_tokens"],
+            bucket["cached_tokens"],
+            bucket["cache_creation_tokens"],
+        )["total_cost"]
+        total_cost += cost
+        print(
+            f"{model:46s} {bucket['reqs']:5d} {bucket['prompt_tokens']:11,} "
+            f"{bucket['completion_tokens']:9,} {bucket['cached_tokens']:11,} {cost:9.3f}"
+        )
+
+    requests = sum(bucket["reqs"] for bucket in totals.values())
+    print(f"\nTOTAL: ${total_cost:.2f} over {requests} requests")
+    if unpriced:
+        # Priced at zero by estimate_model_cost — surface it so the total is not
+        # mistaken for complete.
+        print(f"NOTE: no price entry for {', '.join(sorted(unpriced))} — excluded from the total")
+
+    scored = _mean_agg_score(args.run)
+    if scored:
+        mean, count = scored
+        print(f"mean agg_score: {mean:.3f} over {count} tasks  → ${total_cost / count:.2f}/task")
+
+
+if __name__ == "__main__":
+    main()

crates/libsy/src/algorithms.rs

@@ -10,12 +10,11 @@ pub mod fall_through;
 pub mod llm_class;
 pub mod noop;
 pub mod rand;
-pub mod subagent_override;
 
 pub use fall_through::{FallThrough, FallThroughDecision};
 pub use llm_class::{ClassifierDecision, ClassifierTier, LlmClassifier};
 pub use noop::{Noop, NoopDecision};
 pub use rand::{Random, RandomDecision};
-pub use subagent_override::{SubagentDecision, SubagentOverride};
+pub use util::{AffinityRouter, SubagentOverride};
 
-pub(crate) mod util;
+pub mod util;

crates/libsy/src/algorithms/fall_through.rs

@@ -75,6 +75,19 @@ impl FallThrough {
         self.classifiers.push(classifier);
         self
     }
+
+    /// Registers one dual-role component in *both* the processor chain and the classifier
+    /// cascade.
+    ///
+    /// A component that writes state as a [`Processor`] and reads it back as a
+    /// [`Classifier`] — such as [`AffinityRouter`](crate::algorithms::AffinityRouter) —
+    /// shares that state through the instance, so both roles must be the same `Arc`.
+    /// Registering the two separately is easy to half-wire: omit the processor and the
+    /// classifier silently never sees an assignment. This registers both at once.
+    pub fn with_component<T: Processor + Classifier + 'static>(self, component: Arc<T>) -> Self {
+        self.with_processor(component.clone())
+            .with_classifier(component)
+    }
 }
 #[async_trait]
 impl Algorithm<SharedState> for FallThrough {

crates/libsy/src/algorithms/util.rs

@@ -1,7 +1,8 @@
 // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
 
-mod affinity;
+pub mod affinity;
+pub mod subagent;
 
-#[allow(unused_imports)]
-pub(crate) use affinity::AffinityRouter;
+pub use affinity::AffinityRouter;
+pub use subagent::SubagentOverride;

crates/libsy/src/algorithms/util/affinity.rs

@@ -18,8 +18,6 @@
 //! session. [`AffinityRouter::for_subagents`] narrows affinity to explicitly identified
 //! child agents, leaving root traffic to later classifiers on every turn.
 
-#![allow(dead_code)]
-
 use std::collections::{HashMap, HashSet};
 use std::sync::Mutex;