目录视图接缝 data/catalog_view.py（细案 01 做法 5，总方案 X7）：读目录的四处统一经它，「开目录 / 关目录 / 只用已核实的项 / 目录快照」的对照才做得了

- CatalogView{mode: live | snapshot | off, statuses: all | reviewed, snapshot}：缺省 live + all，行为和以前一样；reviewed 只留已核实、已确认的项（驳回的项原样留着交给各处，照旧经 visible_notes 去掉，冻结快照和 SQL 检查靠驳回的关系编号认出不该现推的关系——为此在 catalog.py 加 restrict_notes，没给 visible_notes 加参数）
- 视图从哪来：运行从 Run.options['catalog_view'] 读（catalog_view.of_run：RunContext.extra 里的发起快照，审批恢复、续跑沿用），ToolContext 多一个 catalog_view，节点（调用工具、llm/agent、多 agent）构造时带上；生成过程从参数读（下一个提交接 answer_ops）
- 四处消费一律经它：运行时 db_schema / db_query（tools/datasource._catalog_of，冻结进表结构快照的也是这一份）、look 表详情（TableKind.load、copilot_context.look_service 收 catalog_view）、自查的 SQL 检查器（sqlcheck.load_checkers 收 catalog_view，改用 catalog.read_catalogs 一次读出）、双盲复算（_Sources 带视图，交给盲算的目录和比较 SQL 用的中文名都按它；核查 agent、反方审稿的 db_schema 也按这次运行的视图）
- 冻结快照：不是缺省视图的在引用 meta 里记下 catalog_view（不进内容哈希，缺省视图不记，快照和以前一字不差），证据面板据此写得出「本次运行目录：关」
- snapshot 模式只定读法（kind=eval_catalog_snapshot 的工件，{sources: {源 id: {tables: {表: {version, notes}}}}}），内容由第 2 波的题集包写；快照不在了照实报错（CatalogSnapshotMissing），不悄悄退回当前目录
- 新增 tests/test_catalog_view.py：四种视图在四处给的是同一份、缺省等于当前目录、运行从 options 读且冻结 meta 记下、lint（消费模块不许直接 read_catalog / read_entry / 查 CatalogNote）；去掉 _catalog_of 里的视图回跑两项失败

backend/app/api/copilot_context.py

@@ -65,16 +65,17 @@ SYSTEM_LOOK_UPGRADE_NEED = "要升级的工作流里，查库节点的 SQL 用
 
 
 async def look_service(sources: list[Any], *, spec: ModelSpec | None, graph: Mapping[str, Any] | None = None,
-                       kinds: tuple[str, ...] = ("table", "capability"), answer: bool = False) -> LookService:
+                       kinds: tuple[str, ...] = ("table", "capability"), answer: bool = False,
+                       catalog_view: Any = None) -> LookService:
     """按入口组装这一轮的查看服务。读目录、读能力登记都用各自的短会话，开流前后都能调。
 
     spec 是助手的模型规格：挑选模型第一次要用时才按它拿（关思考、额度按 look/pick.PICK_MAX_TOKENS）。graph 是现有工作流，概览里列出它查过的表。
-    answer：问数据（意图为回答），能力只列只读的。
+    answer：问数据（意图为回答），能力只列只读的。catalog_view：表详情里给哪一份数据目录（data/catalog_view.py），不给是当前目录。
     """
     picker = Picker(spec)
     made: list[Any] = []
     if "table" in kinds:
-        made.append(await TableKind.load(sources, graph=graph, picker=picker))
+        made.append(await TableKind.load(sources, graph=graph, picker=picker, catalog_view=catalog_view))
     if "capability" in kinds:
         made.append(await CapabilityKind.load(picker=picker, read_only=answer))
     return LookService(made)

backend/app/data/catalog.py

