feat(latency): pass route-selection spend metadata to LiteLLM

Author: eric-liu-nvidia

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

@@ -43,6 +43,7 @@ right validation set. If the change is driven by a launcher need, also read
 | 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. |
 | Per-endpoint attribution on `/metrics` for a Python backend that can't be wrapped by `StatsLlmBackend` | Set `ctx.selected_model = endpoint_id` before returning the response. Also set `ctx.backend_call_latency_ms = upstream_call_ms` so the response processor can compute routing overhead. `LatencyServiceLLMBackend.call` is the reference. |
+| Spend/tokenomics attribution when a front proxy (LiteLLM) sits above Switchyard | Stamp each upstream attempt with the `x-litellm-spend-logs-metadata` request header (selection re-stamped per failover attempt, one `router_correlation_id` uuid per client request) and record the successful attempt's selection in `ctx.metadata[CTX_ROUTE_SELECTION]`; the endpoint layer maps it to the `x-switchyard-*` response headers via `route_selection_headers` in `lib/endpoints/route_selection.py` — stamped by `dispatch.serialize_chain_result` on success and by `upstream_error.handle_chain_exception` when a failure follows a billed upstream success. Header values are sanitized there (client-controlled `router_model` must stay legal header material). `LatencyServiceLLMBackend.call` is the reference; the payload contract lives on `CTX_ROUTE_SELECTION` in `switchyard/lib/proxy_context.py`. |
 | State metrics on `/metrics` | Register a `PrometheusEmitter` via `switchyard.lib.endpoints.prometheus_emitter.register(...)` and unregister on `shutdown()`. This is for backend-owned state, not request-flow counters. |
 | Error-rate / retry-recovery counters | Use `switchyard.lib.endpoints.outcome_metrics`. FastAPI middleware records client outcomes. A retrying Python backend records each upstream attempt itself (and `record_retry_recovered()`) and sets `CTX_UPSTREAM_ATTEMPTS_RECORDED` so the endpoint skips its fallback — `LatencyServiceLLMBackend.call` is the reference. Single-attempt backends (Rust native / passthrough / multi) record nothing themselves; the endpoint fallback (`record_upstream_attempt_success` / `record_upstream_attempt_failure` in `upstream_error.py`, called from `dispatch_chat_request` and `handle_chain_exception`) counts their one attempt. Don't add a `model` label — these counters are layer-aggregate. Keep labels bounded. For a per-model error breakdown, emit a backend-owned counter via the `PrometheusEmitter` instead (labels bounded to config-derived ids: the route id `config.route_model` + endpoint ids, else the `other` sentinel) — `LatencyServiceLLMBackend._render_prometheus_lines` exposes `switchyard_latency_upstream_attempts_total{requested_model,upstream_model,outcome,code}` next to the aggregate, leaving the shared counter model-free. |
 | Per-event error log | Use `switchyard.lib.endpoints.upstream_error_log.log_upstream_attempt_failure(...)` on the failure path. Events belong in logs/traces, not Prometheus sample timestamps. |

docs/internal/latency_service_routing.md

@@ -431,6 +431,52 @@ front of Switchyard (e.g. LiteLLM) should tag its own failures the same way and
 propagate these headers from below. Mid-stream failures after HTTP 200 has been
 committed cannot carry headers and are not annotated today.
 
+## Route-selection spend attribution
+
+For tokenomics reporting, a LiteLLM front proxy needs to tie its provider spend-log
+rows back to the Switchyard logical route that selected them. The backend stamps
+**every upstream attempt** with a spend-logs metadata request header (LiteLLM copies
+its JSON value into the provider spend-log DB row):
+
+```text
+x-litellm-spend-logs-metadata: {"router_model": "<client-facing route id>",
+                                "router_strategy": "latency",
+                                "router_selected_endpoint": "<endpoint id>",
+                                "router_selected_model": "<upstream model>",
+                                "router_selected_provider": "<upstream model's leading path segment>",
+                                "router_correlation_id": "<uuid4>"}
+```
+
+One correlation id is generated per client request and shared by every failover
+attempt; the selection fields are re-stamped per attempt, so each provider row
+records the endpoint it actually hit. `router_model` is the model the client asked
+for, falling back to the configured `route_model` when the body carried no model.
+
+**Successful** responses then return the selection that served the request, so the
+front proxy can enrich its own (parent) spend-log row with the same correlation id
+the provider row received:
+
+| Header | Value |
+|---|---|
+| `x-switchyard-router-model` | `router_model` from the selection |
+| `x-switchyard-selected-model` | `router_selected_model` |
+| `x-switchyard-selected-provider` | `router_selected_provider` |
+| `x-switchyard-router-correlation-id` | `router_correlation_id` |
+
+Streaming responses carry the same headers (the upstream call completes before the
+SSE response is committed). Error responses carry the failure-source headers above;
+the selection headers join them only when an upstream call already **succeeded**
+before the failure (e.g. a response-translation error after a billed 200), so the
+provider row's correlation id stays joinable — a request that never reached a
+successful upstream claims no selection. Because the client controls `router_model`,
+header values are re-validated as header material: a value that is not
+latin-1-encodable or contains control characters is omitted rather than breaking
+(or splitting) the response; the outbound JSON header is immune (`json.dumps`
+escapes it) and still records the raw value. The cross-component contract is
+`CTX_ROUTE_SELECTION` in `switchyard/lib/proxy_context.py` (written by the backend,
+read by the endpoint layer via `switchyard/lib/endpoints/route_selection.py`);
+wire-level proofs live in `tests/test_route_selection_headers.py`.
+
 ## Observability
 
 When `enable_stats` is `true` (the default), `LatencyServiceProfileConfig` wires a

