Merge pull request #177 from EvanProgramming/codex/issue-171-deepseek-controls

fix: normalize DeepSeek controls and sanitize streams

Author: EvanProgramming

interaction.py

@@ -182,6 +182,53 @@ def parse_control_block(text: str, name: str) -> Any | None:
         raise InteractionError(f"{name} contains invalid JSON: {exc.msg}") from exc
 
 
+def normalize_provider_control(text: str) -> tuple[str, dict[str, Any]] | None:
+    """Map a bounded provider JSON shape to one non-executable control."""
+    source = str(text or "").strip()
+    fenced = re.fullmatch(r"```(?:json)?\s*([\s\S]*?)\s*```", source, re.IGNORECASE)
+    if fenced:
+        source = fenced.group(1).strip()
+    try:
+        value = json.loads(source)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(value, dict):
+        return None
+
+    executable_keys = {"action", "args", "arguments", "command", "tool", "tool_calls"}
+    stack: list[Any] = [value]
+    while stack:
+        item = stack.pop()
+        if isinstance(item, dict):
+            if executable_keys & {str(key).lower() for key in item}:
+                return None
+            stack.extend(item.values())
+        elif isinstance(item, list):
+            stack.extend(item)
+
+    mode = str(value.get("mode") or "").strip().lower()
+    if mode == "plan" or value.get("do_not_execute_until_approved") is True:
+        steps = []
+        for index, step in enumerate(value.get("steps") or [], 1):
+            if not isinstance(step, dict):
+                return None
+            steps.append({
+                "id": str(step.get("id") or step.get("step_id") or f"step-{step.get('step', index)}"),
+                "title": step.get("title") or f"Step {index}",
+                "description": step.get("description") or step.get("details"),
+                "acceptance": step.get("acceptance") or step.get("acceptance_criteria"),
+            })
+        return "PlanProposal", {
+            "title": value.get("title") or value.get("plan_name"),
+            "summary": value.get("summary") or value.get("overview"),
+            "assumptions": value.get("assumptions") or [],
+            "steps": steps,
+        }
+    if mode == "ask" or value.get("ask_user") is True:
+        return "AskUser", {"questions": value.get("questions")}
+    return None
+
+
 def route_mode(preference: str, user_input: str, *, pending_plan: bool = False,
                executing_plan: bool = False) -> str:
     if executing_plan:

main.py

@@ -105,7 +105,7 @@ def _terminal_supports_unicode() -> bool:
 from event_store import stable_hash, utc_now
 from interaction import (
     InteractionController, InteractionError, is_plan_acceptance, mode_capabilities,
-    parse_control_block, render_plan, render_question, validate_plan_proposal,
+    normalize_provider_control, parse_control_block, render_plan, render_question, validate_plan_proposal,
     validate_question_request,
 )
 from learning_engine import LearningEngine
@@ -4357,14 +4357,17 @@ class DeepSeekDSMLFilter:
         r"(?i)(?<![\w])(?P<kind>Action|Thought|Plan|TaskList|TaskDone|DefineTool)\s*:"
     )
     _GENERIC_OPEN_RE = re.compile(
-        r"<\s*(?P<kind>invoke|parameter|calls|tool_calls|function_calls)\b[^>]*>",
+        r"<\s*(?P<kind>invoke|parameter|calls|tool_calls|function_calls|tool_use|notes|thought|reasoning)\b[^>]*>",
         re.IGNORECASE,
     )
     _GENERIC_CLOSE_RE = re.compile(
-        r"</\s*(?P<kind>invoke|parameter|calls|tool_calls|function_calls)\s*>",
+        r"</\s*(?P<kind>invoke|parameter|calls|tool_calls|function_calls|tool_use|notes|thought|reasoning)\s*>",
         re.IGNORECASE,
     )
-    _GENERIC_KINDS = ("invoke", "parameter", "calls", "tool_calls", "function_calls")
+    _GENERIC_KINDS = (
+        "invoke", "parameter", "calls", "tool_calls", "function_calls",
+        "tool_use", "notes", "thought", "reasoning",
+    )
     _ACTION_MARKER_RE = _ACTION_MARKER_RE
     _CONTROL_PREFIXES = tuple(
         item[:length].lower()
@@ -4494,7 +4497,10 @@ def _action_end(cls, value: str, start: int, *, final: bool) -> int | None:
     @classmethod
     def _generic_block_end(cls, value: str, match: re.Match[str]) -> int | None:
         kind = match.group("kind").lower()
-        close_kinds = "parameter" if kind == "parameter" else "invoke|calls|tool_calls|function_calls"
+        close_kinds = (
+            "parameter" if kind == "parameter" else
+            "invoke|calls|tool_calls|function_calls|tool_use|notes|thought|reasoning"
+        )
         close = re.compile(rf"</\s*(?:{close_kinds})\s*>", re.IGNORECASE).search(value, match.end())
         return close.end() if close else None
 
@@ -4760,6 +4766,12 @@ def _parse_model_response(text: str) -> dict[str, Any]:
     try:
         question_value = parse_control_block(raw, "AskUser")
         plan_value = parse_control_block(raw, "PlanProposal")
+        if question_value is None and plan_value is None:
+            normalized = normalize_provider_control(raw)
+            if normalized is not None:
+                name, value = normalized
+                question_value = value if name == "AskUser" else None
+                plan_value = value if name == "PlanProposal" else None
         if question_value is not None:
             question = validate_question_request(question_value)
         if plan_value is not None:

server.py

@@ -1127,7 +1127,8 @@ class StreamProjection:
         """Pass plain deltas through while holding model control prefixes."""
 
         prefixes = ("Thought:", "Plan:", "TaskList:", "TaskDone:", "Action:", "DefineTool:",
-                    "AskUser:", "AskUser\n", "PlanProposal:", "PlanProposal\n")
+                    "AskUser:", "AskUser\n", "PlanProposal:", "PlanProposal\n",
+                    "{", "[", "```json")
 
         def __init__(self, sink):
             self.sink = sink
