数据目录审阅队列：推断的项按「使用次数权重 × 风险权重 × 来源权重」排好（有效记录条件、去重规则、业务日期、关联关系最高，模型起草、命名推断比库注释优先），带排在这里的理由，可按影响数字的项 / 名称和说明、按表筛选和分页；批量确认、驳回、撤销审阅逐表按版本乐观锁，同一张表的几条一次写入，一张表冲突不影响别的表；表被哪些运行用过（新的在前，取法同使用次数，抽出 query_snapshot_tables 共用）；审阅进度（常用 20 张表、全部、影响数字的项）

backend/app/api/catalog.py

@@ -13,6 +13,9 @@
 - POST   /catalog/draft-jobs     建后台起草任务（202）；GET /draft-jobs/active、/latest、/{job_id} 读任务；
                                  POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py）
 - POST   /catalog/profile        同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py）
+- GET    /catalog/review-queue   审阅队列：推断的项按使用次数、风险、来源排好；POST /catalog/review/batch 批量审阅，
+                                 GET /catalog/review-progress 审阅进度，GET /catalog/{table}/usage 用过这张表的运行
+                                 （data/catalog_review.py）
 
 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
 不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。
@@ -24,7 +27,7 @@ from __future__ import annotations
 
 from typing import Any, Literal
 
-from fastapi import APIRouter, Body, Depends, Header, HTTPException
+from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query
 from fastapi.exceptions import RequestValidationError
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -35,6 +38,7 @@ from app.api.coded import (
 )
 from app.api.runs import actor_of
 from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions
+from app.data import catalog_review
 from app.data.engine import SnapshotTampered
 from app.db.base import get_session
 from app.db.models import DataSource
@@ -234,6 +238,30 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
     }
 
 