@@ -420,6 +420,18 @@ async def read_catalog(session: AsyncSession, source_id: str) -> dict[str, Catal
     return {row.table_name: _entry(row) for row in rows}
 
 
+async def read_catalogs(session: AsyncSession, source_ids: Iterable[str]) -> dict[str, dict[str, CatalogEntry]]:
+    """几个数据源的目录一次查询读出：{源 id: {表名: CatalogEntry}}，没有目录的源不出现。给模型、给自查读目录的地方经
+    data/catalog_view.py 调它，不直接查 catalog_notes。"""
+    ids = sorted({i for i in source_ids if i})
+    if not ids:
+        return {}
+    out: dict[str, dict[str, CatalogEntry]] = {}
+    for row in (await session.execute(select(CatalogNote).where(CatalogNote.source_id.in_(ids)))).scalars():
+        out.setdefault(row.source_id, {})[row.table_name] = _entry(row)
+    return out
+
+
 async def write_entry(session: AsyncSession, source_id: str, table: str, notes: dict[str, Any], *,
                       if_version: int, actor: str | None, commit: bool = True) -> CatalogEntry:
     """整份写入一张表的目录（乐观锁），提交事务，返回写入后的目录。
@@ -592,6 +604,14 @@ def visible_notes(notes: Mapping[str, Any] | None) -> dict[str, Any]:
     return _assemble({k: v for k, v in _slots(notes).items() if v.get("status") != "rejected"})
 
 
+def restrict_notes(notes: Mapping[str, Any] | None, statuses: Iterable[str]) -> dict[str, Any]:
+    """只留状态在 statuses 里的项（表级项、列级项、关系都按项算）；驳回的项原样留着——给人、给模型看的照旧经
+    visible_notes 去掉，冻结快照和 SQL 检查要靠驳回的关系编号认出不该按外键、命名现推回来的关系（frozen_catalog）。
+    目录视图「只用已核实、已确认的项」用它（data/catalog_view.py）。"""
+    keep = set(statuses) | {"rejected"}
+    return _assemble({k: v for k, v in _slots(notes).items() if v.get("status") in keep})
+
+
 def status_counts(notes: Mapping[str, Any] | None) -> dict[str, int]:
     """各状态的项数（表级项、列级项、关系都算），四种状态都有键。"""
     counts = dict.fromkeys(ITEM_STATUSES, 0)
@@ -2508,8 +2528,9 @@ __all__ = [
     "fk_relations", "frozen_catalog", "infer_name_relations", "initial_status", "is_date_column", "is_profile_note",
     "join_paths",
     "make_item", "merge_notes", "name_words", "now_iso", "parse_path", "plan_patch", "profile_relation_holds",
-    "query_snapshot_meta", "read_catalog", "read_entry", "relation_graph", "relation_id", "render_table_index",
-    "render_table_notes", "resolve_draft_model", "resolve_table_name", "review_entry", "review_item", "same_notes",
+    "query_snapshot_meta", "read_catalog", "read_catalogs", "read_entry", "relation_graph", "relation_id", "render_table_index",
+    "render_table_notes", "resolve_draft_model", "resolve_table_name", "restrict_notes", "review_entry", "review_item",
+    "same_notes",
     "same_profile_note", "save_human_edit", "save_patch", "status_counts", "structure_relation_ids",
     "system_notes_source", "table_usage", "unreviewed_status", "validate_notes", "visible_notes", "write_entry",
 ]

backend/app/data/catalog_view.py

@@ -0,0 +1,181 @@
+"""目录视图：运行和助手读数据目录统一经这一个接缝（docs/design/next-round/01-效果度量与回归评测.md 做法 5，总方案 X7）。
+
+**为什么要接缝。** 目录在四处被读：运行时的 db_schema / db_query 工具（目录同时冻结进表结构快照）、助手查看（look 表详情）、
+自查用的 SQL 检查器、双盲复算（交给盲算的说明、比较两条 SQL 用的中文名）。以前各读各的、没有统一的开关，「开目录和关
+目录」「只用已核实的项」「起草模型 A 的目录和 B 的」这些对照都做不了，评测期间目录被起草或剖析改了也不知道。现在四处都
+经这里取，一处定读哪一份；lint 测试（tests/test_catalog_view.py）不许这几个模块绕过它直接 read_catalog、直接查 CatalogNote。
+
+**视图**（CatalogView）：
+- mode：live 当前目录（缺省，行为和以前一样）；snapshot 一份冻结的目录快照（工件 id 在 snapshot，内容见 SNAPSHOT_KIND）；
+  off 不给目录。
+- statuses：all 全部（驳回的照旧不给模型看）；reviewed 只要已核实（verified）和已确认（confirmed）的项，推断的
+  （proposed）去掉——量的是模型起草、还没审过的那些项是帮忙还是添乱。
+
+**视图从哪来。** 运行从 Run.options['catalog_view'] 读（launch_run 收 options，发起时定下、随 run.started 封存，审批恢复、
+重启续跑都沿用——用上下文变量传的话恢复之后就丢了）；生成过程（问数据的 answer_ops、评测的作答）从参数读。都不给就是 live。
+冻结进表结构快照时，不是缺省视图的在工件引用的 meta 里记下视图（describe），证据面板据此写得出「本次运行目录：关」。
+
+**快照的内容**（第 2 波的评测题集包写，这里只定读法）：kind=eval_catalog_snapshot 的内容寻址工件，
+{"sources": {源 id: {"tables": {表名: {"version": 版本, "notes": 目录}}}}}。快照里没有这个源就是这个源没有目录；
+快照工件不在了照实报错（CatalogSnapshotMissing），不悄悄退回当前目录——那样对照就白做了。
+"""
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from app.data import catalog
+from app.data.catalog import CatalogEntry
+
+logger = logging.getLogger(__name__)
+
+#: Run.options 里放视图的键
+OPTION_KEY = "catalog_view"
+MODES = ("live", "snapshot", "off")
+#: 各状态集合留下哪些项的状态（all 不筛：驳回的由各处照旧经 visible_notes 去掉）
+STATUS_SETS: dict[str, tuple[str, ...] | None] = {"all": None, "reviewed": ("verified", "confirmed")}
+#: 目录快照工件的 kind（细案 01 做法 6）
+SNAPSHOT_KIND = "eval_catalog_snapshot"
+
+
+class CatalogViewInvalid(ValueError):
+    """视图写得不对（服务端代码写错了才会有：客户端写不了 Run.options）。"""
+
+
+class CatalogSnapshotMissing(RuntimeError):
+    """视图指定的目录快照工件不在了、或者内容认不出。"""
+
+
+@dataclass(frozen=True)
+class CatalogView:
+    mode: str = "live"
+    statuses: str = "all"
+    #: mode=snapshot 时快照工件的 id
+    snapshot: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.mode not in MODES:
+            raise CatalogViewInvalid(f"目录视图的 mode 只能是 {'、'.join(MODES)}，当前为「{self.mode}」")
+        if self.statuses not in STATUS_SETS:
+            raise CatalogViewInvalid(f"目录视图的 statuses 只能是 {'、'.join(STATUS_SETS)}，当前为「{self.statuses}」")
+        if (self.mode == "snapshot") != bool(self.snapshot):
+            raise CatalogViewInvalid("目录视图只有 mode=snapshot 时带快照工件 id，而且必须带")
+
+    @property
+    def is_default(self) -> bool:
+        return self.mode == "live" and self.statuses == "all"
+
+    def as_option(self) -> dict[str, Any]:
+        """写进 Run.options['catalog_view'] 的形状。"""
+        return {"mode": self.mode, "statuses": self.statuses, **({"snapshot": self.snapshot} if self.snapshot else {})}
+
+    @classmethod
+    def parse(cls, raw: Any) -> CatalogView:
+        """Run.options['catalog_view']、生成参数 → 视图。None 是缺省（live）；写得不对抛 CatalogViewInvalid。"""
+        if raw is None:
+            return LIVE
+        if isinstance(raw, CatalogView):
+            return raw
+        if not isinstance(raw, Mapping):
+            raise CatalogViewInvalid(f"目录视图需要是一个对象，当前为「{raw}」")
+        unknown = set(raw) - {"mode", "statuses", "snapshot"}
+        if unknown:
+            raise CatalogViewInvalid(f"目录视图里不认识的键：{'、'.join(sorted(unknown))}")
+        return cls(mode=str(raw.get("mode") or "live"), statuses=str(raw.get("statuses") or "all"),
+                   snapshot=str(raw["snapshot"]) if raw.get("snapshot") else None)
+
+
+LIVE = CatalogView()
+
+
+def of_options(options: Mapping[str, Any] | None) -> CatalogView:
+    """一次运行的视图（Run.options 里的 catalog_view）。没有就是 live。"""
+    return CatalogView.parse((options or {}).get(OPTION_KEY))
+
+
+def of_run(run_ctx: Any) -> CatalogView:
+    """运行里的节点构造 ToolContext 时用：RunContext.extra['run_options'] 里的视图（runner 每段执行都放这份）。"""
+    from app.engine.runner import RUN_OPTIONS_KEY
+
+    return of_options((getattr(run_ctx, "extra", None) or {}).get(RUN_OPTIONS_KEY))
+
+
+def describe(view: CatalogView | None) -> dict[str, Any] | None:
+    """记进冻结快照 meta 的说法：缺省视图（live、全部）不记，别的照 as_option 记下。"""
+    view = view or LIVE
+    return None if view.is_default else view.as_option()
+
+
+def restrict(view: CatalogView | None, notes: Mapping[str, Any] | None) -> dict[str, Any]:
+    """一张表的目录在这个视图下留下什么（状态集合；驳回的项原样留着，见 catalog.restrict_notes）。"""
+    allowed = STATUS_SETS[(view or LIVE).statuses]
+    if allowed is None:
+        return dict(notes or {})
+    return catalog.restrict_notes(notes, allowed)
+
+
+def _restricted(view: CatalogView, entries: Mapping[str, CatalogEntry]) -> dict[str, CatalogEntry]:
+    if STATUS_SETS[view.statuses] is None:
+        return dict(entries)
+    out: dict[str, CatalogEntry] = {}
+    for table, entry in entries.items():
+        if notes := restrict(view, entry.notes):
+            out[table] = CatalogEntry(table=entry.table, notes=notes, version=entry.version,
+                                      updated_by=entry.updated_by, updated_at=entry.updated_at)
+    return out
+
+
+def _snapshot_tables(view: CatalogView, source_id: str) -> dict[str, CatalogEntry]:
+    from app.core.artifact_store import load
+
+    try:
+        content = load(view.snapshot or "")
+    except ValueError as e:     # 内容和哈希对不上
+        raise CatalogSnapshotMissing(f"目录快照 {str(view.snapshot)[:12]}… 读不出：{e}") from e
+    if not isinstance(content, dict) or not isinstance(content.get("sources"), dict):
+        raise CatalogSnapshotMissing(f"目录快照 {str(view.snapshot)[:12]}… 不在了或内容认不出，这次不能按它读目录")
+    tables = ((content["sources"].get(source_id) or {}).get("tables")) or {}
+    out: dict[str, CatalogEntry] = {}
+    for table, body in tables.items():
+        if isinstance(body, Mapping) and isinstance(body.get("notes"), Mapping):
+            out[str(table)] = CatalogEntry(table=str(table), notes=dict(body["notes"]),
+                                           version=int(body.get("version") or 0))
+    return out
+
+
+async def read_many(session: Any, view: CatalogView | None, source_ids: Iterable[str]) -> dict[str, dict[str, CatalogEntry]]:
+    """几个数据源在这个视图下的目录：{源 id: {表名: CatalogEntry}}，没有目录的源不出现。live 一次查询读出。"""
+    view = view or LIVE
+    ids = [i for i in dict.fromkeys(source_ids) if i]
+    if view.mode == "off" or not ids:
+        return {}
+    if view.mode == "snapshot":
+        raw = {sid: found for sid in ids if (found := _snapshot_tables(view, sid))}
+    else:
+        raw = await catalog.read_catalogs(session, ids)
+    return {sid: kept for sid, entries in raw.items() if (kept := _restricted(view, entries))}
+
+
+async def read(view: CatalogView | None, source: Any) -> dict[str, CatalogEntry]:
+    """一个数据源在这个视图下的目录：{表名: CatalogEntry}。用自己的短会话读（调用方的会话里已经加载的对象不会因为回滚
+    过期）。live 读不到（库出错）当没有目录、记日志：目录是附加说明，不能挡住查询；快照不在了照实抛 CatalogSnapshotMissing。"""
+    from app.db.base import SessionLocal
+
+    view = view or LIVE
+    source_id = getattr(source, "id", None)
+    if view.mode == "off" or not source_id:
+        return {}
+    if view.mode == "snapshot":
+        return _restricted(view, _snapshot_tables(view, source_id))
+    try:
+        async with SessionLocal() as own:
+            return (await read_many(own, view, [source_id])).get(source_id, {})
+    except Exception:  # noqa: BLE001
+        logger.exception("读取数据源 %s 的数据目录失败，本次不附目录", getattr(source, "name", source_id))
+        return {}
+
+
+__all__ = ["CatalogSnapshotMissing", "CatalogView", "CatalogViewInvalid", "LIVE", "MODES", "OPTION_KEY", "SNAPSHOT_KIND",
+           "STATUS_SETS", "describe", "of_options", "of_run", "read", "read_many", "restrict"]

backend/app/data/sqlcheck.py

@@ -1084,36 +1084,36 @@ def graph_checks(nodes: Iterable[Any], checkers: Mapping[str, SqlChecker], *,
     return out
 
 
-async def load_checkers(session: Any, sources: Iterable[Any]) -> dict[str, SqlChecker]:
+async def load_checkers(session: Any, sources: Iterable[Any], *, catalog_view: Any = None) -> dict[str, SqlChecker]:
     """给一批数据源（DataSource 行）各建一个检查器：{源名: SqlChecker}。没有目录的源不建。
 
-    一次查询读出这些源的全部目录。读不到就当没有目录（返回空 dict）：检查是附加的，不能挡住自查和发布本身。
+    一次查询读出这些源的全部目录，读哪一份由 catalog_view 定（data/catalog_view.py：当前目录、目录快照、关；只用已核实
+    的项），不给是当前目录。读不到当前目录就当没有目录（返回空 dict）：检查是附加的，不能挡住自查和发布本身；视图指定的
+    目录快照不在了照实报错。
     """
-    from sqlalchemy import select
-
-    from app.db.models import CatalogNote
+    from app.data.catalog_view import CatalogSnapshotMissing, CatalogView, read_many
 
     rows = [s for s in sources if getattr(s, "id", None) and dialect_of(getattr(s, "kind", None))]
     if not rows:
         return {}
     try:
-        notes = (await session.execute(
-            select(CatalogNote.source_id, CatalogNote.table_name, CatalogNote.notes)
-            .where(CatalogNote.source_id.in_([r.id for r in rows]))
-        )).all()
+        entries = await read_many(session, CatalogView.parse(catalog_view), [r.id for r in rows])
+    except CatalogSnapshotMissing:
+        raise
     except Exception:  # noqa: BLE001
         logger.exception("读取数据目录失败，本次不做基于目录的 SQL 检查")
         return {}
     by_source: dict[str, dict[str, Any]] = {}
-    for source_id, table, value in notes:
-        if isinstance(value, dict) and value:
-            by_source.setdefault(source_id, {})[table] = value
+    for source_id, tables in entries.items():
+        for table, entry in tables.items():
+            if isinstance(entry.notes, dict) and entry.notes:
+                by_source.setdefault(source_id, {})[table] = entry.notes
     return {r.name: SqlChecker(kind=r.kind, schema_cache=r.schema_cache, notes=by_source[r.id], source_name=r.name)
             for r in rows if r.id in by_source}
 
 
 async def load_checkers_by_name(session: Any, names: Iterable[str]) -> dict[str, SqlChecker]:
-    """按数据源名建检查器（只看启用着的源）。发布门禁用：图里写的是源名。"""
+    """按数据源名建检查器（只看启用着的源）。发布门禁用：图里写的是源名，读的是当前目录。"""
     from sqlalchemy import select
 
     from app.db.models import DataSource

backend/app/engine/nodes/challenge.py

@@ -67,6 +67,7 @@ from app.core.errors import describe_exception
 from app.core.events import EventType
 from app.db.base import SessionLocal
 from app.db.models import DataSource
+from app.data import catalog_view
 from app.engine import deadline, verify_agent
 from app.engine import judge as judging
 from app.engine.context import NodeContext, NodeError
@@ -767,7 +768,8 @@ class _Review:
                 spec=self.spec, system=CHALLENGE_SYSTEM, task=task_text, sources=live,
                 output_schema=CHALLENGE_SCHEMA, max_queries=self.max_queries, timeout_s=allot.timeout_s,
                 daily_max_usd=self.daily_max_usd, emit=deadline.relabel(ctx.emit, allot), run_id=ctx.run.run_id,
-                node_id=ctx.node.id, data_versions=run_versions(ctx.run), masked_values=known)
+                node_id=ctx.node.id, data_versions=run_versions(ctx.run), masked_values=known,
+                catalog_view=catalog_view.of_run(ctx.run))
             outcome = await asyncio.to_thread(_masked_rationale, outcome, known, given=[cand.text], kinds=self.kinds,
                                               masks=self.masks)
             if outcome.error and not outcome.model:

backend/app/engine/nodes/dual_check.py

@@ -68,6 +68,7 @@ from app.core.errors import describe_exception
 from app.core.events import EventType
 from app.db.base import SessionLocal
 from app.db.models import DataSource
+from app.data import catalog_view
 from app.engine import deadline, recompute
 from app.engine.approval import read_decision
 from app.engine.context import NodeContext, NodeError
@@ -346,9 +347,11 @@ def _skipped(ctx: NodeContext, reason: str) -> dict[str, Any]:
 
 
 class _Sources:
-    """一次复算里要读的查询快照、数据源和快照要遮的列：同一份只读一次、只算一次。"""
+    """一次复算里要读的查询快照、数据源和快照要遮的列：同一份只读一次、只算一次。view 是这次运行的目录视图
+    （data/catalog_view.py）：交给盲算的目录、比较两条 SQL 用的中文名都按它读，和运行时工具给模型看的是同一份。"""
 
-    def __init__(self) -> None:
+    def __init__(self, view: Any = None) -> None:
+        self.view = catalog_view.CatalogView.parse(view)
         self.snapshots: dict[str, tuple[Any, str | None]] = {}
         self.rows: dict[str, DataSource | None] = {}
         self.masks: dict[tuple[str, str | None], frozenset[str]] = {}
@@ -782,14 +785,14 @@ async def _constant_columns(snapshot: Mapping[str, Any], sources: _Sources) -> f
 
 async def _catalog_text(sources: _Sources, name: str) -> str:
     """交给盲算 agent 的数据目录：表清单（表名、中文名、粒度）和有目录的表的业务说明（业务日期、有效记录条件、去重、
-    字段含义、码值、关联关系）。太长的截断，其余请它用 db_schema 查。"""
+    字段含义、码值、关联关系）。太长的截断，其余请它用 db_schema 查。读哪一份由这次运行的目录视图定（sources.view），
+    和运行时工具给模型看的是同一份。"""
     from app.data import catalog
 
     row = await sources.source(name)
     if row is None:
         return BLIND_NO_CATALOG_PROMPT
-    async with SessionLocal() as session:
-        entries = await catalog.read_catalog(session, row.id)
+    entries = await catalog_view.read(sources.view, row)
     cache = row.schema_cache if isinstance(row.schema_cache, dict) else {}
     tables = cache.get("tables") or {}
     notes = {table: entry.notes for table, entry in entries.items()}
@@ -843,7 +846,7 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op
         return {}, None, False, usage
     async with SessionLocal() as session:
         spec = await verify_model_spec(session, ctx.config, settings=dict(settings))
-    sources = _Sources()
+    sources = _Sources(catalog_view.of_run(ctx.run))
     catalogs: dict[str, str] = {}
     for job in jobs.values():
         if job["source"] not in catalogs:
@@ -870,7 +873,7 @@ async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Op
             spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]],
             output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=allot.timeout_s,
             daily_max_usd=options.daily_max_usd, emit=deadline.relabel(ctx.emit, allot), run_id=ctx.run.run_id,
