Retain provider timing and reasoning usage for timeout diagnosis

docs/provider-diagnostics.md

@@ -0,0 +1,52 @@
+# Provider timing diagnostics
+
+OpenAI-compatible asynchronous calls retain a bounded `provider_diagnostics`
+array in `x_router.decision_trace`, including failed attempts. The call ledger
+stores the same allowlisted evidence in `calls.routing_summary`, so it remains
+available when an ingress replaces a JSON 504 with an HTML error page. No database
+migration is needed. Diagnostics do not change request bodies, routing, retries,
+deadlines, generation limits or which response content is emitted.
+
+Each entry identifies its attempt, provider and model family and may include:
+
+| Field | Meaning |
+|---|---|
+| `mode` | `buffered`, `buffered_sse` (first-output-bound buffered request), or `streaming` |
+| `phase` | Last observed transport phase: connection pool, connect, TLS, request headers/body, response headers/body; or peer capacity before HTTP starts |
+| `request_sent_ms` | Elapsed time when HTTP Core finished sending the request body |
+| `response_headers_ms` | Elapsed time when response headers arrived |
+| `body_complete_ms` | Elapsed time when HTTP Core completed reading the body |
+| `connect_ms`, `tls_ms` | Connection/TLS operation durations, when a connection was created |
+| `http_status` | Upstream HTTP status, when headers were observed; an incomplete body can have status 200 and still time out |
+| `timeout_source` | Buffered request's total `attempt_deadline`, or HTTPX timeout exception class |
+| `first_reasoning_ms`, `first_output_ms` | First nonempty reasoning or content/tool delta observed by the streaming adapter; independent of whether the router has emitted it publicly |
+| `elapsed_ms` | Adapter elapsed time; phase timestamps share this origin |
+| `upstream_id`, `upstream_provider` | Provider response identifiers when returned; allow correlating with provider support/usage metadata |
+| `tokens_reasoning` | Reported reasoning-token count, including explicit zero |
+| `requested_timeout_ms`, `requested_max_tokens`, `requested_reasoning_effort`, `requested_reasoning_enabled` | Controls passed to the adapter; not a claim that the upstream honored them |
+
+Reasoning usage is also exposed as standard
+`usage.completion_tokens_details.reasoning_tokens` when reported. It is already
+part of completion usage, so it is not added to total tokens or charged twice.
+Missing counts/timings are unknown, not zero. Reused connections do not emit a
+new connect/TLS duration. These fields cannot split an upstream's internal queue
+from its computation without upstream metadata.
+
+For a buffered request, `phase=response_headers` with no `response_headers_ms`
+means no headers arrived before the failure. `phase=response_body` with an
+upstream 200 and no `body_complete_ms` means the response started but was not
+fully read before failure. A pooled-connection wait has no send/header events.
+
+Only HTTPX clients expose HTTP Core phase tracing. Custom/mock clients continue
+to work without those timing fields. Streaming can record an upstream ID before
+completion; buffered JSON may not expose one until the complete response arrives.
+An outer router deadline/cancellation can stop execution before an adapter result
+is returned; this change does not claim to retain a partially cancelled attempt.
+Native tool-call fragments are still accumulated before the final tool-call chunk.
+The existing streaming timeout semantics are unchanged.
+
+At most 32 diagnostic entries are retained per router execution. Fields are
+allowlisted, strings are capped, and numeric fields must be finite/nonnegative.
+Prompts, command arguments, HTTP headers, endpoint URLs, provider error bodies,
+and reasoning text are not copied into diagnostics. A response identifier is a
+correlation identifier, not an authentication credential.

host_store.py

@@ -398,6 +398,11 @@ def routing_summary(trace) -> dict | None:
         if isinstance(step, dict) and step.get('event') == 'attempted'
     ]
     summary['deadline_exceeded'] = trace.get('request_deadline_exceeded') is True
+    from provider_adapters.diagnostics import bounded_diagnostics
+    diagnostics = trace.get('provider_diagnostics')
+    if isinstance(diagnostics, list):
+        summary['provider_diagnostics'] = [bounded_diagnostics(item) for item in diagnostics[:32]
+                                           if isinstance(item, dict)]
     decision = trace.get('automatic')
     if isinstance(decision, dict):
         # A closed allowlist excludes payloads, instructions and arbitrary provider bodies.

llm_router_host.py

@@ -37,6 +37,7 @@
     _provider_error_message,
 )
 from provider_adapters.dispatcher import make_api_kind_dispatcher
