数据目录阶段 1A：db_schema 输出附上数据目录，查询时目录随表结构快照冻结进证据，查询快照记下用到的表供使用次数统计

backend/app/api/datasources.py

@@ -22,6 +22,7 @@ from app.api.runs import actor_of
 from app.core.errors import AUTH, NETWORK, TIMEOUT, classify, explain, raw
 from app.core.config import settings
 from app.core.crypto import encrypt, mask
+from app.data import catalog as data_catalog
 from app.data import introspect as introspect_mod
 from app.data import table_versions
 from app.data.engine import (
@@ -805,10 +806,13 @@ async def get_schema(source_id: str, table: str | None = None,
         raise HTTPException(409, str(e)) from e
     if table:
         meta = introspect_mod.find_table(row, table)
+        # 和 db_schema 工具同一个说法：有数据目录的表，字段清单后面附上目录
+        entries = await data_catalog.read_catalog(session, found.id)
         return {
             "table": table,
             # 给模型看的那段文本照旧；界面画表格用下面的结构化字段
-            "detail": introspect_mod.describe_table(row, table),
+            "detail": introspect_mod.describe_table(row, table,
+                                                    catalog={n: e.notes for n, e in entries.items()}),
             "found": meta is not None,
             "qualified": meta.get("qualified", table) if meta else None,
             "kind": ("view" if meta.get("is_view") else "table") if meta else None,

backend/app/data/introspect.py

@@ -14,6 +14,7 @@
 """
 from __future__ import annotations
 
+from collections.abc import Mapping
 from datetime import datetime, timezone
 from typing import Any
 
@@ -421,8 +422,14 @@ def table_columns(meta: dict[str, Any]) -> list[dict[str, Any]]:
     ]
 
 
-def describe_table(source: Any, table: str) -> str:
-    """单表的完整字段说明。agent 调 db_schema 工具时返回这个。"""
+def describe_table(source: Any, table: str, *, catalog: Mapping[str, Mapping[str, Any]] | None = None) -> str:
+    """单表的完整字段说明。agent 调 db_schema 工具时返回这个。
+
+    catalog 是这个源的数据目录 {表名: notes}（data/catalog.py），给了且这张表有可给模型看的内容时，在字段
+    清单后面附上目录：中文名、粒度、业务主键、业务日期、有效记录条件、每列的含义 / 单位 / 度量类型 / 码值、
+    关联关系，推断的项标出来。导入表格的源，表和列的说明是系统按核对结果生成的：说明照旧拼在表名和列后面，
+    目录只追加说明没有覆盖的、或人工确认过的项（catalog.render_table_notes）。不给 catalog 时输出和以前一样。
+    """
     cache = source.schema_cache or {}
     tables = cache.get("tables") or {}
     # 模型可能传全名（ANALYTICS.V_TRIP_FACT），也可能只传表名，两种都认
@@ -457,6 +464,14 @@ def describe_table(source: Any, table: str) -> str:
             marks.append(col["comment"])
         suffix = f"  {'、'.join(marks)}" if marks else ""
         lines.append(f"  {col['name']}  {col['type']}{suffix}")
+    if catalog:
+        from app.data import catalog as catalog_mod
+
+        key = catalog_mod.resolve_table_name(cache, table)
+        block = catalog_mod.render_table_notes(catalog.get(key) if key else None, meta=meta,
+                                               system_notes=catalog_mod.system_notes_source(source))
+        if block:
+            lines.append(block)
     return "\n".join(lines)
 
 

backend/app/tools/datasource.py

@@ -21,6 +21,7 @@ import inspect
 import json
 import logging
 import weakref
+from collections.abc import Mapping
 from pathlib import Path
 from typing import Any
 
@@ -30,6 +31,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.events import EventType
+from app.data import catalog as data_catalog
 from app.data import introspect, table_versions
 from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query
 from app.data.guard import QueryLimits, SqlRejected, is_write
@@ -154,6 +156,8 @@ RECIPE_QUERY_HINT = " 中文表名和列名请加双引号。查单个值时，
 #: 封存事件 → schema_snapshot → 快照清单 → 各期导入清单 → 原件或清除记录，每一跳都按内容哈希
 _FROZEN_SCHEMA_KEYS = ("schema", "synced_at", "truncated", "total", "import_mode", "import_manifests",
                        "snapshot_manifest")
+#: 表结构快照里放数据目录的顶层键（_store_schema）。不来自 schema_cache：目录存在 catalog_notes 表里
+CATALOG_SNAPSHOT_KEY = "catalog"
 
 
 def _query_description(source: Any) -> str:
@@ -218,14 +222,27 @@ 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(session, row)
         for prefix in prefixes:
             if prefix == QUERY_PREFIX:
-                tools.append(_make_query_tool(source, ctx, fixed=fixed))
+                tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries))
             else:
-                tools.append(_make_schema_tool(source, fixed=fixed))
+                tools.append(_make_schema_tool(source, fixed=fixed, catalog=entries))
     return tools
 
 
+async def _catalog_of(session: AsyncSession, row: DataSource) -> dict[str, data_catalog.CatalogEntry]:
+    """这个源的数据目录，建工具时读一次：db_schema 给模型看的和查询时冻结进表结构快照的是同一版。
+
+    读不到（库还没建这张表之类）就当没有目录，工具照常可用——目录是附加说明，不能挡住查询。
+    """
+    try:
+        return await data_catalog.read_catalog(session, row.id)
+    except Exception:  # noqa: BLE001
+        logger.exception("读取数据源 %s 的数据目录失败，本次不附目录", row.name)
+        return {}
+
+
 # --------------------------------------------------------------------------
 # 上传表格的版本
 # --------------------------------------------------------------------------
@@ -414,10 +431,12 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo
 # --------------------------------------------------------------------------
 
 
-def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> StructuredTool:
+def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
+                     catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool:
     """source 是手工源的 DataSource，或上传源绑定快照的 SourceView（字段同名）。
 
     fixed：上传源的版本是这次运行固定的（找不到时的说法不同）。
+    catalog：建工具时读到的数据目录，随表结构快照冻结（_store_schema）。
     """
     # 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句，
     # 引擎按它告诉界面上限是多少（metadata["timeout_s"]）
@@ -449,7 +468,7 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
         payload["source"] = source.name
         # 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源
         # 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份
-        if schema := await _store_schema(source, ctx):
+        if schema := await _store_schema(source, ctx, catalog=catalog):
             payload["schema_artifact"] = schema
         # 查询快照进工件库：出具体系的数字回指要能下钻到"这个数是哪条 SQL 查出来的"
         try:
@@ -465,9 +484,12 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
                 extra["mask_columns"] = mask
             if version := getattr(source, "snapshot_id", None):
                 extra["data_version"] = version
+            # 工件引用的 meta 记下源和 SQL 里用到的表：数据目录按它数每张表被运行查询过几次
+            # （catalog.table_usage），不用再逐个读工件内容。meta 不进内容哈希，快照本身一字不变
             payload["artifact"] = await put_json(
-                {**payload, **extra} if extra else payload, kind="query_snapshot",
+                {**payload, **extra} if extra else payload, kind=data_catalog.QUERY_SNAPSHOT_KIND,
                 run_id=ctx.run_id or "", node_id=ctx.node_id or "",
+                meta=data_catalog.query_snapshot_meta(source, str(payload.get("sql") or sql)),
             )
         except Exception:  # noqa: BLE001 - 存不下不影响查询本身
             pass
@@ -485,7 +507,8 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
     )
 
 
-async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
+async def _store_schema(source: Any, ctx: ToolContext, *,
+                        catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None:
     """把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。
 
     上传源传进来的是绑定快照的视图，schema_cache 就是快照里冻结的那份：查的是哪一版，冻结的
@@ -493,6 +516,10 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
 
     内容寻址：结构没变的话，同一次运行里查多少次都是同一件，文件只有一份。没探查过结构
     （或者探查失败、一张表都没有）返回 None：没有东西可以冻结，报告也就不核对表名。
+
+    数据目录一起冻结，放在顶层键 catalog 下：{表名: {version, notes（去掉驳回项）}}，只收这份结构里有的、
+    有目录的表（catalog.frozen_catalog）。这次运行的模型看到的是哪一版目录，事后按快照就能回溯。已有字段
+    一个不动；没有目录的源不加这个键，快照和以前一字不差；目录不变（版本和内容都不变）则快照不变。
     """
     cache = source.schema_cache or {}
     tables = cache.get("tables")
@@ -503,15 +530,25 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
 
     content = {"source": source.name, "tables": tables,
                **{k: cache[k] for k in _FROZEN_SCHEMA_KEYS if k in cache}}
+    meta: dict[str, Any] = {"source": source.name, "tables": len(tables)}
+    if frozen := data_catalog.frozen_catalog(catalog or {}, tables):
+        content[CATALOG_SNAPSHOT_KEY] = frozen
+        meta["catalog_tables"] = len(frozen)
     try:
         return await put_json(content, kind=SCHEMA_SNAPSHOT, run_id=ctx.run_id or "", node_id=ctx.node_id or "",
-                              meta={"source": source.name, "tables": len(tables)})
+                              meta=meta)
     except Exception:  # noqa: BLE001 - 存不下不影响查询本身，只是这次报告不核对表名
         return None
 
 
-def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool:
-    """source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。"""
+def _make_schema_tool(source: Any, *, fixed: bool = False,
+                      catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool:
+    """source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。
+
+    catalog：建工具时读到的数据目录，查单表时附在字段清单后面（introspect.describe_table）。
+    """
+    notes = {name: entry.notes for name, entry in (catalog or {}).items()}
+
     async def _run(table: str | list[str] | None = None) -> str:
         if gone := _vanished(source, fixed):
             return f"{DATA_UNAVAILABLE}{gone}"
@@ -520,7 +557,7 @@ def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool:
         # 而模型两种都会写。为此拒绝一次调用，纯属浪费一步
         wanted = [t.strip() for one in wanted for t in str(one).split(",") if t.strip()]
         if wanted:
-            return "\n\n".join(introspect.describe_table(source, t) for t in wanted)
+            return "\n\n".join(introspect.describe_table(source, t, catalog=notes) for t in wanted)
         tables = introspect.table_names(source)
         if not tables:
             # 把实情交出去，而不是让它去点一个它点不到的按钮

backend/tests/test_catalog_api.py

@@ -187,3 +187,13 @@ async def test_draft_without_schema_is_refused(client):
     r = await client.post(f"{_base(sid)}/draft", json={})
     assert r.status_code == 409 and "探查结构" in r.json()["detail"]
     assert (await client.get(_base(sid))).json()["tables"] == []
+
+
+async def test_schema_endpoint_detail_carries_catalog(client, source_id):
+    """数据源页「表结构」里给模型看的那段文本和 db_schema 工具同一个说法，也附上目录。"""
+    before = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json()
+    assert catalog.NOTES_HEADING not in before["detail"]
+    await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
+    after = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json()
+    assert catalog.NOTES_HEADING in after["detail"] and "gate_id → gates.id" in after["detail"]
+    assert after["detail"].startswith(before["detail"])

backend/tests/test_catalog_runtime.py

@@ -0,0 +1,185 @@
+"""数据目录在运行时：db_schema 工具输出附上目录，查询时目录随表结构快照冻结进证据，查询快照记下用到的表。
+
+- describe_table（db_schema 工具的输出）附上这张表的目录；推断的项标「推断，未确认」，驳回的不出现。
+  导入表格的源保留系统生成的说明，目录只追加说明没有覆盖的、或人工确认过的项。
+- 表结构快照新增顶层键 catalog：{表: {version, notes（去掉驳回项）}}。没有目录的源一个字都不加；
+  目录不变则快照不变（内容寻址，哈希相同）。
+"""
+from __future__ import annotations
+
+import json
+import uuid
+from types import SimpleNamespace
+
+from app.core.artifact_store import canonical_json, content_hash, load
+from app.data import catalog
+from app.data.catalog import make_item
+from app.data.engine import engines
+from app.data.introspect import describe_table, introspect
+from app.db.base import SessionLocal
+from app.db.models import Artifact, DataSource
+from app.tools.datasource import _store_schema, build_datasource_tools
+from app.tools.registry import ToolContext
+
+MARK = "（推断，未确认）"
+
+
+def _rel(table, columns, to_table, to_columns, source="fk", status=None):
+    return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table,
+            "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None, "source": source,
+            "status": status or catalog.initial_status(source)}
+
+
+CACHE = {
+    "schema": "main", "synced_at": "2026-10-01T00:00:00+00:00", "truncated": False, "total": 2,
+    "tables": {
+        "visits": {"qualified": "visits", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"],
+                   "columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None},
+                               {"name": "park_id", "type": "INTEGER", "nullable": False, "comment": None},
+                               {"name": "visitor_count", "type": "INTEGER", "nullable": False, "comment": None},
+                               {"name": "status", "type": "INTEGER", "nullable": False, "comment": None}]},
+        "parks": {"qualified": "parks", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"],
+                  "columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None}]},
+    },
+}
+
+NOTES = {
+    "label": make_item("入园记录", "human"),
+    "grain": make_item("每张门票每次检票一行", "llm"),
+    "keys": make_item(["id"], "human"),
+    "business_date": make_item({"column": "visit_time", "rule": "按检票时间计"}, "human"),
+    "valid_filter": make_item("status = 1", "llm"),
+    "dedup": make_item("按票号去重", "llm", "rejected"),
+    "columns": {"visitor_count": {"meaning": make_item("这次检票进园的人数", "llm"), "unit": make_item("人", "human"),
+                                  "measure": make_item("flow", "llm")},
+                "status": {"codes": make_item({"1": "有效", "0": "作废"}, "human")}},
+    "relations": [_rel("visits", ["park_id"], "parks", ["id"])],
+}
+
+
+def _source(cache=CACHE, **kw) -> SimpleNamespace:
+    return SimpleNamespace(id=kw.pop("id", "src-x"), name=kw.pop("name", "scenic"), kind="sqlite", options={},
+                           readonly=True, description="", schema_cache=cache, origin=kw.pop("origin", "manual"), **kw)
+
+
+# ---------------------------------------------------------------- describe_table
+
+
+def test_describe_table_appends_catalog():
+    text = describe_table(_source(), "visits", catalog={"visits": NOTES})
+    head, _, block = text.partition(catalog.NOTES_HEADING)
+    assert head.startswith("表 visits") and "  visitor_count  INTEGER" in head     # 原来的列清单照旧
+    assert "中文名：入园记录" in block and f"粒度：每张门票每次检票一行{MARK}" in block
+    assert "业务主键：id" in block and "业务日期：visit_time；规则：按检票时间计" in block
+    assert f"有效记录条件：status = 1{MARK}" in block
+    assert f"visitor_count：这次检票进园的人数{MARK}；单位：人；度量类型：流量，可跨期加总{MARK}" in block
+    assert "status：码值：1=有效、0=作废" in block
+    assert "park_id → parks.id（多对一）" in block
+    assert "按票号去重" not in text
+
+
+def test_describe_table_without_catalog_is_unchanged():
+    plain = describe_table(_source(), "visits")
+    assert describe_table(_source(), "visits", catalog={}) == plain
+    assert describe_table(_source(), "visits", catalog={"parks": NOTES}) == plain
+    assert catalog.NOTES_HEADING not in plain
+    # 全名、大小写不一致也对得上目录
+    assert catalog.NOTES_HEADING in describe_table(_source(), "MAIN.Visits", catalog={"visits": NOTES})
+
+
+def test_describe_table_keeps_system_notes_of_imported_tables():
+    """导入表格：说明是按核对结果生成的。保留说明，目录不重复它、不和它抢话。"""
+    cache = {"import_mode": "recipe", "tables": {"日报": {
+        "qualified": "日报", "comment": "粒度：每个日期。不要把各入口相加", "primary_key": [],
+        "columns": [{"name": "日期", "type": "TEXT", "nullable": False, "comment": "格式 YYYY-MM-DD"},
+                    {"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}}}
+    notes = {"label": make_item("客流日报", "llm"), "grain": make_item("每天一行", "llm"),
+             "columns": {"客流": {"unit": make_item("人", "llm"), "measure": make_item("flow", "llm")},
+                         "日期": {"meaning": make_item("统计日期", "human")}}}
+    text = describe_table(_source(cache, origin="upload"), "日报", catalog={"日报": notes})
+    assert text.startswith("表 日报（粒度：每个日期。不要把各入口相加）")
+    assert "单位：人次" in text and "格式 YYYY-MM-DD" in text                 # 系统说明原样保留
+    assert "每天一行" not in text                                            # 表说明已写粒度：推断的粒度不追加
+    assert "单位：人" + MARK not in text                                     # 列说明已写单位：推断的单位不追加
+    assert "客流日报" in text and "流量" in text                              # 说明没有覆盖的照常追加
+    assert "日期：统计日期" in text                                          # 人工确认过的追加
+
+
+# ---------------------------------------------------------------- 冻结进表结构快照
+
+
+def _entry(notes, version=1):
+    return {"visits": catalog.CatalogEntry("visits", notes, version)}
+
+
+async def test_store_schema_freezes_catalog_and_is_content_addressed():
+    ctx = SimpleNamespace(run_id="", node_id="")
+    plain = await _store_schema(_source(), ctx)
+    before = {"source": "scenic", "tables": CACHE["tables"],
+              **{k: CACHE[k] for k in ("schema", "synced_at", "truncated", "total")}}
+    assert plain == content_hash(canonical_json(before))                   # 没有目录：和以前一字不差
+    assert await _store_schema(_source(), ctx, catalog={}) == plain
+
+    first = await _store_schema(_source(), ctx, catalog=_entry(NOTES, 3))
+    frozen = load(first)
+    assert frozen["catalog"]["visits"]["version"] == 3
+    assert "dedup" not in frozen["catalog"]["visits"]["notes"]              # 驳回的不冻结
+    assert frozen["tables"] == CACHE["tables"]                              # 已有字段的语义不变
+    # 目录不变：同一件工件
+    assert await _store_schema(_source(), ctx, catalog=_entry(json.loads(json.dumps(NOTES)), 3)) == first
+    # 目录变了（升了版本）：换一件
+    assert await _store_schema(_source(), ctx, catalog=_entry(NOTES, 4)) != first
+    # 只有驳回项、或者表不在快照里：不加这个键
+    assert await _store_schema(_source(), ctx, catalog={
+        "visits": catalog.CatalogEntry("visits", {"dedup": make_item("x", "llm", "rejected")}, 1),
+        "gone": catalog.CatalogEntry("gone", {"label": make_item("x", "human")}, 1)}) == plain
+
+
+def test_snapshot_readers_ignore_catalog_key():
+    """读表结构快照的地方（证据台账的表结构条目、裁判摘录的口径行）只认自己的键，多一个 catalog 不影响。"""
+    from app.data.provenance import judge_lines
+    from app.engine.evidence import schema_entry_fields
+
+    tables = {"日报": {"qualified": "日报", "comment": "粒度：每个日期", "primary_key": [],
+                       "columns": [{"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}}
+    base = {"source": "flow", "tables": tables, "schema": "main"}
+    with_catalog = {**base, "catalog": {"日报": {"version": 2, "notes": {"label": make_item("客流日报", "human")}}}}
+    assert schema_entry_fields(with_catalog) == schema_entry_fields(base)
+    query = {"data_version": "snap1", "columns": ["客流"], "sql": "SELECT 客流 FROM 日报"}
+    lines = judge_lines(query, base, lambda _id: None, tables=["日报"], hidden=set())
+    assert lines and judge_lines(query, with_catalog, lambda _id: None, tables=["日报"], hidden=set()) == lines
+
+
+# ---------------------------------------------------------------- 工具里串起来
+
+
+async def test_tools_show_and_freeze_catalog_and_record_query_tables(scenic_db):
+    async with SessionLocal() as session:
+        row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=scenic_db)
+        session.add(row)
+        await session.commit()
+        row.schema_cache = await introspect(row)
+        await session.commit()
+        await catalog.write_entry(session, row.id, "visits", {"label": make_item("入园记录", "human"),
+                                                               "grain": make_item("每张门票每次检票一行", "llm")},
+                                  if_version=0, actor="王敏")
+        run_id = uuid.uuid4().hex
+        ctx = ToolContext(run_id=run_id, node_id="n1", data_versions=None)
+        tools = {t.name: t for t in await build_datasource_tools(
+            [f"db_query__{row.name}", f"db_schema__{row.name}"], ctx, session)}
+    try:
+        shown = await tools[f"db_schema__{row.name}"].coroutine(table="visits")
+        assert "中文名：入园记录" in shown and f"粒度：每张门票每次检票一行{MARK}" in shown
+        other = await tools[f"db_schema__{row.name}"].coroutine(table="parks")
+        assert catalog.NOTES_HEADING not in other
+
+        result = json.loads(await tools[f"db_query__{row.name}"].coroutine(
+            sql="SELECT v.id FROM visits v JOIN parks p ON p.id = v.park_id LIMIT 3"))
+        frozen = load(result["schema_artifact"])
+        assert frozen["catalog"] == {"visits": {"version": 1, "notes": {
+            "label": make_item("入园记录", "human"), "grain": make_item("每张门票每次检票一行", "llm")}}}
+        async with SessionLocal() as session:
+            ref = await session.get(Artifact, (result["artifact"], run_id))
+        assert ref.meta == {"source_id": row.id, "source": row.name, "tables": ["visits", "parks"]}
+    finally:
+        await engines.invalidate(row.id)