diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py
index 545d545f..91be55cf 100644
--- a/backend/app/api/copilot.py
+++ b/backend/app/api/copilot.py
@@ -980,7 +980,8 @@ async def generate(
 
 _STREAM_PROTOCOL = """\
 输出格式（严格遵守）：
-- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字、不要空行注释
+- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字、不要空行注释，
+  也不要写成工具调用的格式（这里没有工具调用接口，look 也是一行 JSON）
 - 思考和输出共用同一份长度额度：思考里只定方案（加哪些节点、怎么连、每个节点干什么），
   不要在思考里把 JSON、提示词、表达式先写一遍——那些直接写进输出。额度用完时还没输出的改动就全丢了
 - 可用操作：
@@ -2304,7 +2305,7 @@ APP_SYSTEM_PROMPT = """\
 
 APP_PROTOCOL_PROMPT = """\
 输出格式（严格遵守）：
-- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字
+- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字，也不要写成工具调用的格式（这里没有工具调用接口）
 - 可用操作：
   {"op":"plan","summary":"一句话说明打算怎么改"}          ← 第一行
   {"op":"app_add","parent":"组件 id 或页面 id","index":0,"component":{"id":"…","type":"组件名","props":{},"bind":{}}}
diff --git a/backend/app/api/copilot_blueprint.py b/backend/app/api/copilot_blueprint.py
index af3a52f5..d4464616 100644
--- a/backend/app/api/copilot_blueprint.py
+++ b/backend/app/api/copilot_blueprint.py
@@ -71,7 +71,7 @@ BLUEPRINT_SYSTEM_PROMPT = """\
 
 BLUEPRINT_PROTOCOL_PROMPT = """\
 输出格式（严格遵守）：
-- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字
+- 每行一个完整的 JSON 对象，除此之外不输出任何东西——不要 markdown 围栏、不要解释文字，也不要写成工具调用的格式（这里没有工具调用接口）
 - 可用操作：
   {"op":"plan","summary":"一句话说明打算怎么整理"}          ← 第一行
   {"op":"bp_template","template":{"id":"periodic_board","version":1}}
diff --git a/backend/app/api/copilot_gen.py b/backend/app/api/copilot_gen.py
index dcc03d68..fbe0ce58 100644
--- a/backend/app/api/copilot_gen.py
+++ b/backend/app/api/copilot_gen.py
@@ -24,6 +24,7 @@ from __future__ import annotations
 
 import copy
 import json
+import logging
 import re
 import time
 from collections.abc import AsyncIterator
@@ -36,6 +37,8 @@ from app.look.base import LOOK_LIMIT, REPAIR_LOOK_LIMIT, Applied, Applier, Check
 from app.look.service import LookService
 from app.providers.budget import cut_off as _cut_off
 
+logger = logging.getLogger(__name__)
+
 #: 自查最多交回去改几轮。一轮只花一次"只输出改动"的调用；两轮还改不好，多半是需求本身有歧义，该让人看了
 SELF_CHECK_ROUNDS = 2
 #: 读到这几种操作就停下：done 说完了，look 要查看，ask 要等用户回答（蓝图助手的反问是 bp_ask，normal-user-apps N2）
@@ -106,7 +109,9 @@ class StreamDiagnosis:
 
     finish 是最后一次调用的结束原因（finish_reason / stop_reason）；text_chars、thinking_chars 是正文（不算空白）和思考的
     字数；dropped 是跳过的行按原因数；tool_calls 是原生工具调用的
-    名字（按操作读了的也记）；markup 是正文里写成文字的工具调用标记（<tool_call>、DeepSeek 的 DSML……）出现了几次调用。"""
+    名字（按操作读了的也记）；markup 是正文里写成文字的工具调用（<invoke>、<tool_call>、DeepSeek 的 DSML……）有几次——读得出
+    的按操作读（ops 里也算），读不出、只认出了标记的记 1；nudged：起草时一个操作都没认出、当场带纠正请它重写过一次
+    （Generation._converse）。"""
 
     calls: int = 0
     finish: str = ""
@@ -116,14 +121,40 @@ class StreamDiagnosis:
     tool_calls: list[str] = field(default_factory=list)
     dropped: dict[str, int] = field(default_factory=dict)
     markup: int = 0
+    nudged: bool = False
 
     def drop(self, reason: str) -> None:
         self.dropped[reason] = self.dropped.get(reason, 0) + 1
 
+    def absorb(self, other: StreamDiagnosis) -> None:
+        """把一次调用的诊断并进这一轮的（结束原因取最后一次的）。"""
+        self.calls += other.calls
+        self.finish = other.finish or self.finish
+        self.text_chars += other.text_chars
+        self.thinking_chars += other.thinking_chars
+        self.ops += other.ops
+        self.tool_calls.extend(other.tool_calls)
+        for reason, n in other.dropped.items():
+            self.dropped[reason] = self.dropped.get(reason, 0) + n
+        self.markup += other.markup
+
     def as_dict(self) -> dict[str, Any]:
         return {"calls": self.calls, "finish": self.finish or None, "text_chars": self.text_chars,
                 "thinking_chars": self.thinking_chars, "ops": self.ops, "tool_calls": list(self.tool_calls),
-                "dropped": dict(self.dropped), "markup": self.markup}
+                "dropped": dict(self.dropped), "markup": self.markup, "nudged": self.nudged}
+
+    def headline(self) -> str:
+        """报错的那一句（「助手本轮未完成：」后面）：按交回了什么照实说。以前一律是「模型没有给出任何修改，也没有说明原因」，
+        可模型多半是说了想干什么、只是格式不对（部署上撞过：把 look 写成了工具调用的格式）。"""
+        if self.tool_calls or self.markup:
+            return "模型把操作写成了工具调用的格式，没有按约定逐行输出，系统认不出"
+        if not self.text_chars and self.thinking_chars:
+            return "模型只写了思考，没有给出任何操作"
+        if not self.text_chars:
+            return "模型什么都没有输出"
+        if self.dropped:
+            return "模型的输出格式不对，没有一行能认出的操作"
+        return "模型只说了打算，没有给出任何修改"
 
     def summary(self) -> str:
         """给人看的一句：「结束原因 stop；正文 0 字，思考 11 字；认出 0 行操作；……」。"""
@@ -137,6 +168,8 @@ class StreamDiagnosis:
             parts.append("正文里有写成文字的工具调用标记")
         if self.calls > 1:
             parts.append(f"共调用模型 {self.calls} 次")
+        if self.nudged:
+            parts.append("当场带纠正重写过一次，仍然没有操作")
         return "；".join(parts)
 
     def correction(self) -> str:
@@ -184,29 +217,46 @@ async def _drain(stream: Any, already: int) -> None:
             return
 
 
+def _call_op(name: str, fields: Any, diag: StreamDiagnosis, *, what: str) -> dict[str, Any] | None:
+    """一次工具调用 → 操作：名字是操作名的，参数就是操作的字段。名字不认识、参数解析不了的照实记进诊断（what：「工具调用」
+    「工具调用标记」）。"""
+    if name not in OP_NAMES:
+        diag.drop(f"{what}的名字不认识")
+        return None
+    if not isinstance(fields, dict):
+        diag.drop(f"{what}的参数解析不了")
+        return None
+    return {"op": name, **{k: v for k, v in fields.items() if k != "op"}}
+
+
 def _tool_ops(calls: dict[int, dict[str, str]], diag: StreamDiagnosis) -> list[dict[str, Any]]:
-    """原生工具调用（流里的 tool_call_chunks，按 index 拼好的 {name, args}）→ 操作：名字是操作名的，参数就是操作的字段。
-    名字不认识、参数解析不了的照实记进诊断——以前整个工具调用一声不响就丢了。"""
+    """原生工具调用（流里的 tool_call_chunks，按 index 拼好的 {name, args}）→ 操作。以前整个工具调用一声不响就丢了。"""
     out: list[dict[str, Any]] = []
     for index in sorted(calls):
         name, args = calls[index].get("name") or "", calls[index].get("args") or ""
         diag.tool_calls.append(name or "（没有名字）")
-        if name not in OP_NAMES:
-            diag.drop("工具调用的名字不认识")
-            continue
         try:
             fields = json.loads(args) if args.strip() else {}
         except json.JSONDecodeError:
             fields = None
-        if not isinstance(fields, dict):
-            diag.drop("工具调用的参数解析不了")
-            continue
-        out.append({**{k: v for k, v in fields.items() if k != "op"}, "op": name})
+        if (op := _call_op(name, fields, diag, what="工具调用")) is not None:
+            out.append(op)
+    return out
+
+
+def _markup_ops(calls: list[Any], diag: StreamDiagnosis) -> list[dict[str, Any]]:
+    """写成文字的工具调用（engine/toolcalls.take_leaked_calls 读出来的）→ 操作，和原生工具调用同一个规矩。"""
+    out: list[dict[str, Any]] = []
+    for call in calls:
+        diag.markup += 1
+        if (op := _call_op(call.name, call.args, diag, what="工具调用标记")) is not None:
+            diag.ops += 1
+            out.append(op)
     return out
 
 
 async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None = None,
-                    diag: StreamDiagnosis | None = None):
+                    diag: StreamDiagnosis | None = None, raw: list[str] | None = None):
     """流式读模型输出，按行切出操作。done、look、ask 之后即停——后面就算有闲话也不要了；look 之后停下也就是
     「读到这一行就停止本次输出」。停下之后不马上关流：交出那一行、调用方接着要下一条时先把流读完（_drain，内容丢掉，最多
     DRAIN_CHARS 字），流末尾那一段用量才收得到——以前读到这一行就关流，输出 token 一个都拿不到，问数据这一轮的花费多半是下限
@@ -217,21 +267,25 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None
 
     格式跑偏的几种写法照样认（以前一声不响就丢了）：一个操作分成多行写的 JSON（以「{」开头的一行读不成，就接着往后攒，
     攒到读得成为止；中途来了一行自己就读得成的，攒着的那段算跳过）、和围栏写在同一行的操作、原生工具调用里名字是操作名
-    的（_tool_ops，流读完再按先后交出）。diag 给了的话，交回了什么（长度、个数、跳过的原因）记进去，不记原文。
+    的（_tool_ops，流读完再按先后交出）、写成文字的工具调用（DSML、<invoke>、<tool_call>……：从有标记的那一行起攒，一次调用
+    写完就按名字读成操作，见 engine/toolcalls.take_leaked_calls）。diag 给了的话，交回了什么（长度、个数、跳过的原因）记进去，
+    不记原文。raw 给了的话，流读完时把正文的开头一段（最多 4000 字）记进去：一个操作都没认出时当场交回模型纠正用。
 
     没等到 done 就因为额度用完停下的，抛 CopilotIncomplete：截断之前的操作照常转出去，由调用方决定整轮作废。
     """
     from app.engine.state import message_text as _text
     from app.engine.state import thinking_text
-    from app.engine.toolcalls import leaked_markup
+    from app.engine.toolcalls import has_markup, leaked_markup, take_leaked_calls
 
     diag = diag if diag is not None else StreamDiagnosis()
     diag.calls += 1
     buffer = ""
     cut = ""
     pending: list[str] = []
+    markup = ""                     # 写成文字的工具调用里还没写完的那一段
+    markup_before = diag.markup
     calls: dict[int, dict[str, str]] = {}
-    head = ""                       # 正文的开头一段：只用来认写成文字的工具调用标记，不进诊断
+    head = ""                       # 正文的开头一段：认写成文字的工具调用标记、交给 raw 用，不进诊断
 
     def read(line: str) -> dict[str, Any] | None:
         """一行 → 操作（认出来了）或 None；多行的 JSON 攒在 pending 里。"""
@@ -265,6 +319,16 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None
             diag.ops += 1
         return op
 
+    def take(line: str) -> list[dict[str, Any]]:
+        """一行 → 这一行读出来的操作。写成文字的工具调用从有标记的那一行起攒着，一次调用写完才交出（一行里可能写完好几个）；
+        攒着的时候来的行都归它，外层包装和调用之间的说明文字不算跳过。"""
+        nonlocal markup
+        if markup or (not pending and has_markup(line)):
+            leaked, markup = take_leaked_calls(markup + line + "\n")
+            return _markup_ops(leaked, diag)
+        op = read(line)
+        return [op] if op else []
+
     def record(op: dict[str, Any]) -> None:
         if lines is not None:
             lines.append(json.dumps(op, ensure_ascii=False))
@@ -290,17 +354,24 @@ async def _iter_ops(model: Any, messages: list[Any], *, lines: list[str] | None
             buffer += text
             while "\n" in buffer:
                 line, buffer = buffer.split("\n", 1)
-                op = read(line)
-                if op:
+                for op in take(line):
                     record(op)
                     yield op
                     if op.get("op") in _STOP_OPS:
                         await _drain(stream, len(buffer))
                         return
             cut = _cut_off(chunk) or cut
-    if leaked_markup(head):
-        diag.markup += 1
-    tail = [op for op in [read(buffer)] if op]
+    if raw is not None:
+        raw.append(head)
+    tail = take(buffer)
+    if markup:
+        # 流读完了，最后一次调用还差结尾：补上结尾再读一次，还读不出的算跳过
+        leaked, markup = take_leaked_calls(markup, final=True)
+        tail += _markup_ops(leaked, diag)
+        if markup:
+            diag.drop("工具调用标记没写完")
+    if diag.markup == markup_before and leaked_markup(head):
+        diag.markup += 1            # 认出了标记、一次调用都读不出来
     if pending:
         diag.drop("不是完整的 JSON")
         pending.clear()
@@ -863,6 +934,14 @@ class AppChecker:
 # 生成过程
 # ==========================================================================
 
+#: 起草时一次调用读完、到这时还没有一个能落地的操作（只写了思考、写成了工具调用的格式、夹着说明文字、只写了 plan、查看完
+#: 接着写却什么都没认出……）：同一轮里把它的原话交回去、附上这句纠正（correction 是 StreamDiagnosis.correction），请它按格式
+#: 重写一次。只补一次，还不行就照实报 copilot_empty。
+#: 部署上撞过：模型想先 look，却写成了 DSML，整轮当场报「没有给出任何修改」，用户点重试还是同样的几率
+FORMAT_NUDGE_PROMPT = "上面这次输出里没有一行能认出的操作：{correction}。请按输出格式重新给出这一轮的操作。"
+#: 交回去的原话最多多少字
+NUDGE_ECHO_CHARS = 2000
+
 
 class Generation:
     """一轮生成：起草（draft）→ 自查修正（repair）。两段之间调用方可以插自己的判断（只回答了一句、兜底重放……）。
@@ -889,13 +968,18 @@ class Generation:
         self.calls = 0
         #: 起草阶段模型交回了什么（只记长度、个数和原因）：一个操作都没认出来时写进错误详情
         self.diagnosis = StreamDiagnosis()
+        #: 起草阶段交出过能落地的操作（改动、回答、反问、收尾……；plan、look 不算：它们不改任何东西，查看完什么都没写的这一轮
+        #: 照样是空的）。一次调用读完还没有时当场纠正一次（FORMAT_NUDGE_PROMPT）
+        self.moved = False
 
-    def empty(self, message: str = "模型没有给出任何修改，也没有说明原因", *, hint: str = "") -> CopilotIncomplete:
-        """一个操作都没给（copilot_empty）的报错：带诊断（detail、diagnostics）和重来时交给助手的纠正（correction）。"""
+    def empty(self, message: str = "", *, hint: str = "") -> CopilotIncomplete:
+        """一个操作都没给（copilot_empty）的报错：说法按诊断照实写（StreamDiagnosis.headline），带诊断（detail、
+        diagnostics）和重来时交给助手的纠正（correction）。"""
         from app.chat.turn import EMPTY_GENERATION
 
-        return CopilotIncomplete(message, hint=hint, code=EMPTY_GENERATION, detail=self.diagnosis.summary(),
-                                 correction=self.diagnosis.correction(), diagnostics=self.diagnosis.as_dict())
+        return CopilotIncomplete(message or self.diagnosis.headline(), hint=hint, code=EMPTY_GENERATION,
+                                 detail=self.diagnosis.summary(), correction=self.diagnosis.correction(),
+                                 diagnostics=self.diagnosis.as_dict())
 
     async def draft(self, request: str) -> AsyncIterator[dict[str, Any]]:
         async for op in self._converse(request, LookBudget(self.look_limit), repair=False):
@@ -932,34 +1016,56 @@ class Generation:
         return int((time.monotonic() - self.started) * 1000)
 
     async def _converse(self, request: str, budget: LookBudget, *, repair: bool) -> AsyncIterator[dict[str, Any]]:
-        """一段对话：请模型写，读到 look 就停下、查看、把结果作为下一条消息交回，再请它接着写，直到 done / ask / 流结束。"""
+        """一段对话：请模型写，读到 look 就停下、查看、把结果作为下一条消息交回，再请它接着写，直到 done / ask / 流结束。
+
+        起草时一次调用读完、到这时还没有一个能落地的操作（格式跑偏，包括查看完接着写的那一次），先不收场：把它的原话交回去、
+        附一句纠正，请它当场重写一次（FORMAT_NUDGE_PROMPT，每轮一次）。以前直接报「没有给出任何修改」，画布上没有自动重来，
+        点重试也是同样的几率。"""
         messages: list[tuple[str, str]] = [("system", self.system), ("human", request)]
         idle = 0
         while True:
             lines: list[str] = []
+            raw: list[str] = []
             look: dict[str, Any] | None = None
             self.calls += 1
-            async with aclosing(_iter_ops(self.model, messages, lines=lines,
-                                          diag=None if repair else self.diagnosis)) as ops:
-                async for op in ops:
-                    kind = op.get("op")
-                    if kind == "thinking":
-                        yield op
-                    elif kind == "look":
-                        look = op
-                    elif kind in _ASK_OPS:
-                        self.asked = op
-                        yield op
-                    elif repair and kind in ("plan", "done"):
-                        # 修补轮自己的开场和收尾，不是新方案，不覆盖原来的说明
-                        continue
-                    else:
-                        if kind == "done":
-                            self.done = op
-                        if self.applier.apply(op).forward:
+            call = StreamDiagnosis()
+            try:
+                async with aclosing(_iter_ops(self.model, messages, lines=lines, diag=call, raw=raw)) as ops:
+                    async for op in ops:
+                        kind = op.get("op")
+                        if kind not in ("thinking", "plan", "look"):
+                            self.moved = True
+                        if kind == "thinking":
                             yield op
+                        elif kind == "look":
+                            look = op
+                        elif kind in _ASK_OPS:
+                            self.asked = op
+                            yield op
+                        elif repair and kind in ("plan", "done"):
+                            # 修补轮自己的开场和收尾，不是新方案，不覆盖原来的说明
+                            continue
+                        else:
+                            if kind == "done":
+                                self.done = op
+                            if self.applier.apply(op).forward:
+                                yield op
+            finally:
+                if not repair:
+                    self.diagnosis.absorb(call)
+            if call.markup and call.ops:
+                logger.info("助手把 %d 处操作写成了工具调用的格式，已按操作读出 %d 个", call.markup, call.ops)
             if look is None:
-                return
+                if repair or self.moved or self.diagnosis.nudged:
+                    return
+                # 还没有能落地的操作：当场纠正一次。说过话的把原话交回去；什么都没写的（只有思考）把纠正接在最后那条消息后面
+                logger.info("助手起草时一个操作都没认出（%s），当场纠正一次", call.summary())
+                self.diagnosis.nudged = True
+                nudge = FORMAT_NUDGE_PROMPT.format(correction=call.correction())
+                said = raw[0].strip() if raw else ""
+                messages = [*messages, ("ai", said[:NUDGE_ECHO_CHARS]), ("human", nudge)] if said \
+                    else [*messages[:-1], ("human", f"{messages[-1][1]}\n\n{nudge}")]
+                continue
             # 查看可能要等挑选模型：心跳的阶段换成「正在查看」
             yield {"op": "heartbeat", "phase": "looking", "elapsed_ms": self._elapsed()}
             result = await self.looks.look(look, budget)
diff --git a/backend/app/chat/job.py b/backend/app/chat/job.py
index 140ae93f..6bfdfae9 100644
--- a/backend/app/chat/job.py
+++ b/backend/app/chat/job.py
@@ -755,7 +755,8 @@ async def _settled(driver: ChatDriver, execution: ChatTurnExecution
             error = message
         elif term.get("op") == "error":
             error = str(term.get("message") or "助手本轮未完成")
-            cause = {"kind": "copilot", "message": error, "hint": term.get("hint"), "detail": term.get("detail")}
+            cause = {"kind": "copilot", "message": error, "hint": term.get("hint"), "detail": term.get("detail"),
+                     "code": term.get("code"), "diagnostics": term.get("diagnostics")}
         else:
             issues = [str(i.get("message") or "") for i in term.get("issues") or []
                       if isinstance(i, Mapping) and i.get("level") == "error"]
diff --git a/backend/app/chat/record.py b/backend/app/chat/record.py
index 07254265..15840751 100644
--- a/backend/app/chat/record.py
+++ b/backend/app/chat/record.py
@@ -14,7 +14,8 @@
 **失败只记原因（cause），不记说法。** 页面以前记的是翻成人话之后的 failure（标题、原因、怎么办：lib/explain 那一套），说法在
 界面层；服务端只记发生了什么，页面恢复时照同一套 lib/explain 讲（frontend/src/store/chat.ts 的 failureFromCause）：
   {kind: refused, message, code}        开流之前就被拒（模型没配好、限定的数据源都不在）
-  {kind: copilot, message, hint, detail} 生成过程交出 error（助手本轮未完成）
+  {kind: copilot, message, hint, detail, code?, diagnostics?} 生成过程交出 error（助手本轮未完成；一个操作都没给的带
+                                        code=copilot_empty 和诊断 copilot_gen.StreamDiagnosis.as_dict，画布轮次同样记）
   {kind: blocked, issues, self_checked} 生成的图没过检查
   {kind: start, message, code, status}  发起运行被拒（绑定的工具不在、使用者用不了的节点）
   {kind: run, error, detail}            运行失败
diff --git a/backend/app/engine/toolcalls.py b/backend/app/engine/toolcalls.py
index 5cb6b26c..0ccd5fbc 100644
--- a/backend/app/engine/toolcalls.py
+++ b/backend/app/engine/toolcalls.py
@@ -8,6 +8,7 @@ wait_for，而 wait_for 取消之后要等被取消的那一方收完尾：驱
 from __future__ import annotations
 
 import asyncio
+import json
 import re
 from dataclasses import dataclass
 from typing import Any, Awaitable
@@ -135,9 +136,9 @@ def _abandon(job: asyncio.Future[Any], *, cancel: bool = True) -> None:
 # --------------------------------------------------------------------------
 
 _MARKUP = re.compile(
-    r"<[｜|]{1,2}\s*DSML\s*[｜|]{1,2}"                      # DeepSeek 的 DSML
+    r"</?\s*[｜|]{1,2}\s*DSML\s*[｜|]{1,2}"                  # DeepSeek 的 DSML
     r"|<[｜|]\s*tool[▁_ ]calls?[▁_ ]begin\s*[｜|]>"          # DeepSeek 的特殊 token
-    r"|<\s*(?:tool_call|function_call)\s*>"
+    r"|<\s*(?:tool_calls?|function_calls?)\s*>"
     r"|<\s*invoke\s+name\s*=",
     re.IGNORECASE,
 )
@@ -173,3 +174,146 @@ def leaked_markup(text: str | None) -> str | None:
 
 def markup_warning(snippet: str, who: str = "模型") -> str:
     return f"{who}以文本形式输出了工具调用（{snippet[:40]}…），未实际调用工具"
+
+
+# --------------------------------------------------------------------------
+# 把写成文字的工具调用读出来
+#
+# 助手（app/api/copilot_gen）不走工具调用接口：操作（look、add_node……）是正文里一行一行的 JSON。可模型偶尔还是按自己训练时
+# 那套调用格式写（部署上撞过：DeepSeek 想先 look 看表，思考里写着「先输出 look」，正文却是一段 DSML），意图和参数都对，
+# 只是格式不对。助手按名字把它们读成操作；节点里的 agent 仍然只认、不读（真要调工具得走接口，见上面的纠正）。
+#
+# 认得的几种写法：
+#   DSML（DeepSeek）       <｜DSML｜invoke name="look"><｜DSML｜parameter name="need" string="true">…</｜DSML｜parameter>
+#                          </｜DSML｜invoke>，外面可能包一层 function_calls / calls；竖线一到两个、中间可能有空格
+#   <invoke>               同上去掉 DSML 前缀（<function_calls><invoke name="…"><parameter name="…">…）
+#   <tool_call>            <tool_call>{"name": "look", "arguments": {…}}</tool_call>（function_call 同）
+#   DeepSeek 特殊 token    <｜tool▁call▁begin｜>function<｜tool▁sep｜>look ```json {…}```<｜tool▁call▁end｜>，
+#                          或者 <｜tool▁call▁begin｜>look<｜tool▁sep｜>{…}<｜tool▁call▁end｜>
+# --------------------------------------------------------------------------
+
+_DSML_PREFIX = re.compile(r"<(/?)\s*[｜|]{1,2}\s*DSML\s*[｜|]{1,2}\s*", re.IGNORECASE)
+_INVOKE = re.compile(r"<invoke\s+name\s*=\s*[\"']([^\"']*)[\"']\s*>(.*?)</invoke\s*>", re.DOTALL | re.IGNORECASE)
+_PARAMETER = re.compile(r"<parameter\s+name\s*=\s*[\"']([^\"']*)[\"']([^>]*)>(.*?)</parameter\s*>",
+                        re.DOTALL | re.IGNORECASE)
+_PARAMETER_OPEN = re.compile(r"<parameter\b", re.IGNORECASE)
+_PARAMETER_CLOSE = re.compile(r"</parameter\s*>", re.IGNORECASE)
+_STRING_ATTR = re.compile(r"string\s*=\s*[\"'](true|false)[\"']", re.IGNORECASE)
+_JSON_CALL = re.compile(r"<(tool_call|function_call)\s*>(.*?)</\1\s*>", re.DOTALL | re.IGNORECASE)
+_TOKEN_CALL = re.compile(r"<[｜|]\s*tool[▁_ ]call[▁_ ]begin\s*[｜|]>(.*?)<[｜|]\s*tool[▁_ ]sep\s*[｜|]>(.*?)"
+                         r"<[｜|]\s*tool[▁_ ]call[▁_ ]end\s*[｜|]>", re.DOTALL | re.IGNORECASE)
+#: 一次调用的开头：在它之后还没等到结尾的，留着等后面的行
+_CALL_OPENER = re.compile(r"<invoke\s+name\s*=|<(?:tool_call|function_call)\s*>"
+                          r"|<[｜|]\s*tool[▁_ ]call[▁_ ]begin\s*[｜|]>", re.IGNORECASE)
+_FENCE = re.compile(r"^`{3,}[A-Za-z0-9_-]*\s*|\s*`{3,}$")
+#: 参数值像 JSON 的（没写 string 属性时按 JSON 读）
+_JSONISH = re.compile(r"^\s*(?:[\[{\"]|-?\d|true\b|false\b|null\b)")
+
+
+@dataclass(frozen=True)
+class LeakedCall:
+    """写成文字的一次工具调用。args 是 None：参数读不出来（名字照样交出去，由调用方记进诊断）。"""
+
+    name: str
+    args: dict[str, Any] | None
+
+
+def _parameter_value(raw: str, attrs: str) -> Any:
+    """DSML 的 string="true" 是原样的字符串，string="false" 是 JSON；没写的看着像 JSON 就按 JSON 读，读不成照原样。"""
+    value = raw.strip()
+    flag = _STRING_ATTR.search(attrs)
+    if flag and flag.group(1).lower() == "true":
+        return value
+    if flag or _JSONISH.match(value):
+        try:
+            return json.loads(value)
+        except ValueError:
+            return value
+    return value
+
+
+def _invoke_call(name: str, body: str) -> LeakedCall:
+    return LeakedCall(name.strip(), {key.strip(): _parameter_value(raw, attrs)
+                                     for key, attrs, raw in _PARAMETER.findall(body)})
+
+
+def _json_call(body: str) -> LeakedCall:
+    """<tool_call>{"name": …, "arguments": {…}}</tool_call>：arguments 也可能是一段 JSON 字符串。"""
+    try:
+        obj = json.loads(body.strip())
+    except ValueError:
+        return LeakedCall("", None)
+    if not isinstance(obj, dict):
+        return LeakedCall("", None)
+    function = obj.get("function") if isinstance(obj.get("function"), dict) else {}
+    name = str(obj.get("name") or function.get("name") or "")
+    args = next((v for v in (obj.get("arguments"), obj.get("parameters"), obj.get("args"),
+                             function.get("arguments")) if v is not None), {})
+    if isinstance(args, str):
+        try:
+            args = json.loads(args) if args.strip() else {}
+        except ValueError:
+            args = None
+    return LeakedCall(name, args if isinstance(args, dict) else None)
+
+
+def _token_call(head: str, body: str) -> LeakedCall:
+    """DeepSeek 特殊 token 的两种排法：sep 前是 function、名字在 sep 后第一行；或者 sep 前就是名字。参数可能套着围栏。"""
+    head, body = head.strip(), body.strip()
+    if head.lower() in ("", "function"):
+        name, _, body = body.partition("\n")
+    else:
+        name = head
+    body = _FENCE.sub("", body.strip()).strip()
+    try:
+        args = json.loads(body) if body else {}
+    except ValueError:
+        args = None
+    return LeakedCall(name.strip(), args if isinstance(args, dict) else None)
+
+
+#: 流读完了、最后一次调用还差结尾（模型写完参数就停了）时补上的结尾，按开头的写法
+_CLOSERS = (("<invoke", "</invoke>"), ("<tool_call", "</tool_call>"), ("<function_call", "</function_call>"),
+            ("<", "<｜tool▁call▁end｜>"))
+
+
+def take_leaked_calls(text: str, *, final: bool = False) -> tuple[list[LeakedCall], str]:
+    """从一段写成文字的工具调用里取出已经写完的调用（按出现的先后），交回 (调用, 还没写完的那一段)。还没写完的从最后一个
+    没等到结尾的调用开头算起；外层的包装（function_calls、DSML 的 calls）、调用之间的说明文字不留——没有没写完的调用时交回
+    空串。流式读的时候一行一行攒进来，每攒一行取一次。
+
+    final：后面不会再来了。最后一次调用只差结尾的（参数都写完整了，模型写完就停了）补上结尾再读一次；参数写到一半的、
+    补上也读不出名字和参数的，照样交回——拿半截参数当一次调用，比照实说没写完更糟。"""
+    calls, rest = _take(text)
+    if final and rest and len(_PARAMETER_OPEN.findall(rest)) == len(_PARAMETER_CLOSE.findall(rest)):
+        closer = next(c for prefix, c in _CLOSERS if rest.lower().startswith(prefix))
+        more, left = _take(rest + closer)
+        if more and not left and all(c.name and c.args is not None for c in more):
+            return calls + more, ""
+    return calls, rest
+
+
+def _take(text: str) -> tuple[list[LeakedCall], str]:
+    text = _DSML_PREFIX.sub(r"<\1", text)
+    found: list[tuple[int, int, LeakedCall]] = []
+    for m in _INVOKE.finditer(text):
+        found.append((m.start(), m.end(), _invoke_call(m.group(1), m.group(2))))
+    for m in _JSON_CALL.finditer(text):
+        found.append((m.start(), m.end(), _json_call(m.group(2))))
+    for m in _TOKEN_CALL.finditer(text):
+        found.append((m.start(), m.end(), _token_call(m.group(1), m.group(2))))
+    found.sort(key=lambda f: f[0])
+    calls: list[LeakedCall] = []
+    end = 0
+    for start, stop, call in found:
+        if start < end:  # 套在别的调用里面的不另算
+            continue
+        calls.append(call)
+        end = stop
+    opener = _CALL_OPENER.search(text, end)
+    return calls, text[opener.start():] if opener else ""
+
+
+def has_markup(text: str) -> bool:
+    """这一段里有没有工具调用标记（只认标记本身，不认 leaked_markup 那种「假设调用了」的说法）：流式读的时候据此开始攒。"""
+    return bool(_MARKUP.search(text))
diff --git a/backend/app/look/service.py b/backend/app/look/service.py
index 933f3de6..8ece73fd 100644
--- a/backend/app/look/service.py
+++ b/backend/app/look/service.py
@@ -46,6 +46,8 @@ GUIDE_HEAD_PROMPT = (
     "按需查看（look）：要看{labels}的详情时，单独输出一行\n"
     '  {{"op":"look","need":"一句话说要查什么","targets":[{example}]}}\n'
     "  输出这一行就停下，系统把查看结果作为下一条消息交回来，你接着往下写（已经输出的不要重复）。\n"
+    "  · 这里没有工具调用接口：look 和别的操作一样写成正文里单独一行的 JSON，不要写成工具调用的格式"
+    "（<invoke>、<tool_call>、DSML 这类标记）——那样系统认不出\n"
     "  · need 必填。targets 里写了 name 的直接取那几个；只写 kind（可以加 source）不写 name 的，按 need 在这一类里挑；"
     "整个不写 targets 时按 need 挑{default}。sources 可选：限定在哪几个来源（数据源名、工作流名……）里挑\n"
     "  · kind 只能是：{kinds}\n"
diff --git a/backend/tests/fixtures/copilot_answer_ops_baseline.json b/backend/tests/fixtures/copilot_answer_ops_baseline.json
index f1de6562..f76c4b70 100644
--- a/backend/tests/fixtures/copilot_answer_ops_baseline.json
+++ b/backend/tests/fixtures/copilot_answer_ops_baseline.json
@@ -1006,7 +1006,7 @@
   },
   {
    "op": "error",
-   "message": "助手本轮未完成：模型没有给出任何修改，也没有说明原因",
+   "message": "助手本轮未完成：模型的输出格式不对，没有一行能认出的操作",
    "hint": "画布未修改。可以重试，或把需求描述得更具体",
    "detail": ""
   }
diff --git a/backend/tests/test_chat_turn_policy.py b/backend/tests/test_chat_turn_policy.py
index 3fceb663..c886f818 100644
--- a/backend/tests/test_chat_turn_policy.py
+++ b/backend/tests/test_chat_turn_policy.py
@@ -196,13 +196,16 @@ async def test_answer_ops_says_to_retry_an_empty_generation(client, monkeypatch)
     assert err["code"] == turn.EMPTY_GENERATION and err["next"] == "retry", err
     # 原因是重来时交给助手的那句纠正（这次连思考都没有），诊断写在 detail（不含原文）
     assert err["reason"].startswith("什么都没有输出") and "look" in err["reason"], err
-    assert err["message"] == "助手本轮未完成：模型没有给出任何修改，也没有说明原因"
+    # 说法按诊断照实写；起草时已经当场纠正过一次（第二次还是空的），整轮才交 error
+    assert err["message"] == "助手本轮未完成：模型什么都没有输出"
     assert "正文 0 字，思考 0 字" in err["detail"] and err["diagnostics"]["ops"] == 0
+    assert err["diagnostics"]["nudged"] is True and model.calls == 2, err
 
 
 async def test_the_eval_driver_retries_an_empty_generation_once(monkeypatch):
     """评测的作答（app/eval/answer.Driver）走同一个决定：第一次一个操作都没给就带着原因重来一次，原因照实记在 retries；
-    重来还不行才照实报建图失败（助手调用失败），不再试第三次。"""
+    重来还不行才照实报建图失败（助手调用失败），不再试第三次。每一次生成里起草时已经当场纠正过一次（copilot_gen
+    FORMAT_NUDGE_PROMPT），所以一次生成算两次调用。"""
     import json
 
     import app.api.copilot as copilot
@@ -210,7 +213,7 @@ async def test_the_eval_driver_retries_an_empty_generation_once(monkeypatch):
     from tests.test_usage_meter import Scripted
 
     reply = json.dumps({"op": "reply", "text": "上周没有华中区的数据"}, ensure_ascii=False)
-    for replies, path in (([[""], [reply]], "reply"), ([[""]], "build_failed")):
+    for replies, path, calls in (([[""], [""], [reply]], "reply", 3), ([[""]], "build_failed", 4)):
         model = Scripted(model="claude-opus-5", replies=replies)
         seen: list[str] = []
 
@@ -228,14 +231,14 @@ async def test_the_eval_driver_retries_an_empty_generation_once(monkeypatch):
         driver = answer.Driver("上周华中区销售额", datasource_ids=[], ctx=answer.TurnContext(
             suite_id="s", suite_name="虚构题集", eval_run_id="e", case_id=None, case_seq=1, actor=None))
         t = await driver.drive()
-        assert t.path == path and t.attempts == 1 and model.calls == 2, (t.path, t.attempts, model.calls, t.reason)
+        assert t.path == path and t.attempts == 1 and model.calls == calls, (t.path, t.attempts, model.calls, t.reason)
         assert len(t.retries) == 1 and t.retries[0].startswith("什么都没有输出"), t.retries
         # 重来时问题后面附那句纠正，不原样重发
         assert seen[1] == turn.empty_retry_instruction("上周华中区销售额", t.retries[0]) != seen[0], seen
         assert "什么都没有输出" in seen[1]
         if path == "build_failed":
             assert t.copilot_failed and t.reason.startswith("助手本轮未完成"), t.reason
-        assert t.usage["assistant"]["calls"] == 2, "两次生成都是这一题花的"
+        assert t.usage["assistant"]["calls"] == calls, "两次生成（连同各自当场纠正的那一次）都是这一题花的"
 
 
 async def test_review_says_whether_and_how_to_retry(client, monkeypatch):
diff --git a/backend/tests/test_chat_turns.py b/backend/tests/test_chat_turns.py
index 21c22209..e41c6ee4 100644
--- a/backend/tests/test_chat_turns.py
+++ b/backend/tests/test_chat_turns.py
@@ -850,12 +850,13 @@ def test_attempt_records_say_what_the_page_used_to_say():
 
 
 async def test_an_empty_generation_is_retried_once_on_the_server(client, monkeypatch):
-    """助手只写了思考、一个操作都没给（error.next=retry）：服务端带着原因重来一次（turn.retry kind=empty），不用页面在。"""
+    """助手只写了思考、一个操作都没给（error.next=retry）：服务端带着原因重来一次（turn.retry kind=empty），不用页面在。
+    一次建图里起草时已经当场纠正过一次（copilot_gen FORMAT_NUDGE_PROMPT），那一次也空着，整次建图才交 error。"""
     import app.api.copilot as copilot
     from tests.test_usage_meter import Scripted
 
     reply = json.dumps({"op": "reply", "text": "上周没有华中区的数据"}, ensure_ascii=False)
-    model = Scripted(model="claude-opus-5", replies=[[""], [reply]])
+    model = Scripted(model="claude-opus-5", replies=[[""], [""], [reply]])
 
     async def _model(*_a, **_k):
         return model, "claude-opus-5"
@@ -871,7 +872,7 @@ async def test_an_empty_generation_is_retried_once_on_the_server(client, monkeyp
     assert "正文 0 字" in retry["detail"], retry
     row = await _row(turn_id)
     assert row.status == "done" and row.answer == "上周没有华中区的数据" and row.meta["noQuery"] is True
-    assert row.meta["usage"]["copilot"]["calls"] == 2, "两次建图都是这一轮花的"
+    assert row.meta["usage"]["copilot"]["calls"] == 3, "两次建图（连同第一次当场纠正的那一次）都是这一轮花的"
 
 
 async def test_several_stranded_turns_of_one_conversation_can_be_resumed_together(client):
diff --git a/backend/tests/test_copilot_answer_ops.py b/backend/tests/test_copilot_answer_ops.py
index 9a3203d1..2246369a 100644
--- a/backend/tests/test_copilot_answer_ops.py
+++ b/backend/tests/test_copilot_answer_ops.py
@@ -154,7 +154,10 @@ async def test_the_sse_output_is_unchanged_but_for_a_trailing_usage(client, monk
         reason = op.pop("reason")
         if name == "no_ops":
             # 一个操作都没给：原因是重来时交给助手的那句纠正，诊断（不含原文）写进 detail、diagnostics，别的照旧
-            assert "每个操作单独一行" in reason and op.pop("diagnostics")["dropped"] == {"不是 JSON 对象": 1}, op
+            # 起草时当场带纠正重写过一次（同一个剧本再吐一遍），跳过的行是两次的
+            diagnostics = op.pop("diagnostics")
+            assert "每个操作单独一行" in reason and diagnostics["dropped"] == {"不是 JSON 对象": 2}, op
+            assert diagnostics["calls"] == 2 and diagnostics["nudged"] is True, diagnostics
             assert op["detail"].startswith("结束原因"), op
             op["detail"] = ""
         else:
diff --git a/backend/tests/test_copilot_empty_diagnosis.py b/backend/tests/test_copilot_empty_diagnosis.py
index 6d054135..4c11b007 100644
--- a/backend/tests/test_copilot_empty_diagnosis.py
+++ b/backend/tests/test_copilot_empty_diagnosis.py
@@ -93,10 +93,13 @@ async def test_a_native_tool_call_named_after_an_op_becomes_that_op():
     assert diag.dropped == {"工具调用的名字不认识": 1, "工具调用的参数解析不了": 1}, diag.dropped
 
 
-async def test_tool_call_markup_in_text_is_diagnosed():
-    ops, diag = await _ops([AIMessageChunk(content='<tool_call>{"name": "look", "arguments": {}}</tool_call>\n')])
+async def test_tool_call_markup_that_is_not_an_op_is_diagnosed():
+    # 名字是操作名的照操作读（tests/test_copilot_markup_ops.py）；不是的照实记进诊断
+    ops, diag = await _ops([AIMessageChunk(content='<tool_call>{"name": "run_sql", "arguments": {}}</tool_call>\n')])
     assert ops == [] and diag.markup == 1
+    assert diag.dropped == {"工具调用标记的名字不认识": 1}, diag.dropped
     assert "不走工具调用" in diag.correction()
+    assert diag.headline() == "模型把操作写成了工具调用的格式，没有按约定逐行输出，系统认不出"
 
 
 async def test_the_empty_error_carries_the_diagnosis_and_a_correction(client, monkeypatch):  # noqa: F811
@@ -107,9 +110,13 @@ async def test_the_empty_error_carries_the_diagnosis_and_a_correction(client, mo
     got = await _sse(client, BODY)
     [error] = [op for op in got if op["op"] == "error"]
     assert error["code"] == "copilot_empty" and error["next"] == "retry"
-    assert error["message"] == "助手本轮未完成：模型没有给出任何修改，也没有说明原因"
-    assert "正文 0 字" in error["detail"] and "思考 6 字" in error["detail"] and "结束原因 stop" in error["detail"]
-    assert error["diagnostics"]["text_chars"] == 0 and error["diagnostics"]["thinking_chars"] == 6
+    # 说法按诊断照实写，不再一律是「没有给出任何修改，也没有说明原因」
+    assert error["message"] == "助手本轮未完成：模型只写了思考，没有给出任何操作"
+    # 起草时当场纠正过一次（同一个模型又只写了思考）：诊断是两次的
+    assert "正文 0 字" in error["detail"] and "思考 12 字" in error["detail"] and "结束原因 stop" in error["detail"]
+    assert "共调用模型 2 次" in error["detail"] and "当场带纠正重写过一次" in error["detail"], error["detail"]
+    assert error["diagnostics"]["text_chars"] == 0 and error["diagnostics"]["thinking_chars"] == 12
+    assert error["diagnostics"]["nudged"] is True
     assert "只写了思考" in error["reason"]
     # 重来时交给助手的话：问题后面附这句纠正，不是原样重发
     again = policy.empty_retry_instruction("上周各条线的产量", policy.error_reason(error))
diff --git a/backend/tests/test_copilot_markup_ops.py b/backend/tests/test_copilot_markup_ops.py
new file mode 100644
index 00000000..eb63a744
--- /dev/null
+++ b/backend/tests/test_copilot_markup_ops.py
@@ -0,0 +1,230 @@
+"""模型把操作写成了工具调用的格式时，照操作读；一个操作都没认出时当场纠正一次；报错照实说。
+
+部署上撞过：画布上让助手搭一张按天出数的图，连着几轮都是「助手本轮未完成：模型没有给出任何修改，也没有说明原因」，
+又试了一次才成。用同一个模型重放：模型思考里写着「先输出 look」，正文却是一段 DeepSeek 的 DSML 工具调用
+（<｜｜DSML｜｜ invoke name="look"> + need / targets 两个参数），意图和参数都对，只是不是约定的单行 JSON。读流只认 JSON 和
+原生工具调用，于是一个操作都没认出、整轮报错；画布上没有自动重来，点重试还是同样的几率。现在：
+
+- 写成文字的工具调用（DSML、<invoke>、<tool_call>、DeepSeek 特殊 token）一次调用写完就按名字读成操作（名字是操作名的才认），
+  流式一行行攒、读到 look 照样停下去查看；
+- 起草时一次调用读完、一个 plan 以外的操作都没认出：把原话交回去、附一句纠正，请它当场重写一次（每轮一次）；
+- 还是没有，报错的说法按诊断照实写，不再一律是「没有给出任何修改，也没有说明原因」；
+- 提示词里写明这里没有工具调用接口。
+
+需求、库名、表名都是虚构的。
+"""
+from __future__ import annotations
+
+import pytest
+from langchain_core.messages import AIMessageChunk
+
+from app.api.copilot_gen import FORMAT_NUDGE_PROMPT, Generation, GraphApplier, StreamDiagnosis, _iter_ops
+from app.engine.toolcalls import LeakedCall, has_markup, leaked_markup, take_leaked_calls
+from tests.test_copilot_look import START, Turns, _events, _graph_ops, _line, _ops, _tables, client, shop  # noqa: F401
+
+#: 部署上见到的那种写法：竖线两个、DSML 后面带空格，外层包的是 calls，参数带 string 属性
+DSML_LOOK = (
+    "<｜｜DSML｜｜ calls>\n"
+    '<｜｜DSML｜｜ invoke name="look">\n'
+    '<｜｜DSML｜｜ parameter name="need" string="false">订单金额相关的表，要看清表名和字段</｜｜DSML｜｜ parameter>\n'
+    '<｜｜DSML｜｜ parameter name="targets" string="false">[{"kind": "table", "name": "orders"}]</｜｜DSML｜｜ parameter>\n'
+    "</｜｜DSML｜｜ invoke>\n"
+    "</｜｜DSML｜｜ calls>\n"
+)
+LOOK = {"op": "look", "need": "订单金额相关的表，要看清表名和字段", "targets": [{"kind": "table", "name": "orders"}]}
+
+
+class _Model:
+    def __init__(self, chunks: list[AIMessageChunk]) -> None:
+        self.chunks = chunks
+
+    async def astream(self, messages):
+        for chunk in self.chunks:
+            yield chunk
+
+
+async def _read(text: str, *, size: int = 0, lines: list[str] | None = None) -> tuple[list[dict], StreamDiagnosis]:
+    """把 text 当作模型正文读一遍（size>0 时按这么多字切成小段流过来）。"""
+    pieces = [text[i:i + size] for i in range(0, len(text), size)] if size else [text]
+    diag = StreamDiagnosis()
+    model = _Model([AIMessageChunk(content=p) for p in pieces]
+                   + [AIMessageChunk(content="", response_metadata={"finish_reason": "stop"})])
+    ops = [op async for op in _iter_ops(model, [], lines=lines, diag=diag)]
+    return [op for op in ops if op["op"] != "thinking"], diag
+
+
+# ---------------------------------------------------------------- 认标记
+
+@pytest.mark.parametrize("text, call", [
+    # 部署上见到的 DSML
+    (DSML_LOOK, LeakedCall("look", {k: v for k, v in LOOK.items() if k != "op"})),
+    # 标准的 DSML：竖线一个、外层是 function_calls；string="true" 原样是字符串（哪怕看着像数字）
+    ('<｜DSML｜function_calls>\n<｜DSML｜invoke name="look">\n'
+     '<｜DSML｜parameter name="need" string="true">2024 年的订单</｜DSML｜parameter>\n'
+     '<｜DSML｜parameter name="limit" string="true">3</｜DSML｜parameter>\n'
+     "</｜DSML｜invoke>\n</｜DSML｜function_calls>",
+     LeakedCall("look", {"need": "2024 年的订单", "limit": "3"})),
+    # 去掉 DSML 前缀的 <invoke>：没写 string 的，像 JSON 就按 JSON 读
+    ('<function_calls><invoke name="add_node"><parameter name="node">{"id": "q", "type": "llm"}</parameter>'
+     '<parameter name="label">查订单</parameter></invoke></function_calls>',
+     LeakedCall("add_node", {"node": {"id": "q", "type": "llm"}, "label": "查订单"})),
+    # <tool_call>：arguments 可能是一段 JSON 字符串
+    ('<tool_call>{"name": "look", "arguments": "{\\"need\\": \\"订单表\\"}"}</tool_call>',
+     LeakedCall("look", {"need": "订单表"})),
+    ('<tool_call>{"function": {"name": "done", "arguments": {"explanation": "好了"}}}</tool_call>',
+     LeakedCall("done", {"explanation": "好了"})),
+    # DeepSeek 特殊 token 的两种排法
+    ('<｜tool▁calls▁begin｜><｜tool▁call▁begin｜>function<｜tool▁sep｜>look\n```json\n{"need": "订单表"}\n```'
+     "<｜tool▁call▁end｜><｜tool▁calls▁end｜>",
+     LeakedCall("look", {"need": "订单表"})),
+    ('<｜tool▁call▁begin｜>look<｜tool▁sep｜>{"need": "订单表"}<｜tool▁call▁end｜>', LeakedCall("look", {"need": "订单表"})),
+])
+def test_each_markup_style_gives_its_call(text, call):
+    assert has_markup(text)
+    assert take_leaked_calls(text) == ([call], "")
+
+
+def test_an_unfinished_call_is_kept_until_its_end_arrives():
+    calls, rest = take_leaked_calls('说明文字 <｜DSML｜function_calls>\n<｜DSML｜invoke name="look">\n')
+    assert calls == [] and rest.startswith('<invoke name="look">')
+    calls, rest = take_leaked_calls(rest + '<｜DSML｜parameter name="need" string="true">订单</｜DSML｜parameter>\n'
+                                    "</｜DSML｜invoke>\n<｜DSML｜invoke name=\"done\">\n")
+    assert calls == [LeakedCall("look", {"need": "订单"})] and rest.startswith('<invoke name="done">')
+    # 流读完了还差结尾：参数都写完整了的补上结尾再读；参数写到一半、补上也读不出的照样交回（调用方记「没写完」）
+    assert take_leaked_calls(rest, final=True) == ([LeakedCall("done", {})], "")
+    half = '<invoke name="look">\n<parameter name="need" string="true">没写完'
+    assert take_leaked_calls(half, final=True) == ([], half)
+    assert take_leaked_calls("<tool_call>{坏的", final=True) == ([], "<tool_call>{坏的")
+
+
+def test_closing_dsml_tags_and_call_wrappers_count_as_leaked_markup():
+    # 节点里的 agent 也用这个认：DSML 的结尾标签、<function_calls> 包装以前漏认
+    assert leaked_markup("</｜DSML｜invoke>") and leaked_markup("<function_calls>") and leaked_markup("<tool_calls>")
+    assert not has_markup('{"op": "look", "need": "订单"}') and not has_markup("先看一下 orders 表")
+
+
+# ---------------------------------------------------------------- 读流
+
+@pytest.mark.parametrize("size", [0, 7])
+async def test_a_dsml_look_is_read_as_a_look_and_the_stream_stops_there(size):
+    lines: list[str] = []
+    ops, diag = await _read(DSML_LOOK + _line({"op": "add_node", "node": {"id": "ghost"}}) + "\n", size=size,
+                            lines=lines)
+    assert ops == [LOOK]
+    assert diag.markup == 1 and diag.ops == 1 and diag.dropped == {}
+    # 接着写时交回模型的是约定的写法，不是它写的那段标记
+    assert lines == [_line(LOOK)]
+
+
+async def test_markup_and_json_lines_mix():
+    text = (_line({"op": "plan", "summary": "先看表"}) + "\n" + DSML_LOOK.replace('name="look"', 'name="update_node"')
+            .replace('name="need"', 'name="id"').replace('name="targets"', 'name="config"') + _line({"op": "done"}))
+    ops, diag = await _read(text)
+    assert [op["op"] for op in ops] == ["plan", "update_node", "done"]
+    assert ops[1]["id"] == "订单金额相关的表，要看清表名和字段" and diag.markup == 1 and diag.dropped == {}
+
+
+async def test_markup_that_cannot_be_read_is_diagnosed():
+    text = ('<｜DSML｜function_calls>\n<｜DSML｜invoke name="run_sql">\n'
+            '<｜DSML｜parameter name="sql" string="true">SELECT 1</｜DSML｜parameter>\n</｜DSML｜invoke>\n'
+            '<｜DSML｜invoke name="look">\n<｜DSML｜parameter name="need" string="true">没写完')
+    ops, diag = await _read(text)
+    assert ops == [] and diag.markup == 1
+    assert diag.dropped == {"工具调用标记的名字不认识": 1, "工具调用标记没写完": 1}, diag.dropped
+
+
+# ---------------------------------------------------------------- 当场纠正
+
+async def test_a_dsml_look_drives_a_real_look_and_the_graph_lands(shop):
+    """部署上那一幕：第一次调用就把 look 写成了 DSML。现在照样去查看、接着写、图落地，模型只调两次。"""
+    model = Turns([AIMessageChunk(content=DSML_LOOK)], [START, *_graph_ops(shop.name)])
+    nodes: dict = {}
+    gen = Generation(model, system="SYS", applier=GraphApplier(nodes, []), looks=await _tables(shop))
+    events = [op async for op in gen.draft("REQ")]
+    assert len(model.calls) == 2 and not gen.diagnosis.nudged
+    assert model.calls[1][2] == ("ai", _line(LOOK))
+    look = next(e for e in events if e["op"] == "look")
+    assert look["status"] == "ok" and look["objects"][0]["named"] == ["orders"]
+    assert set(nodes) == {"start", "q", "out"}
+
+
+async def test_prose_gets_one_correction_with_its_own_words_handed_back(shop):
+    model = Turns([AIMessageChunk(content="我先看一下订单表的结构，\n再决定怎么搭。\n")], [START, *_graph_ops(shop.name)])
+    nodes: dict = {}
+    gen = Generation(model, system="SYS", applier=GraphApplier(nodes, []), looks=await _tables(shop))
+    [op async for op in gen.draft("REQ")]
+    assert len(model.calls) == 2 and gen.diagnosis.nudged and set(nodes) == {"start", "q", "out"}
+    system, human, ai, nudge = model.calls[1]
+    assert human == ("human", "REQ") and ai == ("ai", "我先看一下订单表的结构，\n再决定怎么搭。")
+    assert nudge[0] == "human" and nudge[1].startswith("上面这次输出里没有一行能认出的操作：")
+    assert "每个操作单独一行" in nudge[1] and nudge[1].endswith("请按输出格式重新给出这一轮的操作。")
+
+
+async def test_an_empty_continuation_after_a_look_is_corrected_too(shop):
+    """真模型重放里撞过：先 look，查看完接着写的那一次交回一大段读不成的东西，整轮照样是空的。look 不算进展，照样纠正。"""
+    look = {"op": "look", "need": "订单金额", "targets": [{"kind": "table", "name": "orders"}]}
+    model = Turns([look], [AIMessageChunk(content='{"op": "add_node", "node": {"id": "q"}}{"op": "done"\n')],
+                  [START, *_graph_ops(shop.name)])
+    nodes: dict = {}
+    gen = Generation(model, system="SYS", applier=GraphApplier(nodes, []), looks=await _tables(shop))
+    [op async for op in gen.draft("REQ")]
+    assert len(model.calls) == 3 and gen.diagnosis.nudged and set(nodes) == {"start", "q", "out"}
+    *_, result, said, nudge = model.calls[2]
+    assert result[1].startswith("查看结果") and said == ("ai", '{"op": "add_node", "node": {"id": "q"}}{"op": "done"')
+    assert nudge[1].startswith("上面这次输出里没有一行能认出的操作：")
+
+
+async def test_only_thinking_gets_the_correction_appended_to_the_request(shop):
+    thinking = AIMessageChunk(content="", additional_kwargs={"reasoning_content": "先看表。"})
+    model = Turns([thinking], [START, *_graph_ops(shop.name)])
+    gen = Generation(model, system="SYS", applier=GraphApplier({}, []), looks=await _tables(shop))
+    [op async for op in gen.draft("REQ")]
+    assert len(model.calls) == 2
+    system, human = model.calls[1]
+    assert human[0] == "human" and human[1].startswith("REQ\n\n上面这次输出里没有一行能认出的操作：只写了思考")
+
+
+async def test_the_correction_is_given_once_and_never_after_real_ops(shop):
+    # 只交了 plan、又什么都没给：纠正一次，还是这样就收场（调用方报 copilot_empty）
+    plan_only = Turns([{"op": "plan", "summary": "打算先看表"}])
+    gen = Generation(plan_only, system="SYS", applier=GraphApplier({}, []), looks=await _tables(shop))
+    [op async for op in gen.draft("REQ")]
+    assert len(plan_only.calls) == 2 and gen.diagnosis.nudged and gen.diagnosis.calls == 2
+    assert "没有给出任何操作" in plan_only.calls[1][-1][1]
+    assert gen.empty().args[0] == "模型只说了打算，没有给出任何修改"
+    # 交过真操作（这里是「无需修改」的 done）不算跑偏，不纠正
+    done = Turns([{"op": "done", "explanation": "不用改"}])
+    gen = Generation(done, system="SYS", applier=GraphApplier({}, []), looks=await _tables(shop))
+    [op async for op in gen.draft("REQ")]
+    assert len(done.calls) == 1 and not gen.diagnosis.nudged
+
+
+async def test_the_canvas_stream_recovers_without_an_error(client, monkeypatch, shop):  # noqa: F811
+    model = Turns([AIMessageChunk(content="好的，我来设计。\n")], [START, *_graph_ops(shop.name)])
+    events = await _events(client, monkeypatch, model, {"instruction": "订单总金额", "intent": "build",
+                                                         "datasource_ids": [shop.id]})
+    assert "error" not in _ops(events) and events[-1]["op"] == "final", _ops(events)
+
+
+# ---------------------------------------------------------------- 报错的说法、提示词
+
+@pytest.mark.parametrize("diag, headline", [
+    (StreamDiagnosis(markup=1, text_chars=200), "模型把操作写成了工具调用的格式，没有按约定逐行输出，系统认不出"),
+    (StreamDiagnosis(tool_calls=["run_sql"]), "模型把操作写成了工具调用的格式，没有按约定逐行输出，系统认不出"),
+    (StreamDiagnosis(thinking_chars=30), "模型只写了思考，没有给出任何操作"),
+    (StreamDiagnosis(), "模型什么都没有输出"),
+    (StreamDiagnosis(text_chars=40, dropped={"不是完整的 JSON": 2}), "模型的输出格式不对，没有一行能认出的操作"),
+    (StreamDiagnosis(text_chars=40, ops=1), "模型只说了打算，没有给出任何修改"),
+])
+def test_the_headline_says_what_actually_came_back(diag, headline):
+    assert diag.headline() == headline
+
+
+def test_the_prompts_say_there_is_no_tool_call_interface():
+    from app.api import copilot, copilot_blueprint
+    from app.look.service import GUIDE_HEAD_PROMPT
+
+    assert "这里没有工具调用接口" in GUIDE_HEAD_PROMPT and "DSML" in GUIDE_HEAD_PROMPT
+    for protocol in (copilot._STREAM_PROTOCOL, copilot.APP_PROTOCOL_PROMPT, copilot_blueprint.BLUEPRINT_PROTOCOL_PROMPT):
+        assert "这里没有工具调用接口" in protocol
+    assert "{correction}" in FORMAT_NUDGE_PROMPT
diff --git a/backend/tests/test_copilot_truncation.py b/backend/tests/test_copilot_truncation.py
index 9bc29a0d..1526b637 100644
--- a/backend/tests/test_copilot_truncation.py
+++ b/backend/tests/test_copilot_truncation.py
@@ -163,7 +163,8 @@ async def test_a_cut_in_the_middle_of_the_ops_is_not_delivered_as_done(client, m
 
 
 async def test_output_that_parses_to_nothing_is_an_error(client, monkeypatch):
-    """格式跑偏（只写了说明文字、没有一行是操作），每一行都被跳过：一个操作都没有，也没有 done。诊断写明跳过了几行。"""
+    """格式跑偏（只写了说明文字、没有一行是操作），每一行都被跳过：一个操作都没有，也没有 done。当场带纠正重写一次还是这样，
+    才报错；诊断写明跳过了几行（两次的）。"""
     model = _Scripted([AIMessageChunk(content="先看一下表结构，\n再决定怎么搭。\n"),
                        AIMessageChunk(content="", response_metadata={"finish_reason": "stop"})])
     events = await _events(client, monkeypatch, model)
@@ -171,7 +172,8 @@ async def test_output_that_parses_to_nothing_is_an_error(client, monkeypatch):
     assert "final" not in kinds
     [err] = [e for e in events if e["op"] == "error"]
     assert err["message"].startswith("助手本轮未完成：") and "画布未修改" in err["hint"]
-    assert "跳过 2 行：2 行不是完整的 JSON" in err["detail"] and "先看一下表" not in err["detail"], err
+    assert "跳过 4 行：4 行不是完整的 JSON" in err["detail"] and "先看一下表" not in err["detail"], err
+    assert err["message"] == "助手本轮未完成：模型的输出格式不对，没有一行能认出的操作" and model.calls == 2, err
 
 
 async def test_a_pretty_printed_op_in_a_fence_is_read(client, monkeypatch):