fix: deepseek DSML leak

src/xagent/core/agent/pattern/react/react.py

@@ -385,6 +385,15 @@ async def _run_tool_calling_loop(
                     "tools": tool_schemas or None,
                     "tool_choice": self.tool_choice if tool_schemas else None,
                 }
+                start_final_answer_stream_immediately = True
+                if force_final_answer_now:
+                    parser_kwargs = self._forced_final_parser_kwargs(
+                        llm=llm,
+                        tool_schemas=base_tool_schemas,
+                    )
+                    if parser_kwargs:
+                        llm_kwargs.update(parser_kwargs)
+                        start_final_answer_stream_immediately = False
                 if tool_schemas:
                     answer_streamer = ReActFinalAnswerStreamer(runtime)
                     response = await runtime.run_streaming_llm_call(
@@ -393,7 +402,11 @@ async def _run_tool_calling_loop(
                         **llm_kwargs,
                     )
                 else:
-                    response = await runtime.stream_final_answer(llm, **llm_kwargs)
+                    response = await runtime.stream_final_answer(
+                        llm,
+                        start_immediately=start_final_answer_stream_immediately,
+                        **llm_kwargs,
+                    )
             except LLMCallInterrupted:
                 if answer_streamer is not None:
                     await answer_streamer.fail("interrupted during LLM stream")
@@ -445,6 +458,8 @@ async def _run_tool_calling_loop(
                 self._remember_tool_call_content(tool_calls, assistant_content)
                 self.status = "acting"
                 self.pending_tool_calls = list(tool_calls)
+                if force_final_answer_now:
+                    self.force_final_answer_next = False
                 await runtime.checkpoint("after_llm", context=context, pattern=self)
                 pending_result = await self._execute_pending_tool_calls(
                     context=context,
@@ -584,6 +599,22 @@ def _schema_tool_names(self, tool_schemas: list[dict[str, Any]]) -> list[str]:
                 names.append(str(function["name"]))
         return names
 
+    def _forced_final_parser_kwargs(
+        self,
+        *,
+        llm: Any,
+        tool_schemas: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        """Return provider-local parser kwargs for final turns that suppress tools."""
+
+        if not tool_schemas:
+            return {}
+        parse_kwargs = getattr(llm, "deepseek_dsml_parse_kwargs", None)
+        if not callable(parse_kwargs):
+            return {}
+        result = parse_kwargs(tool_schemas)
+        return result if isinstance(result, dict) else {}
+
     def _tool_decision_groups_for_tools(self, tools: list[Any]) -> dict[str, str]:
         groups: dict[str, str] = {}
         for tool in tools:

src/xagent/core/agent/runtime.py

@@ -95,7 +95,13 @@ async def run_llm_call(self, llm: Any, **kwargs: Any) -> Any:
         finally:
             self._active_llm_tasks.discard(task)
 
-    async def stream_final_answer(self, llm: Any, **kwargs: Any) -> Any:
+    async def stream_final_answer(
+        self,
+        llm: Any,
+        *,
+        start_immediately: bool = True,
+        **kwargs: Any,
+    ) -> Any:
         """Stream only the final user-facing answer through the outbound boundary."""
 
         if self.outbound_message_handler is None or not self._has_native_stream_chat(
@@ -106,12 +112,13 @@ async def stream_final_answer(self, llm: Any, **kwargs: Any) -> Any:
         from .pattern.final_answer_stream import FinalAnswerStreamSession
 
         stream = FinalAnswerStreamSession(self)
-        if await stream.start() is None:
+        if start_immediately and await stream.start() is None:
             return await self.run_llm_call(llm, **kwargs)
+        defer_text_delta = not start_immediately
 
         async def emit_text_delta(chunk: Any) -> None:
             delta = self._chunk_text_delta(chunk)
-            if delta:
+            if delta and not defer_text_delta:
                 await stream.emit_delta(delta)
 
         try:
@@ -123,6 +130,13 @@ async def emit_text_delta(chunk: Any) -> None:
             raise
         content = self._response_content(response)
 
+        if self._response_tool_calls(response):
+            if stream.started:
+                await stream.fail(
+                    "Final-answer stream stopped because the response contained tool calls."
+                )
+            return response
+
         await stream.finish(content)
         return response
 
@@ -373,6 +387,12 @@ def _response_content(self, response: Any) -> str:
             return str(value)
         return str(response)
 
+    def _response_tool_calls(self, response: Any) -> list[Any]:
+        if not isinstance(response, dict):
+            return []
+        tool_calls = response.get("tool_calls")
+        return list(tool_calls) if tool_calls else []
+
     async def _emit_outbound(self, payload: dict[str, Any]) -> None:
         if self.outbound_message_handler is not None:
             await self._maybe_await(self.outbound_message_handler(payload))

src/xagent/core/model/chat/basic/deepseek.py

@@ -4,6 +4,12 @@
 
 from ...providers import is_placeholder_api_key
 from .base import StreamChunk
+from .deepseek_dsml import (
+    DEEPSEEK_DSML_PARSE_TOOLS_KWARG,
+    DeepSeekDSMLParseError,
+    normalize_deepseek_dsml_response,
+    stream_chunks_from_chat_result,
+)
 from .openai import PROVIDER_STATE_METADATA_KEY, OpenAICompatibleLLM
 
 logger = logging.getLogger(__name__)
@@ -131,6 +137,33 @@ def _prepare_deepseek_kwargs(
 
         return updated_kwargs
 
+    def deepseek_dsml_parse_kwargs(
+        self,
+        tools: list[dict[str, Any]] | None,
+    ) -> dict[str, Any]:
+        """Return parser-only DSML schema metadata for forced-final recovery.
+
+        ReAct forced-final turns intentionally send ``tools=None`` to DeepSeek.
+        If DeepSeek still emits DSML tool markup, this hidden kwarg lets this
+        adapter validate and convert it locally without leaking the schema into
+        provider request parameters.
+        """
+
+        return {DEEPSEEK_DSML_PARSE_TOOLS_KWARG: tools} if tools else {}
+
+    def _pop_deepseek_dsml_parse_tools(
+        self,
+        kwargs: dict[str, Any],
+    ) -> list[dict[str, Any]] | None:
+        raw_tools = kwargs.pop(DEEPSEEK_DSML_PARSE_TOOLS_KWARG, None)
+        if raw_tools is None:
+            return None
+        if not isinstance(raw_tools, list):
+            raise DeepSeekDSMLParseError(
+                f"{DEEPSEEK_DSML_PARSE_TOOLS_KWARG} must be a list of tool schemas."
+            )
+        return raw_tools
+
     def _prepare_messages_for_request(
         self,
         messages: List[Dict[str, Any]],
@@ -219,6 +252,23 @@ def _normalize_response_format(
 
         return response_format, output_config
 
+    def _normalize_deepseek_response(
+        self,
+        response: Any,
+        *,
+        tools: Optional[List[Dict[str, Any]]],
+    ) -> Any:
+        normalized_response, parsed_dsml = normalize_deepseek_dsml_response(
+            response,
+            tools=tools,
+            model_name=self._model_name,
+        )
+        if parsed_dsml and isinstance(normalized_response, dict):
+            provider_state = self._response_provider_state(normalized_response)
+            if provider_state:
+                normalized_response[PROVIDER_STATE_METADATA_KEY] = provider_state
+        return normalized_response
+
     async def chat(
         self,
         messages: List[Dict[str, str]],
@@ -231,12 +281,13 @@ async def chat(
         output_config: Optional[Dict[str, Any]] = None,
         **kwargs: Any,
     ) -> Any:
+        dsml_parse_tools = self._pop_deepseek_dsml_parse_tools(kwargs)
         response_format, output_config = self._normalize_response_format(
             response_format=response_format,
             output_config=output_config,
         )
         kwargs = self._prepare_deepseek_kwargs(kwargs=kwargs)
-        return await super().chat(
+        result = await super().chat(
             messages=messages,
             temperature=temperature,
             max_tokens=max_tokens,
@@ -247,6 +298,10 @@ async def chat(
             output_config=output_config,
             **kwargs,
         )
+        return self._normalize_deepseek_response(
+            result,
+            tools=tools or dsml_parse_tools,
+        )
 
     async def stream_chat(
         self,
@@ -260,6 +315,26 @@ async def stream_chat(
         output_config: Optional[Dict[str, Any]] = None,
         **kwargs: Any,
     ) -> AsyncIterator[StreamChunk]:
+        # DSML recovery needs a complete response. Keep ordinary tool-enabled
+        # streams native; only forced-final turns opt into buffering with the
+        # private sentinel because those calls intentionally suppress provider
+        # tools while still needing local DSML validation.
+        if DEEPSEEK_DSML_PARSE_TOOLS_KWARG in kwargs:
+            result = await self.chat(
+                messages=messages,
+                temperature=temperature,
+                max_tokens=max_tokens,
+                tools=tools,
+                tool_choice=tool_choice,
+                response_format=response_format,
+                thinking=thinking,
+                output_config=output_config,
+                **kwargs,
+            )
+            for chunk in stream_chunks_from_chat_result(result):
+                yield chunk
+            return
+
         response_format, output_config = self._normalize_response_format(
             response_format=response_format,
             output_config=output_config,