数据目录 F3 / B1 服务端：SQL 只写在 Agent 里的模板也提醒目录变化——发布记录分直接引用和可能涉及两组，Agent 和协作成员绑定了查询工具的源记下所有有目录的表，正式运行两组都比、每张表标明属于哪一组；只记了直接引用的旧记录照旧按直接引用比；哪些节点算写死的查询统一交给 tool_queries

backend/app/api/runs.py

@@ -186,7 +186,8 @@ async def start_run(
         await run_manager.note(run.id, "log", **governance_note(
             governed=governed, workflow_name=name, version=version))
     if catalog_changes:
-        # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表
+        # 落在封存范围内：之后回看这次正式出具，能知道当时的目录和发布时不一样、差在哪几张表。
+        # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的（界面分开说）
         await run_manager.note(run.id, "catalog.drift", version=version, count=len(catalog_changes),
                                tables=catalog_changes)
     # 已声明策略的升版记入事件流，出具物上能看到"这期换口径了、怎么处置的"

backend/app/api/workflows.py

@@ -10,7 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.core.errors import graph_error
 from app.api.runs import actor_of
 from app.core.artifact_store import graph_hash
-from app.data.catalog_impact import catalog_drift, catalog_versions_for
+from app.data.catalog_impact import catalog_drift, catalog_versions_for, recorded_versions
 from app.db.base import get_session
 from app.db.models import Run, Workflow, WorkflowVersion
 from app.engine.governance import publish_issues
@@ -210,9 +210,11 @@ class VersionDetailOut(VersionOut):
     #: 画布上改过入口字段时，拿当前画布的字段去跑已发布版本会传错参数
     input_fields: list[dict[str, Any]] = Field(default_factory=list)
     published: bool = False
-    #: 发布时这一版 SQL 用到的表的数据目录版本 {源名: {表名: 版本}}；没发布过、或者记录之前发布的为 null
+    #: 发布时记下的数据目录版本 {"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}：direct 写死的
+    #: SQL 用到的表，possible Agent 可能查询的表（catalog_impact.recorded_versions，只记了直接引用的旧记录 possible
+    #: 为空）；没发布过、或者记录之前发布的为 null
     catalog_versions: dict[str, Any] | None = None
-    #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current}]
+    #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current, impact}]
     catalog_changes: list[dict[str, Any]] = Field(default_factory=list)
 
 
@@ -248,7 +250,7 @@ async def get_version(
         created_at=snapshot.created_at, workflow_id=workflow_id, graph=graph,
         graph_hash=snapshot.graph_hash or graph_hash(graph), input_fields=fields,
         published=workflow.published_version == snapshot.version, level=snapshot.level,
-        catalog_versions=snapshot.catalog_versions,
+        catalog_versions=recorded_versions(snapshot.catalog_versions),
         catalog_changes=await catalog_drift(session, snapshot.catalog_versions),
     )
 
@@ -352,7 +354,7 @@ async def publish_workflow(
         "version": version,
         "graph_hash": snapshot.graph_hash,
         "published_by": workflow.published_by,
-        "catalog_versions": snapshot.catalog_versions,
+        "catalog_versions": recorded_versions(snapshot.catalog_versions),
         "issues": issues,  # 剩下的都是警告
     }
 

backend/app/core/events.py

@@ -69,7 +69,8 @@ class EventType(StrEnum):
     # 出具与口径治理
     ISSUANCE = "issuance"  # 出具契约的判定结果（档位 / 缺数据 / 未回指数字）
     CALIBER_UPGRADE = "caliber.upgrade"  # 钉住的方法卡有新版本，按声明的策略处置
-    # 从发布版本发起的正式运行：发布之后这一版 SQL 用到的表的数据目录有变化（data/catalog_impact.catalog_drift）。
+    # 从发布版本发起的正式运行：发布之后这一版用到的表的数据目录有变化（data/catalog_impact.catalog_drift）。
+    # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的。
     # 只提醒、不拦运行；落在封存范围内，出具物上看得到「当时的目录和发布时不一样」
     CATALOG_DRIFT = "catalog.drift"
     # 报告撰写节点的自查结果：报告文档工件 id、统计、违规清单。落在封存范围内，

backend/app/data/catalog_impact.py

@@ -5,17 +5,20 @@
 - **影响面**：哪些已发布、受管的模板用到了这张表（table_impact）。按每个模板当前的已发布版本算，不看画布上的
   草稿——正式运行跑的是已发布的那一版。调用工具节点写死的 SQL 用到了这张表算「直接引用」；Agent 绑定了这个源
   的查询工具算「可能涉及」（SQL 是运行时写的，静态看不出）；合并查询按它的输入一路追溯。
-- **发布时的目录版本**：发布时记下模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}（catalog_versions_for），
-  从这一版发起正式运行时和当前的比（catalog_drift），有变化只提醒、不拦运行——目录是给人和助手看的说明，
-  不是运行的输入；拦下来等于让一次文字修订卡住正式出具。
+- **发布时的目录版本**：发布时记下模板用到的表的目录版本（catalog_versions_for），和影响面同一个说法分两组：
+  {"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。direct 是写死的 SQL 用到的表；possible 是
+  Agent、协作成员绑定了查询工具的源里所有有目录的表。从这一版发起正式运行时和当前的比（catalog_drift），有变化
+  只提醒、不拦运行——目录是给人和助手看的说明，不是运行的输入；拦下来等于让一次文字修订卡住正式出具。
 
 表名的取法和证据台账同一个口径：evidence.sql_tables 取 FROM / JOIN 后面的表（去掉注释、字符串、CTE），
-去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。
+去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。哪些节点算「写死的查询」以
+sqlcheck.tool_query 为准（tool_queries），发布记录、影响面、SQL 检查和助手的上下文认的是同一种节点。
 """
 from __future__ import annotations
 
 import logging
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Iterator, Mapping
+from types import SimpleNamespace
 from typing import Any
 
 from sqlalchemy import select
@@ -51,20 +54,43 @@ def sql_table_keys(sql: str, tables: Iterable[str] | None) -> set[str]:
     return {hit for s in shorts if (hit := lookup.get(name_key(s))) is not None}
 
 
+def _as_node(raw: Mapping[str, Any]) -> Any:
+    """字典写法的节点 → tool_query 认的样子（有 type、config）。画布 {data: {config}} 和 GraphSpec {config} 都认。"""
+    config = raw.get("config")
+    if not isinstance(config, Mapping):
+        data = raw.get("data")
+        config = data.get("config") if isinstance(data, Mapping) else None
+    return SimpleNamespace(id=raw.get("id"), type=raw.get("type"), config=config if isinstance(config, Mapping) else {})
+
+
+def tool_queries(nodes: Iterable[Any]) -> Iterator[tuple[Any, str, str]]:
+    """一张图里调用工具节点写死的查询：(节点, 源名, SQL)，按图里的顺序。
+
+    判定一律交给 sqlcheck.tool_query（是不是调用工具节点、是不是数据源查询工具、SQL 是不是非空字符串）：发布时
+    记目录版本、影响面、SQL 检查、助手给全字段的表都走这一份，不各写一遍。节点认 GraphSpec 的节点对象，也认还没
+    校验过的字典（画布写法、GraphSpec 写法）——助手改图时手上的是字典，图不一定完整到能过 GraphSpec。
+    """
+    from app.data.sqlcheck import tool_query
+
+    for node in nodes or []:
+        if isinstance(node, Mapping):
+            found = tool_query(_as_node(node))
+        elif node is not None:
+            found = tool_query(node)
+        else:
+            found = None
+        if found:
+            yield node, found[0], found[1]
+
+
 def graph_sql_tables(spec: Any, tables_by_source: Mapping[str, Iterable[str]] | None = None) -> dict[str, set[str]]:
-    """一张图里调用工具节点写死的 SQL 用到的表：{源名: {表名}}。Agent 的 SQL 运行时才有，不在这里。
+    """一张图里调用工具节点写死的 SQL 用到的表：{源名: {表名}}。Agent 的 SQL 运行时才有，不在这里（见 agent_sources）。
 
     tables_by_source 是各源表结构里的表名（schema_cache 的 tables 的键）：给了的源按它规范表名、对不上的不算；
     没给的源用 SQL 里的原名。
     """
-    from app.data.sqlcheck import tool_query
-
     out: dict[str, set[str]] = {}
-    for node in getattr(spec, "nodes", []) or []:
-        found = tool_query(node)
-        if not found:
-            continue
-        source, sql = found
+    for _, source, sql in tool_queries(getattr(spec, "nodes", []) or []):
         known = (tables_by_source or {}).get(source)
         hits = sql_table_keys(sql, list(known) if known is not None else None)
         if hits:
@@ -82,6 +108,36 @@ def _setting(spec: Any, node: Any, key: str) -> Any:
     return (getattr(spec, "defaults", None) or {}).get(key) if value in (None, "") else value
 
 
+def _query_sources(tools: Any) -> set[str]:
+    """工具清单里绑定的数据源查询工具（db_query__<源>）→ 源名。"""
+    from app.tools.datasource import QUERY_PREFIX
+
+    names = [t.strip() for t in tools if isinstance(t, str)] if isinstance(tools, (list, tuple)) else []
+    return {t[len(QUERY_PREFIX):] for t in names if t.startswith(QUERY_PREFIX) and len(t) > len(QUERY_PREFIX)}
+
+
+def agent_bindings(spec: Any) -> list[tuple[Any, str | None, set[str]]]:
+    """Agent 节点和协作成员各自绑定了哪些源的查询工具：[(节点, 成员名或 None, {源名})]，按图里的顺序。
+
+    它们的 SQL 运行时才由模型写出来，静态看不出用哪张表：影响面算「可能涉及」，发布时记下整个源有目录的表。
+    """
+    out: list[tuple[Any, str | None, set[str]]] = []
+    for node in getattr(spec, "nodes", []) or []:
+        kind = _kind(node)
+        if kind == "agent":
+            out.append((node, None, _query_sources(_setting(spec, node, "tools"))))
+        elif kind == "supervisor":
+            for member in _setting(spec, node, "agents") or []:
+                if isinstance(member, Mapping):
+                    out.append((node, str(member.get("name") or ""), _query_sources(member.get("tools"))))
+    return out
+
+
+def agent_sources(spec: Any) -> set[str]:
+    """Agent 节点、协作成员绑定了查询工具的数据源名。"""
+    return {name for _, _, bound in agent_bindings(spec) for name in bound}
+
+
 def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | None) -> list[dict[str, Any]]:
     """一张图里哪些节点用到了某个源的某张表，按图里的顺序。每项 {node_id, label, type, impact, via?, member?}。
 
@@ -90,10 +146,6 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No
     - 合并查询：输入（别名 → 节点）里有用到这张表的 → 取输入里最强的那一级，via 写经由哪些输入。合并的合并
       一路追到底；成环（校验会报错的图）时停下。
     """
-    from app.data.sqlcheck import tool_query
-    from app.tools.datasource import QUERY_PREFIX
-
-    tool = f"{QUERY_PREFIX}{source}"
     wanted = name_key(table)
     known = list(tables) if tables is not None else None
     nodes = list(getattr(spec, "nodes", []) or [])
@@ -104,20 +156,16 @@ def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | No
     def base(node: Any, impact: str, **extra: Any) -> dict[str, Any]:
         return {"node_id": node.id, "label": node.title, "type": _kind(node), "impact": impact, **extra}
 
-    for node in nodes:
-        kind = _kind(node)
-        if kind == "tool":
-            found = tool_query(node)
-            if found and found[0] == source and wanted in {name_key(t) for t in sql_table_keys(found[1], known)}:
-                own[node.id] = base(node, "direct")
-        elif kind == "agent":
-            if tool in [str(t).strip() for t in _setting(spec, node, "tools") or [] if isinstance(t, str)]:
-                own[node.id] = base(node, "possible")
-        elif kind == "supervisor":
-            for member in _setting(spec, node, "agents") or []:
-                if isinstance(member, Mapping) and tool in [str(t).strip() for t in member.get("tools") or []
-                                                            if isinstance(t, str)]:
-                    out.append(base(node, "possible", member=str(member.get("name") or "")))
+    for node, name, sql in tool_queries(nodes):
+        if name == source and wanted in {name_key(t) for t in sql_table_keys(sql, known)}:
+            own[node.id] = base(node, "direct")
+    for node, member, bound in agent_bindings(spec):
+        if source not in bound:
+            continue
+        if member is None:
+            own[node.id] = base(node, "possible")
+        else:
+            out.append(base(node, "possible", member=member))
 
     resolved: dict[str, dict[str, Any] | None] = {}
 