switchyard/lib/backends/latency_service_llm_backend.py

@@ -20,10 +20,13 @@
 Responses-mode endpoints receive the OpenAI Responses API natively.
 """
 
+import json
 import logging
 import random
 import threading
 import time
+import uuid
+from collections.abc import Mapping
 from dataclasses import dataclass
 
 from openai import APIStatusError, AsyncStream
@@ -46,6 +49,7 @@
 from switchyard.lib.proxy_context import (
     CTX_CALLER_API_KEY,
     CTX_ERROR_SOURCE,
+    CTX_ROUTE_SELECTION,
     CTX_UPSTREAM_ATTEMPTS_RECORDED,
     CTX_UPSTREAM_HTTP_BODY,
     CTX_UPSTREAM_HTTP_STATUS,
@@ -63,6 +67,14 @@
 
 log = logging.getLogger(__name__)
 
+#: Request header read by a LiteLLM front proxy; its JSON value is copied into
+#: the provider spend-log DB row, tying that row back to the Switchyard route
+#: that selected it (see :data:`CTX_ROUTE_SELECTION` for the payload contract).
+SPEND_LOGS_METADATA_HEADER = "x-litellm-spend-logs-metadata"
+
+#: ``router_strategy`` value identifying this backend's selection algorithm.
+ROUTER_STRATEGY_LATENCY = "latency"
+
 
 @dataclass(frozen=True)
 class _RouteDecision:
@@ -380,6 +392,10 @@ async def call(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse:
         # Captured before the per-attempt ``body["model"]`` override so the span
         # records the model the client asked for, not the selected endpoint.
         incoming_model = request.model
+        # One correlation id per client request, shared by every failover
+        # attempt's spend-logs header and by the response headers — the join
+        # key between a front proxy's spend-log row and the provider rows.
+        correlation_id = str(uuid.uuid4())
         # Resolve the session-affinity pin once (keyed on the stable conversation
         # prefix). ``None`` when affinity is disabled or the conversation isn't
         # pinned yet, so every attempt routes purely by health + latency.
@@ -426,6 +442,15 @@ async def call(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse:
             target_request_type = self._request_types[model_id]
             body = self._body_for_endpoint_request_type(ctx, request, target_request_type)
             body["model"] = upstream_model
+            # Per-attempt route-selection record: each upstream call is stamped
+            # with the endpoint it actually hits, so a failover retry's
+            # spend-log row doesn't inherit the failed attempt's selection.
+            route_selection = self._route_selection(
+                incoming_model=incoming_model,
+                model_id=model_id,
+                upstream_model=upstream_model,
+                correlation_id=correlation_id,
+            )
             log.debug(
                 "LatencyServiceLLMBackend: attempt=%d model=%s upstream=%s "
                 "request_type=%s stream=%s",
@@ -450,6 +475,9 @@ async def call(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse:
                         target_request_type,
                         api_key=api_key_override,
                         body=body,
+                        extra_headers={
+                            SPEND_LOGS_METADATA_HEADER: json.dumps(route_selection),
+                        },
                     )
                 except APIStatusError as exc:
                     set_tags(attempt_span, {
@@ -547,6 +575,12 @@ async def call(self, ctx: ProxyContext, request: ChatRequest) -> ChatResponse:
                 # Rust ``StatsLlmBackend`` that normally publishes this signal.
                 ctx.backend_call_latency_ms = backend_latency_ms
 
+                # The selection that actually served the request, surfaced to
+                # the endpoint layer as ``x-switchyard-*`` response headers so
+                # a front proxy can enrich its own spend-log row with the same
+                # correlation id the provider row received.
+                ctx.metadata[CTX_ROUTE_SELECTION] = route_selection
+
                 # Pin this conversation to the endpoint that served it so later
                 # turns reuse it (warm cache). Re-pinning on every success also
                 # follows a recovery: if the previous pin degraded and we
@@ -605,21 +639,66 @@ def _body_for_endpoint_request_type(
             )
         return dict(normalized.body)
 
+    def _route_selection(
+        self,
+        *,
+        incoming_model: str | None,
+        model_id: str,
+        upstream_model: str,
+        correlation_id: str,
+    ) -> dict[str, str | None]:
+        """Build the route-selection payload for one upstream attempt.
+
+        This is the JSON value of the outbound ``x-litellm-spend-logs-metadata``
+        header and, for the attempt that succeeds, the
+        :data:`CTX_ROUTE_SELECTION` record behind the ``x-switchyard-*``
+        response headers. ``router_model`` falls back to the configured route id
+        when the client body carried no model; ``router_selected_provider`` is
+        the upstream model's leading path segment (IH/LiteLLM naming, e.g.
+        ``"openai/openai/gpt-5.4"`` → ``"openai"``).
+        """
+        return {
+            "router_model": incoming_model or self._config.route_model,
+            "router_strategy": ROUTER_STRATEGY_LATENCY,
+            "router_selected_endpoint": model_id,
+            "router_selected_model": upstream_model,
+            "router_selected_provider": upstream_model.split("/", 1)[0],
+            "router_correlation_id": correlation_id,
+        }
+
     async def _call_endpoint(
         self,
         model_id: str,
         target_request_type: ChatRequestType,
         *,
         api_key: str | None,
         body: dict[str, object],
+        extra_headers: dict[str, str],
     ) -> object:
+        # ``extra_headers`` rides the client wrapper's ``**kwargs`` into the
+        # OpenAI SDK's per-request header merge (over ``default_headers``).
+        # A passthrough body may itself carry an SDK-style ``extra_headers``
+        # field (shape-preserving translation keeps unknown keys); merge it
+        # under ours rather than letting it collide with the keyword — it used
+        # to ride ``**body`` into the same SDK parameter, and the spend-logs
+        # header must win a name conflict so callers can't spoof it. A
+        # non-mapping value is dropped: it could only ever have broken the
+        # SDK call.
+        client_extra = body.pop("extra_headers", None)
+        headers: dict[str, object] = (
+            {**client_extra, **extra_headers}
+            if isinstance(client_extra, Mapping)
+            else dict(extra_headers)
+        )
         if target_request_type == ChatRequestType.OPENAI_RESPONSES:
             return await self._clients[model_id].aresponses(
                 api_key=api_key,
+                extra_headers=headers,
                 **body,
             )
         return await self._clients[model_id].acompletion(
             api_key=api_key,
+            extra_headers=headers,
             **body,
         )
 

switchyard/lib/endpoints/anthropic_messages_endpoint.py

@@ -106,7 +106,9 @@ async def anthropic_messages(
                         stream,
                         type(result).__name__,
                     )
-                return serialize_chain_result(result, stream=stream, sse_iter=iter_anthropic_sse)
+                return serialize_chain_result(
+                    result, stream=stream, sse_iter=iter_anthropic_sse, ctx=ctx
+                )
             except (SwitchyardContextPoolExhaustedError, SwitchyardContextWindowExceededError) as exc:
                 return context_exhausted_response(exc, inbound="anthropic")
             except Exception as exc:

switchyard/lib/endpoints/dispatch.py

@@ -9,6 +9,7 @@
 from fastapi.responses import JSONResponse, Response, StreamingResponse
 
 from switchyard.lib.endpoints.error_envelope import error_response
+from switchyard.lib.endpoints.route_selection import route_selection_headers
 from switchyard.lib.endpoints.upstream_error import record_upstream_attempt_success
 from switchyard.lib.proxy_context import ProxyContext
 from switchyard.lib.roles import TranslatedResponse
@@ -96,16 +97,24 @@ def serialize_chain_result(
     *,
     stream: bool,
     sse_iter: Callable[[Any], AsyncIterator[str]],
+    ctx: ProxyContext,
 ) -> Response:
     """Serialize a chain result to the appropriate HTTP response.
 
     Returns the result as-is if it is already a ``Response``, wraps it in a
     ``StreamingResponse`` when streaming is requested, or JSON-serializes it.