@@ -1163,7 +1164,9 @@ def _flush(self) -> None:
                 return
             text = self.buffer
             self.buffer = ""
-            if _agent._collect_tool_calls(text) or any(
+            if _agent.normalize_provider_control(text):
+                text = ""
+            elif _agent._collect_tool_calls(text) or any(
                     text.lstrip().lower().startswith(prefix.lower()) for prefix in self.prefixes
             ):
                 text = _agent._clean_final_response(text)

tests/test_interaction.py

@@ -1,3 +1,4 @@
+import json
 import tempfile
 import unittest
 from pathlib import Path
@@ -168,6 +169,27 @@ def test_exact_acceptance_phrases_and_control_parser(self):
             'PlanProposal\n```json\n{"steps": []}\n```', "PlanProposal",
         ), {"steps": []})
 
+    def test_provider_shaped_plan_is_normalized_but_actions_are_not(self):
+        shaped = json.dumps({
+            "plan_name": "Status page", "mode": "plan",
+            "do_not_execute_until_approved": True,
+            "overview": "Build after approval.", "assumptions": ["Python exists"],
+            "steps": [{"step": 1, "title": "Create page", "details": "Write index.html.",
+                       "acceptance_criteria": ["index.html exists"]}],
+        })
+        parsed = main._observe_model_response(shaped)
+        self.assertEqual(parsed["plan_proposal"]["title"], "Status page")
+        self.assertEqual(parsed["plan_proposal"]["steps"][0]["id"], "step-1")
+        self.assertEqual(parsed["tool_calls"], [])
+
+        executable = json.dumps({
+            "mode": "plan", "plan_name": "Unsafe", "overview": "Write now.",
+            "steps": [{"title": "Write", "details": "Write README.",
+                       "acceptance_criteria": ["written"], "action": "write_file"}],
+        })
+        rejected = main._observe_model_response(executable)
+        self.assertIsNone(rejected["plan_proposal"])
+
     def test_control_blocks_combined_with_actions_fail_closed(self):
         parsed = main._observe_model_response(
             'AskUser:\n```json\n{"questions":[{"id":"scope","header":"Scope",'

tests/test_streaming.py

@@ -122,6 +122,18 @@ def chat_stream(self, _messages, _model=None):
 
 
 class StreamingEndpointTests(unittest.TestCase):
+    def test_reasoning_and_tool_use_tags_are_filtered_at_every_boundary(self):
+        marker = (
+            "<notes><thought>private</thought>"
+            "<tool_use><read_file><path>README.md</path></read_file></tool_use></notes>"
+        )
+        expected = "Visible  answer."
+        for split in range(1, len(marker)):
+            stream_filter = main.DeepSeekDSMLFilter()
+            content = stream_filter.feed("Visible " + marker[:split])
+            content += stream_filter.feed(marker[split:] + " answer.", final=True)
+            self.assertEqual(content, expected, split)
+
     def test_first_sse_content_arrives_before_provider_stream_completes(self):
         session_id = "stream-timing-regression"
         provider = _DelayedProvider()

tests/test_tui_protocol.py

@@ -72,6 +72,13 @@ def test_stream_projection_preserves_event_order_and_redacts_receipts(self):
         self.assertEqual([event["event"] for event in events], ["stream_delta", "tool_receipt", "tasks"])
         self.assertEqual(events[0]["request_id"], "turn-1")
 
+    def test_provider_shaped_plan_json_is_not_streamed_as_chat_text(self):
+        callback = self.backend._stream_projection("turn-plan")
+        callback({"event": "content", "chunk": '{"mode":"plan","plan_name":"Safe",'})
+        callback({"event": "content", "chunk": '"overview":"Wait","steps":[]}'})
+        callback({"event": "model_complete"})
+        self.assertEqual(self.output.getvalue(), "")
+
     def test_provider_setup_emits_ready_and_masks_key_flow(self):
         config = tui_backend.agent.ProviderConfig(provider="deepseek")
         with patch.object(tui_backend.agent, "_provider_config", config), \

tui_backend.py

@@ -274,7 +274,8 @@ def _stream_projection(self, request_id: str):
         dsml = agent.DeepSeekDSMLFilter()
         buffer = ""
         prefixes = ("action:", "tasklist:", "taskdone:", "thought:", "plan:", "definetool:",
-                    "askuser:", "askuser\n", "planproposal:", "planproposal\n", "<")
+                    "askuser:", "askuser\n", "planproposal:", "planproposal\n", "<",
+                    "{", "[", "```json")
 
         def emit_text(text: str) -> None:
             nonlocal buffer
@@ -295,7 +296,7 @@ def callback(event: dict[str, Any]) -> None:
             elif kind == "model_complete":
                 emit_text(dsml.feed("", final=True))
                 if buffer:
-                    cleaned = agent._clean_final_response(buffer)
+                    cleaned = "" if agent.normalize_provider_control(buffer) else agent._clean_final_response(buffer)
                     if cleaned:
                         self.emit("stream_delta", request_id, text=_redact(cleaned))
                     buffer = ""