llmcore+ga: strip leaked DSML markers robustly (closing tags, pipe variants, cross-chunk splits) + tests

agent_loop.py

@@ -28,6 +28,22 @@ def dispatch(self, tool_name, args, response, index=0, tool_num=1):
             yield f"未知工具: {tool_name}\n"
             return StepOutcome(None, next_prompt=f"未知工具 {tool_name}", should_exit=False)
 
+def tool_names_from_schema(tools_schema):
+    """Collect tool names from a schema list.
+
+    Accepts both the flat form ({"name": ...}) and the OpenAI form
+    ({"type": "function", "function": {"name": ...}}) -- the shipped
+    assets/tools_schema.json uses the latter, and reading only the top level
+    silently yields an empty set (which made text-channel recovery a no-op).
+    """
+    names = set()
+    for entry in tools_schema or []:
+        if not isinstance(entry, dict): continue
+        fn = entry.get("function")
+        name = (fn.get("name") if isinstance(fn, dict) else None) or entry.get("name")
+        if name: names.add(name)
+    return names
+
 def json_default(o): return list(o) if isinstance(o, set) else str(o)
 def exhaust(g):
     try: 
@@ -66,7 +82,22 @@ def agent_runner_loop(client, system_prompt, user_input, handler, tools_schema,
             if cleaned: yield cleaned + '\n'
         _hook('llm_after', locals())
 
-        if not response.tool_calls: tool_calls = [{'tool_name': 'no_tool', 'args': {}}]
+        if not response.tool_calls:
+            # #837: the model sometimes writes the call as literal markup in the text channel
+            # ("<code_run>...</code_run>" or a truncated "<code_run\">"). Recover and execute it
+            # instead of silently ending the turn; a torn marker still falls through to no_tool,
+            # which asks for a retry.
+            try:
+                from llmcore import split_text_tool_calls
+                _names = tool_names_from_schema(tools_schema)
+                _calls, _partial = split_text_tool_calls(response.content or '', _names)
+            except Exception:
+                _calls, _partial = [], False
+            if _calls:
+                if verbose: yield f"♻️ 文本通道工具调用已恢复执行: {', '.join(n for n, _ in _calls)}\n"
+                tool_calls = [{'tool_name': n, 'args': a, 'id': f'text_call_{turn}_{i}'} for i, (n, a) in enumerate(_calls)]
+            else:
+                tool_calls = [{'tool_name': 'no_tool', 'args': {}}]
         else: tool_calls = [{'tool_name': tc.function.name, 'args': json.loads(tc.function.arguments), 'id': tc.id}
                           for tc in response.tool_calls]
 

frontends/tests/test_dsml_marker_strip.py

@@ -0,0 +1,148 @@
+"""Contracts for DSML tool-call marker stripping (llmcore) and the leak guard (ga.py).
+
+Symptom: DeepSeek sometimes emits its DSML tool-call markup as plain assistant text
+(issue #804). The closed PR #805 added a per-chunk content filter, and upstream f7e4cc5
+added a narrow ``content.rstrip().endswith('</... calls>')`` retry in ``ga.py``. Three
+cases still slipped through and stalled the agent loop:
+
+* closing tags (``</... parameter>``) -- the pattern required ``<`` directly followed by
+  the DSML token, so a leading ``/`` never matched;
+* the double full-width-pipe variant (``|P||P|DSML|P||P|``) -- matching started at the
+  second pipe, leaving a stray ``<`` / ``</`` behind;
+* markers split across SSE chunks -- filtering each chunk independently cannot see them.
+
+These tests load the DSML helpers straight from ``llmcore.py`` (no import, no mykey, no
+side effects) and pin the integration points.
+"""
+from __future__ import annotations
+
+import ast
+import re
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+LLMCORE = ROOT / "llmcore.py"
+GA = ROOT / "ga.py"
+
+PIPE_ASCII = "|"
+PIPE_FULL = chr(0xFF5C)   # fullwidth vertical line
+PIPE_HEAVY = chr(0x2016)  # double vertical line
+
+TAG_DOUBLE = PIPE_FULL * 2 + "DSML" + PIPE_FULL * 2   # ||DSML||
+TAG_SINGLE = PIPE_FULL + "DSML" + PIPE_FULL           # |DSML|
+TAG_HEAVY = PIPE_HEAVY + "DSML" + PIPE_HEAVY          # ||DSML|| (heavy)
+TAG_ASCII = PIPE_ASCII + "DSML" + PIPE_ASCII          # |DSML|
+ALL_TAGS = (TAG_DOUBLE, TAG_SINGLE, TAG_HEAVY, TAG_ASCII)
+
+
+def _dsml_namespace() -> dict:
+    """Exec only the DSML constants/helpers from llmcore.py, without importing it."""
+    tree = ast.parse(LLMCORE.read_text(encoding="utf-8"))
+    wanted_assign = {"_DSML_WS", "_DSML_PIPES", "_DSML_TAG", "_DSML_PATTERN", "_DSML_LEAK"}
+    body = []
+    for node in tree.body:
+        if isinstance(node, ast.Assign) and any(
+            getattr(target, "id", None) in wanted_assign for target in node.targets
+        ):
+            body.append(node)
+        elif isinstance(node, ast.FunctionDef) and node.name == "_strip_dsml_markers":
+            body.append(node)
+        elif isinstance(node, ast.ClassDef) and node.name == "_DsmlStreamFilter":
+            body.append(node)
+    names = [type(node).__name__ + ":" + getattr(node, "name", "") for node in body]
+    assert len(body) == 7, f"unexpected DSML declarations in llmcore.py: {names}"
+    namespace: dict = {"re": re}
+    exec(compile(ast.Module(body=body, type_ignores=[]), "llmcore_dsml.py", "exec"), namespace)
+    return namespace
+
+
+NS = _dsml_namespace()
+STRIP = NS["_strip_dsml_markers"]
+STREAM_FILTER = NS["_DsmlStreamFilter"]
+
+
+def _reset_latch() -> None:
+    NS["_DSML_LEAK"] = False
+
+
+def test_strips_closing_tag_variants():
+    """Regression for issue #804: '</' + token must be removed, not left as '</'."""
+    for tag in ALL_TAGS:
+        assert STRIP("</" + tag + " parameter>") == "", tag
+        assert STRIP("</" + tag + " calls>") == "", tag
+
+
+def test_strips_invoke_and_parameter_variants():
+    for tag in ALL_TAGS:
+        assert STRIP("<" + tag + ' invoke name="ShellCommand" arguments="{}">') == "", tag
+        # markup is removed, the payload between the open/close markers is kept
+        assert STRIP("<" + tag + ' parameter name="cmd">ls -la</' + tag + " parameter>") == "ls -la", tag
+
+
+def test_strips_markers_around_and_inside_text():
+    assert STRIP("before <" + TAG_SINGLE + " tool_calls>after") == "before after"
+    assert STRIP("a</" + TAG_DOUBLE + " parameter>b") == "ab"
+
+
+def test_plain_text_is_untouched_and_does_not_latch():
+    _reset_latch()
+    for text in ("<summary>ok</summary>", "1 < 2 and 3 > 2", "<thinking>x</thinking>", ""):
+        assert STRIP(text) == text
+    assert NS["_DSML_LEAK"] is False
+
+
+def test_latch_is_set_when_a_marker_is_removed():
+    """The latch is what lets ga.do_no_tool retry even if payload text survives."""
+    _reset_latch()
+    assert NS["_DSML_LEAK"] is False
+    STRIP("</" + TAG_DOUBLE + " parameter>")
+    assert NS["_DSML_LEAK"] is True
+
+
+def test_stream_filter_removes_marker_split_across_chunks():
+    stream_filter = STREAM_FILTER()
+    marker = TAG_HEAVY + "tool_calls>"
+    chunks = ["hello<" + marker[:3], marker[3:] + "world"]
+    out = "".join(stream_filter.feed(chunk) for chunk in chunks) + stream_filter.flush()
+    assert "DSML" not in out
+    assert out == "helloworld"
+
+
+def test_stream_filter_removes_marker_split_right_after_open_bracket():
+    stream_filter = STREAM_FILTER()
+    out = "".join(
+        [
+            stream_filter.feed("hi <"),
+            stream_filter.feed(TAG_DOUBLE + " parameter>"),
+            stream_filter.flush(),
+        ]
+    )
+    assert out == "hi "
+
+
+def test_stream_filter_flushes_plain_text_unchanged():
+    stream_filter = STREAM_FILTER()
+    out = "".join([stream_filter.feed("plain "), stream_filter.flush()])
+    assert out == "plain "
+
+
+def test_all_streamed_content_chunks_pass_through_the_filter():
+    """Integration contract: every streamed content yield must be filtered."""
+    source = LLMCORE.read_text(encoding="utf-8")
+    # responses: delta, output_text.done, completed-snapshot fallback, incomplete/chat: delta
+    assert source.count("yield dsml_f.feed(") == 4, "responses delta/done/completed + chat delta"
+    assert source.count("dsml_f.flush()") == 2, "one flush per api_mode branch"
+    assert "_DSML_LEAK = False  # this parse" in source, "latch reset per parse"
+
+
+def test_non_streaming_paths_strip_too():
+    source = LLMCORE.read_text(encoding="utf-8")
+    assert 'content = _strip_dsml_markers(msg.get("content", ""))' in source
+    assert '_txt = _strip_dsml_markers(p["text"])' in source
+
+
+def test_ga_no_tool_guard_uses_latch_and_variant_agnostic_tail_check():
+    source = GA.read_text(encoding="utf-8")
+    assert "getattr(response, 'dsml_leak', False)" in source
+    assert "</?[" + PIPE_ASCII + PIPE_FULL + PIPE_HEAVY + "]{1,2}DSML" in source

frontends/tests/test_text_channel_tool_call.py

@@ -0,0 +1,161 @@
+"""Contracts for text-channel tool-call recovery (#837).
+
+Symptom (observed 2026-10-17, session 423177): DeepSeek ends a turn with the call
+written as literal markup in the *text* channel -- "<code_run\">" (truncated) or a
+complete "<code_run><parameter ...>...</parameter></code_run>" -- and returns no
+tool_call at all.  The loop then treats the turn as a finished answer: nothing runs
+and the agent just stops.
+
+These tests pin the recovery helpers in ``llmcore.py`` and the two stream fixes that
+made the failure invisible (``response.completed`` snapshot fallback, terminal-event
+tracking).
+"""
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+LLMCORE = ROOT / "llmcore.py"
+
+
+def _load_llmcore():
+    spec = importlib.util.spec_from_file_location("llmcore_under_test", LLMCORE)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+llmcore = _load_llmcore()
+split = llmcore.split_text_tool_calls
+TOOLS = {"code_run", "file_read", "file_write"}
+
+
+def test_truncated_marker_is_partial():
+    # exactly the shape that ended session 423177 at 15:34:17
+    calls, partial = split('<summary>x</summary>\n\n<code_run">', TOOLS)
+    assert calls == [] and partial is True
+
+
+def test_bare_opening_tag_is_partial():
+    calls, partial = split('blah\n<code_run>', TOOLS)
+    assert calls == [] and partial is True
+
+
+def test_complete_parameter_form_is_executable():
+    text = '<summary>go</summary>\n<code_run><parameter name="script">print(1)</parameter></code_run>'
+    calls, partial = split(text, TOOLS)
+    assert partial is False
+    assert calls == [("code_run", {"script": "print(1)"})]
+
+
+def test_complete_tag_with_attributes_and_raw_body():
+    text = '<code_run type="python">print("hi")</code_run>'
+    calls, partial = split(text, TOOLS)
+    assert partial is False
+    assert calls == [("code_run", {"script": 'print("hi")', "type": "python"})]
+
+
+def test_json_tool_use_form():
+    payload = json.dumps({"name": "file_read", "arguments": {"path": "a.txt"}})
+    calls, partial = split("<tool_use>" + payload + "</tool_use>", TOOLS)
+    assert partial is False
+    assert calls == [("file_read", {"path": "a.txt"})]
+
+
+def test_quoted_sample_mid_reply_is_ignored():
+    text = '示例：<code_run></code_run> 就是工具调用。\n然后用 file_read 读取。'
+    calls, partial = split(text, TOOLS)
+    assert calls == [] and partial is False
+
+
+def test_text_after_closing_tag_is_ignored():
+    text = '<code_run><parameter name="script">x=1</parameter></code_run>\n以上是脚本。'
+    calls, partial = split(text, TOOLS)
+    assert calls == [] and partial is False
+
+
+def _run_sse(lines):
+    gen = llmcore._parse_openai_sse(lines, "responses")
+    chunks = []
+    try:
+        while True:
+            chunks.append(next(gen))
+    except StopIteration as e:
+        return "".join(chunks), (e.value or [])
+
+
+def _sse(obj):
+    return "data: " + json.dumps(obj)
+
+
+def test_completed_snapshot_recovers_tool_call():
+    # no delta events at all; the call only exists in the terminal snapshot
+    lines = [
+        _sse({"type": "response.completed", "response": {"output": [
+            {"type": "function_call", "name": "code_run", "arguments": '{"script":"print(2)"}', "call_id": "c1"}]}}),
+    ]
+    _, blocks = _run_sse(lines)
+    tool = [b for b in blocks if b.get("type") == "tool_use"]
+    assert tool and tool[0]["name"] == "code_run"
+    assert tool[0]["input"] == {"script": "print(2)"}
+
+
+def test_stream_without_terminal_event_is_flagged():
+    lines = [_sse({"type": "response.output_text.delta", "delta": "partial answer"})]
+    text, blocks = _run_sse(lines)
+    assert "stream ended without a terminal event" in text
+
+
+def test_incomplete_with_content_is_flagged():
+    lines = [
+        _sse({"type": "response.output_text.delta", "delta": "half a sentence"}),
+        _sse({"type": "response.incomplete", "response": {"incomplete_details": {"reason": "max_output_tokens"}}}),
+    ]
+    text, _ = _run_sse(lines)
+    assert "output truncated: max_output_tokens" in text
+
+def _load_agent_loop():
+    spec = importlib.util.spec_from_file_location("agent_loop_under_test", ROOT / "agent_loop.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+agent_loop = _load_agent_loop()
+
+
+def test_schema_names_accept_the_openai_nested_form():
+    # Regression: assets/tools_schema.json ships {"type": "function", "function": {"name": ...}}.
+    # Reading only entry["name"] yielded an empty name set, and split_text_tool_calls
+    # bails out immediately on an empty set -- so recovery silently never ran, and the
+    # leaked markup reached the user (2026-10-17, after the first fix looked applied).
+    names = agent_loop.tool_names_from_schema([
+        {"type": "function", "function": {"name": "code_run"}},
+        {"name": "flat_form_tool"},
+        "junk",
+        {},
+    ])
+    assert names == {"code_run", "flat_form_tool"}
+
+
+def test_schema_names_of_the_shipped_schema():
+    schema = json.loads((ROOT / "assets" / "tools_schema.json").read_text(encoding="utf-8"))
+    names = agent_loop.tool_names_from_schema(schema)
+    assert "code_run" in names and "file_write" in names and len(names) >= 9
+
+
+def test_leaked_block_survives_a_language_tag_line_and_is_recovered():
+    # The real leak ended with "...\n python\n</code_run>": the language tag sits
+    # *inside* the block, so the tail still parses -- provided the name set is real.
+    text = (
+        "Run: list the files.\n\n"
+        ' <code_run>\n import os\n print(os.getcwd())\n\n python\n </code_run> '
+    )
+    schema = json.loads((ROOT / "assets" / "tools_schema.json").read_text(encoding="utf-8"))
+    names = agent_loop.tool_names_from_schema(schema)
+    calls, partial = split(text, names)
+    assert partial is False
+    assert len(calls) == 1 and calls[0][0] == "code_run"
+    assert "print(os.getcwd())" in calls[0][1]["script"]

ga.py

@@ -489,6 +489,37 @@ def do_no_tool(self, args, response):
             return self._retry_or_exit("[ERROR] Tool call format invalid; no tool was called. Regenerate using proper tool calls.")
         if 'max_tokens !!!]' in content[-100:]:
             return self._retry_or_exit("[ERROR] max_tokens limit reached. Use multi small steps to do it.")
+        if getattr(response, 'dsml_leak', False) or re.search(r'</?[|｜‖]{1,2}DSML', content[-200:]):
+            # 模型把工具调用写成了 DSML 标记文本(多种变体/跨 chunk 泄漏)，本回合其实没真正调用工具
+            yield "[Warn] DSML tool-call markup leaked into content, no real tool call. Retrying...\n"
+            return self._retry_or_exit("[ERROR] Your tool call leaked as raw DSML markup text instead of a real tool call. Regenerate this turn and call the tool normally.")
+
+        # 模型把工具调用写成了文本 XML（完整 <code_run>…</code_run> 或截断的 <code_run">），
+        # 本回合其实没有 tool_call（#837 复发）。完整的已由 agent_runner_loop 自动执行，
+        # 这里兜住其它调用路径；截断的必须重试，否则整轮直接"停"。
+        try:
+            from llmcore import split_text_tool_calls
+            _calls, _partial = split_text_tool_calls(content, {n[3:] for n in dir(self) if n.startswith('do_')})
+        except Exception:
+            _calls, _partial = [], False
+        if _calls:
+            # 真的执行，而不是只喊一句"已执行"（这里只有 agent_loop 没接住时才会走到）
+            yield "[Warn] 检测到文本通道的工具调用，转为真实调用执行。\n"
+            for _i, (_n, _a) in enumerate(_calls):
+                try:
+                    _g = self.dispatch(_n, _a, response, index=_i, tool_num=len(_calls))
+                    try:
+                        while True: next(_g)
+                    except StopIteration as _e:
+                        _o = getattr(_e, "value", None)
+                        if getattr(_o, "should_exit", False): return _o
+                except Exception as _err:
+                    yield f"[Warn] 文本通道调用 {_n} 执行失败: {_err}\n"
+                    return self._retry_or_exit("[ERROR] Text-channel tool call failed to execute. Regenerate using a real tool call.")
+            return StepOutcome(None, next_prompt="文本通道的工具调用已执行，请基于结果继续任务。")
+        if _partial:
+            yield "[Warn] 工具调用被写成未闭合的文本标记，重试一次。\n"
+            return self._retry_or_exit("[ERROR] Your tool call was emitted as truncated markup text (e.g. <code_run\">) instead of a real tool call. Regenerate this turn and call the tool normally.")
 
         if self._in_plan_mode() and any(kw in content for kw in ['任务完成', '全部完成', '已完成所有', '🏁']):
             if 'VERDICT' not in content and '[VERIFY]' not in content and '验证subagent' not in content: