From 4c0e83bc80355a27dc1a97d49eee16a05656027a Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 10:28:01 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9B=AE=E5=BD=95=E8=A7=86=E5=9B=BE=E6=8E=A5?= =?UTF-8?q?=E7=BC=9D=20data/catalog=5Fview.py=EF=BC=88=E7=BB=86=E6=A1=88?= =?UTF-8?q?=2001=20=E5=81=9A=E6=B3=95=205=EF=BC=8C=E6=80=BB=E6=96=B9?= =?UTF-8?q?=E6=A1=88=20X7=EF=BC=89=EF=BC=9A=E8=AF=BB=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E7=9A=84=E5=9B=9B=E5=A4=84=E7=BB=9F=E4=B8=80=E7=BB=8F=E5=AE=83?= =?UTF-8?q?=EF=BC=8C=E3=80=8C=E5=BC=80=E7=9B=AE=E5=BD=95=20/=20=E5=85=B3?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=20/=20=E5=8F=AA=E7=94=A8=E5=B7=B2=E6=A0=B8?= =?UTF-8?q?=E5=AE=9E=E7=9A=84=E9=A1=B9=20/=20=E7=9B=AE=E5=BD=95=E5=BF=AB?= =?UTF-8?q?=E7=85=A7=E3=80=8D=E7=9A=84=E5=AF=B9=E7=85=A7=E6=89=8D=E5=81=9A?= =?UTF-8?q?=E5=BE=97=E4=BA=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 7 +- backend/app/data/catalog.py | 25 ++- backend/app/data/catalog_view.py | 181 +++++++++++++++++ backend/app/data/sqlcheck.py | 26 +-- backend/app/engine/nodes/challenge.py | 4 +- backend/app/engine/nodes/dual_check.py | 25 ++- backend/app/engine/nodes/llm.py | 3 + backend/app/engine/nodes/multi.py | 3 + backend/app/engine/nodes/tools.py | 3 + backend/app/engine/verify_agent.py | 8 +- backend/app/look/tables.py | 12 +- backend/app/tools/datasource.py | 25 ++- backend/app/tools/registry.py | 3 + backend/tests/test_catalog_view.py | 270 +++++++++++++++++++++++++ 14 files changed, 548 insertions(+), 47 deletions(-) create mode 100644 backend/app/data/catalog_view.py create mode 100644 backend/tests/test_catalog_view.py diff --git a/backend/app/api/copilot_context.py b/backend/app/api/copilot_context.py index 6c8e807e..7375d5f0 100644 --- a/backend/app/api/copilot_context.py +++ b/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) diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 3247aea4..60617db2 100644 --- a/backend/app/data/catalog.py +++ b/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", ] diff --git a/backend/app/data/catalog_view.py b/backend/app/data/catalog_view.py new file mode 100644 index 00000000..5af6e2f9 --- /dev/null +++ b/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"] diff --git a/backend/app/data/sqlcheck.py b/backend/app/data/sqlcheck.py index 2a300acd..eface254 100644 --- a/backend/app/data/sqlcheck.py +++ b/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 diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index 88753e2b..4b8f21ce 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/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: diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 0152179c..6790cb60 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/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, diff --git a/backend/app/engine/nodes/llm.py b/backend/app/engine/nodes/llm.py index 3c93e39f..bf737196 100644 --- a/backend/app/engine/nodes/llm.py +++ b/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, ) diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index 9815407c..c83efada 100644 --- a/backend/app/engine/nodes/multi.py +++ b/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, ) diff --git a/backend/app/engine/nodes/tools.py b/backend/app/engine/nodes/tools.py index 1b26352d..ec256f30 100644 --- a/backend/app/engine/nodes/tools.py +++ b/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, ) diff --git a/backend/app/engine/verify_agent.py b/backend/app/engine/verify_agent.py index 2127610c..f44d5c03 100644 --- a/backend/app/engine/verify_agent.py +++ b/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, diff --git a/backend/app/look/tables.py b/backend/app/look/tables.py index 25fe1e5c..0d4e61d4 100644 --- a/backend/app/look/tables.py +++ b/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) diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index 53e11ecf..d38ae483 100644 --- a/backend/app/tools/datasource.py +++ b/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) diff --git a/backend/app/tools/registry.py b/backend/app/tools/registry.py index c1215267..c80f0ed4 100644 --- a/backend/app/tools/registry.py +++ b/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) diff --git a/backend/tests/test_catalog_view.py b/backend/tests/test_catalog_view.py new file mode 100644 index 00000000..3edd71d2 --- /dev/null +++ b/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}"