From 4a782f926572288e0668e7ec13dbb4f9a502219b Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Wed, 22 Jul 2026 14:37:55 -0500 Subject: [PATCH 1/5] feat(serve): add --routing-log-file per-request JSONL routing log Signed-off-by: Ryan Lempka --- docs/cli_reference.md | 3 +- switchyard/cli/switchyard_cli.py | 21 ++ .../routing_log_response_processor.py | 101 ++++++++++ tests/test_routing_log_response_processor.py | 187 ++++++++++++++++++ 4 files changed, 311 insertions(+), 1 deletion(-) create mode 100644 switchyard/lib/processors/routing_log_response_processor.py create mode 100644 tests/test_routing_log_response_processor.py diff --git a/docs/cli_reference.md b/docs/cli_reference.md index 4c3b98f4..e02fc8d1 100644 --- a/docs/cli_reference.md +++ b/docs/cli_reference.md @@ -226,6 +226,7 @@ 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** @@ -233,7 +234,7 @@ switchyard [--routing-profiles PATH] serve [--config PATH] - `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. diff --git a/switchyard/cli/switchyard_cli.py b/switchyard/cli/switchyard_cli.py index a7fb6ebc..bbb5576d 100644 --- a/switchyard/cli/switchyard_cli.py +++ b/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")), diff --git a/switchyard/lib/processors/routing_log_response_processor.py b/switchyard/lib/processors/routing_log_response_processor.py new file mode 100644 index 00000000..e840c435 --- /dev/null +++ b/switchyard/lib/processors/routing_log_response_processor.py @@ -0,0 +1,101 @@ +# 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 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: + served_model: str = ctx.selected_model or ctx.metadata.get( + CTX_PROXY_ACTUAL_MODEL, "unknown", + ) + + async def _emit(final: ChatResponse) -> None: + 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 diff --git a/tests/test_routing_log_response_processor.py b/tests/test_routing_log_response_processor.py new file mode 100644 index 00000000..c404a84f --- /dev/null +++ b/tests/test_routing_log_response_processor.py @@ -0,0 +1,187 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the per-request JSONL routing log (`serve --routing-log-file`).""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator +from pathlib import Path + +from openai.types.chat import ChatCompletionChunk +from openai.types.chat.chat_completion_chunk import Choice as ChunkChoice +from openai.types.chat.chat_completion_chunk import ChoiceDelta +from openai.types.completion_usage import CompletionUsage + +from switchyard.cli.switchyard_cli import _build_parser +from switchyard.lib.chat_response import ResponseStream +from switchyard.lib.processors.routing_log_response_processor import ( + RoutingLogResponseProcessor, +) +from switchyard.lib.proxy_context import CTX_PROXY_ACTUAL_MODEL +from switchyard.lib.request_metadata import attach_request_metadata +from switchyard_rust.components import RequestMetadata +from switchyard_rust.core import ChatResponse, ProxyContext + +TASK_HEADERS = { + "x-switchyard-intake-task": "hello-world-abc1", + "proxy_x_session_id": "trial-session-1", +} + + +def _ctx(*, headers: dict[str, str] | None = None, model: str = "gpt-test") -> ProxyContext: + ctx = ProxyContext() + if headers is not None: + attach_request_metadata(ctx, RequestMetadata.from_headers(headers), headers) + ctx.metadata[CTX_PROXY_ACTUAL_MODEL] = model + return ctx + + +def _openai_completion() -> ChatResponse: + return ChatResponse.openai_completion({ + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1700000000, + "model": "gpt-test", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "hi"}, + "finish_reason": "stop", + }], + "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}, + }) + + +def _anthropic_completion() -> ChatResponse: + return ChatResponse.anthropic_completion({ + "id": "msg_test", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "hi"}], + "model": "claude-test", + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 7, + "output_tokens": 3, + "cache_creation_input_tokens": 2, + "cache_read_input_tokens": 4, + }, + }) + + +def _responses_completion() -> ChatResponse: + return ChatResponse.openai_responses_completion({ + "id": "resp_test", + "object": "response", + "created_at": 1700000000, + "status": "completed", + "model": "codex-test", + "output": [{ + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "hi"}], + }], + "parallel_tool_calls": False, + "tool_choice": "auto", + "tools": [], + "usage": {"input_tokens": 6, "output_tokens": 2, "total_tokens": 8}, + }) + + +def _records(log_file: Path) -> list[dict]: + return [json.loads(line) for line in log_file.read_text().splitlines()] + + +async def test_openai_completion_record_carries_task_and_usage(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + ctx = _ctx(headers=TASK_HEADERS) + ctx.metadata["_random_routing_tier"] = "weak" + await RoutingLogResponseProcessor(log_file).process(ctx, _openai_completion()) + + (record,) = _records(log_file) + assert record["task"] == "hello-world-abc1" + assert record["session_id"] == "trial-session-1" + assert record["model"] == "gpt-test" + assert record["tier"] == "weak" + assert record["prompt_tokens"] == 5 + assert record["completion_tokens"] == 3 + assert record["total_tokens"] == 8 + + +async def test_anthropic_usage_sums_cache_siblings(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + await RoutingLogResponseProcessor(log_file).process( + _ctx(headers=TASK_HEADERS, model="claude-test"), _anthropic_completion(), + ) + + (record,) = _records(log_file) + assert record["prompt_tokens"] == 13 # input + cache_creation + cache_read + assert record["completion_tokens"] == 3 + + +async def test_responses_completion_uses_input_output_tokens(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + await RoutingLogResponseProcessor(log_file).process( + _ctx(headers=TASK_HEADERS, model="codex-test"), _responses_completion(), + ) + + (record,) = _records(log_file) + assert record["prompt_tokens"] == 6 + assert record["completion_tokens"] == 2 + + +async def test_missing_headers_log_null_task_and_session(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + await RoutingLogResponseProcessor(log_file).process(_ctx(), _openai_completion()) + + (record,) = _records(log_file) + assert record["task"] is None + assert record["session_id"] is None + + +async def test_streaming_appends_after_drain(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + content_chunk = ChatCompletionChunk( + id="chatcmpl-test", object="chat.completion.chunk", created=1700000000, + model="gpt-test", + choices=[ChunkChoice(index=0, delta=ChoiceDelta(content="hi"), finish_reason="stop")], + ) + usage_chunk = ChatCompletionChunk( + id="chatcmpl-test", object="chat.completion.chunk", created=1700000000, + model="gpt-test", choices=[], + usage=CompletionUsage(prompt_tokens=5, completion_tokens=3, total_tokens=8), + ) + + async def _iter() -> AsyncIterator[ChatCompletionChunk]: + yield content_chunk + yield usage_chunk + + response = ChatResponse.openai_stream(ResponseStream(_iter())) + out = await RoutingLogResponseProcessor(log_file).process( + _ctx(headers=TASK_HEADERS), response, + ) + assert not log_file.exists() # nothing until the stream drains + + forwarded = [chunk async for chunk in out.stream] + assert len(forwarded) == 2 + + (record,) = _records(log_file) + assert record["task"] == "hello-world-abc1" + assert record["total_tokens"] == 8 + + +async def test_appends_one_line_per_request(tmp_path: Path) -> None: + log_file = tmp_path / "routing_requests.jsonl" + processor = RoutingLogResponseProcessor(log_file) + for _ in range(3): + await processor.process(_ctx(headers=TASK_HEADERS), _openai_completion()) + assert len(_records(log_file)) == 3 + + +def test_serve_parser_accepts_routing_log_file() -> None: + parser = _build_parser() + args = parser.parse_args(["serve", "--routing-log-file", "/tmp/routing.jsonl"]) + assert args.routing_log_file == "/tmp/routing.jsonl" From 7fa84aed4906ed22cf4150b0c86bd0a70f40d50a Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Wed, 22 Jul 2026 14:37:55 -0500 Subject: [PATCH 2/5] feat(benchmark): stamp task and session identity headers in the proxy sidecar Signed-off-by: Ryan Lempka --- benchmark/closed_book_proxy/proxy/rewriter.py | 18 ++++++++++++++++-- benchmark/prepare_harbor_dataset.py | 1 + tests/test_prepare_harbor_dataset.py | 13 +++++++++++++ 3 files changed, 30 insertions(+), 2 deletions(-) diff --git a/benchmark/closed_book_proxy/proxy/rewriter.py b/benchmark/closed_book_proxy/proxy/rewriter.py index ed30edaf..86dfe559 100644 --- a/benchmark/closed_book_proxy/proxy/rewriter.py +++ b/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 diff --git a/benchmark/prepare_harbor_dataset.py b/benchmark/prepare_harbor_dataset.py index d87e8f1d..f7c90afb 100644 --- a/benchmark/prepare_harbor_dataset.py +++ b/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": { diff --git a/tests/test_prepare_harbor_dataset.py b/tests/test_prepare_harbor_dataset.py index 899e9318..37f6a55b 100644 --- a/tests/test_prepare_harbor_dataset.py +++ b/tests/test_prepare_harbor_dataset.py @@ -292,3 +292,16 @@ def test_generated_dataset_manifest_records_pins_tasks_and_digests(tmp_path: Pat assert {task["name"] for task in manifest["tasks"]} == {"task-a", "task-b"} assert all(task["dockerfile_digest"].startswith("sha256:") for task in manifest["tasks"]) assert all(task["compose_digest"].startswith("sha256:") for task in manifest["tasks"]) + + +def test_generated_compose_bakes_task_id_into_proxy_env(tmp_path: Path) -> None: + source = tmp_path / "source" + _write_task(source, "task-id-check", "[environment]\n", "FROM ubuntu:22.04\n") + + output = _prepare(tmp_path, source) + compose = yaml.safe_load( + (output / "task-id-check" / "environment" / "docker-compose.yaml").read_text() + ) + + proxy_env = "\n".join(compose["services"]["proxy"]["environment"]) + assert "SWITCHYARD_TASK_ID=task-id-check" in proxy_env From e1928248c90825e641c50db85b04b43794c19fcb Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Wed, 22 Jul 2026 14:37:55 -0500 Subject: [PATCH 3/5] feat(benchmark): roll routing log up into routing_stats_by_task.json at finalize Signed-off-by: Ryan Lempka --- benchmark/README.md | 7 ++++ benchmark/run-baseline.sh | 13 ++++++- benchmark/run_manifest.py | 59 +++++++++++++++++++++++++++++ tests/test_run_manifest.py | 76 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 154 insertions(+), 1 deletion(-) diff --git a/benchmark/README.md b/benchmark/README.md index dd6bff53..6ca511cb 100644 --- a/benchmark/README.md +++ b/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//result.json jobs///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`. diff --git a/benchmark/run-baseline.sh b/benchmark/run-baseline.sh index 4a0a4952..ae71e509 100755 --- a/benchmark/run-baseline.sh +++ b/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 diff --git a/benchmark/run_manifest.py b/benchmark/run_manifest.py index 33b4ee63..7091148a 100644 --- a/benchmark/run_manifest.py +++ b/benchmark/run_manifest.py @@ -223,12 +223,53 @@ 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.""" + 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": [], "models": {}}) + 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" + bucket = entry["models"].setdefault( + model, + { + "tier": record.get("tier") or "", + "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 + 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: if not path.is_file(): print(f"ERROR: manifest not found: {path}") @@ -264,6 +305,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 +396,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 +406,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() diff --git a/tests/test_run_manifest.py b/tests/test_run_manifest.py index 89a2b36b..eb32d2d1 100644 --- a/tests/test_run_manifest.py +++ b/tests/test_run_manifest.py @@ -467,3 +467,79 @@ def test_finalize_copies_proxy_strip_log_artifact(tmp_path: Path) -> None: manifest = json.loads(out.read_text()) assert manifest["closed_book"]["proxy_strip_log_status"] == "present" assert (run_dir / "proxy_strip_log.jsonl").read_text() == '{"removed":["web_search"]}\n' + + +def test_summarize_routing_log_groups_by_task_and_model(tmp_path: Path) -> None: + module = _load_manifest_module() + log = tmp_path / "routing_requests.jsonl" + log.write_text( + "\n".join([ + json.dumps({"task": "task-a", "session_id": "s1", "model": "opus", + "tier": "strong", "prompt_tokens": 10, "completion_tokens": 2, + "total_tokens": 12}), + json.dumps({"task": "task-a", "session_id": "s1", "model": "kimi", + "tier": "weak", "prompt_tokens": 4, "completion_tokens": 1, + "total_tokens": 5}), + json.dumps({"task": "task-b", "session_id": "s2", "model": "opus", + "tier": "strong", "prompt_tokens": 7, "completion_tokens": 3, + "total_tokens": 10}), + "not json", + json.dumps({"model": "opus", "prompt_tokens": 1, "completion_tokens": 1, + "total_tokens": 2}), + ]) + "\n" + ) + + summary = module.summarize_routing_log(log) + + assert summary["total_requests"] == 4 + task_a = summary["tasks"]["task-a"] + assert task_a["requests"] == 2 + assert task_a["sessions"] == ["s1"] + assert task_a["models"]["opus"] == { + "tier": "strong", "calls": 1, "prompt_tokens": 10, + "completion_tokens": 2, "total_tokens": 12, + } + assert task_a["models"]["kimi"]["calls"] == 1 + assert summary["tasks"]["task-b"]["models"]["opus"]["total_tokens"] == 10 + assert summary["tasks"]["unattributed"]["requests"] == 1 + + +def test_finalize_writes_routing_stats_by_task(tmp_path: Path) -> None: + module = _load_manifest_module() + out = tmp_path / "run_manifest.json" + run_dir = tmp_path / "run" + run_dir.mkdir() + log = run_dir / "routing_requests.jsonl" + log.write_text(json.dumps({ + "task": "task-a", "session_id": "s1", "model": "opus", "tier": "strong", + "prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12, + }) + "\n") + by_task = run_dir / "routing_stats_by_task.json" + + module.write_manifest(out, outcomes={"harbor_rc": None}) + rc = module.finalize_manifest( + out, harbor_rc=0, routing_log=log, routing_stats_by_task=by_task, + ) + + assert rc == 0 + summary = json.loads(by_task.read_text()) + assert summary["tasks"]["task-a"]["models"]["opus"]["calls"] == 1 + manifest = json.loads(out.read_text()) + assert manifest["outcomes"]["routing_stats_by_task_json_status"] == "present" + + +def test_finalize_marks_routing_stats_by_task_missing_without_log(tmp_path: Path) -> None: + module = _load_manifest_module() + out = tmp_path / "run_manifest.json" + by_task = tmp_path / "routing_stats_by_task.json" + + module.write_manifest(out, outcomes={"harbor_rc": None}) + rc = module.finalize_manifest( + out, harbor_rc=1, + routing_log=tmp_path / "absent.jsonl", routing_stats_by_task=by_task, + ) + + assert rc == 0 + assert not by_task.exists() + manifest = json.loads(out.read_text()) + assert manifest["outcomes"]["routing_stats_by_task_json_status"] == "missing" From 38f462712cef0718e7687eaaa05f4d12e076b669 Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Wed, 22 Jul 2026 14:56:26 -0500 Subject: [PATCH 4/5] fix(serve): offload routing-log writes off the event loop Signed-off-by: Ryan Lempka --- switchyard/lib/processors/routing_log_response_processor.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/switchyard/lib/processors/routing_log_response_processor.py b/switchyard/lib/processors/routing_log_response_processor.py index e840c435..006f34c9 100644 --- a/switchyard/lib/processors/routing_log_response_processor.py +++ b/switchyard/lib/processors/routing_log_response_processor.py @@ -5,6 +5,7 @@ from __future__ import annotations +import asyncio import json import logging import threading @@ -39,12 +40,13 @@ def __init__(self, log_file: Path | str) -> None: 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: - self._write_record(ctx, served_model, final) + await asyncio.to_thread(self._write_record, ctx, served_model, final) attached = attach_final_response_callback( response, served_model=served_model, callback=_emit, From 7f4c816ab05b919982b58647be0710c1fe06e1d1 Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Wed, 22 Jul 2026 14:56:26 -0500 Subject: [PATCH 5/5] fix(benchmark): key rollup buckets by model and tier Signed-off-by: Ryan Lempka --- benchmark/run_manifest.py | 23 ++++++++++++++++++----- tests/test_run_manifest.py | 34 +++++++++++++++++++++++++++------- 2 files changed, 45 insertions(+), 12 deletions(-) diff --git a/benchmark/run_manifest.py b/benchmark/run_manifest.py index 7091148a..5f286530 100644 --- a/benchmark/run_manifest.py +++ b/benchmark/run_manifest.py @@ -224,7 +224,11 @@ def _copy_if_present(source: Path | None, dest: Path | None) -> str: def summarize_routing_log(log_path: Path) -> dict[str, Any]: - """Aggregate a per-request routing JSONL log into per-task stats.""" + """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(): @@ -239,16 +243,18 @@ def summarize_routing_log(log_path: Path) -> dict[str, Any]: continue total_requests += 1 task = record.get("task") or "unattributed" - entry = tasks.setdefault(task, {"requests": 0, "sessions": [], "models": {}}) + 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" - bucket = entry["models"].setdefault( - model, + tier = record.get("tier") or "" + bucket = entry["_buckets"].setdefault( + (model, tier), { - "tier": record.get("tier") or "", + "model": model, + "tier": tier, "calls": 0, "prompt_tokens": 0, "completion_tokens": 0, @@ -259,6 +265,9 @@ def summarize_routing_log(log_path: Path) -> dict[str, Any]: 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} @@ -271,6 +280,10 @@ def finalize_manifest( 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 diff --git a/tests/test_run_manifest.py b/tests/test_run_manifest.py index eb32d2d1..a26d5021 100644 --- a/tests/test_run_manifest.py +++ b/tests/test_run_manifest.py @@ -495,15 +495,35 @@ def test_summarize_routing_log_groups_by_task_and_model(tmp_path: Path) -> None: task_a = summary["tasks"]["task-a"] assert task_a["requests"] == 2 assert task_a["sessions"] == ["s1"] - assert task_a["models"]["opus"] == { - "tier": "strong", "calls": 1, "prompt_tokens": 10, - "completion_tokens": 2, "total_tokens": 12, - } - assert task_a["models"]["kimi"]["calls"] == 1 - assert summary["tasks"]["task-b"]["models"]["opus"]["total_tokens"] == 10 + assert task_a["models"] == [ + {"model": "kimi", "tier": "weak", "calls": 1, "prompt_tokens": 4, + "completion_tokens": 1, "total_tokens": 5}, + {"model": "opus", "tier": "strong", "calls": 1, "prompt_tokens": 10, + "completion_tokens": 2, "total_tokens": 12}, + ] + assert summary["tasks"]["task-b"]["models"][0]["total_tokens"] == 10 assert summary["tasks"]["unattributed"]["requests"] == 1 +def test_summarize_routing_log_keeps_mixed_tiers_separate(tmp_path: Path) -> None: + module = _load_manifest_module() + log = tmp_path / "routing_requests.jsonl" + log.write_text( + "\n".join([ + json.dumps({"task": "task-a", "model": "opus", "tier": "strong", + "prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}), + json.dumps({"task": "task-a", "model": "opus", "tier": "weak", + "prompt_tokens": 4, "completion_tokens": 1, "total_tokens": 5}), + ]) + "\n" + ) + + buckets = module.summarize_routing_log(log)["tasks"]["task-a"]["models"] + + assert [(b["model"], b["tier"], b["calls"]) for b in buckets] == [ + ("opus", "strong", 1), ("opus", "weak", 1), + ] + + def test_finalize_writes_routing_stats_by_task(tmp_path: Path) -> None: module = _load_manifest_module() out = tmp_path / "run_manifest.json" @@ -523,7 +543,7 @@ def test_finalize_writes_routing_stats_by_task(tmp_path: Path) -> None: assert rc == 0 summary = json.loads(by_task.read_text()) - assert summary["tasks"]["task-a"]["models"]["opus"]["calls"] == 1 + assert summary["tasks"]["task-a"]["models"][0]["calls"] == 1 manifest = json.loads(out.read_text()) assert manifest["outcomes"]["routing_stats_by_task_json_status"] == "present"