diff --git a/docs/provider-diagnostics.md b/docs/provider-diagnostics.md new file mode 100644 index 0000000..8d4547b --- /dev/null +++ b/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. diff --git a/host_store.py b/host_store.py index ceee7cd..6f87547 100644 --- a/host_store.py +++ b/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. diff --git a/llm_router_host.py b/llm_router_host.py index 84048d1..1a23751 100644 --- a/llm_router_host.py +++ b/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,10 +642,14 @@ 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']) @@ -650,6 +657,15 @@ async def execute_async(self, contract: dict, call_override=None) -> dict: 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 diff --git a/provider_adapters/diagnostics.py b/provider_adapters/diagnostics.py new file mode 100644 index 0000000..0830764 --- /dev/null +++ b/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 diff --git a/provider_adapters/openai_compatible.py b/provider_adapters/openai_compatible.py index 641b6a1..71ae8c3 100644 --- a/provider_adapters/openai_compatible.py +++ b/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"), }, } diff --git a/shim.py b/shim.py index be75536..77a4147 100644 --- a/shim.py +++ b/shim.py @@ -1904,6 +1904,8 @@ def _openai_usage(response: dict) -> dict: # x_router already passes that 0 through — the OpenAI block must agree. if response.get("tokens_cached") is not None: usage["prompt_tokens_details"] = {"cached_tokens": response["tokens_cached"]} + if response.get("tokens_reasoning") is not None: + usage["completion_tokens_details"] = {"reasoning_tokens": response["tokens_reasoning"]} return usage diff --git a/tests/test_provider_diagnostics.py b/tests/test_provider_diagnostics.py new file mode 100644 index 0000000..b5aa38f --- /dev/null +++ b/tests/test_provider_diagnostics.py @@ -0,0 +1,210 @@ +"""Real socket boundaries identify where a provider timeout happened.""" +import asyncio +import json + +import httpx +import pytest + +from provider_adapters.openai_compatible import make_async_call_provider, stream_openai_compatible +from provider_adapters.diagnostics import ProviderTiming, bounded_diagnostics, upstream_metadata +from tests.test_antseed_concurrency import _req +from tests.test_streaming import FakeStreamClient, FakeStreamResponse, _openai_lines, OPENAI_REQ +from shim import _openai_usage +from host_store import routing_summary +from tests.test_reasoning_controls import host +from tests.test_compact import _PIN +from shim import ChatRequest, _request_to_contract, create_app +from fastapi.testclient import TestClient + + +@pytest.mark.asyncio +@pytest.mark.parametrize("send_headers", [False, True]) +async def test_timeout_distinguishes_headers_from_incomplete_body(send_headers): + tasks = set() + closed = asyncio.Event() + + async def serve(reader, writer): + tasks.add(asyncio.current_task()) + try: + await reader.readuntil(b"\r\n\r\n") + if send_headers: + writer.write(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + await writer.drain() + while True: + if send_headers: + writer.write(b"1\r\n \r\n") + await writer.drain() + await asyncio.sleep(.01) + except (ConnectionError, asyncio.CancelledError): + pass + finally: + writer.close() + closed.set() + + server = await asyncio.start_server(serve, "127.0.0.1", 0) + try: + req = _req("diagnostic", None, timeout_ms=150) + req.update(base_url=f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}/v1", + reasoning_effort="low", max_tokens=8000) + async with httpx.AsyncClient(trust_env=False) as client: + result = await make_async_call_provider(client=client)(req) + assert result["error_kind"] == "timeout" + d = result["diagnostics"] + assert d["phase"] == ("response_body" if send_headers else "response_headers") + assert d["timeout_source"] in ("attempt_deadline", "ReadTimeout") + assert d["request_sent_ms"] < d["elapsed_ms"] + assert d["requested_reasoning_effort"] == "low" + assert d["requested_max_tokens"] == 8000 + assert d["requested_timeout_ms"] == 150 + if send_headers: + assert d["http_status"] == 200 + assert d["response_headers_ms"] < d["elapsed_ms"] + else: + assert "response_headers_ms" not in d + assert "body_complete_ms" not in d + finally: + server.close() + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + await server.wait_closed() + + +@pytest.mark.asyncio +async def test_buffered_preserves_metadata_without_reasoning_text_or_credentials(): + async def handler(request): + assert json.loads(request.content)["reasoning_effort"] == "low" + return httpx.Response(200, json={ + "id": "gen-for-support", "provider": "Example Provider", + "choices": [{"message": {"content": "ok", "reasoning": "PRIVATE REASONING"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 11, "completion_tokens": 9, + "completion_tokens_details": {"reasoning_tokens": 7}}, + }) + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + result = await make_async_call_provider(client=client)( + {**OPENAI_REQ, "reasoning_effort": "low"}) + d = result["diagnostics"] + assert d["upstream_id"] == "gen-for-support" + assert d["upstream_provider"] == "Example Provider" + assert d["tokens_reasoning"] == 7 + assert "PRIVATE REASONING" not in json.dumps(result) + usage = _openai_usage(result["response"]) + assert usage["completion_tokens"] == 9 + assert usage["completion_tokens_details"] == {"reasoning_tokens": 7} + + +@pytest.mark.asyncio +async def test_stream_observes_reasoning_and_tool_output_without_emitting_reasoning(): + lines = [ + 'data: ' + json.dumps({"id": "gen-stream", "provider": "Example", "choices": [ + {"delta": {"reasoning": "PRIVATE REASONING"}}]}), + 'data: ' + json.dumps({"choices": [{"delta": {"tool_calls": [ + {"index": 0, "id": "tool-1", "function": {"name": "shell", "arguments": "{}"}}]}}]}), + 'data: ' + json.dumps({"choices": [{"delta": {}, "finish_reason": "tool_calls"}], + "usage": {"completion_tokens": 10, "completion_tokens_details": {"reasoning_tokens": 0}}}), + 'data: [DONE]', + ] + emitted = [] + async def emit(text): + emitted.append(text) + result = await stream_openai_compatible(OPENAI_REQ, emit, + client=FakeStreamClient(FakeStreamResponse(200, lines=lines))) + assert result["ok"] and not emitted + d = result["diagnostics"] + assert d["first_reasoning_ms"] <= d["first_output_ms"] <= d["elapsed_ms"] + assert d["upstream_id"] == "gen-stream" + assert d["tokens_reasoning"] == 0 + assert "PRIVATE REASONING" not in json.dumps(result) + + +def test_ledger_diagnostics_are_bounded_and_allowlisted(): + secret = "do-not-store" + diag = {"phase": "response_body", "upstream_id": "x" * 1000, + "elapsed_ms": 12, "tokens_reasoning": 0, "connect_ms": float("nan"), + "tls_ms": True, "messages": secret, "headers": secret, "request": secret} + summary = routing_summary({"provider_diagnostics": [diag] * 40}) + assert len(summary["provider_diagnostics"]) == 32 + d = summary["provider_diagnostics"][0] + assert len(d["upstream_id"]) == 160 + assert d["tokens_reasoning"] == 0 + assert "connect_ms" not in d and "tls_ms" not in d + assert secret not in json.dumps(summary) + assert bounded_diagnostics(None) == {} + assert bounded_diagnostics({"elapsed_ms": 10 ** 1000}) == {} + assert upstream_metadata({"usage": {"completion_tokens_details": "malformed"}}) == {} + + +@pytest.mark.asyncio +async def test_trace_does_not_retain_headers_or_exceptions(): + timing = ProviderTiming({}, 40, "buffered") + await timing.trace("http11.receive_response_headers.complete", { + "return_value": (b"HTTP/1.1", 200, b"OK", [(b"authorization", b"secret")])}) + await timing.trace("connection.connect_tcp.failed", {"exception": ValueError("secret")}) + assert timing.data["http_status"] == 200 + assert "secret" not in json.dumps(timing.data) + + +@pytest.mark.parametrize("failed", [False, True]) +def test_http_trace_retains_provider_evidence_on_success_and_failure(host, monkeypatch, failed): + monkeypatch.setattr("llm_router_host._fold_route_outcome", lambda *a, **kw: None) + async def call(req): + common = {"latency_ms": 20, "diagnostics": { + "mode": "buffered", "phase": "response_body", "upstream_id": "gen-proof", + "messages": "must-not-leak", "elapsed_ms": 20}} + return ({**common, "ok": False, "error_kind": "timeout"} if failed else + {**common, "ok": True, "response": {"text": "ok", "tokens_out": 9, "tokens_reasoning": 7}}) + host.set_async_call_hook(call) + response = TestClient(create_app(host)).post("/v1/chat/completions", json={ + "policy_ir": _PIN, "messages": [{"role": "user", "content": "fixture"}]}) + assert response.status_code == (504 if failed else 200) + data = response.json() + d = data["x_router"]["decision_trace"]["provider_diagnostics"][0] + assert d["upstream_id"] == "gen-proof" and d["attempt"] == 1 + assert d["provider_id"] == "comput3" + assert "must-not-leak" not in response.text + if failed: + assert d["error_kind"] == "timeout" + else: + assert data["usage"]["completion_tokens_details"]["reasoning_tokens"] == 7 + assert routing_summary(data["x_router"]["decision_trace"])["provider_diagnostics"][0] == d + + +@pytest.mark.asyncio +async def test_concurrent_host_requests_keep_diagnostics_separate(host, monkeypatch): + monkeypatch.setattr("llm_router_host._fold_route_outcome", lambda *a, **kw: None) + async def call(req): + ident = req["messages"][0]["content"] + await asyncio.sleep(.01 if ident == "a" else 0) + return {"ok": True, "response": {"text": ident}, "diagnostics": {"upstream_id": ident}} + host.set_async_call_hook(call) + results = await asyncio.gather(*(host.execute_async(_request_to_contract(ChatRequest( + policy_ir=_PIN, messages=[{"role": "user", "content": name}]), "default")) for name in ("a", "b"))) + assert [r["trace"]["provider_diagnostics"][0]["upstream_id"] for r in results] == ["a", "b"] + + +@pytest.mark.asyncio +async def test_exhausted_connection_pool_has_no_request_sent_event(): + writers = set() + async def hold(reader, writer): + writers.add(writer) + await reader.readuntil(b"\r\n\r\n") + writer.write(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n") + await writer.drain() + server = await asyncio.start_server(hold, "127.0.0.1", 0) + url = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}" + try: + async with httpx.AsyncClient(trust_env=False, limits=httpx.Limits(max_connections=1)) as client: + async with client.stream("GET", url): + req = {**OPENAI_REQ, "base_url": url, "timeout_ms": 100} + result = await make_async_call_provider(client=client)(req) + assert result["error_kind"] == "timeout" + d = result["diagnostics"] + assert d["phase"] == "connection_pool" + assert d["timeout_source"] in ("attempt_deadline", "PoolTimeout") + assert "request_sent_ms" not in d and "response_headers_ms" not in d + finally: + server.close() + for writer in writers: + writer.close() + await writer.wait_closed() + await server.wait_closed()