From 4704a43d3e1ed04a35c3a75eb78cda1f8ed1cd54 Mon Sep 17 00:00:00 2001 From: songzhenrui Date: Fri, 16 Oct 2026 16:54:00 +0800 Subject: [PATCH 1/5] llmcore: robustly strip leaked DSML tool-call markers (closing tags, variants, chunk splits) DeepSeek occasionally emits its DSML tool-call markup as plain content text instead of a real tool_calls delta (reported in #804, partially addressed by the closed PR #805 and by f7e4cc5 in ga.py). The previous content filter still missed three cases, so `` (and friends) reached the agent loop as assistant text, no tool call was registered, and the turn stalled. 1. The pattern required '<' to be immediately followed by the DSML token, so closing tags (' --- llmcore.py | 63 ++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 56 insertions(+), 7 deletions(-) diff --git a/llmcore.py b/llmcore.py index 3b0e6b060..1f445e81a 100644 --- a/llmcore.py +++ b/llmcore.py @@ -241,14 +241,54 @@ def _try_parse_tool_args(raw): return parsed return [{"_raw": raw}] +# DSML (DeepSeek tool-call markup) leak guard. +# Leaked shapes seen in the wild: '<|P|DSML|P|invoke ...>', '', +# '', '<|DSML|parameter>'. P = ASCII bar / fullwidth bar / double bar. +_DSML_WS = chr(32) + chr(9) +_DSML_PIPES = chr(124) + chr(0xFF5C) + chr(0x2016) +_DSML_TAG = ']{0,200}?>' +_DSML_PATTERN = re.compile(_DSML_TAG, re.IGNORECASE) +_DSML_LEAK = False # latched while parsing one response; reset at the start of each parse + +def _strip_dsml_markers(text: str) -> str: + # Remove DSML markers that leak into content text (used both on streamed chunks and on the + # final assembled text). Handles the closing-tag variants that previously slipped through, + # e.g. '' and '<|P|DSML|P|invoke name="x">'. + # Latches _DSML_LEAK so callers can detect the leak even when payload text remains. + global _DSML_LEAK + if not text or 'DSML' not in text: return text + out = _DSML_PATTERN.sub('', text) + if out != text: _DSML_LEAK = True + return out + +class _DsmlStreamFilter: + # Chunk-safe stripper: holds back a trailing '<...' without '>' so a marker split across + # SSE chunks is still removed (a plain per-chunk regex cannot see such a split tag). + _HOLD_MAX = 64 + def __init__(self): self.buf = '' + def feed(self, chunk): + self.buf += chunk + cut = self.buf.rfind('<') + hold = '' + if cut >= 0 and '>' not in self.buf[cut:] and len(self.buf) - cut <= self._HOLD_MAX: + hold = self.buf[cut:]; self.buf = self.buf[:cut] + out = _strip_dsml_markers(self.buf); self.buf = hold + return out + def flush(self): + out = _strip_dsml_markers(self.buf); self.buf = '' + return out + def _parse_openai_sse(resp_lines, api_mode="chat_completions"): """Parse OpenAI SSE stream (chat_completions or responses API). Yields text chunks, returns list[content_block]. content_block: {type:'text', text:str} | {type:'tool_use', id:str, name:str, input:dict} """ + global _DSML_LEAK + _DSML_LEAK = False # this parse latches leaks via _strip_dsml_markers content_text = "" if api_mode == "responses": seen_delta = False; fc_buf = {}; current_fc_idx = None; reasoning_text = "" + dsml_f = _DsmlStreamFilter() for line in resp_lines: if not line: continue line = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line @@ -260,10 +300,10 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): etype = evt.get("type", "") if etype == "response.output_text.delta": delta = evt.get("delta", "") - if delta: seen_delta = True; content_text += delta; yield delta + if delta: seen_delta = True; content_text += delta; yield dsml_f.feed(delta) elif etype == "response.output_text.done" and not seen_delta: text = evt.get("text", "") - if text: content_text += text; yield text + if text: content_text += text; yield dsml_f.feed(text) elif etype == "response.reasoning_text.delta": delta = evt.get("delta", "") if delta: reasoning_text += delta @@ -310,9 +350,11 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): _raise_if_retryable_overload(emsg) if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}" break + if (rest := dsml_f.flush()): + content_text += rest; yield rest blocks = [] if reasoning_text: blocks.append({"type": "thinking", "thinking": reasoning_text}) - if content_text: blocks.append({"type": "text", "text": content_text}) + if content_text: blocks.append({"type": "text", "text": _strip_dsml_markers(content_text)}) for idx in sorted(fc_buf): fc = fc_buf[idx] inps = _try_parse_tool_args(fc["args"]) @@ -324,6 +366,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): else: tc_buf = {} # index -> {id, name, args} reasoning_text = "" + dsml_f = _DsmlStreamFilter() for line in resp_lines: if not line: continue line = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line @@ -337,7 +380,7 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): if rc := delta.get("reasoning_content") or delta.get("reasoning", ""): reasoning_text += rc; yield rc if delta.get("content"): - text = delta["content"]; content_text += text; yield text + text = delta["content"]; content_text += text; yield dsml_f.feed(text) for tc in (delta.get("tool_calls") or []): idx = tc.get("index", 0) has_name = bool(tc.get("function", {}).get("name")) @@ -349,9 +392,11 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): if tc.get("id") and not tc_buf[idx]["id"]: tc_buf[idx]["id"] = tc["id"] usage = evt.get("usage") if usage: _record_usage(usage, api_mode) + if (rest := dsml_f.flush()): + content_text += rest; yield rest blocks = [] if reasoning_text: blocks.append({"type": "thinking", "thinking": reasoning_text}) - if content_text: blocks.append({"type": "text", "text": content_text}) + if content_text: blocks.append({"type": "text", "text": _strip_dsml_markers(content_text)}) for idx in sorted(tc_buf): tc = tc_buf[idx] inps = _try_parse_tool_args(tc["args"]) @@ -387,6 +432,8 @@ def _i(v, default=0): STATS.update(inp=inp, cached=cached, out=out) def _parse_openai_json(data, api_mode="chat_completions"): + global _DSML_LEAK + _DSML_LEAK = False blocks = [] if api_mode == "responses": _record_usage(data.get("usage") or {}, api_mode) @@ -394,7 +441,8 @@ def _parse_openai_json(data, api_mode="chat_completions"): if item.get("type") == "message": for p in (item.get("content") or []): if p.get("type") in ("output_text", "text") and p.get("text"): - blocks.append({"type": "text", "text": p["text"]}); yield p["text"] + _txt = _strip_dsml_markers(p["text"]) + if _txt: blocks.append({"type": "text", "text": _txt}); yield _txt elif item.get("type") == "reasoning": for p in (item.get("content") or []): if p.get("type") in ("reasoning_text", "summary_text") and p.get("text"): @@ -420,7 +468,7 @@ def _parse_openai_json(data, api_mode="chat_completions"): reasoning = msg.get("reasoning_content") or msg.get("reasoning", "") if reasoning: blocks.append({"type": "thinking", "thinking": reasoning}) - content = msg.get("content", "") + content = _strip_dsml_markers(msg.get("content", "")) if content: blocks.append({"type": "text", "text": content}); yield content for tc in (msg.get("tool_calls") or []): @@ -901,6 +949,7 @@ def __init__(self, thinking, content, tool_calls, raw, stop_reason='end_turn'): self.thinking = thinking; self.content = content self.tool_calls = tool_calls; self.raw = raw self.stop_reason = 'tool_use' if tool_calls else stop_reason + self.dsml_leak = _DSML_LEAK # latched by _strip_dsml_markers while parsing this response def __repr__(self): return f"" From 8192fff902f876896dc116d04a4d09333b746e58 Mon Sep 17 00:00:00 2001 From: songzhenrui Date: Fri, 16 Oct 2026 16:54:43 +0800 Subject: [PATCH 2/5] ga: make the no-tool DSML leak guard general (latch + marker variants) Complements f7e4cc5, which only retried when the content ended with the exact literal ``. That check misses * a leak anywhere except the very end of the turn, * markers that use `||` (double full-width) or full-width/heavy pipes, * a leaked closing tag of a non-`calls` marker such as `` - the marker that actually stalls the loop: the raw text is fed back into the next request and the model keeps regenerating broken DSML instead of calling a tool. `llmcore` now latches whether the DSML filter had to remove anything while parsing a response (`response.dsml_leak`), so `do_no_tool` can retry on a positive signal instead of guessing from the text tail. The cheap tail check is kept as a fallback for responses produced by other code paths. Refs #804 (closed), #805 (closed unmerged), #836 (open). Co-Authored-By: GenericAgent --- ga.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/ga.py b/ga.py index 68ff37a5c..a10c3131d 100644 --- a/ga.py +++ b/ga.py @@ -489,6 +489,10 @@ 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' Date: Fri, 16 Oct 2026 16:54:44 +0800 Subject: [PATCH 3/5] tests: cover DSML marker stripping and the no-tool leak guard Regression tests for #804 / #805 / #836: variant marker spelling, closing tags of non-`calls` markers, markers split across stream chunks (the exact `` case), the non-streaming JSON path, and the latch that `ga.do_no_tool` relies on. Co-Authored-By: GenericAgent --- frontends/tests/test_dsml_marker_strip.py | 147 ++++++++++++++++++++++ 1 file changed, 147 insertions(+) create mode 100644 frontends/tests/test_dsml_marker_strip.py diff --git a/frontends/tests/test_dsml_marker_strip.py b/frontends/tests/test_dsml_marker_strip.py new file mode 100644 index 000000000..69de6f973 --- /dev/null +++ b/frontends/tests/test_dsml_marker_strip.py @@ -0,0 +1,147 @@ +"""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('')`` retry in ``ga.py``. Three +cases still slipped through and stalled the agent loop: + +* closing tags (````) -- 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 ``<`` / `` 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: '") == "", tag + assert STRIP("") == "", 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") == "ls -la", tag + + +def test_strips_markers_around_and_inside_text(): + assert STRIP("before <" + TAG_SINGLE + " tool_calls>after") == "before after" + assert STRIP("ab") == "ab" + + +def test_plain_text_is_untouched_and_does_not_latch(): + _reset_latch() + for text in ("ok", "1 < 2 and 3 > 2", "x", ""): + 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("") + 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") + assert source.count("yield dsml_f.feed(") == 3, "responses delta/done + 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 " Date: Sat, 17 Oct 2026 15:55:22 +0800 Subject: [PATCH 4/5] llmcore/agent_loop/ga: execute tool calls the model wrote as text (#837) The model sometimes ends a turn with the call written as literal markup in the text channel ("" truncated, or a complete "" block). No tool_call comes back, so the loop treats the turn as a finished answer and the agent stops silently (session 423177, 2026-10-17 15:34). * llmcore.split_text_tool_calls(): recover trailing markup only -- quoted samples earlier in the reply are left alone -- returning (calls, partial) * agent_loop: execute recovered calls like real tool calls instead of no_tool * ga.do_no_tool: retry once when the marker is truncated mid-tag * responses stream: parse the response.completed output snapshot, handle output_item.done, and flag incomplete / terminal-event-less streams instead of letting a torn turn look like a finished answer Tests: frontends/tests/test_text_channel_tool_call.py (9 cases) Co-Authored-By: GenericAgent --- agent_loop.py | 17 ++- frontends/tests/test_dsml_marker_strip.py | 3 +- .../tests/test_text_channel_tool_call.py | 117 ++++++++++++++++ ga.py | 15 +++ llmcore.py | 127 +++++++++++++++++- 5 files changed, 271 insertions(+), 8 deletions(-) create mode 100644 frontends/tests/test_text_channel_tool_call.py diff --git a/agent_loop.py b/agent_loop.py index a744f0d26..acf20c1cb 100644 --- a/agent_loop.py +++ b/agent_loop.py @@ -66,7 +66,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 + # ("..." or a truncated ""). 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 = {t.get('name') for t in tools_schema if isinstance(t, dict)} - {None} + _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] diff --git a/frontends/tests/test_dsml_marker_strip.py b/frontends/tests/test_dsml_marker_strip.py index 69de6f973..5f5664bff 100644 --- a/frontends/tests/test_dsml_marker_strip.py +++ b/frontends/tests/test_dsml_marker_strip.py @@ -130,7 +130,8 @@ def test_stream_filter_flushes_plain_text_unchanged(): 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") - assert source.count("yield dsml_f.feed(") == 3, "responses delta/done + chat delta" + # 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" diff --git a/frontends/tests/test_text_channel_tool_call.py b/frontends/tests/test_text_channel_tool_call.py new file mode 100644 index 000000000..2d1bfb185 --- /dev/null +++ b/frontends/tests/test_text_channel_tool_call.py @@ -0,0 +1,117 @@ +"""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 -- "" (truncated) or a +complete "..." -- 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('x\n\n', TOOLS) + assert calls == [] and partial is True + + +def test_bare_opening_tag_is_partial(): + calls, partial = split('blah\n', TOOLS) + assert calls == [] and partial is True + + +def test_complete_parameter_form_is_executable(): + text = 'go\nprint(1)' + 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 = 'print("hi")' + 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("" + payload + "", TOOLS) + assert partial is False + assert calls == [("file_read", {"path": "a.txt"})] + + +def test_quoted_sample_mid_reply_is_ignored(): + text = '示例： 就是工具调用。\n然后用 file_read 读取。' + calls, partial = split(text, TOOLS) + assert calls == [] and partial is False + + +def test_text_after_closing_tag_is_ignored(): + text = 'x=1\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 diff --git a/ga.py b/ga.py index a10c3131d..b035df5b6 100644 --- a/ga.py +++ b/ga.py @@ -494,6 +494,21 @@ def do_no_tool(self, args, response): 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（完整 … 或截断的 ）， + # 本回合其实没有 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: + yield "[Warn] 检测到文本通道的工具调用，已自动执行。\n" + 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. ) 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: yield "[Warn] Plan模式完成声明拦截。\n" diff --git a/llmcore.py b/llmcore.py index 1f445e81a..7b4a5dd6b 100644 --- a/llmcore.py +++ b/llmcore.py @@ -278,6 +278,91 @@ def flush(self): out = _strip_dsml_markers(self.buf); self.buf = '' return out +# --- text-channel tool-call recovery ------------------------------------------- +# The model sometimes writes the call as literal markup in the assistant *text* +# channel instead of returning a real tool_call, e.g. "...", or the truncated "" that +# ends a turn. Nothing executes, no tool_calls come back, and the agent loop stops +# silently (the #804 / #837 family). These helpers recover that trailing markup. +_TEXT_CALL_TAG = re.compile(r'<\s*([A-Za-z_][\w\-]*)\s*([^<>]*?)/?\s*>', re.S) +_TEXT_PARAM = re.compile(r'<\s*parameter\s+name\s*=\s*["\']([^"\']+)["\']\s*>(.*?)<\s*/\s*parameter\s*>', re.S | re.I) +_TEXT_ATTR = re.compile(r'([A-Za-z_][\w\-]*)\s*=\s*("([^"]*)"|\'([^\']*)\')') + +def _text_tag_args(attrs): + out = {} + for m in _TEXT_ATTR.finditer(attrs or ''): + out[m.group(1)] = m.group(3) if m.group(3) is not None else m.group(4) + return out + +def _text_call_body(name, body): + body = (body or '').strip() + if not body: return None + if body.startswith('{'): + obj = None + for candidate in (body, body.replace('\\n', '\n').replace('\\"', '"')): + try: obj = json.loads(candidate); break + except Exception: continue + if isinstance(obj, dict): + if isinstance(obj.get('arguments'), dict): return obj['arguments'] + if isinstance(obj.get('input'), dict): return obj['input'] + return obj + return None + params = _TEXT_PARAM.findall(body) + if params: return {k: v.strip() for k, v in params} + return {'script': body} if name == 'code_run' else None + +def split_text_tool_calls(text, tool_names): + """Recover a tool call the model emitted as markup in the text channel. + + Only *trailing* markup is considered: a leaked call is the last thing a turn + produces, so code samples quoted earlier in the reply are left alone. + + Returns (calls, partial): + calls -- [(name, args), ...] parsed from the trailing markup (executable) + partial -- True when the turn ends with an unterminated opening tag, e.g. + ''; the model meant to call but the call never completed + """ + if not text or not tool_names: return [], False + tail = text.rstrip() + if not tail.endswith('>'): + m = re.search(r'<\s*([A-Za-z_][\w\-]*)\s*[^<>]*$', tail) + if m and (m.group(1) in tool_names or m.group(1) in ('tool_use', 'tool_call')): + return [], True + return [], False + last = None + for m in _TEXT_CALL_TAG.finditer(tail): + if m.group(1) in tool_names or m.group(1) in ('tool_use', 'tool_call'): last = m + if last is None: return [], False + name = last.group(1) + close = '' % name + end = tail.find(close, last.end()) + if name in ('tool_use', 'tool_call'): + if end < 0: return [], True + if tail[end + len(close):].strip(): return [], False + try: obj = json.loads(tail[last.end():end].strip()) + except Exception: return [], True + if not isinstance(obj, dict): return [], True + cname = obj.get('name') or obj.get('tool') + cargs = obj.get('arguments', obj.get('input', obj.get('args', {}))) + if isinstance(cargs, str): + try: cargs = json.loads(cargs) + except Exception: return [], True + if cname in tool_names and isinstance(cargs, dict): return [(cname, cargs)], False + return [], False + if end < 0: + # opening tag closed but body never came: '' / '' + body = tail[last.end():].strip() + if not body: return [], True + args = _text_call_body(name, body) + if args is None: return [], True + args.update(_text_tag_args(last.group(2))) + return [(name, args)], False + if tail[end + len(close):].strip(): return [], False # text follows: not a leaked call + args = _text_call_body(name, tail[last.end():end]) + if args is None: return [], True + args.update(_text_tag_args(last.group(2))) + return [(name, args)], False + def _parse_openai_sse(resp_lines, api_mode="chat_completions"): """Parse OpenAI SSE stream (chat_completions or responses API). Yields text chunks, returns list[content_block]. @@ -288,13 +373,14 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): content_text = "" if api_mode == "responses": seen_delta = False; fc_buf = {}; current_fc_idx = None; reasoning_text = "" + saw_terminal = False; saw_done = False dsml_f = _DsmlStreamFilter() for line in resp_lines: if not line: continue line = line.decode('utf-8', errors='replace') if isinstance(line, bytes) else line if not line.startswith("data:"): continue data_str = line[5:].lstrip() - if data_str == "[DONE]": break + if data_str == "[DONE]": saw_done = True; break try: evt = json.loads(data_str) except: continue etype = evt.get("type", "") @@ -316,6 +402,14 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): idx = evt.get("output_index", 0) fc_buf[idx] = {"id": item.get("call_id", item.get("id", "")), "name": item.get("name", ""), "args": ""} current_fc_idx = idx + elif etype == "response.output_item.done": + # Not every gateway emits output_item.added; take the final item snapshot too. + item = evt.get("item", {}) + if item.get("type") == "function_call": + idx = evt.get("output_index", len(fc_buf)) + fc_buf.setdefault(idx, {"id": item.get("call_id", item.get("id", "")), "name": item.get("name", ""), "args": ""}) + if item.get("arguments"): fc_buf[idx]["args"] = item["arguments"] + current_fc_idx = idx elif etype == "response.function_call_arguments.delta": idx = evt.get("output_index", current_fc_idx or 0) if idx in fc_buf: fc_buf[idx]["args"] += evt.get("delta", "") @@ -329,20 +423,36 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}" break elif etype == "response.completed": - usage = evt.get("response", {}).get("usage", {}) + saw_terminal = True + resp_obj = evt.get("response") or {} + usage = resp_obj.get("usage", {}) _record_usage(usage, api_mode) + # Some gateways omit delta events and only ship the final snapshot here; + # without this fallback both text and tool calls are silently dropped. + for item in resp_obj.get("output") or []: + itype = item.get("type") + if itype == "function_call": + idx = item.get("output_index", len(fc_buf)) + fc_buf.setdefault(idx, {"id": item.get("call_id", item.get("id", "")), "name": item.get("name", ""), "args": ""}) + if item.get("arguments"): fc_buf[idx]["args"] = item["arguments"] + elif itype == "message" and not seen_delta: + for c in item.get("content") or []: + if c.get("type") in ("output_text", "text") and c.get("text"): + content_text += c["text"]; yield dsml_f.feed(c["text"]) break elif etype == "response.incomplete": # DeepSeek/OpenAI responses stream may end here (no [DONE]); treat as valid terminal, - # record usage and surface a truncation marker instead of an empty-response retry storm. + # record usage and surface a truncation marker so a half-written turn is never silent + # (issue #837: a reply stopping mid-tool-call used to look like a normal answer). + saw_terminal = True usage = (evt.get("response") or {}).get("usage", {}) _record_usage(usage, api_mode) reason = ((evt.get("response") or {}).get("incomplete_details") or {}).get("reason", "") or "unknown" - if not content_text and not fc_buf: - marker = f"[!!! output truncated: {reason}]" - content_text += marker; yield marker + marker = f"[!!! output truncated: {reason}]" + content_text += marker; yield marker break elif etype == "response.failed": + saw_terminal = True usage = (evt.get("response") or {}).get("usage", {}) _record_usage(usage, api_mode) err = ((evt.get("response") or {}).get("error") or {}) @@ -350,6 +460,11 @@ def _parse_openai_sse(resp_lines, api_mode="chat_completions"): _raise_if_retryable_overload(emsg) if emsg: content_text += f"!!!Error: {emsg}"; yield f"!!!Error: {emsg}" break + if not saw_terminal and not saw_done and (content_text or fc_buf): + # Stream died without any terminal event (connection reset, proxy cutoff, server bug). + # Say so instead of letting a truncated turn look like a finished answer. + marker = "[!!! stream ended without a terminal event]" + content_text += marker; yield marker if (rest := dsml_f.flush()): content_text += rest; yield rest blocks = [] From bc7cf81e2098266c1e0159b5ae9140142b14f268 Mon Sep 17 00:00:00 2001 From: SONG ZHENRUI <191720247+ZhulongNT@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:35:29 +0800 Subject: [PATCH 5/5] agent_loop/ga: actually recover text-channel tool calls (schema name bug) The #837 fix was inert: agent_runner_loop built the known-tool set with {entry["name"] for entry in tools_schema}, but the shipped assets/tools_schema.json uses the OpenAI form {"type": "function", "function": {"name": ...}} -- every lookup returned None, the set came out empty, and split_text_tool_calls returns ([], False) immediately for an empty set. So a leaked block was never recovered and the raw markup reached the user; only the truncated-marker retry worked, because that path lives in do_no_tool. * agent_loop: add tool_names_from_schema(), accepting both the flat and the OpenAI nested form, and use it for the recovery call. * ga.py: do_no_tool's recovered-call branch used to yield "already executed" and return without dispatching anything. It now really dispatches each recovered call (it is only reached when the loop did not already handle it) and honours should_exit. * tests: schema name extraction (nested, flat, shipped file) and a leaked block whose body ends with a "python" language tag -- the exact 2026-10-17 shape. Co-Authored-By: GenericAgent --- agent_loop.py | 18 +++++++- .../tests/test_text_channel_tool_call.py | 44 +++++++++++++++++++ ga.py | 16 ++++++- 3 files changed, 75 insertions(+), 3 deletions(-) diff --git a/agent_loop.py b/agent_loop.py index acf20c1cb..7aa6e4000 100644 --- a/agent_loop.py +++ b/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: @@ -73,7 +89,7 @@ def agent_runner_loop(client, system_prompt, user_input, handler, tools_schema, # which asks for a retry. try: from llmcore import split_text_tool_calls - _names = {t.get('name') for t in tools_schema if isinstance(t, dict)} - {None} + _names = tool_names_from_schema(tools_schema) _calls, _partial = split_text_tool_calls(response.content or '', _names) except Exception: _calls, _partial = [], False diff --git a/frontends/tests/test_text_channel_tool_call.py b/frontends/tests/test_text_channel_tool_call.py index 2d1bfb185..2de7113f4 100644 --- a/frontends/tests/test_text_channel_tool_call.py +++ b/frontends/tests/test_text_channel_tool_call.py @@ -115,3 +115,47 @@ def test_incomplete_with_content_is_flagged(): ] 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": 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" + ' \n import os\n print(os.getcwd())\n\n python\n ' + ) + 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"] diff --git a/ga.py b/ga.py index b035df5b6..2ce92cfba 100644 --- a/ga.py +++ b/ga.py @@ -503,8 +503,20 @@ def do_no_tool(self, args, response): except Exception: _calls, _partial = [], False if _calls: - yield "[Warn] 检测到文本通道的工具调用，已自动执行。\n" - return StepOutcome(None, next_prompt="工具已执行，继续任务。") + # 真的执行，而不是只喊一句"已执行"（这里只有 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. ) instead of a real tool call. Regenerate this turn and call the tool normally.")