数据目录阶段 2B：发布门禁接上 SQL 检查——受管级别被 error 挡住、warning 进提示，发布前检查、自动修复和发布同一个口径

backend/app/api/copilot.py

@@ -1109,16 +1109,32 @@ def _blocking_issues(
                                          **({"defaults": defaults} if isinstance(defaults, dict) else {})})
     except Exception:  # noqa: BLE001 - 结构不合法留给收尾那一步报
         return None
+    # 带发布级别时 SQL 检查交给门禁（查全图、按级别定挡不挡），自查这边就不再查一遍，免得同一处报两条
+    authored = authored_issues(spec, sources or [], baseline, checkers=None if level else checkers)
     out = ([i.model_dump() for i in validate_graph(spec).issues if i.level == "error"]
            + _issue_dicts(scope_issues(list(nodes.values()), scope), "datasource_out_of_scope")
-           + [i for i in authored_issues(spec, sources or [], baseline, checkers=checkers) if i["level"] == "error"])
+           + [i for i in authored if i["level"] == "error"])
     if level:
         from app.engine.governance import lint_for_publish
 
-        out += [i.model_dump() for i in lint_for_publish(spec, level=level).issues if i.level == "error"]
+        gate = [i.model_dump() for i in lint_for_publish(spec, level=level, checkers=checkers).issues
+                if i.level == "error"]
+        out += _sql_model_hints(gate, spec, checkers)
     return out
 
 
+def _sql_model_hints(issues: list[dict[str, Any]], spec: GraphSpec, checkers: dict[str, Any] | None,
+                     ) -> list[dict[str, Any]]:
+    """门禁报的 SQL 问题只有给人看的那句（ValidationIssue 不带 for_model）；交回模型改时补上给模型的改法。"""
+    if not checkers or not any(i.get("code") in CHECK_CODES for i in issues):
+        return issues
+    from app.data.sqlcheck import graph_checks
+
+    hints = {(node.id, c.code, c.message): c.for_model for node, c in graph_checks(spec.nodes, checkers)}
+    return [{**i, "for_model": hints[key]} if (key := (i.get("node_id"), i.get("code"), i.get("message"))) in hints
+            else i for i in issues]
+
+
 def _changed(spec: GraphSpec, baseline: list[dict[str, Any]] | None) -> set[str]:
     """这一轮新加或改过配置的节点。改图时只查这些：用户自己写的、这一轮没碰的 SQL 不替他改。"""
     before = {n.get("id"): (n.get("data") or {}).get("config") for n in baseline or [] if isinstance(n, dict)}
@@ -1887,13 +1903,17 @@ async def assist_publish_fix(
     """
     from app.engine.autofix import forbidden_changes, judge
 
+    from app.data.sqlcheck import load_checkers
+
     sources = await _sources(session, None)
+    # 和发布接口同一套 SQL 检查：受管级别里 SQL 的 error 也是挡住发布的问题，一并交给助手改
+    checkers = await load_checkers(session, sources)
     base = copy.deepcopy(graph)
     before_nodes = {str(n["id"]): n for n in copy.deepcopy(base.get("nodes") or []) if isinstance(n, dict)}
     before_edges = [e for e in copy.deepcopy(base.get("edges") or []) if isinstance(e, dict)]
     baseline = list(copy.deepcopy(before_nodes).values())
     errors = _blocking_issues(before_nodes, before_edges, sources=sources, baseline=baseline, level=level,
-                              defaults=base.get("defaults")) or []
+                              defaults=base.get("defaults"), checkers=checkers) or []
     out: dict[str, Any] = {"accepted": False, "reason": None, "summary": "", "questions": [], "graph": graph,
                            "ops": []}
     if not errors:
@@ -1949,7 +1969,7 @@ async def assist_publish_fix(
     proposed = _proposed_graph(base, nodes, edges)
     reasons = forbidden_changes(base, proposed, tried)
     after = _blocking_issues(_nodes_by_id(proposed), proposed.get("edges") or [], sources=sources,
-                             baseline=baseline, level=level, defaults=proposed.get("defaults"))
+                             baseline=baseline, level=level, defaults=proposed.get("defaults"), checkers=checkers)
     if reasons:
         out["reason"] = "助手的修改降低了要求，已作废：" + "；".join(reasons)
     elif after is None:
@@ -2217,7 +2237,11 @@ async def upgrade_evidence(payload: UpgradeIn, session: AsyncSession = Depends(g
             out["notes"] = [n for n in out["notes"] if n["rule"] != "R5" or n["node_id"] in still]
         elif helped["reason"]:
             out["rejected"].append({"fix_id": "assist", "reason": helped["reason"]})
-    issues = publish_issues(GraphSpec.model_validate(out["graph"]), level=payload.level)
+    from app.data.sqlcheck import graph_sources, load_checkers_by_name
+
+    upgraded = GraphSpec.model_validate(out["graph"])
+    checkers = await load_checkers_by_name(session, graph_sources(upgraded.nodes))
+    issues = publish_issues(upgraded, level=payload.level, checkers=checkers)
     out["issues"] = [i.model_dump() for i in issues]
     out["ok"] = not any(i.level == "error" for i in issues)
     return out

backend/app/api/workflows.py

@@ -324,7 +324,8 @@ async def publish_workflow(
     except Exception as e:  # noqa: BLE001
         return {"ok": False, "issues": [{"level": "error", "message": f"无法解析工作流结构：{graph_error(e)}"}]}
 
-    issues = [i.model_dump() for i in publish_issues(spec, level=payload.level)]
+    checkers = await _sql_checkers(session, spec)
+    issues = [i.model_dump() for i in publish_issues(spec, level=payload.level, checkers=checkers)]
     if any(i["level"] == "error" for i in issues):
         return {"ok": False, "level": payload.level, "version": version, "issues": issues}
 
@@ -400,6 +401,14 @@ async def _gate_context(session: AsyncSession, spec: GraphSpec) -> tuple[dict[st
     return versions, cards
 
 
+async def _sql_checkers(session: AsyncSession, spec: GraphSpec) -> dict[str, Any]:
+    """图里调用工具节点写死的 SQL 用到的数据源，各建一个基于数据目录的 SQL 检查器。发布、发布前检查、
+    自动修复都传这一份给门禁（governance.lint_for_publish）：门禁本身不查库，三处口径才一致。"""
+    from app.data.sqlcheck import graph_sources, load_checkers_by_name
+
+    return await load_checkers_by_name(session, graph_sources(spec.nodes))
+
+
 def _unreadable(level: str, e: Exception) -> dict[str, Any]:
     return {"level": level, "ok": False, "fixes": [], "issues": [
         {"level": "error", "node_id": None, "edge_id": None, "field": None, "code": None, "fix": None,
@@ -422,7 +431,8 @@ async def publish_check(
     except Exception as e:  # noqa: BLE001
         return _unreadable(payload.level, e)
     versions, cards = await _gate_context(session, spec)
-    return check(spec, level=payload.level, versions=versions, cards=cards)
+    return check(spec, level=payload.level, versions=versions, cards=cards,
+                 checkers=await _sql_checkers(session, spec))
 
 
 @router.post("/{workflow_id}/autofix")
@@ -446,7 +456,9 @@ async def autofix(
             {"fix_id": fid, "reason": "无法解析工作流结构，不能应用修复"} for fid in payload.apply],
             "remaining": bad["issues"], "fixes": [], "handoff": [], "assist": None, "ok": False}
     versions, cards = await _gate_context(session, spec)
-    out = apply_fixes(graph, payload.apply, payload.choices, level=payload.level, versions=versions, cards=cards)
+    checkers = await _sql_checkers(session, spec)
+    out = apply_fixes(graph, payload.apply, payload.choices, level=payload.level, versions=versions, cards=cards,
+                      checkers=checkers)
     out["assist"] = None
     # 人在选项里选了「交给 Copilot」（handoff），和点「交给 Copilot」是一回事
     if payload.assist or out["handoff"]:
@@ -464,7 +476,8 @@ async def autofix(
             out["changes"] += diff_changes(out["graph"], fixed, fix_id="assist", label="助手的修改")
             out["ops"] += helped["ops"]
             out["applied"].append("assist")
-            after = apply_fixes(fixed, [], level=payload.level, versions=versions, cards=cards)
+            after = apply_fixes(fixed, [], level=payload.level, versions=versions, cards=cards,
+                                checkers=await _sql_checkers(session, GraphSpec.model_validate(fixed)))
             out.update(graph=fixed, remaining=after["remaining"], fixes=after["fixes"], ok=after["ok"])
         elif helped["reason"]:
             out["rejected"].append({"fix_id": "assist", "reason": helped["reason"]})

backend/app/engine/autofix.py

@@ -559,9 +559,13 @@ def annotate(issues: list[Any], fixes: list[dict[str, Any]]) -> list[dict[str, A
 
 
 def check(spec: GraphSpec, *, level: str, versions: dict[str, Any] | None = None,
-          cards: dict[str, list[dict[str, Any]]] | None = None) -> dict[str, Any]:
-    """发布前检查：和真正发布同一套口径的问题，以及能怎么修。只读。"""
-    issues = publish_issues(spec, level=level)
+          cards: dict[str, list[dict[str, Any]]] | None = None,
+          checkers: dict[str, Any] | None = None) -> dict[str, Any]:
+    """发布前检查：和真正发布同一套口径的问题，以及能怎么修。只读。
+
+    checkers：基于数据目录的 SQL 检查器（data/sqlcheck），和发布接口传的是同一份（governance.lint_for_publish）。
+    """
+    issues = publish_issues(spec, level=level, checkers=checkers)
     fixes = plan_fixes(spec, issues, level=level, versions=versions, cards=cards)
     return {"level": level, "ok": not any(i.level == "error" for i in issues),
             "issues": annotate(issues, fixes), "fixes": [public(f) for f in fixes]}
@@ -702,7 +706,8 @@ def _op(graph: dict[str, Any], node_id: str | None, edits: list[tuple[str | None
 
 def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices: dict[str, Any] | None = None, *,
                 level: str, versions: dict[str, Any] | None = None,
-                cards: dict[str, list[dict[str, Any]]] | None = None) -> dict[str, Any]:
+                cards: dict[str, list[dict[str, Any]]] | None = None,
+                checkers: dict[str, Any] | None = None) -> dict[str, Any]:
     """在副本上逐个应用修复，每一个都复核：降低要求、错误没变少或者冒出新错误的，丢弃并写明原因。
 
     返回 {graph, changes, ops, applied, rejected, handoff, remaining, fixes, ok}。graph 只是预览，不落库。
@@ -711,7 +716,7 @@ def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices:
     raw = graph.model_dump(mode="json") if isinstance(graph, GraphSpec) else graph
     current = copy.deepcopy(raw)
     spec = GraphSpec.model_validate(current)
-    issues = publish_issues(spec, level=level)
+    issues = publish_issues(spec, level=level, checkers=checkers)
     plan = plan_fixes(spec, issues, level=level, versions=versions, cards=cards)
     known = {f["id"] for f in plan}
     rank = {code: i for i, code in enumerate(_ORDER)}
@@ -743,7 +748,7 @@ def apply_fixes(graph: dict[str, Any] | GraphSpec, fix_ids: list[str], choices:
         except Exception:  # noqa: BLE001 - 修复规则把图改坏了：丢弃，不交给人一张存不回去的图
             rejected.append({"fix_id": fid, "reason": "修复后的工作流结构无法解析，已放弃这项修复"})
             continue
-        trial_issues = publish_issues(trial_spec, level=level)
+        trial_issues = publish_issues(trial_spec, level=level, checkers=checkers)
         # choice 里人亲手选的值不算「替人决定」（比如把沙箱代码标成取数）；别的降低要求照样拦
         chosen = {(str(nid), path) for nid, path, _, _ in edits} if fix["kind"] == "choice" else set()
         why = "；".join(forbidden_changes(current, trial, chosen=chosen)) or judge(

backend/app/engine/governance.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import re
 from collections import defaultdict
+from collections.abc import Mapping
 from typing import Any, Callable
 
 from sqlalchemy import func, select
@@ -29,7 +30,7 @@ UPGRADE_POLICIES = {"recompute", "dual", "incomparable"}
 UPGRADE_POLICY_LABELS = {k: option_label("upgrade_policy", k) for k in ("recompute", "dual", "incomparable")}
 
 
-def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult:
+def lint_for_publish(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] | None = None) -> ValidationResult:
     """发布门禁。
 
     published：基础可运行性之外只给警告。
@@ -38,6 +39,9 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult:
 
     每条问题带机器编号（code）：发布前自动修复（engine/autofix.py）按它认问题、出修复，
     文案可以改，编号不能随便改。证据门禁 G1–G5 见 _lint_evidence。
+
+    checkers 是各数据源基于数据目录的 SQL 检查器（data/sqlcheck.load_checkers_by_name），由调用方读好传进来：
+    门禁本身不查库。给了的话，调用工具节点里写死的 SQL 对照目录查一遍（_lint_sql）。
     """
     result = ValidationResult()
     strict = level == "governed"
@@ -100,6 +104,7 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult:
                      "（那里会保存取数快照）", code="governed.code_network", node_id=node.id, field="network")
 
     _lint_evidence(spec, flag, strict=strict)
+    _lint_sql(spec, flag, checkers)
 
     if strict and not has_contract_output:
         result.add(
@@ -109,15 +114,34 @@ def lint_for_publish(spec: GraphSpec, *, level: str) -> ValidationResult:
     return result
 
 
-def publish_issues(spec: GraphSpec, *, level: str) -> list[Any]:
+def publish_issues(spec: GraphSpec, *, level: str, checkers: Mapping[str, Any] | None = None) -> list[Any]:
     """发布时拦的全部问题：validate_graph 和 lint_for_publish 的并集，顺序也照发布接口。
 
-    发布、发布前检查、自动修复复核都走这一个口径——检查说能发的，真发布时不会被拦。
+    发布、发布前检查、自动修复复核都走这一个口径——检查说能发的，真发布时不会被拦。checkers 见 lint_for_publish：
+    这几处都要传同一份，否则检查和发布对 SQL 的判断不一致。
     """
-    gate = lint_for_publish(spec, level=level).issues
+    gate = lint_for_publish(spec, level=level, checkers=checkers).issues
     return [*_not_repeated(validate_graph(spec).issues, gate, spec), *gate]
 
 
+def _lint_sql(spec: GraphSpec, flag: Callable[..., None], checkers: Mapping[str, Any] | None) -> None:
+    """调用工具节点里写死的 SQL 对照数据目录检查（data/sqlcheck.py）。
+
+    error 级（依据有确证的重复计算、存量跨期求和）按硬性问题报：受管级别挡住发布，已发布级别只警告——和别的
+    受管要求同一个口径。warning 两档都只提示。info 的依据只是推断（模型起草、命名推断），不进门禁：发布时
+    摆一排「推断，未确认」的提示，只会让人学会无视门禁。编号就是规则编号（fanout_sum…），没有确定性的修复，
+    改 SQL 交给人或助手。Agent 节点的 SQL 要到运行时才写出来，不在这里查。
+    """
+    if not checkers:
+        return
+    from app.data.sqlcheck import graph_checks
+
+    for node, check in graph_checks(spec.nodes, checkers):
+        if check.level == "info":
+            continue
+        flag(check.message, code=check.code, node_id=node.id, hard=check.level == "error", field="args.sql")
+
+
 #: validate 里和门禁 G4、G5 说同一件事的两条提示（schema.evidence_issues）
 _REPEATS = frozenset({"evidence.caliber_compute_input", "evidence.cite_fields_off"})
 

backend/tests/test_publish_sqlcheck.py

@@ -0,0 +1,212 @@
+"""发布门禁接上基于数据目录的 SQL 检查（data/sqlcheck.py）。
+
+- 受管级别：调用工具节点里写死的 SQL 有 error 级问题（一对多关联后重复计算、存量跨期求和）就挡住发布；
+- 已发布级别：同样的问题只是警告，不挡；
+- warning 级的问题两档都进提示，info（依据只是推断）不进门禁；
+- 发布前检查、自动修复和真正发布同一个口径：检查说能发的，真发布时不会被拦。
+"""
+from __future__ import annotations
+
+import copy
+import json
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.data import catalog
+from app.data.engine import engines
+from app.data.introspect import introspect
+from app.data.sqlcheck import SqlChecker
+from app.engine.governance import lint_for_publish, publish_issues
+from app.engine.schema import GraphSpec
+from app.main import app
+from tests.fixtures.catalog import scenic_notes
+
+FANOUT_SQL = ("SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
+              "WHERE o.status = 1")
+CLEAN_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1"
+WARN_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id"
+
+_CACHE: dict[str, dict] = {}
+
+
+async def _schema(path: str) -> dict:
+    if path not in _CACHE:
+        probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path,
+                                host=None, port=None, username=None, password=None, options={}, readonly=True,
+                                description="", schema_cache={}, origin="manual")
+        try:
+            _CACHE[path] = await introspect(probe)
+        finally:
+            await engines.invalidate(probe.id)
+    return _CACHE[path]
+
+
+def node(nid, ntype, label=None, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 10, "y": 20}, "data": {"label": label or nid, "config": config}}
+
+
+def weekly(source: str, sql: str) -> dict:
+    """input → 取数（调用工具，写死的 SQL）→ 口径卡 → 报告撰写 → 出具，其余受管要求都满足。"""
+    nodes = [
+        node("start", "input", "周期"),
+        node("fetch", "tool", "取数", tool=f"db_query__{source}", args={"sql": sql}),
+        node("card", "metrics", "周报口径卡", caliber="周报口径", metrics=[
+            {"id": "gmv", "name": "订单金额", "unit": "元", "expression": "cell(nodes.fetch, 0, 'gmv')"}]),
+        node("write", "report", "报告撰写", instructions="写周报",
+             numbers="strict", on_violation="fail", claims="require_citation"),
+        node("done", "output", "出具", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
+             contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"], "strict": True}),
+    ]
+    edges = [{"id": f"e{i}", "source": a["id"], "target": b["id"]} for i, (a, b) in enumerate(zip(nodes, nodes[1:]))]
+    return {"nodes": nodes, "edges": edges, "defaults": {"approval": "dangerous"}}
+
+
+async def _checkers(scenic_db, status="confirmed"):
+    cache = await _schema(scenic_db)
+    return {"scenic": SqlChecker(kind="sqlite", schema_cache=cache, notes=scenic_notes.notes(status),
+                                 source_name="scenic")}
+
+
+def _sql_issues(issues):
+    return [(i.code, i.level, i.node_id, i.field) for i in issues if i.field == "args.sql"]
+
+
+async def test_governed_level_blocks_error_level_sql_problems(scenic_db):
+    checkers = await _checkers(scenic_db)
+    spec = GraphSpec.model_validate(weekly("scenic", FANOUT_SQL))
+    assert not [i for i in publish_issues(spec, level="governed") if i.level == "error"], "其余受管要求都满足"
+    issues = publish_issues(spec, level="governed", checkers=checkers)
+    assert _sql_issues(issues) == [("fanout_sum", "error", "fetch", "args.sql")]
+    [issue] = [i for i in issues if i.code == "fanout_sum"]
+    assert issue.message.startswith("「订单」关联「订单明细」是一对多")
+    # 已发布级别：同一个问题只警告
+    assert _sql_issues(publish_issues(spec, level="published", checkers=checkers)) == [
+        ("fanout_sum", "warning", "fetch", "args.sql")]
+    # 依据只是推断：info 不进门禁
+    assert _sql_issues(lint_for_publish(spec, level="governed", checkers=await _checkers(scenic_db, "proposed"))
+                       .issues) == []
+    # 没有检查器（数据源没有目录）：和以前一样
+    assert _sql_issues(lint_for_publish(spec, level="governed").issues) == []
+
+
+async def test_warnings_are_hints_on_both_levels(scenic_db):
+    checkers = await _checkers(scenic_db)
+    spec = GraphSpec.model_validate(weekly("scenic", WARN_SQL))
+    for level in ("governed", "published"):
+        assert _sql_issues(publish_issues(spec, level=level, checkers=checkers)) == [
+            ("missing_valid_filter", "warning", "fetch", "args.sql")]
+    clean = GraphSpec.model_validate(weekly("scenic", CLEAN_SQL))
+    assert _sql_issues(publish_issues(clean, level="governed", checkers=checkers)) == []
+
+
+async def test_autofix_check_and_apply_use_the_same_gate(scenic_db):
+    from app.engine.autofix import apply_fixes, check
+
+    checkers = await _checkers(scenic_db)
+    graph = weekly("scenic", FANOUT_SQL)
+    body = check(GraphSpec.model_validate(graph), level="governed", checkers=checkers)
+    assert body["ok"] is False and "fanout_sum" in {i["code"] for i in body["issues"]}
+    out = apply_fixes(copy.deepcopy(graph), [], level="governed", checkers=checkers)
+    assert out["ok"] is False and "fanout_sum" in {i["code"] for i in out["remaining"]}
+
+
+# --------------------------------------------------------------------------
+# 接口：发布、发布前检查、自动修复
+# --------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def scenic_source(scenic_db):
+    from app.db.base import SessionLocal
+    from app.db.models import DataSource
+
+    name = f"scenic_{uuid.uuid4().hex[:8]}"
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=scenic_db, readonly=True)
+        row.schema_cache = await _schema(scenic_db)
+        session.add(row)
+        await session.commit()
+        for table, notes in scenic_notes.notes().items():
+            await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
+        source_id = row.id
+    yield name
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, source_id)
+        row.enabled = False
+        await session.commit()
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
+        yield c
+
+
+async def test_publish_endpoints_block_governed_fanout(client, scenic_source):
+    r = await client.post("/api/workflows", json={"name": f"订单周报-{scenic_source}",
+                                                  "graph": weekly(scenic_source, FANOUT_SQL)})
+    assert r.status_code == 201, r.text
+    wid = r.json()["id"]
+
+    checked = (await client.post(f"/api/workflows/{wid}/publish-check", json={"level": "governed"})).json()
+    assert checked["ok"] is False
+    [issue] = [i for i in checked["issues"] if i["code"] == "fanout_sum"]
+    assert issue["level"] == "error" and issue["node_id"] == "fetch" and issue["field"] == "args.sql"
+    assert issue["fix"] is None, "SQL 写错没有确定性的修复，交给人或助手改"
+
+    fixed = (await client.post(f"/api/workflows/{wid}/autofix", json={"level": "governed", "apply": []})).json()
+    assert fixed["ok"] is False and "fanout_sum" in {i["code"] for i in fixed["remaining"]}
+
+    blocked = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "governed"})).json()
+    assert blocked["ok"] is False and "fanout_sum" in {i.get("code") for i in blocked["issues"]}
+    # 已发布级别不挡，问题作为警告带回
+    published = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "published"})).json()
+    assert published["ok"] is True
+    assert [i["level"] for i in published["issues"] if i.get("code") == "fanout_sum"] == ["warning"]
+
+    # 改好 SQL：受管级别发得出去，剩下的 warning 进提示
+    await client.patch(f"/api/workflows/{wid}", json={"graph": weekly(scenic_source, WARN_SQL)})
+    ok = (await client.post(f"/api/workflows/{wid}/publish", json={"level": "governed"})).json()
+    assert ok["ok"] is True, ok["issues"]
+    assert [i["code"] for i in ok["issues"] if i.get("field") == "args.sql"] == ["missing_valid_filter"]
+
+
+class _Scripted:
+    """按剧本吐操作流的模型；记下每次收到的请求。"""
+
+    def __init__(self, ops):
+        self.ops, self.calls = ops, []
+
+    async def astream(self, messages):
+        from langchain_core.messages import AIMessageChunk
+
+        self.calls.append(messages)
+        for op in self.ops:
+            yield AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n")
+
+
+async def test_publish_fix_assistant_gets_the_sql_error_and_its_fix_is_adopted(client, scenic_source, monkeypatch):
+    """确定性修复修不了 SQL：交给助手。助手看到的是给模型的那句改法，改好的 SQL 复核通过才采纳。"""
+    import app.api.copilot as copilot
+
+    model = _Scripted([{"op": "plan", "summary": "改 SQL"},
+                       {"op": "update_node", "id": "fetch",
+                        "config": {"tool": f"db_query__{scenic_source}", "args": {"sql": CLEAN_SQL}}},
+                       {"op": "done", "explanation": "改为对明细金额求和"}])
+
+    async def _model(*_a, **_k):
+        return model, "mock-fast"
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    wid = (await client.post("/api/workflows", json={"name": f"订单周报-助手-{scenic_source}",
+                                                     "graph": weekly(scenic_source, FANOUT_SQL)})).json()["id"]
+    out = (await client.post(f"/api/workflows/{wid}/autofix",
+                             json={"level": "governed", "apply": [], "assist": True})).json()
+    human = next(text for role, text in model.calls[0] if role == "human")
+    assert "orders 关联 order_items 是一对多" in human
+    assert out["assist"]["ok"] is True and out["applied"] == ["assist"] and out["ok"] is True, out
+    sql = next(n for n in out["graph"]["nodes"] if n["id"] == "fetch")["data"]["config"]["args"]["sql"]
+    assert sql == CLEAN_SQL