+from provider_adapters.diagnostics import bounded_diagnostics
 from provider_adapters.google import make_google_async_call_provider
 from provider_adapters.openai_compatible import (
     _PEER_GATES,
@@ -629,6 +630,8 @@ async def execute_async(self, contract: dict, call_override=None) -> dict:
         # fold so route_cache learns which peer served this conversation. It is a
         # local of this coroutine, so concurrent executes never share it.
         session = contract.get("session")
+        provider_diagnostics = []
+        provider_attempt = 0
         engine_contract = contract
         if contract.get('protocol') == 'decisions':
             # Lua tables cannot represent JSON null or distinguish [] from {}.
@@ -639,17 +642,30 @@ async def execute_async(self, contract: dict, call_override=None) -> dict:
         while True:
             status = step["status"]
             if status == "done":
-                return _to_py(step["result"])
+                result = _to_py(step["result"])
+                if provider_diagnostics:
+                    result.setdefault("trace", {})["provider_diagnostics"] = provider_diagnostics
+                return result
 
             handle = step["state_handle"]
             if status == "call":
+                provider_attempt += 1
                 req = _to_py(step["request"]) or {}
                 if req.get('protocol') == 'decisions':
                     req['decision'] = json.loads(req['decision'])
                 if (contract.get("first_token_timeout_ms") is not None
                         and req.get("first_token_timeout_ms") is None):
                     req["first_token_timeout_ms"] = contract["first_token_timeout_ms"]
                 resp = await self._resolve_call_async(req, call_override, session=session)
+                diagnostic = bounded_diagnostics(resp.get("diagnostics"))
+                if diagnostic and len(provider_diagnostics) < 32:
+                    diagnostic.update(bounded_diagnostics({
+                        "attempt": provider_attempt,
+                        "provider_id": req.get("provider_id"),
+                        "model_family": req.get("model_family"),
+                        "error_kind": resp.get("error_kind"),
+                    }))
+                    provider_diagnostics.append(diagnostic)
                 step = self.router.execute_step(handle, None, _to_lua(self.lua, resp))
             elif status == "wait":
                 until_ms = step["until_ms"] or 0

provider_adapters/diagnostics.py

@@ -0,0 +1,118 @@
+"""Bounded transport/usage evidence, without prompts, credentials or reasoning text."""
+from __future__ import annotations
+
+import math
+import time
+
+
+TEXT_FIELDS = frozenset({
+    "mode", "phase", "timeout_source", "upstream_id", "upstream_provider",
+    "provider_id", "model_family", "error_kind", "requested_reasoning_effort",
+})
+NUMBER_FIELDS = frozenset({
+    "http_status", "elapsed_ms", "request_sent_ms", "response_headers_ms",
+    "body_complete_ms", "connect_ms", "tls_ms", "first_output_ms",
+    "first_reasoning_ms", "tokens_reasoning", "requested_timeout_ms",
+    "requested_max_tokens", "attempt",
+})
+
+
+def bounded_diagnostics(value: object) -> dict:
+    if not isinstance(value, dict):
+        return {}
+    out = {key: value[key][:160] for key in TEXT_FIELDS
+           if isinstance(value.get(key), str)}
+    out.update({key: value[key] for key in NUMBER_FIELDS
+                if isinstance(value.get(key), (int, float))
+                and not isinstance(value[key], bool)
+                and 0 <= value[key] <= 1e15 and math.isfinite(value[key])})
+    if isinstance(value.get("requested_reasoning_enabled"), bool):
+        out["requested_reasoning_enabled"] = value["requested_reasoning_enabled"]
+    return out
+
+
+def upstream_metadata(data: dict) -> dict:
+    usage = data.get("usage") or {}
+    usage = usage if isinstance(usage, dict) else {}
+    details = usage.get("completion_tokens_details") or {}
+    details = details if isinstance(details, dict) else {}
+    return bounded_diagnostics({
+        "upstream_id": data.get("id"),
+        "upstream_provider": data.get("provider"),
+        "tokens_reasoning": details.get("reasoning_tokens"),
+    })
+
+
+class ProviderTiming:
+    def __init__(self, request: dict, timeout: float, mode: str):
+        self.started = time.monotonic()
+        self.starts: dict[str, float] = {}
+        reasoning = request.get("reasoning")
+        reasoning = reasoning if isinstance(reasoning, dict) else {}
+        self.data = bounded_diagnostics({
+            "mode": mode, "phase": "peer_capacity",
+            "requested_timeout_ms": timeout * 1000,
+            "requested_max_tokens": request.get("max_tokens"),
+            "requested_reasoning_effort": reasoning.get("effort", request.get("reasoning_effort")),
+            "requested_reasoning_enabled": reasoning.get("enabled"),
+        })
+
+    def elapsed(self) -> float:
+        return round((time.monotonic() - self.started) * 1000, 3)
+
+    async def trace(self, event: str, info: dict) -> None:
+        # HTTP Core's trace info can contain URLs, headers and exceptions. Read
+        # only the event name and numeric HTTP status; never retain info itself.
+        operation, _, outcome = event.rpartition(".")
+        operation = operation.rsplit(".", 1)[-1]
+        now = time.monotonic()
+        phases = {"connect_tcp": "connect", "start_tls": "tls",
+                  "send_request_headers": "request_headers",
+                  "send_request_body": "request_body",
+                  "receive_response_headers": "response_headers",
+                  "receive_response_body": "response_body"}
+        if operation not in phases:
+            return
+        if outcome == "started":
+            self.data["phase"] = phases[operation]
+            self.starts[operation] = now
+        elif outcome == "complete":
+            if operation in ("connect_tcp", "start_tls") and operation in self.starts:
+                key = "connect_ms" if operation == "connect_tcp" else "tls_ms"
+                self.data[key] = round((now - self.starts[operation]) * 1000, 3)
+            key = {"send_request_body": "request_sent_ms",
+                   "receive_response_headers": "response_headers_ms",
+                   "receive_response_body": "body_complete_ms"}.get(operation)
+            if key:
+                self.data[key] = self.elapsed()
+            if operation == "receive_response_headers":
+                value = info.get("return_value")
+                if isinstance(value, tuple) and len(value) >= 2 and isinstance(value[1], int):
+                    self.data["http_status"] = value[1]
+                elif isinstance(value, tuple) and value and isinstance(value[0], int):
+                    self.data["http_status"] = value[0]  # HTTP/2 has no HTTP version prefix.
+
+    def http_options(self, client) -> dict:
+        # Only HTTPX implements this tracing extension. Custom transports/test
+        # clients retain their existing protocol and receive no extra arguments.
+        import httpx
+        if isinstance(client, httpx.AsyncClient):
+            return {"extensions": {"trace": self.trace}}
+        return {}
+
+    def observe_chunk(self, chunk: dict) -> None:
+        self.data.update(upstream_metadata(chunk))
+        for choice in chunk.get("choices") or []:
+            delta = choice.get("delta") or {}
+            if delta.get("reasoning") or delta.get("reasoning_details"):
+                self.data.setdefault("first_reasoning_ms", self.elapsed())
+            if delta.get("content") or delta.get("tool_calls"):
+                self.data.setdefault("first_output_ms", self.elapsed())
+
+    def attach(self, result: dict) -> dict:
+        nested = result.get("diagnostics") or {}
+        metadata = (result.get("response") or {}).get("upstream") or {}
+        result["diagnostics"] = bounded_diagnostics({
+            **self.data, **nested, **metadata, "elapsed_ms": self.elapsed(),
+        })
+        return result

provider_adapters/openai_compatible.py

@@ -25,6 +25,8 @@
     _provider_error_message,
 )
 
