diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py
index 7ef959d0..f253d126 100644
--- a/backend/app/api/copilot.py
+++ b/backend/app/api/copilot.py
@@ -2251,13 +2251,18 @@ async def assist_upgrade(
     if not feeders:
         out["summary"] = "口径卡的输入均不来自负责计算的沙箱代码，无需交给助手改写"
         return out
+    from app.data.sqlcheck import load_checkers
+
     sources = await _sources(session, None)
+    # 基于数据目录的 SQL 检查器：自查和最终判断都带上，和发布门禁一个口径（受管级别下一对多关联后求和这类 error
+    # 会挡发布）。不带的话，助手改出这种 SQL 照样被采纳，交出去的图要到发布时才被拦
+    checkers = await load_checkers(session, sources)
     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())
     defaults = base.get("defaults")
     base_errors = _blocking_issues(before_nodes, before_edges, sources=sources, baseline=baseline, level=level,
-                                   defaults=defaults) or []
+                                   defaults=defaults, checkers=checkers) or []
     known = {_sig(e) for e in base_errors}
     out["warnings"] = _kept_code(spec)
     request = GenerateIn(instruction="升级为可追溯结构", provider=provider, model=model)
@@ -2309,7 +2314,8 @@ async def assist_upgrade(
         await collect(_upgrade_assist_request(base, spec, feeders), first=True)
         # 自查：和搭图同一套规则，只把这一轮冒出来的 error 交回去改（原来就有的不归这一轮管）
         for _ in range(_SELF_CHECK_ROUNDS):
-            now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults)
+            now = _blocking_issues(nodes, edges, sources=sources, baseline=baseline, level=level, defaults=defaults,
+                                   checkers=checkers)
             fresh = [e for e in now or [] if _sig(e) not in known]
             if not effective or not fresh:
                 break
@@ -2326,7 +2332,7 @@ async def assist_upgrade(
     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:
diff --git a/backend/tests/test_copilot_sqlcheck.py b/backend/tests/test_copilot_sqlcheck.py
index efdd24df..2a6cb072 100644
--- a/backend/tests/test_copilot_sqlcheck.py
+++ b/backend/tests/test_copilot_sqlcheck.py
@@ -241,3 +241,54 @@ async def test_clean_sql_costs_no_extra_call(scenic_source, monkeypatch):
     final = events[-1]
     assert final["autorun"] is True
     assert not [i for i in final["issues"] if i.get("code") in ("fanout_sum", "missing_valid_filter")]
+
+
+# --------------------------------------------------------------------------
+# 一键升级的语义层（assist_upgrade）：自查和最终判断同样带 SQL 检查器，和发布门禁一个口径
+# --------------------------------------------------------------------------
+
+
+def _upgrade_graph(source: str) -> dict:
+    """查库 → 沙箱代码算比率 → 口径卡读代码的产出 → 报告撰写 → 出口。查库的 SQL 本身没问题。"""
+    def n(nid, ntype, x, **config):
+        return {"id": nid, "type": ntype, "position": {"x": x, "y": 80}, "data": {"label": nid, "config": config}}
+
+    nodes = [
+        n("start", "input", 0),
+        n("fetch", "tool", 280, tool=f"db_query__{source}", args={"sql": CLEAN_SQL}),
+        n("calc", "code", 560, language="python", assign_to="calc", code="import json\nprint(json.dumps({'ratio': 3 / 4}))"),
+        n("card", "metrics", 840, metrics=[{"id": "ratio", "name": "客单价", "expression": "vars.calc.ratio"}]),
+        n("write", "report", 1120, instructions="写一句话"),
+        n("out", "output", 1400, fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
+    ]
+    return {"nodes": nodes, "edges": [{"id": f"e{i}", "source": a["id"], "target": b["id"]}
+                                      for i, (a, b) in enumerate(zip(nodes, nodes[1:]))]}
+
+
+async def test_upgrade_assist_checks_sql_like_the_publish_gate(scenic_source, monkeypatch):
+    """助手改写口径卡时顺手把查询改成了一对多关联后求和：受管级别的发布门禁会挡，升级的自查也得交回去改，
+    改不好就作废——以前自查和最终判断都没带 SQL 检查器，这样的改写会被采纳，交出去的图发布时才被拦。"""
+    import app.api.copilot as copilot
+    from app.db.base import SessionLocal
+
+    first = [{"op": "plan", "summary": "把纯算术的代码挪进口径卡"},
+             {"op": "update_node", "id": "card", "config": {"metrics": [
+                 {"id": "ratio", "name": "客单价", "expression": "cell(nodes.fetch, 0, 'gmv') / 4"}]}},
+             {"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL}}},
+             {"op": "done", "explanation": "客单价改由口径卡直接算"}]
+    fix = [{"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL + " "}}},
+           {"op": "done", "explanation": "改了"}]
+    model = _Scripted(first, fix)
+
+    async def _model(*_a, **_k):
+        return model, "mock-fast"
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    async with SessionLocal() as session:
+        out = await copilot.assist_upgrade(session, _upgrade_graph(scenic_source), level="governed")
+    # 自查把一对多求和交回模型，带着给模型的改法
+    assert len(model.calls) >= 2 and "上线前自查" in model.calls[1]
+    assert "orders 关联 order_items 是一对多" in model.calls[1]
+    # 改不好：整个作废，原因写的是这条 SQL 检查
+    assert out["accepted"] is False and out["ok"] is False
+    assert "一对多" in (out["reason"] or ""), out["reason"]