集成收尾：前端约定写入后端实际返回的形状（应用详情只对 manage 给草稿、补算计划与结果、按名字的钩子行、能力字段审阅），去掉多种字段名都认的兼容解析；钩子按（应用, 名字）、应用助手对话按（应用, 建立人）加唯一索引，迁移前去重，插入改在保存点里、撞上就改用已有的那一行；口径卡固定的版本也须已发布且未退役，记级别之前有正式运行凭据的老版本启动时补记为已发布；证据面板和应用视图按列名遮端点快照（同一个函数），不再整张表遮掉；待办计数带上最新一条的时间，新待办通知不再漏；非管理员的搭建者不出模型接入的改动按钮

2026-11-15 00:11:59 +00:00 · 2026-10-14 22:08:33 +08:00
parent 1dd4f93579
commit bff3062a97
32 changed files with 419 additions and 246 deletions
@@ -302,18 +302,22 @@ async def assistant_conversation(app_id: str, session: AsyncSession = Depends(ge
    lookup = (select(Conversation).where(Conversation.kind == "app", Conversation.app_id == app.id,
                                         Conversation.archived.is_(False), mine)
              .order_by(Conversation.last_active_at.desc()).limit(1))
-    # 取或建在进程里按 (应用, 人) 串行：编辑器开发模式下 effect 执行两次、同一个人开两个标签页，都会同时来取。库里没有
-    # 唯一约束，建完再查一遍：别处（另一个进程）抢先建了的，删掉刚建的这段（还没有轮次），用那一段——历史才不会被拆开
+    # 取或建在进程里按 (应用, 人) 串行：编辑器开发模式下 effect 执行两次、同一个人开两个标签页，都会同时来取。库里有
+    # (app_id, created_by) 的部分唯一索引（ux_conversations_app_owner）：别处（另一个进程）抢先建了的，这次插入在保存点里
+    # 被拒，改用那一段——历史才不会被拆开。没署名（关登录又没填）的不受索引约束，靠这把进程内的锁
    async with _assistant_lock(app.id, who):
        row = await session.scalar(lookup)
        if row is None:
-            row = Conversation(kind="app", app_id=app.id, created_by=who, title=f"「{app.name}」的助手")
-            session.add(row)
-            await session.flush()
-            other = await session.scalar(lookup.where(Conversation.id != row.id))
-            if other is not None:
-                await session.delete(row)
-                row = other
+            fresh = Conversation(kind="app", app_id=app.id, created_by=who, title=f"「{app.name}」的助手")
+            try:
+                async with session.begin_nested():
+                    session.add(fresh)
+                    await session.flush()
+                row = fresh
+            except IntegrityError:
+                row = await session.scalar(lookup)
+                if row is None:
+                    raise
            await session.commit()
            await session.refresh(row)
    turns = list((await session.execute(select(ConversationTurn).where(
@@ -2139,6 +2139,13 @@ async def _mask_sets(snap: dict[str, Any] | None, source: Any, masks: _Masks, *,
    artifact 是 snap 的工件 id：给了的，分析结果按快照记在进程里（_MEMO，键见 _Masks.memo_key），之后的请求不再分析。"""
    if snap is not None and snap.get("source") == MERGE_SOURCE:
        return await _merge_mask_sets(snap, masks, depth=depth)
+    if snap is not None and snap.get("capability"):
+        # 端点快照：列是能力登记的表格映射排出来的，没有 SQL 可认、也不该按数据源认（端点不是数据源，按数据源找不到会把
+        # 整张表都当成「数据源不在了」遮掉）。按列名遮：落快照时记下的加能力现在的遮罩字段，确定就是这几列
+        from app.capabilities.registry import snapshot_masked_columns
+
+        base = await snapshot_masked_columns(snap)
+        return _MaskSets(set(base), set(base), base)
    name = source or (snap or {}).get("source")
    current, found = await masks.of(name)
    base = set(current) | (_recorded_masks(snap) if snap is not None else set())
@@ -61,8 +61,8 @@ async def list_tasks(


@router.get("/count")
-async def count_tasks(mine: int = 1, x_actor: str | None = Header(default=None)) -> dict[str, int]:
-    """导航徽标：派给我的、待处理的待办数，按来源分开（apps：应用里的；other：画布、定时任务的审批）。"""
+async def count_tasks(mine: int = 1, x_actor: str | None = Header(default=None)) -> dict[str, Any]:
+    """导航徽标：派给我的、待处理的待办数，按来源分开（apps：应用里的；other：画布、定时任务的审批）；latest 是最新一条的开出时间。"""
    return await task_service.count(principal_of(x_actor))


@@ -164,7 +164,7 @@ async def _lint_subgraph_pins(session: AsyncSession, spec: dict[str, Any], pins:
        except ValueError:
            continue
        for problem in await subgraph_pin_problems(session, graph):
-            out.error(problem.code, f"「{state.label}」里的子工作流「{problem.title}」固定的{problem.reason}：这一版的正式运行"
+            out.error(problem.code, f"「{state.label}」里的{problem.who}固定的{problem.reason}：这一版的正式运行"
                      "发起时会被拒绝。请在那个工作流里改好并发布新版本，再把这里的引用换成新版本", f"pins[{i}]",
                      node_id=problem.node_id)

@@ -869,36 +869,20 @@ def _query_artifact(source: Source, node_id: str, exec_no: Any = None) -> str |
 async def _snapshot(source: Source, artifact: str) -> tuple[dict[str, Any] | None, set[str]]:
    """查询快照和要遮的列（小写）：查询当时记下的加数据源现在设的，按 SQL 认出改了名字的也遮（同证据面板）。端点快照
    按列名遮（见下）。"""
-    from app.api.evidence import _fetch, _recorded_masks
+    from app.api.evidence import _fetch

    snap, ok = _fetch(artifact)
    if ok is not True or not isinstance(snap, dict) or not isinstance(snap.get("rows"), list):
        return None, set()
    if snap.get("capability"):
-        # 端点快照：列是能力登记的表格映射排出来的，没有 SQL 改名这回事，按列名遮就是准的——落快照时记下的，加上能力
-        # 现在的遮罩字段。查询快照那一套（按 SQL 认改了名字的遮罩列）在这里认不出出处，会把整张表都遮掉
-        return snap, _recorded_masks(snap) | await _capability_masks(str(snap["capability"]))
+        # 端点快照：按列名遮（落快照时记下的，加上能力现在的遮罩字段），证据面板用的是同一个函数
+        from app.capabilities.registry import snapshot_masked_columns
+
+        return snap, await snapshot_masked_columns(snap)
    sets = await source.masks.sets(artifact, snap.get("source"), snap)
    return snap, set(sets.hidden)


-async def _capability_masks(name: str) -> set[str]:
-    """端点快照（capabilities/endpoints.py 落的 endpoint_snapshot）按能力现在的遮罩字段要遮的列（小写）。
-
-    快照里的 mask_columns 是落快照那一刻的遮罩（调用方另外并进来）；管理员事后给能力加了遮罩字段，来源运行没换，
-    视图也得跟着遮（P4，同数据源事后加遮罩列）。遮罩字段是 JSON 指针，按能力登记的表格映射认它落在哪几列
-    （masked_columns_of，落快照时记 mask_columns 用的同一个函数）。能力删掉了的只能按快照记下的遮。"""
-    from app.capabilities.endpoints import masked_columns_of
-    from app.capabilities.registry import get_capability, masked_paths
-    from app.db.base import SessionLocal
-
-    async with SessionLocal() as session:
-        cap = await get_capability(session, name)
-    if cap is None:
-        return set()
-    return {c.lower() for c in masked_columns_of(cap.tabular or {}, masked_paths(cap))}
-
-
 async def _cell(ctx: _Page, source: Source, cid: str, ref: Any, conv: Any) -> dict[str, Any]:
    from app.api.evidence import MASKED
    from app.engine.evidence import MISSING, RenderError, cell_eid, render_cell
@@ -80,10 +80,13 @@ async def version_row(session: AsyncSession, workflow_id: str, version: int) ->
 # 子工作流固定的版本（D-q）
 # --------------------------------------------------------------------------

-#: 子工作流节点固定的版本不能用的机器码。发起正式运行（engine/launch.py）、发布工作流与发布前检查（engine/lint_process.py
-#: 的 lint_subgraph_pins）、应用检查（apps/lint.py）都按 subgraph_pin_problems 判，机器码相同
+#: 固定的上游版本不能用的机器码：子工作流节点（workflow_version）、口径卡转抄别处的口径（caliber_from）。发起正式运行
+#: （engine/launch.py）、发布工作流与发布前检查（engine/lint_process.py 的 lint_subgraph_pins）、应用检查（apps/lint.py）
+#: 都按 subgraph_pin_problems 判，机器码相同
 SUBGRAPH_UNPUBLISHED = "subgraph_version_unpublished"
 SUBGRAPH_RETIRED = "subgraph_version_retired"
+CALIBER_UNPUBLISHED = "caliber_version_unpublished"
+CALIBER_RETIRED = "caliber_version_retired"


@dataclass(frozen=True)
@@ -99,40 +102,84 @@ class SubgraphPinProblem:
    reason: str
    #: 怎么改
    advice: str
+    #: 句子的主语：「子工作流「…」」或「口径卡「…」」
+    who: str = ""
+    #: 节点上哪个配置项固定了版本（问题定位到它）
+    field: str = "workflow_version"
+
+
+def pinned_upstream(node: Any) -> tuple[str, str, int] | None:
+    """节点固定的上游版本：(种类 subgraph / caliber, 工作流 id, 版本)；没固定就是 None。子工作流写在 workflow_id /
+    workflow_version，口径卡转抄别处的口径写在 caliber_from 里。"""
+    from app.engine.schema import NodeType
+
+    cfg = node.config
+    if node.type == NodeType.SUBGRAPH:
+        wid, pinned = cfg.get("workflow_id"), cfg.get("workflow_version")
+        kind = "subgraph"
+    elif node.type == NodeType.METRICS and isinstance(cfg.get("caliber_from"), dict):
+        wid, pinned = cfg["caliber_from"].get("workflow_id"), cfg["caliber_from"].get("workflow_version")
+        kind = "caliber"
+    else:
+        return None
+    if not wid or not str(pinned or "").isdigit():
+        return None
+    return kind, str(wid), int(pinned)


 async def subgraph_pin_problems(session: AsyncSession, spec: Any) -> list[SubgraphPinProblem]:
-    """图里每个固定了版本的子工作流节点，固定的那一版必须「已发布且未退役」（is_published）；不是的逐个交回。
-
-    子工作流固定版本就是方法卡的机制：外层发布时认可的是「用它的 vN」。vN 只是一张保存出来的草稿快照、或者已经退役，
-    正式运行照样执行它，就等于绕过了发布门禁、绕过了退役。发起时拦（launch.py），发布时也按同一条拦——不然钉在草稿上的
-    工作流照样发得出去，之后每一次正式运行（定时任务、应用入口、按需视图、补算）都在发起时才被拒。
-    没固定版本的子工作流跟随最新，那是受管级门禁管的事（governed.subgraph_unpinned），这里不管。"""
-    from app.engine.schema import NodeType
+    """图里每个固定了上游版本的节点（子工作流、转抄别处口径的口径卡），固定的那一版必须「已发布且未退役」
+    （is_published）；不是的逐个交回。

+    固定版本就是方法卡的机制：外层发布时认可的是「用它的 vN」。vN 只是一张保存出来的草稿快照、或者已经退役，正式运行
+    照样执行它、照样按它的口径算，就等于绕过了发布门禁、绕过了退役。发起时拦（launch.py），发布时也按同一条拦——不然钉在
+    草稿上的工作流照样发得出去，之后每一次正式运行（定时任务、应用入口、按需视图、补算）都在发起时才被拒。
+    没固定版本的跟随最新，那是受管级门禁管的事（governed.subgraph_unpinned），这里不管。"""
    out: list[SubgraphPinProblem] = []
    for node in spec.nodes:
-        if node.type != NodeType.SUBGRAPH:
+        pin = pinned_upstream(node)
+        if pin is None:
            continue
-        wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version")
-        if not wid or not str(pinned or "").isdigit():
-            continue
-        version = int(pinned)
-        workflow = await session.get(Workflow, str(wid))
-        row = await version_row(session, str(wid), version) if workflow is not None else None
+        kind, wid, version = pin
+        caliber = kind == "caliber"
+        who = f"口径卡「{node.title}」" if caliber else f"子工作流「{node.title}」"
+        field = "caliber_from" if caliber else "workflow_version"
+        unpublished, retired = (CALIBER_UNPUBLISHED, CALIBER_RETIRED) if caliber else (SUBGRAPH_UNPUBLISHED, SUBGRAPH_RETIRED)
+        workflow = await session.get(Workflow, wid)
+        row = await version_row(session, wid, version) if workflow is not None else None
        if workflow is None or row is None:
-            out.append(SubgraphPinProblem(node.id, node.title, SUBGRAPH_UNPUBLISHED, str(wid), version,
-                                          f"工作流版本 v{version} 不存在", "请在节点中重新选择版本"))
+            out.append(SubgraphPinProblem(node.id, node.title, unpublished, wid, version,
+                                          f"工作流版本 v{version} 不存在", "请在节点中重新选择版本", who, field))
        elif row.retired_at is not None:
-            out.append(SubgraphPinProblem(node.id, node.title, SUBGRAPH_RETIRED, str(wid), version,
-                                          f"「{workflow.name}」v{version} 已退役", "请把节点改为固定一个在用的发布版本"))
+            out.append(SubgraphPinProblem(node.id, node.title, retired, wid, version,
+                                          f"「{workflow.name}」v{version} 已退役", "请把节点改为固定一个在用的发布版本",
+                                          who, field))
        elif not is_published(row, current=workflow.published_version):
-            out.append(SubgraphPinProblem(node.id, node.title, SUBGRAPH_UNPUBLISHED, str(wid), version,
+            out.append(SubgraphPinProblem(node.id, node.title, unpublished, wid, version,
                                          f"「{workflow.name}」v{version} 没有发布过",
-                                          "请先发布那个工作流，或把节点改为固定一个已发布的版本"))
+                                          "请先发布那个工作流，或把节点改为固定一个已发布的版本", who, field))
    return out


+async def backfill_legacy_levels(session: AsyncSession) -> int:
+    """给记级别之前发布过的老版本补上 level（启动时跑一次，幂等）。
+
+    level 是后来才加的列：那之前发布过的版本都是空的，按 is_published 只认当前发布版——而子工作流、口径卡钉住的往往是
+    更早的某一版，新规则会把它们误判成「没发布过」。补的依据只有一条，有据可查：这一版发起过正式运行。老代码里正式运行
+    只能从当时的发布版发起，所以它一定发布过。级别记 published：当时是不是受管级查不出来，往低处记（应用的级别取最低，
+    只会少给信任，不会多给）。没有正式运行凭据的照旧为空，不猜。"""
+    from sqlalchemy import exists, update
+
+    from app.db.models import Run
+
+    evidence = exists().where(Run.workflow_id == WorkflowVersion.workflow_id, Run.version == WorkflowVersion.version,
+                              Run.run_class == "formal")
+    result = await session.execute(update(WorkflowVersion)
+                                   .where(WorkflowVersion.level.is_(None), WorkflowVersion.retired_at.is_(None), evidence)
+                                   .values(level="published").execution_options(synchronize_session=False))
+    return result.rowcount or 0
+
+
 # --------------------------------------------------------------------------
 # 账本（L4）：一个工作流版本被谁固定、生效了几次、正式运行了几次。退役确认、版本详情、删除约束、账本接口共用
 # --------------------------------------------------------------------------
@@ -305,6 +305,24 @@ async def endpoint_request(session: AsyncSession, capability: Any, args: dict[st
                     run_id=run_id, node_id=node_id)


+async def snapshot_masked_columns(snap: dict[str, Any]) -> set[str]:
+    """一份端点快照要遮的列（小写）：落快照时记下的（mask_columns），加上能力现在的遮罩字段按表格映射落到的列。
+
+    端点快照的列是能力登记的表格映射排出来的，没有 SQL 改名这回事，按列名遮就是准的。管理员事后给能力加了遮罩字段，
+    来源运行没换，证据面板和视图也得跟着遮（P4，同数据源事后加遮罩列）。能力删掉了的只能按快照记下的遮。
+    证据面板（api/evidence._mask_sets）和应用视图（apps/resolve._snapshot）共用这一份。"""
+    from app.capabilities.endpoints import masked_columns_of
+    from app.db.base import SessionLocal
+
+    recorded = snap.get("mask_columns")
+    out = {c.lower() for c in recorded if isinstance(c, str)} if isinstance(recorded, list) else set()
+    async with SessionLocal() as session:
+        cap = await get_capability(session, str(snap.get("capability") or ""))
+    if cap is not None:
+        out |= {c.lower() for c in masked_columns_of(cap.tabular or {}, masked_paths(cap))}
+    return out
+
+
 def masked_paths(capability: Any) -> list[str]:
    """能力的遮罩字段（JSON 指针）。"""
    return list(getattr(capability, "masked", None) or [])
@@ -252,6 +252,21 @@ _INDEX_MIGRATIONS: list[tuple[str, str, str | tuple[str, ...]]] = [
 ]


+#: 迁移补建的唯一索引：(索引名, 表, 列, 部分索引的条件, 建之前的去重语句)。只在索引还不存在时先去重再建——存量库里可能
+#: 已经有并发留下的重复行，直接建会失败。去重的规矩写在各自的语句里：多出的钩子行删掉（只留最早的那一行，它的密钥照旧
+#: 能用），多出的助手对话归档（不删，历史还在）
+_UNIQUE_INDEX_MIGRATIONS: list[tuple[str, str, tuple[str, ...], str | None, str]] = [
+    ("ux_app_hooks_app_name", "app_hooks", ("app_id", "name"), None,
+     "DELETE FROM app_hooks WHERE id NOT IN (SELECT id FROM (SELECT id, ROW_NUMBER() OVER "
+     "(PARTITION BY app_id, name ORDER BY created_at, id) AS rn FROM app_hooks) WHERE rn = 1)"),
+    ("ux_conversations_app_owner", "conversations", ("app_id", "created_by"), "kind = 'app' AND archived = 0",
+     "UPDATE conversations SET archived = 1 WHERE kind = 'app' AND archived = 0 AND created_by IS NOT NULL "
+     "AND id NOT IN (SELECT id FROM (SELECT id, ROW_NUMBER() OVER (PARTITION BY app_id, created_by "
+     "ORDER BY last_active_at DESC, id) AS rn FROM conversations WHERE kind = 'app' AND archived = 0 "
+     "AND created_by IS NOT NULL) WHERE rn = 1)"),
+]
+
+
 def _migrate(conn) -> None:  # pragma: no cover - 同步回调
    from sqlalchemy import text

@@ -264,6 +279,14 @@ def _migrate(conn) -> None:  # pragma: no cover - 同步回调
        cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))}
        if all(c in cols for c in wanted):
            conn.execute(text(f"CREATE INDEX IF NOT EXISTS {name} ON {table} ({', '.join(wanted)})"))
+    for name, table, columns, where, dedupe in _UNIQUE_INDEX_MIGRATIONS:
+        cols = {row[1] for row in conn.execute(text(f"PRAGMA table_info({table})"))}
+        exists = conn.execute(text("SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = :n"), {"n": name}).first()
+        if exists or not all(c in cols for c in columns):
+            continue
+        conn.execute(text(dedupe))
+        conn.execute(text(f"CREATE UNIQUE INDEX {name} ON {table} ({', '.join(columns)})"
+                          + (f" WHERE {where}" if where else "")))


 async def init_db() -> None:
@@ -803,6 +803,10 @@ class Conversation(Base, TimestampMixin):
    """

    __tablename__ = "conversations"
+    # 一个人在一个应用里只有一段（未归档的）助手对话（D-e）。存量库由 db/base.py 的 _UNIQUE_INDEX_MIGRATIONS 补建。
+    # created_by 为空（关登录又没填署名）的不受约束：SQLite 的唯一索引里 NULL 互不相等，那种情况靠接口里的进程内锁
+    __table_args__ = (Index("ux_conversations_app_owner", "app_id", "created_by", unique=True,
+                            sqlite_where=text("kind = 'app' AND archived = 0")),)

    id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
    title: Mapped[str] = mapped_column(String(200), default="")
@@ -9,7 +9,7 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Any

-from sqlalchemy import ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy import ForeignKey, Index, Integer, String, Text, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column

 from app.db.base import Base, TimestampMixin, UTCDateTime, new_id, utcnow
@@ -86,6 +86,8 @@ class AppHook(Base, TimestampMixin):
    """入站钩子。start：按入口发起实例；signal：按 correlation 解决一条「等外部系统」的待办。"""

    __tablename__ = "app_hooks"
+    # 一个名字一行（D-g）：并发「生成密钥」不会多出第二把能用的密钥。存量库由 db/base.py 的 _UNIQUE_INDEX_MIGRATIONS 补建
+    __table_args__ = (Index("ux_app_hooks_app_name", "app_id", "name", unique=True),)

    id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
    app_id: Mapped[str] = mapped_column(ForeignKey("apps.id", ondelete="CASCADE"), index=True)
@@ -421,7 +421,7 @@ async def _refuse_unpublished_subgraphs(session: AsyncSession, spec: GraphSpec)
    """正式运行里，每个固定了版本的子工作流节点，固定的那一版必须「已发布且未退役」（D-q）。规则在
    apps/versions.subgraph_pin_problems，发布门禁按同一条拦；这里是发起这一刻的那一道（发布之后那一版才退役的）。"""
    for problem in await subgraph_pin_problems(session, spec):
-        raise LaunchRefused(409, f"子工作流「{problem.title}」固定的{problem.reason}，不能发起正式运行。"
+        raise LaunchRefused(409, f"{problem.who}固定的{problem.reason}，不能发起正式运行。"
                                 f"{problem.advice}，再发布这个工作流", problem.code)


@@ -264,16 +264,18 @@ def lint_process(spec: GraphSpec, flag: Callable[..., None], *, level: str, stri
 def lint_subgraph_pins(spec: GraphSpec, flag: Callable[..., None], *, level: str, strict: bool) -> None:
    """子工作流固定的版本不是「已发布且未退役」的（D-q）：各级都报错（always），机器码和发起时拒绝的一样。
    load_subgraphs 没读好、没放进来时不查（门禁本身不查库）。"""
+    from app.apps.versions import pinned_upstream
+
    graphs = _SUBGRAPHS.get()
    nodes = spec.node_map()
    for problem in getattr(graphs, "pin_problems", None) or []:
        node = nodes.get(problem.node_id)
        # 读好的结果对的是读的那张图；拿去查改过的图（自动修复的预览）时，节点改了固定的版本就不再算它
-        if node is None or str(node.config.get("workflow_id") or "") != problem.workflow_id \
-                or str(node.config.get("workflow_version") or "") != str(problem.version):
+        if node is None or pinned_upstream(node) is None \
+                or pinned_upstream(node)[1:] != (problem.workflow_id, problem.version):
            continue
-        flag(f"子工作流「{problem.title}」固定的{problem.reason}：正式运行发起时会被拒绝。{problem.advice}",
-             code=problem.code, node_id=problem.node_id, field="workflow_version", always=True)
+        flag(f"{problem.who}固定的{problem.reason}：正式运行发起时会被拒绝。{problem.advice}",
+             code=problem.code, node_id=problem.node_id, field=problem.field, always=True)


 def _register() -> None:
@@ -42,6 +42,15 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    settings.ensure_dirs()
    await init_db()
    await seed_defaults()
+
+    # 记级别之前发布过的老版本：有正式运行凭据的补记为已发布，免得子工作流、口径卡钉住它们的正式运行被误判成「没发布过」
+    from app.apps.versions import backfill_legacy_levels
+    from app.db.base import SessionLocal as _Sessions
+
+    async with _Sessions() as _s:
+        if levelled := await backfill_legacy_levels(_s):
+            logger.info("给 %d 个老的发布版本补记了级别（有正式运行凭据）", levelled)
+        await _s.commit()
    await _platform_startup()

    # 把存的 embedding 配置读进来。不读的话重启后会悄悄退回本地哈希向量，
@@ -340,8 +340,8 @@ async def rotate_hook(session: AsyncSession, app_id: str, name: str, *, kind: st
    """生成或轮换这个名字的密钥，返回 (行, 密钥明文, 是不是新生成的)。密钥只在这一次交回，库里存密文。

    kind / entry 只是照声明记在行上给人看；收请求时以当前发布版的声明为准（published_declaration）。调用方提交。
-    新插入一行之后再查一遍：别人抢先插了（并发生成，hook_lock 管不到的另一个进程），删掉自己这行、改去轮换最早那一行，
-    交回的仍是这一次的密钥——和两个人先后点「轮换」一样，后点的那把生效。"""
+    (app_id, name) 有唯一索引（ux_app_hooks_app_name）：别人抢先插了这个名字（并发生成，hook_lock 管不到的另一个进程），
+    这次插入在保存点里被库拒掉，改去轮换已有的那一行，交回的仍是这一次的密钥——和两个人先后点「轮换」一样，后点的那把生效。"""
    from app.core.crypto import encrypt

    if kind not in KINDS:
@@ -350,13 +350,16 @@ async def rotate_hook(session: AsyncSession, app_id: str, name: str, *, kind: st
    row = await hook_row(session, app_id, name)
    created = row is None
    if row is None:
-        row = AppHook(app_id=app_id, name=name[:100], kind=kind, enabled=True, secret="")
-        session.add(row)
-        await session.flush()
-        first = await hook_row(session, app_id, name)
-        if first is not None and first.id != row.id:
-            await session.delete(row)
-            row, created = first, False
+        fresh = AppHook(app_id=app_id, name=name[:100], kind=kind, enabled=True, secret="")
+        try:
+            async with session.begin_nested():
+                session.add(fresh)
+                await session.flush()
+            row = fresh
+        except IntegrityError:
+            row, created = await hook_row(session, app_id, name), False
+            if row is None:
+                raise
    row.kind, row.entry_id = kind, (entry if kind == "start" else None)
    row.secret = encrypt(secret) or ""
    await session.flush()
@@ -825,15 +825,22 @@ async def list_tasks(principal: Any, *, assigned: str = "me", statuses: list[str
        return await _describe(session, tasks, principal)


-async def count(principal: Any) -> dict[str, int]:
-    """导航徽标：派给我的、待处理的待办数。apps 是应用里的，other 是不属于应用的（画布、定时任务的审批）。"""
+async def count(principal: Any) -> dict[str, Any]:
+    """导航徽标：派给我的、待处理的待办数。apps 是应用里的，other 是不属于应用的（画布、定时任务的审批）。
+
+    latest 是其中最新一条的开出时间：前端的新待办通知看它变没变，而不只看数——同一拍里一条处理掉、一条新来，数不变，
+    新的那条照样要通知。"""
    async with SessionLocal() as session:
-        stmt = select(Task.app_id.is_not(None), func.count()).where(Task.status == "pending")
+        stmt = select(Task.app_id.is_not(None), func.count(), func.max(Task.created_at)).where(Task.status == "pending")
        if (clause := _mine(principal)) is not None:
            stmt = stmt.where(clause)
-        rows = dict((await session.execute(stmt.group_by(Task.app_id.is_not(None)))).all())
-    apps, other = int(rows.get(True, 0)), int(rows.get(False, 0))
-    return {"mine": apps + other, "apps": apps, "other": other}
+        rows = {has_app: (n, latest) for has_app, n, latest in
+                (await session.execute(stmt.group_by(Task.app_id.is_not(None)))).all()}
+    apps, other = int(rows.get(True, (0, None))[0]), int(rows.get(False, (0, None))[0])
+    stamps = [latest for _, latest in rows.values() if latest is not None]
+    latest = max(stamps) if stamps else None
+    return {"mine": apps + other, "apps": apps, "other": other,
+            "latest": latest.isoformat() if hasattr(latest, "isoformat") else latest}


 async def detail(task_id: str, principal: Any) -> dict[str, Any] | None:
@@ -334,3 +334,52 @@ async def test_a_workflow_in_use_cannot_be_deleted(client):
    assert res.status_code == 409 and "v1 有 1 条生效记录" in res.json()["detail"]
    plain = await workflow({1: "published"}, current=1)
    assert (await client.delete(f"/api/workflows/{plain}")).status_code == 204
+
+
+async def test_caliber_pins_follow_the_same_rule_as_subgraphs():
+    """D-q 延伸到口径卡：转抄别处口径的口径卡固定的那一版，同样必须已发布且未退役；机器码各用一套。"""
+    from types import SimpleNamespace
+
+    from app.apps.versions import CALIBER_UNPUBLISHED, subgraph_pin_problems
+    from app.db.base import SessionLocal
+    from app.db.models import Workflow, WorkflowVersion
+    from app.engine.schema import GraphSpec
+
+    async with SessionLocal() as session:
+        wf = Workflow(name="周销售口径", graph={}, published_version=2)
+        session.add(wf)
+        await session.flush()
+        session.add_all([WorkflowVersion(workflow_id=wf.id, version=1, graph={}),
+                         WorkflowVersion(workflow_id=wf.id, version=2, graph={}, level="published")])
+        await session.commit()
+        graph = GraphSpec.model_validate({"nodes": [
+            {"id": "in", "type": "input", "data": {"label": "入口", "config": {}}},
+            {"id": "kpi", "type": "metrics", "data": {"label": "指标", "config": {
+                "caliber_from": {"workflow_id": wf.id, "workflow_version": 1, "node_id": "card"}}}},
+        ], "edges": [{"source": "in", "target": "kpi"}]})
+        problems = await subgraph_pin_problems(session, graph)
+        assert [(p.code, p.field) for p in problems] == [(CALIBER_UNPUBLISHED, "caliber_from")]
+        assert problems[0].who == "口径卡「指标」"
+        graph.nodes[1].config["caliber_from"]["workflow_version"] = 2
+        assert await subgraph_pin_problems(session, graph) == []
+
+
+async def test_legacy_versions_with_formal_runs_are_backfilled_as_published():
+    """记级别之前发布过的老版本：有正式运行凭据的补记为 published（幂等），没有凭据的不猜。"""
+    from app.apps.versions import backfill_legacy_levels
+    from app.db.base import SessionLocal
+    from app.db.models import Run, Workflow, WorkflowVersion
+
+    async with SessionLocal() as session:
+        wf = Workflow(name="老方法卡", graph={}, published_version=3)
+        session.add(wf)
+        await session.flush()
+        session.add_all([WorkflowVersion(workflow_id=wf.id, version=v, graph={}) for v in (1, 2, 3)])
+        session.add(Run(workflow_id=wf.id, run_class="formal", version=1, graph={}, input={}))
+        await session.commit()
+        assert await backfill_legacy_levels(session) >= 1
+        await session.commit()
+        levels = {v.version: v.level for v in (await session.execute(
+            WorkflowVersion.__table__.select().where(WorkflowVersion.workflow_id == wf.id))).all()}
+        assert levels == {1: "published", 2: None, 3: None}
+        assert await backfill_legacy_levels(session) == 0
@@ -67,7 +67,9 @@ async def publish(graph, *, name="口径库", versions=1, status="published", la
        session.add(wf)
        await session.flush()
        for v in range(1, versions + 1):
-            session.add(WorkflowVersion(workflow_id=wf.id, version=v, graph=later if later and v > 1 else graph))
+            # 每一版都发布过（level 记着）：口径卡固定的版本须「已发布且未退役」（D-q）
+            session.add(WorkflowVersion(workflow_id=wf.id, version=v, graph=later if later and v > 1 else graph,
+                                        level="published"))
        await session.commit()
        return wf.id

@@ -266,3 +266,14 @@ def test_run_compare_reads_endpoint_requests():
    assert _compare_request(q("erp", "/orders", {"d": "1"}), q("erp", "/orders", {"d": "1"})).structure == "same"
    assert _compare_request(q("erp", "/orders", {"d": "1"}), q("erp", "/orders", {"d": "2"})).structure == "params_only"
    assert _compare_request(q("erp", "/orders", {}), q("erp", "/refunds", {})).structure == "changed"
+
+
+async def test_evidence_panel_masks_an_endpoint_snapshot_by_column():
+    """证据面板认端点快照：按列名遮（落快照时记下的加能力现在的遮罩字段），不按数据源找、不把整张表当成「数据源不在了」
+    遮掉。能力登记里查不到的只按快照记下的遮。"""
+    from app.api.evidence import _mask_sets, _Masks
+
+    snap = {"capability": "erp_orders", "source": "erp", "columns": ["单号", "电话", "金额"],
+            "rows": [["A1", "13800000000", 12]], "mask_columns": ["电话"]}
+    sets = await _mask_sets(snap, "erp", _Masks())
+    assert sets.hidden == {"电话"} and sets.sure == {"电话"} and not sets.gone
@@ -252,8 +252,8 @@ async def test_hook_management_functions():


 async def test_one_name_one_row_even_under_races(client, monkeypatch):
-    """库里还没有 (app_id, name) 的唯一约束：并发生成只留一行；插入后发现别人抢先插了，删掉自己、改去轮换那一行；万一
-    多出一行，收请求只认最早那一行，停用按名字改全部行。"""
+    """(app_id, name) 有唯一索引：并发生成只留一行；插入时发现别人抢先插了（库拒掉），改去轮换那一行；库里不可能
+    多出第二行，停用按名字改。"""
    from app.db.base import utcnow
    from app.db.models import AppHook

@@ -269,7 +269,7 @@ async def test_one_name_one_row_even_under_races(client, monkeypatch):
        rows = list((await session.execute(select(AppHook).where(AppHook.app_id == app_id))).scalars())
    assert len(rows) == 1

-    # 另一个进程在「读不到行」和「插入」之间抢先插好了：这一次插入后再查，删掉自己、轮换最早那一行
+    # 另一个进程在「读不到行」和「插入」之间抢先插好了：这一次插入被唯一索引拒掉，改去轮换已有的那一行
    real = hooks.hook_row
    calls = {"n": 0}

@@ -286,21 +286,17 @@ async def test_one_name_one_row_even_under_races(client, monkeypatch):
    async with SessionLocal() as session:
        assert len(list((await session.execute(select(AppHook).where(AppHook.app_id == app_id))).scalars())) == 1

-    # 真的多出一行（多进程各自插入）：它的密钥收不到请求；停用两行一起停
-    from app.core.crypto import encrypt
+    # 库不许多出第二行：多进程各自插入时，后插的那一次被唯一索引拒掉
+    from sqlalchemy.exc import IntegrityError

    async with SessionLocal() as session:
-        stray = AppHook(app_id=app_id, name="paid", kind="signal", enabled=True, secret=encrypt("stray-secret") or "",
-                        created_at=utcnow())
-        session.add(stray)
-        await session.commit()
-        stray_id = stray.id
-    headers, body = signed("stray-secret", {"correlation": "x"}, key="k-stray")
-    assert (await client.post(f"/api/hooks/{stray_id}", content=body, headers=headers)).status_code == 404
+        session.add(AppHook(app_id=app_id, name="paid", kind="signal", enabled=True, secret="x", created_at=utcnow()))
+        with pytest.raises(IntegrityError):
+            await session.commit()
    assert (await client.patch(f"/api/apps/{app_id}/hooks/paid", json={"enabled": False})).status_code == 200
    async with SessionLocal() as session:
        states = [r.enabled for r in (await session.execute(select(AppHook).where(AppHook.app_id == app_id))).scalars()]
-    assert states == [False, False]
+    assert states == [False]


 async def test_a_hook_works_by_what_the_published_version_declares(client, monkeypatch):
@@ -443,7 +443,7 @@ async def test_count_and_list_api(client):
    await _wait(run.id, ("interrupted",))
    task = await _task(run.id)
    count = (await client.get("/api/tasks/count?mine=1")).json()
-    assert set(count) == {"mine", "apps", "other"} and count["other"] >= 1
+    assert set(count) == {"mine", "apps", "other", "latest"} and count["other"] >= 1 and count["latest"]
    listed = (await client.get(f"/api/tasks?assigned=me&run_id={run.id}")).json()["items"]
    assert [t["id"] for t in listed] == [task.id]
    item = listed[0]
@@ -140,8 +140,9 @@ async def upstream_with_two_versions() -> str:
        lib = Workflow(name=f"口径库-{_tag()}", graph=library(), status="published", published_version=2, version=2)
        session.add(lib)
        await session.flush()
-        session.add(WorkflowVersion(workflow_id=lib.id, version=1, graph=library()))
-        session.add(WorkflowVersion(workflow_id=lib.id, version=2, graph=library(version="v4")))
+        # 两版都发布过（level 记着）：口径卡固定的版本须「已发布且未退役」（D-q）
+        session.add(WorkflowVersion(workflow_id=lib.id, version=1, graph=library(), level="published"))
+        session.add(WorkflowVersion(workflow_id=lib.id, version=2, graph=library(version="v4"), level="published"))
        await session.commit()
        return lib.id

@@ -1155,7 +1155,7 @@ function watchApprovals(go: Go): () => void {

 /**
 * 派给自己的新待办：规则同待审批——拿到过一次真列表才记底，之后新来的、标签页在后台时发系统通知。不每拍都取列表：
- * 心跳取的「派给我的」待办数（taskCounts.mine）变了才取一次派给我的待处理待办。搭建者以上已经按待审批发过通知的
+ * 心跳取的「派给我的」待办数或最新一条的时间（taskCounts.mine / latest）变了才取一次派给我的待处理待办。搭建者以上已经按待审批发过通知的
 * （待办带着那条审批）不再发第二条；核实、等外部系统这类没有审批的待办只在这里通知
 */
 function watchTasks(go: Go): () => void {
@@ -1192,7 +1192,7 @@ function watchTasks(go: Go): () => void {
  }
  return useCatalog.subscribe((s, p) => {
    if (s.taskCounts === p.taskCounts || !s.taskCounts) return
-    if (seen && s.taskCounts.mine === p.taskCounts?.mine) return
+    if (seen && s.taskCounts.mine === p.taskCounts?.mine && s.taskCounts.latest === p.taskCounts?.latest) return
    if (needsLogin(useAuth.getState().me)) return
    void pull()
  })
@@ -5,8 +5,8 @@
 * 入口提交校验失败回 422，ApiError 的 body 里带 {errors: FormError[]}。
 */
 import type {
-  AppBindings, AppDetail, AppGrant, AppHookRow, AppSpec, AppSummary, AppVersionRow, ComponentCatalog, Disposition,
-  InstanceRow, LintIssue, PublishResult, ViewModel,
+  AppBindings, AppDetail, AppGrant, AppHookRow, AppSpec, AppSummary, AppVersionRow, BackfillPlan, BackfillResult,
+  ComponentCatalog, Disposition, InstanceRow, LintIssue, PublishResult, ViewModel,
 } from '../typesApp'
 import type { ConversationDetail, EvidenceSegmentDetail } from '../types'
 import { del, get, patch, post, put, qs, type RequestOptions } from './http'
@@ -14,85 +14,17 @@ import { del, get, patch, post, put, qs, type RequestOptions } from './http'
 const at = (id: string) => `/apps/${encodeURIComponent(id)}`
 const hookAt = (id: string, name: string) => `${at(id)}/hooks/${encodeURIComponent(name)}`

-/**
- * 钩子管理里的一行（D-g）：草稿里声明的 ∪ 已经生成过密钥的，按名字合成一行。
- * - declared：草稿的 spec.hooks 里有这个名字；published：当前发布版里有（入站请求按发布版认种类和入口，没有就拒收）；
- * - has_secret：已经生成过密钥（app_hooks 里有这一行）。密钥本身只在生成、轮换时交回一次（AppHookRow.secret）。
- * 服务端几种写法都认（hookStateOf），缺的项按「没有」算
- */
-export interface AppHookState {
-  name: string
-  kind: 'start' | 'signal' | null
-  entry_id: string | null
-  declared: boolean
-  published: boolean
-  has_secret: boolean
-  enabled: boolean
-  /** app_hooks 那一行的 id：入站地址是 /api/hooks/{id}。还没生成密钥时为 null */
-  id: string | null
-  /** 收请求的地址（服务端给的路径）；没给时按 id 拼 */
-  url_path: string | null
-  created_at: string | null
-}
-
-function hookStateOf(raw: Record<string, any>): AppHookState {
-  const id = typeof raw.id === 'string' && raw.id ? raw.id : null
-  const flag = (...keys: string[]) => keys.some((k) => raw[k] === true)
-  return {
-    name: String(raw.name ?? ''),
-    kind: raw.kind === 'start' || raw.kind === 'signal' ? raw.kind : null,
-    entry_id: raw.entry_id ?? raw.entry ?? null,
-    declared: flag('declared', 'in_draft'),
-    published: flag('published', 'in_published'),
-    has_secret: raw.has_secret != null || raw.secret_set != null ? flag('has_secret', 'secret_set') : !!id,
-    enabled: raw.enabled !== false,
-    id,
-    url_path: typeof raw.url_path === 'string' && raw.url_path ? raw.url_path : id ? `/api/hooks/${encodeURIComponent(id)}` : null,
-    created_at: raw.created_at ?? null,
-  }
-}
-
-/** 按新口径补算的结果（D-f）：发起了几次、跳过了几次（同一输入已经补算过） */
-export interface BackfillResult {
-  started: number
-  skipped: number
-  run_ids?: string[]
-  /** 发起到一半被拒（比如版本刚被退役）：已经发起的照算，原因照实交回 */
-  refused?: string | null
-}
-
-/** 补算会发起几次（服务端 apps/backfill.Plan.out，不发起运行）：这个视图、哪个固定版本从几升到几、趋势画几期 */
-export interface BackfillPlan {
-  view: string
-  alias: string
-  source: string
-  from_version: number
-  to_version: number
-  periods: number
-  will_launch: number
-  will_skip: number
-}
-
-/**
- * 应用详情。manage 才有草稿和岗位绑定（D-m）；另带能按新口径补算的视图（backfill，只在取详情时算）。use / observe
- * 只有概况和当前发布版的页面、入口目录（published）
- */
-export type AppDetailOut = AppDetail & {
-  backfill?: BackfillPlan[]
-  published?: { version: number; pages: { id: string; title: string }[]; entries: { id: string; label: string }[] } | null
-}
-
 export const apps = {
  catalog: (opts?: RequestOptions) => get<ComponentCatalog>('/apps/catalog', opts),
  list: (opts?: RequestOptions) => get<{ items: AppSummary[] }>('/apps', opts),
-  get: (id: string, opts?: RequestOptions) => get<AppDetailOut>(at(id), opts),
+  get: (id: string, opts?: RequestOptions) => get<AppDetail>(at(id), opts),
  create: (body: { name: string; description?: string; spec?: AppSpec }) => post<AppDetail>('/apps', body),
  /** 从工作流版本确定性派生一份应用草稿（不落库），编辑器拿去预览 / 新建 */
  derive: (body: { workflow_id: string; version: number; name?: string }) =>
    post<{ spec: AppSpec }>('/apps/derive', body),
  /** 改草稿：spec、岗位绑定、名称等，只带改了的字段 */
  update: (id: string, body: { name?: string; description?: string; icon?: string; draft?: AppSpec;
-    draft_bindings?: AppBindings }) => patch<AppDetailOut>(at(id), body),
+    draft_bindings?: AppBindings }) => patch<AppDetail>(at(id), body),
  archive: (id: string) => post<AppDetail>(`${at(id)}/archive`),
  lint: (id: string) => post<{ issues: LintIssue[]; level: string | null }>(`${at(id)}/lint`),
  publish: (id: string, body: { note?: string; dispositions?: Record<string, Disposition> }) =>
@@ -129,21 +61,14 @@ export const apps = {
   * 按新口径补算（D-f，manage）：这个视图的固定版本相对上一个应用版本升了版、处置为按新口径重算时，取旧版本在同一来源上
   * 最近几次封存的正式运行的输入，用新版本逐个发起正式运行（来源记为按新口径补算），同一输入补算过的跳过
   */
-  backfill: async (id: string, view: string): Promise<BackfillResult> => {
-    const raw = await post<Record<string, any>>(`${at(id)}/views/${encodeURIComponent(view)}/backfill`)
-    const n = (...keys: string[]) => Number(keys.map((k) => raw?.[k]).find((v) => typeof v === 'number') ?? 0)
-    return { started: n('launched', 'started'), skipped: n('skipped'), run_ids: Array.isArray(raw?.run_ids) ? raw.run_ids : undefined,
-             refused: typeof raw?.refused === 'string' ? raw.refused : null }
-  },
+  backfill: (id: string, view: string) =>
+    post<BackfillResult>(`${at(id)}/views/${encodeURIComponent(view)}/backfill`),
  /** 补算会发起几次、跳过几次（不发起运行）：确认框按它写明次数 */
  backfillPlan: (id: string, view: string) => get<BackfillPlan>(`${at(id)}/views/${encodeURIComponent(view)}/backfill`),
  /** 钩子（D-g）：草稿声明 ∪ 已有的行，按名字 */
-  hooks: async (id: string): Promise<AppHookState[]> => {
-    const raw = await get<{ items: Record<string, any>[] } | Record<string, any>[]>(`${at(id)}/hooks`)
-    return (Array.isArray(raw) ? raw : raw?.items ?? []).map(hookStateOf).filter((h) => h.name)
-  },
+  hooks: async (id: string): Promise<AppHookRow[]> => (await get<{ items: AppHookRow[] }>(`${at(id)}/hooks`)).items,
  /** 生成或轮换签名密钥：secret 只在这一次交回；url_path 是收请求的地址 */
-  rotateHook: (id: string, name: string) => post<AppHookRow & { url_path?: string | null }>(`${hookAt(id, name)}/rotate`),
+  rotateHook: (id: string, name: string) => post<AppHookRow>(`${hookAt(id, name)}/rotate`),
  setHookEnabled: (id: string, name: string, enabled: boolean) => patch<AppHookRow>(hookAt(id, name), { enabled }),
  /** 删掉这个名字的密钥行：外部系统再用旧密钥签的请求一律拒收 */
  deleteHook: (id: string, name: string) => del(hookAt(id, name)),
@@ -3,7 +3,9 @@
 * api.endpoints / api.notify 上。形状见 typesApp.ts。端点和渠道的密钥只回掩码；提交掩码值表示保持原值。
 */
 import type { CatalogReviewAction, CatalogSource, CatalogStatus } from '../types'
-import type { CapabilityEffect, CapabilityRow, EndpointRow, NotifyChannelRow } from '../typesApp'
+import type {
+  CapabilityEffect, CapabilityReviewOutcome, CapabilityReviewRow, CapabilityRow, EndpointRow, NotifyChannelRow,
+} from '../typesApp'
 import { del, get, patch, post, put, qs, type RequestOptions } from './http'

 /**
@@ -45,8 +47,6 @@ export interface CapabilityReviewResult {
  code?: string | null
 }

-const str = (v: unknown): string | null => (typeof v === 'string' && v ? v : null)
-const num = (...vs: unknown[]): number => Number(vs.find((v) => typeof v === 'number') ?? 0)

 /** 队列一项的键：字段@位置。字段只有几个固定的英文名，不含 @，拆的时候按第一个 @ 拆 */
 export const reviewPathOf = (field: string, pointer: string) => `${field}@${pointer}`
@@ -55,28 +55,24 @@ export function splitReviewPath(path: string): { field: string; pointer: string
  return at < 0 ? { field: '', pointer: path } : { field: path.slice(0, at), pointer: path.slice(at + 1) }
 }

-function reviewItemOf(raw: Record<string, any>): CapabilityReviewItem {
-  const pointer = String(raw.pointer ?? '')
-  const field = String(raw.field ?? '')
-  const short = String(raw.field_text ?? raw.field ?? '')
+function reviewItemOf(raw: CapabilityReviewRow): CapabilityReviewItem {
  return {
-    name: String(raw.capability ?? raw.name ?? ''),
-    title: str(raw.title),
-    path: reviewPathOf(field, pointer),
-    pointer,
-    field,
-    // 服务端写好的位置说法；老写法只给了字段名时自己拼
-    where: str(raw.field_label) ?? (pointer ? `${pointer} 的${short}` : short),
+    name: raw.capability,
+    title: raw.title,
+    path: reviewPathOf(raw.field, raw.pointer),
+    pointer: raw.pointer,
+    field: raw.field,
+    where: raw.field_label,
    value: raw.value,
-    display: typeof raw.display === 'string' ? raw.display : raw.value == null ? '' : String(raw.value),
-    source: raw.source,
-    status: raw.status ?? 'proposed',
-    note: str(raw.note),
-    priority: num(raw.priority),
-    reasons: Array.isArray(raw.reasons) ? raw.reasons.map(String) : [],
-    usage: num(raw.runs, raw.usage, raw.run_count),
-    effect: raw.effect ?? null,
-    version: num(raw.version),
+    display: raw.display,
+    source: raw.source as CatalogSource,
+    status: raw.status as CatalogStatus,
+    note: raw.note,
+    priority: raw.priority,
+    reasons: raw.reasons,
+    usage: raw.runs,
+    effect: raw.effect,
+    version: raw.version,
  }
 }

@@ -91,25 +87,22 @@ export const capabilities = {
  sync: () => post<{ changed: number }>('/capabilities/sync'),
  /** 待审阅的能力字段（D-i）：顺序服务端排好；capability 只看这一个能力 */
  reviewQueue: async (params: { limit?: number; offset?: number; capability?: string | null }, opts?: RequestOptions) => {
-    const raw = await get<{ total?: number; items?: Record<string, any>[] }>(`/capabilities/review-queue${qs({
+    const raw = await get<{ total: number; items: CapabilityReviewRow[] }>(`/capabilities/review-queue${qs({
      limit: params.limit, offset: params.offset, capability: params.capability || undefined,
    })}`, opts)
-    const items = (raw?.items ?? []).map(reviewItemOf).filter((i) => i.name && i.field)
-    return { total: typeof raw?.total === 'number' ? raw.total : items.length, items }
+    return { total: raw.total, items: raw.items.map(reviewItemOf) }
  },
  /**
   * 成批审阅（只给管理员）：逐个能力按治理版本做乐观锁，逐条交回结果（一个能力冲突只记在那几条上，code 为
   * capability_conflict）；写审计。path 是队列一项的键（字段@位置），这里拆回位置和字段
   */
  reviewBatch: async (items: { name: string; path: string; action: CatalogReviewAction; if_version: number }[]) => {
-    const raw = await post<{ results?: Record<string, any>[] }>('/capabilities/review/batch', {
+    const raw = await post<{ results: CapabilityReviewOutcome[] }>('/capabilities/review/batch', {
      items: items.map((i) => ({ capability: i.name, ...splitReviewPath(i.path), action: i.action, if_version: i.if_version })),
    })
    return {
-      results: (raw?.results ?? []).map((r): CapabilityReviewResult => ({
-        name: String(r.capability ?? r.name ?? ''), path: reviewPathOf(String(r.field ?? ''), String(r.pointer ?? '')),
-        ok: r.ok === true, version: typeof r.version === 'number' ? r.version : null,
-        error: str(r.error) ?? str(r.message), code: str(r.code),
+      results: raw.results.map((r): CapabilityReviewResult => ({
+        name: r.capability, path: reviewPathOf(r.field, r.pointer), ok: r.ok, version: r.version, error: r.error, code: r.code,
      })),
    }
  },
@@ -11,7 +11,8 @@ export const tasks = {
  list: (params?: { assigned?: 'me' | 'all'; status?: string; app_id?: string; run_id?: string; limit?: number },
    opts?: RequestOptions) => get<{ items: Task[] }>(`/tasks${qs(params)}`, opts),
  /** 导航徽标：派给我的、待处理的待办数（按来源分：应用里的 / 不属于应用的） */
-  count: (opts?: RequestOptions) => get<{ mine: number; apps: number; other: number }>('/tasks/count?mine=1', opts),
+  count: (opts?: RequestOptions) =>
+    get<{ mine: number; apps: number; other: number; latest: string | null }>('/tasks/count?mine=1', opts),
  get: (id: string, opts?: RequestOptions) => get<Task>(at(id), opts),
  respond: (id: string, body: TaskRespond) => post<{ task: Task; resolved: boolean }>(`${at(id)}/respond`, body),
  reassign: (id: string, body: { to: { kind: 'user' | 'group'; name: string }[]; note?: string }) =>
@@ -107,6 +107,8 @@ function ProvidersTab() {
  const loaded = useCatalog((s) => s.loaded)
  const [catalog, setCatalog] = useState<any>(null)
  const [editing, setEditing] = useState<Provider | 'new' | null>(null)
+  // 模型接入只有管理员能改（服务端 role_by_method(write=admin)）；开登录的搭建者只看，不出改动的按钮，免得点了才 403
+  const canWrite = atLeast(usePrincipal(), 'admin')

  const loadCatalog = () => api.providers.catalog().then(setCatalog, () => {})
  useEffect(() => { void loadCatalog() }, [])
@@ -144,10 +146,13 @@ function ProvidersTab() {

  return (
    <div>
-      <SectionBar title="模型接入" hint="API Key 加密保存在本地，不会返回给浏览器。状态圆点表示上次连接测试的结果，与是否启用无关。">
-        <button className="btn btn-primary btn-sm" onClick={() => setEditing('new')}>
-          <Plus size={12} aria-hidden /> 添加接入
-        </button>
+      <SectionBar title="模型接入" hint={`API Key 加密保存在本地，不会返回给浏览器。状态圆点表示上次连接测试的结果，与是否启用无关。${
+        canWrite ? '' : '只有管理员能添加、修改和测试模型接入。'}`}>
+        {canWrite && (
+          <button className="btn btn-primary btn-sm" onClick={() => setEditing('new')}>
+            <Plus size={12} aria-hidden /> 添加接入
+          </button>
+        )}
      </SectionBar>

      {!loaded ? (
@@ -158,12 +163,14 @@ function ProvidersTab() {
          title="还没有配置模型"
          source="providers"
          body="接入 Anthropic、OpenAI 或任何 OpenAI 兼容的服务（DeepSeek、通义、本机 Ollama…）后，Agent 才能运行。"
-          action={<button className="btn btn-primary btn-sm" onClick={() => setEditing('new')}><Plus size={12} aria-hidden /> 添加接入</button>}
+          action={canWrite
+            ? <button className="btn btn-primary btn-sm" onClick={() => setEditing('new')}><Plus size={12} aria-hidden /> 添加接入</button>
+            : undefined}
        />
      ) : (
        <div className="space-y-2.5">
          {providers.map((p) => (
-            <ProviderCard key={p.id} provider={p} kindLabel={kindLabel(p.kind)}
+            <ProviderCard key={p.id} provider={p} kindLabel={kindLabel(p.kind)} canWrite={canWrite}
                          onEdit={() => setEditing(p)} onRemove={() => void remove(p)} />
          ))}
        </div>
@@ -189,8 +196,8 @@ function ProvidersTab() {
  )
 }

-function ProviderCard({ provider: p, kindLabel, onEdit, onRemove }: {
-  provider: Provider; kindLabel: string; onEdit: () => void; onRemove: () => void
+function ProviderCard({ provider: p, kindLabel, canWrite, onEdit, onRemove }: {
+  provider: Provider; kindLabel: string; canWrite: boolean; onEdit: () => void; onRemove: () => void
 }) {
  const key = `provider:${p.id}`
  // 后端记着上次测的结果：换了浏览器、清了缓存也还在。本机刚测过的更新就用本机的
@@ -213,14 +220,16 @@ function ProviderCard({ provider: p, kindLabel, onEdit, onRemove }: {
        {!p.enabled && <span className="chip" title={PROVIDER_REF_TEXT.disabledChip}>已停用</span>}
        <span className="flex-1" />
        <HealthPill record={record} checkingSince={checkingSince} />
-        <div className="flex items-center gap-1">
-          <button className="btn btn-sm" disabled={!!checkingSince} onClick={() => void test()}
-                  title={`向 ${p.default_model || '默认模型'} 发送一条测试消息`}>
-            <Plug size={11} aria-hidden /> 测试连接
-          </button>
-          <button className="btn btn-sm btn-ghost" onClick={onEdit}>编辑</button>
-          <DeleteButton label={`删除模型接入 ${p.name}`} onClick={onRemove} />
-        </div>
+        {canWrite && (
+          <div className="flex items-center gap-1">
+            <button className="btn btn-sm" disabled={!!checkingSince} onClick={() => void test()}
+                    title={`向 ${p.default_model || '默认模型'} 发送一条测试消息`}>
+              <Plug size={11} aria-hidden /> 测试连接
+            </button>
+            <button className="btn btn-sm btn-ghost" onClick={onEdit}>编辑</button>
+            <DeleteButton label={`删除模型接入 ${p.name}`} onClick={onRemove} />
+          </div>
+        )}
      </div>
      <div className="mt-1.5 flex flex-wrap gap-x-3 gap-y-1 text-2xs text-faint">
        {p.base_url && <span className="mono break-all">{p.base_url}</span>}
@@ -10,7 +10,7 @@
 import { useEffect, useState } from 'react'
 import { History } from 'lucide-react'
 import { api } from '../../api/client'
-import type { BackfillPlan } from '../../api/apps'
+import type { BackfillPlan } from '../../typesApp'
 import { Spinner, confirmDialog, toast } from '../../components/ui'
 import { APP_EDITOR_TEXT as T } from '../../lib/terms'
 import { useAtLeast } from '../../store/auth'
@@ -121,7 +121,7 @@ export function BackfillButton({ appId, target, onDone }: { appId: string; targe
    setBusy(true)
    try {
      const res = await api.apps.backfill(appId, target.view)
-      toast.ok(T.backfillDone(res.started, res.skipped))
+      toast.ok(T.backfillDone(res.launched, res.skipped))
      if (res.refused) toast.warn(res.refused)
      onDone?.()
    } catch (e) {
@@ -1,7 +1,7 @@
 import { useCallback, useEffect, useState } from 'react'
 import { KeyRound, Plus } from 'lucide-react'
 import { api } from '../../../api/client'
-import type { AppHookState } from '../../../api/apps'
+import type { AppHookRow } from '../../../typesApp'
 import {
  CopyButton, DeleteButton, EmptyState, ErrorState, Field, SectionBar, Skeleton, Spinner, confirmDialog, toast,
 } from '../../../components/ui'
@@ -133,7 +133,7 @@ function Grants() {
 */
 export function HooksSection() {
  const { appId, draft, setDraft, dirty, save } = useEditor()
-  const [rows, setRows] = useState<AppHookState[] | null>(null)
+  const [rows, setRows] = useState<AppHookRow[] | null>(null)
  const [loadError, setLoadError] = useState<unknown>(null)
  const [busy, setBusy] = useState<string | null>(null)
  const [shown, setShown] = useState<{ name: string; secret: string; path: string | null } | null>(null)
@@ -261,7 +261,7 @@ export function HooksSection() {
 /** 一行钩子的服务端状态：发布版里有没有、生成过密钥没有、启用；生成 / 轮换、删除密钥 */
 function HookStatus({ name, state, declared, busy, loaded, onRotate, onToggle, onRemove }: {
  name: string
-  state: AppHookState | undefined
+  state: AppHookRow | undefined
  declared: boolean
  busy: boolean
  loaded: boolean
@@ -273,13 +273,13 @@ function HookStatus({ name, state, declared, busy, loaded, onRotate, onToggle, o
  const has = !!state?.has_secret
  return (
    <div className="flex flex-wrap items-center gap-x-2 gap-y-1 text-2xs" data-app-hook-state={has ? 'secret' : 'none'}
-         data-app-hook-published={state?.published ? '1' : '0'}>
+         data-app-hook-published={state?.in_published ? '1' : '0'}>
      <span className="chip" style={has ? { color: 'var(--ok)', borderColor: 'var(--ok)' } : undefined}>
        {has ? T.hookSecretSet : T.hookSecretNone}
      </span>
      {!declared
        ? <span style={{ color: 'var(--st-waiting)' }}>{T.hookNotDeclared}</span>
-        : state?.published ? <span className="text-faint">{T.hookPublished}</span>
+        : state?.in_published ? <span className="text-faint">{T.hookPublished}</span>
        : <span style={{ color: 'var(--st-waiting)' }}>{T.hookNotPublished}</span>}
      <span className="flex-1" />
      {has && (
@@ -20,8 +20,7 @@ import { matchShortcut } from '../../../lib/keys'
 import { useLeaveGuard } from '../../../lib/leave'
 import { APP_EDITOR_TEXT as T } from '../../../lib/terms'
 import { useCatalog } from '../../../store/catalog'
-import type { AppDetailOut } from '../../../api/apps'
-import type { AppBindings, AppSpec, ComponentCatalog } from '../../../typesApp'
+import type { AppBindings, AppDetail, AppSpec, ComponentCatalog } from '../../../typesApp'
 import { EditorContext, type EditorState } from './context'
 import { PinsSection, PublishedVersionPicker } from './PinsSection'
 import { PagesSection } from './PagesSection'
@@ -117,7 +116,7 @@ function EditApp({ appId }: { appId: string }) {
  const { section: raw } = useParams()
  const navigate = useNavigate()
  const section: Section = (SECTIONS as readonly string[]).includes(raw ?? '') ? raw as Section : 'pages'
-  const [detail, setDetail] = useState<AppDetailOut | null>(null)
+  const [detail, setDetail] = useState<AppDetail | null>(null)
  const [error, setError] = useState<unknown>(null)
  const [draft, setDraftState] = useState<AppSpec | null>(null)
  const [bindings, setBindingsState] = useState<AppBindings>({ slots: {} })
@@ -136,7 +135,7 @@ function EditApp({ appId }: { appId: string }) {
    try {
      const d = await api.apps.get(appId)
      setDetail(d)
-      setDraftState(d.draft)
+      setDraftState(d.draft ?? null)
      setBindingsState(d.draft_bindings ?? { slots: {} })
      setDirty(false)
      setError(null)
@@ -1,12 +1,12 @@
 import { createContext, useContext } from 'react'
-import type { AppDetailOut } from '../../../api/apps'
+import type { AppDetail } from '../../../typesApp'
 import type { AppBindings, AppSpec, ComponentCatalog } from '../../../typesApp'

 /** 应用编辑器各节共用的状态：草稿、岗位绑定、组件目录、保存、选中的组件、切到哪一节 */
 export interface EditorState {
  appId: string
  /** 应用详情（manage）：带着草稿、岗位绑定和能按新口径补算的视图 */
-  detail: AppDetailOut
+  detail: AppDetail
  draft: AppSpec
  /** 改草稿：传函数拿到最新的那一份（助手的操作和人的编辑可能挨着到） */
  setDraft: (fn: (spec: AppSpec) => AppSpec) => void
@@ -45,6 +45,8 @@ export interface TaskCounts {
  mine: number
  apps: number
  other: number
+  /** 其中最新一条的开出时间：数不变、却换了一条（一条处理掉、一条新来）时，新待办通知看它 */
+  latest?: string | null
 }

 /** 心跳间隔：连着时复用 4 秒一次的待审批轮询，不另起请求 */
@@ -213,9 +213,36 @@ export interface AppSummary {
 }

 export interface AppDetail extends AppSummary {
-  draft: AppSpec
-  draft_bindings: AppBindings
-  created_by: string | null
+  /** 草稿和岗位绑定只给 manage（D-m）；use / observe 拿到的详情里没有 */
+  draft?: AppSpec
+  draft_bindings?: AppBindings
+  created_by?: string | null
+  /** 当前发布版的页面和入口目录：没有 manage 的人靠它找首页、入口名 */
+  published?: { version: number; pages: { id: string; title: string }[]; entries: { id: string; label: string }[] } | null
+  /** 能按新口径补算的视图（只在 manage 取详情时算，D-f） */
+  backfill?: BackfillPlan[]
+}
+
+/** 按新口径补算会发起几次（GET /api/apps/{id}/views/{view}/backfill，不发起运行） */
+export interface BackfillPlan {
+  view: string
+  alias: string
+  /** schedule:<id> 或 formal */
+  source: string
+  from_version: number
+  to_version: number
+  /** 趋势画几期 */
+  periods: number
+  will_launch: number
+  will_skip: number
+}
+
+/** POST 同一路径的结果：发起了几次、跳过了几次（同一输入已经补算过）；发起到一半被拒的原因照实交回 */
+export interface BackfillResult extends BackfillPlan {
+  launched: number
+  skipped: number
+  run_ids: string[]
+  refused?: string | null
 }

 export interface AppVersionRow {
@@ -534,14 +561,23 @@ export interface LookEvent {
 // 入站钩子（§2.4）与助手流（§2.9 L1、§3.11）
 // ---------------------------------------------------------------------------

+/** 钩子管理里的一行（D-g）：草稿或发布版里声明的 ∪ 已经生成过密钥的，按名字一行 */
 export interface AppHookRow {
-  id: string
+  /** app_hooks 那一行的 id；还没生成密钥时为 null */
+  id: string | null
  name: string
-  kind: 'start' | 'signal'
+  kind: 'start' | 'signal' | null
  entry_id: string | null
  enabled: boolean
-  created_at: string
-  /** 只在新建和轮换时出现一次 */
+  has_secret: boolean
+  /** 草稿的 spec.hooks 里有这个名字 */
+  in_draft: boolean
+  /** 当前发布版里有（入站请求按发布版认种类和入口，没有就拒收） */
+  in_published: boolean
+  /** 收请求的地址 /api/hooks/{id}；还没生成密钥时为 null */
+  url_path: string | null
+  created_at: string | null
+  /** 只在生成、轮换的回包里出现一次 */
  secret?: string
 }

@@ -568,3 +604,42 @@ export interface AppStreamFinal {
  issues: LintIssue[]
  explanation: string
 }
+
+// ---------------------------------------------------------------------------
+// 能力字段语义的审阅（D-i，GET /api/capabilities/review-queue、POST /api/capabilities/review/batch）
+// ---------------------------------------------------------------------------
+
+export interface CapabilityReviewRow {
+  capability: string
+  system: string
+  title: string
+  effect: CapabilityEffect
+  /** 返回里的位置（JSON 指针） */
+  pointer: string
+  /** label / meaning / unit / measure / codes */
+  field: string
+  field_text: string
+  /** 服务端写好的位置说法（「返回字段 /amount 的单位」） */
+  field_label: string
+  value: unknown
+  display: string
+  source: string
+  status: string
+  note: string | null
+  priority: number
+  reasons: string[]
+  /** 用到这个能力的运行次数 */
+  runs: number
+  /** 能力的治理版本：成批审阅时带上（乐观锁） */
+  version: number
+}
+
+export interface CapabilityReviewOutcome {
+  capability: string
+  pointer: string
+  field: string
+  ok: boolean
+  version: number | null
+  error: string | null
+  code: string | null
+}