合入数据目录阶段 2B：基于数据目录的 SQL 检查——七条规则按依据状态定级，接入助手自查、发布门禁和运行时，出错的来源查询使指标降档

# Conflicts:
#	backend/app/api/evidence.py

backend/app/api/copilot.py

@@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.api.coded import DATASOURCE_SCOPE_EMPTY, CodedHTTPException
 # 起别名：本文件底下有个叫 explain 的接口，generate 里又有个局部变量叫 raw，同名会互相盖掉
 from app.core.errors import explain as explain_error, graph_error, not_configured, raw as raw_error
+from app.data.sqlcheck import CHECK_CODES
 from app.db.base import get_session
 from app.db.models import Workflow
 from app.engine.layout import CORRIDOR_MIN, MIN_ROW_GAP, NODE_W, _height, auto_layout
@@ -1095,6 +1096,7 @@ def _blocking_issues(
     nodes: dict[str, dict[str, Any]], edges: list[dict[str, Any]], scope: set[str] | None = None,
     *, sources: list[Any] | None = None, baseline: list[dict[str, Any]] | None = None,
     level: str | None = None, defaults: dict[str, Any] | None = None,
+    checkers: dict[str, Any] | None = None,
 ) -> list[dict[str, Any]] | None:
     """按运行时同一套校验，挑出会挡住运行的问题。图本身不成形时返回 None。
 
@@ -1112,22 +1114,41 @@ def _blocking_issues(
 
     defaults 是全图默认（graph.defaults）：报告撰写节点的 numbers / claims、审批策略都跟随它，不带上的话
     自查看到的问题和真正的门禁对不上。
+
+    checkers 是各数据源基于数据目录的 SQL 检查器（data/sqlcheck.load_checkers）：调用工具节点里写死的 SQL
+    查出 error 级的问题（一对多关联后重复计算、存量跨期求和）也算挡住，交回模型改；warning、info 不挡。
     """
     try:
         spec = GraphSpec.model_validate({"nodes": list(nodes.values()), "edges": edges,
                                          **({"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")
-           + authored_issues(spec, sources or [], baseline))
+           + [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)}
@@ -1196,7 +1217,7 @@ def _unsourced_reports(spec: GraphSpec, changed: set[str]) -> list[dict[str, Any
 
 
 def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str, Any]] | None = None,
-                    ) -> list[dict[str, Any]]:
+                    *, checkers: dict[str, Any] | None = None) -> list[dict[str, Any]]:
     """助手搭图时要打回、运行时却不挡的写法，按 error 交回模型改：
 
     - sql_unknown_column：调用工具节点的 SQL 里有数据源结构（schema_cache）里查不到的列——凭空
@@ -1204,6 +1225,11 @@ def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str
     - parse_model_text：整形节点按 JSON 解析 agent / llm 写的文字（validate 里只是 warning：旧图照跑）
     - report_no_source：报告撰写节点的上游没有任何证据来源（运行时只给一条警告，报告里的数全没有出处）。
       metrics_from 指向不是口径卡的节点，validate 本来就报 error，自查照样交回去
+
+    另外，给了 checkers（各数据源基于数据目录的 SQL 检查器）时，调用工具节点里写死的 SQL 再对照目录查一遍
+    （code 是 data/sqlcheck.CHECK_CODES 里的规则编号），级别照检查结果：error 交回模型改，warning、info
+    由调用方放进交付的问题清单。列名都对不上的 SQL 先改列名，这一轮不叠加目录检查。Agent 节点的 SQL 要到
+    运行时才写出来，这里拿不到，由数据源查询工具在执行后检查。
     """
     changed = _changed(spec, baseline)
     out: list[dict[str, Any]] = []
@@ -1227,6 +1253,7 @@ def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str
             continue
         found = sql_unknown_columns(sql, source)
         if not found:
+            out += _catalog_issues(node.id, sql, (checkers or {}).get(source.name))
             continue
         unknown, known = found
         listed = "、".join(known[:20]) + (f" 等 {len(known)} 列" if len(known) > 20 else "")
