From 50f34e97c23224301876efe6e6c5cd49618be419 Mon Sep 17 00:00:00 2001 From: arminanton <29869547+arminanton@users.noreply.github.com> Date: Sat, 22 Aug 2026 14:35:56 -0700 Subject: [PATCH] feat(reasoning): expose structured reasoning across API server Expose model reasoning/thinking on /v1/chat/completions and /v1/responses, gated on display.platforms.api_server.show_reasoning (off by default = byte-identical wire). Sweeper fix: both endpoints source STRUCTURED reasoning via the agent's reasoning_callback (fired by run_agent._fire_reasoning_delta from native reasoning_content / thinking deltas), NOT the reasoning.available progress event. conversation_loop derives reasoning.available from assistant_message.content, so sourcing it would serialize answer text as reasoning and drop native thinking deltas. - chat/completions: wires reasoning_callback in both stream and non-stream. Stream emits delta.reasoning_content chunks; non-stream adds a reasoning_content sibling of content. - responses: wires reasoning_callback into the SSE writer, emitted as the spec reasoning event family (output_item.added -> reasoning_summary_part.added -> reasoning_summary_text.delta -> .done -> output_item.done); non-stream via _extract_output_items(include_reasoning). Input hardening skips echoed-back reasoning items (no empty-turn 400s). - gate _reasoning_exposure_enabled() fails closed (never 500). - reasoning_callback threaded through _create_agent / _run_agent. thinking.display control (agent/anthropic_adapter.py): _resolve_thinking_display reads reasoning_config["display"] (summarized|omitted); default summarized, unknown values fall back so a bad value can't make an invalid request. Dropped the usage_pricing reasoning_tokens hunk from the original PR: already on main (3a122ba4a). Addresses the inline comment (use reasoning_callback, not tool_progress_callback / reasoning.available). Tests: 9 thinking.display resolver + 17 api_server reasoning (gate, chat stream/non-stream, responses stream/non-stream, extraction, input hardening). --- agent/anthropic_adapter.py | 31 +- gateway/platforms/api_server.py | 230 ++++++++- tests/gateway/test_api_server_reasoning.py | 554 +++++++++++++++++++++ tests/test_anthropic_thinking_display.py | 82 +++ 4 files changed, 889 insertions(+), 8 deletions(-) create mode 100644 tests/gateway/test_api_server_reasoning.py create mode 100644 tests/test_anthropic_thinking_display.py diff --git a/agent/anthropic_adapter.py b/agent/anthropic_adapter.py index ce311fa1e358..55d6ad3aae8b 100644 --- a/agent/anthropic_adapter.py +++ b/agent/anthropic_adapter.py @@ -2904,6 +2904,30 @@ def convert_messages_to_anthropic( return system, result +# Valid Anthropic thinking.display values (per SDK ThinkingConfig*Param stubs). +_THINKING_DISPLAY_VALUES = ("summarized", "omitted") + + +def _resolve_thinking_display(reasoning_config: Optional[Dict[str, Any]]) -> str: + """Resolve the Anthropic ``thinking.display`` value for a request. + + Reads an optional ``reasoning_config["display"]`` override and validates it + against the SDK-supported values (``"summarized"`` | ``"omitted"``). + + Default is ``"summarized"`` so Hermes always receives the reasoning text for + its CLI activity feed and the API-server reasoning-exposure paths. A caller + that wants to suppress reasoning text on the wire (save returned tokens; the + model still thinks, the signature is still returned for continuity) sets + ``reasoning_config["display"] = "omitted"``. Any unknown value falls back to + ``"summarized"`` so a bad config never produces an invalid request. + """ + if isinstance(reasoning_config, dict): + raw = reasoning_config.get("display") + if isinstance(raw, str) and raw.strip().lower() in _THINKING_DISPLAY_VALUES: + return raw.strip().lower() + return "summarized" + + def build_anthropic_kwargs( model: str, messages: List[Dict], @@ -3102,9 +3126,14 @@ def _to_oauth_wire_name(name: str) -> str: effort = str(reasoning_config.get("effort", "medium")).lower() budget = THINKING_BUDGET.get(effort, 8000) if _supports_adaptive_thinking(model): + # thinking.display defaults to "summarized" so Hermes always + # has reasoning to surface (CLI feed + API-server exposure). + # A caller can pass reasoning_config["display"] = "omitted" to + # suppress the reasoning text on the wire while the model keeps + # thinking and the signature is preserved for continuity. kwargs["thinking"] = { "type": "adaptive", - "display": "summarized", + "display": _resolve_thinking_display(reasoning_config), } adaptive_effort = ADAPTIVE_EFFORT_MAP.get(effort, "medium") # Downgrade xhigh→max on models that don't list xhigh as a diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index 980659c9343d..152f31850d17 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -2811,6 +2811,31 @@ def _request_route_conflict_error( ) return None + def _reasoning_exposure_enabled(self) -> bool: + """True when the API server should expose model reasoning on the wire. + + Honors ``display.platforms.api_server.show_reasoning`` via the standard + per-platform display resolver (#7556). Off by default so the wire + format is byte-identical for existing clients (#37044, #21655); an + operator opts in per platform. + + Broad ``except`` on purpose: a config that fails to load must never 500 + an otherwise-valid /v1/chat/completions or /v1/responses request, so the + gate fails closed (reasoning off) and logs why at debug level. + """ + try: + from gateway.display_config import resolve_display_setting + from gateway.run import _load_gateway_config + + return bool(resolve_display_setting( + _load_gateway_config(), "api_server", "show_reasoning", False, + )) + except Exception as exc: + logger.debug( + "show_reasoning gate resolution failed; defaulting to off: %s", exc + ) + return False + def _create_agent( self, ephemeral_system_prompt: Optional[str] = None, @@ -2819,6 +2844,7 @@ def _create_agent( tool_progress_callback=None, tool_start_callback=None, tool_complete_callback=None, + reasoning_callback=None, gateway_session_key: Optional[str] = None, requested_model: Optional[str] = None, requested_provider: Optional[str] = None, @@ -3132,6 +3158,7 @@ def _resolve_provider_runtime( "tool_progress_callback": tool_progress_callback, "tool_start_callback": tool_start_callback, "tool_complete_callback": tool_complete_callback, + "reasoning_callback": reasoning_callback, "session_db": self._ensure_session_db(), "fallback_model": fallback_model, "reasoning_config": reasoning_config, @@ -5168,6 +5195,14 @@ async def _handle_chat_completions(self, request: "web.Request") -> "web.Respons model_name = body.get("model", self._model_name) created = int(time.time()) + # Resolve reasoning exposure once, before the stream/non-stream branch. + # When on, chat/completions surfaces the model's STRUCTURED reasoning + # (the reasoning_callback / _fire_reasoning_delta path, NOT the + # reasoning.available progress event which carries stripped answer + # content) as delta.reasoning_content (stream) or a reasoning_content + # sibling of content (non-stream). Off by default = byte-identical wire. + show_reasoning = self._reasoning_exposure_enabled() + # Per-client model routing: if the requested model matches a # configured model_routes alias, this request's agent is created # with that route's model/provider instead of the global default. @@ -5203,6 +5238,19 @@ def _on_delta(delta): if delta is not None: _stream_q.put_threadsafe(delta) + # Structured reasoning → delta.reasoning_content chunks. This is + # the agent's reasoning_callback (fired by _fire_reasoning_delta on + # native reasoning_content / thinking deltas), NOT the + # reasoning.available progress event, which conversation_loop + # derives from assistant_message.content and would serialize the + # answer text as reasoning. Gated by show_reasoning; None when off + # so run_agent skips the callback entirely (unchanged wire format). + _on_reasoning = None + if show_reasoning: + def _on_reasoning(text): + if text: + _stream_q.put_threadsafe(("__reasoning__", str(text))) + # Track which tool_call_ids we've emitted a "running" lifecycle # event for, so a "completed" event without a matching "running" # (e.g. internal/filtered tools) is silently dropped instead of @@ -5268,6 +5316,7 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul stream_delta_callback=_on_delta, tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete, + reasoning_callback=_on_reasoning, agent_ref=agent_ref, gateway_session_key=gateway_session_key, **agent_overrides, @@ -5283,13 +5332,25 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul gateway_session_key=gateway_session_key, ) - # Non-streaming: run the agent (with optional Idempotency-Key) + # Non-streaming: run the agent (with optional Idempotency-Key). + # Reasoning is sourced from the agent's structured reasoning_callback + # (fired once with the turn's reasoning text when not streaming — see + # chat_completion_helpers), NOT from reasoning.available. Accumulated + # here and surfaced as a reasoning_content sibling of content when + # show_reasoning is on. + _reasoning_parts: List[str] = [] + + def _collect_reasoning(text): + if text: + _reasoning_parts.append(str(text)) + async def _compute_completion(): return await self._run_agent( user_message=user_message, conversation_history=history, ephemeral_system_prompt=system_prompt, session_id=session_id, + reasoning_callback=_collect_reasoning if show_reasoning else None, gateway_session_key=gateway_session_key, **agent_overrides, route=route, @@ -5374,6 +5435,9 @@ async def _compute_completion(): "message": { "role": "assistant", "content": final_response, + **({ + "reasoning_content": "".join(_reasoning_parts), + } if show_reasoning and _reasoning_parts else {}), }, "finish_reason": finish_reason, } @@ -5445,13 +5509,25 @@ async def _emit(item): """Write a single queue item to the SSE stream. Plain strings are sent as normal ``delta.content`` chunks. + Tagged tuples ``("__reasoning__", text)`` are sent as + ``delta.reasoning_content`` chunks — the shape Open WebUI and + other OpenAI-compatible frontends parse for a collapsible + thinking block. Only enqueued when show_reasoning is on and the + text came from the agent's structured reasoning_callback. Tagged tuples ``("__tool_progress__", payload)`` are sent as a custom ``event: hermes.tool.progress`` SSE event so frontends can display them without storing the markers in conversation history. See #6972 for the original event, #16588 for the ``toolCallId``/``status`` lifecycle fields. """ - if isinstance(item, tuple) and len(item) == 2 and item[0] == "__tool_progress__": + if isinstance(item, tuple) and len(item) == 2 and item[0] == "__reasoning__": + reasoning_chunk = { + "id": completion_id, "object": "chat.completion.chunk", + "created": created, "model": model, + "choices": [{"index": 0, "delta": {"reasoning_content": item[1]}, "finish_reason": None}], + } + await response.write(_sse_frame(reasoning_chunk)) + elif isinstance(item, tuple) and len(item) == 2 and item[0] == "__tool_progress__": await response.write(_sse_frame(item[1], event="hermes.tool.progress")) else: content_chunk = { @@ -5904,6 +5980,87 @@ async def _emit_tool_completed(payload: Dict[str, Any]) -> None: "item": output_item, }) + # ── Reasoning item state (#21655, #7556) ── + # Structured reasoning streams as deltas via the agent's + # reasoning_callback (NOT reasoning.available, which carries the + # stripped answer content). Deltas accumulate into one open + # ``reasoning`` output item per burst; the item closes when the + # model moves on (tool start, answer text, or end of stream). + # Emitted as the spec reasoning event family so Responses clients + # (e.g. Open WebUI) render a live collapsible thinking block: + # output_item.added → reasoning_summary_part.added → + # reasoning_summary_text.delta → .done → output_item.done. + _open_reasoning: Optional[Dict[str, Any]] = None + + async def _emit_reasoning_delta(text: str) -> None: + """Open (if needed) the current reasoning item and append a delta.""" + nonlocal _open_reasoning, output_index + if _open_reasoning is None: + if not text.strip(): + return # never open an item for leading whitespace + idx = output_index + output_index += 1 + _open_reasoning = { + "id": f"rs_{uuid.uuid4().hex[:24]}", + "idx": idx, + "parts": [], + } + await _write_event("response.output_item.added", { + "type": "response.output_item.added", + "output_index": idx, + "item": { + "id": _open_reasoning["id"], + "type": "reasoning", + "summary": [], + "status": "in_progress", + }, + }) + await _write_event("response.reasoning_summary_part.added", { + "type": "response.reasoning_summary_part.added", + "item_id": _open_reasoning["id"], + "output_index": idx, + "summary_index": 0, + "part": {"type": "summary_text", "text": ""}, + }) + _open_reasoning["parts"].append(text) + await _write_event("response.reasoning_summary_text.delta", { + "type": "response.reasoning_summary_text.delta", + "item_id": _open_reasoning["id"], + "output_index": _open_reasoning["idx"], + "summary_index": 0, + "delta": text, + }) + + async def _close_reasoning() -> None: + """Finalize the open reasoning item, if any.""" + nonlocal _open_reasoning + if _open_reasoning is None: + return + text = "".join(_open_reasoning["parts"]) + await _write_event("response.reasoning_summary_text.done", { + "type": "response.reasoning_summary_text.done", + "item_id": _open_reasoning["id"], + "output_index": _open_reasoning["idx"], + "summary_index": 0, + "text": text, + }) + done_item = { + "id": _open_reasoning["id"], + "type": "reasoning", + "summary": [{"type": "summary_text", "text": text}], + "status": "completed", + } + await _write_event("response.output_item.done", { + "type": "response.output_item.done", + "output_index": _open_reasoning["idx"], + "item": done_item, + }) + # The same dict lands in the final envelope so the streamed + # and stored shapes cannot drift (id/status preserved for + # clients that correlate by item id). + emitted_items.append(done_item) + _open_reasoning = None + # Main drain loop — thread-safe queue fed by agent callbacks. async def _dispatch(it) -> None: """Route a queue item to the correct SSE emitter. @@ -5912,19 +6069,25 @@ async def _dispatch(it) -> None: to reduce Open WebUI re-render storms. Tagged tuples with ``__tool_started__`` / ``__tool_completed__`` prefixes are tool lifecycle events and flush the buffer - before emitting. + before emitting. ``__reasoning_delta__`` tuples carry + structured reasoning that accumulates into a reasoning item. """ nonlocal _batch_timer if isinstance(it, tuple) and len(it) == 2 and isinstance(it[0], str): tag, payload = it - # Flush batched text before tool events + # Flush batched text before tool / reasoning-close events if _batch_buf: await _flush_batch() if tag == "__tool_started__": + await _close_reasoning() await _emit_tool_started(payload) elif tag == "__tool_completed__": await _emit_tool_completed(payload) + elif tag == "__reasoning_delta__": + await _emit_reasoning_delta(payload) elif isinstance(it, str): + # Answer text means the current reasoning burst is over. + await _close_reasoning() # Batch text deltas — append to buffer, flush on timer _batch_buf.append(it) if _batch_timer is None: @@ -5993,6 +6156,8 @@ async def _flush_batch() -> None: # Flush any final batched text before processing result if _batch_buf: await _flush_batch() + # A reasoning-only tail (no text/tool after it) closes here. + await _close_reasoning() # Pick up agent result + usage from the completed task try: @@ -6249,6 +6414,12 @@ async def _handle_responses(self, request: "web.Request") -> "web.Response": if isinstance(item, str): input_messages.append({"role": "user", "content": item}) elif isinstance(item, dict): + # Spec-compliant clients echo ``reasoning`` output items back + # in ``input``. They carry no role/content, so treating them + # as messages injects empty user turns (or a 400 when last). + # Skip them outright (#21655). + if str(item.get("type") or "").strip().lower() == "reasoning": + continue role = item.get("role", "user") try: content = _normalize_multimodal_content(item.get("content", "")) @@ -6271,6 +6442,11 @@ async def _handle_responses(self, request: "web.Request") -> "web.Response": status=400, ) for i, entry in enumerate(raw_history): + # Echoed reasoning output items carry no role/content — skip + # them here exactly like the input array, instead of rejecting + # the whole request (#21655). + if isinstance(entry, dict) and str(entry.get("type") or "").strip().lower() == "reasoning": + continue if not isinstance(entry, dict) or "role" not in entry or "content" not in entry: return web.json_response( _openai_error(f"conversation_history[{i}] must have 'role' and 'content' fields"), @@ -6313,6 +6489,10 @@ async def _handle_responses(self, request: "web.Request") -> "web.Response": session_id = stored_session_id or str(uuid.uuid4()) stream = _coerce_request_bool(body.get("stream"), default=False) + # Resolve reasoning exposure once so the streamed events and the final + # envelope can never disagree on the gate. Sources STRUCTURED reasoning + # (reasoning_callback), never reasoning.available. + show_reasoning = self._reasoning_exposure_enabled() route = self._resolve_route(body.get("model")) agent_overrides = _request_agent_overrides( body, @@ -6348,10 +6528,25 @@ def _on_tool_progress(event_type, name, preview, args, **kwargs): The structured Responses stream uses ``tool_start_callback`` and ``tool_complete_callback`` for exact call-id correlation, - so progress events are currently ignored here. + so progress events are ignored here. In particular + ``reasoning.available`` is NOT forwarded: conversation_loop + derives it from the assistant message *content*, not the + model's reasoning. Real reasoning arrives through + ``reasoning_callback`` below (#21655, #7556). """ return + # Structured reasoning deltas → spec reasoning output items. Fed + # by the agent's reasoning_callback (_fire_reasoning_delta), which + # carries native reasoning_content / thinking deltas — NOT + # reasoning.available. None when the gate is off so the wire format + # is byte-identical for existing clients. + _on_reasoning = None + if show_reasoning: + def _on_reasoning(text): + if text: + _stream_q.put_threadsafe(("__reasoning_delta__", str(text))) + def _on_tool_start(tool_call_id, function_name, function_args): """Queue a started tool for live function_call streaming.""" _stream_q.put_threadsafe(("__tool_started__", { @@ -6379,6 +6574,7 @@ def _on_tool_complete(tool_call_id, function_name, function_args, function_resul tool_progress_callback=_on_tool_progress, tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete, + reasoning_callback=_on_reasoning, agent_ref=agent_ref, gateway_session_key=gateway_session_key, **agent_overrides, @@ -6487,7 +6683,9 @@ async def _compute_response(): user_message, result, ) - output_items = self._extract_output_items(result, start_index=output_start_index) + output_items = self._extract_output_items( + result, start_index=output_start_index, include_reasoning=show_reasoning, + ) response_data = { "id": response_id, @@ -7066,11 +7264,18 @@ def _turn_transcript_messages( return out @staticmethod - def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[Dict[str, Any]]: + def _extract_output_items( + result: Dict[str, Any], start_index: int = 0, include_reasoning: bool = False, + ) -> List[Dict[str, Any]]: """ Build the output item array from the agent's messages. Walks *result["messages"]* starting at *start_index* and emits: + - ``reasoning`` items for assistant messages carrying structured + reasoning, when *include_reasoning* is True (#21655; gated by + ``display.platforms.api_server.show_reasoning``). Sourced from the + message's own ``reasoning_content``/``reasoning`` field, never from + reasoning.available. - ``function_call`` items for each tool_call on assistant messages - ``function_call_output`` items for each tool-role message - a final ``message`` item with the assistant's text reply @@ -7082,6 +7287,15 @@ def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[ for msg in messages: role = msg.get("role") + if role == "assistant" and include_reasoning: + reasoning_text = msg.get("reasoning_content") or msg.get("reasoning") + if isinstance(reasoning_text, str) and reasoning_text.strip(): + items.append({ + "id": f"rs_{uuid.uuid4().hex[:24]}", + "type": "reasoning", + "summary": [{"type": "summary_text", "text": reasoning_text}], + "status": "completed", + }) if role == "assistant" and msg.get("tool_calls"): for tc in msg["tool_calls"]: func = tc.get("function", {}) @@ -7207,6 +7421,7 @@ async def _run_agent( tool_progress_callback=None, tool_start_callback=None, tool_complete_callback=None, + reasoning_callback=None, agent_ref: Optional[list] = None, active_run_id: Optional[str] = None, gateway_session_key: Optional[str] = None, @@ -7282,6 +7497,7 @@ def _run(): tool_progress_callback=tool_progress_callback, tool_start_callback=tool_start_callback, tool_complete_callback=tool_complete_callback, + reasoning_callback=reasoning_callback, gateway_session_key=gateway_session_key, requested_model=requested_model, requested_provider=requested_provider, diff --git a/tests/gateway/test_api_server_reasoning.py b/tests/gateway/test_api_server_reasoning.py new file mode 100644 index 000000000000..e72d537ee91a --- /dev/null +++ b/tests/gateway/test_api_server_reasoning.py @@ -0,0 +1,554 @@ +"""Tests for reasoning/thinking exposure across the API server (#48024). + +Covers the sweeper's core requirement: /v1/chat/completions and /v1/responses +must source reasoning from the STRUCTURED ``reasoning_callback`` path (the one +``_fire_reasoning_delta`` drives), NOT from the ``reasoning.available`` progress +event (which conversation_loop derives from the assistant message content). + +Both endpoints are gated on ``display.platforms.api_server.show_reasoning`` and +emit reasoning in the correct streaming shape for each surface: + +- chat/completions: ``delta.reasoning_content`` chunks (stream) + + ``message.reasoning_content`` (non-stream). +- responses: the spec reasoning event family (``response.output_item.added`` → + ``reasoning_summary_part.added`` → ``reasoning_summary_text.delta`` → ``.done`` + → ``output_item.done``) plus ``reasoning`` output items in the envelope. +""" + +import json + +import pytest +from aiohttp.test_utils import TestClient, TestServer + +from gateway.config import PlatformConfig +from gateway.platforms.api_server import APIServerAdapter +from tests.gateway.test_api_server import _create_app + + +@pytest.fixture +def adapter(): + return APIServerAdapter(PlatformConfig(enabled=True)) + + +def _enable_reasoning(adapter): + """Force the show_reasoning gate on for a test.""" + adapter._reasoning_exposure_enabled = lambda: True + + +def _disable_reasoning(adapter): + adapter._reasoning_exposure_enabled = lambda: False + + +# --------------------------------------------------------------------------- +# Gate helper +# --------------------------------------------------------------------------- + + +class TestReasoningExposureGate: + def test_gate_reads_display_setting(self, adapter, monkeypatch): + monkeypatch.setattr( + "gateway.run._load_gateway_config", lambda: {"_fake": True} + ) + monkeypatch.setattr( + "gateway.display_config.resolve_display_setting", + lambda cfg, platform, name, default: True, + ) + assert adapter._reasoning_exposure_enabled() is True + + def test_gate_defaults_off(self, adapter, monkeypatch): + monkeypatch.setattr( + "gateway.run._load_gateway_config", lambda: {} + ) + monkeypatch.setattr( + "gateway.display_config.resolve_display_setting", + lambda cfg, platform, name, default: default, + ) + assert adapter._reasoning_exposure_enabled() is False + + def test_gate_fails_closed_on_config_error(self, adapter, monkeypatch): + def _boom(): + raise RuntimeError("config parse failed") + + monkeypatch.setattr("gateway.run._load_gateway_config", _boom) + # Must not raise — a broken config can't 500 every request. + assert adapter._reasoning_exposure_enabled() is False + + +# --------------------------------------------------------------------------- +# chat/completions — structured reasoning +# --------------------------------------------------------------------------- + + +class TestChatCompletionsReasoning: + @pytest.mark.asyncio + async def test_stream_emits_reasoning_content_from_callback(self, adapter): + """Structured reasoning_callback deltas → delta.reasoning_content.""" + _enable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + async def _mock_run_agent(**kwargs): + rc = kwargs.get("reasoning_callback") + cb = kwargs.get("stream_delta_callback") + # The gateway must wire a real reasoning_callback when the + # gate is on — this is the structured path, not + # reasoning.available. + assert rc is not None + rc("Let me think ") + rc("about this.") + if cb: + cb("Answer.") + return ( + {"final_response": "Answer.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/chat/completions", + json={ + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + "stream": True, + }, + ) + assert resp.status == 200 + body = await resp.text() + + reasoning_deltas = [] + content_deltas = [] + for line in body.splitlines(): + if not line.startswith("data: "): + continue + raw = line[len("data: "):] + if raw.strip() == "[DONE]": + continue + try: + chunk = json.loads(raw) + except json.JSONDecodeError: + continue + for choice in chunk.get("choices", []): + delta = choice.get("delta", {}) + if "reasoning_content" in delta: + reasoning_deltas.append(delta["reasoning_content"]) + if delta.get("content"): + content_deltas.append(delta["content"]) + + assert "".join(reasoning_deltas) == "Let me think about this." + assert "".join(content_deltas) == "Answer." + + @pytest.mark.asyncio + async def test_stream_no_reasoning_when_gate_off(self, adapter): + """Gate off → no reasoning_callback wired, byte-identical wire.""" + _disable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + captured = {} + + async def _mock_run_agent(**kwargs): + captured["reasoning_callback"] = kwargs.get("reasoning_callback") + cb = kwargs.get("stream_delta_callback") + if cb: + cb("Answer.") + return ( + {"final_response": "Answer.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/chat/completions", + json={ + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + "stream": True, + }, + ) + assert resp.status == 200 + body = await resp.text() + + assert captured["reasoning_callback"] is None + assert "reasoning_content" not in body + + @pytest.mark.asyncio + async def test_non_stream_reasoning_content_sibling(self, adapter): + """Non-stream: reasoning_content is a sibling of content.""" + _enable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + async def _mock_run_agent(**kwargs): + rc = kwargs.get("reasoning_callback") + assert rc is not None + rc("Structured thinking.") + return ( + {"final_response": "Final.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/chat/completions", + json={ + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + }, + ) + assert resp.status == 200 + data = await resp.json() + + msg = data["choices"][0]["message"] + assert msg["content"] == "Final." + assert msg["reasoning_content"] == "Structured thinking." + + @pytest.mark.asyncio + async def test_non_stream_no_reasoning_key_when_gate_off(self, adapter): + _disable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + captured = {} + + async def _mock_run_agent(**kwargs): + captured["reasoning_callback"] = kwargs.get("reasoning_callback") + return ( + {"final_response": "Final.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/chat/completions", + json={ + "model": "test", + "messages": [{"role": "user", "content": "hi"}], + }, + ) + assert resp.status == 200 + data = await resp.json() + + assert captured["reasoning_callback"] is None + assert "reasoning_content" not in data["choices"][0]["message"] + + +# --------------------------------------------------------------------------- +# responses — spec reasoning event family +# --------------------------------------------------------------------------- + + +class TestResponsesReasoning: + @pytest.mark.asyncio + async def test_stream_emits_reasoning_event_family(self, adapter): + """Structured reasoning_callback → the spec reasoning event family.""" + _enable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + async def _mock_run_agent(**kwargs): + rc = kwargs.get("reasoning_callback") + cb = kwargs.get("stream_delta_callback") + assert rc is not None + rc("Thinking hard ") + rc("about it.") + if cb: + cb("The answer.") + return ( + {"final_response": "The answer.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Question?", + "stream": True, + }, + ) + assert resp.status == 200 + body = await resp.text() + + events = [ + line[len("event: "):] + for line in body.splitlines() + if line.startswith("event: ") + ] + # The reasoning family must appear, and in order relative to the item. + assert "response.reasoning_summary_part.added" in events + assert "response.reasoning_summary_text.delta" in events + assert "response.reasoning_summary_text.done" in events + + # Collect the reasoning text from the summary deltas. + deltas = [] + for line in body.splitlines(): + if not line.startswith("data: "): + continue + try: + payload = json.loads(line[len("data: "):]) + except json.JSONDecodeError: + continue + if payload.get("type") == "response.reasoning_summary_text.delta": + deltas.append(payload["delta"]) + assert "".join(deltas) == "Thinking hard about it." + + # A reasoning output_item.added must precede the summary part, and its + # item type must be "reasoning". + reasoning_item_added = False + for line in body.splitlines(): + if not line.startswith("data: "): + continue + try: + payload = json.loads(line[len("data: "):]) + except json.JSONDecodeError: + continue + if ( + payload.get("type") == "response.output_item.added" + and payload.get("item", {}).get("type") == "reasoning" + ): + reasoning_item_added = True + assert reasoning_item_added + + @pytest.mark.asyncio + async def test_stream_reasoning_closes_before_text(self, adapter): + """A reasoning burst closes (done) before answer text is emitted.""" + _enable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + async def _mock_run_agent(**kwargs): + rc = kwargs.get("reasoning_callback") + cb = kwargs.get("stream_delta_callback") + rc("Reason.") + if cb: + cb("Text.") + return ( + {"final_response": "Text.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Q?", + "stream": True, + }, + ) + assert resp.status == 200 + body = await resp.text() + + ordered_types = [] + for line in body.splitlines(): + if not line.startswith("data: "): + continue + try: + payload = json.loads(line[len("data: "):]) + except json.JSONDecodeError: + continue + t = payload.get("type") + if t in ( + "response.reasoning_summary_text.done", + "response.output_text.delta", + ): + ordered_types.append(t) + # The reasoning summary must be finalized before the first text delta. + assert ordered_types[0] == "response.reasoning_summary_text.done" + assert "response.output_text.delta" in ordered_types + + @pytest.mark.asyncio + async def test_stream_no_reasoning_when_gate_off(self, adapter): + _disable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + captured = {} + + async def _mock_run_agent(**kwargs): + captured["reasoning_callback"] = kwargs.get("reasoning_callback") + cb = kwargs.get("stream_delta_callback") + if cb: + cb("Answer.") + return ( + {"final_response": "Answer.", "messages": [], "api_calls": 1}, + {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + ) + + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "Q?", + "stream": True, + }, + ) + assert resp.status == 200 + body = await resp.text() + + assert captured["reasoning_callback"] is None + assert "reasoning_summary" not in body + + +# --------------------------------------------------------------------------- +# responses non-stream — _extract_output_items include_reasoning +# --------------------------------------------------------------------------- + + +class TestExtractOutputItemsReasoning: + def test_reasoning_item_from_message_field(self): + result = { + "messages": [ + { + "role": "assistant", + "content": "hi", + "reasoning_content": "my structured reasoning", + }, + ], + "final_response": "hi", + } + items = APIServerAdapter._extract_output_items(result, include_reasoning=True) + reasoning = [it for it in items if it.get("type") == "reasoning"] + assert len(reasoning) == 1 + assert reasoning[0]["summary"][0]["text"] == "my structured reasoning" + assert reasoning[0]["status"] == "completed" + + def test_reasoning_falls_back_to_reasoning_key(self): + result = { + "messages": [ + {"role": "assistant", "content": "hi", "reasoning": "alt field"}, + ], + "final_response": "hi", + } + items = APIServerAdapter._extract_output_items(result, include_reasoning=True) + reasoning = [it for it in items if it.get("type") == "reasoning"] + assert len(reasoning) == 1 + assert reasoning[0]["summary"][0]["text"] == "alt field" + + def test_no_reasoning_item_when_flag_off(self): + result = { + "messages": [ + {"role": "assistant", "content": "hi", "reasoning_content": "x"}, + ], + "final_response": "hi", + } + items = APIServerAdapter._extract_output_items(result, include_reasoning=False) + assert not any(it.get("type") == "reasoning" for it in items) + + def test_no_reasoning_item_when_field_blank(self): + result = { + "messages": [ + {"role": "assistant", "content": "hi", "reasoning_content": " "}, + ], + "final_response": "hi", + } + items = APIServerAdapter._extract_output_items(result, include_reasoning=True) + assert not any(it.get("type") == "reasoning" for it in items) + + @pytest.mark.asyncio + async def test_non_stream_response_includes_reasoning_output_item(self, adapter): + _enable_reasoning(adapter) + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + import unittest.mock as _m + with _m.patch.object(adapter, "_run_agent", new_callable=_m.AsyncMock) as mock_run: + mock_run.return_value = ( + { + "final_response": "answer", + "messages": [ + { + "role": "assistant", + "content": "answer", + "reasoning_content": "the thinking", + }, + ], + "api_calls": 1, + }, + {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}, + ) + resp = await cli.post( + "/v1/responses", + json={"model": "hermes-agent", "input": "Q?"}, + ) + assert resp.status == 200 + data = await resp.json() + + types = [it.get("type") for it in data["output"]] + assert "reasoning" in types + + +# --------------------------------------------------------------------------- +# responses input hardening — echoed reasoning items are skipped +# --------------------------------------------------------------------------- + + +class TestResponsesInputHardening: + @pytest.mark.asyncio + async def test_echoed_reasoning_item_in_input_is_skipped(self, adapter): + """A trailing reasoning item in ``input`` must not 400 the request.""" + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + captured = {} + + import unittest.mock as _m + + async def _mock_run_agent(**kwargs): + captured["user_message"] = kwargs.get("user_message") + captured["history"] = kwargs.get("conversation_history") + return ( + {"final_response": "ok", "messages": [], "api_calls": 1}, + {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}, + ) + + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": [ + {"role": "user", "content": "hi"}, + { + "type": "reasoning", + "id": "rs_abc", + "summary": [{"type": "summary_text", "text": "prev"}], + }, + ], + }, + ) + assert resp.status == 200 + + # The reasoning item was dropped, so the real user turn is the input. + assert captured["user_message"] == "hi" + + @pytest.mark.asyncio + async def test_echoed_reasoning_item_in_history_is_skipped(self, adapter): + app = _create_app(adapter) + async with TestClient(TestServer(app)) as cli: + import unittest.mock as _m + + async def _mock_run_agent(**kwargs): + return ( + {"final_response": "ok", "messages": [], "api_calls": 1}, + {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}, + ) + + with _m.patch.object(adapter, "_run_agent", side_effect=_mock_run_agent): + resp = await cli.post( + "/v1/responses", + json={ + "model": "hermes-agent", + "input": "next", + "conversation_history": [ + {"role": "user", "content": "earlier"}, + { + "type": "reasoning", + "id": "rs_xyz", + "summary": [{"type": "summary_text", "text": "t"}], + }, + ], + }, + ) + # Without the skip, the reasoning entry (no role/content) 400s. + assert resp.status == 200 diff --git a/tests/test_anthropic_thinking_display.py b/tests/test_anthropic_thinking_display.py new file mode 100644 index 000000000000..8f78e69e8ab2 --- /dev/null +++ b/tests/test_anthropic_thinking_display.py @@ -0,0 +1,82 @@ +"""Unit tests for the Anthropic thinking.display resolver (#48024). + +reasoning_config["display"] controls whether Claude returns the reasoning text +("summarized") or suppresses it on the wire ("omitted"). Default is summarized so +Hermes always has reasoning to surface in its CLI + API-exposure paths. + +Run: python -m pytest tests/test_anthropic_thinking_display.py -q +""" +from __future__ import annotations + +import pytest + +from agent.anthropic_adapter import _resolve_thinking_display, build_anthropic_kwargs + + +# ── _resolve_thinking_display ────────────────────────────────────────── + +def test_default_is_summarized_when_no_config(): + assert _resolve_thinking_display(None) == "summarized" + assert _resolve_thinking_display({}) == "summarized" + assert _resolve_thinking_display({"effort": "high"}) == "summarized" + + +def test_explicit_omitted(): + assert _resolve_thinking_display({"display": "omitted"}) == "omitted" + + +def test_explicit_summarized(): + assert _resolve_thinking_display({"display": "summarized"}) == "summarized" + + +def test_case_and_whitespace_normalized(): + assert _resolve_thinking_display({"display": " OMITTED "}) == "omitted" + assert _resolve_thinking_display({"display": "Summarized"}) == "summarized" + + +def test_invalid_value_falls_back_to_summarized(): + # never produce an invalid wire value + assert _resolve_thinking_display({"display": "verbose"}) == "summarized" + assert _resolve_thinking_display({"display": 123}) == "summarized" + assert _resolve_thinking_display({"display": None}) == "summarized" + + +# ── end-to-end through build_anthropic_kwargs (adaptive model) ───────── + +def _thinking_for(reasoning_config): + kw = build_anthropic_kwargs( + model="claude-opus-4-7", # adaptive-thinking model + messages=[{"role": "user", "content": "hi"}], + tools=None, + max_tokens=4096, + reasoning_config=reasoning_config, + ) + return kw.get("thinking") + + +def test_kwargs_default_display_summarized(): + th = _thinking_for({"effort": "high"}) + assert th and th["type"] == "adaptive" + assert th["display"] == "summarized" + + +def test_kwargs_display_omitted_passes_through(): + th = _thinking_for({"effort": "high", "display": "omitted"}) + assert th and th["type"] == "adaptive" + assert th["display"] == "omitted" + + +def test_kwargs_display_invalid_falls_back(): + th = _thinking_for({"effort": "high", "display": "loud"}) + assert th and th["type"] == "adaptive" + assert th["display"] == "summarized" + + +def test_kwargs_display_unchanged_when_reasoning_disabled(): + # enabled:False → no adaptive thinking kwarg (display moot) + th = _thinking_for({"enabled": False, "effort": "high"}) + assert th is None or th.get("type") != "adaptive" + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-q"]))