+    Any route selection recorded on *ctx* is stamped as ``x-switchyard-*``
+    response headers (streaming included — the backend call completed before
+    the response object is built, so the selection is final). ``ctx`` is
+    required so a new endpoint cannot silently opt out of spend attribution.
     """
     if isinstance(result, Response):
         return result
+    headers = route_selection_headers(ctx)
     if stream and hasattr(result, "__aiter__"):
-        return StreamingResponse(sse_iter(result), media_type="text/event-stream")
+        return StreamingResponse(
+            sse_iter(result), media_type="text/event-stream", headers=headers
+        )
     if hasattr(result, "model_dump"):
-        return JSONResponse(content=result.model_dump())
-    return JSONResponse(content=result)
+        return JSONResponse(content=result.model_dump(), headers=headers)
+    return JSONResponse(content=result, headers=headers)

switchyard/lib/endpoints/openai_chat_endpoint.py

@@ -89,7 +89,7 @@ async def chat_completions(
             try:
                 result: Any = await dispatch_chat_request(obj, chat_request, ctx)
                 return serialize_chain_result(
-                    result, stream=stream, sse_iter=iter_chat_completion_sse
+                    result, stream=stream, sse_iter=iter_chat_completion_sse, ctx=ctx
                 )
             except (SwitchyardContextPoolExhaustedError, SwitchyardContextWindowExceededError) as exc:
                 return context_exhausted_response(exc, inbound="openai")

switchyard/lib/endpoints/responses_endpoint.py

@@ -84,7 +84,9 @@ async def responses(
                         stream,
                         type(result).__name__,
                     )
-                return serialize_chain_result(result, stream=stream, sse_iter=iter_preframed_sse)
+                return serialize_chain_result(
+                    result, stream=stream, sse_iter=iter_preframed_sse, ctx=ctx
+                )
             except (SwitchyardContextPoolExhaustedError, SwitchyardContextWindowExceededError) as exc:
                 return context_exhausted_response(exc, inbound="openai-responses")
             except Exception as exc:

switchyard/lib/endpoints/route_selection.py

@@ -0,0 +1,68 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Route-selection response headers for spend/tokenomics attribution.
+
+Maps the :data:`CTX_ROUTE_SELECTION` record a routing backend stored on
+``ctx`` (see :mod:`switchyard.lib.proxy_context`) to the ``x-switchyard-*``
+response headers a front proxy such as LiteLLM copies into its parent
+spend-log row. Shared by the success serializer (``dispatch``) and the
+error path (``upstream_error``) — a failure that happens *after* a billed
+upstream success must still expose the selection, or the provider spend-log
+row's correlation id becomes unjoinable.
+"""
+
+from collections.abc import Mapping
+
+from switchyard.lib.proxy_context import CTX_ROUTE_SELECTION, ProxyContext
+
+#: Response headers exposing the route selection behind an upstream call,
+#: carrying the same correlation id the provider row received via the
+#: outbound ``x-litellm-spend-logs-metadata`` header.
+ROUTER_MODEL_HEADER = "x-switchyard-router-model"
+SELECTED_MODEL_HEADER = "x-switchyard-selected-model"
+SELECTED_PROVIDER_HEADER = "x-switchyard-selected-provider"
+ROUTER_CORRELATION_ID_HEADER = "x-switchyard-router-correlation-id"
+
+_ROUTE_SELECTION_RESPONSE_HEADERS = (
+    (ROUTER_MODEL_HEADER, "router_model"),
+    (SELECTED_MODEL_HEADER, "router_selected_model"),
+    (SELECTED_PROVIDER_HEADER, "router_selected_provider"),
+    (ROUTER_CORRELATION_ID_HEADER, "router_correlation_id"),
+)
+
+
+def _is_header_value_safe(value: str) -> bool:
+    """Whether *value* can be emitted as an HTTP/1.1 response-header value.
+
+    ``router_model`` echoes the client-supplied model string, so it must be
+    re-validated as header material: Starlette encodes response-header values
+    as latin-1 (a non-encodable value would fail response construction after
+    the upstream call already succeeded and was billed), and CTL characters —
+    CR/LF above all — would be a response-splitting vector on permissive
+    ASGI stacks.
+    """
+    try:
+        value.encode("latin-1")
+    except UnicodeEncodeError:
+        return False
+    return not any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value)
+
+
+def route_selection_headers(ctx: ProxyContext) -> dict[str, str]:
+    """Response headers for the route selection recorded on *ctx*, if any.
+
+    Empty when no routing backend recorded a selection (passthrough chains,
+    failures before any upstream success). A recorded field that is absent or
+    not emittable as a header value is skipped — headers never carry
+    placeholder or unsafe values.
+    """
+    selection = ctx.metadata.get(CTX_ROUTE_SELECTION)
+    if not isinstance(selection, Mapping):
+        return {}
+    headers: dict[str, str] = {}
+    for header_name, selection_key in _ROUTE_SELECTION_RESPONSE_HEADERS:
+        value = selection.get(selection_key)
+        if isinstance(value, str) and value and _is_header_value_safe(value):
+            headers[header_name] = value
+    return headers