@@ -224,66 +272,135 @@ def _schema_tables(view: Any) -> list[str] | None:
     return list(tables) if isinstance(tables, Mapping) and tables else None
 
 
-async def catalog_versions_for(session: AsyncSession, spec: Any) -> dict[str, dict[str, int]]:
-    """发布时要记下的目录版本：{源名: {表名: 版本}}，只收模板 SQL 用到的表；还没有目录的表记 0。
+#: 发布记录的两组，和影响面的两级同名：direct 写死的 SQL 用到的表，possible Agent 可能查询的表
+CatalogVersions = dict[str, dict[str, dict[str, int]]]
 
-    记 0 是有意的：发布之后有人给这张表补了目录，同样算「自发布以来有变化」。找不到的源（停用、删了）不记：
-    没有目录可比，正式运行时工具自己会报源不存在。读库出错时返回空 dict——记不下版本不该挡住发布。
+
+def _pinned_group(value: Any) -> dict[str, dict[str, Any]] | None:
+    """{源名: {表名: 版本}} 的一组；形状不对返回 None。"""
+    if not isinstance(value, Mapping) or not all(isinstance(t, Mapping) for t in value.values()):
+        return None
+    return {str(k): dict(v) for k, v in value.items()}
+
+
+def recorded_versions(recorded: Any) -> CatalogVersions | None:
+    """发布时记下的目录版本，统一成 {"direct": {...}, "possible": {...}}。没记过返回 None。
+
+    认两种写法：
+    - 现在的写法：顶层只有 direct、possible 两个键，每组 {源名: {表名: 版本}}。
+    - 可能涉及上线之前的写法：{源名: {表名: 版本}}，只记了写死的 SQL 用到的表——整个算 direct，possible 为空：
+      那时没记 Agent 可能查询的表，不知道当时是哪一版，不能说它们变了。
+    两种写法靠第二层分得开：旧写法第二层是版本号（整数），新写法第二层是 {表名: 版本}。源名可以叫 direct，
+    所以不能只看顶层的键。
+    """
+    if not isinstance(recorded, Mapping):
+        return None
+    if recorded and set(recorded) <= set(IMPACTS):
+        groups = {k: _pinned_group(recorded.get(k, {})) for k in IMPACTS}
+        if all(g is not None for g in groups.values()):
+            return {k: g or {} for k, g in groups.items()}
+    legacy = _pinned_group(recorded)
+    return {"direct": legacy, "possible": {}} if legacy is not None else None
+
+
+async def catalog_versions_for(session: AsyncSession, spec: Any) -> CatalogVersions | None:
+    """发布时要记下的目录版本：{"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。
+
+    - direct：调用工具节点写死的 SQL 用到的表；还没有目录的表记 0。记 0 是有意的：发布之后有人给这张表补了
+      目录，同样算「自发布以来有变化」。
+    - possible：Agent 节点、协作成员绑定了某个源的查询工具（影响面里的「可能涉及」）。SQL 运行时才写，看不出用哪
+      张表，于是记下这个源里**所有有目录的表**；direct 里已经有的不重复记。发布时还没有目录的表不逐张记 0（一个
+      库几百张表），比对时「之后新建了目录」照样算变化（catalog_drift）。源里一张有目录的表都没有时记一个空的
+      {}，表示这个源在看着。
+
+    找不到的源（停用、删了）不记：没有目录可比，正式运行时工具自己会报源不存在。读库出错时返回 None（和没记过
+    一样，不提醒）——记不下版本不该挡住发布。
     """
     from app.data import catalog
 
     try:
-        names = graph_sql_tables(spec).keys()
-        sources = await _sources_by_name(session, names)
+        possible_names = agent_sources(spec)
+        sources = await _sources_by_name(session, [*graph_sql_tables(spec).keys(), *possible_names])
         tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()}
         used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None})
-        out: dict[str, dict[str, int]] = {}
+        entries_of: dict[str, dict[str, Any]] = {}
+
+        async def entries(name: str) -> dict[str, Any]:
+            if name not in entries_of:
+                entries_of[name] = await catalog.read_catalog(session, sources[name][0].id)
+            return entries_of[name]
+
+        direct: dict[str, dict[str, int]] = {}
         for name, tables in sorted(used.items()):
-            if name not in sources:
-                continue
-            entries = await catalog.read_catalog(session, sources[name][0].id)
-            out[name] = {t: (entries[t].version if t in entries else 0) for t in sorted(tables)}
-        return out
+            if name in sources:
+                found = await entries(name)
+                direct[name] = {t: (found[t].version if t in found else 0) for t in sorted(tables)}
+        possible: dict[str, dict[str, int]] = {}
+        for name in sorted(possible_names):
+            if name in sources:
+                pinned = direct.get(name, {})
+                possible[name] = {t: e.version for t, e in sorted((await entries(name)).items()) if t not in pinned}
+        return {"direct": direct, "possible": possible}
     except Exception:  # noqa: BLE001
         logger.exception("记录发布时的目录版本失败，本次不记")
-        return {}
+        return None
+
+
+def _label(entry: Any) -> str | None:
+    """表现在的中文名；没有或被驳回时为 None。"""
+    label = (entry.notes.get("label") if entry else None) or {}
+    return label.get("value") if isinstance(label, dict) and label.get("status") != "rejected" else None
 
 
 async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]:
-    """发布时记下的目录版本和现在比：变了的表 [{source, source_id, table, label, published, current}]。
+    """发布时记下的目录版本和现在比：变了的表 [{source, source_id, table, label, published, current, impact}]。
 
-    没记过（这个功能之前发布的版本）返回空列表——不知道当时是哪一版，不能说它变了。现在找不到的源跳过（理由同
-    catalog_versions_for）。label 是表现在的中文名（没有或被驳回时为 None）。读库出错返回空列表：提醒是附加的。
+    impact 是这张表属于哪一组（direct 直接引用 / possible Agent 可能查询的），direct 在前，同组按源名、表名。
+    possible 的源里发布时没记、现在有了目录的表也算（published 为 0），direct 里已有的表不重复报。
+
+    没记过（这个功能之前发布的版本）返回空列表——不知道当时是哪一版，不能说它变了；只记了直接引用的旧记录
+    同理不报可能涉及（recorded_versions）。现在找不到的源跳过（理由同 catalog_versions_for）。label 是表现在的
+    中文名（没有或被驳回时为 None）。读库出错返回空列表：提醒是附加的。
     """
     from app.data import catalog
 
-    if not isinstance(recorded, Mapping) or not recorded:
+    groups = recorded_versions(recorded)
+    if not groups or not any(groups.values()):
         return []
     try:
-        sources = await _sources_by_name(session, recorded.keys())
+        sources = await _sources_by_name(session, [*groups["direct"], *groups["possible"]])
+        entries_of: dict[str, dict[str, Any]] = {}
+
+        async def entries(name: str) -> dict[str, Any]:
+            if name not in entries_of:
+                entries_of[name] = await catalog.read_catalog(session, sources[name][0].id)
+            return entries_of[name]
+
         out: list[dict[str, Any]] = []
-        for name in sorted(recorded):
-            pinned = recorded[name]
-            if name not in sources or not isinstance(pinned, Mapping):
-                continue
-            row = sources[name][0]
-            entries = await catalog.read_catalog(session, row.id)
-            for table in sorted(pinned):
-                then = pinned[table]
-                entry = entries.get(table)
-                now = entry.version if entry else 0
-                if isinstance(then, int) and then == now:
+        for impact in IMPACTS:
+            for name in sorted(groups[impact]):
+                if name not in sources:
                     continue
-                label = (entry.notes.get("label") if entry else None) or {}
-                out.append({"source": name, "source_id": row.id, "table": table,
-                            "label": label.get("value") if isinstance(label, dict) and label.get("status") != "rejected"
-                            else None,
-                            "published": then if isinstance(then, int) else None, "current": now})
+                pinned = dict(groups[impact][name])
+                found = await entries(name)
+                if impact == "possible":
+                    # 发布时没有目录、之后新建了的表：记录里没有它，按发布时「尚无目录」算
+                    skip = groups["direct"].get(name, {})
+                    pinned.update({t: 0 for t in found if t not in pinned and t not in skip})
+                for table in sorted(pinned):
+                    then = pinned[table]
+                    entry = found.get(table)
+                    now = entry.version if entry else 0
+                    if isinstance(then, int) and then == now:
+                        continue
+                    out.append({"source": name, "source_id": sources[name][0].id, "table": table,
+                                "label": _label(entry), "published": then if isinstance(then, int) else None,
+                                "current": now, "impact": impact})
         return out
     except Exception:  # noqa: BLE001
         logger.exception("比对发布时的目录版本失败，本次不提醒")
         return []
 
 
-__all__ = ["IMPACTS", "catalog_drift", "catalog_versions_for", "graph_sql_tables", "sql_table_keys", "table_impact",
-           "template_refs"]
+__all__ = ["IMPACTS", "CatalogVersions", "agent_bindings", "agent_sources", "catalog_drift", "catalog_versions_for",
+           "graph_sql_tables", "recorded_versions", "sql_table_keys", "table_impact", "template_refs", "tool_queries"]

backend/tests/test_catalog_publish_drift.py

@@ -1,10 +1,14 @@
 """发布时记下目录版本，正式运行时提示目录的变化（阶段 4A）。
 
-- 发布（把某一版立为正式版本）时，记下这个模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}，存在
-  workflow_versions.catalog_versions；还没有目录的表记 0（之后补了目录也算变化）。
-- 从这一版发起正式运行时和当前的目录版本比：有变化发一条 catalog.drift 事件（列出表和前后版本），**不拦运行**；
-  目录没变不发。
+- 发布（把某一版立为正式版本）时，记下这个模板用到的表的目录版本，存在 workflow_versions.catalog_versions，
+  分两组：{"direct": {源名: {表名: 版本}}, "possible": {源名: {表名: 版本}}}。
+  - direct：调用工具节点写死的 SQL 用到的表；还没有目录的表记 0（之后补了目录也算变化）。
+  - possible：Agent 节点、协作成员绑定了某个源的查询工具——SQL 运行时才写，发布时记下这个源里所有有目录的表
+    （直接引用的不重复记）。之后新建了目录的表同样算变化。
+- 从这一版发起正式运行时和当前的目录版本比：有变化发一条 catalog.drift 事件（列出表、前后版本、属于哪一组），
+  **不拦运行**；目录没变不发。
 - 版本详情给出发布时的目录版本，以及哪些表在那之后变了。
+- 没有记录的老版本不提醒；只有直接引用那一种写法（{源名: {表名: 版本}}）的记录照旧按直接引用比，不提醒可能涉及。
 """
 from __future__ import annotations
 
@@ -81,18 +85,52 @@ def node(nid, ntype, **config):
     return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
 
 
+def _chain(*nodes) -> dict:
+    return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+
+
 def daily(source: str) -> dict:
-    nodes = [
+    return _chain(
         node("start", "input"),
         node("fetch", "tool", tool=f"db_query__{source}", args={"sql": SQL}),
         node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.fetch }}"}]),
-    ]
-    return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
+    )
 
 
-async def _publish(client, source) -> str:
+def agent_only(source: str) -> dict:
+    """SQL 只写在 Agent 里：图上没有一条写死的查询。"""
+    return _chain(
+        node("start", "input"),
+        node("ask", "agent", prompt="查一下昨天各景区的入园人数", tools=[f"db_query__{source}"]),
+        node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.ask.text }}"}]),
+    )
+
+
+def team(source: str) -> dict:
+    """协作成员里有一个绑定了查询工具。"""
+    return _chain(
+        node("start", "input"),
+        node("crew", "supervisor", agents=[
+            {"name": "取数员", "prompt": "查入园人数", "tools": [f"db_query__{source}"]},
+            {"name": "撰稿人", "prompt": "写一句总结", "tools": []},
+        ]),
+        node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.crew.text }}"}]),
+    )
+
+
+def mixed(source: str) -> dict:
+    """写死的查询用到入园记录和景区，Agent 还绑着同一个源。"""
+    return _chain(
+        node("start", "input"),
+        node("fetch", "tool", tool=f"db_query__{source}", args={"sql": SQL}),
+        node("ask", "agent", prompt="再看看别的表", tools=[f"db_query__{source}"]),
+        node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.ask.text }}"}]),
+    )
+
+
+async def _publish(client, source, graph: dict | None = None) -> str:
     wf = (await client.post("/api/workflows", json={"name": f"入园日报-{uuid.uuid4().hex[:6]}",
-                                                    "graph": daily(source.name)})).json()["id"]
+                                                    "graph": graph or daily(source.name)})).json()["id"]
     source.made.append(wf)
     body = (await client.post(f"/api/workflows/{wf}/publish", json={"level": "published"})).json()
     assert body["ok"] is True, body
