From 37fb7adfd6d4b3776578634c26c8f0ef9ec77b20 Mon Sep 17 00:00:00 2001
From: Teknium <127238744+teknium1@users.noreply.github.com>
Date: Mon, 7 Sep 2026 03:14:19 -0700
Subject: [PATCH] fix: structured reasoning no longer breaks chat consumers

Normalize incoming reasoning at the shared heading boundary and completed
extraction, and flatten auxiliary content and reasoning before accumulation.
Reuse the existing text flattener with no implicit fragment separators.

Combine the earliest related work from zsuroy (#85791), the diagnosis and
patch from 2025hcsmile2010-hue (#104711, #104848), and completed extraction
work from liuhao1024 (#104717) as a slim redo, not a verbatim cherry-pick.

Two invariant tests exercise the real SDK and local HTTP fixture across
main streaming, Relay collection, auxiliary sync/async and completed output.
The standalone matrix improves from 32/84 to 84/84, preserving answers.

Co-authored-by: suroy <suroy@qq.com>
Co-authored-by: 2025hcsmile2010-hue <2025hcsmile2010@gmail.com>
Co-authored-by: liuhao1024 <sunsky.lau@gmail.com>
---
 agent/agent_runtime_helpers.py                |   3 +
 agent/auxiliary_client.py                     |   7 +-
 agent/reasoning_summaries.py                  |   8 +-
 evals/providers/reasoning_shapes.py           | 169 ++++++++++++++++++
 .../agent/test_reasoning_shape_boundaries.py  |  14 ++
 .../docs/developer-guide/provider-runtime.md  |  10 ++
 6 files changed, 208 insertions(+), 3 deletions(-)
 create mode 100644 evals/providers/reasoning_shapes.py
 create mode 100644 tests/agent/test_reasoning_shape_boundaries.py

diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py
index 52c7762d9b..f097dee742 100644
--- a/agent/agent_runtime_helpers.py
+++ b/agent/agent_runtime_helpers.py
@@ -1211,6 +1211,9 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
     parts: List[str] = []
 
     def _add(text) -> None:
+        from agent.message_content import flatten_message_text
+
+        text = flatten_message_text(text, sep="")
         if text and text not in parts:
             parts.append(text)
     _add(getattr(assistant_message, "reasoning", None))
diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py
index fd6ecd4672..7f587a9752 100644
--- a/agent/auxiliary_client.py
+++ b/agent/auxiliary_client.py
@@ -6422,12 +6422,15 @@ class _ChatStreamAccumulator:
         if delta is None:
             return
         made_progress = False
-        piece = getattr(delta, "content", None)
+        from agent.message_content import flatten_message_text
+
+        piece = flatten_message_text(getattr(delta, "content", None), sep="")
         if piece:
             self.content_parts.append(piece)
             made_progress = True
         reasoning_piece = getattr(delta, "reasoning", None) or getattr(delta, "reasoning_content", None)
-        if reasoning_piece and isinstance(reasoning_piece, str):
+        reasoning_piece = flatten_message_text(reasoning_piece, sep="")
+        if reasoning_piece:
             self.reasoning_parts.append(reasoning_piece)
             made_progress = True
         # Evaluate both unconditionally: they accumulate state, not just progress.
diff --git a/agent/reasoning_summaries.py b/agent/reasoning_summaries.py
index f58d150d7d..ef6c7f7d89 100644
--- a/agent/reasoning_summaries.py
+++ b/agent/reasoning_summaries.py
@@ -10,10 +10,14 @@ the blank-line join Hermes' own Responses adapter does.
 
 from __future__ import annotations
 
+from typing import Any
+
+from agent.message_content import flatten_message_text
+
 __all__ = ["separate_glued_reasoning_blocks"]
 
 
-def separate_glued_reasoning_blocks(previous: str, delta: str) -> str:
+def separate_glued_reasoning_blocks(previous: str, delta: Any) -> str:
     """Return *delta*, prefixed with a paragraph break when it glues onto *previous*.
 
     A break is inserted when *delta* opens a *closed* bold heading and *previous* is mid-line
@@ -21,5 +25,7 @@ def separate_glued_reasoning_blocks(previous: str, delta: str) -> str:
     alone: its deltas carry their own whitespace, and a fragment that merely opens emphasis
     (``**`` alone) is not a part boundary — summary parts carry the whole heading in one delta.
     """
+    # Relays also emit content-part lists/dicts; fragments carry their own whitespace.
+    delta = flatten_message_text(delta, sep="")
     glued = previous and delta and not previous[-1].isspace() and delta.startswith("**") and "**" in delta[2:]
     return f"\n\n{delta}" if glued else delta
diff --git a/evals/providers/reasoning_shapes.py b/evals/providers/reasoning_shapes.py
new file mode 100644
index 0000000000..7d9851e4b8
--- /dev/null
+++ b/evals/providers/reasoning_shapes.py
@@ -0,0 +1,169 @@
+"""Local HTTP/SDK reasoning-shape probe; no vendor inference or credentials.
+
+Run: python evals/providers/reasoning_shapes.py --output /tmp/reasoning.json
+Run the same file in a fresh interpreter on base and fix checkouts for A/B.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+import tempfile
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+
+CASES = {
+    "string": ("**First**", "**Second**", "**First****Second**"),
+    "text-parts": ([{"type": "text", "text": "**First**"}],
+                   [{"type": "text", "text": "**Second**"}], "**First****Second**"),
+    "split-fragments": (["Hel", "lo"], [" wor", "ld"], "Hello world"),
+    "dict": ({"type": "text", "text": "Hello"}, {"text": " world"}, "Hello world"),
+    "empty": ([], None, ""),
+    "nontext-parts": ([{"type": "image_url", "image_url": {"url": "ignored"}}], [], ""),
+}
+
+
+def run_matrix(surfaces):
+    from openai import AsyncOpenAI, OpenAI
+    from run_agent import AIAgent
+    from agent.agent_runtime_helpers import extract_reasoning
+    from agent.auxiliary_client import _aggregate_chat_stream, _aggregate_chat_stream_async
+    from agent.chat_completion_helpers import interruptible_streaming_api_call
+    from agent.chat_completion_helpers_relay import RelayChatAccumulator
+
+    active = {"message": {}, "deltas": [], "answer": "LOCAL_CAPTURE_OK"}
+    requests = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, format, *args):
+            pass
+
+        def do_POST(self):
+            request = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            requests.append({"path": self.path, "body": request})
+            self.send_response(200)
+            stream = request.get("stream", False)
+            self.send_header("Content-Type", "text/event-stream" if stream else "application/json")
+            self.end_headers()
+            common = {"id": "local", "created": 1, "model": "probe"}
+            if stream:
+                for delta in active["deltas"] + [{"content": active["answer"]}]:
+                    chunk = {**common, "object": "chat.completion.chunk", "choices": [
+                        {"index": 0, "delta": delta, "finish_reason": None}]}
+                    self.wfile.write(("data: " + json.dumps(chunk) + "\n\n").encode())
+                chunk = {**common, "object": "chat.completion.chunk", "choices": [
+                    {"index": 0, "delta": {}, "finish_reason": "stop"}]}
+                self.wfile.write(("data: " + json.dumps(chunk) + "\n\ndata: [DONE]\n\n").encode())
+            else:
+                message = {"role": "assistant", "content": "LOCAL_CAPTURE_OK", **active["message"]}
+                response = {**common, "object": "chat.completion", "choices": [
+                    {"index": 0, "message": message, "finish_reason": "stop"}]}
+                self.wfile.write(json.dumps(response).encode())
+            self.wfile.flush()
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    url = f"http://127.0.0.1:{server.server_port}/v1"
+    rows = []
+    client = OpenAI(base_url=url, api_key="local-fixture", max_retries=0)
+    agent = AIAgent(base_url=url, api_key="local-fixture", provider="custom", model="probe",
+                    api_mode="chat_completions", enabled_toolsets=[], quiet_mode=True,
+                    skip_memory=True, skip_context_files=True, skip_background_review=True)
+    kwargs = {"model": "probe", "messages": [{"role": "user", "content": "probe"}]}
+
+    async def aux_async():
+        async with AsyncOpenAI(base_url=url, api_key="local-fixture", max_retries=0) as async_client:
+            stream = await async_client.chat.completions.create(**kwargs, stream=True)
+            return await _aggregate_chat_stream_async(stream, model="probe")
+
+    try:
+        for surface in surfaces:
+            fields = ("reasoning", "reasoning_content")
+            if surface == "nonstream":
+                fields += tuple(f"reasoning_details.{key}" for key in ("summary", "thinking", "content", "text"))
+            for field in fields:
+                for name, (first, second, plain) in CASES.items():
+                    active["deltas"] = [{field: first}, {field: second}]
+                    active["answer"] = ([{"type": "text", "text": "LOCAL_"}, "CAPTURE_OK"]
+                                        if name == "split-fragments" else "LOCAL_CAPTURE_OK")
+                    # Completed responses exercise one field containing the same fragments.
+                    active["message"] = ({"reasoning_details": [{field.split(".")[1]: first}]}
+                                         if "." in field else {field: first})
+                    callbacks = []
+                    agent.reasoning_callback = callbacks.append
+                    expected = plain
+                    if surface in ("main", "relay") and name in ("string", "text-parts"):
+                        expected = "**First**\n\n**Second**"
+                    if surface == "nonstream":
+                        expected = {"string": "**First**", "text-parts": "**First**",
+                                    "split-fragments": "Hello", "dict": "Hello",
+                                    "empty": "", "nontext-parts": ""}[name]
+                    row = {"surface": surface, "field": field, "case": name, "expected": expected}
+                    try:
+                        if surface == "main":
+                            response = interruptible_streaming_api_call(agent, dict(kwargs))
+                            message = response.choices[0].message
+                            reasoning = extract_reasoning(agent, message) or ""
+                            row["callbacks"] = callbacks
+                        elif surface == "relay":
+                            acc = RelayChatAccumulator()
+                            with client.chat.completions.create(**kwargs, stream=True) as stream:
+                                for chunk in stream:
+                                    acc.observe(chunk.model_dump(warnings=False))
+                            message = acc.finalize()["choices"][0]["message"]
+                            reasoning = message["reasoning_content"] or ""
+                        elif surface == "nonstream":
+                            message = client.chat.completions.create(**kwargs).choices[0].message
+                            reasoning = extract_reasoning(agent, message) or ""
+                        else:
+                            if surface == "aux-sync":
+                                stream = client.chat.completions.create(**kwargs, stream=True)
+                                response = _aggregate_chat_stream(stream, model="probe")
+                            else:
+                                response = asyncio.run(aux_async())
+                            message = response.choices[0].message
+                            reasoning = message.reasoning or ""
+                        content = message["content"] if isinstance(message, dict) else message.content
+                        row.update(reasoning=reasoning, content=content)
+                        row["ok"] = reasoning == expected and content == "LOCAL_CAPTURE_OK"
+                        if surface == "main":
+                            row["ok"] &= all(isinstance(value, str) for value in callbacks)
+                            if row["ok"]:
+                                row["ok"] &= "".join(callbacks) == expected
+                    except Exception as error:
+                        row.update(ok=False, error=f"{type(error).__name__}: {error}")
+                    rows.append(row)
+    finally:
+        client.close()
+        agent.client.close()
+        server.shutdown()
+        server.server_close()
+        thread.join()
+    return {"fidelity": "Local HTTP fixture + real SDK + production consumers; not vendor inference",
+            "module": str(Path(__file__).resolve()), "requests": requests, "results": rows}
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", required=True)
+    args = parser.parse_args()
+    for key in list(os.environ):
+        if any(token in key.upper() for token in ("TOKEN", "SECRET", "API_KEY", "AUTH")):
+            os.environ.pop(key, None)
+    with tempfile.TemporaryDirectory(prefix="reasoning-wire-") as home:
+        os.environ.update(HOME=home, HERMES_HOME=home)
+        sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+        result = run_matrix(("main", "relay", "aux-sync", "aux-async", "nonstream"))
+    Path(args.output).write_text(json.dumps(result, indent=2), encoding="utf-8")
+    print(json.dumps({"cases": len(result["results"]),
+                      "passed": sum(row["ok"] for row in result["results"]),
+                      "output": args.output}))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/agent/test_reasoning_shape_boundaries.py b/tests/agent/test_reasoning_shape_boundaries.py
new file mode 100644
index 0000000000..cd842b3321
--- /dev/null
+++ b/tests/agent/test_reasoning_shape_boundaries.py
@@ -0,0 +1,14 @@
+"""Structured reasoning keeps text and answer channels intact on every chat consumer."""
+from evals.providers.reasoning_shapes import run_matrix
+
+
+def test_stream_reasoning_shapes_preserve_text_and_answer():
+    result = run_matrix(("main", "relay", "aux-sync", "aux-async"))
+    failures = [row for row in result["results"] if not row["ok"]]
+    assert not failures, failures
+
+
+def test_completed_reasoning_shapes_preserve_text_and_answer():
+    result = run_matrix(("nonstream",))
+    failures = [row for row in result["results"] if not row["ok"]]
+    assert not failures, failures
diff --git a/website/docs/developer-guide/provider-runtime.md b/website/docs/developer-guide/provider-runtime.md
index 3e2c723a6b..5cb620a6d1 100644
--- a/website/docs/developer-guide/provider-runtime.md
+++ b/website/docs/developer-guide/provider-runtime.md
@@ -27,6 +27,16 @@ Primary implementation:
 
 If you are trying to add a new first-class inference provider, read [Adding Providers](./adding-providers.md) and the [Model Provider Plugin guide](./model-provider-plugin.md) alongside this page.
 
+## Chat-completions reasoning shapes
+
+OpenAI-compatible relays can return `reasoning` or `reasoning_content` as strings,
+text-part dictionaries, or lists of text parts and string fragments. Hermes flattens
+these fields before string operations in the main stream, Relay recording, synchronous
+and asynchronous auxiliary streams, and completed-response reasoning extraction.
+Fragments retain their explicit whitespace; normalization adds no intra-field separator.
+Main-stream and Relay recording retain the existing paragraph breaks between complete
+bold reasoning headings. Reasoning stays separate from the visible answer.
+
 ## Resolution precedence
 
 At a high level, provider resolution uses: