core/agent: full reasoning_text.done replaces streamed fragments

src/plyngent/agent/loop.py

@@ -16,6 +16,7 @@ from plyngent.lmproto.openai_compatible.model import (
     ChatCompletionChunk,
     ChatCompletionResponse,
     ChatCompletionsParam,
+    DeltaMessage,
     StreamOptions,
     StreamToolCallDelta,
     ToolChatMessage,
@@ -254,6 +255,18 @@ async def _non_stream_round(
     )
 
 
+def _append_reasoning_delta(
+    reasoning_parts: list[str],
+    delta: DeltaMessage,
+) -> list[str]:
+    """Accumulate a reasoning delta; a ``full`` (``.done``) event replaces fragments."""
+    if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
+        if delta.reasoning_full is True:
+            return [delta.reasoning_content]
+        return [*reasoning_parts, delta.reasoning_content]
+    return reasoning_parts
+
+
 async def _stream_round(
     client: AnyLLMClient,
     param: ChatCompletionsParam,
@@ -295,7 +308,7 @@ async def _stream_round(
             saw_terminal = True
         delta = choice.delta
         if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
-            reasoning_parts.append(delta.reasoning_content)
+            reasoning_parts = _append_reasoning_delta(reasoning_parts, delta)
             yield ReasoningDeltaEvent(content=delta.reasoning_content)
         if isinstance(delta.content, str) and delta.content:
             content_parts.append(delta.content)

src/plyngent/agent/responses_dispatch.py

@@ -47,13 +47,15 @@ if TYPE_CHECKING:
 
 # Responses stream event types that carry reasoning text deltas.
 # OpenAI streams ``response.reasoning_summary_text.delta``; DeepSeek streams
-# ``response.reasoning_text.delta`` and emits the full text once via
-# ``response.reasoning_text.done``.
+# ``response.reasoning_text.delta`` fragments and emits the FULL chain-of-thought
+# once via ``response.reasoning_text.done``. Fragments append; a ``.done`` event
+# carries the complete text and must replace the accumulated fragments (otherwise
+# the CoT is concatenated twice).
 _REASONING_DELTA_TYPES = {
     "response.reasoning_summary_text.delta",
     "response.reasoning_text.delta",
-    "response.reasoning_text.done",
 }
+_REASONING_FULL_TYPES = {"response.reasoning_text.done"}
 # Terminal events that carry the full Response object (status, tool calls, usage).
 # ``response.failed`` also carries a Response but is surfaced as an error.
 _FINAL_EVENT_TYPES = {"response.completed", "response.incomplete"}
@@ -98,12 +100,19 @@ def _decode_final_event(event: ResponseStreamEvent) -> ResponseModel | None:
         return None
 
 
-def _stream_reasoning_content(event: ResponseStreamEvent) -> str | None:
-    """Return reasoning text from a reasoning stream event (delta or full-text done)."""
-    if event.type not in _REASONING_DELTA_TYPES:
-        return None
+def _stream_reasoning_content(event: ResponseStreamEvent) -> tuple[str | None, bool]:
+    """Return (reasoning_text, is_full) from a reasoning stream event.
+
+    ``.delta`` events carry incremental fragments (``is_full=False``); ``.done``
+    events carry the full chain-of-thought (``is_full=True``) and must replace
+    the accumulated fragments.
+    """
+    if event.type not in _REASONING_DELTA_TYPES and event.type not in _REASONING_FULL_TYPES:
+        return None, False
     content = event.delta if isinstance(event.delta, str) else event.text
-    return content if isinstance(content, str) and content else None
+    if not isinstance(content, str) or not content:
+        return None, False
+    return content, event.type in _REASONING_FULL_TYPES
 
 
 def _terminal_chunks(final: ResponseModel, *, model: str) -> list[ChatCompletionChunk]:
@@ -148,10 +157,10 @@ async def _stream_as_chat_chunks(
         if etype == "response.output_text.delta" and isinstance(event.delta, str) and event.delta:
             yield text_delta_chunk(model=model, content=event.delta)
             continue
-        reasoning = _stream_reasoning_content(event)
+        reasoning, reasoning_full = _stream_reasoning_content(event)
         if reasoning is not None:
             saw_reasoning = True
-            yield reasoning_delta_chunk(model=model, content=reasoning)
+            yield reasoning_delta_chunk(model=model, content=reasoning, full=reasoning_full)
             continue
 
     if final is None:

src/plyngent/agent/stream_chunks.py

@@ -58,7 +58,13 @@ def text_delta_chunk(*, model: str, content: str, created: int = 0) -> ChatCompl
     )
 
 
-def reasoning_delta_chunk(*, model: str, content: str, created: int = 0) -> ChatCompletionChunk:
+def reasoning_delta_chunk(
+    *,
+    model: str,
+    content: str,
+    created: int = 0,
+    full: bool = False,
+) -> ChatCompletionChunk:
     return ChatCompletionChunk(
         id="bridge-stream",
         object="chat.completion.chunk",
@@ -67,7 +73,7 @@ def reasoning_delta_chunk(*, model: str, content: str, created: int = 0) -> Chat
         choices=[
             ChunkChoice(
                 index=0,
-                delta=DeltaMessage(reasoning_content=content),
+                delta=DeltaMessage(reasoning_content=content, reasoning_full=full or UNSET),
             )
         ],
     )

src/plyngent/lmproto/openai_compatible/model.py

@@ -239,6 +239,9 @@ class DeltaMessage(Struct):
     # Providers often send content=null / reasoning_content=null on partial chunks.
     content: str | None | Unset = UNSET
     reasoning_content: str | None | Unset = UNSET
+    # True when reasoning_content is the FULL chain-of-thought (e.g. DeepSeek's
+    # ``response.reasoning_text.done``), replacing any streamed fragments.
+    reasoning_full: bool | Unset = UNSET
     tool_calls: list[StreamToolCallDelta] | Unset = UNSET
 
 

tests/test_agent/test_responses_dispatch.py

@@ -186,7 +186,11 @@ async def test_dispatch_responses_stream_text_and_usage() -> None:
 async def _streamed_reasoning(
     events: list[ResponseStreamEvent],
 ) -> str:
-    """Run one Responses stream and join all streamed ``reasoning_content`` deltas."""
+    """Run one Responses stream and join streamed ``reasoning_content`` deltas.
+
+    Mirrors the agent loop: ``.done`` (``reasoning_full``) replaces accumulated
+    fragments instead of appending.
+    """
     import msgspec
 
     final = _completed_response(text="ok")
@@ -202,7 +206,10 @@ async def _streamed_reasoning(
             continue
         delta = chunk.choices[0].delta
         if isinstance(delta.reasoning_content, str) and delta.reasoning_content:
-            parts.append(delta.reasoning_content)
+            if delta.reasoning_full is True:
+                parts = [delta.reasoning_content]
+            else:
+                parts.append(delta.reasoning_content)
     return "".join(parts)
 
 
@@ -224,6 +231,19 @@ async def test_dispatch_responses_stream_reasoning_deltas() -> None:
     assert reasoning == "step one step two"
 
 
+async def test_dispatch_responses_stream_reasoning_full_replaces_fragments() -> None:
+    """A ``.done`` event with the full CoT must replace streamed fragments, not
+    append — otherwise the chain-of-thought is concatenated twice."""
+    reasoning = await _streamed_reasoning(
+        [
+            ResponseStreamEvent(type="response.reasoning_text.delta", delta="partial "),
+            ResponseStreamEvent(type="response.reasoning_text.delta", delta="thinking"),
+            ResponseStreamEvent(type="response.reasoning_text.done", delta="full chain of thought"),
+        ]
+    )
+    assert reasoning == "full chain of thought"
+
+
 async def test_dispatch_responses_stream_terminal_only_reasoning() -> None:
     """DeepSeek sometimes delivers reasoning only on the terminal response."""
     reasoning = await _streamed_reasoning([])