-            node_id=ctx.node.id, data_versions=run_versions(ctx.run))
+            node_id=ctx.node.id, data_versions=run_versions(ctx.run), catalog_view=sources.view)
 
     keys = list(jobs)
     await one(keys[0], jobs[keys[0]])
@@ -1078,7 +1081,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
     """
     rows = [r for r in checked["rows"] if r["status"] == DIFFERENT]
     results = checked["results"]
-    sources = _Sources()
+    sources = _Sources(catalog_view.of_run(ctx.run))
     infos: dict[str, dict[str, Any]] = {}
     metrics: list[dict[str, Any]] = []
     summary: list[dict[str, Any]] = []
@@ -1179,15 +1182,15 @@ async def _source_info(sources: _Sources, name: str) -> dict[str, Any]:
     （uploaded：导入时日期一律存成文字）和它导入时认作日期的列（upload_dates，按当前版本的配方读，见
     dual_check_facts.upload_dates）——业务日期据此认上传表格的日期列，认不出的原因照实说（_dated_problem）。两项都进
     候选：审批卡的预告按发起审批时读的这一份；裁决后的真写按写入时的版本重读 upload_dates，和现读的表结构一起以写入时
-    为准（dual_check_facts.record_choices）。数据源没了就都不给，照样比原文。"""
-    from app.data import catalog
+    为准（dual_check_facts.record_choices）。数据源没了就都不给，照样比原文。目录按这次运行的目录视图读（sources.view），
+    比较两条 SQL 用到的中文名和运行时给模型看的是同一份。"""
     from app.data.sql_semantics import catalog_labels
 
     row = await sources.source(name)
     if row is None:
         return {"kind": None, "schema": None, "names": {}, "notes": {}, "uploaded": False, "upload_dates": {}}
+    entries = await catalog_view.read(sources.view, row)
     async with SessionLocal() as session:
-        entries = await catalog.read_catalog(session, row.id)
         dates = await upload_dates(session, row)
     notes = {table: entry.notes for table, entry in entries.items()}
     return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None,

backend/app/engine/nodes/llm.py

@@ -48,6 +48,7 @@ from app.providers.factory import (
     get_chat_model,
 )
 from app.engine.sql_problems import event_checks
+from app.data import catalog_view
 from app.tools.datasource import QUERY_PREFIX, run_masks_results, run_reads_only, run_scope_fixed, run_versions
 from app.tools.registry import (
     ToolArgsError,
@@ -341,6 +342,8 @@ async def _resolve_tools(ctx: NodeContext, state: GraphState) -> list[BaseTool]:
         mask_results=run_masks_results(ctx.run),
         force_readonly=run_reads_only(ctx.run),
         fixed_scope=fixed,
+        # 读哪一份数据目录：发起时定下的（data/catalog_view.py），审批恢复、续跑都沿用
+        catalog_view=catalog_view.of_run(ctx.run),
         # 数据源工具运行中途补固定版本时经它留一条事件（tools/datasource.py 的 _pin_late）
         emit=ctx.emit,
     )

backend/app/engine/nodes/multi.py

@@ -28,6 +28,7 @@ from app.engine.toolcalls import (
 from app.providers import catalog
 from app.providers.factory import ModelSpec, bind_tools_safely, get_chat_model
 from app.engine.sql_problems import event_checks
+from app.data import catalog_view
 from app.tools.datasource import QUERY_PREFIX, run_masks_results, run_reads_only, run_scope_fixed, run_versions
 from app.tools.registry import (
     ToolArgsError,
@@ -317,6 +318,8 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]:
                 mask_results=run_masks_results(ctx.run),
                 force_readonly=run_reads_only(ctx.run),
                 fixed_scope=run_scope_fixed(ctx.run),
+                # 读哪一份数据目录：发起时定下的（data/catalog_view.py），审批恢复、续跑都沿用
+                catalog_view=catalog_view.of_run(ctx.run),
                 # 数据源工具运行中途补固定版本时经它留一条事件（tools/datasource.py 的 _pin_late）
                 emit=ctx.emit,
             )

backend/app/engine/nodes/tools.py

@@ -20,6 +20,7 @@ from app.sandbox.base import SandboxLimits
 from app.sandbox.manager import sandbox_manager
 from app.engine.sql_problems import event_checks
 from app.capabilities.registry import is_platform_query, snapshot_owned
+from app.data import catalog_view
 from app.tools.datasource import (DATA_UNAVAILABLE, SCHEMA_PREFIX, run_masks_results, run_reads_only,
                                   run_scope_fixed, run_versions)
 from app.tools.registry import (
@@ -54,6 +55,8 @@ def _tool_ctx(ctx: NodeContext) -> ToolContext:
         mask_results=run_masks_results(ctx.run),
         force_readonly=run_reads_only(ctx.run),
         fixed_scope=fixed,
+        # 读哪一份数据目录：发起时定下的（data/catalog_view.py），审批恢复、续跑都沿用
+        catalog_view=catalog_view.of_run(ctx.run),
         # 数据源工具运行中途补固定版本时经它留一条事件（tools/datasource.py 的 _pin_late）
         emit=ctx.emit,
     )

backend/app/engine/verify_agent.py

@@ -758,6 +758,7 @@ async def run_verify_agent(
     node_id: str = "",
     data_versions: dict[str, Any] | Versions | None = Versions.UNSET,
     masked_values: Any = None,
+    catalog_view: Any = None,
 ) -> VerifyOutcome:
     """让一个只读 agent 查库，按 output_schema 交回有出处的结果。永远不抛。
 
