diff --git a/agent_loop.py b/agent_loop.py index a744f0d26..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: @@ -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 + # ("..." 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 = 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] diff --git a/frontends/tests/test_dsml_marker_strip.py b/frontends/tests/test_dsml_marker_strip.py new file mode 100644 index 000000000..5f5664bff --- /dev/null +++ b/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('')`` 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") + # 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 "" (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 + +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"] + +# --- variants seen on 2026-10-27: as the parameter container.""" + text = ('' + 'python') + calls, partial = split(text, TOOLS) + assert partial is False + assert calls == [("code_run", {"script": "print(1)", "type": "python"})] + + +def test_bare_script_tag_becomes_script_arg(): + text = '\n\n' + calls, partial = split(text, TOOLS) + assert partial is False + assert calls == [("code_run", {"script": "x = 1"})] + + +def test_stray_closing_tags_after_call_are_ignored(): + """'' / '' tails used to make the call look like prose.""" + text = ('x = 1\n' + '\n') + calls, partial = split(text, TOOLS) + assert partial is False + assert calls == [("code_run", {"script": "x = 1"})] + + +def test_truncated_mid_parameter_value_is_partial(): + """max_tokens cut the stream inside a value: '180'.""" + text = ('print(1)' + '180') + calls, partial = split(text, TOOLS) + assert calls == [] and partial is True + + +def test_fenced_quoted_sample_is_not_executed(): + text = ('示例：\n```\nx = 1\n' + '```\n以上就是格式。') + calls, partial = split(text, TOOLS) + assert calls == [] and partial is False + + +def test_strip_code_spans_keeps_trailing_fence(): + quoted = '写法是 `` 这样' + assert 'DSML' not in llmcore.strip_code_spans(quoted) + trailing = '正文\n```\nx\n```' + assert '' in llmcore.strip_code_spans(trailing) diff --git a/ga.py b/ga.py index 15fdb71bf..a48582759 100644 --- a/ga.py +++ b/ga.py @@ -482,7 +482,40 @@ 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.") + from llmcore import strip_code_spans + # 只认「裸写」的标记：回复里用反引号/围栏**引用**该标记（写文档、报 issue）不是泄漏 + _visible_tail = strip_code_spans(content[-4000:])[-200:] + if getattr(response, 'dsml_leak', False) or re.search(r'… 或截断的 ）， + # 本回合其实没有 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. ) instead of a real tool call. Regenerate this turn and call the tool normally.") # 2. 检测"包含较大代码块但未调用工具"的情况 # 关键特征：恰好1个大代码块 + 代码块直接结尾（后面只有空白） code_block_pattern = r"```[a-zA-Z0-9_]*\n[\s\S]{50,}?```" diff --git a/llmcore.py b/llmcore.py index 611e83556..342658f9c 100644 --- a/llmcore.py +++ b/llmcore.py @@ -270,29 +270,193 @@ 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 + +# --- 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*("([^"]*)"|\'([^\']*)\')') + +_TEXT_JUNK_TAIL = re.compile(r'^(?:\s*)*\s*$', re.S) +# 只剥「后面还有正文」的围栏块：结尾处的围栏里可能就是泄漏的调用本身 +_TEXT_FENCE_RE = re.compile(r'```.*?```(?=\s*\S)', re.S) + +def strip_code_spans(text): + """去掉引用用的代码块/行内代码。 + + 在文档、issue、commit message 里**引用**标记不是泄漏，只有裸写的标记才算。 + 结尾处的围栏块会被保留，因为那往往就是模型把调用包进 ``` 里发出的。 + """ + text = _TEXT_FENCE_RE.sub(' ', text or '') + return re.sub(r'`[^`\n]*`', ' ', text) + +def _quoted_markup(text, pos): + """pos 处的标记是否只是引用（围栏块或行内代码），而非模型真的在调用工具。""" + if pos <= 0: return False + line_start = text.rfind('\n', 0, pos) + 1 + if text[line_start:pos].count('`') % 2 == 1: return True # 行内代码 + if text.count('```', 0, pos) % 2 == 0: return False # 不在围栏里 + close = text.find('```', pos) + return close >= 0 and bool(text[close + 3:].strip()) # 围栏后面还有正文 + +def _text_tail_is_junk(rest): + """闭标签之后只剩 这类残留收尾标签。""" + return bool(_TEXT_JUNK_TAIL.match(rest or '')) + +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) + args = {k: v.strip() for k, v in params} if params else {} + if name == 'code_run': + # 变体：模型用 当参数容器，而不是 + m = re.search(r'<\s*script\b[^>]*>(.*?)<\s*/\s*script\s*>', body, re.S | re.I) + if m is not None and 'script' not in args: args['script'] = m.group(1).strip() + if not args: return {'script': body if m is None else m.group(1).strip()} + return args or 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 + # 截断在参数值中间（'180'）：开标签在，闭标签永远没来 + for m in _TEXT_CALL_TAG.finditer(tail): + if m.group(1) not in tool_names or _quoted_markup(tail, m.start()): continue + if ('' % m.group(1)) in tail[m.end():]: continue + if len(tail) - m.end() <= 1200 and '\n\n' not in tail[m.end():]: + 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 + if _quoted_markup(tail, last.start()): 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 not _text_tail_is_junk(tail[end + len(close):]): 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 not _text_tail_is_junk(tail[end + len(close):]): return [], False # 后面还有正文：不算调用 + 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]. 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 = "" + 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", "") 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 @@ -305,6 +469,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", "") @@ -318,20 +490,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 {}) @@ -339,9 +527,16 @@ 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 = [] 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"]) @@ -353,6 +548,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 @@ -366,7 +562,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")) @@ -378,9 +574,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"]) @@ -416,6 +614,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) @@ -423,7 +623,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"): @@ -449,7 +650,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 []): @@ -930,6 +1131,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""