From b3dcddb52e1fcd84a2e4e31c161c2e8f0be4bbac Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 22:16:58 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=AC=E5=9B=9E=E5=A4=8D=E6=A0=B8=E4=BF=AE?= =?UTF-8?q?=E6=AD=A3=EF=BC=88=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E4=BA=BA?= =?UTF-8?q?=E5=B7=A5=E8=8A=82=E7=82=B9=E4=B8=8D=E5=86=8D=E6=8C=A1=E5=8F=AC?= =?UTF-8?q?=E5=9B=9E=E3=80=81=E8=AF=BB=E4=B8=8D=E5=87=BA=E5=8F=AC=E5=9B=9E?= =?UTF-8?q?=E5=8D=A1=E8=AE=B0=E5=A4=B1=E8=B4=A5=E3=80=81=E6=97=A7=E7=89=88?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E6=8E=A5=E7=9D=80=E7=AE=97=E3=80=81=E5=9B=A2?= =?UTF-8?q?=E9=98=9F=E6=B8=A0=E9=81=93=E4=B8=80=E6=AC=A1=E5=8F=AC=E5=9B=9E?= =?UTF-8?q?=E5=8F=AA=E6=8E=A8=E4=B8=80=E6=9D=A1=E3=80=81=E7=9B=AE=E5=BD=95?= =?UTF-8?q?=E6=A0=B8=E5=AF=B9=E6=8C=89=E6=95=B0=E6=8D=AE=E6=BA=90=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E6=8E=92=E9=98=9F=EF=BC=9B=E8=B4=A6=E6=9C=AC=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3=E5=B8=A6=E4=B8=8A=E5=8F=AC=E5=9B=9E=E6=A0=87=E8=AE=B0?= =?UTF-8?q?=EF=BC=8C=E5=AF=B9=E8=B4=A6=E7=BB=93=E6=9E=9C=E5=8F=AF=E4=BB=A5?= =?UTF-8?q?=E5=8F=8D=E9=A6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - W4R-01：召回只查新版图里会不会写外部系统（apps/backfill 抽出 graph_external_write，graph_side_effect 复用它）。召回图只有两版 的取数和口径卡，图里别处的人工节点、按流程使用的问题碰不到它；口径卡上游的人工节点 derive 本来就认作数据部分不能复用。 「按新版重新出具」走完整的正式运行，仍按 graph_side_effect 全套拒绝。原来带审批的口径升版后一组都不召回，L6 对它们失效。 - W4R-09：收尾时召回卡读不出、或缺了某个比较的旧值 / 新值，照实记重算失败（读不出召回卡），原答案一个不动。原来两边都是空， 按渲染比被当成「没变」，给每个原答案永久盖上「已核对，不影响」，收过尾之后再也不核对。 - W4R-12：这一版已经不是当前发布版（之后又发布了更新的版本）时不再发起召回、不给接着算，批次里写 continue_reason，接着算的 接口 409 带同一句。同启动时接着做只认当前发布版：新版本的召回覆盖更早的所有版本，再按旧版开「口径升到 vN」的待办只是添乱。 - W4R-13：开登录时一组答案几个主人就几条待办，全局渠道却按这一批（recall.batch_of_key：同一次召回、同一个来由）只在最早开的 那条上推一次，答案、待办按这一批汇总，几条的链到待办箱；别的几条连 task.opened 也不再推全局渠道。个人渠道照旧每人一条。 - W4R-11：目录审阅之后的核对按数据源排队、批量审阅的几张表合成一遍（catalog_pass，同一数据源一把锁）；查询快照按数据源在库里 筛、按运行结束时间分页、每张表凑够 BASIS_MAX_RUNS 次运行就停；冻结的表结构快照这一遍里按工件 id 只读一次。原来批量审阅 几十张表就同时起几十个后台任务，每个都全表扫查询快照、反复读同一份几 MB 的表结构快照。 - W4R-03（本包那一半）：GET /api/runs/{id}/ledger 加 recall：checked（已按 vN 核对，不影响）、basis（依据有变：哪一种、几项、 新出现几条错误级问题）、tasks（看的这个人手上说的是这个答案、还没处理的 recall 待办）。不放数、不放表名。几种状态都带。 - W4R-10（本包那一半）：新事件 ledger.feedback（对账运行上的封存后事件），POST /api/ledger/answers/{run}/reconcile/{rid}/feedback {verdict: useful | unclear} 记反馈，同一个人改主意只算最后一次（app/ledger/feedback.py）；账本接口的 reconciled 每一项带 feedback {mine, useful, unclear}；统计 E2 的 feedback 数到它，去掉「第 4 波接上」的占位说法。离线核对页、封存后事件摘要 都认得这个事件。 测试：test_ledger_recall 新增七个用例、改了人工节点那一个（召回照算、重新出具照实拒绝）；新增 test_ledger_feedback； test_ledger_api 的 E2 断言跟上。每个修复都先写失败的测试、去掉修复回跑确认抓得住。后端全量 14237 通过。 --- backend/app/api/ledger.py | 28 ++- backend/app/apps/backfill.py | 33 ++- backend/app/bundle/build.py | 4 + backend/app/bundle/verifier.js | 1 + backend/app/core/events.py | 4 + backend/app/ledger/feedback.py | 108 ++++++++ backend/app/ledger/recall.py | 349 +++++++++++++++++++++----- backend/app/ledger/stats.py | 11 +- backend/app/ledger/view.py | 23 +- backend/app/notify/service.py | 56 ++++- backend/tests/test_ledger_api.py | 3 +- backend/tests/test_ledger_feedback.py | 119 +++++++++ backend/tests/test_ledger_recall.py | 277 +++++++++++++++++++- 13 files changed, 912 insertions(+), 104 deletions(-) create mode 100644 backend/app/ledger/feedback.py create mode 100644 backend/tests/test_ledger_feedback.py diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 5917c529..0b063434 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -10,6 +10,8 @@ - POST /api/ledger/answers/{run_id}/issue（answers_router，登录即可，第 3 波）：按口径正式出具（app/ledger/issue.py，拍板 L1、L5：参数只从比对工件取，使用者只能按他有权使用的应用固定着的版本出具）； - POST /api/ledger/answers/{run_id}/reconcile（同上）：口径对账，点开才跑一次确定性派生的只读图（app/ledger/reconcile.py）； +- POST /api/ledger/answers/{run_id}/reconcile/{reconcile_id}/feedback（同上，第 4 波）：对账结果上的「说明有用 / 看不懂」 + （app/ledger/feedback.py，统计 E2）； - GET /api/ledger/certify-queue、POST /api/ledger/certify-queue/dismiss（router，搭建者，第 3 波）：可沉淀的问题、「不是口径」 （app/ledger/certify.py）。 - GET /api/ledger/recall/{workflow_id}/{version}、POST …/continue（router，搭建者，第 4 波）：口径升到这一版之后的召回批次—— @@ -172,6 +174,28 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]: return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)} +class FeedbackIn(BaseModel): + """对账结果上的反馈：说明有用（useful）还是看不懂（unclear）。""" + + verdict: str = Field(min_length=1, max_length=20) + + +@answers_router.post("/{run_id}/reconcile/{reconcile_id}/feedback") +async def reconcile_feedback(run_id: str, reconcile_id: str, payload: FeedbackIn) -> dict[str, Any]: + """对账结果上的「说明有用 / 看不懂」（app/ledger/feedback.py，统计 E2）：记成对账运行上的封存后事件，同一个人改主意只算 + 最后一次。交回这次对账现在的反馈 {mine, useful, unclear}。不认的反馈 422；不是这个答案的对账 404；读不了这个答案 403； + 对账还没跑完 409。""" + from app.api.coded import CodedHTTPException + from app.api.deps import current_principal + from app.ledger import feedback + + try: + out = await feedback.give(run_id, reconcile_id, payload.verdict, principal=current_principal()) + except feedback.FeedbackRefused as e: + raise CodedHTTPException(e.status_code, e.message, e.code) from e + return {"run_id": reconcile_id, "feedback": out} + + @router.get("/recall/{workflow_id}/{version}") async def recall_batches(workflow_id: str, version: int, session: AsyncSession = Depends(get_session) ) -> dict[str, Any]: @@ -188,7 +212,7 @@ async def recall_batches(workflow_id: str, version: int, session: AsyncSession = async def recall_continue(workflow_id: str, version: int, session: AsyncSession = Depends(get_session) ) -> dict[str, Any]: """接着算：把还没重算的组再发起一批（每批 recall_limit 组，上限是 0 时按缺省的组数），交回发起了的召回运行和新的批次 - 情况。都算过了的回 409。""" + 情况。都算过了的、这一版已经不是当前发布版的（之后又发布了更新的版本）回 409，detail 写为什么。""" from app.api.coded import CodedHTTPException from app.api.deps import current_principal from app.ledger import recall @@ -197,7 +221,7 @@ async def recall_continue(workflow_id: str, version: int, session: AsyncSession raise HTTPException(404, "工作流不存在，可能已被删除") current = await recall.batches(session, workflow_id, version) if not current["can_continue"]: - raise CodedHTTPException(409, current.get("problem") or "没有还没重算的组", recall.REFUSED) + raise CodedHTTPException(409, current.get("continue_reason") or "没有还没重算的组", recall.REFUSED) launched = await recall.launch(workflow_id, version, batch="manual", actor=current_principal().username) async with SessionLocal() as fresh: view = await recall.batches(fresh, workflow_id, version) diff --git a/backend/app/apps/backfill.py b/backend/app/apps/backfill.py index 47349ed9..97630d6e 100644 --- a/backend/app/apps/backfill.py +++ b/backend/app/apps/backfill.py @@ -120,7 +120,6 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st 判断沿用应用检查对流程钉住项的那一套（apps/lint._lint_process）：人工节点、写外部系统的能力 （capabilities.registry.writes_externally，和运行时、发布检查同一个判断）、process.* 问题（子工作流先读好， process.human_in_subgraph 才查得到）。level 是那一版发布时的级别（门禁按它查），没有按 published。""" - from app.capabilities.registry import effective_effect, get_capability, writes_externally from app.engine.governance import publish_issues from app.engine.lint_process import load_subgraphs, with_subgraphs from app.engine.schema import GraphSpec, NodeType @@ -131,6 +130,34 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st return "的图无法解析" if humans := [n.title for n in graph.nodes if n.type == NodeType.HUMAN]: return f"里有人工节点「{'、'.join(humans[:3])}」" + if reason := await _external_write(session, graph): + return reason + with with_subgraphs(await load_subgraphs(session, graph)): + issues = publish_issues(graph, level=level or "published") + if process := [i for i in issues if (i.code or "").startswith("process.")]: + return f"按流程使用不安全：{process[0].message.rstrip('。')}" + return None + + +async def graph_external_write(session: AsyncSession, graph_json: Any) -> str | None: + """一张图里有没有调用工具节点会写外部系统（graph_side_effect 里的那一条）：有的交回一句原因，没有返回 None。 + + 口径召回（ledger/recall）只用这一条：召回图是从口径卡往上派生的取数和口径卡，图里别处的人工节点、按流程使用的问题 + 碰不到它（细案做法 7 不重算的只有缺入口字段、数据部分不能复用、会写外部系统三种）；替人整张图重跑的补算、按口径出具 + 仍然走 graph_side_effect 全套。""" + from app.engine.schema import GraphSpec + + try: + graph = GraphSpec.model_validate(graph_json) + except ValueError: + return "的图无法解析" + return await _external_write(session, graph) + + +async def _external_write(session: AsyncSession, graph: Any) -> str | None: + from app.capabilities.registry import effective_effect, get_capability, writes_externally + from app.engine.schema import NodeType + for node in graph.nodes: name = node.config.get("tool") if node.type == NodeType.TOOL else None if not isinstance(name, str) or not name: @@ -138,10 +165,6 @@ async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: st cap = await get_capability(session, name) if cap is not None and writes_externally(cap.name, effective_effect(cap), cap.endpoint_id): return f"的「{node.title}」会写外部系统（{name}）" - with with_subgraphs(await load_subgraphs(session, graph)): - issues = publish_issues(graph, level=level or "published") - if process := [i for i in issues if (i.code or "").startswith("process.")]: - return f"按流程使用不安全：{process[0].message.rstrip('。')}" return None diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index 9dd73f2e..b088980a 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -1197,6 +1197,10 @@ def _summary(etype: str, data: Any) -> str: items = data.get("items") count = len(items) if isinstance(items, list) else 0 return f"依据的{kind or '东西'}有变" + (f"（{count} 项）" if count else "") + if etype == EventType.LEDGER_FEEDBACK: + # 对账结果上的反馈（next-round 02 做法 10）：只说是哪一种，不带谁给的 + verdict = {"useful": "说明有用", "unclear": "看不懂"}.get(str(data.get("verdict") or "")) + return f"反馈：{verdict}" if verdict else "对账说明的反馈" if etype == EventType.LOG and isinstance(data.get("message"), str): return _clip(data["message"]) return "" diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index f4ae077d..24bf3a60 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -66,6 +66,7 @@ 'issuance': '出具判定', 'caliber.upgrade': '口径卡升版处置', 'catalog.drift': '数据目录变化', 'schema.drift': '表结构变化', 'report.checked': '报告自查', 'evidence.judged': '按需裁判', 'file.extracted': '读入附件', 'ledger.matched': '比对已发布口径', 'ledger.recalled': '口径召回核对', 'ledger.basis_changed': '答案的依据有变', + 'ledger.feedback': '对账说明的反馈', 'log': '日志', 'usage': '用量', } const T = { diff --git a/backend/app/core/events.py b/backend/app/core/events.py index 4cbd7c7a..fa295c27 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -102,6 +102,10 @@ class EventType(StrEnum): # structure 用到的表或列被删、列的类型变了；values 触及有效记录条件的新码值；catalog_edit 处理变化时改了目录。带数据源、 # 表、哪几项（路径和叫法），按新目录跑 SQL 检查新出现的错误级问题条数。同上，封存之后追加，不进封存范围 LEDGER_BASIS_CHANGED = "ledger.basis_changed" + # 对账结果上的「说明有用 / 看不懂」（02 做法 10 的 E2，ledger/feedback）：追加在对账运行（origin_kind=reconcile）上， + # {verdict: useful | unclear, by, verified, source_run}。同一个人改主意再记一条，统计只算他最后一次。同上，封存之后 + # 追加，是批注不是证据 + LEDGER_FEEDBACK = "ledger.feedback" # 应用平台（docs/design/app-platform-and-look.md §3.11） # 运行来源：发起时写进 launch_run 的 notes，在封存范围内（engine/origin.py） diff --git a/backend/app/ledger/feedback.py b/backend/app/ledger/feedback.py new file mode 100644 index 00000000..216b8057 --- /dev/null +++ b/backend/app/ledger/feedback.py @@ -0,0 +1,108 @@ +"""对账结果上的「说明有用 / 看不懂」（docs/design/next-round/02-口径账本.md 做法 10 的 E2）。 + +对账（ledger/reconcile）把这次的答案和一份相关口径在同一份数据上各算一次，结果卡片上写差在哪、差多少。看的人觉得这段 +说明有没有帮他看懂，点一下「说明有用」或「看不懂」——这是 E2 里唯一要人给的数，衡量差异说明写得好不好。 + +- 记在哪：对账运行上的封存后事件 ledger.feedback（{verdict, by, verified, source_run}，engine/run_append，post_seal）。 + 运行本身就是记录，不另建反馈表（同统计「全部现算、不另存」）；批注不是证据，封存核对照旧一致。 +- 谁能给：能读这个答案的人（同发起对账，ledger/reconcile：使用者对自己的答案也能对账）；对账要是这个答案的（origin.source_run）， + 而且已经跑完封存。 +- 怎么数：同一个人（开登录时按用户名；关登录时署名是自报的，按署名、没署名的算同一个）改主意再记一条，只算他最后一次。 + 账本接口 reconciled 每一项带上 {mine, useful, unclear}（feedback_of），统计 E2 数整个窗口（counts）。 +""" +from __future__ import annotations + +from collections.abc import Collection, Iterable +from typing import Any + +from sqlalchemy import select + +from app.core.events import EventType +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent + +#: 两种反馈 +VERDICTS = ("useful", "unclear") +#: 被拒的机器码 +REFUSED = "ledger_feedback_refused" +NOT_FOUND = "这次对账不存在，或者不是这个答案的对账" +NO_ACCESS = "你没有权限查看这个答案" +NOT_SEALED = "对账还没运行完，等结果出来再评说明有没有用" +BAD_VERDICT = "反馈只有两种：说明有用、看不懂" + + +class FeedbackRefused(Exception): + def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None: + super().__init__(message) + self.status_code = status_code + self.message = message + self.code = code + + +def _who(principal: Any) -> str: + return str(getattr(principal, "username", None) or "") + + +async def give(answer_id: str, reconcile_id: str, verdict: str, *, principal: Any) -> dict[str, Any]: + """记一条反馈（见模块说明），交回这次对账现在的反馈 {mine, useful, unclear}。不能记的抛 FeedbackRefused。""" + from app.api.deps import can_read_run + from app.engine.run_append import append_events + from app.ledger.reconcile import ORIGIN as RECONCILE + + if verdict not in VERDICTS: + raise FeedbackRefused(422, BAD_VERDICT) + async with SessionLocal() as session: + answer = await session.get(Run, answer_id) + rec = await session.get(Run, reconcile_id) + origin = rec.origin if rec is not None and isinstance(rec.origin, dict) else {} + if answer is None or rec is None or rec.origin_kind != RECONCILE or origin.get("source_run") != answer_id: + raise FeedbackRefused(404, NOT_FOUND) + if getattr(principal, "mode", "off") != "off" and not await can_read_run(principal, answer): + raise FeedbackRefused(403, NO_ACCESS) + data = {"verdict": verdict, "by": _who(principal) or None, "verified": bool(getattr(principal, "verified", False)), + "source_run": answer_id} + if not await append_events(reconcile_id, [(EventType.LEDGER_FEEDBACK, None, data)], require="sealed"): + raise FeedbackRefused(409, NOT_SEALED) + async with SessionLocal() as session: + return (await feedback_of(session, [reconcile_id], principal))[reconcile_id] + + +async def _latest(session: Any, run_ids: Collection[str]) -> dict[str, dict[str, str]]: + """{对账运行: {谁: 他最后一次给的}}。""" + out: dict[str, dict[str, str]] = {} + if not run_ids: + return out + rows = (await session.execute(select(RunEvent.run_id, RunEvent.data).where( + RunEvent.run_id.in_(list(run_ids)), RunEvent.type == EventType.LEDGER_FEEDBACK).order_by(RunEvent.seq))).all() + for run_id, data in rows: + if isinstance(data, dict) and data.get("verdict") in VERDICTS: + out.setdefault(run_id, {})[str(data.get("by") or "")] = str(data["verdict"]) + return out + + +async def feedback_of(session: Any, run_ids: Iterable[str], principal: Any) -> dict[str, dict[str, Any]]: + """每次对账的反馈：{对账运行: {mine: 看的这个人最后一次给的（没给过为 None）, useful: 几个人, unclear: 几个人}}。""" + ids = list(dict.fromkeys(run_ids)) + latest = await _latest(session, ids) + me = _who(principal) + out: dict[str, dict[str, Any]] = {} + for run_id in ids: + votes = latest.get(run_id, {}) + out[run_id] = {"mine": votes.get(me), "useful": sum(v == "useful" for v in votes.values()), + "unclear": sum(v == "unclear" for v in votes.values())} + return out + + +async def counts(session: Any, run_ids: Collection[str]) -> dict[str, Any]: + """统计 E2 的反馈（ledger/stats._e2）：这些对账运行上，说明有用、看不懂各几个（每人只算最后一次），给过反馈的对账几次， + 有用的比率（分母为 0 时为 None）。""" + latest = await _latest(session, run_ids) + useful = sum(v == "useful" for votes in latest.values() for v in votes.values()) + unclear = sum(v == "unclear" for votes in latest.values() for v in votes.values()) + total = useful + unclear + return {"available": True, "useful": useful, "unclear": unclear, "runs": sum(1 for v in latest.values() if v), + "useful_rate": round(useful / total, 4) if total else None} + + +__all__ = ["BAD_VERDICT", "NOT_FOUND", "NOT_SEALED", "REFUSED", "VERDICTS", "FeedbackRefused", "counts", "feedback_of", + "give"] diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index bfce4cba..938a81b4 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -13,8 +13,10 @@ metrics_from 指向召回卡；不加报告、不调模型）。两版在同一次运行里查同一份数据，差值只反映口径；数据随时间的变化另外拿 原答案当时的值对照着说（task_payload 的 original）。 - 不能重算的照实记「无法重算：原因」，不发起：口径的数据部分不能复用（derive.NotDerivable）、新版里没有这张口径卡、两版 - 没有同名的指标、新版入口要的参数这一组里没有或不合规（compose.check_inputs 同一套校验）、新版的图里有人工节点或会写外部 - 系统（apps/backfill.graph_side_effect，和按新口径补算、按口径出具同一个判断：替人重跑不代人做这些事）。 + 没有同名的指标、新版入口要的参数这一组里没有或不合规（compose.check_inputs 同一套校验）、新版的图里会写外部系统 + （apps/backfill.graph_external_write，和按新口径补算、按口径出具同一个判断）。新版图里别处有人工节点、按流程使用不安全的 + 照样重算：召回图只有两版的取数和口径卡，碰不到它们（口径卡上游的人工节点 derive 已经认作数据部分不能复用）；「按新版 + 重新出具」走完整的正式运行，仍按 graph_side_effect 全套拒绝（ledger/issue.check_target）。 - 上限：每次发布最多自动重算 ledger.recall_limit 组（设置，缺省 50；0 表示不自动重算，只列出受影响的答案）。超出的写 「另有 M 组未重算」，在版本历史里点「接着算」手动接着发起下一批（每批同样的组数，上限是 0 时按缺省值）。 - **计划从已有的召回运行重算**（照 apps/backfill 的写法，不建任务表）：一组算没算过，看有没有 origin 指向（W、V、N、 @@ -25,13 +27,14 @@ 不同就是数变了。数变了：给这一组答案的主人开 recall 待办（挂在召回运行上；开登录时每个主人一条，只列他的答案；关登录时 一条派给任何人），推 ledger.recall 通知。数没变：只记账——在每个原答案的运行上追加 ledger.recalled（「已按 vN 核对， 不影响」）。召回运行自己也追加一条 ledger.recalled 记结论（只记变没变、几个答案、几条待办，不记数），它同时是「收过尾」的 - 记号。召回运行没成功的照实记重算失败。 + 记号。召回运行没成功的、成功了却读不出召回卡（或缺了某个比较的旧值、新值）的照实记重算失败，原答案一个不动。 **二、目录事实被驳回或修改**（只管影响数字的那一组：有效记录条件、去重、业务日期、关联、业务主键、度量类型，同 data/catalog_review.FIELD_GROUP 的 facts）：审阅写入之后（data/catalog 的审阅、人工编辑、保存提案，catalog_review 的批量审阅） 在后台找出查过这张表、而且查询快照冻结的目录里看得到这一项的探索运行，追加 ledger.basis_changed（kind=catalog），界面标 「依据有变」；对这些运行执行过的 SQL 按新目录跑一遍 SQL 检查（data/sqlcheck，纯计算），比冻结目录下多出错误级问题的才开 -一条 recall 待办（reason=catalog，只有「知道了」）通知答案的主人（拍板 L7：不自动重问）。 +一条 recall 待办（reason=catalog，只有「知道了」）通知答案的主人（拍板 L7：不自动重问）。同一个数据源排着做、批量审阅的几张表 +合成一遍（catalog_pass）：查询快照按数据源在库里筛、凑够就停，冻结的表结构快照（大库的能有几 MB）同一份只读一次。 **三、表结构变动**：消费 data/source_changes（list_changes(since_seq) 按 seq 游标拉，游标记在设置表里；探查、剖析任务结束、 处理变化改了目录时触发）。只看 triggers_recall 认的：表或列被删、列的类型变了标「依据的表结构已变」（kind=structure）； @@ -41,7 +44,11 @@ data/catalog_review.FIELD_GROUP 的 facts）：审阅写入之后（data/catalog recall 待办（reason=structure）通知发布人。 **通知**：ledger.recall 只推口径升版、数真变了的那一种待办（消息照 notify/service 的规矩：标题、状态、条数、链接，不放 -新旧值和差值）；目录和表结构的两种待办照别的待办推 task.opened。 +新旧值和差值）；目录和表结构的两种待办照别的待办推 task.opened。全局渠道按一批待办（一次召回开给几个主人的那几条， +batch_of_key）只推一次，个人渠道每个主人各一条（notify/service.notify_recall）。 + +**答案上的标记**（answer_marks）：账本接口 GET /api/runs/{id}/ledger 的 recall——已按 vN 核对、依据有变、看的这个人手上说的 +是这个答案的 recall 待办，给问数据的轮次和证据面板「口径」一步。 **自动的那几样**（发布后自动那一批、探查剖析之后拉表结构变化、启动时接着做）和调度循环同一个开关（automatic， AGENTLAB_SCHEDULER）：它们是在后台替人查业务库，测试、拷来的数据目录（检查脚本）不做；关着时版本历史里照样列出受影响的 @@ -95,6 +102,11 @@ NEW_INPUTS = "新版 v{version} 的入口要的参数这一组里没有或不合 NEW_GONE = "新版本不存在，可能已被删除" LAUNCH_REFUSED = "没能发起召回运行：{reason}" RUN_FAILED = "重算失败：{reason}" +#: 召回运行成功了，召回卡（或其中某个比较的旧值、新值）却读不出来：照实算重算失败，不当成数没变 +CARD_UNREADABLE = "读不出召回卡" +#: 这一版已经不是当前发布版：不再接着算（新版本的召回本来就覆盖更早的所有版本，同 resume 只接着做当前发布版） +SUPERSEDED = "已有更新的发布版 v{latest}，召回在 v{latest} 那一版接着看" +NOT_CURRENT = "v{version} 已不是当前发布版，召回不再接着算" #: 收尾 CHANGED_TITLE = "口径「{name}」升到 v{version}，答案的数变了" CATALOG_TITLE = "答案依据的数据目录有变：按新目录检查出了新的问题" @@ -152,6 +164,8 @@ class Plan: workflow_id: str workflow_name: str new_version: int + #: 工作流现在的发布版（不是 new_version 的不再接着算，见 stale） + published_version: int | None = None groups: list[Group] = field(default_factory=list) #: 每个旧版本的口径卡对、旧版本自己的问题（不是口径、数据部分不能复用……） pairs: dict[int, list[Pair]] = field(default_factory=dict) @@ -165,6 +179,15 @@ class Plan: """还没算、能算的组（按最近一次用到的先后）。""" return [g for g in self.groups if g.run is None and g.problem is None] + def stale(self) -> str | None: + """这一版已经不是当前发布版时不再发起（手动接着算也不）的原因：之后又发布了更新的版本，新版本的召回覆盖更早的所有 + 版本（同 resume 只接着做当前发布版），再按这一版开「口径升到 vN」的待办只是添乱。是当前发布版为 None。""" + if self.published_version == self.new_version: + return None + if self.published_version is not None and self.published_version > self.new_version: + return SUPERSEDED.format(latest=self.published_version) + return NOT_CURRENT.format(version=self.new_version) + def _aware(moment: datetime | None) -> datetime | None: if moment is None: @@ -212,19 +235,20 @@ async def recall_runs(session: Any, workflow_id: str, new_version: int | None = async def build_plan(session: Any, workflow_id: str, new_version: int) -> Plan: """W 升到 N 的召回计划（见模块说明）。只读，不发起。""" from app.apps import versions as app_versions - from app.apps.backfill import graph_side_effect + from app.apps.backfill import graph_external_write from app.ledger import compose, derive workflow = await session.get(Workflow, workflow_id) plan = Plan(workflow_id=workflow_id, workflow_name=workflow.name if workflow is not None else "", - new_version=int(new_version)) + new_version=int(new_version), + published_version=workflow.published_version if workflow is not None else None) new_row = await app_versions.version_row(session, workflow_id, int(new_version)) if workflow else None if new_row is None: plan.problem = NEW_GONE return plan found = await ledger_uses.find_uses(session, workflow_id, below=int(new_version)) plan.found = found - side = await graph_side_effect(session, new_row.graph, level=new_row.level) + side = await graph_external_write(session, new_row.graph) for version in found.versions(): old_row = await app_versions.version_row(session, workflow_id, version) cards = source_cards(old_row.graph if old_row is not None else None) @@ -405,7 +429,7 @@ async def launch(workflow_id: str, new_version: int, *, batch: str = "auto", act async with SessionLocal() as session: plan = await build_plan(session, workflow_id, int(new_version)) limit = int((await ledger_settings(session))["recall_limit"]) - if plan.problem is not None: + if plan.problem is not None or plan.stale() is not None: return [] if batch == "auto": size = max(0, limit - plan.auto_launched) @@ -496,7 +520,8 @@ def _group_state(group: Group) -> tuple[str, str | None]: async def batches(session: Any, workflow_id: str, version: int) -> dict[str, Any]: """版本历史里「这一版的召回」：每组的结论（数变了 / 没变 / 无法重算：原因 / 重算失败 / 还在算 / 未重算），「另有 M 组 - 未重算」和能不能接着算。接口只给搭建者（api/ledger 的 router）：参数原样给，同他在召回运行的输入里读到的一样；使用者 + 未重算」和能不能接着算（还有没算的组却不能接着算时 continue_reason 写为什么：之后又发布了更新的版本……）。 + 接口只给搭建者（api/ledger 的 router）：参数原样给，同他在召回运行的输入里读到的一样；使用者 看的是 recall 待办（task_payload，按遮罩换过）。""" from app.api.settings import LEDGER_DEFAULTS, ledger_settings @@ -526,10 +551,11 @@ async def batches(session: Any, workflow_id: str, version: int) -> dict[str, Any "latest": group.latest.isoformat() if group.latest else None, "state": state, **({"reason": reason} if reason else {}), **extra}) pending = counts["pending"] + why_not = (plan.problem or plan.stale()) if pending > 0 else None return {"workflow_id": workflow_id, "workflow_name": plan.workflow_name, "version": int(version), "limit": limit, "batch_size": limit or int(LEDGER_DEFAULTS["recall_limit"]), "problem": plan.problem, "caliber": bool(plan.pairs) or bool(plan.groups), "counts": counts, "groups": items, - "remaining": pending, "can_continue": pending > 0 and plan.problem is None, + "remaining": pending, "can_continue": pending > 0 and why_not is None, "continue_reason": why_not, "truncated": bool(plan.found and plan.found.truncated)} @@ -644,7 +670,14 @@ async def settle(run_id: str) -> str | None: await _mark(run_id, {**base, "outcome": "failed", "changed": None, "error": str(run.error or "运行没有成功结束")[:300]}) return "failed" - results = compare(origin.get("metrics") or [], await card_values(run_id)) + meta = origin.get("metrics") or [] + values = await card_values(run_id) + # 读不出召回卡、少了哪个比较的旧值或新值：两边都是空，按渲染比会被当成「没变」，给每个原答案盖上「已核对，不影响」 + # ——照实记重算失败，原答案一个不动（收过尾就不再核对，盖错了收不回来） + if not meta or any(f"{side}_{m.get('k')}" not in values for m in meta for side in ("old", "new")): + await _mark(run_id, {**base, "outcome": "failed", "changed": None, "error": CARD_UNREADABLE}) + return "failed" + results = compare(meta, values) changed = any(r["changed"] for r in results) answers = list(dict.fromkeys(str(u.get("run_id")) for u in origin.get("uses") or [] if u.get("run_id"))) tasks: list[str] = [] @@ -738,6 +771,13 @@ def _owner_of_key(key: str) -> tuple[str, str]: return (parts[-2], parts[-1]) if len(parts) >= 4 else ("anyone", "") +def batch_of_key(key: str) -> str: + """correlation_key 去掉派给谁的那两段：同一次召回（同一个来由、同一次运行、同一条依据）给各个主人开的那一批待办是同一个 + 值。全局渠道按它只推一次（notify/service.notify_recall）。""" + parts = (key or "").split(":") + return ":".join(parts[:-2]) if len(parts) >= 4 else key or "" + + async def task_payload(session: Any, task: Any, principal: Any) -> dict[str, Any]: """recall 待办给处理人看的内容（待办箱、待办详情的 payload，process/tasks._describe 调）：旧口径的值、新口径的值、两版之差 和原答案当时的值（每个都带出处：哪次运行、哪张卡、哪个指标，点得开），这一组的参数，能做的动作。使用者看到的值照样 @@ -896,6 +936,72 @@ def _cell(item: Mapping[str, Any], value: Any, text: Any, hide: Any, *, missing: return out +# ========================================================================== +# 答案上的召回标记（账本接口 GET /api/runs/{id}/ledger 的 recall，ledger/view.run_view） +# ========================================================================== + + +async def answer_marks(session: Any, run_id: str, principal: Any) -> dict[str, Any]: + """一个答案（探索运行、正式运行）身上的召回标记，给问数据的轮次和证据面板「口径」一步（细案做法 7：界面标「依据有变」、 + 「已按 vN 核对，不影响」）： + + - checked：数没变、追加在这次运行上的 ledger.recalled（role=answer）——按哪份口径、从哪一版核对到哪一版、哪次召回、什么时候； + - basis：ledger.basis_changed——哪一种依据（目录事实、表结构、取值、处理时改了目录）、几项、按新目录新出现几条错误级问题、 + 什么时候； + - tasks：看的这个人手上还没处理的、说的是这个答案的 recall 待办（数变了的那种挂在召回运行上，列着这个答案；目录有变的 + 挂在这次运行上），界面链到待办箱。 + + 不放数（新旧值、差值只在待办里给，按遮罩换过），不放表名、取值。""" + from app.process.tasks import Task, _assignments, handles + + checked: list[dict[str, Any]] = [] + basis: list[dict[str, Any]] = [] + rows = (await session.execute(select(RunEvent.type, RunEvent.data, RunEvent.ts).where( + RunEvent.run_id == run_id, RunEvent.type.in_((EventType.LEDGER_RECALLED, EventType.LEDGER_BASIS_CHANGED))) + .order_by(RunEvent.seq))).all() + def at(ts: Any) -> str | None: + return datetime.fromtimestamp(ts, timezone.utc).isoformat(timespec="seconds") \ + if isinstance(ts, (int, float)) else None + + workflows: set[str] = set() + for etype, data, ts in rows: + data = data if isinstance(data, dict) else {} + if etype == EventType.LEDGER_RECALLED and data.get("role") == "answer": + checked.append({k: data.get(k) for k in ("workflow_id", "from_version", "to_version", "recall_run")} + | {"changed": bool(data.get("changed")), "at": at(ts)}) + workflows.add(str(data.get("workflow_id") or "")) + elif etype == EventType.LEDGER_BASIS_CHANGED: + items = data.get("items") + basis.append({"kind": data.get("kind"), "items": len(items) if isinstance(items, list) else 0, + "new_errors": int(data.get("new_errors") or 0), "at": at(ts)}) + names = dict((await session.execute(select(Workflow.id, Workflow.name).where( + Workflow.id.in_(sorted(w for w in workflows if w))))).all()) if workflows - {""} else {} + for item in checked: + item["workflow_name"] = names.get(str(item.get("workflow_id") or "")) + # 说的是这个答案、还没处理的 recall 待办：挂在这次运行上的（目录有变），和召回运行 origin.uses 里列着它的（数变了）。 + # 运行 id 是 32 位十六进制，在 uses 的 JSON 文字里找子串不会误中，找到的再逐条核对 + candidates = (await session.execute( + select(Task, Run.origin).join(Run, Run.id == Task.run_id) + .where(Task.kind == TASK_KIND, Task.status == "pending", + (Task.run_id == run_id) | ((Run.origin_kind == ORIGIN) + & (func.instr(func.json_extract(Run.origin, "$.uses"), run_id) > 0))) + .order_by(Task.created_at, Task.id))).all() + mine = await _assignments(session, [t.id for t, _ in candidates]) + tasks: list[dict[str, Any]] = [] + for task, origin in candidates: + key = task.correlation_key or "" + reason = key.split(":", 1)[0] if key.split(":", 1)[0] in REASONS else "caliber" + if task.run_id != run_id: + origin = origin if isinstance(origin, dict) else {} + if not any(u.get("run_id") == run_id for u in _uses_of(origin, _owner_of_key(key))): + continue + if not handles(mine.get(task.id, []), principal): + continue + tasks.append({"id": task.id, "reason": reason, + **({"to_version": (origin or {}).get("to_version")} if reason == "caliber" else {})}) + return {"checked": checked, "basis": basis, "tasks": tasks} + + # ========================================================================== # 「按新版重新出具」「知道了」 # ========================================================================== @@ -999,51 +1105,85 @@ def fact_changes(before: Mapping[str, Any] | None, after: Mapping[str, Any] | No return out -async def _explore_snapshots(session: Any, source: Any, table: str, *, before: datetime | None = None - ) -> list[tuple[str, str]]: - """查过 source 这张表的探索运行（计入统计的、成功封存的）和那条查询快照：[(运行, 快照)]，新的在前，最多 BASIS_MAX_RUNS - 次运行。表名按 name_key 比，表已经不在表结构里也认（表被删了正要找用过它的）。""" +#: 查询快照一页取几行（按运行结束时间从新到旧翻，每张表凑够 BASIS_MAX_RUNS 次运行就不再往下翻） +SNAPSHOT_PAGE = 500 + + +async def _explore_snapshots(session: Any, source: Any, tables: Iterable[str], *, before: datetime | None = None + ) -> dict[str, list[tuple[str, str]]]: + """查过 source 这几张表的探索运行（计入统计的、成功封存的）和那几条查询快照：{表: [(运行, 快照)]}，新的在前，每张表最多 + BASIS_MAX_RUNS 次运行。表名按 name_key 比，表已经不在表结构里也认（表被删了正要找用过它的）。 + + 数据源在库里就筛掉（引用的 meta 里记着源 id 和源名，query_snapshot_meta）：部署久了查询快照有几万条，别的源的不该读回来； + 按运行结束时间、运行 id 排序分页翻（同一次运行的几条挨在一起），每张表都凑够了就停。表名不在库里筛：meta 里的表名是 + SQL 里写的样子（带 schema 前缀、大小写不一），按 name_key 比只能在这里做。""" + from sqlalchemy import or_ + from app.data.catalog import QUERY_SNAPSHOT_KIND from app.data.names import name_key from app.db.models import Artifact from app.engine.origin import counted_runs - want = name_key(table) + want: dict[str, list[str]] = {} + for t in dict.fromkeys(tables): + want.setdefault(name_key(t), []).append(t) + out: dict[str, list[tuple[str, str]]] = {t: [] for names in want.values() for t in names} + runs: dict[str, set[str]] = {t: set() for t in out} stmt = (select(Artifact.id, Artifact.meta, Run.id, Run.finished_at).join(Run, Run.id == Artifact.run_id) .where(Artifact.kind == QUERY_SNAPSHOT_KIND, Run.run_class == "exploratory", Run.status == "succeeded", - Run.manifest_hash.is_not(None), counted_runs()).order_by(Run.finished_at.desc())) - out: list[tuple[str, str]] = [] - runs: set[str] = set() + Run.manifest_hash.is_not(None), counted_runs(), + or_(func.json_extract(Artifact.meta, "$.source_id") == source.id, + func.json_extract(Artifact.meta, "$.source") == source.name)) + .order_by(Run.finished_at.desc(), Run.id, Artifact.id)) cut = _aware(before) - for artifact, meta, run_id, finished in (await session.execute(stmt)).all(): - if cut is not None and (_aware(finished) is None or _aware(finished) > cut): - continue - meta = meta if isinstance(meta, dict) else {} - if meta.get("source_id") != source.id and meta.get("source") != source.name: - continue - if not any(name_key(str(t).rsplit(".", 1)[-1]) == want for t in meta.get("tables") or []): - continue - if run_id not in runs and len(runs) >= BASIS_MAX_RUNS: - continue - runs.add(run_id) - out.append((run_id, artifact)) + offset = 0 + last_run: str | None = None + while want: + page = (await session.execute(stmt.limit(SNAPSHOT_PAGE).offset(offset))).all() + offset += len(page) + for artifact, meta, run_id, finished in page: + if run_id != last_run and all(len(seen) >= BASIS_MAX_RUNS for seen in runs.values()): + return out + last_run = run_id + if cut is not None and (_aware(finished) is None or _aware(finished) > cut): + continue + meta = meta if isinstance(meta, dict) else {} + if meta.get("source_id") != source.id and meta.get("source") != source.name: + continue + hit = {name_key(str(t).rsplit(".", 1)[-1]) for t in meta.get("tables") or []} + for table in (t for key in hit & want.keys() for t in want[key]): + if run_id not in runs[table] and len(runs[table]) >= BASIS_MAX_RUNS: + continue + runs[table].add(run_id) + # 翻页之间进来了新的运行，偏移会错开一两行：同一条不收两次 + if (run_id, artifact) not in out[table]: + out[table].append((run_id, artifact)) + if len(page) < SNAPSHOT_PAGE: + break return out -def _frozen_of(snapshot_id: str) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: - """查询快照 → (快照内容, 它当时冻结的表结构快照)。读不出为 None。""" +def _frozen_of(snapshot_id: str, cache: dict[str, Any] | None = None + ) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + """查询快照 → (快照内容, 它当时冻结的表结构快照)。读不出为 None。cache 是这一遍里读过的工件（{工件 id: 内容}）：工件按内容 + 寻址、不会变，几十次运行冻结的常常是同一份表结构快照（大库的能有几 MB），同一遍里只读一次。""" from app.core.artifact_store import load - try: - snap = load(snapshot_id) - except ValueError: - return None, None + def read(artifact_id: str) -> Any: + if cache is not None and artifact_id in cache: + return cache[artifact_id] + try: + content = load(artifact_id) + except ValueError: + content = None + if cache is not None: + cache[artifact_id] = content + return content + + snap = read(snapshot_id) if not isinstance(snap, dict): return None, None - try: - schema = load(str(snap.get("schema_artifact") or "")) if snap.get("schema_artifact") else None - except ValueError: - schema = None + schema = read(str(snap["schema_artifact"])) if snap.get("schema_artifact") else None return snap, schema if isinstance(schema, dict) else None @@ -1096,28 +1236,77 @@ async def _answer_owner(session: Any, run_id: str) -> list[tuple[str, str]]: return use.owners +@dataclass +class CatalogEdit: + """一张表这一次影响数字的改动（目录审阅写入、新码值、处理变化时改了目录）：catalog_changed 一遍里可以有好几张表。""" + + table: str + changes: Sequence[Mapping[str, Any]] + catalog_version: int | None = None + kind: str = "catalog" + change_id: str | None = None + + async def catalog_changed(source_id: str, table: str, changes: Sequence[Mapping[str, Any]], *, catalog_version: int | None, kind: str = "catalog", change_id: str | None = None) -> int: """目录里这张表影响数字的几项被驳回或改了（见模块说明二）：冻结目录里看得到其中某项的探索运行追加 ledger.basis_changed； - 按新目录多出错误级问题的开 recall 待办（reason=catalog）。返回标了几次运行。""" + 按新目录多出错误级问题的开 recall 待办（reason=catalog）。返回标了几次运行。同一个数据源的排着做（_catalog_lock）。""" + return await catalog_pass(source_id, [CatalogEdit(table=table, changes=changes, catalog_version=catalog_version, + kind=kind, change_id=change_id)]) + + +#: 同一个数据源的目录核对排着做：{事件循环: {数据源: 锁}}。批量审阅几十张表、几个人同时审阅，每一遍都要翻查询快照、读冻结 +#: 的表结构快照，同时跑几十遍会把单进程的服务拖住 +_catalog_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]]" = \ + weakref.WeakKeyDictionary() + + +def _catalog_lock(source_id: str) -> asyncio.Lock: + return _catalog_locks.setdefault(asyncio.get_running_loop(), {}).setdefault(source_id, asyncio.Lock()) + + +async def catalog_pass(source_id: str, edits: Sequence[CatalogEdit]) -> int: + """一遍核对同一个数据源的几张表（catalog_changed、审阅之后排着的那一批）：查询快照只翻一次，冻结的表结构快照同一份只读 + 一次。返回标了几次运行。""" + async with _catalog_lock(source_id): + return await _catalog_pass(source_id, edits) + + +async def _catalog_pass(source_id: str, edits: Sequence[CatalogEdit]) -> int: from app.data import catalog from app.db.models import DataSource - paths = [str(c.get("path")) for c in changes if c.get("path")] - if not paths and kind == "catalog": + # 目录审阅那一种得有改了的项（路径）才核对；新码值、处理时改了目录的没有逐项路径，照样核对 + edits = [e for e in edits if e.kind != "catalog" or any(c.get("path") for c in e.changes)] + if not edits: return 0 async with SessionLocal() as session: source = await session.get(DataSource, source_id) if source is None: return 0 - found = await _explore_snapshots(session, source, table) + found = await _explore_snapshots(session, source, list(dict.fromkeys(e.table for e in edits))) current = {t: e.notes for t, e in (await catalog.read_catalog(session, source_id)).items()} - by_run: dict[str, list[tuple[dict[str, Any], dict[str, Any]]]] = {} - for run_id, snapshot in found: - snap, frozen = await asyncio.to_thread(_frozen_of, snapshot) - if snap is None or frozen is None: - continue - by_run.setdefault(run_id, []).append((snap, frozen)) + cache: dict[str, Any] = {} + marked = 0 + for edit in edits: + by_run: dict[str, list[tuple[dict[str, Any], dict[str, Any]]]] = {} + for run_id, snapshot in found.get(edit.table) or []: + snap, frozen = await asyncio.to_thread(_frozen_of, snapshot, cache) + if snap is None or frozen is None: + continue + by_run.setdefault(run_id, []).append((snap, frozen)) + marked += await _mark_catalog(source, edit, by_run, current) + return marked + + +async def _mark_catalog(source: Any, edit: CatalogEdit, + by_run: Mapping[str, list[tuple[dict[str, Any], dict[str, Any]]]], + current: Mapping[str, Mapping[str, Any]]) -> int: + """一张表的这次改动：冻结目录里看得到改动项的运行追加 ledger.basis_changed，按新目录多出错误级问题的开待办。""" + from app.engine.run_append import append_events + + table, changes = edit.table, edit.changes + paths = [str(c.get("path")) for c in changes if c.get("path")] marked = 0 labels = {str(c.get("path")): c for c in changes} for run_id, items in by_run.items(): @@ -1128,17 +1317,15 @@ async def catalog_changed(source_id: str, table: str, changes: Sequence[Mapping[ for snap, frozen in items: if isinstance(snap.get("sql"), str): errors += new_errors(snap["sql"], frozen, current, source.kind) - match = {"kind": kind, "source_id": source_id, "table": table, - **({"catalog_version": catalog_version} if catalog_version is not None else {}), - **({"change": change_id} if change_id else {})} + match = {"kind": edit.kind, "source_id": source.id, "table": table, + **({"catalog_version": edit.catalog_version} if edit.catalog_version is not None else {}), + **({"change": edit.change_id} if edit.change_id else {})} async with SessionLocal() as session: if await _basis_marked(session, run_id, match): continue data = {**match, "source": source.name, "items": [{"path": p, "label": labels.get(p, {}).get("label"), "change": labels.get(p, {}).get("change")} for p in seen], "new_errors": len(errors), "checks": errors[:10]} - from app.engine.run_append import append_events - written = await append_events(run_id, [(EventType.LEDGER_BASIS_CHANGED, None, data)], require="sealed") if not written: continue @@ -1159,20 +1346,48 @@ async def _open_basis_task(run_id: str, seq: int) -> list[str]: return await open_tasks(run, reason="catalog", owners=owners, title=CATALOG_TITLE, extra=str(seq)) +#: 审阅之后排着的目录改动：{事件循环: {数据源: [CatalogEdit]}}。一个数据源一个后台任务（_drain_catalog），批量审阅逐表写完 +#: 逐表调 after_review，这期间（不让出事件循环）进来的几张表合成一遍做；做的时候又进来的，做完这一遍接着做 +_catalog_queues: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, list[CatalogEdit]]]" = \ + weakref.WeakKeyDictionary() + + def after_review(source_id: str, table: str, before: Mapping[str, Any] | None, after: Mapping[str, Any] | None, version: int | None) -> int: - """审阅写入之后（data/catalog、catalog_review 调）：有影响数字的项被驳回或改了，在后台标依据有变。返回改了几项。 - 出错只记日志：审阅已经写进去了。""" + """审阅写入之后（data/catalog、catalog_review 调）：有影响数字的项被驳回或改了，排进这个数据源的后台任务标依据有变（同一个 + 数据源的几张表合成一遍）。返回改了几项。出错只记日志：审阅已经写进去了。""" try: changes = fact_changes(before, after) if changes: - spawn(catalog_changed(source_id, table, changes, catalog_version=version)) + queues = _catalog_queues.setdefault(asyncio.get_running_loop(), {}) + edit = CatalogEdit(table=table, changes=changes, catalog_version=version) + if source_id in queues: + queues[source_id].append(edit) + else: + queues[source_id] = [edit] + spawn(_drain_catalog(source_id)) return len(changes) except Exception: # noqa: BLE001 - 辅助步骤 logger.exception("审阅写入后排队标依据有变时出错（%s · %s）", source_id, table) return 0 +async def _drain_catalog(source_id: str) -> None: + """一个数据源排着的目录改动一遍一遍做完。一遍出错只记日志，接着做后面排进来的。""" + queues = _catalog_queues.setdefault(asyncio.get_running_loop(), {}) + try: + while edits := queues.get(source_id): + queues[source_id] = [] + try: + await catalog_pass(source_id, edits) + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 - 辅助步骤 + logger.exception("审阅之后标依据有变时出错（%s）", source_id) + finally: + queues.pop(source_id, None) + + # ========================================================================== # 三、表结构变动 # ========================================================================== @@ -1222,7 +1437,8 @@ async def _mark_structure(change: Mapping[str, Any]) -> int: return 0 detected = change.get("detected_at") at = datetime.fromisoformat(detected) if isinstance(detected, str) else None - found = await _explore_snapshots(session, source, str(change["table"]), before=at) + table = str(change["table"]) + found = (await _explore_snapshots(session, source, [table], before=at)).get(table) or [] marked = 0 for run_id in dict.fromkeys(r for r, _ in found): match = {"kind": "structure", "change": change.get("id")} @@ -1459,9 +1675,10 @@ async def shutdown() -> None: await asyncio.gather(*tasks, return_exceptions=True) -__all__ = ["BASIS_KINDS", "BATCHES", "CURSOR_KEY", "OUTCOMES", "ORIGIN", "REASONS", "TASK_KIND", "Group", "Pair", "Plan", - "RecallRefused", "after_publish", "after_review", "automatic", "basis_kind", "batches", "build_graph", "build_plan", - "card_values", "catalog_changed", "compare", "consume_source_changes", "drain", "fact_changes", "launch", - "new_errors", "normalize", "open_tasks", "original_values", "outcome_of", "preview", "process_change", - "recall_card", "recall_runs", "reissue", "reissue_action", "resume", "schedule_settle", "settle", "shutdown", +__all__ = ["BASIS_KINDS", "BATCHES", "CURSOR_KEY", "OUTCOMES", "ORIGIN", "REASONS", "TASK_KIND", "CatalogEdit", + "Group", "Pair", "Plan", "RecallRefused", "after_publish", "after_review", "answer_marks", "automatic", + "basis_kind", "batch_of_key", "batches", "build_graph", "build_plan", "card_values", "catalog_changed", + "catalog_pass", "compare", "consume_source_changes", "drain", "fact_changes", "launch", "new_errors", + "normalize", "open_tasks", "original_values", "outcome_of", "preview", "process_change", "recall_card", + "recall_runs", "reissue", "reissue_action", "resume", "schedule_settle", "settle", "shutdown", "source_cards", "spawn", "start_resume", "task_payload", "version_recall_counts"] diff --git a/backend/app/ledger/stats.py b/backend/app/ledger/stats.py index 9f5e4832..180d984a 100644 --- a/backend/app/ledger/stats.py +++ b/backend/app/ledger/stats.py @@ -11,8 +11,8 @@ 的查询里各自的比率；证明等价的绑定是在哪一级配上的；没证明的原因（不可比、参数绑不上、配上了但不等价）；不等价的差异类型 分布（time / filter / aggregate / join / other / parse，每份口径的差异每类只计一次）；逐个数的标签各几个、整题等价几次。 - E2 对账：对账运行（origin=reconcile，第 3 波起由 ledger/reconcile 发起）发起、完成、失败各几次，对过账的答案几个，其中参数 - 经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比。「说明有用 / 看不懂」的反馈要等对账面板 - （第 4 波）记成对账运行上的封存后事件，这之前照实写没有数据。 + 经人核对过的、对照的指标由人选的各几次；相关口径的差异说明里兜底说明的占比；对账结果上「说明有用 / 看不懂」的反馈 + （第 4 波，ledger/feedback：对账运行上的封存后事件 ledger.feedback，每人只算最后一次）各几个、有用的比率。 - E3 重复度：问数据运行里执行过的查询按（数据源、normalize_sql 文本）分组，出现在 ≥ REPEAT_RUNS 次运行、来自 ≥ REPEAT_CONVERSATIONS 段对话的算重复的形状，给出这样的形状几个、有重复形状的运行占几成；问题聚类（app/chat/questions：全站唯一的聚类，只取问题 原文、不含 history、排除系统派生的运行）里 ≥ REPEAT_RUNS 个问题的类占几成。认证队列的转化（第 3 波，ledger/certify）：固化成 @@ -42,8 +42,6 @@ ASPECT_GROUP = {"time": "time", "time_range": "time", "filter": "filter", "aggre "group_by": "aggregate", "join": "join", "tables": "join", "parse": "parse"} DIFF_GROUPS = ("time", "filter", "aggregate", "join", "other", "parse") STATUSES = ("proven", "composed", "related", "none", "skipped") -#: 没有数据时的说法（这一块的功能还没做） -NOT_YET_RECONCILE = "对账面板的反馈在第 4 波接上，目前没有数据" #: 聚类最多取几个问题（同治理接口的上限） QUESTION_LIMIT = 500 @@ -154,8 +152,11 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic stmt = stmt.where(Run.created_at >= scope["since"]) if scope.get("until") is not None: stmt = stmt.where(Run.created_at < scope["until"]) + from app.ledger import feedback as ledger_feedback + by_status = {s: int(n) for s, n in (await session.execute(stmt.group_by(Run.status))).all()} origins = [o for (o,) in (await session.execute(stmt.with_only_columns(Run.origin))).all() if isinstance(o, dict)] + reconcile_ids = [r for (r,) in (await session.execute(stmt.with_only_columns(Run.id))).all()] answers = {o.get("source_run") for o in origins if o.get("source_run")} explained = fallback = 0 for content in contents: @@ -170,7 +171,7 @@ async def _e2(session: Any, contents: list[dict[str, Any]], **scope: Any) -> dic "pairs_chosen": sum(1 for o in origins if o.get("pairs_chosen")), "explained": explained, "fallback": fallback, "fallback_rate": _ratio(fallback, explained), - "feedback": {"available": False, "note": NOT_YET_RECONCILE}} + "feedback": await ledger_feedback.counts(session, reconcile_ids)} #: 查询快照 → (数据源, 形状)。工件按内容寻址、不会变，记下来不会过期 diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index d1c8f356..640731b3 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -6,7 +6,11 @@ ledger 键、报告导出的片段面板，三处读同一份比对工件、按 界面层不放逻辑：标签、整题判断、能做的动作都由这里给。actions（第 3 波）：整题等价时按那份口径正式出具（能不能、能不能 自动、不能的原因，ledger/issue.plan_issue，按看的这个人的权限判）；有相关口径的查询看差多少（口径对账，ledger/reconcile）。 -issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态）。 +issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态，第 4 波起带上结果上的「说明有用 / +看不懂」：看的这个人给过的、各几个，ledger/feedback）。 + +recall（第 4 波，ledger/recall.answer_marks）：这个答案身上的召回标记——「已按 vN 核对，不影响」（checked）、依据有变 +（basis）、看的这个人手上说的是它的 recall 待办（tasks）。几种状态都带：没比对过、不比对的答案照样可能被标依据有变。 """ from __future__ import annotations @@ -108,11 +112,16 @@ async def run_view(run_id: str) -> dict[str, Any]: from app.api.coded import CodedHTTPException from app.api.evidence import RUN_NOT_FOUND, _sealed + from app.api.deps import current_principal + from app.ledger.recall import answer_marks + async with SessionLocal() as session: run = await session.get(Run, run_id) - if run is None: - raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND) - base: dict[str, Any] = {"run_id": run_id, "note": ledger_labels.POST_SEAL_NOTE} + if run is None: + raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND) + principal = current_principal() + marks = await answer_marks(session, run_id, principal) + base: dict[str, Any] = {"run_id": run_id, "note": ledger_labels.POST_SEAL_NOTE, "recall": marks} recorded = await store.latest(run_id) if recorded is None: why = ledger_match.ineligible(run) @@ -125,7 +134,7 @@ async def run_view(run_id: str) -> dict[str, Any]: return {**base, "state": "pending"} if recorded.content is None: return {**base, "state": "unreadable", "reason": recorded.problem, **_stamp(recorded)} - from app.api.deps import current_principal + from app.ledger import feedback as ledger_feedback from app.ledger import issue as ledger_issue from app.ledger import reconcile as ledger_reconcile @@ -133,12 +142,14 @@ async def run_view(run_id: str) -> dict[str, Any]: sealed = await _sealed(run_id) hider = await _hider(sealed) names = await _names(_workflow_ids(content)) - principal = current_principal() plans = await ledger_issue.plans_for(run_id, content, principal=principal, hide=hider.hide, artifact=recorded.artifact, retro=recorded.retro) async with SessionLocal() as session: issued = [ledger_issue.issued_view(r) for r in await ledger_issue.issued_runs(session, run_id)] reconciled = [ledger_reconcile.reconciled_view(r) for r in await ledger_reconcile.reconciled_runs(session, run_id)] + # 对账结果上的「说明有用 / 看不懂」（ledger/feedback）：看的这个人给过哪一个、各几个 + votes = await ledger_feedback.feedback_of(session, [r["run_id"] for r in reconciled], principal) + reconciled = [{**r, "feedback": votes[r["run_id"]]} for r in reconciled] return { **base, "state": "matched", **_stamp(recorded), "calibers": [{**{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level", "graph_hash")}, diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index b8bfeac4..5490dacd 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -347,7 +347,12 @@ async def notify_recall(task_id: str, users: list[str]) -> int: 自己的个人渠道；然后照别的待办推 task.opened（已经推过 ledger.recall 的渠道不再推，同一个渠道只推一次）。目录、表结构变了 的那两种只推 task.opened。 - 消息照模块说明的规矩（X6）：口径名（搭建时起的工作流名）、新旧版本号、这条待办列了几个答案、召回编号、去处理的链接；不放 + 全局渠道按这一批待办只推一次（同一次召回运行、同一个来由开的那几条，recall.batch_of_key）：开登录时一组答案有几个主人就 + 开几条待办，团队群却只该收到一条——只在这一批里最早开的那条（开出时间、id 排序，几条并发推也认同一条）上推全局渠道， + 答案、待办按这一批汇总，链到待办箱（各人在那里看到派给自己的那条）；别的几条连 task.opened 也不再推全局渠道。个人渠道 + 照旧每个主人各一条，只数他的答案、链到他的那条待办。 + + 消息照模块说明的规矩（X6）：口径名（搭建时起的工作流名）、新旧版本号、几个答案（几条待办）、召回编号、去处理的链接；不放 旧口径的值、新口径的值、差值和原答案的值。返回推了几个渠道。""" from app.ledger import recall @@ -361,26 +366,51 @@ async def notify_recall(task_id: str, users: list[str]) -> int: if task is None: return 0 run = await session.get(Run, task.run_id) - reason = (task.correlation_key or "").split(":", 1)[0] + key = task.correlation_key or "" + reason = key.split(":", 1)[0] + batch = recall.batch_of_key(key) + siblings = [t for t, k in (await session.execute( + select(Task.id, Task.correlation_key).where(Task.run_id == task.run_id, Task.kind == recall.TASK_KIND) + .order_by(Task.created_at, Task.id))).all() if recall.batch_of_key(k or "") == batch] + first = not siblings or siblings[0] == task_id channels = await _channels(session, event, run, users) if reason == "caliber" else [] + # 不是这一批最早的那条：全局渠道已经（或正由那一条）推过了，这条只推个人渠道，task.opened 也跳过全局渠道 + shared = [] if first else list((await session.execute(select(NotifyChannel.id).where( + NotifyChannel.scope == "global"))).scalars()) + channels = [c for c in channels if first or c.scope != "global"] origin = run.origin if run is not None and isinstance(run.origin, dict) else {} name = str(await session.scalar(select(Workflow.name).where(Workflow.id == origin.get("workflow_id"))) or "") if origin.get("workflow_id") else "" base = await _base_url(session) if channels else None pushed = 0 if channels: - answers = len(recall._uses_of(origin, recall._owner_of_key(task.correlation_key or ""))) - lines = [("状态", "待处理"), ("口径", f"「{name[:80]}」"), - ("版本", f"v{origin.get('from_version')} → v{origin.get('to_version')}"), ("答案", f"{answers} 个"), - ("召回", f"#{task.run_id[:6]}")] - message = {"event": event, "status": "pending", "ok": False, - "title": recall.CHANGED_TITLE.format(name=name[:80] or "口径", version=origin.get("to_version")), - "lines": lines, "links": [("去处理", f"{base}/apps/inbox/{task_id}")] if base else [], - "subject": {"task_id": task_id, "run_id": task.run_id, "workflow_id": origin.get("workflow_id")}} - pushed = await _push(channels, message, subject_kind="task", subject_id=task_id, event=event, - run_id=task.run_id, task_id=task_id) + head = [("状态", "待处理"), ("口径", f"「{name[:80]}」"), + ("版本", f"v{origin.get('from_version')} → v{origin.get('to_version')}")] + tail = [("召回", f"#{task.run_id[:6]}")] + mine = len(recall._uses_of(origin, recall._owner_of_key(key))) + everyone = len(recall._uses_of(origin, ("anyone", ""))) + + def message(answers: int, tasks: int, link: tuple[str, str] | None) -> dict[str, Any]: + return {"event": event, "status": "pending", "ok": False, + "title": recall.CHANGED_TITLE.format(name=name[:80] or "口径", version=origin.get("to_version")), + "lines": [*head, ("答案", f"{answers} 个"), *([("待办", f"{tasks} 条")] if tasks > 1 else []), *tail], + "links": [link] if base and link else [], + "subject": {"task_id": task_id, "run_id": task.run_id, "workflow_id": origin.get("workflow_id")}} + + own = ("去处理", f"{base}/apps/inbox/{task_id}") + group = [c for c in channels if c.scope == "global"] + if group: + # 这一批只有这一条（关登录时派给任何人）就链到它；几条的链到待办箱，各人看派给自己的那条 + many = len(siblings) > 1 + pushed += await _push(group, message(everyone, len(siblings), ("去待办箱", f"{base}/apps?tab=inbox") + if many else own), + subject_kind="task", subject_id=task_id, event=event, run_id=task.run_id, + task_id=task_id) + if personal := [c for c in channels if c.scope != "global"]: + pushed += await _push(personal, message(mine, 1, own), subject_kind="task", subject_id=task_id, + event=event, run_id=task.run_id, task_id=task_id) return pushed + await notify_task(task_id, "task.opened", users=users, - exclude=frozenset(c.id for c in channels)) + exclude=frozenset(c.id for c in channels) | frozenset(shared)) async def notify_run(run_id: str, event: str) -> int: diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py index 5a5a2001..aeedd136 100644 --- a/backend/tests/test_ledger_api.py +++ b/backend/tests/test_ledger_api.py @@ -233,7 +233,8 @@ async def test_stats_count_what_the_fixtures_did(client, monkeypatch): assert e1["with_candidates"] == 3 and e1["rates"]["proven"] == round(1 / 3, 4) assert e1["diff_aspects"]["filter"] == 1 and e1["labels"]["caliber"] == 1 and e1["labels"]["unknown"] == 1 assert e1["whole_equivalent"] == 1 - assert body["e2"]["launched"] == 0 and body["e2"]["feedback"]["available"] is False + assert body["e2"]["launched"] == 0 and body["e2"]["feedback"] == {"available": True, "useful": 0, "unclear": 0, + "runs": 0, "useful_rate": None} assert body["e3"]["chat_runs"] == 0 and body["e3"]["threshold"] == {"runs": 3, "conversations": 2} usage = {u["caliber"]: u for u in body["e4"]["usage"]} assert usage[f"{wf}@1"]["proven_runs"] == 1 diff --git a/backend/tests/test_ledger_feedback.py b/backend/tests/test_ledger_feedback.py new file mode 100644 index 00000000..805ead19 --- /dev/null +++ b/backend/tests/test_ledger_feedback.py @@ -0,0 +1,119 @@ +"""对账结果的「说明有用 / 看不懂」（next-round 02 做法 10 的 E2；app/ledger/feedback.py）。 + +真的跑：虚构店铺库里的问数据答案（算上了退款）和受管口径「周销售」（只算已支付）对一次账，再在结果上给反馈： +- 反馈记成对账运行上的封存后事件 ledger.feedback（批注不是证据，不进封存范围）；同一个人改主意只算最后一次； +- 账本接口 reconciled 每一项带上反馈（看的这个人给过的、各几个），界面照着画选中的那一个； +- 统计 E2 数到它（不再写「第 4 波接上」）； +- 不认的反馈、不是这个答案的对账、还没跑完的对账照实拒绝。 +数据都是编的。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.core.events import EventType +from app.db.base import SessionLocal +from app.db.models import RunEvent +from app.engine.runner import run_manager +from app.ledger import match as ledger_match +from app.main import app +from tests import ledger_samples as S + +CHECKED = {"start": "2026-09-01", "end": "2026-09-08"} + + +@pytest.fixture(autouse=True) +async def engine_up(tmp_path): + await run_manager.setup() + path = tmp_path / "shop.db" + await S.fresh_source(path) + yield path + await ledger_match.drain() + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def reconciled(client, monkeypatch) -> tuple[str, str]: + """问一次（算上了退款）、按核对过的参数和「周销售」对一次账，等它跑完。返回 (答案, 对账运行)。""" + wf = await S.publish() + row = await S.explore(monkeypatch, S.WITH_REFUNDS) + body = (await client.get(f"/api/runs/{row.id}/ledger")).json() + [action] = [a for a in body["actions"] if a["kind"] == "reconcile"] + res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", + json={"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, + "confirm": True}) + assert res.status_code == 201, res.text + done = await S.wait(res.json()["run_id"]) + assert done.status == "succeeded", done.error + return row.id, done.id + + +async def test_feedback_is_a_post_seal_note_counted_by_e2(client, monkeypatch): + answer, rec = await reconciled(client, monkeypatch) + before = (await client.get("/api/ledger/stats")).json()["e2"]["feedback"] + assert before["available"] is True and "第 4 波" not in str(before) + + res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "useful"}) + assert res.status_code == 200, res.text + assert res.json() == {"run_id": rec, "feedback": {"mine": "useful", "useful": 1, "unclear": 0}} + async with SessionLocal() as session: + [data] = (await session.execute(select(RunEvent.data).where( + RunEvent.run_id == rec, RunEvent.type == EventType.LEDGER_FEEDBACK))).scalars() + assert data["verdict"] == "useful" and data["post_seal"] is True and data["source_run"] == answer + + # 改主意：同一个人只算最后一次 + res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "unclear"}) + assert res.json()["feedback"] == {"mine": "unclear", "useful": 0, "unclear": 1} + [item] = (await client.get(f"/api/runs/{answer}/ledger")).json()["reconciled"] + assert item["run_id"] == rec and item["feedback"] == {"mine": "unclear", "useful": 0, "unclear": 1} + + after = (await client.get("/api/ledger/stats")).json()["e2"]["feedback"] + assert after["useful"] - before["useful"] == 0 and after["unclear"] - before["unclear"] == 1 + assert after["runs"] - before["runs"] == 1 and after["available"] is True + + +async def test_feedback_refusals(client, monkeypatch): + answer, rec = await reconciled(client, monkeypatch) + bad = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "great"}) + assert bad.status_code == 422 + other = await S.explore(monkeypatch, S.ELSEWHERE, text="门店 [[v:Q1.r0.n]] 家。") + res = await client.post(f"/api/ledger/answers/{other.id}/reconcile/{rec}/feedback", json={"verdict": "useful"}) + assert res.status_code == 404 and res.json()["code"] == "ledger_feedback_refused", "不是这个答案的对账" + res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{answer}/feedback", json={"verdict": "useful"}) + assert res.status_code == 404, "答案自己不是对账运行" + # 还没封存的对账（正在跑、刚发起）：没有结果可评 + from app.db.models import Run + + async with SessionLocal() as session: + row = await session.get(Run, rec) + sealed = row.manifest_hash + row.manifest_hash = None + await session.commit() + try: + res = await client.post(f"/api/ledger/answers/{answer}/reconcile/{rec}/feedback", json={"verdict": "useful"}) + assert res.status_code == 409 and "还没运行完" in res.json()["detail"], res.text + finally: + async with SessionLocal() as session: + (await session.get(Run, rec)).manifest_hash = sealed + await session.commit() + + +def test_offline_page_and_summary_know_the_feedback_event(): + import re + from pathlib import Path + + from app.bundle.build import _summary + + script = (Path(__file__).resolve().parents[1] / "app" / "bundle" / "verifier.js").read_text(encoding="utf-8") + block = re.search(r"const EVENT_LABEL = \{(.*?)\n \}", script, re.S) + labels = dict(re.findall(r"'([a-z_.]+)': '([^']+)'", block.group(1))) + assert labels["ledger.feedback"] == "对账说明的反馈" + assert _summary("ledger.feedback", {"verdict": "useful"}) == "反馈：说明有用" + assert _summary("ledger.feedback", {"verdict": "unclear"}) == "反馈：看不懂" diff --git a/backend/tests/test_ledger_recall.py b/backend/tests/test_ledger_recall.py index 867d09d5..0052f306 100644 --- a/backend/tests/test_ledger_recall.py +++ b/backend/tests/test_ledger_recall.py @@ -7,7 +7,8 @@ - 召回运行：派生图里新旧两版的取数和口径卡各一份（前缀不撞），召回卡里算两版之差；同一份数据上两版各算一次； - 收尾：数变了才开 recall 待办（挂在召回运行上，旧值、新值、差值、原答案当时的值都给），推 ledger.recall，消息里没有数； 数没变的在原答案上记 ledger.recalled（已按 vN 核对，不影响），不开待办； -- 不能重算的照实记：新版入口要的参数这一组里没有、图里有人工节点（有副作用）； +- 不能重算的照实记：新版入口要的参数这一组里没有、图里会写外部系统；图里别处有人工节点的照样重算（召回图碰不到它）， + 只是「按新版重新出具」不代人审批； - 待办「按新版重新出具」走按口径出具同一套判法、「知道了」； - 目录里影响数字的一项被改了：冻结目录里看得到它的答案标「依据有变」，按新目录多出错误级问题的才开待办； - 表结构变了：用过这张表的答案标「依据的表结构已变」，口径试算算不出来才开待办； @@ -146,6 +147,32 @@ async def test_limit_continue_and_idempotent_replanning(monkeypatch): assert await recall.launch(wf, 2, batch="manual") == [] +async def test_superseded_version_offers_no_continue(client, monkeypatch): + """v2 自动算了一批、还剩几组没算，接着又发布了 v3：v2 那一版不再给「接着算」（v3 的召回覆盖更早的所有版本，同启动时 + 接着做只认当前发布版），写明去 v3 那一版看；手动点也照样拒绝，不再开「口径升到 v2」的待办。""" + await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=1) + wf = await S.publish() + await answer(monkeypatch) + await answer(monkeypatch, "2026-09-08", "2026-09-15") + await publish_v2(wf) + assert len(await run_recall(wf)) == 1 + async with SessionLocal() as session: + view = await recall.batches(session, wf, 2) + assert view["remaining"] == 1 and view["can_continue"] is True and view["continue_reason"] is None + await S.publish(graph=S.caliber_graph(WITH_REFUNDED), version=3, workflow_id=wf) + async with SessionLocal() as session: + view = await recall.batches(session, wf, 2) + assert view["remaining"] == 1 and view["can_continue"] is False + assert view["continue_reason"] == "已有更新的发布版 v3，召回在 v3 那一版接着看" + res = await client.post(f"/api/ledger/recall/{wf}/2/continue") + assert res.status_code == 409 and res.json()["detail"] == view["continue_reason"], res.text + assert await recall.launch(wf, 2, batch="manual") == [], "直接调也不再发起" + async with SessionLocal() as session: + assert len(await recall.recall_runs(session, wf, 2)) == 1 + current = await recall.batches(session, wf, 3) + assert current["can_continue"] is True and current["continue_reason"] is None + + async def test_zero_limit_lists_but_does_not_recompute(monkeypatch): await ledger_settings(auto_issue=False, issue_daily_limit=100_000, recall_limit=0) wf = await S.publish() @@ -189,6 +216,9 @@ async def test_changed_numbers_open_a_recall_task_with_all_three_values(client, assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "数变了的不在原答案上记「不影响」" [task] = await recall_tasks(done.id) assert task.title == "口径「周销售」升到 v2，答案的数变了" + # 问数据的轮次、证据面板读的账本接口：这个答案有一条没处理的召回待办（界面链到待办箱），没有「已核对」 + marks = (await client.get(f"/api/runs/{a1.id}/ledger")).json()["recall"] + assert marks == {"checked": [], "basis": [], "tasks": [{"id": task.id, "reason": "caliber", "to_version": 2}]} res = await client.get(f"/api/tasks/{task.id}") assert res.status_code == 200, res.text payload = res.json()["payload"] @@ -220,11 +250,43 @@ async def test_unchanged_numbers_are_only_booked_on_the_answer(monkeypatch): [booked] = await events(a1.id, EventType.LEDGER_RECALLED) assert booked == {"role": "answer", "changed": False, "recall_run": done.id, "workflow_id": wf, "from_version": 1, "to_version": 2, "post_seal": True} + # 账本接口带上「已按 v2 核对，不影响」：问数据的轮次、证据面板不用再去翻运行时间线 + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + marks = (await c.get(f"/api/runs/{a1.id}/ledger")).json()["recall"] + [checked] = marks["checked"] + assert {k: checked[k] for k in ("workflow_id", "workflow_name", "from_version", "to_version", "changed", + "recall_run")} == {"workflow_id": wf, "workflow_name": "周销售", "from_version": 1, + "to_version": 2, "changed": False, "recall_run": done.id} + assert checked["at"] and marks["basis"] == [] and marks["tasks"] == [] # 再收一次尾不会重复记 assert await recall.settle(done.id) is None assert len(await events(a1.id, EventType.LEDGER_RECALLED)) == 1 +@pytest.mark.parametrize("values", [{}, {"old_0": {"value": 1, "rendered": "1.00元"}}], ids=["none", "half"]) +async def test_unreadable_recall_card_is_a_failure_not_unchanged(monkeypatch, values): + """召回运行成功了、召回卡却读不出来（封存里找不到 recall 节点、读不出封存，或少了某个比较的旧值 / 新值）：照实记重算 + 失败，不当成「数没变」去给原答案盖「已核对，不影响」，也不开待办。""" + wf = await S.publish() + a1 = await answer(monkeypatch) + await publish_v2(wf) + + async def unreadable(run_id): + return dict(values) + + monkeypatch.setattr(recall, "card_values", unreadable) + [done] = await run_recall(wf) + assert done.status == "succeeded", done.error + [settled] = [d for d in await events(done.id, EventType.LEDGER_RECALLED) if d.get("role") == "recall"] + assert settled["outcome"] == "failed" and settled["changed"] is None and settled["error"] == "读不出召回卡" + assert await events(a1.id, EventType.LEDGER_RECALLED) == [], "读不出的不在原答案上记「不影响」" + assert await recall_tasks(done.id) == [] + async with SessionLocal() as session: + view = await recall.batches(session, wf, 2) + [group] = view["groups"] + assert group["state"] == "failed" and group["reason"] == "重算失败：读不出召回卡" + + async def test_incompatible_entry_and_side_effects_are_not_recomputed(monkeypatch): wf = await S.publish() await answer(monkeypatch) @@ -240,14 +302,59 @@ async def test_incompatible_entry_and_side_effects_are_not_recomputed(monkeypatc assert view["counts"]["not_recomputable"] == 1 and view["groups"][0]["reason"] == group.problem assert view["can_continue"] is False - wf2 = await S.publish(name="周销售（要审批）") + + +async def test_human_node_elsewhere_still_recomputes_but_reissue_refuses(client, monkeypatch): + """新版图里在口径卡下游有审批（取数 → 口径卡 → 报告 → 人工确认 → 出口）：召回图只有两版的取数和口径卡，人工节点碰不到， + 照样重算、数变了照样开待办（做法 7 不重算的只有三种：缺入口字段、数据部分不能复用、会写外部系统）；「按新版重新出具」 + 走完整的正式运行，仍然不代人审批，照实写不能出具。""" + wf = await S.publish(name="周销售（要审批）") await answer(monkeypatch) human = S.caliber_graph(WITH_REFUNDED) - human["nodes"].insert(3, S.node("ok", "human", mode="approve", title="确认")) - await S.publish(graph=human, version=2, workflow_id=wf2) + human["nodes"].insert(4, S.node("ok", "human", mode="approve", title="确认")) + human["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(human["nodes"], human["nodes"][1:])] + await S.publish(graph=human, version=2, workflow_id=wf) async with SessionLocal() as session: - plan = await recall.build_plan(session, wf2, 2) - assert plan.groups and all("人工节点" in (g.problem or "") for g in plan.groups) + plan = await recall.build_plan(session, wf, 2) + assert plan.groups and all(g.problem is None for g in plan.groups), [g.problem for g in plan.groups] + assert plan.version_problem == {} + [done] = await run_recall(wf) + assert done.status == "succeeded", done.error + assert not [n for n in done.graph["nodes"] if n["type"] == "human"], "召回图里没有人工节点" + [task] = await recall_tasks(done.id) + payload = (await client.get(f"/api/tasks/{task.id}")).json()["payload"] + reissue = next(a for a in payload["actions"] if a["decision"] == "reissue") + assert reissue["available"] is False and "人工节点" in reissue["reason"], reissue + res = await client.post(f"/api/tasks/{task.id}/respond", json={"decision": "reissue"}) + assert res.status_code == 409 and "人工节点" in res.json()["detail"], res.text + + +async def test_writing_to_an_external_system_is_not_recomputed(monkeypatch): + """新版图里调用工具节点绑的能力会写外部系统（capabilities.registry.writes_externally，同按新口径补算、按口径出具）： + 照实记「无法重算」，不发起。""" + from types import SimpleNamespace + + from app.capabilities import registry + + wf = await S.publish() + await answer(monkeypatch) + graph = S.caliber_graph(WITH_REFUNDED) + graph["nodes"].insert(4, S.node("pay", "tool", tool="erp_pay", args={})) + graph["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(graph["nodes"], graph["nodes"][1:])] + await publish_v2(wf, graph=graph) + real = registry.get_capability + + async def get_capability(session, name): + if name == "erp_pay": + return SimpleNamespace(name=name, effect="write", effect_system="write", endpoint_id=None) + return await real(session, name) + + monkeypatch.setattr(registry, "get_capability", get_capability) + async with SessionLocal() as session: + plan = await recall.build_plan(session, wf, 2) + assert plan.groups and all("会写外部系统（erp_pay）" in (g.problem or "") for g in plan.groups), \ + [g.problem for g in plan.groups] + assert await recall.launch(wf, 2) == [] async def test_plan_from_formal_runs_and_exporters_as_owners(monkeypatch): @@ -362,6 +469,74 @@ async def test_recall_notification_has_no_numbers(monkeypatch): assert "周销售" in mine[0]["title"] and "v2" in mine[0]["title"] +async def test_recall_pushes_the_shared_channel_once_per_recall_run(monkeypatch): + """开登录时一组答案有几个主人就开几条待办，但全局渠道（团队群）按召回运行只收一条（答案、待办都按这次召回汇总，链到 + 待办箱）；订阅了 task.opened 的全局渠道也不再按每条待办各推一次。每个主人自己的个人渠道各收他那一条。""" + from app.core.crypto import encrypt + from app.db.models import NotifyChannel, User + from app.notify import service + + sent: list[dict] = [] + + async def capture(channel, message): + sent.append({"channel": channel.name, **message}) + from app.notify.core import Outcome + + return Outcome(True, 200, 1, None) + + monkeypatch.setattr("app.notify.send", capture) + wf = await S.publish() + first = await answer(monkeypatch) + second = await answer(monkeypatch) # 同一周：同一组，两个主人 + made: list[str] = [] + async with SessionLocal() as session: + for name, run_id in (("analyst_a", first.id), ("analyst_b", second.id)): + user = User(username=name, role="user") + session.add(user) + await session.flush() + (await session.get(Run, run_id)).started_by = name + channel = NotifyChannel(name=f"个人-{name}", scope="user", owner_id=user.id, format="generic", + url=encrypt(f"https://hooks.example.com/{name}"), events=["ledger.recall"]) + session.add(channel) + await session.flush() + made.append(channel.id) + for name, events_ in (("团队-召回", ["ledger.recall", "task.opened"]), ("团队-待办", ["task.opened"])): + channel = NotifyChannel(name=name, scope="global", url=encrypt("https://hooks.example.com/team"), + events=events_) + session.add(channel) + await session.flush() + made.append(channel.id) + await session.commit() + monkeypatch.setenv("AGENTLAB_AUTH_MODE", "local") + try: + await publish_v2(wf) + [done] = await run_recall(wf) + await service.drain() + finally: + async with SessionLocal() as session: + for channel_id in made: + await session.delete(await session.get(NotifyChannel, channel_id)) + await session.commit() + tasks = await recall_tasks(done.id) + assert len(tasks) == 2, "每个主人一条待办" + team = [m for m in sent if m["channel"] == "团队-召回"] + assert [m["event"] for m in team] == ["ledger.recall"], [m["event"] for m in team] + lines = dict(team[0]["lines"]) + assert lines["答案"] == "2 个" and lines["待办"] == "2 条", lines + assert [label for label, _ in team[0]["links"]] == ["去待办箱"] or team[0]["links"] == [] + assert [m["event"] for m in sent if m["channel"] == "团队-待办"] == ["task.opened"], "只订阅了新待办的团队渠道也只收一条" + for name in ("analyst_a", "analyst_b"): + mine = [m for m in sent if m["channel"] == f"个人-{name}"] + assert [m["event"] for m in mine] == ["ledger.recall"] and dict(mine[0]["lines"])["答案"] == "1 个", mine + # 账本接口里的召回待办只列看的这个人手上的：第一个答案的主人看得到自己那条，另一个主人看不到 + from app.api.deps import Principal + + async with SessionLocal() as session: + own = await recall.answer_marks(session, first.id, Principal(username="analyst_a", role="user", mode="local")) + other = await recall.answer_marks(session, first.id, Principal(username="analyst_b", role="user", mode="local")) + assert len(own["tasks"]) == 1 and own["tasks"][0]["reason"] == "caliber" and other["tasks"] == [] + + # -------------------------------------------------------------------------- # 目录事实被改、表结构变了 # -------------------------------------------------------------------------- @@ -411,6 +586,13 @@ async def test_catalog_fact_change_marks_answers_and_notifies_only_on_new_errors assert await events(unrelated.id, EventType.LEDGER_BASIS_CHANGED) == [], "没查过这张表的不标" [task] = await recall_tasks(row.id) assert task.correlation_key.startswith("catalog:") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + marks = (await c.get(f"/api/runs/{row.id}/ledger")).json()["recall"] + quiet = (await c.get(f"/api/runs/{unrelated.id}/ledger")).json()["recall"] + [basis] = marks["basis"] + assert {k: basis[k] for k in ("kind", "items", "new_errors")} == {"kind": "catalog", "items": 1, "new_errors": 1} + assert basis["at"] and marks["tasks"] == [{"id": task.id, "reason": "catalog"}] and marks["checked"] == [] + assert quiet == {"checked": [], "basis": [], "tasks": []} async with SessionLocal() as session: from app.api.deps import Principal @@ -427,6 +609,89 @@ async def test_catalog_fact_change_marks_answers_and_notifies_only_on_new_errors assert len(await events(row.id, EventType.LEDGER_BASIS_CHANGED)) == 1 +async def test_catalog_edits_on_one_source_share_one_pass(engine_up, monkeypatch): + """批量审阅一口气改了同一个数据源的几张表：只在后台做一遍（同一个数据源排着做，几张表合在一次里查），查询快照按数据源 + 在库里筛，冻结的表结构快照（大库的能有几 MB）同一份只读一次。""" + from collections import Counter + + from app.core import artifact_store + from app.data import catalog + + source_id = await S.set_source(engine_up) + await _catalog(source_id, {"business_date": _item({"column": "paid_at"}), + "columns": {"amount": {"measure": _item("flow")}}}) + async with SessionLocal() as session: + await catalog.write_entry(session, source_id, "stores", {"valid_filter": _item("id > 0")}, if_version=0, + actor="tester") + week1 = await answer(monkeypatch) + week2 = await answer(monkeypatch, "2026-09-08", "2026-09-15") + shops = await S.explore(monkeypatch, S.ELSEWHERE, text="门店 [[v:Q1.r0.n]] 家。", origin="chat") + scans: list[list[str]] = [] + loads: list[str] = [] + real_scan, real_load = recall._explore_snapshots, artifact_store.load + + async def scan(session, source, tables, **kw): + scans.append(sorted(tables)) + return await real_scan(session, source, tables, **kw) + + def load(artifact_id): + loads.append(artifact_id) + return real_load(artifact_id) + + monkeypatch.setattr(recall, "_explore_snapshots", scan) + monkeypatch.setattr(artifact_store, "load", load) + async with SessionLocal() as session: + orders = await catalog.read_entry(session, source_id, "orders") + stores = await catalog.read_entry(session, source_id, "stores") + rejected = {**orders.notes, "business_date": {**orders.notes["business_date"], "status": "rejected"}} + # 批量审阅逐表写完逐表调审阅之后的钩子（catalog.after_review → recall.after_review），中间不让出事件循环 + assert recall.after_review(source_id, "orders", orders.notes, rejected, orders.version + 1) == 1 + assert recall.after_review(source_id, "stores", stores.notes, + {"valid_filter": {**stores.notes["valid_filter"], "status": "rejected"}}, + stores.version + 1) == 1 + await recall.drain() + assert scans == [["orders", "stores"]], "同一个数据源的几张表合成一遍" + for run in (week1, week2): + [marked] = await events(run.id, EventType.LEDGER_BASIS_CHANGED) + assert marked["table"] == "orders" and marked["items"][0]["path"] == "business_date" + [marked] = await events(shops.id, EventType.LEDGER_BASIS_CHANGED) + assert marked["table"] == "stores" and marked["items"][0]["path"] == "valid_filter" + schemas = set() + for snapshot in {a for a in loads}: + content = real_load(snapshot) + if isinstance(content, dict) and content.get("schema_artifact"): + schemas.add(content["schema_artifact"]) + assert schemas and all(Counter(loads)[sid] == 1 for sid in schemas), Counter(loads) + + +async def test_snapshot_scan_pages_newest_first_and_stops_when_full(engine_up, monkeypatch): + """翻查询快照：只取这个数据源的、新的在前，每张表凑够 BASIS_MAX_RUNS 次运行就不再往下翻。""" + from app.db.models import DataSource + + source_id = await S.set_source(engine_up) + runs = [await answer(monkeypatch, f"2026-09-0{d}", "2026-09-08") for d in (1, 2, 3)] + pages: list[int] = [] + monkeypatch.setattr(recall, "BASIS_MAX_RUNS", 2) + monkeypatch.setattr(recall, "SNAPSHOT_PAGE", 1) + async with SessionLocal() as session: + source = await session.get(DataSource, source_id) + real_execute = session.execute + + async def counted(stmt, *args, **kwargs): + pages.append(1) + return await real_execute(stmt, *args, **kwargs) + + monkeypatch.setattr(session, "execute", counted) + found = await recall._explore_snapshots(session, source, ["ORDERS"]) + read = len(pages) + other = DataSource(id="elsewhere", name="elsewhere") + nothing = await recall._explore_snapshots(session, other, ["orders"]) + assert [r for r, _ in found["ORDERS"]] == [runs[2].id, runs[1].id], "新的在前、凑够两次运行就停" + # 一页一行：第三页读到第三次运行时两次已经凑够，停下，不再翻第四页 + assert read == 3, f"凑够了不再往下翻（翻了 {read} 页）" + assert nothing == {"orders": []}, "别的数据源的快照不读回来" + + def test_fact_changes_only_counts_items_that_affect_numbers(): before = {"valid_filter": _item("status <> 'open'"), "label": _item("订单"), "columns": {"amount": {"measure": _item("flow"), "label": _item("金额")}}}