画完先自查：条件写错在校验时就拦下，Copilot 交付前按同一套规则查一遍、查出问题先改

开发库里四次运行死在同一句"循环条件写错了：表达式里不允许出现 Set"。条件原文是
`{{ vars.gate }} != "ok"`——把模板写法带进了表达式：`{{ x }}` 在 Python 里是"装着
一个集合的集合"。Copilot 的提示词写着"模板语法在任意字符串里用"，模型照做而已；
校验又不看表达式，写错了只有跑到那一步才知道，前面的步骤白跑。

1. 表达式：解析和静态检查收成一个 parse_expression，运行时、校验、Copilot 自查共用，
   不会出现"校验说能跑、跑起来报错"。多套的 {{ }} 意思无歧义，在语法树上剥掉（字符串
   里的花括号不动），按 vars.x 理解并提示。真写错的给能照改的提示：| 过滤器 → len(x)、
   集合 → 列表、不认识的函数列出能用的、= 当 ==、引号里的 {{ }} 永远不成立、推导式、
   切片、私有属性。and / or 短路掉的那半边也查。

2. 校验：按节点类型和模式挑出真正求值的字段（分支条件、while 条件、整形表达式、口径卡
   指标、skip_if）。写错是 error、挡住运行；漏写 vars. 的名字提示会取到空值；while 没写
   条件、整形没填表达式也说出来。foreach 的 condition、模板模式的 expression 不求值，不查。

3. 代码节点：assign_to 拿的是 stdout。「测试 1」修好条件后跑通了、结果却是空的——代码
   最后一行是裸表达式，那是 notebook 的行为。设了 assign_to 却找不到任何输出的 Python
   代码，校验直接拦下，提示 print(json.dumps(结果))。宁可漏判不可误判：子进程、exec、
   附带文件里的 print 都算有输出；整行只有一个 {{ }} 的代码是运行时才拿到的（⑤号示例
   整段就是 {{ vars.code }}），看不见，不猜。库里 20 张工作流无一误拦。

4. Copilot 自查：搭完用同一套校验过一遍，error 级的问题连同图交回模型——只修这些、只
   输出改动，最多两轮。改好了才自动运行；改不好照实交付、列出剩下的问题，不自动开跑。
   没问题的图不多花一次调用；只是多套了 {{ }} 的不返工。提示词改成"模板只在普通文本
   字段里用"，并列出表达式字段的对 / 错写法和代码节点必须 print。

5. 前端：自查这一段在 Copilot 的过程里看得见——发现几处问题、交回去改、改好了没有；
   改不好标红并说明没有自动运行。"搭好了，共几步"那一行不会被挤掉。

测试：新增 test_expressions（开发库里真实死过的四条条件、静态提示、按模式挑字段、代码
输出、带 {{ }} 的循环真跑一遍）与 test_copilot_selfcheck（脚本化模型：改好 → 自动运行；
改不好 → 不运行、最多两轮；干净的图一次调用；多套 {{ }} 不返工）。test_continue_run 里
用来制造运行时失败的表达式换成 `2 ** 100000 > 1`——原来那条现在在校验阶段就被拦下了。
check-decode 补了自查的解码断言。

README.md

@@ -62,7 +62,7 @@ conda env create -f environment.yml   # 首次：创建 agentlab 环境并安装
 | **表格入库** | Excel / CSV 上传后转为可用 SQL 查询的表，自动生成 `db_query__<name>` 工具——数字由 SQL 计算得出，而非模型读取 |
 | **长期记忆** | 跨运行的记忆读写与自动去重；Copilot 建图时主动记录应当记住的事实 |
 | **Skill 管理** | 将方法论抽象为可复用单元，挂载到节点上注入 system prompt |
-| **自然语言编排** | Copilot：一句话生成或改写整张图，自动排版与校验 |
+| **自然语言编排** | Copilot：一句话生成或改写整张图，自动排版；交付前按运行时同一套规则自查，查出会跑挂的问题先交回去改，改不好就不自动运行 |
 | **结构化成果** | JSON Schema 校验，不合格时将错误回传模型自动返工 |
 | **产出复核** | 运行结束后规则扫描事件流，发现异常才调用模型复核；复核可重组答案但**不得引入原答案中不存在的数字** |
 | **追踪与成本** | 全量事件落库可回放；逐节点耗时、token 用量、按目录价折算的成本 |