@@ -1240,6 +1267,14 @@ def authored_issues(spec: GraphSpec, sources: list[Any], baseline: list[dict[str
     return out
 
 
+def _catalog_issues(node_id: str, sql: str, checker: Any) -> list[dict[str, Any]]:
+    """一个调用工具节点的 SQL 对照数据目录查出来的问题，形状和别的自查问题一样（node_id、field 都在），
+    另带 table、column、relation_id、sql_excerpt 给界面定位。"""
+    if checker is None:
+        return []
+    return [{**check, "node_id": node_id, "edge_id": None, "field": "args.sql"} for check in checker.check_dicts(sql)]
+
+
 # --------------------------------------------------------------------------
 # SQL 里的列名对照数据源结构
 #
@@ -1614,10 +1649,14 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
     """
     from fastapi.responses import StreamingResponse
 
+    from app.data.sqlcheck import load_checkers
+
     sources = await _sources(session, payload.datasource_ids)
     scope = {r.name for r in sources} if payload.datasource_ids else None
     tool_list = _tool_catalog(sources)
     datasources = _datasource_section(sources) + _scope_note(sources, scope is not None)
+    # 基于数据目录的 SQL 检查器：在开流之前读好目录（流里不再碰这个会话），自查和收尾共用
+    checkers = await load_checkers(session, sources)
     system = (
         "你是一个 agent 工作流编排专家。根据用户需求，以操作流的方式逐步搭出一张可执行的工作流图。\n\n"
         f"{NODE_REFERENCE}\n\n可用工具：\n{tool_list}{datasources}\n\n"
@@ -1736,7 +1775,7 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
         # 循环 / 分支条件里套了 {{ }}），前面的步骤白跑，报错还停在半路
         repaired = 0
         for round_no in range(1, _SELF_CHECK_ROUNDS + 1):
-            errors = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline)
+            errors = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline, checkers=checkers)
             if not errors:
                 break
             yield _sse({"op": "check", "status": "repairing", "round": round_no, "issues": errors})
@@ -1754,7 +1793,7 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
                             "message": f"自查修正失败：{explain_error(e)[0]}", "detail": raw_error(e)})
                 break
             repaired = round_no
-        remaining = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline)
+        remaining = _blocking_issues(nodes, edges, scope, sources=sources, baseline=baseline, checkers=checkers)
         # 工具绑定变化不挡运行，但要在自查这一步就说出来：工具被改没了的图照样
         # 能跑，只是跑出来的是模型「假设」查过库的答案
         changes = tool_changes(baseline, list(nodes.values()))
@@ -1778,9 +1817,11 @@ async def generate_stream(payload: GenerateIn, session: AsyncSession = Depends(g
             issues = [i.model_dump() for i in validate_graph(spec).issues if i.level != "info"]
             issues += _issue_dicts(scope_issues(list(nodes.values()), scope),
                                    "datasource_out_of_scope")
-            # 自查没改好的猜列名留在画布上：运行时不挡（结构缓存可能旧了），但要看得见
-            issues += [{**i, "level": "warning"} for i in authored_issues(spec, sources, baseline)
-                       if i["code"] == "sql_unknown_column"]
+            # 自查没改好的猜列名留在画布上：运行时不挡（结构缓存可能旧了），但要看得见。
+            # 对照数据目录的 SQL 检查照原级别列出：warning、info 本来就不返工，改不好的 error 也在这里看得见
+            authored = authored_issues(spec, sources, baseline, checkers=checkers)
+            issues += [{**i, "level": "warning"} for i in authored if i["code"] == "sql_unknown_column"]
+            issues += [i for i in authored if i["code"] in CHECK_CODES]
             # 跳过的节点要说出来，不能安静地少一步。code 给前端认：这一类要单独
             # 提示「少了一步」，不能和普通校验警告混在一起
             issues += [
@@ -1877,13 +1918,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:
@@ -1939,7 +1984,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:
@@ -2207,7 +2252,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/evidence.py

@@ -844,6 +844,9 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s
         # 拿截断的查询结果整组算出来的：面板标「结果不完整」并写明原因。完整的指标没有这两个键
         **({"incomplete": True, "incomplete_reason": source.get("incomplete_reason") or ""}
            if source.get("incomplete") else {}),
+        # 来源查询没通过 SQL 检查：面板标出来并写明原因（问题和改法在查询步骤的 checks 里）。没问题的没有这两个键
+        **({"sql_check_failed": True, "sql_check_reason": source.get("sql_check_reason") or ""}
+           if source.get("sql_check_failed") else {}),
         "node_id": entry.get("node_id"), "artifact": artifact or None, "eid": cite.get("eid"),
         "eid_ok": cite.get("eid") == entry.get("eid") == make_eid("metric", artifact or None, {"metric": metric_id}),
         "hash_ok": hash_ok and metric is not None,
@@ -923,6 +926,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
     hint（期 4，P4-SPEC 2.8.2）：调用方是带文档标记的单元格片段时为真。这时快照若是上传表格的（有 data_version），
     步骤多一个 `provenance: true`，界面据此才去请求推断来源接口；其余情况一个键都不加，形状和期 4 之前完全相同
     （期 4 之前的文档、手工源、指标链都不多发请求）。不看封存状态：未封存的运行照常推断，响应里照实标明。
+
+    checks：快照里记下的 SQL 检查结果（数据目录阶段 2B，见 _query_checks），只在查出问题时有这个键。
     """
     catalog = doc.get("catalog") or {}
     alias = _query_alias(catalog, artifact)
@@ -977,7 +982,8 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
                 masked=[columns[i] for i in sorted(masked)],
                 # 查询时从驱动的原始值记下的列类型（老快照没有）：文本列里的 "2026" 按文本显示，不当数
                 column_types=snap.get("column_types") if isinstance(snap.get("column_types"), dict) else {},
-                **({"provenance": True} if hint and snap.get("data_version") and not merged else {}))
+                **({"provenance": True} if hint and snap.get("data_version") and not merged else {}),
+                **({"checks": checks} if (checks := _query_checks(snap)) else {}))
     if len(index) > MAX_WINDOW_ROWS:
         index, step["window_truncated"] = index[:MAX_WINDOW_ROWS], True
     step["row_index"] = index
@@ -987,6 +993,19 @@ async def _query_step(artifact: str, cells: list[tuple[Any, Any]], doc: dict[str
     return step
 
 
+#: 查询步骤里每条 SQL 检查结果交给界面的字段（data/sqlcheck.SqlCheck.as_dict 去掉 for_model）
+_CHECK_FIELDS = ("code", "level", "message", "table", "column", "relation_id", "sql_excerpt")
+
+
+def _query_checks(snap: dict[str, Any]) -> list[dict[str, Any]]:
+    """查询快照里记下的 SQL 检查结果（数据源查询工具执行后对照数据目录查的）。给模型的那句改法不上界面。
+    快照里没有（没查出问题、或者是升级前的快照）返回空列表，步骤里就不加这个键。"""
+    checks = snap.get("checks")
+    if not isinstance(checks, list):
+        return []
+    return [{k: c[k] for k in _CHECK_FIELDS if k in c} for c in checks if isinstance(c, dict) and c.get("code")]
+
+
 # --------------------------------------------------------------------------
 # 合并查询的结果：表级来历（输入、合并 SQL）和逐格来历（被引用的格追到输入的哪一格）
 # --------------------------------------------------------------------------

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/evidence.py

@@ -902,6 +902,9 @@ def build_catalog(
                 # 只在不完整时有这两个键，完整的指标条目和以前一字不差
                 **({"incomplete": True, "incomplete_reason": metric.get("incomplete_reason") or ""}
                    if metric.get("incomplete") else {}),
+                # 来源查询没通过 SQL 检查（data/sqlcheck.py 的 error 级问题）：同样只在有问题时有这两个键
+                **({"sql_check_failed": True, "sql_check_reason": metric.get("sql_check_reason") or ""}
+                   if metric.get("sql_check_failed") else {}),
             }
 
     queries = retrievals = 0
@@ -3109,6 +3112,9 @@ def catalog_prompt(catalog: dict[str, Any], *, budget: int = 12000, cells_allowe
                 # 不完整的指标照样能引用（单格、口径都对），但写作者得知道它不是全量：不写行数，免得照抄成裸数字
                 caveat = "（基于被截断的查询结果计算，结果不完整：引用时须如实说明，不要写成全部数据）" \
                     if e.get("incomplete") else ""
+                # 取数的 SQL 没通过检查（重复计算、存量跨期加总）：数是算错的可能很大，写作者引用时得交代
+                if e.get("sql_check_failed"):
+                    caveat += "（所依据的查询未通过 SQL 检查，结果存疑：引用时须如实说明，不要当作确定的结论）"
                 lines.append(f"- [[{e['alias']}]] {e.get('name')} = {e.get('rendered')}{caveat}")
     inputs = [e for e in catalog.values() if e.get("kind") == "input"]
     if inputs:

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/app/engine/issuance.py

@@ -462,6 +462,17 @@ def incomplete_gaps(metrics: list[dict[str, Any]]) -> list[str]:
             for m in metrics if isinstance(m, dict) and m.get("incomplete")]
 
 
+def sql_check_gaps(metrics: list[dict[str, Any]]) -> list[str]:
+    """契约收来的指标里来源查询没通过 SQL 检查的（口径卡标了 sql_check_failed：取数的那条 SQL 对照数据目录查出
+    error 级问题，比如一对多关联之后对「一」那一侧求和），每个一句缺口。
+
+    和 incomplete_gaps 同一个道理：数字照样回指得上口径卡，口径卡照样复算一致，但算它的那条查询本身就把数算错了
+    （重复计算、存量跨期加总），盖完整出具的章等于替一个错数作保。原因用口径卡记下的那句，写明是哪条检查。
+    """
+    return [f"指标「{m.get('name') or m.get('id')}」{m.get('sql_check_reason') or '所依据的查询未通过 SQL 检查'}"
+            for m in metrics if isinstance(m, dict) and m.get("sql_check_failed")]
+
+
 def _claims_effect(
     policy: str | dict[str, Any] | None,
     unsupported: list[Any] | int | None,

backend/app/engine/nodes/io.py

@@ -147,7 +147,7 @@ def _apply_contract(
 ) -> dict[str, Any]:
     from datetime import datetime, timezone
 
-    from app.engine.issuance import decide_tier, incomplete_gaps, trace_numbers
+    from app.engine.issuance import decide_tier, incomplete_gaps, sql_check_gaps, trace_numbers
 
     nodes = state.get("nodes") or {}
     # 写了 report_from 就是引用模式：按报告文档逐段复核，不再按数值回指叙述
@@ -188,6 +188,8 @@ def _apply_contract(
         gaps.append(f"指标来源未解析：{'、'.join(unresolved[:5])}")
     # 拿截断的查询结果整组算出来的指标：数回指得上，但算的只是取回的那一截，不能完整出具
     gaps.extend(incomplete_gaps(metrics))
+    # 来源查询没通过 SQL 检查（error 级：重复计算、存量跨期加总）的指标：口径卡复算得一致，算的却是错数
+    gaps.extend(sql_check_gaps(metrics))
 
     # 引用模式下，缺输入记成空值的指标（口径卡 on_missing=null）就是缺：报告里引用不了它，
     # 也不能因为 id 在清单里就算「齐了」。旧模式照旧只看 id，行为不变

backend/app/engine/nodes/metrics.py

@@ -49,6 +49,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
 
     表达式把截断的查询结果当成整组用了（len、sum、max……，见 expressions.TruncatedUse）的指标，
     值照算，但标 incomplete 并写明原因：它算的只是取回的那一截。出具契约据此降档（issuance.incomplete_gaps）。
+
+    取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、
+    存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。
     """
     definitions = ctx.cfg("metrics", []) or []
     caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1")
@@ -71,6 +74,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
 
     metrics: list[dict[str, Any]] = []
     errors: list[str] = []
+    #: 查询快照工件 id → 它的 error 级检查结果。一张卡里多个指标常常取自同一次查询，快照只读一次
+    checked: dict[str, list[dict[str, Any]]] = {}
     for definition in definitions:
         metric_id = str(definition.get("id") or "").strip()
         expr = str(definition.get("expression") or "").strip()
@@ -127,6 +132,12 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
             sources = list(use.sources.values())
             incomplete = {"incomplete": True, "incomplete_reason": _incomplete_reason(sources),
                           "truncated_sources": sources}
+        inputs = [_input_of(path, values[path], state, ctx, tctx) for path in paths]
+        sql_check: dict[str, Any] = {}
+        if failed := _failed_queries(inputs, list(use.sources.values()), paths, tctx, checked):
+            sql_check = {"sql_check_failed": True, "sql_check_reason": _sql_check_reason(failed),
+                         "sql_check_sources": [{"artifact": artifact, "codes": list(dict.fromkeys(
+                             str(c.get("code")) for c in found))} for artifact, found in failed]}
         metrics.append(
             {
                 "id": metric_id,
@@ -137,7 +148,7 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 "decimals": decimals,
                 "format": display["format"],
                 "rendered": _rendered(value, definition, display),
-                "inputs": [_input_of(path, values[path], state, ctx, tctx) for path in paths],
+                "inputs": inputs,
                 "substituted": substituted,
                 "recompute_ok": _recompute(substituted, tctx, value, decimals),
                 # 值是空的就是缺输入，不论 on_missing 是哪一种：fail 模式下直接取到空值的
@@ -145,6 +156,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 "status": "ok" if value is not None else "missing_input",
                 # 只在不完整时出现：完整的指标和以前一字不差，老运行复核时目录也对得上
                 **incomplete,
+                # 只在来源查询没通过 SQL 检查时出现，同上
+                **sql_check,
             }
         )
 
@@ -158,12 +171,16 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
          else f"- {m['id']}（{m['name']}）= {MISSING}（缺少输入，无值）")
         # 不写行数：叙述节点照抄进正文就是一个没有出处的数
         + ("（不完整：基于被截断的查询结果计算）" if m.get("incomplete") else "")
+        + ("（存疑：所依据的查询未通过 SQL 检查）" if m.get("sql_check_failed") else "")
         for m in metrics
     ]
     for m in metrics:
         if m.get("incomplete"):
             ctx.emit(EventType.LOG, level="warn", code="metric_incomplete", metric=m["id"],
                      message=f"指标「{m['name']}」{m['incomplete_reason']}")
+        if m.get("sql_check_failed"):
+            ctx.emit(EventType.LOG, level="warn", code="metric_sql_check", metric=m["id"],
+                     message=f"指标「{m['name']}」{m['sql_check_reason']}")
     card = {
         "kind": "metric_set",
         "caliber": caliber,
@@ -331,6 +348,72 @@ def _incomplete_reason(sources: list[dict[str, Any]]) -> str:
     return f"基于 {len(fetched)} 份被截断的查询结果计算（分别只取回了{'、'.join(fetched)}），结果不完整"
 
 
+#: 取值链的最后一段：.字段 或 [下标]
+_LAST_STEP = re.compile(r"(?:\.[^.\[\]]+|\[[^\[\]]*\])$")
+
+
+def _failed_queries(inputs: list[dict[str, Any]], truncated: list[dict[str, Any]], paths: list[str],
+                    tctx: dict[str, Any], checked: dict[str, list[dict[str, Any]]],
+                    ) -> list[tuple[str, list[dict[str, Any]]]]:
+    """这个指标用到的查询里，SQL 检查有 error 的：[(查询快照工件 id, error 级检查结果)]。
+
+    用到了哪些查询，三处来历合起来认：输入的出处（cell() 取的格、Agent 经 cite_fields 交来的字段都带着快照 id）、
+    截断跟踪记下的整组、取值链上的查询结果本身（数据整形解析过的 vars.res.rows[0][0]，往上找到 vars.res 那份
+    带 artifact 的结果）。检查结果从快照里读，不信内存里那份：快照按哈希寻址，就是查询当时存下的那份。
+    读不出来的快照不算（另有别的检查报快照的问题）。Agent 没开 cite_fields 时说不清数出自哪次查询，不算。
+    """
+    artifacts = [item.get("artifact") for item in inputs] + [s.get("artifact") for s in truncated]
+    artifacts += [_result_artifact(path, tctx) for path in paths if cell_parts(path) is None]
+    out = []
+    for artifact in dict.fromkeys(a for a in artifacts if isinstance(a, str) and a):
+        if artifact not in checked:
+            checked[artifact] = _error_checks(artifact)
+        if checked[artifact]:
+            out.append((artifact, checked[artifact]))
+    return out
+
+
+def _result_artifact(path: str, tctx: dict[str, Any]) -> str | None:
+    """顺着取值链往上找（vars.res.rows[0][0] → vars.res.rows[0] → vars.res.rows → vars.res），第一个查询结果的快照 id。
+    数据源工具交回的是 JSON 文本（调用工具节点的输出），解析过的是对象（数据整形之后），两种都认。"""
+    current = path
+    for _ in range(8):
+        value = _value_of(current, tctx)
+        if isinstance(value, str) and value.lstrip().startswith("{") and '"artifact"' in value:
+            try:
+                value = json.loads(value)
+            except ValueError:
+                value = None
+        if isinstance(value, dict) and isinstance(value.get("artifact"), str) and isinstance(value.get("rows"), list):
+            return value["artifact"]
+        shorter = _LAST_STEP.sub("", current)
+        if not shorter or shorter == current:
+            return None
+        current = shorter
+    return None
+
+
+def _error_checks(artifact: str) -> list[dict[str, Any]]:
+    try:
+        snapshot = artifact_store.load(artifact)
+    except Exception:  # noqa: BLE001 - 快照读不出来、被改过：这里不报，证据层另有说法
+        return []
+    checks = snapshot.get("checks") if isinstance(snapshot, dict) else None
+    return [c for c in checks if isinstance(c, dict) and c.get("level") == "error"] if isinstance(checks, list) else []
+
+
+def _sql_check_reason(failed: list[tuple[str, list[dict[str, Any]]]]) -> str:
+    """「所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算），
+    结果不可靠」：出具声明、运行日志、证据面板都用这句。只摘问题那半句，改法留在查询步骤里看。"""
+    problems = list(dict.fromkeys(str(c.get("message") or "").split("。")[0] for _, found in failed for c in found))
+    problems = [p for p in problems if p]
+    if not problems:
+        return "所依据的查询未通过 SQL 检查，结果不可靠"
+    if len(problems) == 1:
+        return f"所依据的查询未通过 SQL 检查（{problems[0]}），结果不可靠"
+    return f"所依据的查询有 {len(problems)} 处未通过 SQL 检查（{problems[0]}等），结果不可靠"
+
+
 def _truncated_field(path: str, state: GraphState, ctx: NodeContext) -> dict[str, Any] | None:
     """agent 经 cite_fields 交来、被截断切开的数组字段：{rows, artifact}。别的取值链返回 None。
 

backend/app/tools/datasource.py

@@ -33,6 +33,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.events import EventType
 from app.data import catalog as data_catalog
 from app.data import introspect, table_versions
+from app.data.sqlcheck import SqlChecker
 from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query
 from app.data.guard import QueryLimits, SqlRejected, is_write
 from app.data.tabular import UNSHAPED_NOTE
@@ -53,6 +54,10 @@ DATA_UNAVAILABLE = "查询失败（数据不可用）："
 #: 运行中途补固定一个源时发的那条日志事件的 code（_pin_late）
 LATE_PIN_CODE = "data_version_pinned"
 
+#: 每次查询之后基于数据目录的 SQL 检查最多等这么久。正常一条 SQL 几毫秒；等不到就放弃这一次检查，
+#: 查询结果照常交回——检查是附加的，不能拖慢查询
+SQL_CHECK_TIMEOUT_S = 2.0
+
 
 class _QueryArgs(BaseModel):
     sql: str = Field(description="要执行的 SQL，一次一条")
@@ -445,6 +450,15 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
     # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句，
     # 引擎按它告诉界面上限是多少（metadata["timeout_s"]）
     seconds = query_timeout(source)
+    # 基于数据目录的 SQL 检查器：和冻结进表结构快照的是同一版目录。第一次查询时才建（关系图也是用到才建）
+    checker: list[SqlChecker] = []
+
+    def _checker() -> SqlChecker:
+        if not checker:
+            checker.append(SqlChecker(kind=source.kind, schema_cache=source.schema_cache,
+                                      notes={name: entry.notes for name, entry in (catalog or {}).items()},
+                                      source_name=source.name))
+        return checker[0]
 
     async def _run(sql: str, limit: int | None = None) -> str:
         if gone := _vanished(source, fixed):
@@ -470,6 +484,11 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
 
         payload = result.to_payload()
         payload["source"] = source.name
+        # 对照数据目录检查这条 SQL（一对多关联后重复计算、存量跨期求和……）。结果交给模型（Agent 据此改写），
+        # 也随查询快照存下：证据面板看得到，口径卡据此给指标标「存疑」、出具降档。没查出问题不加这个键，
+        # 快照和以前一字不差
+        if checks := await _sql_checks(_checker(), sql):
+            payload["checks"] = checks
         # 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源
         # 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份
         if schema := await _store_schema(source, ctx, catalog=catalog):
@@ -511,6 +530,20 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
     )
 
 
+async def _sql_checks(checker: SqlChecker, sql: str) -> list[dict[str, Any]]:
+    """一条查询的 SQL 检查结果（as_dict 的形状）。在线程里跑、限时 SQL_CHECK_TIMEOUT_S：解析是纯计算，放在事件循环里
+    会挡住同一进程里别的运行。出错、超时都返回空列表——检查不能让一次成功的查询变成失败。"""
+    if not checker.enabled:
+        return []
+    try:
+        return await asyncio.wait_for(asyncio.to_thread(checker.check_dicts, sql), timeout=SQL_CHECK_TIMEOUT_S)
+    except asyncio.TimeoutError:
+        logger.warning("SQL 检查超过 %.1f 秒，本次跳过（数据源 %s）", SQL_CHECK_TIMEOUT_S, checker.source_name)
+    except Exception:  # noqa: BLE001
+        logger.exception("SQL 检查出错，本次跳过（数据源 %s）", checker.source_name)
+    return []
+
+
 async def _store_schema(source: Any, ctx: ToolContext, *,
                         catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None:
     """把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。

backend/pyproject.toml

@@ -30,6 +30,10 @@ dependencies = [
     "jsonschema>=4.23",
     "beautifulsoup4>=4.12",
     "cryptography>=43.0",
+    # 数据目录的 SQL 检查（data/sqlcheck.py）按数据源方言解析 SQL。sqlglot 的大版本常改语法树节点和
+    # 方言行为（节点改名、某个写法从解析成 Anonymous 变成专门的节点），规则是照着这些节点写的，
+    # 上限卡到下一个大版本——升级要重新锁、跑一遍 test_sqlcheck
+    "sqlglot>=30.21,<31",
 ]
 
 [project.optional-dependencies]

backend/requirements.lock

@@ -261,6 +261,8 @@ soupsieve==2.9.2
     # via beautifulsoup4
 sqlalchemy==2.0.52
     # via agentlab-backend (pyproject.toml)
+sqlglot==30.21.0
+    # via agentlab-backend (pyproject.toml)
 sqlite-vec==0.1.9
     # via langgraph-checkpoint-sqlite
 sse-starlette==3.4.11

backend/tests/fixtures/catalog/scenic_notes.py

@@ -0,0 +1,127 @@
+"""景区业务库（scenic.py）的一份数据目录，给 SQL 检查（data/sqlcheck.py）的测试用。
+
+只写检查要用到的几张表：订单与订单明细（一对多，外键）、入园记录（有效记录条件、码值、业务日期）、
+库存快照（存量）、会员等级（比率）、停车记录（两个时间列，业务日期按出场时间）、渠道（文字码值）。
+支付记录、会员这些表故意不写目录：关系要靠表结构现推（外键 verified、命名推断 proposed）。
+
+notes(status) 里「检查拿来当依据的项」一律用给定的状态：confirmed / verified / proposed / rejected。
+来源跟着状态走（confirmed 记人工填写，verified 记外键或数据剖析，其余记模型起草），和真实目录里的组合一致。
+单独改某一项的状态用 override：{("orders", "columns.total_amount.measure"): "proposed"} 这样的路径。
+表名、数据全部虚构。
+"""
+from __future__ import annotations
+
+import copy
+from typing import Any
+
+from app.data import catalog
+from app.data.catalog import make_item
+
+_SOURCE_OF = {"confirmed": "human", "verified": "profile", "proposed": "llm", "rejected": "llm"}
+
+
+def _item(value: Any, status: str) -> dict[str, Any]:
+    return make_item(value, _SOURCE_OF[status], status)
+
+
+def _rel(table: str, columns: list[str], to_table: str, to_columns: list[str], cardinality: str,
+         status: str) -> dict[str, Any]:
+    source = {"confirmed": "human", "verified": "fk"}.get(status, "name")
+    return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table,
+            "to_columns": to_columns, "cardinality": cardinality, "coverage": None, "source": source,
+            "status": status}
+
+
+#: 订单明细 → 订单 这条关系的编号（断言 relation_id 用）
+ORDER_ITEMS_TO_ORDERS = catalog.relation_id("order_items", ["order_id"], "orders", ["id"])
+
+
+def notes(status: str = "confirmed", override: dict[tuple[str, str], str] | None = None) -> dict[str, dict[str, Any]]:
+    """{表名: notes}。名称类的项（中文名）一律人工确认，只有检查的依据随 status 变。"""
+    s = status
+    out: dict[str, dict[str, Any]] = {
+        "orders": {
+            "label": _item("订单", "confirmed"),
+            "grain": _item("每笔订单一行", s),
+            "keys": _item(["id"], s),
+            "kind": _item("fact", s),
+            "business_date": _item({"column": "ordered_at", "rule": "按下单时间计"}, s),
+            "valid_filter": _item("status = 1", s),
+            "columns": {
+                "total_amount": {"label": _item("订单金额", "confirmed"), "unit": _item("元", s),
+                                 "measure": _item("flow", s)},
+                "status": {"label": _item("订单状态", "confirmed"), "measure": _item("status", s),
+                           "codes": _item({"1": "已支付", "2": "已退款"}, s)},
+                "ordered_at": {"label": _item("下单时间", "confirmed")},
+            },
+        },
+        "order_items": {
+            "label": _item("订单明细", "confirmed"),
+            "grain": _item("每笔订单的每个票种一行", s),
+            "keys": _item(["id"], s),
+            "columns": {
+                "amount": {"label": _item("明细金额", "confirmed"), "measure": _item("flow", s)},
+                "qty": {"label": _item("张数", "confirmed"), "measure": _item("flow", s)},
+            },
+            "relations": [_rel("order_items", ["order_id"], "orders", ["id"], "many_to_one", s)],
+        },
+        "visits": {
+            "label": _item("入园记录", "confirmed"),
+            "keys": _item(["id"], s),
+            "business_date": _item({"column": "visit_time", "rule": "按检票时间计"}, s),
+            "valid_filter": _item("status = 1", s),
+            "columns": {
+                "visitor_count": {"label": _item("入园人数", "confirmed"), "measure": _item("flow", s)},
+                "status": {"label": _item("检票状态", "confirmed"), "codes": _item({"1": "有效", "0": "作废"}, s)},
+                "visit_time": {"label": _item("检票时间", "confirmed")},
+            },
+            "relations": [_rel("visits", ["park_id"], "parks", ["id"], "many_to_one", s)],
+        },
+        "inventory_snapshots": {
+            "label": _item("库存快照", "confirmed"),
+            "kind": _item("snapshot", s),
+            "business_date": _item({"column": "snapshot_date", "rule": "每天营业结束时盘点"}, s),
+            "columns": {
+                "on_hand": {"label": _item("在库数量", "confirmed"), "measure": _item("stock", s)},
+                "snapshot_date": {"label": _item("盘点日期", "confirmed")},
+            },
+        },
+        "member_levels": {
+            "label": _item("会员等级", "confirmed"),
+            "columns": {"discount_rate": {"label": _item("折扣率", "confirmed"), "measure": _item("ratio", s)}},
+        },
+        "parking_records": {
+            "label": _item("停车记录", "confirmed"),
+            "business_date": _item({"column": "exited_at", "rule": "按出场时间计费"}, s),
+            "columns": {
+                "fee": {"label": _item("停车费", "confirmed"), "measure": _item("flow", s)},
+                "entered_at": {"label": _item("入场时间", "confirmed")},
+                "exited_at": {"label": _item("出场时间", "confirmed")},
+            },
+        },
+        "channels": {
+            "label": _item("渠道", "confirmed"),
+            "columns": {"channel_type": {"label": _item("渠道类型", "confirmed"),
+                                         "codes": _item({"线上直销": "官网和小程序", "线下": "窗口售票",
+                                                         "分销": "旅行社和第三方平台"}, s)}},
+        },
+    }
+    for (table, path), wanted in (override or {}).items():
+        out[table] = set_status(out[table], path, wanted)
+    return out
+
+
+def set_status(table_notes: dict[str, Any], path: str, status: str) -> dict[str, Any]:
+    """改一项的状态（不改入参）。path：表级字段名、columns.<列>.<字段>，或 relations.<编号>。"""
+    notes_ = copy.deepcopy(table_notes)
+    parts = path.split(".")
+    if parts[0] == "relations":
+        target = next(r for r in notes_["relations"] if r["id"] == parts[1])
+    elif parts[0] == "columns":
+        target = notes_["columns"][parts[1]][parts[2]]
+    else:
+        target = notes_[parts[0]]
+    target["status"] = status
+    target["source"] = {"confirmed": "human", "verified": "fk"}.get(status, "llm") if parts[0] == "relations" \
+        else _SOURCE_OF[status]
+    return notes_

backend/tests/test_copilot_sqlcheck.py

@@ -0,0 +1,245 @@
+"""助手搭完图的自查接上基于数据目录的 SQL 检查（data/sqlcheck.py）。
+
+- 调用工具节点里写死的 SQL 走现有的自查：error 级（一对多关联后重复计算、存量跨期求和）交回模型改，进修正轮；
+  warning、info 不返工，放进 final 的问题清单。
+- Agent 节点的 SQL 要到运行时才由模型写出来，静态拿不到，不查（运行时由数据源查询工具检查）。
+- 这一轮没碰的节点不查：用户自己写的 SQL 不替他改（和列名核对同一个规矩）。
+- 没有目录的数据源什么都不查。
+"""
+from __future__ import annotations
+
+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 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")
+#: 改好的：对明细求和；没按订单状态筛，留一条 warning
+FIXED_SQL = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id"
+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"
+
+_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 _spec(*nodes):
+    from app.engine.schema import GraphSpec
+
+    return GraphSpec.model_validate({"nodes": list(nodes), "edges": [
+        {"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]})
+
+
+def _node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "data": {"label": nid, "config": config}}
+
+
+def _fetch(sql, source="scenic"):
+    return _node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql})
+
+
+async def _checkers(scenic_db, status="confirmed"):
+    cache = await _schema(scenic_db)
+    source = SimpleNamespace(id="s1", name="scenic", kind="sqlite", schema_cache=cache)
+    checker = SqlChecker(kind="sqlite", schema_cache=cache, notes=scenic_notes.notes(status), source_name="scenic")
+    return source, {"scenic": checker}
+
+
+# --------------------------------------------------------------------------
+# authored_issues / _blocking_issues
+# --------------------------------------------------------------------------
+
+
+async def test_sql_check_errors_block_and_warnings_do_not(scenic_db):
+    from app.api.copilot import _blocking_issues, authored_issues
+
+    source, checkers = await _checkers(scenic_db)
+    spec = _spec(_node("start", "input"), _fetch(FANOUT_SQL), _node("out", "output", fields=[]))
+    [issue] = authored_issues(spec, [source], checkers=checkers)
+    assert issue["code"] == "fanout_sum" and issue["level"] == "error"
+    assert issue["node_id"] == "fetch" and issue["field"] == "args.sql"
+    assert issue["table"] == "orders" and issue["column"] == "total_amount" and issue["sql_excerpt"]
+    assert issue["message"].startswith("「订单」关联「订单明细」是一对多") and "order_items" in issue["for_model"]
+    nodes = {n.id: n.model_dump(mode="json") for n in spec.nodes}
+    assert [i["code"] for i in _blocking_issues(nodes, [], sources=[source], checkers=checkers)] == ["fanout_sum"]
+
+    # 只有 warning：不挡（不返工），但 authored_issues 照样带出来，收尾时放进 final
+    warn = _spec(_node("start", "input"), _fetch(FIXED_SQL), _node("out", "output", fields=[]))
+    [w] = authored_issues(warn, [source], checkers=checkers)
+    assert (w["code"], w["level"]) == ("missing_valid_filter", "warning")
+    nodes = {n.id: n.model_dump(mode="json") for n in warn.nodes}
+    assert _blocking_issues(nodes, [], sources=[source], checkers=checkers) == []
+
+    # 依据只是推断：降为 info，也不挡
+    _, proposed = await _checkers(scenic_db, "proposed")
+    assert {i["level"] for i in authored_issues(spec, [source], checkers=proposed)} == {"info"}
+    # 不给检查器（没有目录）：和以前一样什么都不查
+    assert authored_issues(spec, [source]) == []
+
+
+async def test_unknown_columns_come_first(scenic_db):
+    """列名都对不上的 SQL 先改列名：这时不再叠加目录检查的结果，免得一轮里塞给模型两套说法。"""
+    from app.api.copilot import authored_issues
+
+    source, checkers = await _checkers(scenic_db)
+    bad = ("SELECT SUM(o.total_amt) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id "
+           "WHERE o.status = 1")
+    spec = _spec(_node("start", "input"), _fetch(bad), _node("out", "output", fields=[]))
+    assert [i["code"] for i in authored_issues(spec, [source], checkers=checkers)] == ["sql_unknown_column"]
+
+
+async def test_agent_prompts_and_untouched_nodes_are_not_checked(scenic_db):
+    from app.api.copilot import authored_issues
+
+    source, checkers = await _checkers(scenic_db)
+    agent = _node("ask", "agent", prompt=f"用这条 SQL 查：{FANOUT_SQL}", tools=["db_query__scenic"])
+    spec = _spec(_node("start", "input"), agent, _node("out", "output", fields=[]))
+    assert authored_issues(spec, [source], checkers=checkers) == []
+    # 用户自己写的、这一轮没改的 SQL 不替他改
+    fetch = _fetch(FANOUT_SQL)
+    untouched = _spec(_node("start", "input"), fetch, _node("out", "output", fields=[]))
+    assert authored_issues(untouched, [source], baseline=[fetch], checkers=checkers) == []
+    # 模板占位照常检查其余部分
+    templated = _fetch("SELECT SUM(o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id "
+                       "WHERE o.status = 1 AND o.ordered_at >= '{{ input.since }}'")
+    out = authored_issues(_spec(_node("start", "input"), templated, _node("out", "output", fields=[])), [source],
+                          checkers=checkers)
+    assert [i["code"] for i in out] == ["fanout_sum"]
+
+
+# --------------------------------------------------------------------------
+# 端到端：假模型搭图，error 进修正轮，warning 留在 final
+# --------------------------------------------------------------------------
+
+
+class _Scripted:
+    """第一轮吐 first；收到自查请求（human 里有「上线前自查」）时吐 fix。"""
+
+    def __init__(self, first, fix):
+        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 op in (self.fix if "上线前自查" in human else self.first):
+            yield AIMessageChunk(content=json.dumps(op, ensure_ascii=False) + "\n")
+
+
+def _ops(source, sql):
+    return [
+        {"op": "plan", "summary": "查订单金额"},
+        {"op": "add_node", "node": {"id": "start", "type": "input", "label": "输入", "config": {}}},
+        {"op": "add_node", "node": {"id": "fetch", "type": "tool", "label": "取数",
+                                    "config": {"tool": f"db_query__{source}", "args": {"sql": sql}}}},
+        {"op": "add_edge", "edge": {"source": "start", "target": "fetch"}},
+        {"op": "add_node", "node": {"id": "out", "type": "output", "label": "成果",
+                                    "config": {"fields": [{"name": "r", "value": "{{ nodes.fetch }}"}]}}},
+        {"op": "add_edge", "edge": {"source": "fetch", "target": "out"}},
+        {"op": "done", "explanation": "查一下", "run": True},
+    ]
+
+
+@pytest.fixture
+async def scenic_source(scenic_db):
+    """库里一个带目录的景区数据源：表结构真实探查，目录按 scenic_notes 写进 catalog_notes。"""
+    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()
+
+
+async def _stream(monkeypatch, model, source):
+    import app.api.copilot as copilot
+    from app.main import app
+
+    async def _model(*_a, **_k):
+        return model, "mock-fast"
+
+    monkeypatch.setattr(copilot, "get_chat_model", _model)
+    events: list[dict] = []
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        async with client.stream("POST", "/api/copilot/generate-stream",
+                                 json={"instruction": "上周订单金额多少", "intent": "answer",
+                                       "datasource_ids": [source]}) as r:
+            async for line in r.aiter_lines():
+                if line.startswith("data: "):
+                    events.append(json.loads(line[6:]))
+    return events
+
+
+async def test_fanout_is_sent_back_and_fixed_before_running(scenic_source, monkeypatch):
+    model = _Scripted(_ops(scenic_source, FANOUT_SQL),
+                      [{"op": "update_node", "id": "fetch", "config": {"args": {"sql": FIXED_SQL}}},
+                       {"op": "done", "explanation": "改好了"}])
+    events = await _stream(monkeypatch, model, scenic_source)
+    checks = [e for e in events if e["op"] == "check"]
+    assert [c["status"] for c in checks] == ["repairing", "passed"], checks
+    [issue] = checks[0]["issues"]
+    assert issue["code"] == "fanout_sum" and issue["node_id"] == "fetch"
+    # 交回模型的是 for_model：写 SQL 里的真名和改法
+    assert "orders 关联 order_items 是一对多" in model.calls[1] and "「fetch」" in model.calls[1]
+    final = events[-1]
+    assert final["op"] == "final" and final["autorun"] is True
+    sql = next(n for n in final["graph"]["nodes"] if n["id"] == "fetch")["data"]["config"]["args"]["sql"]
+    assert sql == FIXED_SQL
+    # 剩下的 warning 不返工，留在 final 里看得见
+    [warn] = [i for i in final["issues"] if i.get("code") == "missing_valid_filter"]
+    assert warn["level"] == "warning" and warn["node_id"] == "fetch" and "status = 1" in warn["message"]
+    assert not [i for i in final["issues"] if i["level"] == "error"]
+
+
+async def test_a_fanout_that_is_not_fixed_stops_autorun(scenic_source, monkeypatch):
+    model = _Scripted(_ops(scenic_source, FANOUT_SQL),
+                      [{"op": "update_node", "id": "fetch", "config": {"args": {"sql": FANOUT_SQL + " "}}},
+                       {"op": "done", "explanation": "改了"}])
+    events = await _stream(monkeypatch, model, scenic_source)
+    checks = [e for e in events if e["op"] == "check"]
+    assert [c["status"] for c in checks] == ["repairing", "repairing", "failed"], checks
+    final = events[-1]
+    assert final["autorun"] is False
+    assert [i["code"] for i in final["issues"] if i["level"] == "error"] == ["fanout_sum"]
+
+
+async def test_clean_sql_costs_no_extra_call(scenic_source, monkeypatch):
+    model = _Scripted(_ops(scenic_source, CLEAN_SQL), [])
+    events = await _stream(monkeypatch, model, scenic_source)
+    assert len(model.calls) == 1
+    assert [e["status"] for e in events if e["op"] == "check"] == ["passed"]
+    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")]

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

backend/tests/test_runtime_sqlcheck.py

@@ -0,0 +1,288 @@
+"""运行时的 SQL 检查：数据源查询工具每次执行成功后对照数据目录检查一遍。
+
+- 结果写进查询结果（模型看得到，Agent 能据此改写 SQL）和查询快照的 checks 字段；没查出问题时不加这个键，
+  快照和以前一字不差。检查出错、超时都不影响查询本身。
+- 证据接口的查询步骤带上 checks（不带给模型的那句 for_model）。
+- 契约指标的来源查询有 error 级问题：口径卡把这个指标标 sql_check_failed 并写明原因，出具按 gaps 降档；
+  写作目录提醒写作者，证据面板的指标步骤标出来。和阶段 0 的「结果不完整」走同一条路。
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from langchain_core.messages import AIMessage
+from sqlalchemy import select
+
+from app.core import artifact_store
+from app.data import catalog
+from app.data.engine import engines
+from app.data.introspect import introspect
+from app.db.base import SessionLocal
+from app.db.models import DataSource, Run, RunEvent
+from app.engine.evidence import iter_segments
+from app.engine.issuance import decide_tier, sql_check_gaps
+from app.engine.runner import run_manager
+from app.main import app
+from app.tools.datasource import build_datasource_tools
+from app.tools.registry import ToolContext
+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"
+REASON = ("所依据的查询未通过 SQL 检查（「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算），"
+          "结果不可靠")
+
+_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]
+
+
+async def _source(path: str, *, with_catalog: bool = True) -> tuple[str, str]:
+    name = f"scenic_{uuid.uuid4().hex[:8]}"
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=path, readonly=True, enabled=True)
+        row.schema_cache = await _schema(path)
+        session.add(row)
+        await session.commit()
+        if with_catalog:
+            for table, notes in scenic_notes.notes().items():
+                await catalog.write_entry(session, row.id, table, notes, if_version=0, actor="王敏")
+        return row.id, name
+
+
+@pytest.fixture
+async def scenic_source(scenic_db):
+    source_id, name = await _source(scenic_db)
+    yield name
+    await engines.invalidate(source_id)
+
+
+async def _query_tool(name: str):
+    ctx = ToolContext(run_id=uuid.uuid4().hex, node_id="n1", data_versions=None)
+    async with SessionLocal() as session:
+        [tool] = await build_datasource_tools([f"db_query__{name}"], ctx, session)
+    return tool
+
+
+# --------------------------------------------------------------------------
+# 数据源查询工具
+# --------------------------------------------------------------------------
+
+
+async def test_query_result_and_snapshot_carry_checks(scenic_source):
+    tool = await _query_tool(scenic_source)
+    result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
+    assert result["rows"] and result["artifact"]
+    [check] = result["checks"]
+    assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders"
+    assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "order_items" in check["for_model"]
+    assert artifact_store.load(result["artifact"])["checks"] == result["checks"]
+
+    clean = json.loads(await tool.coroutine(sql=CLEAN_SQL))
+    assert "checks" not in clean and "checks" not in artifact_store.load(clean["artifact"])
+
+
+async def test_sources_without_catalog_are_not_checked(scenic_db):
+    source_id, name = await _source(scenic_db, with_catalog=False)
+    try:
+        result = json.loads(await (await _query_tool(name)).coroutine(sql=FANOUT_SQL))
+        assert result["rows"] and "checks" not in result
+    finally:
+        await engines.invalidate(source_id)
+
+
+async def test_a_failing_or_slow_check_never_breaks_the_query(scenic_source, monkeypatch):
+    from app.data import sqlcheck
+    from app.tools import datasource
+
+    tool = await _query_tool(scenic_source)
+
+    def boom(self, sql):
+        raise RuntimeError("规则缺陷")
+
+    monkeypatch.setattr(sqlcheck.SqlChecker, "check", boom)
+    result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
+    assert result["rows"] and "checks" not in result
+
+    def slow(self, sql):
+        time.sleep(1.0)
+        return []
+
+    monkeypatch.setattr(sqlcheck.SqlChecker, "check", slow)
+    monkeypatch.setattr(datasource, "SQL_CHECK_TIMEOUT_S", 0.05)
+    started = time.monotonic()
+    result = json.loads(await tool.coroutine(sql=FANOUT_SQL))
+    assert result["rows"] and "checks" not in result
+    assert time.monotonic() - started < 0.9, "检查超时就放弃，不等它跑完"
+
+
+# --------------------------------------------------------------------------
+# 出具：契约指标的来源查询有 error 级问题时降档
+# --------------------------------------------------------------------------
+
+
+def test_sql_check_failures_become_readable_gaps():
+    metrics = [{"id": "gmv", "name": "订单金额", "value": 1, "sql_check_failed": True, "sql_check_reason": REASON},
+               {"id": "n", "name": "订单数", "value": 2}]
+    assert sql_check_gaps(metrics) == [f"指标「订单金额」{REASON}"]
+    assert sql_check_gaps(metrics[1:]) == []
+    tier = decide_tier(missing_required=[], missing_expected=[], unmatched=[], strict=False,
+                       gaps=sql_check_gaps(metrics))
+    assert tier == "degraded"
+
+
+@pytest.fixture
+async def engine_up():
+    await run_manager.setup()
+    yield
+    await run_manager.shutdown()
+
+
+def node(nid, ntype, **config):
+    return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+
+def chain(nodes: list[dict]) -> dict:
+    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
+GMV = {"id": "gmv", "name": "订单金额", "unit": "元", "format": "plain", "expression": "cell(nodes.fetch, 0, 'gmv')"}
+#: 同一份结果经数据整形解析后按下标取：来历记成 transform，要顺着取值链找到那份查询结果
+GMV2 = {"id": "gmv2", "name": "订单金额（整形后）", "unit": "元", "format": "plain",
+        "expression": "vars.res.rows[0][0]"}
+WEEKLY = "本周订单金额 [[m:gmv]]。"
+
+
+def weekly(source: str, sql: str) -> dict:
+    return chain([
+        node("start", "input"),
+        node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}),
+        node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"),
+        node("card", "metrics", caliber="订单口径", caliber_version="v1", metrics=[GMV, GMV2]),
+        node("write", "report", instructions="写订单周报"),
+        node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
+             contract={"report_from": "write", "metrics_from": ["card"], "required": ["gmv"]}),
+    ])
+
+
+class Writer:
+    """报告写作者：每次都照 text 写；记下看到的提示。"""
+
+    def __init__(self, monkeypatch, text: str = WEEKLY):
+        from app.providers import mock_model
+
+        self.prompts: list[str] = []
+
+        def decide(model, messages):
+            self.prompts.append("\n".join(str(m.content) for m in messages))
+            return AIMessage(content=text)
+
+        monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
+
+
+async def wait(run_id: str) -> Run:
+    for _ in range(400):
+        await asyncio.sleep(0.05)
+        async with SessionLocal() as session:
+            row = await session.get(Run, run_id)
+        if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
+            return row
+    raise AssertionError(f"run {run_id} 没有结束：{row.status}")
+
+
+async def events(run_id: str, etype: str, node_id: str | None = None) -> list[RunEvent]:
+    async with SessionLocal() as session:
+        q = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype)
+        if node_id:
+            q = q.where(RunEvent.node_id == node_id)
+        return list((await session.execute(q.order_by(RunEvent.seq))).scalars())
+
+
+async def output_of(run_id: str, node_id: str) -> dict:
+    [finished] = await events(run_id, "node.finished", node_id)
+    return artifact_store.load(finished.data["artifact"])
+
+
+async def test_a_fanout_behind_a_contract_metric_degrades_the_issuance(scenic_source, engine_up, monkeypatch):
+    writer = Writer(monkeypatch)
+    run = await run_manager.start(graph=weekly(scenic_source, FANOUT_SQL), input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+
+    card = await output_of(row.id, "card")
+    metrics = {m["id"]: m for m in card["metrics"]}
+    [query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")]
+    for mid in ("gmv", "gmv2"):
+        assert metrics[mid]["sql_check_failed"] is True and metrics[mid]["sql_check_reason"] == REASON
+        assert metrics[mid]["sql_check_sources"] == [{"artifact": query_end["query_artifact"],
+                                                     "codes": ["fanout_sum"]}]
+    assert "（存疑：所依据的查询未通过 SQL 检查）" in card["text"]
+    warns = [e.data for e in await events(row.id, "log", "card") if e.data.get("code") == "metric_sql_check"]
+    assert [w["message"] for w in warns] == [f"指标「订单金额」{REASON}", f"指标「订单金额（整形后）」{REASON}"]
+
+    issuance = row.output["_issuance"]
+    assert issuance["tier"] == "degraded", issuance
+    assert f"指标「订单金额」{REASON}" in issuance["gaps"]
+    # 写作目录提醒写作者：这个数存疑，引用时要说明
+    assert "所依据的查询未通过 SQL 检查，结果存疑" in writer.prompts[0]
+
+    # 证据面板：指标步骤标出来，查询步骤带着检查结果（不带给模型的那句）
+    doc = artifact_store.load(row.output["_evidence"]["doc_artifact"])
+    assert doc["catalog"]["m:gmv"]["sql_check_failed"] is True
+    seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv")
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
+    metric = body["chain"][0]
+    assert metric["step"] == "metric" and metric["sql_check_failed"] is True
+    assert metric["sql_check_reason"] == REASON
+    [query] = [s for s in body["chain"] if s["step"] == "query"]
+    [check] = query["checks"]
+    assert check["code"] == "fanout_sum" and check["level"] == "error" and check["table"] == "orders"
+    assert check["message"].startswith("「订单」关联「订单明细」是一对多") and "for_model" not in check
+
+
+async def test_the_same_card_on_a_clean_query_issues_formally(scenic_source, engine_up, monkeypatch):
+    Writer(monkeypatch)
+    run = await run_manager.start(graph=weekly(scenic_source, CLEAN_SQL), input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    card = await output_of(row.id, "card")
+    assert all("sql_check_failed" not in m for m in card["metrics"])
+    issuance = row.output["_issuance"]
+    assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance
+    doc = artifact_store.load(row.output["_evidence"]["doc_artifact"])
+    seg = next(s for s in iter_segments(doc) if s.get("ref") == "m:gmv")
+    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
+        body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
+    assert "sql_check_failed" not in body["chain"][0]
+    assert all("checks" not in s for s in body["chain"] if s["step"] == "query")
+
+
+async def test_warnings_alone_do_not_degrade(scenic_source, engine_up, monkeypatch):
+    """只有 warning（这里是没按订单状态筛）：查询结果和快照里照样带着，但不影响出具档位。"""
+    Writer(monkeypatch)
+    sql = "SELECT SUM(i.amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id"
+    run = await run_manager.start(graph=weekly(scenic_source, sql), input_payload={})
+    row = await wait(run.id)
+    assert row.status == "succeeded", row.error
+    [query_end] = [e.data for e in await events(row.id, "tool.end", "fetch")]
+    snapshot = artifact_store.load(query_end["query_artifact"])
+    assert [c["code"] for c in snapshot["checks"]] == ["missing_valid_filter"]
+    assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]

backend/tests/test_sqlcheck.py

@@ -0,0 +1,505 @@
+"""基于数据目录的 SQL 检查（data/sqlcheck.py）：七条规则的正例、反例，级别随依据的状态变化，以及解析不了的写法。
+
+用合成的景区业务库（tests/fixtures/catalog/scenic.py）和它的一份目录（scenic_notes.py）。检查只做有把握的判断：
+解析不了、遇到不支持的写法、列对不上表，一律跳过不报——它的 error 会把图打回去改、挡住受管级别的发布、让出具降档，
+误报的代价比漏报大。
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.data.engine import engines
+from app.data.introspect import introspect
+from app.data.sqlcheck import CHECK_CODES, SqlChecker, check_sql, dialect_of
+from tests.fixtures.catalog import scenic_notes
+from tests.fixtures.catalog.scenic_notes import ORDER_ITEMS_TO_ORDERS
+
+_CACHE: dict[str, dict] = {}
+
+
+@pytest.fixture
+async def cache(scenic_db):
+    """景区库的表结构缓存（真实探查一次：带外键和唯一约束，关系现推要用）。每个测试进程只探查一次。"""
+    if scenic_db not in _CACHE:
+        source = SimpleNamespace(id="sqlcheck-scenic", name="scenic", kind="sqlite", database=scenic_db, host=None,
+                                 port=None, username=None, password=None, options={}, readonly=True, description="",
+                                 schema_cache={}, origin="manual")
+        try:
+            _CACHE[scenic_db] = await introspect(source)
+        finally:
+            await engines.invalidate(source.id)
+    return _CACHE[scenic_db]
+
+
+def run(cache, sql, status="confirmed", override=None, kind="sqlite"):
+    checker = SqlChecker(kind=kind, schema_cache=cache, notes=scenic_notes.notes(status, override),
+                         source_name="scenic")
+    return checker.check(sql)
+
+
+def found(cache, sql, status="confirmed", override=None, kind="sqlite"):
+    """[(code, level, table, column)]，按检查给出的顺序。"""
+    return [(c.code, c.level, c.table, c.column) for c in run(cache, sql, status, override, kind)]
+
+
+def codes(cache, sql, status="confirmed", override=None):
+    return {c.code for c in run(cache, sql, status, override)}
+
+
+# --------------------------------------------------------------------------
+# 结果的形状
+# --------------------------------------------------------------------------
+
+
+async def test_result_shape_and_wording(cache):
+    sql = ("SELECT o.channel_id, SUM(o.total_amount) AS amt FROM orders o "
+           "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 GROUP BY o.channel_id")
+    [check] = run(cache, sql)
+    out = check.as_dict()
+    assert set(out) == {"code", "level", "message", "for_model", "table", "column", "relation_id", "sql_excerpt"}
+    assert out["code"] == "fanout_sum" and out["level"] == "error"
+    assert out["table"] == "orders" and out["column"] == "total_amount"
+    assert out["relation_id"] == ORDER_ITEMS_TO_ORDERS
+    assert "SUM(o.total_amount)" in out["sql_excerpt"]
+    assert out["message"].startswith("「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算。")
+    # 给模型的写 SQL 里的真名和改法；给人看的不出现工具名和对模型的指令
+    assert "orders" in out["for_model"] and "order_items" in out["for_model"]
+    assert "db_schema__" not in out["message"] and "db_" not in out["message"]
+    # 没有的可选字段不出现
+    single = run(cache, "SELECT AVG(discount_rate) FROM member_levels")[0].as_dict()
+    assert "relation_id" not in single and single["column"] == "discount_rate"
+    assert set(CHECK_CODES) == {"fanout_sum", "stock_summed", "join_unconfirmed", "ratio_aggregated",
+                                "missing_valid_filter", "unknown_code", "wrong_date_column"}
+
+
+def test_dialects():
+    assert dialect_of("sqlite") == "sqlite"
+    assert dialect_of("postgres") == dialect_of("postgresql") == "postgres"
+    assert dialect_of("mysql") == dialect_of("mariadb") == "mysql"
+    assert dialect_of("oracle") == "oracle"
+    assert dialect_of("mssql") is None and dialect_of(None) is None
+
+
+# --------------------------------------------------------------------------
+# fanout_sum：沿一对多关联之后对「一」那一侧的度量列求和、计数
+# --------------------------------------------------------------------------
+
+FANOUT = [
+    "SELECT SUM(o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 先写明细、后连订单：订单照样随明细重复
+    "SELECT SUM(o.total_amount) FROM order_items i JOIN orders o ON o.id = i.order_id WHERE o.status = 1",
+    # 包一层 COALESCE、CASE 里只用订单的列：还是订单的数
+    "SELECT SUM(COALESCE(o.total_amount, 0)) FROM orders o LEFT JOIN order_items i ON o.id = i.order_id "
+    "WHERE o.status = 1",
+    "SELECT SUM(CASE WHEN o.status = 1 THEN o.total_amount END) FROM orders o JOIN order_items i "
+    "ON i.order_id = o.id AND i.qty > 0",
+    # 逗号连接、条件写在 WHERE 里
+    "SELECT SUM(o.total_amount) FROM orders o, order_items i WHERE i.order_id = o.id AND o.status = 1",
+]
+
+
+@pytest.mark.parametrize("sql", FANOUT)
+async def test_fanout_sum_positive(cache, sql):
+    assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, sql)
+
+
+async def test_fanout_count_of_the_one_side_key(cache):
+    sql = "SELECT COUNT(o.id) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1"
+    [check] = run(cache, sql)
+    assert (check.code, check.level, check.table, check.column) == ("fanout_sum", "error", "orders", "id")
+    assert "计数会重复计算" in check.message
+
+
+async def test_fanout_through_two_hops_with_an_inferred_foreign_key(cache):
+    """支付记录没写目录：订单 → 支付记录的关系按外键现推（verified），明细经订单连支付记录，明细的金额照样重复。"""
+    sql = ("SELECT SUM(i.amount) FROM order_items i JOIN orders o ON o.id = i.order_id "
+           "JOIN payments p ON p.order_id = o.id WHERE o.status = 1")
+    [check] = [c for c in run(cache, sql) if c.code == "fanout_sum"]
+    assert (check.level, check.table, check.column) == ("error", "order_items", "amount")
+    assert "「订单」关联「payments」是一对多，「订单明细」的每一行会随之重复" in check.message
+    assert "去掉与「payments」的关联" in check.message
+
+
+FANOUT_NEGATIVE = [
+    # 对「多」那一侧求和：本来就在明细粒度上
+    "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    "SELECT SUM(i.amount) FROM order_items i JOIN orders o ON o.id = i.order_id WHERE o.status = 1",
+    # 不关联
+    "SELECT SUM(o.total_amount) FROM orders o WHERE o.status = 1",
+    # 去重计数、去重求和：写的人已经在处理重复
+    "SELECT COUNT(DISTINCT o.id) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 两边的列一起算：按明细逐行算出来的数
+    "SELECT SUM(i.qty * o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 按明细的主键分组：每组只有一行明细，订单不重复
+    "SELECT i.id, SUM(o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 "
+    "GROUP BY i.id",
+    # 明细先在子查询里按订单汇总再连：一对一
+    "SELECT SUM(o.total_amount), SUM(x.n) FROM orders o JOIN (SELECT order_id, COUNT(*) AS n FROM order_items "
+    "GROUP BY order_id) x ON x.order_id = o.id WHERE o.status = 1",
+    # 计数所有行：说不清是要数订单还是数明细
+    "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 窗口函数：不是分组聚合
+    "SELECT o.id, SUM(o.total_amount) OVER (PARTITION BY o.channel_id) FROM orders o "
+    "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 反连接：只留没有明细的订单，每笔订单至多一行
+    "SELECT SUM(o.total_amount) FROM orders o LEFT JOIN order_items i ON i.order_id = o.id "
+    "WHERE i.id IS NULL AND o.status = 1",
+]
+
+
+@pytest.mark.parametrize("sql", FANOUT_NEGATIVE)
+async def test_fanout_sum_negative(cache, sql):
+    assert "fanout_sum" not in codes(cache, sql)
+
+
+async def test_fanout_level_follows_the_status_of_its_basis(cache):
+    sql = FANOUT[0]
+    assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, sql, "verified")
+    assert ("fanout_sum", "info", "orders", "total_amount") in found(cache, sql, "proposed")
+    # 关系确认了、度量类型只是推断：依据里有推断就降为 info
+    assert ("fanout_sum", "info", "orders", "total_amount") in found(
+        cache, sql, override={("orders", "columns.total_amount.measure"): "proposed"})
+    # 关系被驳回：不知道是一对多，不报重复计算；关联本身对不上目录，单独提示
+    rejected = found(cache, sql, override={("order_items", f"relations.{ORDER_ITEMS_TO_ORDERS}"): "rejected"})
+    assert not [f for f in rejected if f[0] == "fanout_sum"]
+    assert ("join_unconfirmed", "warning", "order_items", None) in rejected
+    # 度量类型被驳回：不知道订单金额能不能加，不报
+    assert "fanout_sum" not in codes(cache, sql, override={("orders", "columns.total_amount.measure"): "rejected"})
+
+
+# --------------------------------------------------------------------------
+# stock_summed：存量跨时间求和
+# --------------------------------------------------------------------------
+
+STOCK = [
+    "SELECT SUM(on_hand) FROM inventory_snapshots",
+    'SELECT "storeID", SUM(on_hand) FROM inventory_snapshots GROUP BY "storeID"',
+    "SELECT SUM(s.on_hand) FROM inventory_snapshots s WHERE s.snapshot_date BETWEEN '2026-09-01' AND '2026-09-30'",
+]
+
+
+@pytest.mark.parametrize("sql", STOCK)
+async def test_stock_summed_positive(cache, sql):
+    assert ("stock_summed", "error", "inventory_snapshots", "on_hand") in found(cache, sql)
+
+
+STOCK_NEGATIVE = [
+    "SELECT snapshot_date, SUM(on_hand) FROM inventory_snapshots GROUP BY snapshot_date",
+    "SELECT SUM(on_hand) FROM inventory_snapshots WHERE snapshot_date = '2026-09-30'",
+    "SELECT substr(snapshot_date, 1, 7) AS m, SUM(on_hand) FROM inventory_snapshots GROUP BY substr(snapshot_date, 1, 7)",
+    "SELECT AVG(on_hand) FROM inventory_snapshots",
+    "SELECT MAX(on_hand) FROM inventory_snapshots",
+    "SELECT snapshot_date, SUM(on_hand) OVER (PARTITION BY snapshot_date) FROM inventory_snapshots",
+    # 经日期维度表分组：维度表的日期和快照日期相等
+    "SELECT d.biz_date, SUM(s.on_hand) FROM inventory_snapshots s JOIN dim_dates d ON d.biz_date = s.snapshot_date "
+    "GROUP BY d.biz_date",
+]
+
+
+@pytest.mark.parametrize("sql", STOCK_NEGATIVE)
+async def test_stock_summed_negative(cache, sql):
+    assert "stock_summed" not in codes(cache, sql)
+
+
+async def test_stock_level_follows_status(cache):
+    sql = STOCK[0]
+    assert ("stock_summed", "error", "inventory_snapshots", "on_hand") in found(cache, sql, "verified")
+    assert ("stock_summed", "info", "inventory_snapshots", "on_hand") in found(cache, sql, "proposed")
+    assert "stock_summed" not in codes(cache, sql, "rejected")
+    # 业务日期被驳回：不知道按哪一列算时间，不报
+    assert "stock_summed" not in codes(cache, sql, override={("inventory_snapshots", "business_date"): "rejected"})
+
+
+# --------------------------------------------------------------------------
+# join_unconfirmed：关联条件对不上目录里的关系
+# --------------------------------------------------------------------------
+
+async def test_join_unconfirmed_positive(cache):
+    # 列连错了：订单的主键对明细的主键
+    wrong = found(cache, "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.id = o.id WHERE o.status = 1")
+    assert ("join_unconfirmed", "warning", "order_items", None) in wrong
+    [check] = [c for c in run(cache, "SELECT o.id FROM orders o JOIN members m ON m.id = o.channel_id "
+                                     "WHERE o.status = 1") if c.code == "join_unconfirmed"]
+    assert check.level == "warning" and "o.channel_id" in check.sql_excerpt
+    assert check.table == "members" and "「订单」与「members」的关联条件" in check.message
+    # 只对上命名推断出来的关系（入园记录.member_id → 会员）：提示一句，级别 info
+    proposed = found(cache, "SELECT m.name FROM visits v JOIN members m ON m.id = v.member_id WHERE v.status = 1")
+    assert ("join_unconfirmed", "info", "members", None) in proposed
+
+
+JOIN_NEGATIVE = [
+    "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1",
+    # 多写的过滤条件不影响对关系
+    "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id AND i.qty > 0 WHERE o.status = 1",
+    # 没写目录的表之间按外键现推（verified）
+    "SELECT SUM(p.paid_amount) FROM orders o JOIN payments p ON p.order_id = o.id WHERE o.status = 1",
+    # 两张表都没写目录：没有可对照的目录，不判
+    "SELECT COUNT(*) FROM tour_groups t JOIN agencies a ON a.channel_id = t.agency_id",
+    # 子查询、CTE 当一边：不知道它的粒度，不判
+    "WITH x AS (SELECT order_id, SUM(amount) AS a FROM order_items GROUP BY order_id) "
+    "SELECT SUM(x.a) FROM orders o JOIN x ON x.order_id = o.id WHERE o.status = 1",
+    # 自己连自己：目录不记这种关系
+    "SELECT c.name FROM categories c JOIN categories p ON p.id = c.\"parentId\"",
+    # 没有等值条件
+    "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.amount > o.total_amount WHERE o.status = 1",
+    # OR 连起来的条件：说不清按哪条连
+    "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.order_id = o.id OR i.id = o.id WHERE o.status = 1",
+]
+
+
+@pytest.mark.parametrize("sql", JOIN_NEGATIVE)
+async def test_join_unconfirmed_negative(cache, sql):
+    assert "join_unconfirmed" not in codes(cache, sql)
+
+
+async def test_join_level_follows_status(cache):
+    sql = "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1"
+    assert "join_unconfirmed" not in codes(cache, sql, "confirmed")
+    assert "join_unconfirmed" not in codes(cache, sql, "verified")
+    # 结果记在 JOIN 引入的那张表上
+    assert ("join_unconfirmed", "info", "order_items", None) in found(cache, sql, "proposed")
+    assert ("join_unconfirmed", "warning", "order_items", None) in found(cache, sql, "rejected")
+
+
+# --------------------------------------------------------------------------
+# ratio_aggregated：对比率求和、求平均
+# --------------------------------------------------------------------------
+
+@pytest.mark.parametrize("sql", [
+    "SELECT AVG(discount_rate) FROM member_levels",
+    "SELECT SUM(discount_rate) FROM member_levels",
+    "SELECT ROUND(AVG(l.discount_rate), 2) FROM member_levels l",
+])
+async def test_ratio_aggregated_positive(cache, sql):
+    assert found(cache, sql) == [("ratio_aggregated", "warning", "member_levels", "discount_rate")]
+
+
+@pytest.mark.parametrize("sql", [
+    "SELECT level_name, discount_rate FROM member_levels",
+    "SELECT MAX(discount_rate) FROM member_levels",
+    # 加权：比率乘上数量再加，是正确的算法
+    "SELECT SUM(m.id * l.discount_rate) FROM members m JOIN member_levels l ON l.id = m.member_level_id",
+])
+async def test_ratio_aggregated_negative(cache, sql):
+    assert "ratio_aggregated" not in codes(cache, sql)
+
+
+async def test_ratio_level_follows_status(cache):
+    sql = "SELECT AVG(discount_rate) FROM member_levels"
+    assert found(cache, sql, "verified") == [("ratio_aggregated", "warning", "member_levels", "discount_rate")]
+    assert found(cache, sql, "proposed") == [("ratio_aggregated", "info", "member_levels", "discount_rate")]
+    assert found(cache, sql, "rejected") == []
+
+
+# --------------------------------------------------------------------------
+# missing_valid_filter：没按有效记录条件筛
+# --------------------------------------------------------------------------
+
+@pytest.mark.parametrize("sql", [
+    "SELECT COUNT(*) FROM visits",
+    "SELECT SUM(visitor_count) FROM visits WHERE park_id = 1",
+    "SELECT v.park_id, SUM(v.visitor_count) FROM visits v GROUP BY v.park_id",
+])
+async def test_missing_valid_filter_positive(cache, sql):
+    [check] = run(cache, sql)
+    assert (check.code, check.level, check.table) == ("missing_valid_filter", "warning", "visits")
+    assert "有效记录条件「status = 1」" in check.message
+
+
+@pytest.mark.parametrize("sql", [
+    "SELECT COUNT(*) FROM visits WHERE status = 1",
+    "SELECT COUNT(*) FROM visits v WHERE v.status IN (0, 1)",
+    # 按状态分组：写的人就是要看每种状态
+    "SELECT status, COUNT(*) FROM visits GROUP BY status",
+    # 条件写在聚合里
+    "SELECT SUM(CASE WHEN status = 1 THEN visitor_count ELSE 0 END) FROM visits",
+    # CTE 里筛过
+    "WITH ok AS (SELECT * FROM visits WHERE status = 1) SELECT COUNT(*) FROM ok",
+    # 子查询原样透出、外层再筛
+    "SELECT COUNT(*) FROM (SELECT * FROM visits) x WHERE x.status = 1",
+])
+async def test_missing_valid_filter_negative(cache, sql):
+    assert "missing_valid_filter" not in codes(cache, sql)
+
+
+async def test_valid_filter_level_follows_status(cache):
+    sql = "SELECT COUNT(*) FROM visits"
+    assert found(cache, sql, "verified") == [("missing_valid_filter", "warning", "visits", None)]
+    assert found(cache, sql, "proposed") == [("missing_valid_filter", "info", "visits", None)]
+    assert found(cache, sql, "rejected") == []
+
+
+# --------------------------------------------------------------------------
+# unknown_code：写了码值表里没有的值
+# --------------------------------------------------------------------------
+
+async def test_unknown_code_positive(cache):
+    [check] = run(cache, "SELECT COUNT(*) FROM visits WHERE status = 3")
+    assert (check.code, check.level, check.table, check.column) == ("unknown_code", "warning", "visits", "status")
+    assert "「3」" in check.message and "1=有效、0=作废" in check.message
+    [listed] = run(cache, "SELECT COUNT(*) FROM visits WHERE status IN (1, 5, 7)")
+    assert "「5」、「7」" in listed.message and "「1」" not in listed.message
+    assert found(cache, "SELECT COUNT(*) FROM channels WHERE channel_type = '线上'") == [
+        ("unknown_code", "warning", "channels", "channel_type")]
+
+
+@pytest.mark.parametrize("sql", [
+    "SELECT COUNT(*) FROM visits WHERE status = 1",
+    "SELECT COUNT(*) FROM visits WHERE status = '1'",
+    "SELECT COUNT(*) FROM visits WHERE 0 = status OR status = 1",
+    "SELECT COUNT(*) FROM visits WHERE status IN (0, 1)",
+    "SELECT COUNT(*) FROM visits WHERE status = 1.0",
+    "SELECT COUNT(*) FROM channels WHERE channel_type = '分销'",
+    # 模板和参数：值要到运行时才知道
+    "SELECT COUNT(*) FROM visits WHERE status = '{{ input.status }}'",
+    "SELECT COUNT(*) FROM visits WHERE status = {{ input.status }}",
+    "SELECT COUNT(*) FROM visits WHERE status = ?",
+    "SELECT COUNT(*) FROM visits WHERE status = :status",
+])
+async def test_unknown_code_negative(cache, sql):
+    assert "unknown_code" not in codes(cache, sql)
+
+
+async def test_unknown_code_level_follows_status(cache):
+    sql = "SELECT COUNT(*) FROM channels WHERE channel_type = '线上'"
+    assert found(cache, sql, "verified") == [("unknown_code", "warning", "channels", "channel_type")]
+    assert found(cache, sql, "proposed") == [("unknown_code", "info", "channels", "channel_type")]
+    assert found(cache, sql, "rejected") == []
+
+
+# --------------------------------------------------------------------------
+# wrong_date_column：按另一个时间列分组、筛选
+# --------------------------------------------------------------------------
+
+@pytest.mark.parametrize("sql", [
+    "SELECT substr(entered_at, 1, 10) AS d, SUM(fee) FROM parking_records GROUP BY substr(entered_at, 1, 10)",
+    "SELECT SUM(p.fee) FROM parking_records p WHERE p.entered_at >= '2026-09-01'",
+])
+async def test_wrong_date_column_positive(cache, sql):
+    [check] = run(cache, sql)
+    assert (check.code, check.level, check.table, check.column) == (
+        "wrong_date_column", "warning", "parking_records", "entered_at")
+    assert "业务日期按「出场时间」计" in check.message and "「入场时间」" in check.message
+
+
+@pytest.mark.parametrize("sql", [
+    "SELECT substr(exited_at, 1, 10) AS d, SUM(fee) FROM parking_records GROUP BY substr(exited_at, 1, 10)",
+    "SELECT SUM(fee) FROM parking_records WHERE exited_at >= '2026-09-01' AND entered_at IS NOT NULL",
+    # 两个时间都用上了：写的人知道两者的区别
+    "SELECT SUM(fee) FROM parking_records WHERE exited_at >= '2026-09-01' AND entered_at < '2026-09-30'",
+    "SELECT entered_at, exited_at, fee FROM parking_records",
+])
+async def test_wrong_date_column_negative(cache, sql):
+    assert "wrong_date_column" not in codes(cache, sql)
+
+
+async def test_wrong_date_level_follows_status(cache):
+    sql = "SELECT SUM(p.fee) FROM parking_records p WHERE p.entered_at >= '2026-09-01'"
+    assert found(cache, sql, "verified") == [("wrong_date_column", "warning", "parking_records", "entered_at")]
+    assert found(cache, sql, "proposed") == [("wrong_date_column", "info", "parking_records", "entered_at")]
+    assert found(cache, sql, "rejected") == []
+
+
+# --------------------------------------------------------------------------
+# 复杂写法：照样分析得出，或者跳过；不报错、不误报
+# --------------------------------------------------------------------------
+
+CLEAN_COMPLEX = [
+    # CTE + 窗口函数 + 子查询
+    """WITH daily AS (
+         SELECT substr(o.ordered_at, 1, 10) AS d, SUM(i.amount) AS amt
+         FROM orders o JOIN order_items i ON i.order_id = o.id
+         WHERE o.status = 1 GROUP BY substr(o.ordered_at, 1, 10))
+       SELECT d, amt, SUM(amt) OVER (ORDER BY d) AS running,
+              (SELECT COUNT(*) FROM visits v WHERE v.status = 1) AS visits
+       FROM daily ORDER BY d""",
+    # UNION
+    "SELECT 'web' AS ch, SUM(total_amount) FROM orders WHERE status = 1 AND channel_id = 1 "
+    "UNION ALL SELECT 'box', SUM(total_amount) FROM orders WHERE status = 1 AND channel_id = 3",
+    # 相关子查询、EXISTS
+    "SELECT o.id FROM orders o WHERE o.status = 1 AND EXISTS (SELECT 1 FROM order_items i WHERE i.order_id = o.id)",
+    # 引号括起来的驼峰标识符
+    'SELECT t.tag_name, COUNT(*) FROM member_tag_links l JOIN "memberTags" t ON t.id = l."memberTagId" '
+    "GROUP BY t.tag_name",
+]
+
+
+@pytest.mark.parametrize("sql", CLEAN_COMPLEX)
+async def test_complex_but_clean_sql_reports_nothing(cache, sql):
+    assert run(cache, sql) == []
+
+
+async def test_problems_inside_cte_and_union_are_found(cache):
+    cte = ("WITH t AS (SELECT o.channel_id, SUM(o.total_amount) AS amt FROM orders o "
+           "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 GROUP BY o.channel_id) SELECT * FROM t")
+    assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, cte)
+    union = ("SELECT SUM(on_hand) FROM inventory_snapshots UNION ALL "
+             "SELECT SUM(on_hand) FROM inventory_snapshots WHERE snapshot_date = '2026-09-30'")
+    assert found(cache, union) == [("stock_summed", "error", "inventory_snapshots", "on_hand")]
+
+
+@pytest.mark.parametrize("sql", [
+    "SELECT FROM WHERE",
+    "SELEC total_amount FROM orders",
+    "SELECT SUM(o.total_amount FROM orders o JOIN order_items i ON i.order_id = o.id",
+    "SELECT 1; SELECT SUM(on_hand) FROM inventory_snapshots",
+    "DELETE FROM visits",
+    "",
+    "   ",
+    # 方言特性：SQLite 读不了 Oracle 的 q 引号
+    "SELECT SUM(on_hand) FROM inventory_snapshots WHERE q'[x]' = 'x'",
+    # 模板里的控制语句：拼出来是什么样说不清
+    "SELECT SUM(on_hand) FROM inventory_snapshots {% if x %}WHERE snapshot_date = '2026-09-30'{% endif %}",
+    # 结构里没有的表、别名对不上的列
+    "SELECT SUM(x.on_hand) FROM inventory_snapshots s",
+    "SELECT SUM(on_hand) FROM not_a_table",
+])
+async def test_unparseable_or_unsupported_sql_is_skipped(cache, sql):
+    assert run(cache, sql) == []
+
+
+async def test_other_dialects(cache):
+    pg = "SELECT s.on_hand::numeric FROM inventory_snapshots s WHERE s.snapshot_date = $1"
+    assert run(cache, pg, kind="postgres") == []
+    assert found(cache, "SELECT SUM(s.on_hand)::numeric FROM inventory_snapshots s", kind="postgres") == [
+        ("stock_summed", "error", "inventory_snapshots", "on_hand")]
+    mysql = "SELECT `v`.`status`, COUNT(*) FROM `visits` `v` WHERE `v`.`status` = 3 GROUP BY `v`.`status`"
+    assert found(cache, mysql, kind="mysql") == [("unknown_code", "warning", "visits", "status")]
+    # 能读就照常分析，读不了就跳过：只要求不抛
+    assert isinstance(run(cache, "SELECT SUM(on_hand) FROM inventory_snapshots WHERE q'[it''s]' = 'x'",
+                          kind="oracle"), list)
+    assert run(cache, "SELECT SUM(on_hand) FROM inventory_snapshots", kind="mssql") == []
+
+
+async def test_no_catalog_means_no_checks(cache):
+    sql = "SELECT SUM(o.total_amount) FROM orders o JOIN order_items i ON i.id = o.id"
+    assert SqlChecker(kind="sqlite", schema_cache=cache, notes={}).check(sql) == []
+    assert check_sql(sql, kind="sqlite", schema_cache=cache, notes=None) == []
+    # 没探查过结构：对不上表，一样不判
+    assert check_sql(sql, kind="sqlite", schema_cache={}, notes=scenic_notes.notes()) == []
+
+
+async def test_duplicates_are_folded_and_errors_come_first(cache):
+    sql = ("SELECT SUM(o.total_amount), SUM(o.total_amount) * 2, AVG(l.discount_rate) "
+           "FROM orders o JOIN order_items i ON i.order_id = o.id JOIN members m ON m.id = o.member_id "
+           "JOIN member_levels l ON l.id = m.member_level_id WHERE o.status = 1")
+    out = found(cache, sql)
+    assert [f for f in out if f[0] == "fanout_sum"] == [("fanout_sum", "error", "orders", "total_amount")]
+    levels = [f[1] for f in out]
+    assert levels == sorted(levels, key=["error", "warning", "info"].index)
+
+
+async def test_checker_builds_relation_graph_lazily_and_reuses_it(cache, monkeypatch):
+    from app.data import catalog
+
+    calls = []
+    real = catalog.relation_graph
+    monkeypatch.setattr(catalog, "relation_graph", lambda *a, **k: calls.append(1) or real(*a, **k))
+    checker = SqlChecker(kind="sqlite", schema_cache=cache, notes=scenic_notes.notes())
+    checker.check("SELECT AVG(discount_rate) FROM member_levels")
+    assert calls == []                              # 单表查询用不到关系图
+    checker.check(FANOUT[0])
+    checker.check(FANOUT[1])
+    assert calls == [1]