feat: stream and persist reasoning blocks

backend/app/adapters/base.py

@@ -99,12 +99,25 @@ async def emit_token_delta(
         step_index: int,
         delta: str,
         role: str = "assistant",
+        part: str = "text",
         **data: Any,
     ) -> None:
-        """Stream an incremental token chunk to SSE subscribers (no DB write)."""
+        """Stream an incremental token chunk to SSE subscribers (no DB write).
+
+        ``part`` discriminates fine-grained streams: ``text`` (default) for the
+        visible reply, ``reasoning`` for model thinking / chain-of-thought.
+        Reasoning deltas are still ephemeral; persist the finished block with
+        ``emit_message(..., kind="reasoning")``.
+        """
         await self.emit(
             "token.delta",
-            {"step_index": step_index, "delta": delta, "role": role, **data},
+            {
+                "step_index": step_index,
+                "delta": delta,
+                "role": role,
+                "part": part,
+                **data,
+            },
         )
 
     async def emit_message(

backend/app/adapters/langgraph_adapter.py

@@ -18,6 +18,7 @@
   },
   "system_prompt": "You are a helpful coordinator.",
   "stream_tokens": true,       // emit token.delta SSE; defer tokens to step.updated
+  "stream_reasoning": false,   // mock / providers: also stream part=reasoning blocks
   "tools": ["echo"],           // builtin or MCP keys (mcp/{server}/{tool})
   "mcp_servers": [             // optional MCP stdio/SSE/HTTP servers
     {
@@ -156,6 +157,7 @@ def __init__(self, output: dict[str, Any]) -> None:
 @dataclass
 class ModelResponse:
     content: str
+    reasoning: str = ""
     tool_calls: list[dict[str, Any]] = field(default_factory=list)
     tokens_in: int = 0
     tokens_out: int = 0
@@ -593,6 +595,7 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
             total_out = 0
             total_cost = 0.0
             final_reply = ""
+            final_reasoning = ""
             had_recoverable_feedback = False
             policy = run_state.tool_error_policy
             last_response: ModelResponse | None = None
@@ -698,6 +701,7 @@ async def _run_one_tool(tc: dict[str, Any]) -> ToolOutcome:
                         continue
 
                     final_reply = response.content or ""
+                    final_reasoning = response.reasoning or ""
                     messages.append({"role": "assistant", "content": final_reply})
                     break
                 else:
@@ -710,12 +714,13 @@ async def _run_one_tool(tc: dict[str, Any]) -> ToolOutcome:
                     messages.append({"role": "assistant", "content": final_reply})
 
                 latency_ms = int((time.monotonic() - started) * 1000)
-                used_model = (
-                    (last_response.model if last_response else None) or model
+                cost_usd = estimate_cost_usd(model, total_in, total_out)
+                await _emit_assistant_messages(
+                    ctx,
+                    step_index=step_idx,
+                    content=final_reply,
+                    reasoning=final_reasoning,
                 )
-                routing_meta = last_response.routing if last_response else None
-                cost_usd = round(total_cost, 6)
-                await ctx.emit_message(role="assistant", content=final_reply, step_index=step_idx)
                 await ctx.emit_step_updated(
                     index=step_idx,
                     tokens_in=total_in,
@@ -827,7 +832,12 @@ async def handler(state: dict[str, Any]) -> dict[str, Any]:
                 model=used_model,
                 model_routing=response.routing,
             )
-            await ctx.emit_message(role="assistant", content=reply, step_index=step_idx)
+            await _emit_assistant_messages(
+                ctx,
+                step_index=step_idx,
+                content=reply,
+                reasoning=response.reasoning,
+            )
             await ctx.emit_step_completed(
                 index=step_idx,
                 node=spec.id,
@@ -997,17 +1007,23 @@ async def _invoke_model_once(
             message = data["choices"][0]["message"]
             tool_calls = _parse_openai_tool_calls(message.get("tool_calls"))
             content = message.get("content") or ""
+            reasoning = _extract_reasoning_text(message)
             usage = data.get("usage") or {}
             prompt_text = json.dumps(messages, default=str)
             tokens_in = int(
                 usage.get("prompt_tokens") or estimate_tokens(prompt_text)
             )
             tokens_out = int(
                 usage.get("completion_tokens")
-                or estimate_tokens(content or json.dumps(tool_calls, default=str))
+                or estimate_tokens(
+                    (content or "")
+                    + reasoning
+                    + (json.dumps(tool_calls, default=str) if tool_calls else "")
+                )
             )
             return ModelResponse(
                 content=content,
+                reasoning=reasoning,
                 tool_calls=tool_calls,
                 tokens_in=tokens_in,
                 tokens_out=tokens_out,
@@ -1064,12 +1080,24 @@ async def _invoke_mock(
         else:
             reply = f"[mock:{model}]{suffix} {user_input}"
         tokens_in = estimate_tokens(json.dumps(messages, default=str))
-        tokens_out = estimate_tokens(reply)
+        reasoning = ""
+        if bool(ctx.agent_config.get("stream_reasoning")):
+            preview = user_input[:48].replace("\n", " ")
+            reasoning = f"[think:{model}] weigh options for: {preview}"
+        tokens_out = estimate_tokens(reply + reasoning)
         if stream_tokens:
+            if reasoning:
+                for chunk in _chunk_text(reasoning):
+                    await ctx.emit_token_delta(
+                        step_index=step_index, delta=chunk, part="reasoning"
+                    )
             for chunk in _chunk_text(reply):
-                await ctx.emit_token_delta(step_index=step_index, delta=chunk)
+                await ctx.emit_token_delta(
+                    step_index=step_index, delta=chunk, part="text"
+                )
         return ModelResponse(
             content=reply,
+            reasoning=reasoning,
             tokens_in=tokens_in,
             tokens_out=tokens_out,
             model=model,
@@ -1090,6 +1118,7 @@ async def _invoke_openai_streaming(
             "stream_options": {"include_usage": True},
         }
         parts: list[str] = []
+        reasoning_parts: list[str] = []
         tokens_in = 0
         tokens_out = 0
 
@@ -1119,20 +1148,34 @@ async def _invoke_openai_streaming(
                         )
                     for choice in chunk.get("choices") or []:
                         delta = choice.get("delta") or {}
+                        reasoning_piece = _extract_reasoning_text(delta)
+                        if reasoning_piece:
+                            reasoning_parts.append(reasoning_piece)
+                            await ctx.emit_token_delta(
+                                step_index=step_index,
+                                delta=reasoning_piece,
+                                part="reasoning",
+                            )
                         content = delta.get("content")
                         if content:
                             parts.append(content)
                             await ctx.emit_token_delta(
-                                step_index=step_index, delta=content
+                                step_index=step_index,
+                                delta=content,
+                                part="text",
                             )
 
         reply = "".join(parts)
+        reasoning = "".join(reasoning_parts)
         if not tokens_in:
             tokens_in = estimate_tokens(str(payload.get("messages", "")))
         if not tokens_out:
-            tokens_out = estimate_tokens(reply)
+            tokens_out = estimate_tokens(reply + reasoning)
         return ModelResponse(
-            content=reply, tokens_in=tokens_in, tokens_out=tokens_out
+            content=reply,
+            reasoning=reasoning,
+            tokens_in=tokens_in,
+            tokens_out=tokens_out,
         )
 
 
@@ -1325,3 +1368,36 @@ def _initial_graph_state(ctx: AdapterContext) -> dict[str, Any]:
 def _chunk_text(text: str, *, size: int = 8) -> list[str]:
     """Split text into small chunks for mock streaming."""
     return [text[i : i + size] for i in range(0, len(text), size)] or [text]
+
+
+def _extract_reasoning_text(payload: dict[str, Any] | None) -> str:
+    """Pull reasoning / thinking text from OpenAI-compatible message or delta.
+
+    Providers disagree on the field name (``reasoning_content``, ``reasoning``,
+    Anthropic-style ``thinking``). Empty / missing values are ignored.
+    """
+    if not payload:
+        return ""
+    for key in ("reasoning_content", "reasoning", "thinking"):
+        value = payload.get(key)
+        if isinstance(value, str) and value:
+            return value
+    return ""
+
+
+async def _emit_assistant_messages(
+    ctx: AdapterContext,
+    *,
+    step_index: int,
+    content: str,
+    reasoning: str = "",
+) -> None:
+    """Persist optional reasoning block, then the visible assistant reply."""
+    if reasoning.strip():
+        await ctx.emit_message(
+            role="assistant",
+            content=reasoning,
+            step_index=step_index,
+            kind="reasoning",
+        )
+    await ctx.emit_message(role="assistant", content=content, step_index=step_index)

backend/app/services/thread_service.py

@@ -217,9 +217,11 @@ async def load_thread_window(
     ) -> list[dict[str, Any]]:
         """Return OpenAI-style chat dicts for prior thread turns, window-trimmed.
 
-        Seeds only complete user/assistant turns (no system / prompt_echo / tool).
-        Cross-run tool chains are incomplete and break model APIs, so they are
-        omitted from L1. Used by the worker when constructing ``AdapterContext``.
+        Seeds only complete user/assistant turns (no system / prompt_echo /
+        reasoning / tool). Cross-run tool chains are incomplete and break model
+        APIs, so they are omitted from L1. Reasoning blocks stay in the Run
+        transcript for the console but are not replayed into the next prompt.
+        Used by the worker when constructing ``AdapterContext``.
         """
         max_messages = get_settings().thread_messages_max
         # Over-fetch slightly so role filtering still fills the cap.
@@ -246,7 +248,7 @@ async def load_thread_window(
         messages: list[dict[str, Any]] = []
         for msg, _run in result.all():
             extra = msg.extra or {}
-            if extra.get("kind") == "prompt_echo":
+            if extra.get("kind") in ("prompt_echo", "reasoning"):
                 continue
             payload = message_row_to_dict(msg)
             # Drop incomplete tool-call metadata from prior runs.

backend/tests/test_langgraph_adapter.py

@@ -294,6 +294,7 @@ async def test_langgraph_streams_token_deltas_and_defers_step_tokens():
     ]
     assert len(token_events) > 0
     assert all(e["step_index"] == 0 for e in token_events)
+    assert all(e.get("part", "text") == "text" for e in token_events)
     reply = (result.output or {}).get("reply", "")
     assert "".join(e["delta"] for e in token_events) == reply
 
@@ -315,6 +316,49 @@ async def test_langgraph_streams_token_deltas_and_defers_step_tokens():
     assert completed[0]["tokens_out"] == updated[0]["tokens_out"]
 
 
+@pytest.mark.asyncio
+async def test_langgraph_streams_and_persists_reasoning_blocks():
+    adapter = LangGraphAdapter()
+    ctx = _RecordingContext()
+    ctx.agent_config = {
+        "model": "openai/gpt-4o-mini",
+        "stream_tokens": True,
+        "stream_reasoning": True,
+    }
+    result = await adapter.run(ctx)
+    assert result.status == RunStatus.SUCCEEDED
+
+    reasoning_deltas = [
+        data
+        for event, data in ctx.events
+        if event == "token.delta" and data.get("part") == "reasoning"
+    ]
+    text_deltas = [
+        data
+        for event, data in ctx.events
+        if event == "token.delta" and data.get("part", "text") == "text"
+    ]
+    assert reasoning_deltas
+    assert text_deltas
+    reply = (result.output or {}).get("reply", "")
+    assert "".join(e["delta"] for e in text_deltas) == reply
+
+    reasoning_msgs = [
+        m
+        for m in _message_events(ctx.events)
+        if (m.get("extra") or {}).get("kind") == "reasoning"
+    ]
+    assert len(reasoning_msgs) == 1
+    assert "".join(e["delta"] for e in reasoning_deltas) == reasoning_msgs[0]["content"]
+
+    assistant = [
+        m
+        for m in _message_events(ctx.events)
+        if m["role"] == "assistant" and (m.get("extra") or {}).get("kind") != "reasoning"
+    ]
+    assert assistant[-1]["content"] == reply
+
+
 @pytest.mark.asyncio
 async def test_langgraph_stream_tokens_disabled():
     adapter = LangGraphAdapter()

backend/tests/test_threads.py

@@ -243,7 +243,7 @@ async def test_thread_message_pagination(client):
 
 
 @pytest.mark.asyncio
-async def test_load_thread_window_skips_tool_and_prompt_echo():
+async def test_load_thread_window_skips_tool_prompt_echo_and_reasoning():
     from app.db.base import Base
     from app.db.session import SessionLocal, engine
     from app.models import Agent, Message, Run, Thread
@@ -301,7 +301,14 @@ async def test_load_thread_window_skips_tool_and_prompt_echo():
                     content='{"ok":true}',
                     tool_call_id="c1",
                 ),
-                Message(run_id=run.id, index=4, role="assistant", content="done"),
+                Message(
+                    run_id=run.id,
+                    index=4,
+                    role="assistant",
+                    content="secret chain of thought",
+                    extra={"kind": "reasoning"},
+                ),
+                Message(run_id=run.id, index=5, role="assistant", content="done"),
             ]
         )
         await session.commit()

docs/api-contract.md

@@ -381,9 +381,15 @@ The supported `type` values are:
 `token.delta` is SSE-only (not persisted). Payload:
 
 ```json
-{ "step_index": 0, "delta": "Hel", "role": "assistant" }
+{ "step_index": 0, "delta": "Hel", "role": "assistant", "part": "text" }
 ```
 
+`part` discriminates fine-grained streams: `"text"` (default, visible reply)
+or `"reasoning"` (model thinking / chain-of-thought). Finished reasoning is
+persisted as a `message.created` with `extra.kind = "reasoning"` so the
+console can show a collapsible block after reconnect; live deltas remain
+ephemeral.
+
 `step.updated` flushes deferred metrics on a running step (tokens, latency)
 before `step.completed`:
 
@@ -592,11 +598,15 @@ ops dashboards.
   name: string | null;
   content: string;
   tool_call_id: string | null;
-  extra: Record<string, unknown>;
+  extra: Record<string, unknown>;  // kind?: "prompt_echo" | "reasoning" | …
   created_at: string;
 }
 ```
 
+`extra.kind = "reasoning"` marks a persisted reasoning / thinking block
+(content is the full chain-of-thought). These rows stay on the Run transcript
+for the console but are excluded from Thread L1 window seeding.
+
 ### `MessagePage`
 
 ```ts