+from provider_adapters.diagnostics import ProviderTiming, upstream_metadata
+
 Emit = Callable[[str], Awaitable[None]]
 
 
@@ -428,19 +430,21 @@ async def call(request: dict) -> dict:
         # adapter. Acquiring here as well would deadlock a cap=1 peer against
         # this same request.
         uses_streaming_backend = request.get("first_token_timeout_ms") is not None
+        timing = ProviderTiming(request, timeout, "buffered_sse" if uses_streaming_backend else "buffered")
         slot = None
         if not uses_streaming_backend:
             slot, gate_error = await _acquire_peer_capacity(request, timeout)
             if gate_error:
-                return _peer_capacity_error(
-                    str(peer_id or ""), int(cap or 0), gate_error, t0)
+                return timing.attach(_peer_capacity_error(
+                    str(peer_id or ""), int(cap or 0), gate_error, t0))
         try:
             try:
                 # HTTPX limits inactivity between reads; a trickling response can
                 # exceed it indefinitely. Bound the complete buffered call, including
                 # time already spent waiting for peer capacity.
                 remaining = max(0.0, timeout - (_time.monotonic() - t0))
-                async with asyncio.timeout(remaining):
+                deadline = asyncio.timeout(remaining)
+                async with deadline:
                     if uses_streaming_backend:
                         # Reuse the streaming backend (defined below in this module) to
                         # get a first-token bound, discarding deltas — a non-stream call.
@@ -455,29 +459,33 @@ async def _ignore_delta(_delta: str) -> None:
                             extra_headers=_extra,
                             timeout_s=timeout_s,
                             token_providers=token_providers,
-                            provider_rules=provider_rules,
+                            provider_rules=provider_rules, _timing=timing,
                         )
                     else:
                         from byo_http import buyer_client, is_byo_buyer
                         if is_byo_buyer(request, _env_get):
                             async with buyer_client() as buyer:
-                                resp = await buyer.post(url, json=body, headers=headers, timeout=timeout)
+                                timing.data["phase"] = "connection_pool"
+                                resp = await buyer.post(url, json=body, headers=headers, timeout=timeout, **timing.http_options(buyer))
                         elif client is not None:
+                            timing.data["phase"] = "connection_pool"
                             resp = await client.post(
-                                url, json=body, headers=headers, timeout=timeout)
+                                url, json=body, headers=headers, timeout=timeout, **timing.http_options(client))
                         else:
                             async with httpx.AsyncClient() as c:
+                                timing.data["phase"] = "connection_pool"
                                 resp = await c.post(
-                                    url, json=body, headers=headers, timeout=timeout)
+                                    url, json=body, headers=headers, timeout=timeout, **timing.http_options(c))
                         rules = (provider_rules or {}).get(request.get("provider_id")) or {}
                         result = _parse_openai_response(
                             resp, _elapsed_ms(t0), error_map=rules.get("error_map"))
-            except (TimeoutError, httpx.TimeoutException):
+            except (TimeoutError, httpx.TimeoutException) as exc:
+                timing.data["timeout_source"] = "attempt_deadline" if deadline.expired() else type(exc).__name__
                 result = _err("timeout", 0, _elapsed_ms(t0),
                               f"POST {url} timed out")
             except (httpx.NetworkError, httpx.RequestError) as e:
                 result = _err("network_error", 0, _elapsed_ms(t0), str(e))
-            return result
+            return timing.attach(result)
         finally:
             if slot is not None:
                 await slot.release()
@@ -506,6 +514,27 @@ async def stream_openai_compatible(
     timeout_s: float = 45.0,
     token_providers: dict | None = None,
     provider_rules: dict[str, dict] | None = None,
+    _timing: ProviderTiming | None = None,
+) -> dict:
+    timing = _timing or ProviderTiming(request, (request.get("timeout_ms") or timeout_s * 1000) / 1000, "streaming")
+    result = await _stream_openai_compatible_impl(
+        request, emit, env_get=env_get, extra_headers=extra_headers, client=client,
+        timeout_s=timeout_s, token_providers=token_providers,
+        provider_rules=provider_rules, _timing=timing)
+    return timing.attach(result)
+
+
+async def _stream_openai_compatible_impl(
+    request: dict,
+    emit: Emit,
+    *,
+    client: Any = None,
+    env_get=None,
+    extra_headers: dict | None = None,
+    timeout_s: float = 45.0,
+    token_providers: dict | None = None,
+    provider_rules: dict[str, dict] | None = None,
+    _timing: ProviderTiming,
 ) -> dict:
     """The OpenAI-compatible STREAMING wire backend (sibling of `call`). Returns the
     SAME complete-response dict the non-streaming backend does, so the core's
@@ -556,15 +585,17 @@ def _saw_output() -> bool:
         return saw_output
 
     def _timeout_err() -> dict:
+        _timing.data["timeout_source"] = "first_output_deadline"
         return first_token_timeout_err(first_timeout_s, _latency())
 
     try:
         try:
             async with AsyncExitStack() as stack:
                 try:
+                    _timing.data["phase"] = "connection_pool"
                     resp = await before_first_output(stack.enter_async_context(
                         client.stream("POST", url, json=body, headers=headers,
-                                      timeout=timeout)), first_timeout_s, t0, _saw_output)
+                                      timeout=timeout, **_timing.http_options(client))), first_timeout_s, t0, _saw_output)
                 except (asyncio.TimeoutError, TimeoutError):
                     return _timeout_err()
                 if not (200 <= resp.status_code < 300):
@@ -593,6 +624,7 @@ def _timeout_err() -> dict:
                         chunk = json.loads(data)
                     except ValueError:
                         continue
+                    _timing.observe_chunk(chunk)
                     if raw_model is None:
                         raw_model = chunk.get("model")
                     if chunk.get("usage"):
@@ -644,6 +676,8 @@ def _timeout_err() -> dict:
                 "tokens_cached": _cached_tokens(usage),
                 "cost_reported": usage.get("cost"),
                 "raw_model": raw_model,
+                "upstream": {k: _timing.data[k] for k in ("upstream_id", "upstream_provider", "tokens_reasoning") if k in _timing.data},
+                "tokens_reasoning": _timing.data.get("tokens_reasoning"),
             },
         }
     finally:
@@ -697,6 +731,8 @@ def _parse_openai_response(
                 "tokens_cached": _cached_tokens(usage),
                 "cost_reported": usage.get("cost"),
                 "raw_model": data.get("model"),
+                "upstream": upstream_metadata(data),
+                "tokens_reasoning": upstream_metadata(data).get("tokens_reasoning"),
             },
         }