fix: preserve DeepSeek thinking events

app/chat_service.py

@@ -41,6 +41,7 @@ from langgraph.store.memory import InMemoryStore
 from langgraph.types import Command
 
 from .chat_types import ChatEvent, ChatRequest, ChatTurn, SourceMatch
+from .deepseek_chat import TutorChatDeepSeek
 from .memory_presets import (
     DEFAULT_MEMORY_PRESET,
     MEMORY_PRESETS,
@@ -78,7 +79,7 @@ from .prompts import build_system_prompt, ensure_kb_agents_instructions
 from .provider_events import (
     GoogleSearchActivity,
     extract_anthropic_source_matches,
-    extract_thought_summaries,
+    extract_reasoning_deltas,
 )
 from .config import (
     BM25_INDEX_PATH,
@@ -906,6 +907,12 @@ def is_anthropic_model(model_name: str) -> bool:
     return provider == "anthropic"
 
 
+def is_deepseek_model(model_name: str) -> bool:
+    provider_model = normalize_model_name(model_name)
+    provider, _, _actual_model = provider_model.partition(":")
+    return provider == "deepseek"
+
+
 def supports_gemini_tool_combination(model_name: str) -> bool:
     """True when a Gemini model can mix built-in tools with function tools.
 
@@ -981,18 +988,23 @@ def _build_chat_model_client(provider_model: str, include_thoughts: bool = False
             max_retries=12,
         )
     if provider == "deepseek":
-        # DeepSeek first-party API (OpenAI-compatible, base https://api.deepseek.com).
+        # Use the provider-specific adapter: the generic ChatOpenAI wrapper
+        # intentionally discards DeepSeek's non-standard reasoning_content
+        # field even though the transport itself is OpenAI-compatible.
         # stream_usage=True so the streamed response carries token usage ->
         # context_stats / cost telemetry populates, including the cached-prefix
         # tokens that drive the cost comparison (DeepSeek caches prefixes
         # automatically; cache-hit input is ~50x cheaper than cache-miss). Reads
         # DEEPSEEK_API_KEY.
-        return ChatOpenAI(
+        return TutorChatDeepSeek(
             model=actual_model,
-            temperature=1,
+            temperature=None if include_thoughts else 1,
             base_url="https://api.deepseek.com",
             api_key=os.environ.get("DEEPSEEK_API_KEY"),
             stream_usage=True,
+            extra_body={
+                "thinking": {"type": "enabled" if include_thoughts else "disabled"}
+            },
             # A few retries ride out transient 429/5xx on long agentic sessions
             # without failing the whole session (we run evals at concurrency 1).
             max_retries=6,
@@ -1284,7 +1296,20 @@ class DeepSeekCacheIsolationMiddleware(AgentMiddleware):
         if not user_id:
             return request
         settings = dict(getattr(request, "model_settings", None) or {})
-        extra_body = dict(settings.get("extra_body") or {})
+        model = getattr(request, "model", None)
+        model_extra_body: Any = None
+        # Runnable bindings/fallbacks wrap the underlying provider model. Keep
+        # its DeepSeek thinking toggle when adding the experiment-only user_id;
+        # a call-time extra_body otherwise replaces the model-level mapping.
+        for _ in range(4):
+            model_extra_body = getattr(model, "extra_body", None)
+            if model_extra_body is not None:
+                break
+            model = getattr(model, "runnable", None) or getattr(model, "bound", None)
+            if model is None:
+                break
+        extra_body = dict(model_extra_body or {})
+        extra_body.update(settings.get("extra_body") or {})
         extra_body["user_id"] = user_id
         settings["extra_body"] = extra_body
         return request.override(model_settings=settings)
@@ -2690,6 +2715,7 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
     include_reasoning = bool(request.include_reasoning) and (
         is_google_genai_model(request.model_name)
         or is_anthropic_model(request.model_name)
+        or is_deepseek_model(request.model_name)
     )
     # Gemini streams each thought summary as one complete block; Anthropic
     # streams partial fragments of a single thought. The encoder uses this to
@@ -2777,7 +2803,7 @@ async def stream_chat(request: ChatRequest) -> AsyncIterator[ChatEvent]:
 
                 step = str(metadata.get("langgraph_node", ""))
                 if include_reasoning:
-                    thought_text = "\n\n".join(extract_thought_summaries(token.content))
+                    thought_text = "\n\n".join(extract_reasoning_deltas(token))
                     if thought_text:
                         if first_token_at is None:
                             first_token_at = time.monotonic()

app/deepseek_chat.py

@@ -0,0 +1,58 @@
+"""DeepSeek chat-model compatibility helpers.
+
+``ChatDeepSeek`` preserves the provider's streamed ``reasoning_content`` in
+LangChain message metadata.  DeepSeek thinking-mode tool loops additionally
+require that metadata to be serialized back onto assistant tool-call messages;
+the upstream integration does not currently do that replay step.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from langchain_core.language_models import LanguageModelInput
+from langchain_core.messages import BaseMessage
+from langchain_deepseek import ChatDeepSeek
+
+
+class TutorChatDeepSeek(ChatDeepSeek):
+    """ChatDeepSeek with reasoning replay for thinking-mode tool calls."""
+
+    @staticmethod
+    def _thinking_enabled(payload: dict[str, Any]) -> bool:
+        extra_body = payload.get("extra_body")
+        if not isinstance(extra_body, dict):
+            return False
+        thinking = extra_body.get("thinking")
+        return isinstance(thinking, dict) and thinking.get("type") == "enabled"
+
+    def _original_messages(self, input_: LanguageModelInput) -> list[BaseMessage]:
+        return self._convert_input(input_).to_messages()
+
+    def _get_request_payload(
+        self,
+        input_: LanguageModelInput,
+        *,
+        stop: list[str] | None = None,
+        **kwargs: Any,
+    ) -> dict[str, Any]:
+        payload = super()._get_request_payload(input_, stop=stop, **kwargs)
+        if not self._thinking_enabled(payload):
+            return payload
+
+        original_messages = self._original_messages(input_)
+        for index, message in enumerate(payload.get("messages") or []):
+            if message.get("role") != "assistant" or not message.get("tool_calls"):
+                continue
+            reasoning_content = ""
+            if index < len(original_messages):
+                value = original_messages[index].additional_kwargs.get(
+                    "reasoning_content"
+                )
+                if isinstance(value, str):
+                    reasoning_content = value
+            # DeepSeek requires the top-level field to exist on every replayed
+            # assistant tool-call message in thinking mode. An empty value is
+            # still preferable to omitting the field and receiving a 400.
+            message["reasoning_content"] = reasoning_content
+        return payload

app/provider_events.py

@@ -40,6 +40,22 @@ def extract_thought_summaries(content: Any) -> list[str]:
     return thoughts
 
 
+def extract_reasoning_deltas(message: Any) -> list[str]:
+    """Extract provider reasoning fragments from a streamed LangChain message.
+
+    Gemini and Anthropic expose reasoning as typed content blocks. DeepSeek's
+    provider adapter preserves its sibling ``reasoning_content`` response field
+    in ``additional_kwargs`` instead.
+    """
+    thoughts = extract_thought_summaries(getattr(message, "content", None))
+    additional_kwargs = getattr(message, "additional_kwargs", None)
+    if isinstance(additional_kwargs, dict):
+        reasoning_content = additional_kwargs.get("reasoning_content")
+        if isinstance(reasoning_content, str) and reasoning_content.strip():
+            thoughts.append(reasoning_content)
+    return thoughts
+
+
 def extract_web_search_queries(response_metadata: Any) -> list[str]:
     """Pull the queries Gemini ran against google_search from grounding metadata."""
     if not isinstance(response_metadata, dict):

pyproject.toml

@@ -14,6 +14,7 @@ dependencies = [
     "huggingface-hub",
     "langchain",
     "langchain-anthropic",
+    "langchain-deepseek>=1.1.0",
     "langchain-google-genai",
     "langchain-openai",
     "llama-index",

tests/test_chat_service.py

@@ -10,6 +10,7 @@ from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, Too
 from langchain_core.runnables.fallbacks import RunnableWithFallbacks
 
 from app.config import DEEPSEEK_DIRECT_MODEL_NAME, GEMINI_FALLBACK_MODEL_NAME
+from app.deepseek_chat import TutorChatDeepSeek
 from app.chat_service import (
     THREAD_IDLE_TTL_SECONDS,
     _claim_kb_command_budget,
@@ -130,6 +131,32 @@ class FakeAnswerAgent(FakeAgent):
         }
 
 
+class FakeDeepSeekReasoningAgent(FakeAgent):
+    async def astream(self, *args, **kwargs):
+        for fragment in ("Inspect ", "the course sources."):
+            yield {
+                "type": "messages",
+                "data": (
+                    AIMessageChunk(
+                        content="",
+                        additional_kwargs={"reasoning_content": fragment},
+                    ),
+                    {"langgraph_node": "model"},
+                ),
+            }
+        yield {
+            "type": "messages",
+            "data": (
+                AIMessageChunk(content="Final answer"),
+                {"langgraph_node": "model"},
+            ),
+        }
+        yield {
+            "type": "updates",
+            "data": {"model": {"messages": [AIMessage(content="Final answer")]}},
+        }
+
+
 class FakeToolThenTextAgent(FakeAgent):
     """Streams a tool-call token first, then streamed answer text, so a turn
     sets first_token_at (the tool call) strictly before first_text_at (the
@@ -421,9 +448,12 @@ class ChatServiceTestCase(unittest.TestCase):
             model = build_chat_model(DEEPSEEK_DIRECT_MODEL_NAME)
 
         self.assertIsInstance(model, RunnableWithFallbacks)
+        self.assertIsInstance(model.runnable, TutorChatDeepSeek)
         self.assertEqual(model.runnable.model_name, "deepseek-v4-flash")
+        self.assertEqual(str(model.runnable.api_base), "https://api.deepseek.com")
         self.assertEqual(
-            str(model.runnable.openai_api_base), "https://api.deepseek.com"
+            model.runnable.extra_body,
+            {"thinking": {"type": "disabled"}},
         )
         self.assertEqual(len(model.fallbacks), 1)
         self.assertEqual(
@@ -442,8 +472,58 @@ class ChatServiceTestCase(unittest.TestCase):
             model = build_chat_model(DEEPSEEK_DIRECT_MODEL_NAME)
 
         self.assertNotIsInstance(model, RunnableWithFallbacks)
+        self.assertIsInstance(model, TutorChatDeepSeek)
         self.assertEqual(model.model_name, "deepseek-v4-flash")
 
+    def test_deepseek_reasoning_mode_is_explicitly_enabled(self) -> None:
+        with patch.dict(
+            os.environ,
+            {"DEEPSEEK_API_KEY": "deepseek-test-key"},
+            clear=True,
+        ):
+            model = build_chat_model(
+                DEEPSEEK_DIRECT_MODEL_NAME,
+                include_thoughts=True,
+            )
+
+        self.assertIsInstance(model, TutorChatDeepSeek)
+        self.assertIsNone(model.temperature)
+        self.assertEqual(
+            model.extra_body,
+            {"thinking": {"type": "enabled"}},
+        )
+
+    def test_deepseek_thinking_tool_call_replays_reasoning_content(self) -> None:
+        model = TutorChatDeepSeek(
+            model="deepseek-v4-flash",
+            api_key="deepseek-test-key",
+            extra_body={"thinking": {"type": "enabled"}},
+        )
+        messages = [
+            HumanMessage(content="Find the course repository"),
+            AIMessage(
+                content="",
+                tool_calls=[
+                    {
+                        "id": "call_1",
+                        "name": "retrieve_tutor_context",
+                        "args": {"query": "course repository"},
+                    }
+                ],
+                additional_kwargs={
+                    "reasoning_content": "I should search the selected course."
+                },
+            ),
+            ToolMessage(content="result", tool_call_id="call_1"),
+        ]
+
+        payload = model._get_request_payload(messages)
+
+        self.assertEqual(
+            payload["messages"][1]["reasoning_content"],
+            "I should search the selected course.",
+        )
+
     def test_deepseek_direct_default_uses_gemini_when_deepseek_key_missing(
         self,
     ) -> None:
@@ -757,6 +837,40 @@ class ChatServiceTestCase(unittest.TestCase):
         # latency never exceeds time-to-first-answer.
         self.assertLessEqual(first_token, ttft)
 
+    def test_stream_chat_emits_deepseek_reasoning_content(self) -> None:
+        agent = FakeDeepSeekReasoningAgent([])
+        build_agent_mock = MagicMock(return_value=agent)
+        self.addCleanup(_drop_thread_record, "thread_deepseek_reasoning")
+        request = ChatRequest(
+            query="Find the course repository",
+            source_keys=("peft",),
+            model_name=DEEPSEEK_DIRECT_MODEL_NAME,
+            include_reasoning=True,
+            enabled_tools=(),
+        )
+
+        async def collect_events():
+            return [event async for event in stream_chat(request)]
+
+        with (
+            patch("app.chat_service.build_agent", build_agent_mock),
+            patch(
+                "app.chat_service.new_thread_id",
+                return_value="thread_deepseek_reasoning",
+            ),
+        ):
+            events = asyncio.run(collect_events())
+
+        reasoning = [
+            event.data["text"] for event in events if event.type == "reasoning_delta"
+        ]
+        self.assertEqual(reasoning, ["Inspect ", "the course sources."])
+        self.assertEqual(
+            [event.data["text"] for event in events if event.type == "text_delta"],
+            ["Final answer"],
+        )
+        self.assertTrue(build_agent_mock.call_args.kwargs["include_thoughts"])
+
     def test_stream_chat_resolves_shell_citation_after_final_answer(self) -> None:
         agent = FakeStreamingAgent([])
         self.addCleanup(_drop_thread_record, "thread_rg")

tests/test_memory_variants.py

@@ -383,14 +383,24 @@ class ExperimentCompactionMiddlewareTests(unittest.TestCase):
             context=SimpleNamespace(cache_user_id="eval_abc", kb_session_id="turn")
         )
         request = SimpleNamespace(
-            runtime=runtime, model_settings={}, messages=[], system_message=None
+            runtime=runtime,
+            model=SimpleNamespace(extra_body={"thinking": {"type": "enabled"}}),
+            model_settings={},
+            messages=[],
+            system_message=None,
         )
         request.override = lambda **updates: SimpleNamespace(
             runtime=runtime,
             model_settings=updates.get("model_settings", request.model_settings),
         )
         isolated = DeepSeekCacheIsolationMiddleware()._isolate(request)
-        self.assertEqual(isolated.model_settings["extra_body"], {"user_id": "eval_abc"})
+        self.assertEqual(
+            isolated.model_settings["extra_body"],
+            {
+                "thinking": {"type": "enabled"},
+                "user_id": "eval_abc",
+            },
+        )
 
     def test_agent_request_guard_fails_before_model_handler(self) -> None:
         runtime = SimpleNamespace(

uv.lock

@@ -22,6 +22,7 @@ dependencies = [
     { name = "huggingface-hub" },
     { name = "langchain" },
     { name = "langchain-anthropic" },
+    { name = "langchain-deepseek" },
     { name = "langchain-google-genai" },
     { name = "langchain-openai" },
     { name = "llama-index" },
@@ -61,6 +62,7 @@ requires-dist = [
     { name = "huggingface-hub" },
     { name = "langchain" },
     { name = "langchain-anthropic" },
+    { name = "langchain-deepseek", specifier = ">=1.1.0" },
     { name = "langchain-google-genai" },
     { name = "langchain-openai" },
     { name = "llama-index" },
@@ -2016,6 +2018,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/13/d6/bdf6f0481cc57ef300d6b1eb48cf1400c0409be715d6eb3cabadd1142a09/langchain_core-1.4.8-py3-none-any.whl", hash = "sha256:d84c28b05e3ba8d4271d0827aad5b592ccdaaf986e76768c23503f0a2045e8aa", size = 557416, upload-time = "2026-06-18T19:39:21.902Z" },
 ]
 
+[[package]]
+name = "langchain-deepseek"
+version = "1.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "langchain-core" },
+    { name = "langchain-openai" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/15/f4/200792013d86406aa02aba8d1b536f8877a832372702189aef0c8c7a8329/langchain_deepseek-1.1.0.tar.gz", hash = "sha256:5ec8128cabae3d71366e2f297dea2153649be317a4be7f135cfcad1160523bf6", size = 138078, upload-time = "2026-06-03T19:07:11.887Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/8c/1a/d18b7f985c35c635503a6cd426161933ffc7785f228d92447c63b026299a/langchain_deepseek-1.1.0-py3-none-any.whl", hash = "sha256:14813cb413a97a5cce95118da253cfd64dce50537b7381b7c5d0ecf11d2a7032", size = 10061, upload-time = "2026-06-03T19:07:11.04Z" },
+]
+
 [[package]]
 name = "langchain-google-genai"
 version = "4.2.7"
@@ -3135,7 +3150,7 @@ name = "pexpect"
 version = "4.9.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "ptyprocess" },
+    { name = "ptyprocess", marker = "sys_platform != 'win32'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" }
 wheels = [