@@ -775,6 +776,8 @@ async def run_verify_agent(
     - masked_values：调用方已经认出的遮罩值（engine/masking 的 _Secrets，审批质询、反方审稿收的报告快照里的值）。
       查询工具从第一条查询起就按它们按值遮：agent 查到的备注、别的表里写着同一个号码时，模型看不到。随查询时遮掉的值
       一起交回（VerifyOutcome.masked_values）。不改它。
+    - catalog_view：db_schema 给 agent 看哪一份数据目录（data/catalog_view.py）。运行里的节点传这次运行的视图
+      （catalog_view.of_run）；不给是当前目录（质询接口）。
     """
     state = _State(started=time.monotonic(), seen=SeenValues(
         given=masked_values if masked_values is not None else _Secrets()))
@@ -782,7 +785,7 @@ async def run_verify_agent(
         out = await _verify(state, spec=spec, system=system, task=task, sources=sources,
                             output_schema=output_schema, max_queries=max_queries, timeout_s=timeout_s,
                             max_cost_usd=max_cost_usd, daily_max_usd=daily_max_usd, max_steps=max_steps, emit=emit,
-                            run_id=run_id, node_id=node_id, data_versions=data_versions)
+                            run_id=run_id, node_id=node_id, data_versions=data_versions, catalog_view=catalog_view)
     except Exception as e:  # noqa: BLE001 - 永远不抛：库写不进去、建工具出错这类意外也落在返回值里
         logger.warning("核查出错", exc_info=True)
         out = state.outcome(error=f"核查出错（{describe_exception(e)}）")
@@ -798,6 +801,7 @@ async def run_verify_agent(
 async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sources: Iterable[str],
                   output_schema: dict[str, Any], max_queries: int | None, timeout_s: float | None,
                   max_cost_usd: float | None, daily_max_usd: float | None | Default, max_steps: int | None,
+                  catalog_view: Any = None,
                   emit: Emit | None, run_id: str, node_id: str,
                   data_versions: dict[str, Any] | Versions | None) -> VerifyOutcome:
     deadline = state.started + timeout_s if timeout_s is not None else None
@@ -834,7 +838,7 @@ async def _verify(state: _State, *, spec: ModelSpec, system: str, task: str, sou
 
     # ---- 工具：只读的 db_schema / db_query，查询结果先遮再交给模型（遮掉的值并进 state.seen）
     wanted = list(dict.fromkeys(s for s in sources if isinstance(s, str) and s))
-    ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit,
+    ctx = ToolContext(run_id=run_id, node_id=node_id, data_versions=data_versions, emit=emit, catalog_view=catalog_view,
                       extra={MASKED_VALUES_KEY: state.seen})
     async with SessionLocal() as session:
         tools = await build_datasource_tools([f"{p}{s}" for s in wanted for p in (SCHEMA_PREFIX, QUERY_PREFIX)], ctx,

backend/app/look/tables.py

@@ -159,12 +159,16 @@ class TableKind:
 
     @classmethod
     async def load(cls, sources: list[Any], *, graph: Mapping[str, Any] | None = None,
-                   picker: Picker | None = None) -> TableKind:
-        """读各源的数据目录。graph 是现有工作流（改图时的原图、追问时上一轮的图），概览里列出它查过的表。"""
-        from app.tools.datasource import _catalog_of
+                   picker: Picker | None = None, catalog_view: Any = None) -> TableKind:
+        """读各源的数据目录。graph 是现有工作流（改图时的原图、追问时上一轮的图），概览里列出它查过的表。
 
+        读哪一份目录由 catalog_view 定（data/catalog_view.py：当前目录、目录快照、关；只用已核实的项），不给是当前目录。
+        查看结果里的目录和运行时 db_schema 给模型看的、自查用的是同一份——评测的「目录对照」量的才是整份贡献。"""
+        from app.data.catalog_view import CatalogView, read
+
+        view = CatalogView.parse(catalog_view)
         seeds = graph_tables(graph, sources)
-        plans = [SourcePlan(source=s, entries=await _catalog_of(s) if introspect.table_names(s) else {},
+        plans = [SourcePlan(source=s, entries=await read(view, s) if introspect.table_names(s) else {},
                             seeds=seeds.get(s.name, [])) for s in sources]
         return cls(plans, picker=picker)
 

backend/app/tools/datasource.py

@@ -277,7 +277,7 @@ async def build_datasource_tools(
         except table_versions.SnapshotMissing as e:
             tools.extend(_unavailable_tool(row, prefix, str(e)) for prefix in prefixes)
             continue
-        entries = await _catalog_of(row)
+        entries = await _catalog_of(row, ctx.catalog_view)
         for prefix in prefixes:
             if prefix == QUERY_PREFIX:
                 tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries, force_readonly=force_readonly,
@@ -287,20 +287,17 @@ async def build_datasource_tools(
     return tools
 
 
-async def _catalog_of(row: DataSource) -> dict[str, data_catalog.CatalogEntry]:
+async def _catalog_of(row: Any, view: Any = None) -> dict[str, data_catalog.CatalogEntry]:
     """这个源的数据目录，建工具时读一次：db_schema 给模型看的和查询时冻结进表结构快照的是同一版。
 
-    读不到就当没有目录，工具照常可用——目录是附加说明，不能挡住查询。用单独的会话读：出错要回滚，
-    而回滚会让调用方会话里已经加载的对象（row）全部过期，之后再读属性就是异步环境里的懒加载。
+    读哪一份由目录视图定（view，data/catalog_view.py：当前目录、目录快照、关；只用已核实的项），不给就是当前目录。
+    读不到就当没有目录，工具照常可用——目录是附加说明，不能挡住查询；视图指定的目录快照不在了照实报错（不悄悄换成当前
+    目录）。用单独的会话读：出错要回滚，而回滚会让调用方会话里已经加载的对象（row）全部过期，之后再读属性就是异步环境里
+    的懒加载。
     """
-    from app.db.base import SessionLocal
+    from app.data import catalog_view
 
-    try:
-        async with SessionLocal() as own:
-            return await data_catalog.read_catalog(own, row.id)
-    except Exception:  # noqa: BLE001
-        logger.exception("读取数据源 %s 的数据目录失败，本次不附目录", row.name)
-        return {}
+    return await catalog_view.read(catalog_view.CatalogView.parse(view), row)
 
 
 # --------------------------------------------------------------------------
@@ -753,6 +750,12 @@ async def _store_schema(source: Any, ctx: ToolContext, *,
     if frozen := data_catalog.frozen_catalog(catalog or {}, tables):
         content[CATALOG_SNAPSHOT_KEY] = frozen
         meta["catalog_tables"] = len(frozen)
+    # 这次运行读的不是缺省的当前目录（评测的对照：关、目录快照、只用已核实的项）：记在引用的 meta 里，证据面板写得出
+    # 「本次运行目录：关」。不进内容哈希，缺省视图不记，快照和以前一字不差
+    from app.data.catalog_view import CatalogView, describe
+
+    if (view := describe(CatalogView.parse(ctx.catalog_view))) is not None:
+        meta["catalog_view"] = view
     try:
         return await put_json(content, kind=SCHEMA_SNAPSHOT, run_id=ctx.run_id or "", node_id=ctx.node_id or "",
                               meta=meta)

backend/app/tools/registry.py

@@ -49,6 +49,9 @@ class ToolContext:
     #: 个人应用的刷新、个人蓝图的试运行）。记忆、知识库的内容不经遮罩：kb_search 的 collection 参数（调用工具节点写的、
     #: 参数名纠错改出来的、图级 defaults 补上的、模型挑的）一律拒，只能检索 collection 这一个
     fixed_scope: bool = False
+    #: 读哪一份数据目录（data/catalog_view.CatalogView）：运行里由节点从 Run.options 取来（catalog_view.of_run），db_schema
+    #: 给模型看的、查询冻结进表结构快照的、查询时的 SQL 检查都按它。None 是当前目录（缺省，行为同以前）
+    catalog_view: Any = None
     extra: dict[str, Any] = field(default_factory=dict)
 
 

backend/tests/test_catalog_view.py

@@ -0,0 +1,270 @@
+"""目录视图接缝（data/catalog_view.py；细案 01 做法 5，总方案 X7）。
+
+目录在四处被读：运行时的 db_schema / db_query（目录同时冻结进表结构快照）、助手查看的表详情、自查的 SQL 检查器、双盲复算。
+这里守：四种视图（当前目录、只用已核实和已确认的项、关、目录快照）在四处给出的是同一份；缺省是当前目录、行为不变；运行
+从 Run.options 读视图（随发起定下，审批恢复、续跑沿用），冻结快照的 meta 记下不是缺省的视图；lint：这几个消费模块不许
+绕过视图直接 read_catalog、直接查 CatalogNote。
+
+库、表、列、目录内容都是虚构的。
+"""
+from __future__ import annotations
+
+import asyncio
+import re
+import sqlite3
+import uuid
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+
+from app.core.artifact_store import put_json
+from app.data import catalog, catalog_view
+from app.data.catalog import make_item
+from app.data.catalog_view import CatalogSnapshotMissing, CatalogView, CatalogViewInvalid
+from app.data.introspect import introspect
+from app.db.base import SessionLocal
+from app.db.models import Artifact, DataSource, Run
+from app.engine.runner import run_manager
+from tests.fixtures.sources import drop_source
+
+APP = Path(__file__).resolve().parents[1] / "app"
+
+#: 已确认的一项、推断的一项、驳回的一项：三种视图各留哪些一眼看得出
+NOTES = {
+    "label": make_item("门店订单", "human"),                         # confirmed
+    "grain": make_item("每笔订单一行", "llm"),                         # proposed
+    "description": make_item("旧说明", "llm", "rejected"),             # rejected
+    "columns": {"amount": {"label": make_item("实收金额", "human")},
+                "status": {"label": make_item("订单状态", "llm")}},
+}
+SNAPSHOT_NOTES = {"label": make_item("快照里的订单", "human")}
+
+VIEWS = {
+    "live": CatalogView(),
+    "reviewed": CatalogView(statuses="reviewed"),
+    "off": CatalogView(mode="off"),
+}
+
+
+@pytest.fixture
+async def shop(tmp_path):
+    path = tmp_path / "view_shop.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, amount REAL NOT NULL, status INTEGER);"
+                     "INSERT INTO orders VALUES (1, 30.0, 1), (2, 20.0, 9);")
+    db.commit()
+    db.close()
+    async with SessionLocal() as session:
+        row = DataSource(name=f"viewshop_{uuid.uuid4().hex[:6]}", kind="sqlite", database=str(path))
+        session.add(row)
+        await session.commit()
+        row.schema_cache = await introspect(row)
+        await session.commit()
+        await session.refresh(row)
+        await catalog.write_entry(session, row.id, "orders", NOTES, if_version=0, actor=None)
+        session.expunge(row)
+    yield row
+    await drop_source(row.id)
+
+
+async def _snapshot_view(source) -> CatalogView:
+    sid = await put_json({"sources": {source.id: {"tables": {"orders": {"version": 7, "notes": SNAPSHOT_NOTES}}}}},
+                         kind=catalog_view.SNAPSHOT_KIND)
+    return CatalogView(mode="snapshot", snapshot=sid)
+
+
+def _says(text: str) -> dict[str, bool]:
+    return {word: word in text for word in ("门店订单", "每笔订单一行", "旧说明", "实收金额", "订单状态", "快照里的订单")}
+
+
+EXPECT = {
+    "live": {"门店订单": True, "每笔订单一行": True, "旧说明": False, "实收金额": True, "订单状态": True, "快照里的订单": False},
+    "reviewed": {"门店订单": True, "每笔订单一行": False, "旧说明": False, "实收金额": True, "订单状态": False,
+                 "快照里的订单": False},
+    "off": dict.fromkeys(("门店订单", "每笔订单一行", "旧说明", "实收金额", "订单状态", "快照里的订单"), False),
+    "snapshot": {"门店订单": False, "每笔订单一行": False, "旧说明": False, "实收金额": False, "订单状态": False,
+                 "快照里的订单": True},
+}
+
+
+async def _all_views(source) -> dict[str, CatalogView]:
+    return {**VIEWS, "snapshot": await _snapshot_view(source)}
+
+
+# --------------------------------------------------------------------------
+# 视图本身
+# --------------------------------------------------------------------------
+
+
+def test_parse_and_defaults():
+    assert CatalogView.parse(None) is catalog_view.LIVE and catalog_view.LIVE.is_default
+    assert CatalogView.parse({"mode": "off"}) == CatalogView(mode="off")
+    assert catalog_view.describe(None) is None and catalog_view.describe(CatalogView()) is None
+    assert catalog_view.describe(CatalogView(mode="off")) == {"mode": "off", "statuses": "all"}
+    assert catalog_view.of_options({"catalog_view": {"statuses": "reviewed"}}) == CatalogView(statuses="reviewed")
+    for bad in ({"mode": "nope"}, {"statuses": "some"}, {"mode": "snapshot"}, {"mode": "live", "snapshot": "x"},
+                {"extra": 1}, "off"):
+        with pytest.raises(CatalogViewInvalid):
+            CatalogView.parse(bad)
+
+
+async def test_a_missing_snapshot_is_an_error_not_the_live_catalog(shop):
+    with pytest.raises(CatalogSnapshotMissing):
+        await catalog_view.read(CatalogView(mode="snapshot", snapshot="0" * 64), shop)
+
+
+# --------------------------------------------------------------------------
+# 四处一致
+# --------------------------------------------------------------------------
+
+
+async def test_runtime_db_schema_follows_the_view(shop):
+    from app.tools.datasource import SCHEMA_PREFIX, build_datasource_tools
+    from app.tools.registry import ToolContext, Versions
+
+    for name, view in (await _all_views(shop)).items():
+        ctx = ToolContext(run_id="", node_id="", data_versions=Versions.CURRENT, catalog_view=view)
+        async with SessionLocal() as session:
+            [tool] = await build_datasource_tools([f"{SCHEMA_PREFIX}{shop.name}"], ctx, session)
+        assert _says(await tool.coroutine(table="orders")) == EXPECT[name], name
+
+
+async def test_look_table_detail_follows_the_view(shop):
+    from app.look.service import LookService
+    from app.look.tables import TableKind
+
+    for name, view in (await _all_views(shop)).items():
+        looks = LookService([await TableKind.load([shop], catalog_view=view)])
+        got = await looks.look({"op": "look", "need": "订单", "targets": [{"kind": "table", "name": "orders"}]})
+        assert _says(got.text) == EXPECT[name], name
+
+
+async def test_sql_checkers_follow_the_view(shop):
+    from app.data.sqlcheck import load_checkers
+
+    for name, view in (await _all_views(shop)).items():
+        async with SessionLocal() as session:
+            checkers = await load_checkers(session, [shop], catalog_view=view)
+        if name == "off":
+            assert checkers == {}
+            continue
+        notes = checkers[shop.name]._raw["orders"]
+        assert _says(str(notes)) | {"旧说明": False} == EXPECT[name] | {"旧说明": False}, name
+        if name == "reviewed":
+            # 驳回的项原样留着交给检查器（它自己按驳回处理），推断的去掉
+            assert notes["description"]["status"] == "rejected" and "grain" not in notes
+
+
+async def test_dual_check_blind_catalog_follows_the_view(shop):
+    from app.engine.nodes.dual_check import BLIND_NO_CATALOG_PROMPT, _catalog_text, _source_info, _Sources
+
+    for name, view in (await _all_views(shop)).items():
+        sources = _Sources(view)
+        text = await _catalog_text(sources, shop.name)
+        assert _says(text) == EXPECT[name], name
+        info = await _source_info(sources, shop.name)
+        assert ("orders" in info["notes"]) == (name != "off"), name
+        if name == "off":
+            # 关目录：表清单（表结构里的事实）照给，目录的说明一句都没有
+            assert "orders" in text and "数据目录（" not in text and BLIND_NO_CATALOG_PROMPT not in text
+
+
+async def test_the_default_is_the_live_catalog_everywhere(shop):
+    """不给视图：和以前一样读当前目录。"""
+    from app.engine.nodes.dual_check import _catalog_text, _Sources
+    from app.look.service import LookService
+    from app.look.tables import TableKind
+    from app.tools.datasource import _catalog_of
+
+    assert _says(str({t: e.notes for t, e in (await _catalog_of(shop)).items()}))["门店订单"]
+    looks = LookService([await TableKind.load([shop])])
+    got = await looks.look({"op": "look", "need": "订单", "targets": [{"kind": "table", "name": "orders"}]})
+    assert _says(got.text) == EXPECT["live"]
+    assert _says(await _catalog_text(_Sources(), shop.name)) == EXPECT["live"]
+
+
+# --------------------------------------------------------------------------
+# 运行：从 Run.options 读视图，冻结快照记下
+# --------------------------------------------------------------------------
+
+
+def _graph(source: str) -> dict:
+    def node(nid, ntype, **config):
+        return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
+
+    return {"nodes": [node("start", "input"),
+                      node("q", "tool", tool=f"db_query__{source}", args={"sql": "SELECT SUM(amount) AS 金额 FROM orders"},
+                           assign_to="rows"),
+                      node("out", "output", fields=[{"name": "rows", "value": "{{ vars.rows }}"}])],
+            "edges": [{"source": "start", "target": "q"}, {"source": "q", "target": "out"}]}
+
+
+async def _frozen(options: dict | None, source) -> tuple[dict, dict]:
+    from app.core.artifact_store import load
+    from app.engine.evidence import SCHEMA_SNAPSHOT
+    from app.engine.launch import launch_run
+
+    async with SessionLocal() as session:
+        run = await launch_run(session, graph=_graph(source.name), input_payload={}, actor="测试",
+                               origin_kind="canvas", options=options)
+        run_id = run.id
+    for _ in range(200):
+        async with SessionLocal() as session:
+            status = (await session.get(Run, run_id)).status
+        if status in ("succeeded", "failed", "cancelled"):
+            break
+        await asyncio.sleep(0.05)
+    assert status == "succeeded", status
+    async with SessionLocal() as session:
+        ref = (await session.execute(select(Artifact).where(
+            Artifact.run_id == run_id, Artifact.kind == SCHEMA_SNAPSHOT))).scalars().first()
+    assert ref is not None
+    return load(ref.id), ref.meta or {}
+
+
+async def test_a_run_reads_its_view_from_options_and_the_freeze_says_so(shop):
+    await run_manager.setup()
+    try:
+        content, meta = await _frozen(None, shop)
+        assert "orders" in content.get("catalog", {}), "缺省：当前目录照旧冻结"
+        assert "catalog_view" not in meta, "缺省视图不记，meta 和以前一样"
+
+        content, meta = await _frozen({"catalog_view": {"mode": "off"}}, shop)
+        assert "catalog" not in content, "关目录：快照里没有目录"
+        assert meta["catalog_view"] == {"mode": "off", "statuses": "all"}
+
+        content, meta = await _frozen({"catalog_view": {"statuses": "reviewed"}}, shop)
+        frozen = content["catalog"]["orders"]["notes"]
+        assert "grain" not in frozen and frozen["label"]["value"] == "门店订单"
+        assert meta["catalog_view"] == {"mode": "live", "statuses": "reviewed"}
+    finally:
+        await run_manager.shutdown()
+
+
+def test_of_run_reads_the_run_options():
+    from types import SimpleNamespace
+
+    from app.engine.runner import RUN_OPTIONS_KEY
+
+    run = SimpleNamespace(extra={RUN_OPTIONS_KEY: {"catalog_view": {"mode": "off"}}})
+    assert catalog_view.of_run(run) == CatalogView(mode="off")
+    assert catalog_view.of_run(SimpleNamespace(extra={})) == catalog_view.LIVE
+
+
+# --------------------------------------------------------------------------
+# lint：消费模块只经视图读目录
+# --------------------------------------------------------------------------
+
+#: 给模型、给自查读目录的模块（四处消费 + 它们的帮手）。要读目录一律经 data/catalog_view.py
+CONSUMERS = ("tools/datasource.py", "look/tables.py", "data/sqlcheck.py", "engine/nodes/dual_check.py",
+             "engine/verify_agent.py", "api/copilot_context.py")
+FORBIDDEN = re.compile(r"\bread_catalogs?\s*\(|\bread_entry\s*\(|\bCatalogNote\b")
+
+
+@pytest.mark.parametrize("module", CONSUMERS)
+def test_consumers_never_read_the_catalog_directly(module):
+    text = (APP / module).read_text(encoding="utf-8")
+    code = "\n".join(line for line in text.splitlines() if not line.lstrip().startswith("#"))
+    hits = [m.group(0) for m in FORBIDDEN.finditer(code)]
+    assert not hits, f"{module} 绕过了目录视图：{hits}"