From d9ab059ac11ea4bc713943f9be2d2eafe313dcb8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Albert=20Castellana=20LLu=C3=ADs?= Date: Mon, 21 Sep 2026 10:09:11 +0200 Subject: [PATCH 1/2] Preserve provider reasoning and Codex summaries in buffered responses --- README.md | 23 +++++++++ codex_backend.py | 36 +++++++++++++- provider_adapters/openai_compatible.py | 19 ++++++- responses_api.py | 4 +- shim.py | 12 +++-- tests/test_reasoning_capture.py | 68 ++++++++++++++++++++++++++ 6 files changed, 153 insertions(+), 9 deletions(-) create mode 100644 tests/test_reasoning_capture.py diff --git a/README.md b/README.md index b126dfa..5ad4e8e 100644 --- a/README.md +++ b/README.md @@ -224,3 +224,26 @@ chromedriver` for the browser pass — works as before.)* See [generic typed decision/data flows](docs/TYPED-FLOWS.md) for classification, selection and conditional generation. See [decision routing within generative flows](docs/DECISION-FLOWS.md) to select an economical or capable generation policy from conversation and tool history. + +### Provider-exposed reasoning in buffered chat + +OpenAI-compatible adapters retain the upstream assistant message (including +`reasoning`, `reasoning_content` and `reasoning_details`) and usage details through +the public buffered chat response. A reasoning-only response is returned with +its original finish reason, so callers can diagnose an exhausted output budget +instead of losing the result to an empty-content fallback. The stream-backed +buffered adapter also retains exposed reasoning. + +The existing Codex OAuth backend forwards explicit `reasoning.effort` and +`reasoning.summary` controls (or `reasoning_effort`); when effort is specified, +summary defaults to `auto`. It preserves returned Responses reasoning items as +`x_reasoning_items` on chat responses, and exposes readable summaries as typed +`message.reasoning_details`. Encrypted items remain opaque. Responses output +also retains the original reasoning items. These fields describe only data the +provider returned, never undisclosed internal chain-of-thought. + +This does not introduce a new OAuth login path, enable reasoning by default for +all routes, or deploy the router. The Codex endpoint still has its existing +parameter restrictions. The host application remains responsible for private +trace persistence and spend controls. No provider credentials are included in +these returned artifacts. diff --git a/codex_backend.py b/codex_backend.py index 37acb4e..018a49a 100644 --- a/codex_backend.py +++ b/codex_backend.py @@ -168,6 +168,13 @@ def build_codex_body(request: dict) -> dict: tc = _to_responses_tool_choice(request.get("tool_choice")) if tc is not None: body["tool_choice"] = tc + controls = request.get("reasoning") or {} + reasoning = {k: controls[k] for k in ("effort", "summary") if k in controls} + if request.get("reasoning_effort") is not None: + reasoning.setdefault("effort", request["reasoning_effort"]) + if reasoning: + reasoning.setdefault("summary", "auto") + body["reasoning"] = reasoning # The ChatGPT-account Codex endpoint rejects some public Responses API # params even though they are accepted elsewhere. Do not forward max_tokens # as max_output_tokens, and do not forward temperature; live endpoint errors @@ -213,6 +220,8 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict: # function_call items keyed by their streaming item id, in arrival order. fcalls: dict = {} fcorder: list = [] + reasoning_items: dict = {} + summaries: dict = {} for line in lines: line = line.strip() @@ -229,6 +238,12 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict: if etype == "response.output_text.delta": if ev.get("delta"): text_parts.append(ev["delta"]) + elif etype == "response.reasoning_summary_text.delta": + key = (ev.get("item_id", "reasoning"), ev.get("summary_index", 0)) + summaries.setdefault(key, []).append(ev.get("delta") or "") + elif etype == "response.output_item.done" and (ev.get("item") or {}).get("type") == "reasoning": + item = ev["item"] + reasoning_items[item.get("id", "reasoning")] = item elif etype == "response.output_item.added": item = ev.get("item") or {} if item.get("type") == "function_call": @@ -248,9 +263,12 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict: iid = ev.get("item_id") if iid in fcalls and ev.get("arguments") is not None: fcalls[iid]["done"] = ev["arguments"] - elif etype == "response.completed": + elif etype in ("response.completed", "response.incomplete"): resp = ev.get("response") or {} usage = resp.get("usage") or usage + for item in resp.get("output") or []: + if item.get("type") == "reasoning": + reasoning_items[item.get("id", "reasoning")] = item if resp.get("status") == "incomplete": finish_reason = "length" elif etype in ("response.failed", "error"): @@ -260,6 +278,14 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict: if err is not None: return err + for (iid, index), parts in summaries.items(): + if iid not in reasoning_items: + reasoning_items[iid] = {"id": iid, "type": "reasoning", "summary": []} + if not any(p.get("text") == "".join(parts) for p in reasoning_items[iid].get("summary", [])): + # Final items are authoritative; delta-only streams remain explicitly partial. + if not reasoning_items[iid].get("summary") or reasoning_items[iid].get("partial"): + reasoning_items[iid].setdefault("summary", []).append({"type": "summary_text", "text": "".join(parts)}) + reasoning_items[iid]["partial"] = True tool_calls: "list[dict] | None" = None if fcorder: tool_calls = [] @@ -286,6 +312,12 @@ def aggregate_codex_sse(lines: Iterable[str], latency_ms: int) -> dict: "tokens_cached": _cached_tokens(usage), "cost_reported": usage.get("cost"), "raw_model": None, + "reasoning_items": list(reasoning_items.values()), + "provider_message": {"reasoning_details": [ + {"type": "reasoning.summary", "summary": part.get("text", ""), "source": "codex"} + for item in reasoning_items.values() for part in item.get("summary", []) + ]} if reasoning_items else {}, + "tokens_reasoning": (usage.get("output_tokens_details") or {}).get("reasoning_tokens"), }, } @@ -301,7 +333,7 @@ def _codex_line_has_output_delta(line: str) -> bool: ev = json.loads(payload) except ValueError: return False - return ev.get("type") == "response.output_text.delta" and bool(ev.get("delta")) + return ev.get("type") in ("response.output_text.delta", "response.reasoning_summary_text.delta") and bool(ev.get("delta")) def make_codex_async_call_provider( diff --git a/provider_adapters/openai_compatible.py b/provider_adapters/openai_compatible.py index 71ae8c3..55340ff 100644 --- a/provider_adapters/openai_compatible.py +++ b/provider_adapters/openai_compatible.py @@ -571,6 +571,8 @@ async def _stream_openai_compatible_impl( emitted = False text_parts: list[str] = [] + reasoning_parts: dict[str, list[str]] = {} + reasoning_details: list[dict] = [] tool_calls_acc: dict[int, dict] = {} finish_reason = None usage: dict = {} @@ -633,6 +635,13 @@ def _timeout_err() -> dict: delta = choice.get("delta") or {} if choice.get("finish_reason"): finish_reason = choice["finish_reason"] + for key in ("reasoning", "reasoning_content"): + if isinstance(delta.get(key), str): + reasoning_parts.setdefault(key, []).append(delta[key]) + saw_output = True + if isinstance(delta.get("reasoning_details"), list): + reasoning_details.extend(delta["reasoning_details"]) + saw_output = True content = delta.get("content") if content: saw_output = True @@ -661,7 +670,7 @@ def _timeout_err() -> dict: tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)] or None text = "".join(text_parts) - if not text.strip() and not tool_calls: + if not text.strip() and not tool_calls and not reasoning_parts and not reasoning_details: return _err("bad_response", 200, _latency(), "empty assistant content") return { "ok": True, @@ -676,6 +685,9 @@ def _timeout_err() -> dict: "tokens_cached": _cached_tokens(usage), "cost_reported": usage.get("cost"), "raw_model": raw_model, + "provider_message": {**{k: "".join(v) for k, v in reasoning_parts.items()}, + **({"reasoning_details": reasoning_details} if reasoning_details else {})}, + "provider_usage": usage, "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"), }, @@ -716,7 +728,8 @@ def _parse_openai_response( usage = data.get("usage") or {} text = msg.get("content") or "" tool_calls = msg.get("tool_calls") - if not str(text).strip() and not tool_calls: + if not str(text).strip() and not tool_calls and not any( + msg.get(k) for k in ("reasoning", "reasoning_content", "reasoning_details")): return _err("bad_response", status, latency, "empty assistant content") return { "ok": True, @@ -731,6 +744,8 @@ def _parse_openai_response( "tokens_cached": _cached_tokens(usage), "cost_reported": usage.get("cost"), "raw_model": data.get("model"), + "provider_message": msg, + "provider_usage": usage, "upstream": upstream_metadata(data), "tokens_reasoning": upstream_metadata(data).get("tokens_reasoning"), }, diff --git a/responses_api.py b/responses_api.py index 188861e..318afc1 100644 --- a/responses_api.py +++ b/responses_api.py @@ -200,7 +200,7 @@ def result_to_responses_object( or requested_model or "") ts = created_at if created_at is not None else int((now or time.time)()) - output: list[dict] = [] + output: list[dict] = list(resp.get("reasoning_items") or []) if text: output.append({ "type": "message", @@ -239,6 +239,8 @@ def result_to_responses_object( # cache reads correctly (mirrors the chat path's usage.prompt_tokens_details). if resp.get("tokens_cached"): usage["input_tokens_details"] = {"cached_tokens": resp["tokens_cached"]} + if resp.get("tokens_reasoning") is not None: + usage["output_tokens_details"] = {"reasoning_tokens": resp["tokens_reasoning"]} if usage: obj["usage"] = usage return obj diff --git a/shim.py b/shim.py index 77a4147..3a3c57c 100644 --- a/shim.py +++ b/shim.py @@ -1892,7 +1892,7 @@ def _openai_usage(response: dict) -> dict: shared by the unary chat body, the streaming final chunk and /v1/compact. Empty dict when the provider reported no token counts (caller omits the key, per the additive wire contract).""" - usage: dict = {} + usage: dict = dict(response.get("provider_usage") or {}) for src_key, dst_key in (("tokens_in", "prompt_tokens"), ("tokens_out", "completion_tokens"), ("tokens_total", "total_tokens")): @@ -1903,9 +1903,9 @@ def _openai_usage(response: dict) -> dict: # explicit tokens_cached: 0 means caching was evaluated with no hits, and # 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"]} + usage["prompt_tokens_details"] = {**(usage.get("prompt_tokens_details") or {}), "cached_tokens": response["tokens_cached"]} if response.get("tokens_reasoning") is not None: - usage["completion_tokens_details"] = {"reasoning_tokens": response["tokens_reasoning"]} + usage["completion_tokens_details"] = {**(usage.get("completion_tokens_details") or {}), "reasoning_tokens": response["tokens_reasoning"]} return usage @@ -1971,7 +1971,8 @@ def _router_response_to_openai(result: dict, requested_model: str, response = result.get("response") or {} chosen = result.get("chosen") or {} - message: dict = {"role": "assistant", "content": response.get("text") or ""} + message: dict = dict(response.get("provider_message") or {}) + message.update(role="assistant", content=response.get("text") or "") if response.get("tool_calls"): message["tool_calls"] = response["tool_calls"] @@ -1992,6 +1993,9 @@ def _router_response_to_openai(result: dict, requested_model: str, }], } + if response.get("reasoning_items"): + out["x_reasoning_items"] = response["reasoning_items"] + usage = _openai_usage(response) if usage: out["usage"] = usage diff --git a/tests/test_reasoning_capture.py b/tests/test_reasoning_capture.py new file mode 100644 index 0000000..e50b3bb --- /dev/null +++ b/tests/test_reasoning_capture.py @@ -0,0 +1,68 @@ +"""No paid calls: provider-exposed reasoning survives the buffered wire boundary.""" +import httpx +import pytest +from provider_adapters.openai_compatible import make_async_call_provider +from shim import _router_response_to_openai +from tests.test_streaming import OPENAI_REQ + +@pytest.mark.asyncio +@pytest.mark.parametrize("content,finish", [('{}', 'stop'), ('', 'length')]) +async def test_reasoning_and_usage_survive_adapter_and_public_chat(content, finish): + message = {"role": "assistant", "content": content, "reasoning": "Exposed explanation", + "reasoning_details": [{"type": "reasoning.summary", "summary": "summary"}, + {"type": "reasoning.encrypted", "data": "opaque"}]} + usage = {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30, + "completion_tokens_details": {"reasoning_tokens": 15}, "cost": 0.001} + async def respond(request): + return httpx.Response(200, json={"model": "fixture", "choices": [ + {"message": message, "finish_reason": finish}], "usage": usage}) + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + result = await make_async_call_provider(client=client)(OPENAI_REQ) + assert result['ok'] + public = _router_response_to_openai(result, 'fixture') + assert public['choices'][0]['message'] == message + assert public['choices'][0]['finish_reason'] == finish + assert public['usage']['completion_tokens_details']['reasoning_tokens'] == 15 + assert public['usage']['cost'] == .001 + +from tests.test_reasoning_controls import host +from tests.test_compact import _PIN +from fastapi.testclient import TestClient +from shim import create_app + +def test_reasoning_survives_real_engine_and_chat_endpoint(host): + async def provider(req): + return {"ok": True, "response": {"text": "{}", "provider_message": { + "reasoning": "explanation", "reasoning_details": [{"type": "reasoning.summary", "summary": "summary"}]}, + "provider_usage": {"completion_tokens_details": {"reasoning_tokens": 8}}, "tokens_out": 10}} + host.set_async_call_hook(provider) + response = TestClient(create_app(host)).post('/v1/chat/completions', json={"messages": [], "policy_ir": _PIN}) + assert response.status_code == 200 + assert response.json()['choices'][0]['message']['reasoning'] == 'explanation' + +@pytest.mark.asyncio +async def test_buffered_sse_keeps_reasoning(): + import json + from tests.test_streaming import FakeStreamClient, FakeStreamResponse + chunks = [{"choices": [{"delta": {"reasoning": "Think "}}]}, + {"choices": [{"delta": {"reasoning": "again", "content": "{}"}, "finish_reason": "stop"}]}] + client = FakeStreamClient(FakeStreamResponse(200, ['data: '+json.dumps(c) for c in chunks]+['data: [DONE]'])) + result = await make_async_call_provider(client=client)({**OPENAI_REQ, "first_token_timeout_ms": 1000}) + assert result['ok'] + assert result['response']['provider_message']['reasoning'] == 'Think again' + +def test_codex_oauth_summary_survives_without_exposing_internal_thought(): + import json + from codex_backend import build_codex_body, aggregate_codex_sse + req = build_codex_body({"served_model_id": "fixture", "reasoning_effort": "low"}) + assert req['reasoning'] == {'effort': 'low', 'summary': 'auto'} + item = {"id": "r1", "type": "reasoning", "summary": [{"type": "summary_text", "text": "Exposed summary"}], "encrypted_content": "opaque"} + events = [{"type": "response.reasoning_summary_text.delta", "item_id": "r1", "summary_index": 0, "delta": "Exposed summary"}, + {"type": "response.output_item.done", "item": item}, + {"type": "response.output_text.delta", "delta": "{}"}, + {"type": "response.completed", "response": {"output": [item], "usage": {"output_tokens": 25, "output_tokens_details": {"reasoning_tokens": 20}}}}] + result = aggregate_codex_sse(['data: '+json.dumps(e) for e in events], 1) + public = _router_response_to_openai(result, 'fixture') + assert public['x_reasoning_items'] == [item] + assert public['choices'][0]['message']['reasoning_details'][0]['summary'] == 'Exposed summary' + assert public['usage']['completion_tokens_details']['reasoning_tokens'] == 20 From 4c2364b58f49d6548a5187a04611c28e70620b9e Mon Sep 17 00:00:00 2001 From: jmlago Date: Mon, 21 Sep 2026 11:09:26 +0200 Subject: [PATCH 2/2] fix: validate reasoning payloads and replay Responses items consistently --- README.md | 6 ++- provider_adapters/openai_compatible.py | 26 +++++++--- responses_api.py | 20 ++++++- tests/test_provider_diagnostics.py | 10 ++-- tests/test_reasoning_capture.py | 72 +++++++++++++++++++++++++- tests/test_responses_api.py | 37 +++++++++++++ tests/test_responses_shim.py | 21 ++++++++ 7 files changed, 180 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 5ad4e8e..11ec23d 100644 --- a/README.md +++ b/README.md @@ -233,13 +233,17 @@ the public buffered chat response. A reasoning-only response is returned with its original finish reason, so callers can diagnose an exhausted output budget instead of losing the result to an empty-content fallback. The stream-backed buffered adapter also retains exposed reasoning. +Empty or whitespace-only reasoning and detail metadata without a payload do not +count as output: they retain empty-response fallback and first-output deadlines. The existing Codex OAuth backend forwards explicit `reasoning.effort` and `reasoning.summary` controls (or `reasoning_effort`); when effort is specified, summary defaults to `auto`. It preserves returned Responses reasoning items as `x_reasoning_items` on chat responses, and exposes readable summaries as typed `message.reasoning_details`. Encrypted items remain opaque. Responses output -also retains the original reasoning items. These fields describe only data the +also retains the original reasoning items. Its buffered SSE replay emits each +reasoning item and summary before subsequent text or tool items, with contiguous +output indexes. These fields describe only data the provider returned, never undisclosed internal chain-of-thought. This does not introduce a new OAuth login path, enable reasoning by default for diff --git a/provider_adapters/openai_compatible.py b/provider_adapters/openai_compatible.py index 55340ff..c5b72e4 100644 --- a/provider_adapters/openai_compatible.py +++ b/provider_adapters/openai_compatible.py @@ -30,6 +30,20 @@ Emit = Callable[[str], Awaitable[None]] +def _has_reasoning_content(message: dict) -> bool: + """Empty deltas and detail metadata alone are not generated output.""" + def nonblank(value): + return isinstance(value, str) and bool(value.strip()) + + if any(nonblank(message.get(key)) for key in ("reasoning", "reasoning_content")): + return True + details = message.get("reasoning_details") + return isinstance(details, list) and any( + isinstance(part, dict) and any(nonblank(part.get(key)) + for key in ("text", "summary", "data", "encrypted_content")) + for part in details) + + def _resolve_auth_headers( request: dict, env_get: Callable[[str], str | None], @@ -638,9 +652,9 @@ def _timeout_err() -> dict: for key in ("reasoning", "reasoning_content"): if isinstance(delta.get(key), str): reasoning_parts.setdefault(key, []).append(delta[key]) - saw_output = True if isinstance(delta.get("reasoning_details"), list): reasoning_details.extend(delta["reasoning_details"]) + if _has_reasoning_content(delta): saw_output = True content = delta.get("content") if content: @@ -670,7 +684,9 @@ def _timeout_err() -> dict: tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)] or None text = "".join(text_parts) - if not text.strip() and not tool_calls and not reasoning_parts and not reasoning_details: + provider_message = {**{k: "".join(v) for k, v in reasoning_parts.items()}, + **({"reasoning_details": reasoning_details} if reasoning_details else {})} + if not text.strip() and not tool_calls and not _has_reasoning_content(provider_message): return _err("bad_response", 200, _latency(), "empty assistant content") return { "ok": True, @@ -685,8 +701,7 @@ def _timeout_err() -> dict: "tokens_cached": _cached_tokens(usage), "cost_reported": usage.get("cost"), "raw_model": raw_model, - "provider_message": {**{k: "".join(v) for k, v in reasoning_parts.items()}, - **({"reasoning_details": reasoning_details} if reasoning_details else {})}, + "provider_message": provider_message, "provider_usage": usage, "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"), @@ -728,8 +743,7 @@ def _parse_openai_response( usage = data.get("usage") or {} text = msg.get("content") or "" tool_calls = msg.get("tool_calls") - if not str(text).strip() and not tool_calls and not any( - msg.get(k) for k in ("reasoning", "reasoning_content", "reasoning_details")): + if not str(text).strip() and not tool_calls and not _has_reasoning_content(msg): return _err("bad_response", status, latency, "empty assistant content") return { "ok": True, diff --git a/responses_api.py b/responses_api.py index 318afc1..a7277c4 100644 --- a/responses_api.py +++ b/responses_api.py @@ -283,7 +283,25 @@ def _emit(event: str, data: dict) -> str: return frame for out_index, item in enumerate(obj.get("output") or []): - if item.get("type") == "message": + if item.get("type") == "reasoning": + yield _emit("response.output_item.added", + {"output_index": out_index, "item": {**item, "summary": []}}) + for summary_index, part in enumerate(item.get("summary") or []): + position = {"item_id": item["id"], "output_index": out_index, + "summary_index": summary_index} + text = part.get("text", "") + yield _emit("response.reasoning_summary_part.added", + {**position, "part": {**part, "text": ""}}) + if text: + yield _emit("response.reasoning_summary_text.delta", + {**position, "delta": text}) + yield _emit("response.reasoning_summary_text.done", + {**position, "text": text}) + yield _emit("response.reasoning_summary_part.done", + {**position, "part": part}) + yield _emit("response.output_item.done", + {"output_index": out_index, "item": item}) + elif item.get("type") == "message": text = (item.get("content") or [{}])[0].get("text", "") yield _emit("response.output_item.added", {"output_index": out_index, "item": {**item, "content": []}}) diff --git a/tests/test_provider_diagnostics.py b/tests/test_provider_diagnostics.py index b5aa38f..4cc3a7e 100644 --- a/tests/test_provider_diagnostics.py +++ b/tests/test_provider_diagnostics.py @@ -71,7 +71,7 @@ async def serve(reader, writer): @pytest.mark.asyncio -async def test_buffered_preserves_metadata_without_reasoning_text_or_credentials(): +async def test_buffered_preserves_reasoning_only_in_response_not_diagnostics(): async def handler(request): assert json.loads(request.content)["reasoning_effort"] == "low" return httpx.Response(200, json={ @@ -87,7 +87,9 @@ async def handler(request): 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) + assert result["response"]["provider_message"]["reasoning"] == "PRIVATE REASONING" + assert "PRIVATE REASONING" not in json.dumps(d) + assert "PRIVATE REASONING" not in json.dumps(routing_summary({"provider_diagnostics": [d]})) usage = _openai_usage(result["response"]) assert usage["completion_tokens"] == 9 assert usage["completion_tokens_details"] == {"reasoning_tokens": 7} @@ -114,7 +116,9 @@ async def emit(text): 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) + assert result["response"]["provider_message"]["reasoning"] == "PRIVATE REASONING" + assert "PRIVATE REASONING" not in json.dumps(d) + assert "PRIVATE REASONING" not in json.dumps(routing_summary({"provider_diagnostics": [d]})) def test_ledger_diagnostics_are_bounded_and_allowlisted(): diff --git a/tests/test_reasoning_capture.py b/tests/test_reasoning_capture.py index e50b3bb..cab6831 100644 --- a/tests/test_reasoning_capture.py +++ b/tests/test_reasoning_capture.py @@ -34,11 +34,20 @@ def test_reasoning_survives_real_engine_and_chat_endpoint(host): async def provider(req): return {"ok": True, "response": {"text": "{}", "provider_message": { "reasoning": "explanation", "reasoning_details": [{"type": "reasoning.summary", "summary": "summary"}]}, - "provider_usage": {"completion_tokens_details": {"reasoning_tokens": 8}}, "tokens_out": 10}} + "provider_usage": {"completion_tokens_details": {"reasoning_tokens": 8}}, "tokens_out": 10}, + "diagnostics": {"upstream_id": "fixture-id", "messages": "explanation", + "headers": {"Authorization": "fixture-credential"}}} host.set_async_call_hook(provider) response = TestClient(create_app(host)).post('/v1/chat/completions', json={"messages": [], "policy_ir": _PIN}) assert response.status_code == 200 assert response.json()['choices'][0]['message']['reasoning'] == 'explanation' + import json + from host_store import routing_summary + trace = response.json()['x_router']['decision_trace'] + assert trace['provider_diagnostics'][0]['upstream_id'] == 'fixture-id' + assert 'explanation' not in json.dumps(trace) + assert 'explanation' not in json.dumps(routing_summary(trace)) + assert 'fixture-credential' not in response.text @pytest.mark.asyncio async def test_buffered_sse_keeps_reasoning(): @@ -66,3 +75,64 @@ def test_codex_oauth_summary_survives_without_exposing_internal_thought(): assert public['x_reasoning_items'] == [item] assert public['choices'][0]['message']['reasoning_details'][0]['summary'] == 'Exposed summary' assert public['usage']['completion_tokens_details']['reasoning_tokens'] == 20 + + +EMPTY_REASONING = [ + {"reasoning": ""}, {"reasoning": " \n\t"}, {"reasoning_content": ""}, + {"reasoning_details": []}, {"reasoning_details": [{}]}, + {"reasoning_details": [{"type": "reasoning.text", "text": " "}]}, + {"reasoning_details": [{"type": "reasoning.encrypted", "id": "r1", "index": 0}]}, +] +REAL_REASONING = [ + {"reasoning": "Exposed explanation"}, {"reasoning_content": " Exposed explanation\n"}, + {"reasoning_details": [{"type": "reasoning.summary", "summary": "Summary"}]}, + {"reasoning_details": [{"type": "reasoning.text", "text": "Explanation"}]}, + {"reasoning_details": [{"type": "reasoning.encrypted", "data": "opaque"}]}, +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stream_backed", [False, True]) +@pytest.mark.parametrize("fields", EMPTY_REASONING + REAL_REASONING) +async def test_reasoning_only_requires_nonempty_payload(fields, stream_backed): + import json + from tests.test_streaming import FakeStreamClient, FakeStreamResponse + message = {"role": "assistant", "content": "", **fields} + expected_ok = fields in REAL_REASONING + if stream_backed: + chunks = [{"choices": [{"delta": fields, "finish_reason": "length"}]}] + client = FakeStreamClient(FakeStreamResponse(200, + ['data: '+json.dumps(c) for c in chunks] + ['data: [DONE]'])) + result = await make_async_call_provider(client=client)({**OPENAI_REQ, "first_token_timeout_ms": 1000}) + else: + async with httpx.AsyncClient(transport=httpx.MockTransport(lambda _: httpx.Response(200, + json={"choices": [{"message": message, "finish_reason": "length"}]}))) as client: + result = await make_async_call_provider(client=client)(OPENAI_REQ) + assert result["ok"] is expected_ok + if expected_ok: + public = _router_response_to_openai(result, "fixture") + assert public["choices"][0]["finish_reason"] == "length" + for key, value in fields.items(): + assert public["choices"][0]["message"][key] == value + else: + assert result["error_kind"] == "bad_response" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("fields", EMPTY_REASONING) +async def test_empty_reasoning_does_not_disable_first_output_deadline(fields): + import asyncio + import json + from tests.test_streaming import FakeStreamClient, FakeStreamResponse + + class DelayedOutput(FakeStreamResponse): + async def aiter_lines(self): + yield 'data: ' + json.dumps({"choices": [{"delta": fields}]}) + await asyncio.sleep(.05) + yield 'data: ' + json.dumps({"choices": [{"delta": {"content": "late"}}]}) + yield 'data: [DONE]' + + result = await make_async_call_provider(client=FakeStreamClient(DelayedOutput(200)))( + {**OPENAI_REQ, "first_token_timeout_ms": 10}) + assert not result["ok"] and result["error_kind"] == "timeout" + assert result["diagnostics"]["timeout_source"] == "first_output_deadline" diff --git a/tests/test_responses_api.py b/tests/test_responses_api.py index 0b320e9..e09aa2c 100644 --- a/tests/test_responses_api.py +++ b/tests/test_responses_api.py @@ -5,6 +5,7 @@ from __future__ import annotations import json +import pytest import sys from pathlib import Path @@ -338,3 +339,39 @@ def test_sse_events_sequence_numbers_increase(): created_at=1) seqs = [d["sequence_number"] for _, d in _parse_sse(list(ra.responses_sse_events(obj)))] assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs) + + +@pytest.mark.parametrize("summary", [[], [{"type": "summary_text", "text": ""}], + [{"type": "summary_text", "text": "First"}, {"type": "summary_text", "text": "Second"}]]) +def test_reasoning_sse_reconstructs_every_output_item_and_summary(summary): + import copy + reasoning = {"id": "rs_1", "type": "reasoning", "summary": summary, + "encrypted_content": "opaque"} + result = _result(text="Answer", tool_calls=[{"id": "call_1", "type": "function", + "function": {"name": "shell", "arguments": "{}"}}]) + result["response"]["reasoning_items"] = [reasoning] + obj = ra.result_to_responses_object(result) + original = copy.deepcopy(obj) + reconstructed = [] + summaries = {} + events = _parse_sse(list(ra.responses_sse_events(obj))) + for event, data in events: + if event == "response.output_item.added": + assert data["output_index"] == len(reconstructed) + reconstructed.append(data["item"]) + elif event == "response.reasoning_summary_part.added": + assert reconstructed[data["output_index"]]["id"] == data["item_id"] + summaries[data["summary_index"]] = data["part"]["text"] + elif event == "response.reasoning_summary_text.delta": + summaries[data["summary_index"]] += data["delta"] + elif event == "response.reasoning_summary_text.done": + assert summaries[data["summary_index"]] == data["text"] + elif event == "response.reasoning_summary_part.done": + assert summaries[data["summary_index"]] == data["part"]["text"] + elif event == "response.output_item.done": + reconstructed[data["output_index"]] = data["item"] + assert reconstructed == obj["output"] == events[-1][1]["response"]["output"] + assert [summaries[i] for i in range(len(summary))] == [p["text"] for p in summary] + assert reconstructed[0]["encrypted_content"] == "opaque" + assert obj == original + assert [d["sequence_number"] for _, d in events] == list(range(1, len(events)+1)) diff --git a/tests/test_responses_shim.py b/tests/test_responses_shim.py index f6c90c5..3396c5e 100644 --- a/tests/test_responses_shim.py +++ b/tests/test_responses_shim.py @@ -178,3 +178,24 @@ async def _slow_error(contract): seqs = [d["sequence_number"] for d in datas if "sequence_number" in d] assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs), \ f"sequence_numbers must be strictly increasing, got {seqs}" + + +def test_responses_stream_replays_reasoning_before_text_and_tools(client, host): + item = {"id": "rs_fixture", "type": "reasoning", "summary": [ + {"type": "summary_text", "text": "Exposed summary"}], "encrypted_content": "opaque"} + result = _ok(text="Answer", tool_calls=[{"id": "call_1", "type": "function", + "function": {"name": "shell", "arguments": "{}"}}]) + result["response"]["reasoning_items"] = [item] + _seed(host, result) + response = client.post("/v1/responses", json={"input": "hi", "stream": True}) + assert response.status_code == 200 + events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith("data: ")] + added = [e for e in events if e["type"] == "response.output_item.added"] + done = [e for e in events if e["type"] == "response.output_item.done"] + assert [e["output_index"] for e in added] == [0, 1, 2] + assert [e["item"]["type"] for e in added] == ["reasoning", "message", "function_call"] + completed = next(e["response"] for e in events if e["type"] == "response.completed") + assert [e["item"] for e in done] == completed["output"] + assert completed["output"][0] == item + assert any(e["type"] == "response.reasoning_summary_text.delta" and + e["delta"] == "Exposed summary" for e in events)