+# ---- 审阅队列、审阅进度（要登记在 GET /catalog/{table} 前面：不然 review-queue、review-progress 会被当成表名）
+
+
+@router.get("/{source_id}/catalog/review-queue")
+async def catalog_review_queue(source_id: str,
+                               limit: int = Query(50, ge=1, le=catalog_review.QUEUE_LIMIT_MAX),
+                               offset: int = Query(0, ge=0),
+                               kind: Literal["all", "facts", "names"] = Query("all"),
+                               table: str | None = Query(None, max_length=255),
+                               session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """审阅队列：只列推断的项，按使用次数、风险、来源排好（data/catalog_review.py）。kind 筛影响数字的项（facts）或名称
+    和说明（names），table 只看一张表。"""
+    _, source = await _resolved(session, source_id)
+    return await catalog_review.review_queue(session, source, kind=kind, table=table or None, limit=limit,
+                                             offset=offset)
+
+
+@router.get("/{source_id}/catalog/review-progress")
+async def catalog_review_progress(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """审阅进度：常用 20 张表、全部、影响数字的项各自确认了多少。"""
+    _, source = await _resolved(session, source_id)
+    return await catalog_review.review_progress(session, source)
+
+
 @router.get("/{source_id}/catalog/{table}")
 async def get_catalog_table(source_id: str, table: str,
                             session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
@@ -527,6 +555,38 @@ async def review_catalog_item(source_id: str, table: str, payload: CatalogReview
     return await _detail(session, source, key, entry)
 
 
+class CatalogBatchReviewItem(BaseModel):
+    table_name: str = Field(min_length=1, max_length=255)
+    path: str = Field(min_length=1, max_length=600)
+    action: Literal["confirm", "reject", "reset"]
+    if_version: int = Field(ge=0)
+
+
+class CatalogBatchReviewIn(BaseModel):
+    items: list[CatalogBatchReviewItem] = Field(min_length=1, max_length=catalog_review.BATCH_MAX)
+
+
+@router.post("/{source_id}/catalog/review/batch")
+async def review_catalog_batch(source_id: str, payload: CatalogBatchReviewIn,
+                               x_actor: str | None = Header(default=None),
+                               session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """批量审阅：逐条结果和请求一一对应。逐表按 if_version 乐观锁，同一张表的几条一次写入；一张表冲突、路径不对、表不存在
+    只记在那几条上（ok 为假，code 是 catalog_conflict / catalog_path_invalid / table_missing），不影响别的。"""
+    _, source = await _resolved(session, source_id)
+    items = [catalog_review.BatchItem(i.table_name, i.path, i.action, i.if_version) for i in payload.items]
+    return {"results": await catalog_review.review_batch(session, source, items, actor=actor_of(x_actor))}
+
+
+@router.get("/{source_id}/catalog/{table}/usage")
+async def catalog_table_runs(source_id: str, table: str,
+                             limit: int = Query(20, ge=1, le=catalog_review.RUNS_LIMIT_MAX),
+                             session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """用过这张表的运行（total 是几次运行），新的在前；取法同使用次数（catalog.table_usage）。"""
+    _, source = await _resolved(session, source_id)
+    key, _ = await _table_key(session, source, table)
+    return await catalog_review.table_runs(session, source, key, limit=limit)
+
+
 def _tables_of(source: Any) -> dict[str, Any] | None:
     """核对提案用的表结构；没有表结构（尚未探查）时不核对列名。"""
     return ((source.schema_cache or {}).get("tables") or None) if source is not None else None

backend/app/data/catalog.py

@@ -2249,27 +2249,27 @@ def query_snapshot_meta(source: Any, sql: str) -> dict[str, Any]:
     return {"source_id": getattr(source, "id", None), "source": getattr(source, "name", None), "tables": tables}
 
 
-async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]:
-    """每张表被运行查询过的次数：{schema_cache 的表名: 次数}，没被查过的表不出现。
+async def query_snapshot_tables(session: AsyncSession, source: Any) -> list[tuple[str, frozenset[str]]]:
+    """这个源上属于运行的每一行查询快照引用：(运行 id, 这条查询用到的表——schema_cache 的表名)，没对上任何表的不收。
 
-    取法：数据源查询工具每次成功查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里
-    按 (内容哈希, 运行) 各记一行。数「属于某次运行（runs 里有这次运行）、属于这个源、SQL 里用到了这张表」的
-    行：同一次运行里结果一模一样的重复查询只算一次，工具库里的试用（不属于任何运行）不算。
+    使用次数（table_usage）和「这张表被哪些运行用过」（data/catalog_review.py）共用这一份取法：数据源查询工具每次成功
+    查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里按 (内容哈希, 运行) 各记一行；只收属于某次运行
+    （runs 里有这次运行）、属于这个源的行，工具库里的试用（不属于任何运行）不算。
 
-    表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用
-    没有 meta，退回读工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉
-    schema 前缀后按 name_key（只把 ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。
+    表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用没有 meta，退回读
+    工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉 schema 前缀后按 name_key（只把
+    ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。
     """
     rows = (await session.execute(
-        select(Artifact.id, Artifact.meta).join(Run, Run.id == Artifact.run_id)
+        select(Artifact.id, Artifact.meta, Artifact.run_id).join(Run, Run.id == Artifact.run_id)
         .where(Artifact.kind == QUERY_SNAPSHOT_KIND)
     )).all()
     tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {}
     lookup: dict[str, str] = {}
     for name in tables:
         lookup.setdefault(name_key(name), name)
-    counts: Counter[str] = Counter()
-    for artifact_id, meta in rows:
+    out: list[tuple[str, frozenset[str]]] = []
+    for artifact_id, meta, run_id in rows:
         meta = meta if isinstance(meta, dict) else {}
         if isinstance(meta.get("tables"), list):
             source_id, source_name, names = meta.get("source_id"), meta.get("source"), meta["tables"]
@@ -2281,8 +2281,21 @@ async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]:
                 continue
         elif source_name != source.name:
             continue
-        hit = {lookup.get(name_key(str(n).rsplit(".", 1)[-1])) for n in names}
-        counts.update(t for t in hit if t is not None)
+        hit = frozenset(t for t in (lookup.get(name_key(str(n).rsplit(".", 1)[-1])) for n in names) if t is not None)
+        if hit:
+            out.append((str(run_id), hit))
+    return out
+
+
+async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]:
+    """每张表被运行查询过的次数：{schema_cache 的表名: 次数}，没被查过的表不出现。
+
+    按查询快照引用的行数（query_snapshot_tables）：同一次运行里结果一模一样的重复查询只算一次，同一次运行里不同的查询
+    各算一次，工具库里的试用不算。
+    """
+    counts: Counter[str] = Counter()
+    for _run, hit in await query_snapshot_tables(session, source):
+        counts.update(hit)
     return dict(counts)
 
 
@@ -2337,7 +2350,8 @@ __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",
+    "query_snapshot_meta", "query_snapshot_tables", "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",
     "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_review.py

@@ -0,0 +1,404 @@
+"""数据目录的审阅队列、批量审阅、表被哪些运行用过、审阅进度。
+
+**为什么要有它。** 起草（数据库注释、外键、命名推断、数据剖析、模型）一次写出成百上千项，绝大多数停在「推断」：SQL 检查的
+错误级规则只认已确认、有确证的项，推断的项只作提示。逐表点开确认太慢，也看不出先确认哪几项。队列把推断的项摊平，按
+「这张表常不常用 × 这一项错了影响多大 × 这一项从哪来」排好，最该先看的在前面；勾几项一次确认、驳回或撤销审阅。
+
+- **队列**（review_queue，GET /catalog/review-queue）：只列推断（proposed）的项。priority = 使用次数权重 × 风险权重 ×
+  来源权重，从高到低；同分按表名、路径排，结果稳定。
+  - 使用次数权重：1 + log2(1 + 使用次数)。使用次数同目录清单（catalog.table_usage）：按对数增长，常用表在前，但不至于
+    让常用表上的中文名压过所有表上影响数字的项。
+  - 风险权重（RISK_WEIGHT）：有效记录条件、去重规则、业务日期、关联关系最高（直接改变数字），业务主键、度量类型次之
+    （计数和能不能加总），码值、粒度、表类型再次，中文名、说明、含义、单位最低。
+  - 来源权重（SOURCE_WEIGHT）：模型起草、命名推断是猜的，比数据库注释优先；数据剖析（码值候选、业务日期、没核实的关系）
+    居中。
+  - 两组（kind）：facts 影响数字的项（有效记录条件、去重规则、业务日期、关联关系、业务主键、度量类型）；names 名称和说明
+    （中文名、说明、含义、单位、码值、粒度、表类型）。表的「说明」和中文名、含义一样只是说明文字，归 names。
+  - 表结构里已经没有的表、列上的项不列（确认了也没有用，目录页上另有清理的提示）。
+  - reasons 用中文说为什么排在这里：常用表（使用 N 次）、影响数字的项（有效记录条件）、模型起草，未确认……
+- **批量审阅**（review_batch，POST /catalog/review/batch）：逐表按 if_version 乐观锁，同一张表的几条在一次写入里完成，
+  一张表冲突、写不进去不影响别的表；路径不对、表不存在逐条报（results 和请求一一对应）。审阅的规则同单项审阅
+  （catalog.review_item）。
+- **被哪些运行用过**（table_runs，GET /catalog/{table}/usage）：取法同使用次数（catalog.query_snapshot_tables），按运行
+  去重，新的在前。
+- **进度**（review_progress，GET /catalog/review-progress）：使用次数前 TOP_N 张表（只算被运行用到过的）、全部、影响
+  数字的项各自的确认情况。项的范围和队列一样：表结构里没有的表、列不算。
+"""
+from __future__ import annotations
+
+import math
+from collections import Counter
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.data import catalog
+from app.db.models import Run
+
+#: 队列的筛选：全部、影响数字的项、名称和说明
+KINDS = ("all", "facts", "names")
+#: 队列一页最多几项；批量审阅一次最多几条；「被哪些运行用过」一次最多几次运行
+QUEUE_LIMIT_MAX = 200
+BATCH_MAX = 200
+RUNS_LIMIT_MAX = 100
+#: 进度里的「常用表」取使用次数前几张
+TOP_N = 20
+
+#: 每一项属于哪一组、风险几档：键是 (t, 字段) / (c, 字段) / (r, "")，档 4 最高
+FIELD_GROUP: dict[tuple[str, str], tuple[str, int]] = {
+    ("t", "valid_filter"): ("facts", 4), ("t", "dedup"): ("facts", 4), ("t", "business_date"): ("facts", 4),
+    ("r", ""): ("facts", 4),
+    ("t", "keys"): ("facts", 3), ("c", "measure"): ("facts", 3),
+    ("c", "codes"): ("names", 2), ("t", "grain"): ("names", 2), ("t", "kind"): ("names", 2),
+    ("t", "label"): ("names", 1), ("t", "description"): ("names", 1), ("c", "label"): ("names", 1),
+    ("c", "meaning"): ("names", 1), ("c", "unit"): ("names", 1),
+}
+RISK_WEIGHT = {4: 4.0, 3: 3.0, 2: 2.0, 1: 1.0}
+#: 来源权重：没写的（外键约束、人工填写，它们本来就不是推断）按 1
+SOURCE_WEIGHT = {"llm": 1.5, "name": 1.5, "profile": 1.25, "comment": 1.0}
+
+#: 来源的界面叫法（和 frontend/src/lib/terms.ts 的 CATALOG_SOURCE_LABEL 一致，有测试核对）
+SOURCE_LABEL = {"comment": "数据库注释", "fk": "外键约束", "name": "命名推断", "profile": "数据剖析", "llm": "模型起草",
+                "human": "人工填写"}
+RELATION_LABEL = "关联关系"
+#: reasons 的说法
+REASON_TOP = "常用表（使用 {n} 次）"
+REASON_USED = "使用 {n} 次"
+REASON_RISK = {4: "影响数字的项（{label}）", 3: "影响计数和汇总的项（{label}）", 2: "帮助理解取值的项（{label}）",
+               1: "名称和说明（{label}）"}
+REASON_PENDING_CODES = "{n} 个码值的含义待填写"
+REASON_SOURCE = "{source}，未确认"
+#: 码值含义为空（数据剖析写进来的候选）时的说法，同界面的 CODES_TEXT.pending
+CODE_PENDING = "含义待填写"
+
+#: 批量审阅逐条结果的机读码
+CODE_CONFLICT = "catalog_conflict"
+CODE_PATH = "catalog_path_invalid"
+CODE_INVALID = "catalog_invalid"
+CODE_TABLE_MISSING = "table_missing"
+
+Slot = catalog.Slot
+
+
+# ==========================================================================
+# 一项的路径、叫法、显示
+# ==========================================================================
+
+
+def slot_path(key: Slot) -> str:
+    """槽位 → 审阅接口的路径写法（catalog.parse_path 的反过来）。"""
+    if key[0] == "t":
+        return key[1]
+    if key[0] == "c":
+        return f"columns.{key[1]}.{key[2]}"
+    return f"relations.{key[1]}"
+
+
+def field_of(key: Slot) -> tuple[str, str]:
+    """槽位 → FIELD_GROUP 的键。"""
+    if key[0] == "r":
+        return ("r", "")
+    return (key[0], key[-1])
+
+
+def field_name(key: Slot) -> str:
+    """这一项的字段名：表级、列级的项是字段名，关联关系是 relation。"""
+    return "relation" if key[0] == "r" else key[-1]
+
+
+def field_label(key: Slot) -> str:
+    """字段的界面叫法：中文名、码值、有效记录条件、关联关系……"""
+    if key[0] == "t":
+        return catalog.TABLE_FIELD_LABEL.get(key[1], key[1])
+    if key[0] == "c":
+        return catalog.COLUMN_FIELD_LABEL.get(key[2], key[2])
+    return RELATION_LABEL
+
+
+def item_value(key: Slot, item: Mapping[str, Any]) -> Any:
+    """一项的值：关联关系是两端、基数和覆盖率，别的是 value。"""
+    if key[0] == "r":
+        return {k: item.get(k) for k in ("columns", "to_table", "to_columns", "cardinality", "coverage")}
+    return item.get("value")
+
+
+def _percent(ratio: float) -> str:
+    """覆盖率写成百分数：整数不带小数，其余留一位；不到 100% 的不写成 100%。"""
+    text = f"{ratio * 100:.1f}".rstrip("0").rstrip(".")
+    return "99.9%" if text == "100" and ratio < 1 else f"{text}%"
+
+
+def relation_text(rel: Mapping[str, Any]) -> str:
+    """关联关系写成一行：member_id → members.id（多对一，覆盖率 98%）。"""
+    left = "、".join(str(c) for c in rel.get("columns") or [])
+    right = "、".join(f"{rel.get('to_table')}.{c}" for c in rel.get("to_columns") or [])
+    extra = []
+    if rel.get("cardinality") in catalog.CARDINALITY_LABEL:
+        extra.append(catalog.CARDINALITY_LABEL[rel["cardinality"]])
+    coverage = rel.get("coverage")
+    if isinstance(coverage, (int, float)) and not isinstance(coverage, bool):
+        extra.append(f"覆盖率 {_percent(float(coverage))}")
+    return f"{left} → {right}" + (f"（{'，'.join(extra)}）" if extra else "")
+
+
+def display(key: Slot, value: Any) -> str:
+    """一项的值写成一行字，用界面上的说法（明细表、可累加、多对一……）。"""
+    if key[0] == "r":
+        return relation_text(value if isinstance(value, Mapping) else {})
+    field_ = key[-1]
+    if field_ == "kind":
+        return catalog.UI_KIND_LABEL.get(value, str(value))
+    if field_ == "measure":
+        return catalog.UI_MEASURE_LABEL.get(value, str(value))
+    if field_ == "keys" and isinstance(value, list):
+        return "、".join(str(v) for v in value)
+    if field_ == "codes" and isinstance(value, Mapping):
+        return "、".join(f"{k}={v if str(v).strip() else CODE_PENDING}" for k, v in value.items())
+    if field_ == "business_date" and isinstance(value, Mapping):
+        parts = [str(value.get("column") or "")]
+        if value.get("rule"):
+            parts.append(f"规则：{value['rule']}")
+        if value.get("timezone"):
+            parts.append(f"时区：{value['timezone']}")
+        return "；".join(parts)
+    return value if isinstance(value, str) else str(value)
+
+
+# ==========================================================================
+# 排序
+# ==========================================================================
+
+
+def usage_weight(usage: int) -> float:
+    return 1.0 + math.log2(1 + max(0, int(usage or 0)))
+
+
+def priority(usage: int, key: Slot, source: Any) -> float:
+    """一项的优先级：使用次数权重 × 风险权重 × 来源权重（不取整；排序和交给界面时取四位小数）。"""
+    _, tier = FIELD_GROUP[field_of(key)]
+    return usage_weight(usage) * RISK_WEIGHT[tier] * SOURCE_WEIGHT.get(str(source), 1.0)
+
+
+def top_tables(tables: Mapping[str, Any], usage: Mapping[str, int]) -> list[str]:
+    """使用次数前 TOP_N 张表：只算被运行用到过的，同样多的按表结构里的顺序。"""
+    order = {name: i for i, name in enumerate(tables)}
+    used = [name for name in tables if usage.get(name, 0) > 0]
+    return sorted(used, key=lambda n: (-usage[n], order[n]))[:TOP_N]
+
+
+def _reasons(key: Slot, item: Mapping[str, Any], usage: int, top: bool) -> list[str]:
+    out: list[str] = []
+    if usage > 0:
+        out.append((REASON_TOP if top else REASON_USED).format(n=f"{usage:,}"))
+    _, tier = FIELD_GROUP[field_of(key)]
+    out.append(REASON_RISK[tier].format(label=field_label(key)))
+    if key[0] == "c" and key[2] == "codes" and isinstance(item.get("value"), Mapping):
+        pending = sum(1 for v in item["value"].values() if not str(v).strip())
+        if pending:
+            out.append(REASON_PENDING_CODES.format(n=pending))
+    source = str(item.get("source") or "")
+    out.append(REASON_SOURCE.format(source=SOURCE_LABEL.get(source, source)))
+    return out
+
+
+def live_slots(notes: Mapping[str, Any] | None, meta: Mapping[str, Any] | None) -> list[tuple[Slot, dict[str, Any]]]:
+    """一张表目录里还算数的项：表结构里已经没有的列上的项不算（不知道表结构时都算）。"""
+    columns = catalog._column_names(meta)
+    return [(key, item) for key, item in catalog._slots(notes).items()
+            if not (key[0] == "c" and columns is not None and key[1] not in columns)]
+
+
+# ==========================================================================
+# 队列
+# ==========================================================================
+
+
+async def review_queue(session: AsyncSession, source: Any, *, kind: str = "all", table: str | None = None,
+                       limit: int = 50, offset: int = 0) -> dict[str, Any]:
+    """审阅队列：{total, items}。source 是读表结构用的源（上传源是绑定当前快照的视图）。"""
+    cache = getattr(source, "schema_cache", None) or {}
+    tables = cache.get("tables") or {}
+    wanted = None
+    if table:
+        wanted = catalog.resolve_table_name(cache, table) or table
+    entries = await catalog.read_catalog(session, source.id)
+    usage = await catalog.table_usage(session, source)
+    top = set(top_tables(tables, usage))
+    items: list[dict[str, Any]] = []
+    for name, entry in entries.items():
+        meta = tables.get(name)
+        if meta is None or (wanted is not None and name != wanted):
+            continue
+        used = usage.get(name, 0)
+        for key, item in live_slots(entry.notes, meta):
+            if item.get("status") != "proposed":
+                continue
+            group, _ = FIELD_GROUP[field_of(key)]
+            if kind != "all" and group != kind:
+                continue
+            value = item_value(key, item)
+            items.append({
+                "table_name": name,
+                "path": slot_path(key),
+                "field": field_name(key),
+                "field_label": field_label(key),
+                # 列级的项是哪一列（表级的项、关联关系为 null）
+                "column": key[1] if key[0] == "c" else None,
+                "value": value,
+                "display": display(key, value),
+                "source": item.get("source"),
+                "status": item.get("status"),
+                "note": item.get("note") if isinstance(item.get("note"), str) and item.get("note") else None,
+                "priority": round(priority(used, key, item.get("source")), 4),
+                "reasons": _reasons(key, item, used, name in top),
+                "table_usage": used,
+                "version": entry.version,
+            })
+    items.sort(key=lambda i: (-i["priority"], i["table_name"], i["path"]))
+    return {"total": len(items), "items": items[offset:offset + limit]}
+
+
+# ==========================================================================
+# 批量审阅
+# ==========================================================================
+
+
+@dataclass(frozen=True)
+class BatchItem:
+    table_name: str
+    path: str
+    action: str
+    if_version: int
+
+
+def _result(item: BatchItem, *, ok: bool, version: int | None = None, error: str | None = None,
+            code: str | None = None) -> dict[str, Any]:
+    return {"table_name": item.table_name, "path": item.path, "ok": ok, "version": version if ok else None,
+            "error": error, "code": code}
+
+
+async def review_batch(session: AsyncSession, source: Any, items: Iterable[BatchItem], *,
+                       actor: str | None) -> list[dict[str, Any]]:
+    """批量审阅，返回逐条结果（和请求一一对应）。
+
+    表名按表结构认（全名、大小写不一致都认）；表结构里没有、但还留着目录的也认（同单项审阅）。同一张表的几条按请求里的
+    顺序在当前目录上依次审阅，一次写入：版本不符（if_version 和库里的不一样）、路径不对的那几条不做，其余照做；写入时撞上
+    别人刚改过，这张表的几条都记冲突。"""
+    items = list(items)
+    cache = getattr(source, "schema_cache", None) or {}
+    results: list[dict[str, Any] | None] = [None] * len(items)
+    groups: dict[str, list[int]] = {}
+    known: dict[str, str | None] = {}
+    for i, item in enumerate(items):
+        if item.table_name not in known:
+            key = catalog.resolve_table_name(cache, item.table_name)
+            if key is None and await catalog.read_entry(session, source.id, item.table_name) is not None:
+                key = item.table_name
+            known[item.table_name] = key
+        key = known[item.table_name]
+        if key is None:
+            results[i] = _result(item, ok=False, code=CODE_TABLE_MISSING,
+                                 error=f"数据源中没有表 {item.table_name}，可能已被删除或尚未探查结构")
+            continue
+        groups.setdefault(key, []).append(i)
+    at = catalog.now_iso()
+    for key, indexes in groups.items():
+        entry = await catalog.read_entry(session, source.id, key)
+        current = entry.version if entry else 0
+        notes: dict[str, Any] = entry.notes if entry else {}
+        applied: list[int] = []
+        for i in indexes:
+            item = items[i]
+            if item.if_version != current:
+                results[i] = _result(item, ok=False, code=CODE_CONFLICT, error=str(catalog.CatalogConflict(key, current)))
+                continue
+            try:
+                notes = catalog.review_item(notes, item.path, item.action, at=at)
+            except catalog.CatalogPathError as e:
+                results[i] = _result(item, ok=False, code=CODE_PATH, error=str(e))
+                continue
+            applied.append(i)
+        if not applied:
+            continue
+        try:
+            written = await catalog.write_entry(session, source.id, key, notes, if_version=current, actor=actor)
+        except catalog.CatalogConflict as e:
+            for i in applied:
+                results[i] = _result(items[i], ok=False, code=CODE_CONFLICT, error=str(e))
+            continue
+        except catalog.CatalogInvalid as e:
+            for i in applied:
+                results[i] = _result(items[i], ok=False, code=CODE_INVALID, error=str(e))
+            continue
+        for i in applied:
+            results[i] = _result(items[i], ok=True, version=written.version)
+    return [r for r in results if r is not None]
+
+
+# ==========================================================================
+# 被哪些运行用过
+# ==========================================================================
+
+
+async def table_runs(session: AsyncSession, source: Any, table: str, *, limit: int = 20) -> dict[str, Any]:
+    """用过这张表的运行：{total, runs}，新的在前（按开始时间，没开始的按建立时间）。取法同使用次数。"""
+    run_ids = {run for run, hit in await catalog.query_snapshot_tables(session, source) if table in hit}
+    if not run_ids:
+        return {"total": 0, "runs": []}
+    started = func.coalesce(Run.started_at, Run.created_at)
+    rows = (await session.execute(
+        select(Run.id, Run.workflow_id, Run.workflow_name, Run.run_class, Run.status, Run.started_at)
+        .where(Run.id.in_(run_ids)).order_by(started.desc(), Run.id).limit(limit)
+    )).all()
+    return {
+        "total": len(run_ids),
+        "runs": [{"run_id": r.id, "workflow_id": r.workflow_id, "workflow_name": r.workflow_name or "",
+                  "run_class": r.run_class, "status": r.status,
+                  "started_at": r.started_at.isoformat() if r.started_at else None} for r in rows],
+    }
+
+
+# ==========================================================================
+# 进度
+# ==========================================================================
+
+
+async def review_progress(session: AsyncSession, source: Any) -> dict[str, Any]:
+    """审阅进度：常用 TOP_N 张表（只算被运行用到过的）、全部、影响数字的项。项按状态计数；verified（有确证）另给一项。"""
+    tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {}
+    entries = await catalog.read_catalog(session, source.id)
+    usage = await catalog.table_usage(session, source)
+    top = set(top_tables(tables, usage))
+    every: Counter[str] = Counter()
+    head: Counter[str] = Counter()
+    facts: Counter[str] = Counter()
+    for name, entry in entries.items():
+        meta = tables.get(name)
+        if meta is None:
+            continue
+        for key, item in live_slots(entry.notes, meta):
+            status = str(item.get("status"))
+            every[status] += 1
+            if name in top:
+                head[status] += 1
+            if FIELD_GROUP[field_of(key)][0] == "facts":
+                facts[status] += 1
+    return {
+        "top_n": TOP_N,
+        "top": {"tables": len(top), "items_total": sum(head.values()), "items_confirmed": head["confirmed"],
+                "items_proposed": head["proposed"], "items_verified": head["verified"]},
+        "all": {"items_total": sum(every.values()), "items_confirmed": every["confirmed"],
+                "items_proposed": every["proposed"], "items_rejected": every["rejected"],
+                "items_verified": every["verified"]},
+        "facts": {"proposed": facts["proposed"], "confirmed": facts["confirmed"]},
+    }
+
+
+__all__ = ["BATCH_MAX", "BatchItem", "CODE_CONFLICT", "CODE_INVALID", "CODE_PATH", "CODE_TABLE_MISSING",
+           "FIELD_GROUP", "KINDS", "QUEUE_LIMIT_MAX", "RISK_WEIGHT", "RUNS_LIMIT_MAX", "SOURCE_LABEL", "SOURCE_WEIGHT",
+           "TOP_N", "display", "field_label", "field_name", "field_of", "item_value", "live_slots", "priority",
+           "relation_text", "review_batch", "review_progress", "review_queue", "slot_path", "table_runs", "top_tables",
+           "usage_weight"]

backend/tests/test_catalog_review_queue.py

@@ -0,0 +1,357 @@
+"""数据目录的审阅队列、批量审阅、表被哪些运行用过、审阅进度（data/catalog_review.py）。
+
+- GET  /catalog/review-queue：只列推断（proposed）的项，按「使用次数权重 × 风险权重 × 来源权重」从高到低，同分按表名、
+  路径；facts（有效记录条件、去重规则、业务日期、关联关系、业务主键、度量类型）/ names（中文名、说明、含义、单位、码值、
+  粒度、表类型）两组；按表筛选；分页；reasons 写为什么排在这里；表结构里已经没有的表、列不列
+- POST /catalog/review/batch：逐表按 if_version 乐观锁，一张表冲突不影响别的表；同一张表的几条一次写入；路径不对、表不存在
+  逐条报
+- GET  /catalog/{table}/usage：用过这张表的运行，新的在前，取法同 catalog.table_usage
+- GET  /catalog/review-progress：常用 20 张表、全部、影响数字的项各自的确认情况
+
+表名、列名、业务数字都是虚构的。
+"""
+from __future__ import annotations
+
+import re
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from urllib.parse import quote
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.data import catalog, catalog_review
+from app.data.catalog import make_item
+from app.db.base import SessionLocal
+from app.db.models import Artifact, DataSource, Run
+from app.main import app as _fastapi_app
+from tests.fixtures.sources import drop_source
+
+ACTOR = {"X-Actor": quote("林晓")}
+TERMS = Path(__file__).resolve().parents[2] / "frontend" / "src" / "lib" / "terms.ts"
+
+
+def _col(name: str, type_: str = "INTEGER") -> dict:
+    return {"name": name, "type": type_, "nullable": True, "comment": None}
+
+
+def _tables() -> dict:
+    return {
+        "orders": {"qualified": "orders", "primary_key": ["id"], "comment": None, "is_view": False,
+                   "foreign_keys": [], "unique": [],
+                   "columns": [_col("id"), _col("status"), _col("amount", "REAL"), _col("member_id"),
+                               _col("paid_at", "TEXT")]},
+        "members": {"qualified": "members", "primary_key": ["id"], "comment": None, "is_view": False,
+                    "foreign_keys": [], "unique": [],
+                    "columns": [_col("id"), _col("nick", "TEXT"), _col("tier")]},
+        "coupons": {"qualified": "coupons", "primary_key": ["id"], "comment": None, "is_view": False,
+                    "foreign_keys": [], "unique": [], "columns": [_col("id"), _col("face_value", "REAL")]},
+    }
+
+
+def _member_rel(status: str = "proposed", source: str = "name") -> dict:
+    return {"id": catalog.relation_id("orders", ["member_id"], "members", ["id"]), "columns": ["member_id"],
+            "to_table": "members", "to_columns": ["id"], "cardinality": "many_to_one", "coverage": 0.98,
+            "source": source, "status": status}
+
+
+ORDERS = {
+    "label": make_item("订单", "llm"),
+    "grain": make_item("每笔订单一行", "llm", "confirmed"),
+    "valid_filter": make_item("status <> 9", "llm", note="剖析核对：排除 3.2% 的行（共 12,345 行）"),
+    "kind": make_item("fact", "llm"),
+    "columns": {
+        "status": {"label": make_item("状态", "comment"), "codes": make_item({"1": "已支付", "9": ""}, "profile")},
+        "amount": {"measure": make_item("flow", "llm"), "unit": make_item("元", "llm", "rejected")},
+        # 表结构里已经没有这一列：不进队列
+        "legacy_flag": {"label": make_item("旧标记", "llm")},
+    },
+    "relations": [_member_rel()],
+}
+MEMBERS = {"label": make_item("会员", "llm"), "keys": make_item(["id"], "name"),
+           "columns": {"nick": {"label": make_item("昵称", "comment")}}}
+COUPONS = {"label": make_item("优惠券", "comment")}
+#: 表结构里已经没有这张表：不进队列
+GONE = {"label": make_item("旧表", "llm")}
+
+_CREATED: list[str] = []
+
+
+@pytest.fixture(autouse=True)
+async def _cleanup():
+    yield
+    while _CREATED:
+        await drop_source(_CREATED.pop())
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
+        yield c
+
+
+async def _write(sid: str, table: str, notes: dict) -> None:
+    async with SessionLocal() as session:
+        entry = await catalog.read_entry(session, sid, table)
+        await catalog.write_entry(session, sid, table, notes, if_version=entry.version if entry else 0, actor=None)
+
+
+async def _use(sid: str, name: str, table_runs: dict[str, int]) -> list[str]:
+    """给每张表造几次运行查询它（每次运行一份查询快照），返回运行 id（先建的在前、开始得早）。"""
+    runs: list[str] = []
+    base = datetime(2026, 9, 1, tzinfo=timezone.utc)
+    async with SessionLocal() as session:
+        for table, n in table_runs.items():
+            for _ in range(n):
+                rid = uuid.uuid4().hex
+                session.add(Run(id=rid, graph={}, workflow_name=f"周报-{table}", run_class="formal",
+                                status="succeeded", started_at=base + timedelta(hours=len(runs))))
+                session.add(Artifact(id=uuid.uuid4().hex, run_id=rid, kind=catalog.QUERY_SNAPSHOT_KIND,
+                                     meta={"source_id": sid, "source": name, "tables": [table]}))
+                runs.append(rid)
+        await session.commit()
+    return runs
+
+
+async def _source(*, usage: dict[str, int] | None = None) -> str:
+    name = f"rq_{uuid.uuid4().hex[:10]}"
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=":memory:", schema_cache={"tables": _tables()})
+        session.add(row)
+        await session.commit()
+        sid = row.id
+    _CREATED.append(sid)
+    for table, notes in (("orders", ORDERS), ("members", MEMBERS), ("coupons", COUPONS), ("gone_table", GONE)):
+        await _write(sid, table, notes)
+    if usage:
+        await _use(sid, name, usage)
+    return sid
+
+
+def _base(sid: str) -> str:
+    return f"/api/datasources/{sid}/catalog"
+
+
+# --------------------------------------------------------------------------
+# 排序规则（纯函数）
+# --------------------------------------------------------------------------
+
+
+def test_priority_multiplies_usage_risk_and_source_weights():
+    p = catalog_review.priority
+    # 风险：有效记录条件 > 业务主键 > 码值 > 中文名
+    assert p(0, ("t", "valid_filter"), "llm") > p(0, ("t", "keys"), "llm") > p(0, ("c", "x", "codes"), "llm") \
+        > p(0, ("c", "x", "label"), "llm")
+    assert p(0, ("r", "r1"), "name") == p(0, ("t", "dedup"), "name") == p(0, ("t", "business_date"), "name")
+    assert p(0, ("c", "x", "measure"), "llm") == p(0, ("t", "keys"), "llm")
+    assert p(0, ("t", "grain"), "llm") == p(0, ("t", "kind"), "llm") == p(0, ("c", "x", "codes"), "llm")
+    assert p(0, ("c", "x", "meaning"), "llm") == p(0, ("c", "x", "unit"), "llm") == p(0, ("t", "label"), "llm")
+    # 来源：模型起草、命名推断比数据库注释优先
+    assert p(0, ("t", "label"), "llm") == p(0, ("t", "label"), "name") > p(0, ("t", "label"), "comment")
+    # 使用次数：用得多的在前，但按对数增长，不压过风险
+    assert p(37, ("t", "label"), "llm") > p(1, ("t", "label"), "llm") > p(0, ("t", "label"), "llm")
+    assert p(37, ("t", "label"), "llm") == pytest.approx(
+        p(0, ("t", "label"), "llm") * p(37, ("t", "label"), "comment") / p(0, ("t", "label"), "comment"))
+
+
+def test_source_labels_follow_the_ui():
+    text = TERMS.read_text(encoding="utf-8")
+    block = re.search(r"export const CATALOG_SOURCE_LABEL\b[^=]*=\s*\{(.*?)\n\}", text, re.S)
+    assert block
+    ui = dict(re.findall(r"(\w+):\s*'([^']+)'", block.group(1)))
+    assert ui == catalog_review.SOURCE_LABEL
+
+
+# --------------------------------------------------------------------------
+# 队列
+# --------------------------------------------------------------------------
+
+
+async def test_queue_lists_only_proposed_items_by_priority(client):
+    sid = await _source(usage={"orders": 37, "members": 2})
+    r = await client.get(f"{_base(sid)}/review-queue")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    items = body["items"]
+    paths = [(i["table_name"], i["path"]) for i in items]
+    # 已确认、已驳回的不列；表结构里没有的表、列不列
+    assert ("orders", "grain") not in paths and ("orders", "columns.amount.unit") not in paths
+    assert not any(t == "gone_table" for t, _ in paths)
+    assert ("orders", "columns.legacy_flag.label") not in paths
+    assert body["total"] == len(items) == 11
+    assert all(i["status"] == "proposed" for i in items)
+    # 从高到低，同分按表名、路径
+    keys = [(-i["priority"], i["table_name"], i["path"]) for i in items]
+    assert keys == sorted(keys)
+    first = items[0]
+    rel_id = _member_rel()["id"]
+    assert (first["table_name"], first["path"]) in {("orders", "valid_filter"), ("orders", f"relations.{rel_id}")}
+    vf = next(i for i in items if i["path"] == "valid_filter")
+    assert vf["field"] == "valid_filter" and vf["field_label"] == "有效记录条件"
+    assert vf["value"] == "status <> 9" and vf["display"] == "status <> 9"
+    assert vf["source"] == "llm" and vf["note"] == "剖析核对：排除 3.2% 的行（共 12,345 行）"
+    assert vf["table_usage"] == 37 and vf["version"] == 1
+    assert vf["reasons"] == ["常用表（使用 37 次）", "影响数字的项（有效记录条件）", "模型起草，未确认"]
+    # 常用表上的中文名排在没人用的表上的中文名前面；同一张表上影响数字的项排在中文名前面
+    order = [(i["table_name"], i["path"]) for i in items]
+    assert order.index(("orders", "label")) < order.index(("coupons", "label"))
+    assert order.index(("members", "keys")) < order.index(("members", "label"))
+
+
+async def test_queue_values_are_displayed_in_ui_words(client):
+    sid = await _source()
+    items = {(i["table_name"], i["path"]): i for i in (await client.get(f"{_base(sid)}/review-queue")).json()["items"]}
+    rel = items[("orders", f"relations.{_member_rel()['id']}")]
+    assert rel["field"] == "relation" and rel["field_label"] == "关联关系"
+    assert rel["value"] == {"columns": ["member_id"], "to_table": "members", "to_columns": ["id"],
+                            "cardinality": "many_to_one", "coverage": 0.98}
+    assert rel["display"] == "member_id → members.id（多对一，覆盖率 98%）"
+    assert rel["reasons"][-1] == "命名推断，未确认"
+    kind = items[("orders", "kind")]
+    assert kind["display"] == "明细表" and kind["value"] == "fact"
+    codes = items[("orders", "columns.status.codes")]
+    assert codes["field"] == "codes" and codes["field_label"] == "码值" and codes["column"] == "status"
+    assert codes["display"] == "1=已支付、9=含义待填写"
+    assert "1 个码值的含义待填写" in codes["reasons"] and "数据剖析，未确认" in codes["reasons"]
+    assert items[("orders", "columns.amount.measure")]["display"] == "可累加"
+    assert items[("members", "keys")]["display"] == "id"
+    assert items[("members", "label")]["reasons"] == ["名称和说明（中文名）", "模型起草，未确认"]
+    assert items[("coupons", "label")]["reasons"] == ["名称和说明（中文名）", "数据库注释，未确认"]
+
+
+async def test_queue_filters_by_kind_and_table_and_pages(client):
+    sid = await _source(usage={"orders": 3})
+    facts = (await client.get(f"{_base(sid)}/review-queue", params={"kind": "facts"})).json()
+    assert {i["field"] for i in facts["items"]} == {"valid_filter", "relation", "measure", "keys"}
+    names = (await client.get(f"{_base(sid)}/review-queue", params={"kind": "names"})).json()
+    assert {i["field"] for i in names["items"]} == {"label", "codes", "kind"}
+    assert facts["total"] + names["total"] == 11
+    only = (await client.get(f"{_base(sid)}/review-queue", params={"table": "MEMBERS"})).json()
+    assert {i["table_name"] for i in only["items"]} == {"members"} and only["total"] == 3
+    everything = (await client.get(f"{_base(sid)}/review-queue")).json()["items"]
+    page = (await client.get(f"{_base(sid)}/review-queue", params={"limit": 4, "offset": 4})).json()
+    assert page["total"] == 11 and page["items"] == everything[4:8]
+    bad = await client.get(f"{_base(sid)}/review-queue", params={"kind": "other"})
+    assert bad.status_code == 422
+    assert (await client.get(f"{_base(sid)}/review-queue", params={"limit": 0})).status_code == 422
+    missing = await client.get("/api/datasources/nope/catalog/review-queue")
+    assert missing.status_code == 404
+
+
+# --------------------------------------------------------------------------
+# 批量审阅
+# --------------------------------------------------------------------------
+
+
+async def test_batch_review_writes_each_table_once_and_isolates_conflicts(client):
+    sid = await _source()
+    rel_path = f"relations.{_member_rel()['id']}"
+    body = {"items": [
+        {"table_name": "orders", "path": "valid_filter", "action": "confirm", "if_version": 1},
+        {"table_name": "orders", "path": rel_path, "action": "reject", "if_version": 1},
+        {"table_name": "orders", "path": "columns.nothing.label", "action": "confirm", "if_version": 1},
+        {"table_name": "members", "path": "label", "action": "confirm", "if_version": 7},      # 版本不对
+        {"table_name": "coupons", "path": "label", "action": "reject", "if_version": 1},
+        {"table_name": "ghost", "path": "label", "action": "confirm", "if_version": 0},
+    ]}
+    r = await client.post(f"{_base(sid)}/review/batch", json=body, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    results = r.json()["results"]
+    assert [(x["table_name"], x["path"], x["ok"]) for x in results] == [
+        ("orders", "valid_filter", True), ("orders", rel_path, True), ("orders", "columns.nothing.label", False),
+        ("members", "label", False), ("coupons", "label", True), ("ghost", "label", False)]
+    # 同一张表的两条一次写进去：版本只加 1
+    assert results[0]["version"] == results[1]["version"] == 2
+    assert results[2]["code"] == "catalog_path_invalid" and "找不到" in results[2]["error"]
+    assert results[2]["version"] is None
+    assert results[3]["code"] == "catalog_conflict" and "重新载入" in results[3]["error"]
+    assert results[4]["version"] == 2 and results[4]["error"] is None and results[4]["code"] is None
+    assert results[5]["code"] == "table_missing"
+    async with SessionLocal() as session:
+        orders = await catalog.read_entry(session, sid, "orders")
+        members = await catalog.read_entry(session, sid, "members")
+        coupons = await catalog.read_entry(session, sid, "coupons")
+    assert orders.version == 2 and orders.updated_by == "林晓"
+    assert orders.notes["valid_filter"]["status"] == "confirmed"
+    assert orders.notes["relations"][0]["status"] == "rejected"
+    assert members.version == 1 and members.notes["label"]["status"] == "proposed"
+    assert coupons.notes["label"]["status"] == "rejected"
+    # 撤销审阅：回到来源的初始状态
+    again = await client.post(f"{_base(sid)}/review/batch", json={"items": [
+        {"table_name": "orders", "path": "valid_filter", "action": "reset", "if_version": 2}]})
+    assert again.json()["results"][0]["ok"] is True
+    async with SessionLocal() as session:
+        assert (await catalog.read_entry(session, sid, "orders")).notes["valid_filter"]["status"] == "proposed"
+
+
+async def test_batch_review_limits(client):
+    sid = await _source()
+    many = [{"table_name": "orders", "path": "label", "action": "confirm", "if_version": 1}] * 201
+    assert (await client.post(f"{_base(sid)}/review/batch", json={"items": many})).status_code == 422
+    assert (await client.post(f"{_base(sid)}/review/batch", json={"items": []})).status_code == 422
+    wrong = [{"table_name": "orders", "path": "label", "action": "approve", "if_version": 1}]
+    assert (await client.post(f"{_base(sid)}/review/batch", json={"items": wrong})).status_code == 422
+    ok = [{"table_name": "orders", "path": "label", "action": "confirm", "if_version": 1}] * 200
+    r = await client.post(f"{_base(sid)}/review/batch", json={"items": ok})
+    assert r.status_code == 200 and all(x["ok"] for x in r.json()["results"])
+
+
+# --------------------------------------------------------------------------
+# 被哪些运行用过
+# --------------------------------------------------------------------------
+
+
+async def test_table_usage_lists_runs_newest_first(client):
+    sid = await _source()
+    async with SessionLocal() as session:
+        name = (await session.get(DataSource, sid)).name
+    runs = await _use(sid, name, {"orders": 3, "members": 1})
+    # 同一次运行查了两次 orders：只算一次运行
+    async with SessionLocal() as session:
+        session.add(Artifact(id=uuid.uuid4().hex, run_id=runs[0], kind=catalog.QUERY_SNAPSHOT_KIND,
+                             meta={"source_id": sid, "source": name, "tables": ["orders", "members"]}))
+        # 别的数据源的同名表不算
+        other = uuid.uuid4().hex
+        session.add(Run(id=other, graph={}, workflow_name="别处", status="succeeded"))
+        session.add(Artifact(id=uuid.uuid4().hex, run_id=other, kind=catalog.QUERY_SNAPSHOT_KIND,
+                             meta={"source_id": "elsewhere", "source": "elsewhere", "tables": ["orders"]}))
+        await session.commit()
+    r = await client.get(f"{_base(sid)}/orders/usage")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["total"] == 3
+    assert [x["run_id"] for x in body["runs"]] == [runs[2], runs[1], runs[0]]
+    first = body["runs"][0]
+    assert first["workflow_name"] == "周报-orders" and first["run_class"] == "formal"
+    assert first["status"] == "succeeded" and first["started_at"].startswith("2026-09-01T02:00")
+    assert set(first) == {"run_id", "workflow_id", "workflow_name", "run_class", "status", "started_at"}
+    limited = (await client.get(f"{_base(sid)}/orders/usage", params={"limit": 1})).json()
+    assert limited["total"] == 3 and len(limited["runs"]) == 1
+    members = (await client.get(f"{_base(sid)}/members/usage")).json()
+    assert members["total"] == 2
+    assert (await client.get(f"{_base(sid)}/ghost/usage")).status_code == 404
+
+
+# --------------------------------------------------------------------------
+# 进度
+# --------------------------------------------------------------------------
+
+
+async def test_review_progress_counts_top_tables_all_and_facts(client):
+    sid = await _source(usage={"orders": 5})
+    r = await client.get(f"{_base(sid)}/review-progress")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["top_n"] == 20
+    # 常用表只算被运行用到过的表：这里只有 orders（9 项：表结构里没有的列不算）
+    assert body["top"] == {"tables": 1, "items_total": 9, "items_confirmed": 1, "items_proposed": 7,
+                           "items_verified": 0}
+    assert body["all"] == {"items_total": 13, "items_confirmed": 1, "items_proposed": 11, "items_rejected": 1,
+                           "items_verified": 0}
+    # 影响数字的项：有效记录条件、关联关系、度量类型、业务主键
+    assert body["facts"] == {"proposed": 4, "confirmed": 0}
+    await client.post(f"{_base(sid)}/review/batch", json={"items": [
+        {"table_name": "orders", "path": "valid_filter", "action": "confirm", "if_version": 1}]})
+    after = (await client.get(f"{_base(sid)}/review-progress")).json()
+    assert after["top"]["items_confirmed"] == 2 and after["facts"] == {"proposed": 3, "confirmed": 1}