@@ -155,6 +155,18 @@ flowchart LR
 {{ vars.items | json }}     过滤器：json / compact / upper / length / first …
 ```
 
+**条件和表达式不是模板。** 分支条件、while 循环条件、整形表达式、口径卡指标、跳过条件
+这几处是直接求值的表达式，变量直接写：`vars.gate != 'ok'`、`len(vars.items) > 0`。
+多套了 `{{ }}` 意思没有歧义，照 `vars.gate` 理解、只给一条提示；真写错的在校验阶段就拦下，
+不会等跑到那一步才炸——`| length` 这类过滤器（改写 `len(x)`）、`{'a', 'b'}` 集合（改用列表）、
+不认识的函数、把 `==` 写成 `=`、引号里的 `{{ }}`（它不会被渲染，条件永远不成立）。
+校验、Copilot 自查和运行时用的是同一个解析器，不会出现"校验说能跑、跑起来报错"。
+
+**代码节点的结果要 print 出来。** `assign_to` 拿到的是 stdout：脚本最后一行写个裸表达式
+不会输出（那是 notebook 的行为），变量会是空的。要交给下游一个对象就
+`print(json.dumps(结果))`，stdout 是 JSON 时会自动解析成对象。设了 `assign_to` 却找不到
+任何输出的 Python 代码，校验会直接拦下。
+
 连线即并行：单个节点连出多条边构成 fan-out，多条边汇入同一节点时自动等待汇聚——
 汇合节点等图里其余待执行的任务都跑完再执行一次，支路一长一短也不会被触发两遍；
 分支之后的二选一汇合不会去等没走的那条。代价是它也会等图里与它无关、恰好还在跑的支路。

backend/app/api/copilot.py

@@ -122,7 +122,9 @@ NODE_REFERENCE = """\
 - code：沙箱里跑代码。config: {language: python|bash|node, code, timeout, network, assign_to}
   assign_to 拿到的是 stdout（尾部的换行已去掉）。**stdout 是 JSON 时会解析成对象**，
   这时下游要用 {{ vars.x.字段 }} 取字段，拿它比字符串永远不成立。要判一个简单结论，
-  就 print 一个短字符串（比如 print('ok')）并让下游比它
+  就 print 一个短字符串（比如 print('ok')）并让下游比它。
+  **结果必须 print 出来**：脚本最后一行写个裸表达式（state、{"a": 1}）不会输出，那是
+  notebook 的行为——assign_to 会拿到空值。要交给下游一个对象，就 print(json.dumps(结果))
 - branch：条件分支。config: {mode: expression|llm, cases:[{key, condition, label}]}
 - loop：循环。config: {mode: foreach|while, items, item_var, condition, max_iterations}
 - retrieve：知识库检索。config: {query, collection, limit, assign_to}
@@ -141,13 +143,25 @@ NODE_REFERENCE = """\
   "deepseek-chat"）会得到 401 invalid_model，整个节点跑不起来。
 - 只有用户在需求里明确点名了某个模型，才把他说的那个字符串原样填进去。
 
-模板语法（在任意字符串里用）：
+模板语法（在 prompt、template、args、message、query 这类**普通文本字段**里用）：
 - {{ input.字段名 }}        入口输入
 - {{ vars.变量名 }}         某节点 assign_to 写入的变量
 - {{ nodes.节点id.text }}   某节点的输出
 - {{ last_message }}        最近一条消息文本
 - 过滤器：{{ vars.x | json }}
 
+**表达式字段不是模板，不要套 {{ }}**：branch 的 cases[].condition、loop（while 模式）的
+condition、transform（expression 模式）的 expression、口径卡的 metrics[].expression、任意节点的
+skip_if。这些字段按受限的 Python 表达式求值，变量直接写路径：
+- 对：vars.gate == "ok"            错：{{ vars.gate }} == "ok"（会被当成集合，跑不起来）
+- 对：vars.state.count <= 1000     错：'{{ vars.x }}' == "ok"（引号里的模板不会渲染，永远不成立）
+- 对：len(vars.items) > 0          错：{{ vars.items | length }} > 0（表达式里没有 | 过滤器）
+- 能用：and / or / not、== != < <= > >=、in [列表]、x if 条件 else y；函数只有 len str int float
+  round min max sum any all lower upper contains startswith endswith matches get
+- 字符串加引号；判断相等写 ==；"是不是其中之一"用列表 x in ['a', 'b']，不要用 {…}
+
+图搭完会用和运行时同一套规则自查：表达式写错、必填没填、引用了不存在的变量，都会被打回来让你改。
+
 **引用必须有来源**（这条会被校验，不满足整张图跑不起来）：
 - 写 {{ vars.X }} 之前，必须有某个节点的 config.assign_to 正好是 "X"
 - 写 {{ input.X }} 之前，X 必须在入口节点的 fields 里声明过
@@ -640,14 +654,13 @@ async def _iter_ops(model: Any, messages: list[Any]):
 _HEARTBEAT_SECONDS = 3.0
 
 
-async def _with_heartbeat(source: Any):
+async def _with_heartbeat(source: Any, phase: str = "planning"):
     """模型沉默时按拍补心跳，让前端能显示阶段和已用时长。
 
     必须在这一层做而不是在 _iter_ops 里判断时间差：模型不吐 chunk 时那个
     async for 的循环体根本不执行，压根轮不到检查。
     """
     started = time.monotonic()
-    phase = "planning"
     iterator = source.__aiter__()
     while True:
         pending = asyncio.ensure_future(iterator.__anext__())
@@ -675,6 +688,42 @@ async def _with_heartbeat(source: Any):
             }
 
 
+#: 自查最多交回去改几轮。一轮只花一次"只输出改动"的调用；两轮还改不好，多半是
+#: 需求本身有歧义，该让人看了，而不是继续替他烧钱
+_SELF_CHECK_ROUNDS = 2
+
+
+def _sse(obj: dict[str, Any]) -> str:
+    return f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"
+
+
+def _blocking_issues(nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]]) -> list[Any] | None:
+    """按运行时同一套校验，挑出会挡住运行的问题。图本身不成形时返回 None。"""
+    try:
+        spec = GraphSpec.model_validate({"nodes": list(nodes.values()), "edges": edges})
+    except Exception:  # noqa: BLE001 - 结构不合法留给收尾那一步报
+        return None
+    return [i for i in validate_graph(spec).issues if i.level == "error"]
+
+
+def _issue_line(issue: Any) -> str:
+    return f"「{issue.node_id}」{issue.message}" if issue.node_id else issue.message
+
+
+def _repair_request(
+    nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], errors: list[Any],
+) -> str:
+    graph = {"nodes": list(nodes.values()), "edges": edges}
+    return (
+        "这是你刚搭好的工作流：\n"
+        f"{json.dumps(_slim(graph), ensure_ascii=False, indent=2)}\n\n"
+        "上线前自查（和真正运行时同一套规则）发现下面这些问题，不改的话这张图跑不起来：\n"
+        + "\n".join(f"- {_issue_line(i)}" for i in errors)
+        + "\n\n只修这些问题，别的不要动。用操作流输出修正：改节点用 update_node"
+          "（config 整体替换，要带上这个节点完整的 config）；缺节点、缺边就补上。最后一行输出 done。"
+    )
+
+
 @router.post("/generate-stream")
 async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(get_session)):
     """流式版生成：SSE 逐操作推送，前端边收边把节点摆上画布。
@@ -784,6 +833,40 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
             for op in seen_ops:
                 _apply_op(nodes, edges, op)
 
+        # 自查：按运行时同一套规则把图过一遍，挡住运行的问题交回模型改，改完再查。
+        # 以前这些问题照样交付、照样自动开跑——跑到那一步才炸（开发库里四次运行死在
+        # 循环 / 分支条件里套了 {{ }}），前面的步骤白跑，报错还停在半路
+        repaired = 0
+        for round_no in range(1, _SELF_CHECK_ROUNDS + 1):
+            errors = _blocking_issues(nodes, edges)
+            if not errors:
+                break
+            yield _sse({"op": "check", "status": "repairing", "round": round_no,
+                        "issues": [_issue_line(i) for i in errors]})
+            fix = [("system", system), ("human", _repair_request(nodes, edges, errors))]
+            try:
+                async for op in _with_heartbeat(_iter_ops(model, fix), phase="repairing"):
+                    kind = op.get("op")
+                    if kind in ("thinking", "heartbeat"):
+                        yield _sse(op)
+                    # plan / done 是修补轮自己的开场和收尾，不是新方案，不覆盖原来的说明
+                    elif kind not in ("plan", "done", "reply") and _apply_op(nodes, edges, op):
+                        yield _sse(op)
+            except Exception as e:  # noqa: BLE001 - 修不成就照实交付，下面列出剩下的问题
+                yield _sse({"op": "check", "status": "error",
+                            "message": f"自查修正没跑成：{type(e).__name__}: {e}"})
+                break
+            repaired = round_no
+        remaining = _blocking_issues(nodes, edges)
+        if remaining:
+            # 改不好也照实交付、列出问题，但不自动运行：明知跑不起来的图，开跑只会
+            # 在半路报一个用户看不懂的错
+            autorun = False
+            yield _sse({"op": "check", "status": "failed",
+                        "issues": [_issue_line(i) for i in remaining]})
+        elif remaining is not None:
+            yield _sse({"op": "check", "status": "passed", "repaired": repaired})
+
         # 收尾：排版 + 校验，把最终图整体交付
         try:
             spec = auto_layout(

backend/app/engine/expressions.py

@@ -176,11 +176,107 @@ _SAFE_FUNCS: dict[str, Any] = {
 
 _MAX_NODES = 200
 
+#: 表达式能引用的根名字，就是 state.template_context 的那几个键（有测试钉住两边一致）。
+#: 不在这里、也不是白名单函数或 true/false/null 的名字，运行时一律取到 None——
+#: `gate == "ok"` 这种漏写了 vars. 的条件永远不成立，而且不报错
+EXPRESSION_ROOTS = frozenset(
+    {"input", "nodes", "vars", "output", "loops", "usage", "messages", "last_message"}
+)
+_LITERAL_NAMES = frozenset({"true", "True", "false", "False", "null", "None"})
+
+_ALLOWED_NODES: tuple[type[ast.AST], ...] = (
+    ast.Expression, ast.Constant, ast.Name, ast.Load, ast.Attribute, ast.Subscript,
+    ast.BinOp, ast.UnaryOp, ast.BoolOp, ast.And, ast.Or, ast.Compare, ast.IfExp,
+    ast.Call, ast.keyword, ast.List, ast.Tuple, ast.Dict,
+    *_BIN_OPS, *_UNARY_OPS, *_CMP_OPS,
+)
+_NODE_NAMES = {
+    "ListComp": "列表推导式", "SetComp": "集合推导式", "DictComp": "字典推导式",
+    "GeneratorExp": "生成器表达式", "Lambda": "lambda", "JoinedStr": "f-string",
+    "NamedExpr": ":=", "Slice": "切片 a[i:j]", "Starred": "* 展开", "Await": "await",
+}
+
 
 class ExpressionError(ValueError):
     pass
 
 
+class _Unbrace(ast.NodeTransformer):
+    """`{{ vars.x }}` 在 Python 里是"装着一个集合的集合"。表达式不允许集合，所以
+    它在表达式里只可能是一种笔误：把模板写法带了进来——模型和人都这么写（开发库
+    里四次运行就死在"表达式里不允许出现 Set"）。意思毫无歧义，按 vars.x 理解。
+
+    在语法树上剥，不在文本上剥：字符串字面量里的花括号原样保留。
+    """
+
+    def __init__(self) -> None:
+        self.found: list[str] = []
+
+    def visit_Set(self, node: ast.Set) -> ast.AST:
+        self.generic_visit(node)
+        inner = node.elts[0] if len(node.elts) == 1 else None
+        if isinstance(inner, ast.Set) and len(inner.elts) == 1:
+            self.found.append(ast.unparse(inner.elts[0]))
+            return inner.elts[0]
+        return node
+
+
+def _check_static(tree: ast.Expression) -> list[str]:
+    """不求值，只看结构：违规直接抛 ExpressionError，返回不认识的名字（留给提示用）。
+
+    运行时求值会短路（and/or、三元只走一边），没走到的那一半写错了也发现不了；
+    画布校验和 Copilot 自查要的是"整条表达式都能跑"，所以得把整棵树走一遍。
+    """
+    unknown: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.BinOp) and type(node.op) not in _BIN_OPS:
+            if isinstance(node.op, ast.BitOr):
+                raise ExpressionError(
+                    "表达式里没有 | 过滤器（那是模板的写法）：取长度写 len(x)，大小写写 upper(x) / lower(x)"
+                )
+            raise ExpressionError(f"不支持的运算符 {type(node.op).__name__}")
+        if not isinstance(node, _ALLOWED_NODES):
+            if isinstance(node, ast.Set):
+                raise ExpressionError("表达式里不支持集合 {…}：要判断是不是其中之一，用列表 x in ['a', 'b']")
+            name = type(node).__name__
+            raise ExpressionError(f"表达式里不允许出现 {_NODE_NAMES.get(name, name)}")
+        if isinstance(node, ast.Call):
+            if not isinstance(node.func, ast.Name) or node.func.id not in _SAFE_FUNCS:
+                called = node.func.id if isinstance(node.func, ast.Name) else ast.unparse(node.func)
+                raise ExpressionError(f"不认识的函数 {called}()，能用的只有：{'、'.join(_SAFE_FUNCS)}")
+        elif isinstance(node, ast.Attribute) and node.attr.startswith("_"):
+            raise ExpressionError("不允许访问私有属性")
+        elif (isinstance(node, ast.Constant) and isinstance(node.value, str)
+              and _TEMPLATE_RE.search(node.value)):
+            raise ExpressionError(
+                f"字符串 {node.value!r} 里的 {{{{ }}}} 不会被渲染——表达式不过模板，直接写路径，比如 vars.x"
+            )
+        elif (isinstance(node, ast.Name) and node.id not in EXPRESSION_ROOTS
+              and node.id not in _SAFE_FUNCS and node.id not in _LITERAL_NAMES):
+            unknown.append(node.id)
+    return list(dict.fromkeys(unknown))
+
+
+def parse_expression(expr: str) -> tuple[ast.Expression, list[str], list[str]]:
+    """解析 + 静态检查：(语法树, 被剥掉的 {{ }}, 不认识的名字)。结构违规抛 ExpressionError。
+
+    运行时求值、画布校验、Copilot 自查都走这一个入口——三处各写一套规则的话，
+    迟早出现"校验说能跑、跑起来报错"。
+    """
+    text = (expr or "").strip()
+    try:
+        tree = ast.parse(text, mode="eval")
+    except SyntaxError as e:
+        hint = "（判断相等要写 ==）" if re.search(r"(?<![=!<>])=(?!=)", text) else ""
+        raise ExpressionError(f"表达式语法错误：{e.msg}{hint}") from e
+    unbrace = _Unbrace()
+    tree = ast.fix_missing_locations(unbrace.visit(tree))
+    if sum(1 for _ in ast.walk(tree)) > _MAX_NODES:
+        raise ExpressionError("表达式太复杂")
+    unknown = _check_static(tree)
+    return tree, unbrace.found, unknown
+
+
 def eval_expression(expr: str, ctx: dict[str, Any]) -> Any:
     """求值一个受限的 Python 表达式。
 
@@ -190,13 +286,7 @@ def eval_expression(expr: str, ctx: dict[str, Any]) -> Any:
     expr = (expr or "").strip()
     if not expr:
         return None
-    try:
-        tree = ast.parse(expr, mode="eval")
-    except SyntaxError as e:
-        raise ExpressionError(f"表达式语法错误：{e.msg}") from e
-
-    if sum(1 for _ in ast.walk(tree)) > _MAX_NODES:
-        raise ExpressionError("表达式太复杂")
+    tree, _, _ = parse_expression(expr)
 
     def _eval(node: ast.AST) -> Any:
         if isinstance(node, ast.Expression):

backend/app/engine/schema.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 from collections import defaultdict
 from enum import StrEnum
 from typing import Any, Iterable, Literal
@@ -210,6 +211,44 @@ def back_edges(
     return back
 
 
+def expression_fields(node: GraphNode) -> list[tuple[str, str]]:
+    """这个节点上按表达式求值的字段：(给人看的名字, 原文)。
+
+    跟着模式走：foreach 循环的 condition、模板模式整形的 expression 不参与求值，
+    拿它们报错就是误报。
+    """
+    cfg = node.config
+    out: list[tuple[str, str]] = []
+
+    def add(label: str, value: Any) -> None:
+        if isinstance(value, str) and value.strip():
+            out.append((label, value))
+
+    add("跳过条件", cfg.get("skip_if"))
+    if node.type == NodeType.BRANCH and cfg.get("mode", "expression") == "expression":
+        for case in cfg.get("cases") or []:
+            case = case or {}
+            add(f"分支「{case.get('label') or case.get('key') or '?'}」的条件", case.get("condition"))
+    elif node.type == NodeType.LOOP and cfg.get("mode", "foreach") == "while":
+        add("循环条件", cfg.get("condition"))
+    elif node.type == NodeType.TRANSFORM and cfg.get("mode", "expression") == "expression":
+        add("整形表达式", cfg.get("expression"))
+    elif node.type == NodeType.METRICS:
+        for d in cfg.get("metrics") or []:
+            d = d or {}
+            add(f"指标「{d.get('id') or '?'}」的表达式", d.get("expression"))
+    return out
+
+
+# 代码节点可能往 stdout 写东西的迹象。宁可漏判不可误判：这条是 error 级、会挡住运行，
+# 所以间接的写法（子进程、exec、附带文件里的 print）都算。整行只有一个 {{ }} 的，
+# 那行代码是运行时从上游拿的（⑤号示例整段就是 {{ vars.code }}），看不见，不猜；
+# 而 `state = {{ vars.state | json }}` 这种只是填一个值，照查
+_MAY_WRITE_STDOUT = re.compile(
+    r"\b(print|pprint|exec)\s*\(|\bstdout\b|\bos\.(system|write|popen)\b|\bsubprocess\b"
+    r"|^\s*\{\{.*\}\}\s*$", re.M)
+
+
 def validate_graph(spec: GraphSpec) -> ValidationResult:
     """编译前的静态检查。错误会挡住运行，警告只在画布上提示。"""
     result = ValidationResult()
@@ -283,6 +322,52 @@ def validate_graph(spec: GraphSpec) -> ValidationResult:
                     node_id=node.id,
                 )
 
+    # 表达式：用运行时同一个 parse_expression 先解析一遍。以前写错了只有跑到那一步
+    # 才知道——前面的步骤白跑、钱白花，报错还停在半路（开发库里四次运行死在
+    # "表达式里不允许出现 Set"，就是条件里套了 {{ }}）
+    from app.engine.expressions import ExpressionError, parse_expression
+
+    for node in spec.nodes:
+        for label, text in expression_fields(node):
+            try:
+                _, unwrapped, unknown = parse_expression(text)
+            except ExpressionError as e:
+                result.add(f"{label}写错了：{e}（原文：{text}）", node_id=node.id)
+                continue
+            if unwrapped:
+                result.add(
+                    f"{label}里的 {{{{ }}}} 是多余的——这里是表达式、不是模板，"
+                    f"已按 {'、'.join(unwrapped)} 理解", level="warning", node_id=node.id,
+                )
+            if unknown:
+                result.add(
+                    f"{label}里的 {'、'.join(unknown)} 不是能用的名字，运行时会取到空值"
+                    "（变量要写全：vars.x、input.x、nodes.某节点.text）",
+                    level="warning", node_id=node.id,
+                )
+        cfg = node.config
+        if node.type == NodeType.LOOP and cfg.get("mode", "foreach") == "while" \
+                and not str(cfg.get("condition") or "").strip():
+            result.add("while 循环没有写条件，循环体一次都不会跑", level="warning", node_id=node.id)
+        if node.type == NodeType.TRANSFORM and cfg.get("mode", "expression") == "expression" \
+                and not str(cfg.get("expression") or "").strip():
+            result.add("整形节点没有填表达式", node_id=node.id)
+        # assign_to 拿到的是 stdout。脚本最后一行写个裸表达式不会输出（那是 notebook
+        # 的行为）——变量是空的，下游的循环条件一上来就不成立、成果也是空的，而整次
+        # 运行照样"成功"。开发库里的「测试 1」就是这样：修好条件之后跑通了，结果为空
+        if node.type == NodeType.CODE and cfg.get("assign_to") \
+                and (cfg.get("language") or "python").lower() == "python":
+            files = cfg.get("files")
+            source = "\n".join([str(cfg.get("code") or ""),
+                                 *(map(str, files.values()) if isinstance(files, dict) else ())])
+            if not _MAY_WRITE_STDOUT.search(source):
+                var = cfg["assign_to"]
+                result.add(
+                    f"代码把输出交给 vars.{var}，可代码里没有 print：Python 脚本最后一行的表达式"
+                    f"不会自动输出，vars.{var} 会是空的。把结果 print 出来；要交给下游一个对象，"
+                    "就 print(json.dumps(结果))", node_id=node.id,
+                )
+
     # 孤儿节点
     for node in spec.nodes:
         if node.type in (NodeType.INPUT, NodeType.OUTPUT):

backend/tests/test_continue_run.py

@@ -75,7 +75,10 @@ def graph_with(*, condition: str):
     }
 
 
-BROKEN = "这不是表达式 >>> 啊"
+# 过得了静态检查、一跑就报错的条件：语法没问题，指数超了上限。以前这里是一句
+# 语法错误——那种图现在在校验阶段就被拦下、根本启动不了（这正是要的），
+# 而这组测试要的是"跑到半路失败、改了配置接着跑"
+BROKEN = "2 ** 100000 > 1"
 FIXED = "false"
 
 

backend/tests/test_copilot_selfcheck.py

@@ -0,0 +1,140 @@
+"""Copilot 搭完图先自查：和运行时同一套规则，有问题交回去改，改好再交付。
+
+以前校验结果只是附在 final 上交给前端，error 级的问题照样交付、照样自动开跑，
+跑到那一步才炸——开发库里四次运行就死在条件表达式上，前面的步骤全白跑。
+
+这里守四件事：写错了会被交回去改、改好的图才自动运行；改不好也照实交付但不自动
+运行；没问题的图不多花一次调用；只是多套了 {{ }} 的不算错，不为它返工。
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.main import app
+
+
+@pytest.fixture
+async def client():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+def _graph_ops(condition: str) -> list[str]:
+    """一张 while 循环的图，条件由参数给。"""
+    ops = [
+        {"op": "plan", "summary": "数一数"},
+        {"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入",
+                                    "config": {"fields": [{"name": "question"}]}}},
+        {"op": "add_node", "node": {"id": "init", "type": "transform", "label": "准备",
+                                    "config": {"mode": "expression", "expression": "[1, 2]",
+                                               "assign_to": "items"}}},
+        {"op": "add_edge", "edge": {"source": "start", "target": "init"}},
+        {"op": "add_node", "node": {"id": "lp", "type": "loop", "label": "循环",
+                                    "config": {"mode": "while", "condition": condition,
+                                               "max_iterations": 2}}},
+        {"op": "add_edge", "edge": {"source": "init", "target": "lp"}},
+        {"op": "add_node", "node": {"id": "body", "type": "transform", "label": "循环体",
+                                    "config": {"mode": "expression", "expression": "1"}}},
+        {"op": "add_edge", "edge": {"source": "lp", "target": "body", "sourceHandle": "body"}},
+        {"op": "add_edge", "edge": {"source": "body", "target": "lp"}},
+        {"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果",
+                                    "config": {"fields": [{"name": "结果", "value": "{{ vars.items }}"}]}}},
+        {"op": "add_edge", "edge": {"source": "lp", "target": "out", "sourceHandle": "done"}},
+        {"op": "done", "explanation": "数一数", "run": True},
+    ]
+    return [json.dumps(o, ensure_ascii=False) for o in ops]
+
+
+def _fix_ops(condition: str) -> list[str]:
+    return [json.dumps(o, ensure_ascii=False) for o in (
+        {"op": "plan", "summary": "改条件"},
+        {"op": "update_node", "id": "lp",
+         "config": {"mode": "while", "condition": condition, "max_iterations": 2}},
+        {"op": "done", "explanation": "改好了"},
+    )]
+
+
+class _Scripted:
+    """第一轮吐 first；收到自查请求（human 里有"上线前自查"）时吐 fix。"""
+
+    def __init__(self, first: list[str], fix: list[str]) -> None:
+        self.first, self.fix = first, fix
+        self.calls: list[str] = []
+
+    async def astream(self, messages):
+        from langchain_core.messages import AIMessageChunk
+
+        human = next(text for role, text in messages if role == "human")
+        self.calls.append(human)
+        for line in (self.fix if "上线前自查" in human else self.first):
+            yield AIMessageChunk(content=line + "\n")
+
+
+async def _generate(client, monkeypatch, model: _Scripted) -> list[dict]:
+    import app.api.copilot as copilot
+
+    async def _model(*_a, **_k):
+        return model, "mock-fast"
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    events: list[dict] = []
+    async with client.stream("POST", "/api/copilot/generate-stream",
+                             json={"instruction": "数一数"}) as r:
+        async for line in r.aiter_lines():
+            if line.startswith("data: "):
+                events.append(json.loads(line[6:]))
+    return events
+
+
+def _condition(final: dict) -> str:
+    return next(n for n in final["graph"]["nodes"] if n["id"] == "lp")["data"]["config"]["condition"]
+
+
+async def test_a_broken_condition_is_sent_back_and_fixed_before_delivery(client, monkeypatch):
+    model = _Scripted(_graph_ops("{{ vars.items | length }} > 5"), _fix_ops("len(vars.items) > 5"))
+    events = await _generate(client, monkeypatch, model)
+
+    checks = [e for e in events if e["op"] == "check"]
+    assert [c["status"] for c in checks] == ["repairing", "passed"], checks
+    assert any("len(x)" in i for i in checks[0]["issues"]), "交回去的问题里得带着能照改的提示"
+    assert "len(x)" in model.calls[1], "自查请求没把问题原文交给模型"
+
+    final = events[-1]
+    assert final["op"] == "final"
+    assert _condition(final) == "len(vars.items) > 5"
+    assert final["autorun"] is True
+    assert not [i for i in final["issues"] if i["level"] == "error"]
+
+
+async def test_when_the_fix_does_not_take_it_is_delivered_but_not_run(client, monkeypatch):
+    model = _Scripted(_graph_ops("foo(vars.items)"), _fix_ops("bar(vars.items)"))
+    events = await _generate(client, monkeypatch, model)
+
+    checks = [e for e in events if e["op"] == "check"]
+    assert [c["status"] for c in checks] == ["repairing", "repairing", "failed"], checks
+    assert len(model.calls) == 3, "最多交回去改两轮"
+    final = events[-1]
+    assert final["autorun"] is False, "明知跑不起来的图不能自动开跑"
+    assert [i for i in final["issues"] if i["level"] == "error"]
+
+
+async def test_a_clean_graph_costs_no_extra_call(client, monkeypatch):
+    model = _Scripted(_graph_ops("len(vars.items) > 5"), [])
+    events = await _generate(client, monkeypatch, model)
+    assert len(model.calls) == 1
+    assert [e["status"] for e in events if e["op"] == "check"] == ["passed"]
+    assert events[-1]["autorun"] is True
+
+
+async def test_template_braces_are_accepted_without_a_rework(client, monkeypatch):
+    """多套了 {{ }} 意思没有歧义，照样能跑——提示一句就行，不为它返工一轮。"""
+    model = _Scripted(_graph_ops("{{ vars.items }} != []"), [])
+    events = await _generate(client, monkeypatch, model)
+    assert len(model.calls) == 1
+    final = events[-1]
+    assert final["autorun"] is True
+    assert any("是多余的" in i["message"] for i in final["issues"])

backend/tests/test_expressions.py

@@ -0,0 +1,221 @@
+"""条件表达式：写错了要在画图时就说出来，而不是跑到那一步才炸。
+
+开发库里四次运行死在同一句"循环条件写错了：表达式里不允许出现 Set"。条件原文
+是 `{{ vars.gate }} != "ok"`——把模板写法带进了表达式。`{{ x }}` 在 Python 里是
+"装着一个集合的集合"，而表达式不允许集合，于是跑到循环那一步才报错，前面的步骤
+全白跑。Copilot 的提示词还写着"模板语法在任意字符串里用"，模型照做而已。
+
+这里守三件事：这种写法意思无歧义，按 vars.x 理解并提示；真写错的在校验阶段就
+拦下（运行前校验会挡住它，Copilot 自查会把它交回去改）；运行时、校验、自查用的
+是同一个 parse_expression，不会出现"校验说能跑、跑起来报错"。
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.db.base import SessionLocal
+from app.db.models import Run
+from app.engine.expressions import (
+    EXPRESSION_ROOTS,
+    ExpressionError,
+    eval_expression,
+    parse_expression,
+)
+from app.engine.runner import run_manager
+from app.engine.schema import GraphSpec, validate_graph
+from app.engine.state import template_context
+
+CTX = {"vars": {"gate": "fail", "ok": "fail", "state": {"count": 999}, "items": [1, 2]},
+       "input": {}, "nodes": {}}
+
+
+@pytest.mark.parametrize("expr, expected", [
+    # 开发库里真实死过的四条
+    ('{{ vars.gate }} != "ok"', True),
+    ("{{ vars.ok }} == 'fail'", True),
+    ("{{ vars.ok }} == 'ok'", False),
+    ("{{ vars.state.count }} <= 1000", True),
+])
+def test_template_braces_in_an_expression_mean_the_bare_path(expr, expected):
+    assert eval_expression(expr, CTX) is expected
+    _, unwrapped, _ = parse_expression(expr)
+    assert unwrapped, "剥掉了 {{ }} 就得说出来，校验据此给提示"
+
+
+def test_braces_inside_a_string_literal_are_left_alone():
+    """在语法树上剥，不在文本上剥：字符串里的花括号是字面量。"""
+    assert eval_expression("contains(vars.gate, '{') or true", CTX) is True
+
+
+@pytest.mark.parametrize("expr, hint", [
+    ("{{ vars.items | length }} > 0", "len(x)"),         # 模板过滤器
+    ("vars.gate in {'ok', 'fail'}", "用列表"),           # 集合写法
+    ("foo(vars.gate)", "能用的只有"),                    # 不认识的函数
+    ("'{{ vars.gate }}' == 'ok'", "不会被渲染"),         # 引号里的模板永远不成立
+    ("vars.gate = 'ok'", "=="),                         # 赋值当比较
+    ("[x for x in vars.items]", "列表推导式"),
+    ("vars.items[0:1]", "切片"),
+    ("vars.__class__", "私有属性"),
+])
+def test_real_mistakes_are_caught_statically_with_a_usable_hint(expr, hint):
+    with pytest.raises(ExpressionError, match=hint.replace("(", r"\(").replace(")", r"\)")):
+        parse_expression(expr)
+
+
+def test_the_short_circuited_half_is_checked_too():
+    """运行时 and 短路，右边从来走不到；静态检查得把整棵树走一遍。"""
+    assert eval_expression("false and true", CTX) is False
+    with pytest.raises(ExpressionError):
+        parse_expression("false and foo(1)")
+
+
+def test_names_outside_the_context_are_reported():
+    """漏写 vars. 的 gate 运行时是 None，条件永远不成立——不报错，但得提示。"""
+    _, _, unknown = parse_expression("gate == 'ok' and len(vars.items) > 0 and true")
+    assert unknown == ["gate"]
+
+
+def test_expression_roots_match_the_runtime_context():
+    """EXPRESSION_ROOTS 是 template_context 的键的抄本，两边一旦不一致，提示就会误报。"""
+    assert EXPRESSION_ROOTS == set(template_context({}))
+
+
+# --------------------------------------------------------------------------
+# 校验：按节点类型和模式挑字段
+# --------------------------------------------------------------------------
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0},
+            "data": {"label": nid, "config": config}}
+
+
+def issues_of(*nodes):
+    chain = [node("start", "input", fields=[]), *nodes, node("out", "output", fields=[])]
+    spec = GraphSpec.model_validate({
+        "nodes": chain,
+        "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(chain, chain[1:])],
+    })
+    return [(i.level, i.message) for i in validate_graph(spec).issues]
+
+
+def test_a_broken_loop_condition_blocks_the_run():
+    found = issues_of(node("lp", "loop", mode="while", condition="{{ vars.items | length }} > 0"))
+    assert any(level == "error" and "循环条件写错了" in msg and "len(x)" in msg for level, msg in found), found
+
+
+def test_braces_are_a_warning_not_an_error():
+    found = issues_of(
+        node("g", "transform", expression="'ok'", assign_to="gate"),
+        node("br", "branch", cases=[{"key": "yes", "condition": '{{ vars.gate }} == "ok"'}]),
+    )
+    assert not [m for lv, m in found if lv == "error"], found
+    assert any("是多余的" in m and "vars.gate" in m for _, m in found), found
+
+
+def test_fields_that_are_not_evaluated_in_this_mode_are_not_checked():
+    """foreach 循环的 condition、模板模式整形的 expression 不参与求值，报了就是误报。"""
+    found = issues_of(
+        node("lp", "loop", mode="foreach", items="[1]", condition="这不是表达式 >>>"),
+        node("t", "transform", mode="template", template="x", expression="这也不是 >>>"),
+    )
+    assert not [m for lv, m in found if "写错了" in m], found
+
+
+def test_branch_in_llm_mode_is_not_checked_either():
+    found = issues_of(node("br", "branch", mode="llm", cases=[{"key": "a", "condition": "随便写的描述"}]))
+    assert not [m for lv, m in found if "写错了" in m], found
+
+
+def test_skip_if_and_metrics_are_checked():
+    found = issues_of(node("t", "transform", expression="1", skip_if="vars.x = 1"))
+    assert any("跳过条件写错了" in m for _, m in found), found
+    found = issues_of(node("m", "metrics", metrics=[{"id": "r", "expression": "foo(1)"}]))
+    assert any("指标「r」的表达式写错了" in m for _, m in found), found
+
+
+def test_empty_while_condition_and_empty_transform_expression():
+    found = issues_of(node("lp", "loop", mode="while", condition=""))
+    assert any(lv == "warning" and "一次都不会跑" in m for lv, m in found), found
+    found = issues_of(node("t", "transform", mode="expression", expression=""))
+    assert any(lv == "error" and "没有填表达式" in m for lv, m in found), found
+
+
+# --------------------------------------------------------------------------
+# 真跑一遍：以前死在 Set 上的那种循环
+# --------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+async def test_a_loop_written_with_template_braces_now_runs(engine_up):
+    graph = {
+        "nodes": [
+            node("start", "input", fields=[{"name": "question"}]),
+            node("init", "transform", expression="0", assign_to="n"),
+            node("lp", "loop", mode="while", condition="{{ vars.n }} < 2", max_iterations=5),
+            node("inc", "transform", expression="vars.n + 1", assign_to="n"),
+            node("out", "output", fields=[{"name": "结果", "value": "{{ vars.n }}"}]),
+        ],
+        "edges": [
+            {"source": "start", "target": "init"},
+            {"source": "init", "target": "lp"},
+            {"source": "lp", "target": "inc", "sourceHandle": "body"},
+            {"source": "inc", "target": "lp"},
+            {"source": "lp", "target": "out", "sourceHandle": "done"},
+        ],
+    }
+    run = await run_manager.start(graph=graph, input_payload={"question": "x"})
+    for _ in range(300):
+        async with SessionLocal() as session:
+            row = await session.get(Run, run.id)
+        if row.status in ("succeeded", "failed"):
+            break
+        await asyncio.sleep(0.05)
+    assert row.status == "succeeded", row.error
+    assert row.output["结果"] == "2"
+
+
+# --------------------------------------------------------------------------
+# 代码节点：assign_to 拿的是 stdout，结果得 print 出来
+# --------------------------------------------------------------------------
+
+
+def _code(code: str, **cfg):
+    return node("c", "code", code=code, **{"assign_to": "state", **cfg})
+
+
+def test_code_that_hands_over_nothing_is_caught():
+    """开发库里的「测试 1」：条件修好之后跑通了，结果却是空的——代码最后一行是裸表达式。"""
+    found = issues_of(_code('count = 1\n{"count": count}'))
+    assert any(lv == "error" and "没有 print" in m and "vars.state" in m for lv, m in found), found
+    # 同一张图的循环体：模板只是填进一个值，代码本身照样没 print
+    found = issues_of(_code('state = {{ vars.state | json }}\nstate["count"] += 1\nstate'))
+    assert any("没有 print" in m for _, m in found), found
+    # 沙箱把空的和大写的 language 都当 python 跑，这里也得这么认
+    for language in ("", "Python"):
+        found = issues_of(_code('{"count": 1}', language=language))
+        assert any("没有 print" in m for _, m in found), (language, found)
+
+
+@pytest.mark.parametrize("code, cfg", [
+    ('import json\nprint(json.dumps({"count": 1}))', {}),
+    ("import sys\nsys.stdout.write('ok')", {}),
+    ("from sys import stdout\nstdout.write('ok')", {}),
+    ("import subprocess\nsubprocess.run(['echo', 'ok'])", {}),   # 子进程写的也是 stdout
+    ("import main", {"files": {"main.py": "print('ok')"}}),     # print 在附带文件里
+    ("{{ vars.code }}", {}),                  # ⑤号示例：整段代码是上游模型现写的，看不见
+    ("exec({{ vars.code | json }})", {}),
+    ("x = 1", {"assign_to": ""}),            # 不交给下游就不需要输出
+    ("ls -la", {"language": "bash"}),        # bash 的命令本身就会输出，不按 print 判
+])
+def test_code_that_does_hand_something_over_passes(code, cfg):
+    found = issues_of(_code(code, **cfg))
+    assert not [m for _, m in found if "没有 print" in m], found

frontend/src/run/Composer.tsx

@@ -284,4 +284,5 @@ export const PHASE_TEXT: Record<string, string> = {
   building: '正在放置节点…',
   wiring: '正在连接数据流…',
   finalizing: '正在排版和校验…',
+  repairing: '自查发现问题，正在修正…',
 }

frontend/src/run/decode.ts

@@ -1029,6 +1029,7 @@ const PHASE_LABEL: Record<string, string> = {
   building: '正在搭建流程',
   wiring: '正在连接数据流',
   finalizing: '正在排版和校验',
+  repairing: '正在按自查结果修正',
 }
 
 /**
@@ -1117,13 +1118,39 @@ export function decodeCopilot(ops: CopilotOp[]): Step[] {
                    title: `流程搭好了，加了 ${nodeCount} 步`,
                    detail: String(op.explanation ?? '') || undefined })
         break
+      case 'check': {
+        // 服务端自查：用运行时同一套规则过一遍，有问题交回模型改。
+        // 这件事得看得见——否则多出来的那十几秒像是卡住了，改了什么也无从知道
+        const issues = Array.isArray(op.issues) ? op.issues.map(String) : []
+        if (op.status === 'repairing') {
+          out.push({ id: `ck-${i}`, seq: i, kind: 'note', status: 'done',
+                     title: `自查发现 ${issues.length} 处问题，交回去改`,
+                     detail: issues.join('\n') || undefined })
+        } else if (op.status === 'passed') {
+          closeLifecycles(out, 'done')
+          out.push({ id: `ck-${i}`, seq: i, kind: 'note', status: 'done',
+                     title: op.repaired ? '自查通过：问题已经改好' : '自查通过' })
+        } else {
+          closeLifecycles(out, 'failed')
+          out.push({ id: `ck-${i}`, seq: i, kind: 'error', level: 'error', status: 'failed',
+                     title: op.status === 'failed'
+                       ? `自查后还有 ${issues.length} 处问题，没有自动运行`
+                       : String(op.message ?? '自查没能完成'),
+                     detail: issues.join('\n') || undefined })
+        }
+        break
+      }
       case 'final': {
         // 后端排版校验后的最终图才知道整张图有几步。nodeCount 只是这一轮
-        // 新增的数量——在"改图"场景下说"共 2 步"是错的，图上明明有四个节点
+        // 新增的数量——在"改图"场景下说"共 2 步"是错的，图上明明有四个节点。
+        // 自查的记录会排在"搭好了"后面，所以往回找那一行，不能只看最后一条
         const total = op.graph?.nodes?.length
         const last = out[out.length - 1]
-        if (total && last?.kind === 'lifecycle') {
-          last.title = nodeCount && nodeCount < total
+        // 流没等到 done 就断了时没有那一行，沿用原来的做法：改写最后一条生命周期
+        const built = [...out].reverse().find((s) => s.id.startsWith('cd-'))
+          ?? (last?.kind === 'lifecycle' ? last : undefined)
+        if (total && built) {
+          built.title = nodeCount && nodeCount < total
             ? `流程搭好了，加了 ${nodeCount} 步，整张图共 ${total} 步`
             : `流程搭好了，共 ${total} 步`
         }

frontend/src/store/studio.ts

@@ -156,7 +156,7 @@ interface StudioState {
     error: string
     model: string
     // 从提交到第一个操作之间有 5~30 秒，得让用户看见这段在发生什么
-    phase: 'connecting' | 'planning' | 'building' | 'wiring' | 'finalizing' | ''
+    phase: 'connecting' | 'planning' | 'building' | 'wiring' | 'finalizing' | 'repairing' | ''
     thinking: string        // 模型的思考原文（只有支持 thinking 的模型有）
     elapsedMs: number
     lastInstruction: string // 失败后"用同一需求重试"要用
@@ -560,6 +560,14 @@ export const useStudio = create<StudioState>((set, get) => ({
           case 'plan':
             set({ copilot: { ...s.copilot, phase: 'planning', lastOp: op.summary ?? '规划中' } })
             break
+          case 'check':
+            // 服务端自查：用运行时同一套规则过一遍，有问题交回模型改
+            set({ copilot: { ...s.copilot,
+              phase: op.status === 'repairing' ? 'repairing' : s.copilot.phase,
+              lastOp: op.status === 'repairing'
+                ? `自查发现 ${(op.issues ?? []).length} 处问题，正在修正`
+                : op.status === 'passed' ? '自查通过' : '自查后仍有问题，先别急着运行' } })
+            break
           case 'add_node': {
             const n = op.node
             // 类型不认识就不放上画布：节点卡片、属性面板都按类型查定义，查不到

scripts/check-decode.mjs

@@ -354,6 +354,37 @@ console.log('\n=== Copilot 操作流 ===')
   check('从零新建时不啰嗦，只说共几步',
     built[built.length - 1].title === '流程搭好了，共 2 步',
     built[built.length - 1].title)
+
+  // 自查：服务端用运行时同一套规则过一遍，有问题交回模型改。多出来的这段得看得见，
+  // 而且它排在"搭好了"后面，不能把那一行的"共几步"挤没了
+  const checked = mod.decodeCopilot([
+    { op: 'add_node', node: { id: 'a', type: 'input' } },
+    { op: 'add_node', node: { id: 'lp', type: 'loop' } },
+    { op: 'done', explanation: '' },
+    { op: 'check', status: 'repairing', round: 1, issues: ['「lp」循环条件写错了：表达式里没有 | 过滤器'] },
+    { op: 'heartbeat', phase: 'repairing', elapsed_ms: 3000 },
+    { op: 'update_node', id: 'lp' },
+    { op: 'check', status: 'passed', repaired: 1 },
+    { op: 'final', graph: { nodes: [{ id: 'a' }, { id: 'lp' }] } },
+  ])
+  const titles = checked.map((s) => s.title)
+  check('自查发现问题要说出来', titles.some((t) => t.includes('自查发现 1 处问题')), titles.join(' | '))
+  check('交回去改的是哪条要看得到',
+    checked.some((s) => (s.detail ?? '').includes('循环条件写错了')))
+  check('改好了要说', titles.includes('自查通过：问题已经改好'), titles.join(' | '))
+  check('自查插在后面也不挤掉"共几步"', titles.includes('流程搭好了，共 2 步'), titles.join(' | '))
+  check('自查结束后没有还在转的行', !checked.some((s) => s.status === 'running'),
+    checked.filter((s) => s.status === 'running').map((s) => s.title).join(','))
+
+  const stuck = mod.decodeCopilot([
+    { op: 'done', explanation: '' },
+    { op: 'check', status: 'failed', issues: ['「lp」循环条件写错了', '「t」整形节点没有填表达式'] },
+    { op: 'final', graph: { nodes: [{ id: 'lp' }, { id: 't' }] } },
+  ])
+  const failedCheck = stuck.find((s) => s.kind === 'error')
+  check('改不好要明说、说清没有自动运行',
+    !!failedCheck?.title.includes('还有 2 处问题') && failedCheck.title.includes('没有自动运行'),
+    failedCheck?.title)
 }
 
 console.log(failed ? `\n✗ ${failed} 项未通过` : '\n✓ 解码器全部通过')