feat(benchmark): task-level routing stats for Harbor runs

benchmark/README.md

@@ -273,6 +273,8 @@ run_manifest.json
 server.log
 harbor.log
 routing_stats_final.json
+routing_requests.jsonl
+routing_stats_by_task.json
 jobs/<job-name>/result.json
 jobs/<job-name>/<task-id>/agent/trajectory.json
 ```
@@ -284,6 +286,11 @@ deterministic LLM-classifier routes, `server.classifier_prompts` records each ro
 prompt, prompt SHA-256, `max_request_chars`, and `recent_turn_window` for reproducibility. Direct
 runs mark routing stats as `not-requested`.
 
+Switchyard runs also write `routing_requests.jsonl` (one record per router request: task, session,
+selected model, tier, token usage) and roll it up into `routing_stats_by_task.json` at finalize.
+The task identity comes from the per-task proxy sidecar, which stamps `x-switchyard-intake-task`
+and a per-trial `proxy_x_session_id` on model-bound requests.
+
 ## Docker Image Notes
 
 Baseline runs build `switchyard-baseline:local` from `benchmark/switchyard-server.Dockerfile`.

benchmark/closed_book_proxy/proxy/rewriter.py

@@ -8,6 +8,7 @@
 import json
 import os
 import re
+import uuid
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
@@ -41,6 +42,8 @@
     "/v1/messages",
     "/v1/responses",
 )
+INTAKE_TASK_HEADER = "x-switchyard-intake-task"
+SESSION_ID_HEADER = "proxy_x_session_id"
 
 
 def _truthy(value: str | None) -> bool:
@@ -63,6 +66,10 @@ def __init__(self) -> None:
             os.environ.get("SWITCHYARD_PROXY_STRIP_LOG", "/etc/proxy-public/strip.jsonl")
         )
         self.allowed_hosts = self._load_allowed_hosts()
+        self.task_id = os.environ.get("SWITCHYARD_TASK_ID", "")
+        # One id per proxy boot; the sidecar boots once per trial, so this
+        # distinguishes trials of the same task.
+        self.session_id = uuid.uuid4().hex
 
     def _load_allowed_hosts(self) -> set[str]:
         hosts: set[str] = set()
@@ -112,9 +119,16 @@ def http_connect(self, flow: http.HTTPFlow) -> None:
         self._deny_if_needed(flow)
 
     def request(self, flow: http.HTTPFlow) -> None:
-        if self._deny_if_needed(flow) or not self.closed_book:
+        if self._deny_if_needed(flow):
             return
-        if not any(flow.request.path.endswith(suffix) for suffix in STRIP_PATH_SUFFIXES):
+        # flow.request.path includes the query string (e.g. /v1/messages?beta=true).
+        path = flow.request.path.split("?", 1)[0]
+        if not any(path.endswith(suffix) for suffix in STRIP_PATH_SUFFIXES):
+            return
+        if self.task_id:
+            flow.request.headers[INTAKE_TASK_HEADER] = self.task_id
+            flow.request.headers[SESSION_ID_HEADER] = self.session_id
+        if not self.closed_book:
             return
         if "application/json" not in flow.request.headers.get("content-type", ""):
             return

benchmark/prepare_harbor_dataset.py

@@ -411,6 +411,7 @@ def _merge_compose(task_dir: Path, proxy_allowlist_hosts: tuple[str, ...]) -> di
             "CLOSED_BOOK_MODE=${CLOSED_BOOK_MODE:-1}",
             "OPENAI_BASE_URL=${OPENAI_BASE_URL:-}",
             "SWITCHYARD_BASE_URL=${SWITCHYARD_BASE_URL:-}",
+            f"SWITCHYARD_TASK_ID={task_dir.name}",
             "VERIFIER_PROXY_TOKEN=${SWITCHYARD_VERIFIER_PROXY_TOKEN:-}",
         ],
         "healthcheck": {

benchmark/run-baseline.sh

@@ -503,6 +503,12 @@ SERVER_LOG="${RUN_DIR}/server.log"
 HARBOR_LOG="${RUN_DIR}/harbor.log"
 HARBOR_RESULT_JSON="${RUN_DIR}/harbor_result.json"
 ROUTING_STATS_JSON="${RUN_DIR}/routing_stats_final.json"
+ROUTING_LOG_JSONL="${RUN_DIR}/routing_requests.jsonl"
+ROUTING_STATS_BY_TASK_JSON="${RUN_DIR}/routing_stats_by_task.json"
+if [[ "${SWITCHYARD_ENABLED}" -eq 1 ]]; then
+    SERVER_CMD+=(--routing-log-file "${ROUTING_LOG_JSONL}")
+    SERVER_DOCKER_CMD+=(--routing-log-file "${ROUTING_LOG_JSONL}")
+fi
 DOCKER_RUN_ID="$(printf '%s-%s' "${TS##*_}" "$$" | tr -c '[:alnum:]_.-' '-')"
 SWITCHYARD_DOCKER_NETWORK="${SWITCHYARD_DOCKER_NETWORK:-switchyard-${DOCKER_RUN_ID}}"
 SWITCHYARD_DOCKER_CONTAINER="${SWITCHYARD_DOCKER_CONTAINER:-switchyard-${DOCKER_RUN_ID}}"
@@ -822,6 +828,8 @@ REPO_ROOT=$(q "${REPO_ROOT}")
 SERVER_LOG=$(q "${SERVER_LOG}")
 HARBOR_LOG=$(q "${HARBOR_LOG}")
 ROUTING_STATS_JSON=$(q "${ROUTING_STATS_JSON}")
+ROUTING_LOG_JSONL=$(q "${ROUTING_LOG_JSONL}")
+ROUTING_STATS_BY_TASK_JSON=$(q "${ROUTING_STATS_BY_TASK_JSON}")
 HARBOR_JOB_DIR=$(q "${HARBOR_JOB_DIR}")
 SKIP_HEALTH_CHECK=$(q "${SKIP_HEALTH_CHECK}")
 BOOK_MODE=$(q "${BOOK_MODE}")
@@ -877,6 +885,7 @@ if [[ "\${SERVER_ENABLED}" == "1" ]]; then
         --network-alias "\${SWITCHYARD_DOCKER_SERVICE_NAME}"
         -p "127.0.0.1:$(q "${PORT}"):$(q "${PORT}")"
         -v "\${REPO_ROOT}:\${REPO_ROOT}:ro"
+        -v "\${RUN_DIR}:\${RUN_DIR}"
     )
     if [[ -n "\${ROUTING_PROFILES_DIR}" && "\${ROUTING_PROFILES_DIR}" != "\${REPO_ROOT}"* ]]; then
         DOCKER_RUN_ARGS+=(-v "\${ROUTING_PROFILES_DIR}:\${ROUTING_PROFILES_DIR}:ro")
@@ -1029,7 +1038,9 @@ FINALIZE_CMD=(python3 benchmark/run_manifest.py finalize \\
     --harbor-job-dir "\${HARBOR_JOB_DIR}")
 if [[ "\${SERVER_ENABLED}" == "1" ]]; then
     curl -fsS "\${SERVER_STATS_URL}" -o "\${ROUTING_STATS_JSON}" >/dev/null 2>&1 || true
-    FINALIZE_CMD+=(--routing-stats "\${ROUTING_STATS_JSON}")
+    FINALIZE_CMD+=(--routing-stats "\${ROUTING_STATS_JSON}"
+        --routing-log "\${ROUTING_LOG_JSONL}"
+        --routing-stats-by-task "\${ROUTING_STATS_BY_TASK_JSON}")
 fi
 "\${FINALIZE_CMD[@]}" || true
 

benchmark/run_manifest.py

@@ -223,13 +223,67 @@ def _copy_if_present(source: Path | None, dest: Path | None) -> str:
     return "present"
 
 
+def summarize_routing_log(log_path: Path) -> dict[str, Any]:
+    """Aggregate a per-request routing JSONL log into per-task stats.
+
+    Each task entry lists one bucket per (model, tier) pair so a model that
+    served more than one tier is not merged into a single row.
+    """
+    tasks: dict[str, dict[str, Any]] = {}
+    total_requests = 0
+    for raw in log_path.read_text().splitlines():
+        raw = raw.strip()
+        if not raw:
+            continue
+        try:
+            record = json.loads(raw)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(record, dict):
+            continue
+        total_requests += 1
+        task = record.get("task") or "unattributed"
+        entry = tasks.setdefault(task, {"requests": 0, "sessions": [], "_buckets": {}})
+        entry["requests"] += 1
+        session = record.get("session_id")
+        if session and session not in entry["sessions"]:
+            entry["sessions"].append(session)
+        model = record.get("model") or "unknown"
+        tier = record.get("tier") or ""
+        bucket = entry["_buckets"].setdefault(
+            (model, tier),
+            {
+                "model": model,
+                "tier": tier,
+                "calls": 0,
+                "prompt_tokens": 0,
+                "completion_tokens": 0,
+                "total_tokens": 0,
+            },
+        )
+        bucket["calls"] += 1
+        for key in ("prompt_tokens", "completion_tokens", "total_tokens"):
+            value = record.get(key)
+            bucket[key] += value if isinstance(value, int) else 0
+    for entry in tasks.values():
+        buckets = entry.pop("_buckets")
+        entry["models"] = [buckets[key] for key in sorted(buckets)]
+    return {"total_requests": total_requests, "tasks": tasks}
+
+
 def finalize_manifest(
     path: Path,
     *,
     harbor_rc: int | None,
     harbor_job_dir: Path | None = None,
     routing_stats: Path | None = None,
+    routing_log: Path | None = None,
+    routing_stats_by_task: Path | None = None,
 ) -> int:
+    """Copy run artifacts into place, roll up the routing log, and record outcomes.
+
+    Returns 0 on success, 1 when the manifest is missing.
+    """
     if not path.is_file():
         print(f"ERROR: manifest not found: {path}")
         return 1
@@ -264,6 +318,20 @@ def finalize_manifest(
                 break
         closed_book["proxy_strip_log_status"] = _copy_if_present(strip_source, strip_dest)
 
+    if routing_stats_by_task is not None:
+        status = "missing"
+        if routing_log is not None and routing_log.is_file():
+            try:
+                summary = summarize_routing_log(routing_log)
+                routing_stats_by_task.write_text(
+                    json.dumps(summary, indent=2, sort_keys=False) + "\n"
+                )
+                status = "present"
+            except OSError:
+                status = "missing"
+        outcomes["routing_stats_by_task_json"] = str(routing_stats_by_task.resolve())
+        outcomes["routing_stats_by_task_json_status"] = status
+
     outcomes["harbor_rc"] = harbor_rc
     outcomes["completed_at"] = _iso_timestamp()
     path.write_text(json.dumps(manifest, indent=2, sort_keys=False) + "\n")
@@ -341,6 +409,8 @@ def _cli_main(argv: list[str] | None = None) -> int:
     finalize.add_argument("--harbor-rc", type=int, default=None)
     finalize.add_argument("--harbor-job-dir", type=Path, default=None)
     finalize.add_argument("--routing-stats", type=Path, default=None)
+    finalize.add_argument("--routing-log", type=Path, default=None)
+    finalize.add_argument("--routing-stats-by-task", type=Path, default=None)
 
     ns = parser.parse_args(argv)
     if ns.command == "finalize":
@@ -349,6 +419,8 @@ def _cli_main(argv: list[str] | None = None) -> int:
             harbor_rc=ns.harbor_rc,
             harbor_job_dir=ns.harbor_job_dir,
             routing_stats=ns.routing_stats,
+            routing_log=ns.routing_log,
+            routing_stats_by_task=ns.routing_stats_by_task,
         )
     if ns.command != "write":
         parser.print_help()

docs/cli_reference.md

@@ -226,14 +226,15 @@ switchyard [--routing-profiles PATH] serve [--config PATH]
 | `--host`, `--port`/`-p`, `--reload` | [Transport](#transport-server-verbs). |
 | `--inbound FORMAT` | Valid only for legacy route-bundle serve (`--routing-profiles`); `serve --config` actively rejects it with an error. For legacy serve, the flag is a no-op — all request APIs are always registered regardless of the value (accepted for backwards compat only). |
 | `--workers` / `-w` | uvicorn worker count. |
+| `--routing-log-file PATH` | Append one JSONL routing record per request (task, session, selected model, tier, token usage) to PATH. Route-bundle serve only; rejected by `serve --config`. |
 | `--intake-enabled` / `--enable-intake`, `--intake-base-url`, `--intake-workspace`, `--intake-api-key`, `--intake-target-url` | [Intake sink](#intake-sink-serve-and-launchers). |
 
 **Notes**
 
 - `serve` always registers `POST /v1/chat/completions`, `POST /v1/messages`, `POST /v1/responses`, `GET /v1/models`, and `GET /health`. There is no flag to expose just one request API.
 - `GET /v1/stats` and `GET /v1/routing/stats` are available on both serve paths.
 - The deprecated route-bundle path accepts `--inbound` for compatibility but ignores it; all supported request APIs are always registered.
-- `serve --config` does not support `--reload`, `--workers > 1`, Intake options, `--enable-rl-logging`, or any explicit `--inbound` value.
+- `serve --config` does not support `--reload`, `--workers > 1`, Intake options, `--enable-rl-logging`, `--routing-log-file`, or any explicit `--inbound` value.
 - Rust-defined and Python-defined profiles use the same profile-config schema. The profile `type` decides which implementation builds that profile.
 - Python-defined profiles are registered via `@profile_config`; the shipped `header-routing` profile is an example. A config can mix a Rust-defined profile and a Python-defined profile, and both profile ids are routable on the same served host and port.
 

switchyard/cli/switchyard_cli.py

@@ -411,6 +411,12 @@ def _cmd_serve(args: argparse.Namespace) -> None:
     rl_request, rl_response = build_rl_logging_processors(resolve_rl_log_dir(args))
     request_processors = [*intake_request, *rl_request]
     response_processors = [*intake_response, *rl_response]
+    if getattr(args, "routing_log_file", None):
+        from switchyard.lib.processors.routing_log_response_processor import (
+            RoutingLogResponseProcessor,
+        )
+
+        response_processors.append(RoutingLogResponseProcessor(args.routing_log_file))
     if routing_profiles:
         table = load_route_bundle_table(
             routing_profiles,
@@ -485,6 +491,11 @@ def _cmd_serve_profile_config(args: argparse.Namespace) -> None:
             "serve --config does not support --enable-rl-logging. "
             "Use serve --routing-profiles for local RL trace logging."
         )
+    if getattr(args, "routing_log_file", None):
+        raise SystemExit(
+            "serve --config does not support --routing-log-file. "
+            "Use serve --routing-profiles for per-request routing logs."
+        )
 
     table = _profile_config_route_table(args.config)
     logger.info(
@@ -886,6 +897,16 @@ def _build_parser() -> argparse.ArgumentParser:
             "Path to a Switchyard v2 profile config (YAML, JSON, or TOML)."
         ),
     )
+    serve.add_argument(
+        "--routing-log-file",
+        dest="routing_log_file",
+        default=None,
+        metavar="PATH",
+        help=(
+            "Append one JSONL routing record per request (task, session, "
+            "selected model, tier, token usage) to PATH."
+        ),
+    )
     serve.add_argument(
         "--workers", "-w", type=int,
         default=int(os.environ.get("SWITCHYARD_WORKERS", "1")),

switchyard/lib/processors/routing_log_response_processor.py

@@ -0,0 +1,103 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Response-side processor that appends one JSONL routing record per request."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import threading
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from switchyard.lib.chat_response.streaming_response_accumulator import (
+    attach_final_response_callback,
+)
+from switchyard.lib.proxy_context import CTX_PROXY_ACTUAL_MODEL, ProxyContext
+from switchyard.lib.request_metadata import CTX_REQUEST_METADATA
+from switchyard_rust.core import ChatResponse
+
+logger = logging.getLogger(__name__)
+
+
+class RoutingLogResponseProcessor:
+    """Append one JSON line per completed request to ``log_file``.
+
+    Each record carries the routing decision (selected model and tier), the
+    caller-supplied task and session identity headers, and token usage, so a
+    benchmark harness can attribute router traffic to individual tasks.
+    Streaming responses log once the stream drains; write failures are logged
+    and never break the proxied response.
+    """
+
+    def __init__(self, log_file: Path | str) -> None:
+        self._log_file = Path(log_file)
+        self._log_file.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+
+    async def process(self, ctx: ProxyContext, response: ChatResponse) -> ChatResponse:
+        """Log one routing record for the completed request; return the response unchanged."""
+        served_model: str = ctx.selected_model or ctx.metadata.get(
+            CTX_PROXY_ACTUAL_MODEL, "unknown",
+        )
+
+        async def _emit(final: ChatResponse) -> None:
+            await asyncio.to_thread(self._write_record, ctx, served_model, final)
+
+        attached = attach_final_response_callback(
+            response, served_model=served_model, callback=_emit,
+        )
+        if not attached:
+            await _emit(response)
+        return response
+
+    def _write_record(self, ctx: ProxyContext, served_model: str, response: ChatResponse) -> None:
+        metadata = ctx.metadata.get(CTX_REQUEST_METADATA)
+        record = {
+            "ts": datetime.now(UTC).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
+            "task": getattr(getattr(metadata, "intake", None), "task", None),
+            "session_id": getattr(metadata, "session_id", None),
+            "model": served_model,
+            "tier": ctx.metadata.get("_random_routing_tier", "") or (ctx.selected_target or ""),
+            **_usage_tokens(response.body),
+        }
+        try:
+            line = json.dumps(record, separators=(",", ":"))
+            with self._lock, self._log_file.open("a", encoding="utf-8") as handle:
+                handle.write(line + "\n")
+        except OSError as exc:
+            logger.warning("Routing log: failed to append to %s: %s", self._log_file, exc)
+
+
+def _usage_tokens(body: object) -> dict[str, int]:
+    """Token counts from any native usage shape (OpenAI Chat/Responses, Anthropic)."""
+    usage = _field(body, "usage")
+    prompt = _int_field(usage, "prompt_tokens")
+    completion = _int_field(usage, "completion_tokens")
+    if not prompt and not completion:
+        prompt = (
+            _int_field(usage, "input_tokens")
+            + _int_field(usage, "cache_creation_input_tokens")
+            + _int_field(usage, "cache_read_input_tokens")
+        )
+        completion = _int_field(usage, "output_tokens")
+    return {
+        "prompt_tokens": prompt,
+        "completion_tokens": completion,
+        "total_tokens": prompt + completion,
+    }
+
+
+def _field(value: object, name: str) -> Any:
+    if isinstance(value, Mapping):
+        return value.get(name)
+    return getattr(value, name, None)
+
+
+def _int_field(value: object, name: str) -> int:
+    field = _field(value, name)
+    return field if isinstance(field, int) else 0