@@ -106,12 +144,13 @@ async def _events(run_id: str, kind: str) -> list[dict]:
         return [e.data for e in rows]
 
 
-async def _settle(run_id: str) -> Run:
+async def _settle(run_id: str, *, sealed: bool = True) -> Run:
+    """等运行结束。sealed=False 时不等封存：Agent 没有可用的模型，跑到它那一步就失败，只看开头的事件。"""
     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:
+        if row.status in ("succeeded", "failed") and (row.manifest_seq is not None or not sealed):
             return row
     raise AssertionError(f"run {run_id} 没跑完：{row.status}")
 
@@ -125,15 +164,35 @@ async def _edit_visits(source_id: str) -> int:
         return saved.version
 
 
+async def _recorded(wf: str) -> dict | None:
+    async with SessionLocal() as session:
+        return (await session.execute(select(WorkflowVersion.catalog_versions).where(
+            WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one()
+
+
+async def _catalogued(source_id: str) -> dict[str, int]:
+    """这个源现在所有有目录的表：{表名: 版本}。"""
+    async with SessionLocal() as session:
+        return {t: e.version for t, e in (await catalog.read_catalog(session, source_id)).items()}
+
+
+async def _formal(client, wf: str, *, sealed: bool = True) -> str:
+    r = await client.post("/api/runs", json={"workflow_id": wf, "run_class": "formal"})
+    assert r.status_code == 201, r.text                 # 不拦
+    run_id = r.json()["id"]
+    await _settle(run_id, sealed=sealed)
+    return run_id
+
+
 async def test_publish_records_catalog_versions(client, source):
     wf = await _publish(client, source)
     async with SessionLocal() as session:
         snap = (await session.execute(select(WorkflowVersion).where(
             WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one()
-    # 入园记录有目录（第 1 版）；景区表还没有目录，记 0
-    assert snap.catalog_versions == {source.name: {"parks": 0, "visits": 1}}
+    # 入园记录有目录（第 1 版）；景区表还没有目录，记 0。没有 Agent 查库，可能涉及的一组是空的
+    assert snap.catalog_versions == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}}
     detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
-    assert detail["catalog_versions"] == {source.name: {"parks": 0, "visits": 1}}
+    assert detail["catalog_versions"] == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}}
     assert detail["catalog_changes"] == []
 
 
@@ -149,7 +208,7 @@ async def test_formal_run_warns_on_catalog_change_but_runs(client, source):
     assert drift["count"] == 1 and drift["version"] == 1
     [table] = drift["tables"]
     assert table == {"source": source.name, "source_id": source.id, "table": "visits", "label": "入园记录",
-                     "published": 1, "current": now}
+                     "published": 1, "current": now, "impact": "direct"}
     # 版本详情也写明哪些表在发布之后变了
     detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
     assert [(c["table"], c["published"], c["current"]) for c in detail["catalog_changes"]] == [("visits", 1, now)]
@@ -167,6 +226,88 @@ async def test_catalog_added_after_publish_counts_as_change(client, source):
     assert [(t["table"], t["published"], t["current"], t["label"]) for t in drift["tables"]] == [("parks", 0, 1, "景区")]
 
 
+# --------------------------------------------------------------------------
+# SQL 只写在 Agent 里：可能涉及的表
+# --------------------------------------------------------------------------
+
+
+async def test_agent_only_template_records_every_catalogued_table_as_possible(client, source):
+    """影响面把绑定了查询工具的 Agent 算作「可能涉及」，发布记录得是同一个说法：记下这个源所有有目录的表。"""
+    wf = await _publish(client, source, agent_only(source.name))
+    recorded = await _recorded(wf)
+    assert recorded == {"direct": {}, "possible": {source.name: await _catalogued(source.id)}}
+    assert "visits" in recorded["possible"][source.name] and "parks" not in recorded["possible"][source.name]
+    detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
+    assert detail["catalog_versions"] == recorded and detail["catalog_changes"] == []
+
+
+async def test_agent_only_template_warns_after_a_catalog_change(client, source):
+    wf = await _publish(client, source, agent_only(source.name))
+    now = await _edit_visits(source.id)
+    run_id = await _formal(client, wf, sealed=False)
+    [drift] = await _events(run_id, "catalog.drift")
+    assert drift["count"] == 1
+    assert drift["tables"] == [{"source": source.name, "source_id": source.id, "table": "visits", "label": "入园记录",
+                                "published": 1, "current": now, "impact": "possible"}]
+    detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
+    assert [(c["table"], c["impact"]) for c in detail["catalog_changes"]] == [("visits", "possible")]
+
+
+async def test_catalog_created_after_publish_counts_for_possible_tables(client, source):
+    """发布时没有目录的表不逐张记 0（一个库几百张表），之后有人给它建了目录同样算变化。"""
+    wf = await _publish(client, source, agent_only(source.name))
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, source.id, "parks", {"label": catalog.make_item("景区", "human")},
+                                  if_version=0, actor="王敏")
+    run_id = await _formal(client, wf, sealed=False)
+    [drift] = await _events(run_id, "catalog.drift")
+    assert [(t["table"], t["published"], t["current"], t["impact"]) for t in drift["tables"]] == [
+        ("parks", 0, 1, "possible")]
+
+
+async def test_team_members_with_the_query_tool_count_as_possible(client, source):
+    wf = await _publish(client, source, team(source.name))
+    assert await _recorded(wf) == {"direct": {}, "possible": {source.name: await _catalogued(source.id)}}
+
+
+async def test_direct_tables_are_not_repeated_as_possible(client, source):
+    """同一个源既有写死的查询、又有 Agent：写死的 SQL 用到的表按直接引用记，可能涉及里不再记一遍。"""
+    wf = await _publish(client, source, mixed(source.name))
+    recorded = await _recorded(wf)
+    assert recorded["direct"] == {source.name: {"parks": 0, "visits": 1}}
+    rest = {t: v for t, v in (await _catalogued(source.id)).items() if t not in ("parks", "visits")}
+    assert recorded["possible"] == {source.name: rest} and rest
+    now = await _edit_visits(source.id)
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, source.id, "parks", {"label": catalog.make_item("景区", "human")},
+                                  if_version=0, actor="王敏")
+    run_id = await _formal(client, wf, sealed=False)
+    [drift] = await _events(run_id, "catalog.drift")
+    # 两张都是直接引用：parks 新建了目录也不会再按可能涉及报一遍
+    assert [(t["table"], t["published"], t["current"], t["impact"]) for t in drift["tables"]] == [
+        ("parks", 0, 1, "direct"), ("visits", 1, now, "direct")]
+
+
+async def test_direct_only_records_from_before_still_warn_as_direct(client, source):
+    """可能涉及上线之前发布的版本只记了直接引用（{源名: {表名: 版本}}）：照旧按直接引用比，不补可能涉及。"""
+    wf = await _publish(client, source, mixed(source.name))
+    async with SessionLocal() as session:
+        snap = (await session.execute(select(WorkflowVersion).where(
+            WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one()
+        snap.catalog_versions = {source.name: {"parks": 0, "visits": 1}}
+        await session.commit()
+    detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json()
+    assert detail["catalog_versions"] == {"direct": {source.name: {"parks": 0, "visits": 1}}, "possible": {}}
+    now = await _edit_visits(source.id)
+    async with SessionLocal() as session:
+        await catalog.write_entry(session, source.id, "gates", {"label": catalog.make_item("闸机", "human")},
+                                  if_version=0, actor="王敏")
+    run_id = await _formal(client, wf, sealed=False)
+    [drift] = await _events(run_id, "catalog.drift")
+    assert [(t["table"], t["impact"]) for t in drift["tables"]] == [("visits", "direct")]
+    assert drift["tables"][0]["current"] == now
+
+
 async def test_no_change_no_warning(client, source):
     wf = await _publish(client, source)
     r = await client.post("/api/runs", json={"workflow_id": wf, "run_class": "formal"})