From 906ec072ce9eb415e6eb701fd9d928d47503c76f Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 16:55:06 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8C=89=E5=8F=A3=E5=BE=84=E6=AD=A3=E5=BC=8F?= =?UTF-8?q?=E5=87=BA=E5=85=B7=E4=B8=8E=E5=8F=A3=E5=BE=84=E5=AF=B9=E8=B4=A6?= =?UTF-8?q?=EF=BC=88=E5=8F=A3=E5=BE=84=E8=B4=A6=E6=9C=AC=E5=81=9A=E6=B3=95?= =?UTF-8?q?=205=E3=80=816=EF=BC=8C=E6=8B=8D=E6=9D=BF=20L1=E3=80=81L5?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：问数据的答案整题等价于已发布口径时，要能按那一版发起一次完整走过门禁的正式运行（路径一、二a）；只是相关、不等价时，要能看「差多少」（路径二b）。两条路都不能让客户端冒充来源、自己传参数，也不能在 Python 里算差值。 正式出具（新增 ledger/issue.py）： - POST /api/ledger/answers/{run_id}/issue {workflow_id, version}：参数只从比对工件的绑定取，补入口缺省值再过 apps/forms 校验；经 launch_run 发起正式运行，来源 ledger_issue，origin 记 source_run、conversation_id、turn_id、binding_hash、口径和认法、是否自动；按 expect_hash 钉 graph_hash。 - 不能出具照实写原因：整题不等价、点名别的口径、版本退役或没发布、图里有人工节点或写入（apps/backfill._side_effect 抽成共用的 graph_side_effect）、缺入口字段、超过每日上限、权限（L5：使用者只能按他有权使用的应用固定着的版本出具，复用应用授权，origin 记 granted_by_app；别的写「请搭建者出具」）。 - 自动出具（L1）：比对完（只在运行一封存就比的那一次）整题独立等价、版本受管、设置开着、当天没超上限才自动；助手选用的只给一键。不自动的原因写进 actions 的 auto_reason。 - 同一运行、同一口径、同一组参数只发起一次（input_key 去重），进程内按（运行、工作流、版本）串行。 - 设置加 ledger 一组：自动出具开关、每日上限、比对时限（ledger/match 读它）。 口径对账（新增 ledger/reconcile.py，derive 扩 reconcile_graph / ours_block / assemble 的出口契约）： - POST /api/ledger/answers/{run_id}/reconcile：按那一版确定性派生只读图（这次的取数 ours_q1 → 口径的取数 → 口径卡 → 对账卡，出口契约 metrics_from，不加报告、不调模型），来源 reconcile 的探索运行；对账卡算这次、口径、差值、相对差。 - 参数是启发得来的、不全、含遮罩值时先让人核对（needs_check，没核对的 409）；指标式不是取数一格、差异有兜底时要人选对照哪个指标，选择记进 origin。 - 多行结果：被引用的行按维度列经合并查询 LEFT JOIN 对齐、按引用先后排，最多 10 行；维度是遮罩列、合并键不唯一照实拒绝。使用者发起带 MASK_RESULTS / READ_ONLY，搭建者也带 READ_ONLY。 - GET /api/runs/{id}/ledger 的 actions 补上出具、对账两类，issued、reconciled 列出发起过的运行。 测试：tests/test_ledger_issue.py（自动出具、只给一键的几种、拒绝原因、使用者的应用授权、并发只发一次、设置）、tests/test_ledger_reconcile.py（核对参数、派生图确定与契约、差值和相对差、出处链、要人选列、多行对齐、遮罩维度和重复键拒绝、使用者只读遮罩）；第 2 波 test_ledger_api 的 actions 断言按这一波改。 --- backend/app/api/ledger.py | 67 +++- backend/app/api/settings.py | 62 +++ backend/app/apps/backfill.py | 17 +- backend/app/ledger/derive.py | 62 ++- backend/app/ledger/issue.py | 372 ++++++++++++++++++ backend/app/ledger/match.py | 19 +- backend/app/ledger/reconcile.py | 518 +++++++++++++++++++++++++ backend/app/ledger/view.py | 19 +- backend/tests/test_ledger_api.py | 3 +- backend/tests/test_ledger_issue.py | 306 +++++++++++++++ backend/tests/test_ledger_reconcile.py | 216 +++++++++++ 11 files changed, 1637 insertions(+), 24 deletions(-) create mode 100644 backend/app/ledger/issue.py create mode 100644 backend/app/ledger/reconcile.py create mode 100644 backend/tests/test_ledger_issue.py create mode 100644 backend/tests/test_ledger_reconcile.py diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 1af77f15..814ee9b3 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -3,10 +3,15 @@ 不另建存储：ActionRecord 本身带着工作流、版本、graph_hash、能力版本、payload_hash、回执；「口径 → 应用」就是固定这一版的 应用版本。用在发布弹窗、退役确认、版本列表三处。 -口径账本（docs/design/next-round/02-口径账本.md）在这里加两个只读接口： +口径账本（docs/design/next-round/02-口径账本.md）在这里加的接口： - GET /api/runs/{run_id}/ledger（runs_router，和证据同一道门）：一次运行里每个数的口径标签、整题判断、每条查询的比对、能做的 - 动作（app/ledger/view.py）； -- GET /api/ledger/stats（router，搭建者）：账本自身给出的统计 E1–E4，从比对工件、运行、审计现算（app/ledger/stats.py）。 + 动作、出具过的正式运行、对过的账（app/ledger/view.py）； +- GET /api/ledger/stats（router，搭建者）：账本自身给出的统计 E1–E4，从比对工件、运行、审计现算（app/ledger/stats.py）； +- 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）； +- GET /api/ledger/certify-queue、POST /api/ledger/certify-queue/dismiss（router，搭建者，第 3 波）：可沉淀的问题、「不是口径」 + （app/ledger/certify.py）。这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。 """ from __future__ import annotations @@ -15,6 +20,7 @@ from typing import Any from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -100,3 +106,58 @@ async def ledger_stats(since: datetime | None = Query(default=None, description= origins = [o.strip() for o in (origin or "").split(",") if o.strip()] or None return await compute_stats(session, since=since, until=until, origins=origins, principal=current_principal()) + + +class IssueIn(BaseModel): + """按哪份口径出具：只能选口径，参数由服务端从比对工件里取（不收客户端传的参数）。""" + + workflow_id: str = Field(min_length=1, max_length=32) + version: int = Field(ge=1) + + +@answers_router.post("/{run_id}/issue") +async def issue_answer(run_id: str, payload: IssueIn) -> dict[str, Any]: + """按口径正式出具这次问数据的答案（app/ledger/issue.py）：经 launch_run 发起正式运行，来源 ledger_issue。同一运行、同一 + 口径、同一组参数只发起一次，再点交回已经发起的那一次（created 为假）。不能出具时回 409（整题不等价、版本退役、有人工 + 节点或写入、缺入口字段……）、403（权限不够：使用者只能按他有权使用的应用固定着的版本出具）、429（超过每日上限）， + detail 写明原因。""" + from app.api.coded import CodedHTTPException + from app.api.deps import current_principal + from app.ledger import issue as ledger_issue + + try: + run, created = await ledger_issue.issue(run_id, payload.workflow_id, payload.version, + principal=current_principal()) + except ledger_issue.IssueRefused as e: + raise CodedHTTPException(e.status_code, e.message, e.code) from e + return {"run_id": run.id, "created": created, "issued": ledger_issue.issued_view(run)} + + +class ReconcileIn(BaseModel): + """对哪条查询、和哪份相关口径对账。inputs + confirm：参数要核对时（动作里 needs_check）人核对、改过的参数；pairs：认不出 + 哪一列对哪个指标时人选的 [{ours: 这次的列, metric: 口径卡上的指标 id}]（动作里写明要不要选）。""" + + query: str = Field(min_length=1, max_length=64) + workflow_id: str = Field(min_length=1, max_length=32) + version: int = Field(ge=1) + inputs: dict[str, Any] | None = None + confirm: bool = False + pairs: list[dict[str, Any]] | None = None + + +@answers_router.post("/{run_id}/reconcile", status_code=201) +async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]: + """口径对账（app/ledger/reconcile.py）：按那一版口径确定性派生只读图，发起一次探索运行（来源 reconcile），差值在对账卡 + 里算。参数是推断出来的、不全的要先核对（409，code ledger_reconcile_check_inputs）；认不出列怎么配的要人选（409， + ledger_reconcile_needs_pairs）；多行时维度是遮罩列、键不唯一照实拒绝（409）。""" + from app.api.coded import CodedHTTPException + from app.api.deps import current_principal + from app.ledger import reconcile as ledger_reconcile + + try: + run = await ledger_reconcile.reconcile(run_id, ledger_reconcile.Request( + query=payload.query, workflow_id=payload.workflow_id, version=payload.version, inputs=payload.inputs, + confirm=payload.confirm, pairs=payload.pairs), principal=current_principal()) + except ledger_reconcile.ReconcileRefused as e: + raise CodedHTTPException(e.status_code, e.message, e.code) from e + return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)} diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index c45a685a..0ed08090 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -718,6 +718,61 @@ async def apps_settings(session: AsyncSession) -> dict[str, Any]: return apps_values(row.value if row else None) +# -------------------------------------------------------------------------- +# 口径账本：按口径自动正式出具的开关、每日上限，口径比对的时限（docs/design/next-round/02-口径账本.md 做法 5、2） +# -------------------------------------------------------------------------- + +#: 设置里 ledger 这一组。auto_issue：问数据的答案整题等价于受管口径（Agent 独立写的 SQL 被证明等价、版本受管、图里没有人工 +#: 节点和写入）时自动按口径发起一次正式运行（拍板 L1；不满足的只给一键）。issue_daily_limit：每天最多按口径正式出具几次 +#: （自动的、一键的合计，0 表示不再出具）：正式运行含报告撰写的模型费用。match_timeout_s：一次运行的口径比对最多算几秒， +#: 超时的照实记没比完（ledger/match.py）。召回的上限在第 4 波加。只有管理员能改（整个设置接口的改动都要管理员） +LEDGER_GROUP = "ledger" +LEDGER_DEFAULTS: dict[str, Any] = {"auto_issue": True, "issue_daily_limit": 20, "match_timeout_s": 8} +LEDGER_SETTING_LABEL = {"auto_issue": "自动按口径正式出具", "issue_daily_limit": "按口径正式出具每日上限", + "match_timeout_s": "口径比对时限（秒）"} +#: 时限的范围：太短比不完一份口径，太长占着比对的线程 +LEDGER_TIMEOUT_RANGE = (1, 60) + + +def _timeout_ok(value: Any) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) \ + and LEDGER_TIMEOUT_RANGE[0] <= value <= LEDGER_TIMEOUT_RANGE[1] + + +def ledger_problem(value: Any) -> str | None: + """设置页写进来的 ledger 这一组有什么问题，没问题返回 None。""" + if not isinstance(value, dict): + return f"「口径账本」设置格式有误，需要是一个对象，当前为「{value}」" + for key, item in value.items(): + if key not in LEDGER_DEFAULTS: + return (f"「口径账本」设置中不支持「{key}」，可设置的项：" + + "、".join(f"「{LEDGER_SETTING_LABEL[k]}」" for k in LEDGER_DEFAULTS)) + if key == "auto_issue" and not isinstance(item, bool): + return f"「{LEDGER_SETTING_LABEL[key]}」只能是开或关，当前为「{item}」" + if key == "issue_daily_limit" and not _daily_limit_ok(item): + return (f"「{LEDGER_SETTING_LABEL[key]}」需要是 0 到 {APPS_DAILY_LIMIT_MAX:,} 之间的整数（0 表示不再按口径出具），" + f"当前为「{item}」") + if key == "match_timeout_s" and not _timeout_ok(item): + return (f"「{LEDGER_SETTING_LABEL[key]}」需要是 {LEDGER_TIMEOUT_RANGE[0]} 到 {LEDGER_TIMEOUT_RANGE[1]} 之间的数，" + f"当前为「{item}」") + return None + + +def ledger_values(stored: Any) -> dict[str, Any]: + """库里存的（或刚写进来的）ledger 一组 → 补齐默认；存坏了的项按默认（自动出具的开关存坏了按关，宁可少出具）。""" + stored = stored if isinstance(stored, dict) else {} + auto, limit, timeout = stored.get("auto_issue"), stored.get("issue_daily_limit"), stored.get("match_timeout_s") + return {"auto_issue": auto if isinstance(auto, bool) else (LEDGER_DEFAULTS["auto_issue"] if auto is None else False), + "issue_daily_limit": limit if _daily_limit_ok(limit) else LEDGER_DEFAULTS["issue_daily_limit"], + "match_timeout_s": timeout if _timeout_ok(timeout) else LEDGER_DEFAULTS["match_timeout_s"]} + + +async def ledger_settings(session: AsyncSession) -> dict[str, Any]: + """设置里的 ledger 一组，没存过的补默认。按口径出具、比对从这里读。""" + row = await session.get(Setting, LEDGER_GROUP) + return ledger_values(row.value if row else None) + + # -------------------------------------------------------------------------- # 用户设置 # -------------------------------------------------------------------------- @@ -763,6 +818,8 @@ DEFAULT_SETTINGS: dict[str, Any] = { CATALOG_DRAFT_GROUP: catalog_draft_values(None), # 应用：按需视图的每日上限、使用者能不能建个人应用（管理员在设置页改） APPS_GROUP: apps_values(None), + # 口径账本：自动按口径正式出具的开关、每日上限，口径比对的时限 + LEDGER_GROUP: ledger_values(None), } @@ -825,6 +882,7 @@ async def get_settings(session: AsyncSession = Depends(get_session)) -> dict[str merged[PUBLIC_URL_GROUP] = public_url_values(stored.get(NOTIFY_GROUP)) merged[CATALOG_DRAFT_GROUP] = catalog_draft_values(stored.get(CATALOG_DRAFT_GROUP)) merged[APPS_GROUP] = apps_values(stored.get(APPS_GROUP)) + merged[LEDGER_GROUP] = ledger_values(stored.get(LEDGER_GROUP)) for key, value in stored.items(): merged.setdefault(key, value) return merged @@ -924,6 +982,10 @@ async def put_settings( if problem := apps_problem(payload.values[APPS_GROUP]): raise HTTPException(422, problem) payload.values[APPS_GROUP] = apps_values(payload.values[APPS_GROUP]) + if LEDGER_GROUP in payload.values: + if problem := ledger_problem(payload.values[LEDGER_GROUP]): + raise HTTPException(422, problem) + payload.values[LEDGER_GROUP] = ledger_values(payload.values[LEDGER_GROUP]) if found := await _provider_problem(session, payload.values): raise CodedHTTPException(422, found[1], PROVIDER_INVALID, field=found[0]) for key, value in payload.values.items(): diff --git a/backend/app/apps/backfill.py b/backend/app/apps/backfill.py index 6777986c..47349ed9 100644 --- a/backend/app/apps/backfill.py +++ b/backend/app/apps/backfill.py @@ -107,19 +107,26 @@ async def _upgrade(session: AsyncSession, app: Any, version: AppVersion, view_id async def _side_effect(session: AsyncSession, state: Any) -> str | None: """新版本的图补算时会不会碰到人或外部系统：会的交回一句原因（接在「「周报 v3」」后面），不会的返回 None。 + 作为流程固定的一律算；别的按图判（graph_side_effect）。""" + if state.pin.get("use") == "process": + return "是作为流程固定的" + return await graph_side_effect(session, state.graph, level=state.level) + + +async def graph_side_effect(session: AsyncSession, graph_json: Any, *, level: str | None = None) -> str | None: + """一张图替人重跑一次会不会碰到人或外部系统：会的交回一句原因（接在「「周报 v3」」后面），不会的返回 None。按新口径 + 补算（这里）和按口径正式出具（ledger/issue.py：自动出具、一键出具都不代人做这些事）共用这一个判断。 判断沿用应用检查对流程钉住项的那一套（apps/lint._lint_process）：人工节点、写外部系统的能力 （capabilities.registry.writes_externally，和运行时、发布检查同一个判断）、process.* 问题（子工作流先读好， - process.human_in_subgraph 才查得到）。""" + 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 - if state.pin.get("use") == "process": - return "是作为流程固定的" try: - graph = GraphSpec.model_validate(state.graph) + graph = GraphSpec.model_validate(graph_json) except ValueError: return "的图无法解析" if humans := [n.title for n in graph.nodes if n.type == NodeType.HUMAN]: @@ -132,7 +139,7 @@ async def _side_effect(session: AsyncSession, state: Any) -> str | None: 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=state.level or "published") + 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/ledger/derive.py b/backend/app/ledger/derive.py index b3c41202..4d22ff4c 100644 --- a/backend/app/ledger/derive.py +++ b/backend/app/ledger/derive.py @@ -131,10 +131,11 @@ class Pinned: @dataclass class Block: - """派生图里的一段：照抄（或这次执行过）的取数节点 + 一张口径卡。rename 是拷出来的节点改了名的映射（不带前缀时为空）。""" + """派生图里的一段：照抄（或这次执行过）的取数节点 + 一张口径卡。rename 是拷出来的节点改了名的映射（不带前缀时为空）。 + card_id 为 None 的段没有口径卡（对账里「这次的取数」那一段只有一条查询）。""" nodes: list[dict[str, Any]] - card_id: str + card_id: str | None rename: dict[str, str] = field(default_factory=dict) #: 这一段出自哪里（记进 DerivedGraph.origin） source: dict[str, Any] | None = None @@ -300,9 +301,11 @@ def _free_id(wanted: str, taken: set[str]) -> str: def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str, Any]] = (), - defaults: Mapping[str, Any] | None = None, tail: Sequence[dict[str, Any]] = ()) -> DerivedGraph: + defaults: Mapping[str, Any] | None = None, tail: Sequence[dict[str, Any]] = (), + contract: Mapping[str, Any] | None = None) -> DerivedGraph: """入口 → 各段（首尾相接）→ tail（对账卡、召回卡）→ 出口。出口每张口径卡（含 tail 里的 metrics 节点）一个字段， - 取那张卡的指标清单。几段的节点 id 撞了直接报错：两份口径放进同一张图要带前缀（caliber_block 的 prefix）。""" + 取那张卡的指标清单；contract 给了就写成出口的出具契约（对账写 metrics_from 指向对账卡，不加报告、不调模型）。几段的 + 节点 id 撞了直接报错：两份口径放进同一张图要带前缀（caliber_block 的 prefix）。""" if kind not in KINDS: raise ValueError(f"不认识的派生图种类：{kind}") body: list[dict[str, Any]] = [n for b in blocks for n in b.nodes] + [copy.deepcopy(n) for n in tail] @@ -313,15 +316,18 @@ def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str taken = set(ids) entry = _free_id(ENTRY_ID, taken) exit_ = _free_id(EXIT_ID, taken | {entry}) - cards = [b.card_id for b in blocks] + [str(n["id"]) for n in tail if n.get("type") == "metrics"] + cards = [b.card_id for b in blocks if b.card_id] + [str(n["id"]) for n in tail if n.get("type") == "metrics"] titles = {str(n["id"]): str((n.get("data") or {}).get("label") or n["id"]) for n in body} outputs, seen = [], set() for cid in cards: name = _free_id(titles.get(cid, cid), seen) seen.add(name) outputs.append({"name": name, "value": f"{{{{ nodes.{cid}.text }}}}"}) + exit_config: dict[str, Any] = {"fields": outputs} + if contract: + exit_config["contract"] = copy.deepcopy(dict(contract)) chain = [_node(entry, "input", "入口", fields=[copy.deepcopy(dict(f)) for f in fields]), *body, - _node(exit_, "output", "出口", fields=outputs)] + _node(exit_, "output", "出口", **exit_config)] for i, node in enumerate(chain): node["position"] = {"x": i * _STEP_X, "y": 0} graph: dict[str, Any] = { @@ -335,6 +341,45 @@ def assemble(kind: str, blocks: Sequence[Block], *, fields: Sequence[Mapping[str entry_fields=tuple(copy.deepcopy(dict(f)) for f in fields), origin=origin) +#: 对账图里「这次的取数」节点、对账卡、多行时按键对齐的合并查询的 id（细案 02 做法 6） +OURS_ID = "ours_q1" +RECONCILE_CARD = "reconcile" +ALIGN_ID = "align" + + +def reconcile_prefix(pinned: Pinned) -> str: + """对账图里口径那一段的前缀：口径的节点 id 和 ours_q1 / align / reconcile / 入口出口撞了才带 cal_，不撞不带。""" + reserved = {OURS_ID, ALIGN_ID, RECONCILE_CARD, ENTRY_ID, EXIT_ID} + clash = any(str(n.get("id")) in reserved for n in pinned.fetches) or pinned.node_id in reserved + return "cal_" if clash else "" + + +def ours_block(sql: str, tool: str, *, label: str = "这次的取数") -> Block: + """对账里「这次的取数」：这次执行过的 SQL 原文，放进一个调用工具节点（id ours_q1）。没有口径卡。""" + return Block(nodes=[_node(OURS_ID, "tool", label, tool=tool, args={"sql": sql})], card_id=None, + source={"kind": "executed", "node_id": OURS_ID}) + + +def reconcile_graph(pinned: Pinned, ours: Block, tail: Sequence[dict[str, Any]], *, + inputs: Mapping[str, Any]) -> DerivedGraph: + """口径对账（细案 02 做法 6）：入口（口径的入口字段，缺省值是这次的参数）→ 这次的取数 → 口径的取数（模板里保留 + {{ input.x }}，运行时按入口渲染）→ 口径卡（caliber_from 固定那一版）→ tail（多行时的对齐合并查询、对账卡）→ 出口 + （契约 metrics_from 指向对账卡；不加报告、不调模型）。两边在同一次运行里先后查，差值在对账卡里算。 + + 口径的节点 id 和 ours_q1 / align / reconcile 撞了的，口径那一段带前缀 cal_（改名映射照常写进 caliber_from.nodes）。 + 同样的输入永远得到同一张图。""" + block = caliber_block(pinned, prefix=reconcile_prefix(pinned)) + fields = [] + for f in pinned.entry_fields: + field_ = copy.deepcopy(dict(f)) + if field_.get("name") in inputs: + field_["default"] = inputs[field_["name"]] + fields.append(field_) + card = next((n for n in tail if n.get("type") == "metrics" and n.get("id") == RECONCILE_CARD), None) + return assemble("reconcile", [ours, block], fields=fields, defaults=pinned.defaults, tail=tail, + contract={"metrics_from": RECONCILE_CARD} if card is not None else None) + + def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> DerivedGraph: """口径探针：一份口径的取数 + 口径卡（评测口径题的期望，细案 01 做法 8）。节点 id 和原图一样，入口字段照抄。 同样的 pinned 永远得到同一张图（确定性）。""" @@ -342,6 +387,7 @@ def caliber_probe(pinned: Pinned, *, upgrade_policy: str | None = None) -> Deriv fields=pinned.entry_fields, defaults=pinned.defaults) -__all__ = ["Block", "DerivedGraph", "ENTRY_ID", "EXIT_ID", "GONE", "KINDS", "NOT_CARD", "NOT_REUSABLE", "NotDerivable", - "Pinned", "TEMPLATE", "assemble", "caliber_block", "caliber_probe", "entry_fields", "load_pinned", +__all__ = ["ALIGN_ID", "Block", "DerivedGraph", "ENTRY_ID", "EXIT_ID", "GONE", "KINDS", "NOT_CARD", "NOT_REUSABLE", + "NotDerivable", "OURS_ID", "Pinned", "RECONCILE_CARD", "TEMPLATE", "assemble", "caliber_block", + "caliber_probe", "entry_fields", "load_pinned", "ours_block", "reconcile_graph", "reconcile_prefix", "render_inputs", "template_problem"] diff --git a/backend/app/ledger/issue.py b/backend/app/ledger/issue.py new file mode 100644 index 00000000..43815e09 --- /dev/null +++ b/backend/app/ledger/issue.py @@ -0,0 +1,372 @@ +"""按口径正式出具：路径一、二a（docs/design/next-round/02-口径账本.md 做法 5，拍板 L1、L5）。 + +问数据的答案整题等价于一份已发布口径（ledger/labels.whole_of：主报告里每个数都来自同一口径的同一版、参数一致）时，可以 +按那份口径发起一次正式运行——同一个问题，从此有了一份按已发布版本出具的、完整走过门禁的答案。 + +**参数只从比对工件里取。** 发起用的输入 = 那一版入口字段的缺省值 + 比对时绑定出来的参数（whole.binding），再过 apps/forms +同一套表单校验；不收客户端传的参数，前端只能选「按哪份口径」——不然谁都能拿一次问数据去替一份正式口径「签字」任意输入。 +绑定的值是这次运行查到过的遮罩值的，不代填出具。 + +**经 launch_run 发起**（正式运行的全部门禁：已发布未退役、升版处置、子工作流固定版本、工具在不在），来源 ledger_issue +（服务端专用，engine/origin），origin 记 source_run、conversation_id、turn_id、binding_hash、口径和怎么认的、是不是自动的； +按 expect_hash 钉住那一版的 graph_hash（比对时考虑的就是这张图）。出具档位照正式运行原样走。 + +**能不能出具**（plan_issue，一键和自动同一套，不满足的照实写原因）： +- 这次运行是比对过的探索运行，整题等价于点名的这份口径（工作流、版本都对）； +- 那一版已发布、未退役； +- 图里没有人工节点、不会写外部系统（apps/backfill.graph_side_effect，和按新口径补算同一个判断）：替人重跑一次的正式 + 运行不该替人去审批、往外部系统写； +- 补了缺省值之后表单校验通过（缺入口字段的照实说缺哪个）； +- 权限（拍板 L5）：搭建者及以上（关登录时同）都可以；使用者只能按他有权使用的应用固定着的这一版出具（复用应用授权， + apps/access），别的写「请搭建者出具」；都要读得到这次运行； +- 当天没超上限（设置 ledger.issue_daily_limit，自动的、一键的合计）。 +**自动出具**（auto_issue，拍板 L1）另要：设置里开着、认法是独立等价（how=independent：Agent 自己写的 SQL 被证明等价； +「助手选用」的只给一键——选哪份口径是助手凭问法判断的，由人点一下代替证明）、版本是受管级别。只在运行一封存就比完的 +那一次触发（ledger/match）。不满足的写进 actions 的 auto_reason。 + +**同一运行、同一口径、同一组参数只发起一次**：按 input_key（apps/resolve，补算用的同一个指纹）认，已经发起过、没失败的 +直接交回那一次。进程内按（运行、工作流、版本）串行，自动和一键同时到也只发一次（单进程部署，同 apps/backfill 的锁）。 + +结果挂回答案：GET /api/runs/{id}/ledger 的 issued 列出 origin.source_run 等于这次运行的正式运行，档位文字用 TIER_TEXT。 +""" +from __future__ import annotations + +import asyncio +import logging +import weakref +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy import func, select + +from app.db.base import SessionLocal +from app.db.models import Run, Workflow + +logger = logging.getLogger(__name__) + +ORIGIN = "ledger_issue" +#: 出具被拒的机器码（接口回 409 / 403 / 429 时带它） +REFUSED = "ledger_issue_refused" +FORBIDDEN = "ledger_issue_forbidden" +DAILY_LIMIT = "ledger_issue_daily_limit" +#: 算「同一次」时看哪几种状态（失败、取消的可以再发） +_LIVE = ("queued", "running", "interrupted", "succeeded") + +#: 不能出具的原因 +NOT_MATCHED = "这次运行还没有比对口径，比对完才能按口径出具" +NOT_ELIGIBLE = "这次运行不能按口径出具：{why}" +NOT_WHOLE = "整题没有等价于这份口径：{why}" +OTHER_CALIBER = "整题等价的是另一份口径，不是「{name}」v{version}" +MASKED_BINDING = "参数里有这次运行查到过的遮罩值，不代填出具" +GONE = "口径所在的工作流不存在，可能已被删除" +VERSION_GONE = "「{name}」没有 v{version} 这个版本" +RETIRED = "「{name}」v{version} 已退役，不再接受新的正式运行" +UNPUBLISHED = "「{name}」v{version} 没有发布过，正式运行只能从已发布的版本发起" +SIDE_EFFECT = "「{name}」v{version}{reason}，按口径出具不代人做这些事" +INPUTS = "按这次的参数补上缺省值后，入口字段还不齐：{problems}" +NO_RUN_ACCESS = "没有权限查看这次运行" +ASK_BUILDER = "这份口径没有被你能使用的应用固定，不能由你按口径出具，请搭建者出具" +OVER_LIMIT = "今天的按口径出具已达到上限（{limit} 次），请明天再试，或请管理员调整上限" +#: 不自动出具的原因（能一键出具时写进 actions 的 auto_reason） +AUTO_OFF = "设置里关了自动按口径出具" +AUTO_COMPOSED = "数来自助手选用的口径，只给一键出具：选用哪份口径是助手按问法判断的，要由人确认" +AUTO_LEVEL = "这一版是已发布级别，自动出具只限受管级别的口径" +AUTO_RETRO = "这次比对是事后按现在的口径回看的，不自动出具" + + +class IssueRefused(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 + + +@dataclass +class Plan: + """按一份口径出具的打算：能不能（reason 为空就能）、发起用的输入、为什么不自动（auto_reason，能自动的为空）。""" + + run_id: str + workflow_id: str + version: int + caliber: str + how: str | None = None + level: str | None = None + workflow_name: str = "" + inputs: dict[str, Any] = field(default_factory=dict) + binding: dict[str, Any] = field(default_factory=dict) + expect_hash: str | None = None + match_artifact: str | None = None + #: 不能出具的原因、HTTP 状态码、机器码 + reason: str | None = None + status: int = 409 + code: str = REFUSED + auto_reason: str | None = None + #: 使用者靠哪个应用的授权出具（apps/access）：记进 origin + granted_by_app: str | None = None + #: 已经发起过的同一次（没失败的） + existing: str | None = None + + @property + def available(self) -> bool: + return self.reason is None + + def refuse(self, message: str, status: int = 409, code: str = REFUSED) -> "Plan": + if self.reason is None: + self.reason, self.status, self.code = message, status, code + return self + + def action(self) -> dict[str, Any]: + """交给界面的那一项动作（GET 账本的 actions）。""" + from app.ledger.labels import LEVEL_TEXT + + out: dict[str, Any] = {"kind": "issue", "workflow_id": self.workflow_id, "version": self.version, + "caliber": self.caliber, "workflow_name": self.workflow_name, "how": self.how, + "level": self.level, "level_text": LEVEL_TEXT.get(str(self.level or ""), ""), + "available": self.available, "auto": self.available and self.auto_reason is None} + if self.reason: + out["reason"] = self.reason + elif self.auto_reason: + out["auto_reason"] = self.auto_reason + if self.existing: + out["issued_run"] = self.existing + return out + + +def _local_midnight() -> datetime: + return datetime.now().astimezone().replace(hour=0, minute=0, second=0, microsecond=0).astimezone(timezone.utc) + + +async def issued_today(session: Any) -> int: + return int(await session.scalar(select(func.count(Run.id)).where( + Run.origin_kind == ORIGIN, Run.created_at >= _local_midnight())) or 0) + + +async def issued_runs(session: Any, run_id: str) -> list[Run]: + """这次运行按口径出具过的正式运行（origin.source_run 是它），新的在前。origin_kind 有索引，先按它筛再看 origin。""" + rows = (await session.execute(select(Run).where(Run.origin_kind == ORIGIN).order_by(Run.created_at.desc()))).scalars() + return [r for r in rows if isinstance(r.origin, dict) and r.origin.get("source_run") == run_id] + + +def issued_view(run: Run) -> dict[str, Any]: + """GET 账本 issued 里的一项：哪一次正式运行、状态、出具档位（TIER_TEXT，完整出具 / 降档出具 / 不予出具）。""" + from app.apps.resolve import TIER_TEXT + + output = run.output if isinstance(run.output, dict) else {} + issuance = output.get("_issuance") if isinstance(output.get("_issuance"), dict) else {} + tier = issuance.get("tier") + origin = run.origin if isinstance(run.origin, dict) else {} + return {"run_id": run.id, "status": run.status, "end_reason": getattr(run, "end_reason", None), + "workflow_id": run.workflow_id, "version": run.version, + "tier": tier if tier in TIER_TEXT else None, "tier_text": TIER_TEXT.get(str(tier)), + "auto": bool(origin.get("auto")), + "created_at": run.created_at.isoformat() if run.created_at else None} + + +async def _starter(session: Any, run: Run) -> Any: + """自动出具时按谁的权限判（拍板 L5）：这次运行的发起人。关登录时同管理员；账号不在了的按使用者算（宁可少出具）。""" + from app.api.deps import Principal + from app.auth import current_mode + from app.auth.sessions import principal_of + from app.db.models import User + + mode = current_mode() + if mode == "off": + return Principal(username=run.started_by, role="admin", verified=False, mode="off", via="off") + user = (await session.execute(select(User).where(User.username == run.started_by))).scalar_one_or_none() \ + if run.started_by else None + if user is None: + return Principal(username=run.started_by, role="user", verified=False, mode=mode) + return await principal_of(session, user, mode) + + +async def _granting_app(session: Any, principal: Any, workflow_id: str, version: int) -> str | None: + """使用者能按这一版出具，靠的是哪个应用的授权：固定着这一版、他至少能使用的应用（apps/access）。没有为 None。""" + from app.apps.access import app_role, at_least + from app.apps.versions import pinning_apps + from app.db.models_apps import App + + for item in await pinning_apps(session, workflow_id, version): + app = await session.get(App, item["app_id"]) + if app is not None and at_least(await app_role(session, principal, app), "use"): + return app.id + return None + + +async def plan_issue(session: Any, run: Run | None, content: dict[str, Any] | None, workflow_id: str, version: int, *, + principal: Any, hide: Any = None, artifact: str | None = None, retro: bool = False) -> Plan: + """按 workflow_id@version 出具这次运行的打算（见模块说明）。content 是这次运行最新的比对工件内容（读不出为 None）， + hide 判一个绑定值是不是这次运行查到过的遮罩值。只读，不发起。""" + from app.api.deps import can_read_run + from app.api.settings import ledger_settings + from app.apps import versions as app_versions + from app.apps.backfill import graph_side_effect + from app.apps.forms import input_schema, validate_form + from app.apps.resolve import input_key + from app.ledger import match as ledger_match + from app.ledger.derive import entry_fields + from app.ledger.labels import caliber_key + + plan = Plan(run_id=run.id if run else "", workflow_id=workflow_id, version=int(version), + caliber=caliber_key(workflow_id, int(version)), match_artifact=artifact) + if run is None: + return plan.refuse(NOT_ELIGIBLE.format(why=ledger_match.NOT_FOUND), 404) + if principal.mode != "off" and not await can_read_run(principal, run): + return plan.refuse(NO_RUN_ACCESS, 403, FORBIDDEN) + if (why := ledger_match.ineligible(run)) is not None: + return plan.refuse(NOT_ELIGIBLE.format(why=why)) + if content is None: + return plan.refuse(NOT_MATCHED) + whole = content.get("whole") or {} + workflow = await session.get(Workflow, workflow_id) + plan.workflow_name = workflow.name if workflow is not None else "" + if not whole.get("equivalent"): + return plan.refuse(NOT_WHOLE.format(why=whole.get("reason") or "没有比对结论")) + if (whole.get("workflow_id"), whole.get("version")) != (workflow_id, int(version)): + return plan.refuse(OTHER_CALIBER.format(name=plan.workflow_name or workflow_id, version=version)) + plan.how, plan.level = whole.get("how"), whole.get("level") + plan.binding = dict(whole.get("binding") or {}) + if hide is not None and any(hide(v) for v in plan.binding.values()): + return plan.refuse(MASKED_BINDING) + if workflow is None: + return plan.refuse(GONE, 404) + row = await app_versions.version_row(session, workflow_id, int(version)) + if row is None: + return plan.refuse(VERSION_GONE.format(name=workflow.name, version=version), 404) + if row.retired_at is not None: + return plan.refuse(RETIRED.format(name=workflow.name, version=version)) + if not app_versions.is_published(row, current=workflow.published_version): + return plan.refuse(UNPUBLISHED.format(name=workflow.name, version=version)) + plan.level = row.level or plan.level + plan.expect_hash = row.graph_hash + if reason := await graph_side_effect(session, row.graph, level=row.level): + return plan.refuse(SIDE_EFFECT.format(name=workflow.name, version=version, reason=reason)) + fields = list(entry_fields(row.graph or {})) + values = {str(f["name"]): f.get("default") for f in fields if f.get("default") not in (None, "")} + values.update({k: v for k, v in plan.binding.items() if v not in (None, "")}) + plan.inputs = values + if errors := validate_form(input_schema(fields), values): + return plan.refuse(INPUTS.format(problems="；".join(e["message"] for e in errors))) + if principal.mode != "off" and not principal.at_least("builder"): + plan.granted_by_app = await _granting_app(session, principal, workflow_id, int(version)) + if plan.granted_by_app is None: + return plan.refuse(ASK_BUILDER, 403, FORBIDDEN) + key = input_key(values) + plan.existing = next((r.id for r in await issued_runs(session, run.id) + if (r.workflow_id, r.version) == (workflow_id, int(version)) and r.status in _LIVE + and input_key(r.input) == key), None) + settings = await ledger_settings(session) + if plan.existing is None and await issued_today(session) >= int(settings["issue_daily_limit"]): + return plan.refuse(OVER_LIMIT.format(limit=settings["issue_daily_limit"]), 429, DAILY_LIMIT) + plan.auto_reason = (AUTO_OFF if not settings["auto_issue"] else AUTO_COMPOSED if plan.how != "independent" + else AUTO_LEVEL if plan.level != "governed" else AUTO_RETRO if retro else None) + return plan + + +async def _conversation_of(session: Any, run: Run) -> dict[str, str]: + """这次运行出自哪段对话的哪一轮：运行来源里封存的（origin_ref，第 2 波）；老运行退回 conversation_turns.run_id。""" + from app.db.models import ConversationTurn + + origin = run.origin if isinstance(run.origin, dict) else {} + if origin.get("conversation_id") and origin.get("turn_id"): + return {"conversation_id": str(origin["conversation_id"]), "turn_id": str(origin["turn_id"])} + turn = (await session.execute(select(ConversationTurn).where(ConversationTurn.run_id == run.id) + .order_by(ConversationTurn.created_at.desc()))).scalars().first() + return {"conversation_id": turn.conversation_id, "turn_id": turn.id} if turn is not None else {} + + +#: 同一（运行、工作流、版本）的出具串行：{事件循环: {键: 锁}}（asyncio.Lock 不能跨循环用，测试每个用例一个循环） +_locks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[tuple[str, str, int], asyncio.Lock]]" = \ + weakref.WeakKeyDictionary() + + +def _lock(key: tuple[str, str, int]) -> asyncio.Lock: + group = _locks.setdefault(asyncio.get_running_loop(), {}) + return group.setdefault(key, asyncio.Lock()) + + +async def _hider(run_id: str) -> Any: + from app.api.evidence import _sealed + from app.ledger.view import _hider as hider_of + + return (await hider_of(await _sealed(run_id))).hide + + +async def issue(run_id: str, workflow_id: str, version: int, *, principal: Any, auto: bool = False) -> tuple[Run, bool]: + """按 workflow_id@version 出具这次运行：(正式运行, 是不是这一次新发起的)。不能出具抛 IssueRefused；已经发起过同一次的 + 交回那一次（第二项为 False）。auto：自动出具（自动的条件也要满足）。""" + from app.engine.launch import LaunchRefused, launch_run + from app.ledger import store + + async with _lock((run_id, workflow_id, int(version))): + recorded = await store.latest(run_id) + hide = await _hider(run_id) + async with SessionLocal() as session: + run = await session.get(Run, run_id) + plan = await plan_issue(session, run, recorded.content if recorded else None, workflow_id, version, + principal=principal, hide=hide, artifact=recorded.artifact if recorded else None, + retro=bool(recorded and recorded.retro)) + if plan.reason is not None: + raise IssueRefused(plan.status, plan.reason, plan.code) + if plan.existing is not None: + return await session.get(Run, plan.existing), False + if auto and plan.auto_reason is not None: + raise IssueRefused(409, plan.auto_reason) + from app.apps.resolve import input_key + + origin = {"source_run": run_id, **await _conversation_of(session, run), "caliber": plan.caliber, + "how": plan.how, "binding_hash": input_key(plan.binding), "match": plan.match_artifact, + "auto": auto, **({"granted_by_app": plan.granted_by_app} if plan.granted_by_app else {})} + actor = run.started_by if auto else principal.username + try: + launched = await launch_run(session, workflow_id=workflow_id, run_class="formal", version=int(version), + input_payload=plan.inputs, actor=actor, origin_kind=ORIGIN, origin=origin, + expect_hash=plan.expect_hash) + except LaunchRefused as e: + raise IssueRefused(e.status_code, e.message, e.code or REFUSED) from e + return launched, True + + +async def plans_for(run_id: str, content: dict[str, Any] | None, *, principal: Any, hide: Any, artifact: str | None, + retro: bool) -> list[Plan]: + """GET 账本的 actions 里出具那几项：整题等价的那一份口径能不能出具、能不能自动（整题不等价的没有这一项）。""" + whole = (content or {}).get("whole") or {} + if not whole.get("equivalent") or not whole.get("workflow_id") or whole.get("version") is None: + return [] + async with SessionLocal() as session: + run = await session.get(Run, run_id) + return [await plan_issue(session, run, content, str(whole["workflow_id"]), int(whole["version"]), + principal=principal, hide=hide, artifact=artifact, retro=retro)] + + +async def auto_issue(run_id: str) -> str | None: + """比对完、整题等价时看能不能自动出具（拍板 L1），能就发起；返回新发起的正式运行 id。不能的、出错的只记日志：自动出具 + 是辅助步骤，不影响运行本身和答案（照实说为什么没自动出具在 GET 账本的 actions 里）。""" + from app.ledger import store + + try: + recorded = await store.latest(run_id) + whole = ((recorded.content if recorded else None) or {}).get("whole") or {} + if not whole.get("equivalent"): + return None + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if run is None: + return None + principal = await _starter(session, run) + launched, fresh = await issue(run_id, str(whole["workflow_id"]), int(whole["version"]), principal=principal, + auto=True) + return launched.id if fresh else None + except IssueRefused as e: + logger.info("运行 %s 没有自动按口径出具：%s", run_id, e.message) + return None + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 - 辅助步骤 + logger.exception("运行 %s 自动按口径出具出错", run_id) + return None + + +__all__ = ["DAILY_LIMIT", "FORBIDDEN", "ORIGIN", "REFUSED", "IssueRefused", "Plan", "auto_issue", "issue", + "issued_runs", "issued_today", "issued_view", "plan_issue", "plans_for"] diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py index 96605533..fc2643d9 100644 --- a/backend/app/ledger/match.py +++ b/backend/app/ledger/match.py @@ -19,7 +19,7 @@ MAX_CANDIDATES 份（按工作流、版本、节点排好的先后）。 schema 用这条查询当时冻结的表结构（表结构快照），列的中文名取同一份快照里冻结的数据目录。只读封存范围内的快照（api/evidence 的 _sealed 读法），读不出、被改过的快照不比，记 skipped。 -**资源。** 每次运行限时 TIMEOUT_S（缺省 8 秒），在线程里跑，同时最多 CONCURRENCY（2）个（照 scheduler/auto_compare 的做法： +**资源。** 每次运行限时（设置 ledger.match_timeout_s，缺省 TIMEOUT_S = 8 秒），在线程里跑，同时最多 CONCURRENCY（2）个（照 scheduler/auto_compare 的做法： 信号量按事件循环分开，线程池两个线程）。时限按绑定逐组看：过了时限，还没比的口径记 skipped（比对超时）；整次比对卡在线程里 过了时限再加 GRACE_S 还没回来，整次记 skipped。解析失败同样照实记 skipped 和原因。都只记日志和 skipped，不影响运行本身。 @@ -477,15 +477,21 @@ async def _kinds(session: Any) -> dict[str, str | None]: return {name: kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind))).all()} -async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float = TIMEOUT_S) -> str | None: +async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float | None = None) -> str | None: """比对一次运行，存工件、追加 ledger.matched；返回工件 id。不比对（不是探索运行、没成功、没封存、没有查询）、比对期间 - 运行换了封存时返回 None。比对本身出的问题都记成 skipped，不抛。""" + 运行换了封存时返回 None。比对本身出的问题都记成 skipped，不抛。timeout 不给按设置（ledger.match_timeout_s）。 + + 运行一封存就比的（trigger=sealed）比完接着看能不能按口径自动正式出具（ledger/issue.auto_issue，拍板 L1）：事后补比、 + 按需比的不自动出具——那是按现在的口径回看老答案，不该替人补发正式运行。""" from app.api.evidence import _sealed + from app.api.settings import ledger_settings async with SessionLocal() as session: run = await session.get(Run, run_id) if ineligible(run) is not None: return None + if timeout is None: + timeout = float((await ledger_settings(session))["match_timeout_s"]) sealed = await _sealed(run_id) if not sealed.queries: return None @@ -500,7 +506,12 @@ async def match_run(run_id: str, *, trigger: str = "sealed", timeout: float = TI except TimeoutError: logger.warning("运行 %s 的口径比对超过 %g 秒仍未完成，记为没比完", run_id, timeout + GRACE_S) content = _timed_out(sealed) - return await store.save(run_id, content, seal=(sealed.run_manifest_seq, sealed.manifest_hash), trigger=trigger) + artifact = await store.save(run_id, content, seal=(sealed.run_manifest_seq, sealed.manifest_hash), trigger=trigger) + if artifact is not None and trigger == "sealed" and (content.get("whole") or {}).get("equivalent"): + from app.ledger import issue + + await issue.auto_issue(run_id) + return artifact def schedule(run_id: str, *, trigger: str = "sealed") -> bool: diff --git a/backend/app/ledger/reconcile.py b/backend/app/ledger/reconcile.py new file mode 100644 index 00000000..1bf35caa --- /dev/null +++ b/backend/app/ledger/reconcile.py @@ -0,0 +1,518 @@ +"""口径对账：路径二b，差在哪、差多少（docs/design/next-round/02-口径账本.md 做法 6，总方案 X3）。 + +**差在哪**在比对时就有（ledger/match：explain_sql_diff 的条目，这次 / 口径两边说），不跑查询。**差多少**要点开才跑： +按那一版口径确定性派生一张只读小图（ledger/derive.reconcile_graph，全站唯一的派生图生成器），跑一次探索运行 +（来源 reconcile，服务端专用、系统派生、不进使用统计）： + + 入口（口径的入口字段，缺省值是这次的参数）→「这次的取数」ours_q1（执行过的 SQL 原文）→ 口径的取数（模板里保留 + {{ input.x }}）→ 口径卡（caliber_from 固定那一版）→（多行时）按维度对齐的合并查询 align → 对账卡（这次、口径、差值、 + 相对差）→ 出口（契约 metrics_from 指向对账卡；不加报告、不调模型） + +两边在同一次运行里查，差值在对账卡里算（叙述层无算术，也不在 Python 里算：compare.py 是同一工作流两次运行配对用的）。 +面板上每个数都点得开 /evidence/metrics 的出处链。 + +**参数。** 默认用比对时绑定出来的参数（缺的补缺省值）。参数是启发得来的（绑定在 heuristic 一级配上）、有没绑上的、补完 +缺省值表单还不齐、或者里面有遮罩值时，要人先核对：动作里写 needs_check 和原因，发起时要带 confirm 和 inputs（按口径入口 +的字段校验，ledger/compose.check_inputs 同一套），界面写「参数按这次的 SQL 推断，请核对」。 + +**列怎么配。** 只有一行结果时：口径卡上的指标式就是取数第一行某一列（cell(nodes.取数, 0, '列')）的，按那一列在口径取数 +结果里的位置配这次结果同一位置的列（等价时已实测列的先后计；不等价的两条 SQL 选择列表一样时位置照样对得上）。差异 +有兜底说明（fallback：两条 SQL 对不上结构）、列数不同、一个都配不上时不猜，要人选（pairs：{ours: 这次的列, metric: 口径卡 +上的指标 id}），这个选择记进对账运行的 origin。 + +**多行结果**（这次的结果不止一行）：被报告引用了的那几行（最多 MAX_ROWS 行），按维度列（不是数的列）经合并查询和口径 +取数的结果按键对齐（LEFT JOIN，按引用的先后排），每个被引用的行、每个被引用的列出一组对账指标。两边的列按位置配（列数 +不同时不猜，照实拒绝）。维度列是遮罩列、或者这次的结果里合并键不唯一时照实拒绝：前者对齐要用到遮住的值，后者对不准 +是哪一行。被引用的列是遮罩列的同样不对账。 + +只读：派生图只查库、不写；使用者发起时带上 MASK_RESULTS / READ_ONLY（同问数据），搭建者也带 READ_ONLY。 +""" +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from typing import Any + +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import Run, Workflow + +ORIGIN = "reconcile" +#: 被拒的机器码 +REFUSED = "ledger_reconcile_refused" +CHECK_INPUTS = "ledger_reconcile_check_inputs" +NEEDS_PAIRS = "ledger_reconcile_needs_pairs" +FORBIDDEN = "ledger_reconcile_forbidden" +#: 多行对账最多几行、几列（每行每列出四个对账指标） +MAX_ROWS = 10 +MAX_COLUMNS = 5 + +#: 给人看的说法 +NOT_FOUND = "运行记录不存在" +NO_ACCESS = "没有权限查看这次运行" +NOT_MATCHED = "这次运行还没有比对口径，比对完才能对账" +NO_QUERY = "这次运行的比对记录里没有这条查询" +NOT_RELATED = "这条查询没有和「{name}」v{version} 比出差异（只有相关口径才对账）" +NO_CARD = "「{name}」v{version} 里没有读这个取数节点的口径卡，对不了账" +SNAPSHOT = "这条查询的快照读不出来（文件缺失或与哈希不一致），对不了账" +CHECK_REASON = "参数按这次的 SQL 推断，请核对" +CHECK_MASKED = "参数里有遮罩值，请核对参数" +CHECK_INCOMPLETE = "按这次的 SQL 没能推断出全部参数，请补全并核对" +CONFIRM_FIRST = "参数按这次的 SQL 推断，发起前请先核对参数" +PAIRS_FALLBACK = "两条 SQL 的结构对不上，认不出这次的哪一列对应口径的哪个指标，请选择要对照的指标" +PAIRS_NONE = "认不出这次的哪一列对应口径的哪个指标，请选择要对照的指标" +PAIR_COLUMN = "这次的结果里没有列「{column}」" +PAIR_MASKED = "列「{column}」设置了遮罩，不能对账" +PAIR_METRIC = "口径卡「{card}」上没有指标「{metric}」" +PAIRS_SHAPE = "要对照的指标写成 [{ours: 这次的列, metric: 口径卡上的指标 id}]" +ROWS_NONE = "报告里没有引用这条查询的格，不知道要对哪几行" +ROWS_NO_KEY = "这次的结果有多行，但没有可以用来对齐的维度列（不是数的列）" +ROWS_MASKED_KEY = "用来对齐的维度列「{column}」设置了遮罩，不能对账" +ROWS_DUPLICATE = "这次的结果里按维度列（{columns}）有重复的行，对不准是哪一行" +ROWS_COLUMNS = "这次的结果和口径的取数列数不同（{ours} 列对 {theirs} 列），按位置对不上" +ROWS_CALIBER_COLUMNS = "认不出口径取数的结果列，按位置对不上" + + +class ReconcileRefused(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 + + +# ========================================================================== +# 动作：GET 账本里能对的账 +# ========================================================================== + + +def _needs_check(cand: dict[str, Any], fields: list[dict[str, Any]], hide: Any) -> str | None: + """参数要不要人先核对：要的回原因（见模块说明），不要的为 None。fields 是比对工件里记的入口字段（name、required、 + has_default）。""" + binding = cand.get("binding") or {} + if hide is not None and any(hide(v) for v in binding.values()): + return CHECK_MASKED + if cand.get("unbound") or not binding and any(f.get("name") for f in fields): + return CHECK_INCOMPLETE + if any(f.get("required") and not f.get("has_default") and f.get("name") not in binding for f in fields): + return CHECK_INCOMPLETE + if cand.get("via") in (None, "heuristic"): + return CHECK_REASON + return None + + +def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None) -> list[dict[str, Any]]: + """比对工件里每条有相关口径的查询、每份相关口径一项「看差多少」：{kind: reconcile, query, workflow_id, version, node_id, + needs_check, check_reason?, binding（遮过的）}。界面据此摆按钮、要不要先核对参数。""" + from app.ledger.labels import LEVEL_TEXT, binding_view + + hide = hide or (lambda _value: False) + fields = {c.get("caliber"): c.get("fields") or [] for c in content.get("calibers") or []} + out: list[dict[str, Any]] = [] + for query in content.get("queries") or []: + if query.get("status") not in ("related",): + continue + for cand in query.get("candidates") or []: + if cand.get("status") != "related" or cand.get("comparable") is False: + continue + why = _needs_check(cand, list(fields.get(cand.get("caliber")) or []), hide) + item = {"kind": "reconcile", "query": query.get("artifact"), "caliber": cand.get("caliber"), + "workflow_id": cand.get("workflow_id"), "version": cand.get("version"), + "node_id": cand.get("node_id"), "workflow_name": names.get(str(cand.get("workflow_id") or "")), + "level_text": LEVEL_TEXT.get(str(cand.get("level") or ""), ""), + "needs_check": why is not None, "binding": binding_view(cand.get("binding"), hide), + "fallback": bool((cand.get("diff") or {}).get("fallback"))} + if why is not None: + item["check_reason"] = why + out.append(item) + return out + + +async def reconciled_runs(session: Any, run_id: str) -> list[Run]: + """对这次运行对过的账（origin.source_run 是它），新的在前。""" + rows = (await session.execute(select(Run).where(Run.origin_kind == ORIGIN).order_by(Run.created_at.desc()))).scalars() + return [r for r in rows if isinstance(r.origin, dict) and r.origin.get("source_run") == run_id] + + +def reconciled_view(run: Run) -> dict[str, Any]: + origin = run.origin if isinstance(run.origin, dict) else {} + return {"run_id": run.id, "status": run.status, "query": origin.get("query"), "caliber": origin.get("caliber"), + "workflow_id": origin.get("workflow_id"), "version": origin.get("version"), + "inputs_from": origin.get("inputs_from"), "rows": origin.get("rows"), + "created_at": run.created_at.isoformat() if run.created_at else None} + + +# ========================================================================== +# 派生图 +# ========================================================================== + + +@dataclass +class Pair: + """一组对照：这次结果的哪一列，对口径卡上的哪个指标（metric）或口径取数结果的哪一列（column）。""" + + ours: str + metric: str | None = None + column: str | None = None + #: 口径卡上那个指标的位置（metric 配对时） + index: int | None = None + name: str = "" + unit: str = "" + decimals: Any = None + + def describe(self) -> dict[str, Any]: + return {"ours": self.ours, **({"metric": self.metric} if self.metric else {}), + **({"caliber": self.column} if self.column else {})} + + +@dataclass +class Plan: + """一次对账要派生的东西（纯计算的结果，测试锁住确定性）。""" + + pairs: list[Pair] + rows: list[int] | None = None + keys: list[tuple[str, str]] = field(default_factory=list) + key_values: list[tuple[Any, ...]] = field(default_factory=list) + + +def _lit(value: Any) -> str: + """一个值写进合并查询（SQLite）的字面量。""" + if value is None: + return "NULL" + if isinstance(value, bool): + return "1" if value else "0" + if isinstance(value, (int, float)): + return repr(value) + return "'" + str(value).replace("'", "''") + "'" + + +def _ident(name: str) -> str: + return '"' + str(name).replace('"', '""') + '"' + + +def _equals(alias: str, column: str, value: Any) -> str: + return f"{alias}.{_ident(column)} IS NULL" if value is None else f"{alias}.{_ident(column)} = {_lit(value)}" + + +def align_sql(keys: list[tuple[str, str]], pairs: list[Pair], values: list[tuple[Any, ...]]) -> str: + """多行对账的合并查询：这次（o）LEFT JOIN 口径（c）按维度列对齐，只留被引用的那几行，按引用的先后排。""" + select_ = [f"o.{_ident(ours)} AS {_ident(ours)}" for ours, _ in keys] + select_ += [f"o.{_ident(p.ours)} AS {_ident(f'ours_{j}')}, c.{_ident(p.column or '')} AS {_ident(f'cal_{j}')}" + for j, p in enumerate(pairs)] + on = " AND ".join(f"o.{_ident(ours)} = c.{_ident(theirs)}" for ours, theirs in keys) + rows = ["(" + " AND ".join(_equals("o", ours, v) for (ours, _), v in zip(keys, row)) + ")" for row in values] + order = " ".join(f"WHEN {cond} THEN {i}" for i, cond in enumerate(rows)) + return (f"SELECT {', '.join(select_)} FROM o LEFT JOIN c ON {on} WHERE {' OR '.join(rows)} " + f"ORDER BY CASE {order} END") + + +def _metric(mid: str, name: str, expression: str, *, unit: str = "", decimals: Any = None, + fmt: str | None = None) -> dict[str, Any]: + out: dict[str, Any] = {"id": mid, "name": name, "unit": unit, "expression": expression} + if decimals is not None: + out["decimals"] = decimals + if fmt: + out["format"] = fmt + return out + + +def _four(n: str, label: str, ours: str, theirs: str, pair: Pair) -> list[dict[str, Any]]: + """一组对照的四个对账指标：这次、口径、差值（这次 − 口径）、相对差（差值 ÷ 口径，口径是 0 时没有值）。""" + diff = f"({ours}) - ({theirs})" + return [ + _metric(f"ours_{n}", f"{label} · 这次", ours, unit=pair.unit, decimals=pair.decimals), + _metric(f"cal_{n}", f"{label} · 口径", theirs, unit=pair.unit, decimals=pair.decimals), + _metric(f"diff_{n}", f"{label} · 差值", diff, unit=pair.unit, decimals=pair.decimals), + _metric(f"rel_{n}", f"{label} · 相对差", f"({diff}) / ({theirs}) if ({theirs}) else None", decimals=4, + fmt="percent_of_ratio"), + ] + + +def tail_nodes(plan: Plan, *, card_id: str, fetch_id: str) -> list[dict[str, Any]]: + """派生图的 tail：（多行时）对齐的合并查询 + 对账卡。card_id、fetch_id 是图里口径卡、口径取数节点的 id。""" + from app.ledger.derive import ALIGN_ID, OURS_ID, RECONCILE_CARD, _node + + metrics: list[dict[str, Any]] = [] + nodes: list[dict[str, Any]] = [] + if plan.rows is None: + for j, pair in enumerate(plan.pairs): + ours = f"cell(nodes.{OURS_ID}, 0, {pair.ours!r})" + theirs = (f"nodes.{card_id}.metrics[{pair.index}].value" if pair.metric is not None + else f"cell(nodes.{fetch_id}, 0, {pair.column!r})") + metrics += _four(str(j), pair.name or pair.ours, ours, theirs, pair) + else: + nodes.append(_node(ALIGN_ID, "merge", "按维度对齐", inputs={"o": OURS_ID, "c": fetch_id}, + sql=align_sql(plan.keys, plan.pairs, plan.key_values))) + for i, row in enumerate(plan.key_values): + where = "，".join(f"{ours}={v}" for (ours, _), v in zip(plan.keys, row)) + for j, pair in enumerate(plan.pairs): + metrics += _four(f"{i}_{j}", f"{where} · {pair.name or pair.ours}", + f"cell(nodes.{ALIGN_ID}, {i}, 'ours_{j}')", f"cell(nodes.{ALIGN_ID}, {i}, 'cal_{j}')", + pair) + nodes.append(_node(RECONCILE_CARD, "metrics", "对账卡", caliber="口径对账", caliber_version="v1", on_missing="null", + metrics=metrics)) + return nodes + + +def _numeric(rows: list[Any], index: int) -> bool: + values = [r[index] for r in rows if isinstance(r, (list, tuple)) and index < len(r) and r[index] is not None] + return bool(values) and all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values) + + +def _plain_cell(expression: str, fetch: str) -> str | None: + """指标式就是取数第一行某一列（cell(nodes., 0, '列')）时交回列名，别的写法为 None。""" + import ast + + from app.engine.expressions import cell_parts + + parts = cell_parts(expression.strip()) + if parts is None or parts[0] != f"nodes.{fetch}" or parts[1].strip() != "0": + return None + try: + column = ast.literal_eval(parts[2]) + except (ValueError, SyntaxError): + return None + return column if isinstance(column, str) else None + + +def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str], caliber_columns: list[str] | None, + metrics: list[dict[str, Any]], fetch: str, card: str, fallback: bool, chosen: Any, + referenced: list[tuple[int, str]]) -> Plan: + """配列、（多行时）定行和键：交回 Plan，配不了的抛 ReconcileRefused（见模块说明）。纯计算。""" + def check_column(column: str) -> None: + if column not in ours_columns: + raise ReconcileRefused(422, PAIR_COLUMN.format(column=column)) + if column.lower() in hidden: + raise ReconcileRefused(409, PAIR_MASKED.format(column=column)) + + by_id = {str(m.get("id")): (i, m) for i, m in enumerate(metrics)} + multi = len(ours_rows) > 1 + if multi: + return _plan_rows(ours_columns, ours_rows, hidden, caliber_columns, referenced, check_column) + if chosen is not None: + if not isinstance(chosen, list) or not chosen or not all(isinstance(p, dict) for p in chosen): + raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS) + pairs = [] + for item in chosen: + ours, metric = str(item.get("ours") or ""), str(item.get("metric") or "") + check_column(ours) + if metric not in by_id: + raise ReconcileRefused(422, PAIR_METRIC.format(card=card, metric=metric)) + index, m = by_id[metric] + pairs.append(Pair(ours=ours, metric=metric, index=index, name=str(m.get("name") or metric), + unit=str(m.get("unit") or ""), decimals=m.get("decimals"))) + return Plan(pairs=pairs) + if fallback: + raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS) + pairs = [] + if caliber_columns is not None and len(caliber_columns) == len(ours_columns): + for index, m in enumerate(metrics): + column = _plain_cell(str(m.get("expression") or ""), fetch) + if column is None or column not in caliber_columns: + continue + ours = ours_columns[caliber_columns.index(column)] + if ours.lower() in hidden: + continue + pairs.append(Pair(ours=ours, metric=str(m.get("id")), index=index, name=str(m.get("name") or m.get("id")), + unit=str(m.get("unit") or ""), decimals=m.get("decimals"))) + if not pairs: + raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS) + return Plan(pairs=pairs) + + +def _plan_rows(ours_columns: list[str], rows: list[Any], hidden: set[str], caliber_columns: list[str] | None, + referenced: list[tuple[int, str]], check_column: Any) -> Plan: + if caliber_columns is None: + raise ReconcileRefused(409, ROWS_CALIBER_COLUMNS) + if len(caliber_columns) != len(ours_columns): + raise ReconcileRefused(409, ROWS_COLUMNS.format(ours=len(ours_columns), theirs=len(caliber_columns))) + keys = [i for i in range(len(ours_columns)) if not _numeric(rows, i)] + if not keys: + raise ReconcileRefused(409, ROWS_NO_KEY) + if masked := [ours_columns[i] for i in keys if ours_columns[i].lower() in hidden]: + raise ReconcileRefused(409, ROWS_MASKED_KEY.format(column=masked[0])) + tuples = [tuple(r[i] if isinstance(r, (list, tuple)) and i < len(r) else None for i in keys) for r in rows] + if len(set(map(json.dumps, [list(t) for t in tuples]))) != len(tuples): + raise ReconcileRefused(409, ROWS_DUPLICATE.format(columns="、".join(ours_columns[i] for i in keys))) + wanted_rows = list(dict.fromkeys(r for r, _ in referenced if isinstance(r, int) and 0 <= r < len(rows)))[:MAX_ROWS] + columns = list(dict.fromkeys(c for _, c in referenced if c in ours_columns and ours_columns.index(c) not in keys)) + if not wanted_rows or not columns: + raise ReconcileRefused(409, ROWS_NONE) + for column in columns: + check_column(column) + pairs = [Pair(ours=c, column=caliber_columns[ours_columns.index(c)], name=c) for c in columns[:MAX_COLUMNS]] + return Plan(pairs=pairs, rows=wanted_rows, keys=[(ours_columns[i], caliber_columns[i]) for i in keys], + key_values=[tuples[r] for r in wanted_rows]) + + +def _referenced(sealed: Any, artifact: str) -> list[tuple[int, str]]: + """报告里引用了这条查询的哪几格：[(行, 列)]，按正文的先后（单元格、表格里的格）。""" + from app.api.evidence import _reports + from app.engine.evidence import iter_units + + out: list[tuple[int, str]] = [] + for report in _reports(sealed): + doc = report.doc if report.hash_ok is True else None + if not isinstance(doc, dict): + continue + catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} + for _, unit in iter_units(doc): + for seg in unit.get("segments") or []: + cite = seg.get("cite") if isinstance(seg, dict) and isinstance(seg.get("cite"), dict) else {} + entry = catalog.get(cite.get("alias")) if isinstance(catalog.get(cite.get("alias")), dict) else {} + locator = cite.get("locator") if isinstance(cite.get("locator"), dict) else {} + if cite.get("kind") == "cell" and entry.get("artifact") == artifact \ + and isinstance(locator.get("row"), int) and isinstance(locator.get("column"), str): + out.append((locator["row"], locator["column"])) + return out + + +def _caliber_columns(template: str, inputs: dict[str, Any], fields: list[dict[str, Any]], dialect: str | None, + schema: dict[str, Any] | None) -> list[str] | None: + """口径取数按这次的参数渲染之后选出来的列名（按位置）；认不出的为 None。""" + from app.data.sql_semantics import _read, _result_columns, _schema_info, render_bound + + try: + rendered = render_bound(template, inputs, fields) + return _result_columns(rendered, _read(dialect), _schema_info(schema)) + except Exception: # noqa: BLE001 - 认不出就按认不出说 + return None + + +# ========================================================================== +# 发起 +# ========================================================================== + + +@dataclass +class Request: + query: str + workflow_id: str + version: int + inputs: dict[str, Any] | None = None + confirm: bool = False + pairs: Any = None + + +async def reconcile(run_id: str, req: Request, *, principal: Any) -> Run: + """对这次运行的一条查询和一份相关口径对一次账：派生只读图、发起探索运行（来源 reconcile）。不能对的抛 ReconcileRefused。""" + from app.api.deps import can_read_run + from app.api.evidence import _Masks, _sealed + from app.core.artifact_store import load + from app.engine.launch import LaunchRefused, launch_run + from app.ledger import compose, derive, store + from app.ledger.issue import _conversation_of + from app.ledger.labels import caliber_key + from app.ledger.match import _frozen + from app.ledger.view import _hider + from app.tools.datasource import MASK_RESULTS_OPTION, QUERY_PREFIX, READ_ONLY_OPTION + + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if run is None: + raise ReconcileRefused(404, NOT_FOUND) + if principal.mode != "off" and not await can_read_run(principal, run): + raise ReconcileRefused(403, NO_ACCESS, FORBIDDEN) + recorded = await store.latest(run_id) + if recorded is None or recorded.content is None: + raise ReconcileRefused(409, NOT_MATCHED) + content = recorded.content + key = caliber_key(req.workflow_id, int(req.version)) + query = next((q for q in content.get("queries") or [] if q.get("artifact") == req.query), None) + if query is None: + raise ReconcileRefused(404, NO_QUERY) + async with SessionLocal() as session: + workflow = await session.get(Workflow, req.workflow_id) + name = workflow.name if workflow is not None else req.workflow_id + cand = next((c for c in query.get("candidates") or [] if c.get("caliber") == key and c.get("status") == "related"), + None) + if cand is None: + raise ReconcileRefused(409, NOT_RELATED.format(name=name, version=req.version)) + summary = next((c for c in content.get("calibers") or [] if c.get("caliber") == key), {}) + pinned = None + async with SessionLocal() as session: + for card_id in summary.get("cards") or []: + try: + candidate = await derive.load_pinned(session, req.workflow_id, int(req.version), str(card_id)) + except derive.NotDerivable: + continue + if any(node == cand.get("node_id") for node, _, _ in candidate.reads): + pinned = candidate + break + if pinned is None: + raise ReconcileRefused(409, NO_CARD.format(name=name, version=req.version)) + + # 参数：比对时绑定出来的（补缺省值），要核对的要人带着核对过的参数再来 + sealed = await _sealed(run_id) + hider = await _hider(sealed) + why = _needs_check(cand, list(summary.get("fields") or []), hider.hide) + if req.confirm and req.inputs is not None: + try: + inputs = compose.check_inputs(pinned, req.inputs) + except compose.ComposeRefused as e: + raise ReconcileRefused(422, e.message, CHECK_INPUTS) from e + inputs_from = "checked" + elif why is not None: + raise ReconcileRefused(409, CONFIRM_FIRST if why == CHECK_REASON else why, CHECK_INPUTS) + else: + try: + inputs = compose.check_inputs(pinned, dict(cand.get("binding") or {})) + except compose.ComposeRefused as e: + raise ReconcileRefused(409, f"{CHECK_INCOMPLETE}：{e.message}", CHECK_INPUTS) from e + inputs_from = "binding" + + # 这次的那条查询：快照、遮罩列、被引用的格 + try: + snap = load(req.query) + except ValueError: + snap = None + if not isinstance(snap, dict) or not isinstance(snap.get("sql"), str) or req.query not in sealed.queries: + raise ReconcileRefused(409, SNAPSHOT) + source = str(snap.get("source") or query.get("source") or "") + columns = [str(c) for c in snap.get("columns") or []] + rows = snap.get("rows") if isinstance(snap.get("rows"), list) else [] + masks = _Masks(sealed) + hidden = set((await masks.sets(req.query, source, snap)).hidden) + kind = await masks.kind(source) + schema, _ = _frozen(snap.get("schema_artifact")) + fetch = next(n for n in pinned.fetches if n["id"] == cand.get("node_id")) + template = str(((fetch.get("data") or {}).get("config") or {}).get("args", {}).get("sql") or "") + plan = plan_pairs(ours_columns=columns, ours_rows=rows, hidden=hidden, + caliber_columns=_caliber_columns(template, inputs, list(pinned.entry_fields), kind, schema), + metrics=pinned.metrics, fetch=str(fetch["id"]), card=pinned.card_title, + fallback=bool((cand.get("diff") or {}).get("fallback")), chosen=req.pairs, + referenced=_referenced(sealed, req.query)) + graph = build(pinned, sql=snap["sql"], tool=f"{QUERY_PREFIX}{source}", plan=plan, inputs=inputs, + fetch=str(fetch["id"])) + + async with SessionLocal() as session: + origin = {"source_run": run_id, **await _conversation_of(session, run), "query": req.query, "caliber": key, + "workflow_id": req.workflow_id, "version": int(req.version), "inputs_from": inputs_from, + "binding_via": cand.get("via"), "pairs": [p.describe() for p in plan.pairs], + **({"rows": plan.rows} if plan.rows is not None else {}), + **({"pairs_chosen": True} if req.pairs is not None else {}), "derive": graph.origin} + options = {READ_ONLY_OPTION: True} + if principal.mode != "off" and not principal.at_least("builder"): + options[MASK_RESULTS_OPTION] = True + try: + return await launch_run(session, graph=graph.graph, input_payload=inputs, run_class="exploratory", + actor=principal.username, origin_kind=ORIGIN, origin=origin, options=options, + workflow_name=f"口径对账：{name} v{req.version}") + except LaunchRefused as e: + raise ReconcileRefused(e.status_code, e.message, e.code or REFUSED) from e + + +def build(pinned: Any, *, sql: str, tool: str, plan: Plan, inputs: dict[str, Any], fetch: str) -> Any: + """按 Plan 派生对账图（ledger/derive.reconcile_graph）。fetch 是这条查询比的那个口径取数节点（那一版里的 id）。 + 纯计算，同样的输入同一张图。""" + from app.ledger import derive + + prefix = derive.reconcile_prefix(pinned) + tail = tail_nodes(plan, card_id=f"{prefix}{pinned.node_id}", fetch_id=f"{prefix}{fetch}") + return derive.reconcile_graph(pinned, derive.ours_block(sql, tool), tail, inputs=inputs) + + +__all__ = ["CHECK_INPUTS", "MAX_COLUMNS", "MAX_ROWS", "NEEDS_PAIRS", "ORIGIN", "REFUSED", "Pair", "Plan", + "ReconcileRefused", "Request", "actions_of", "align_sql", "build", "plan_pairs", "reconcile", + "reconciled_runs", "reconciled_view", "tail_nodes"] diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index 6bfec90c..33e860c6 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -4,7 +4,9 @@ ledger 键、报告导出的片段面板，三处读同一份比对工件、按 遮罩按读的那一刻重滤（同证据面板，api/evidence._Masks）：绑定的值是这次运行查到过的遮罩值时不给值（也就不能拿它代填出具）， 差异条目里的 SQL 片段走 _Shown 同一套换法。工作流名按现在的取（工件里不存名字）。 -界面层不放逻辑：标签、整题判断、能做的动作（这一波还没有出具和对账，actions 是空的）都由这里给。 +界面层不放逻辑：标签、整题判断、能做的动作都由这里给。actions（第 3 波）：整题等价时按那份口径正式出具（能不能、能不能 +自动、不能的原因，ledger/issue.plan_issue，按看的这个人的权限判）；有相关口径的查询看差多少（口径对账，ledger/reconcile）。 +issued 是这次运行按口径出具过的正式运行，reconciled 是对过的账（各自的运行、状态）。 """ from __future__ import annotations @@ -122,10 +124,20 @@ 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 issue as ledger_issue + from app.ledger import reconcile as ledger_reconcile + content = recorded.content 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)] return { **base, "state": "matched", **_stamp(recorded), "calibers": [{**{k: c.get(k) for k in ("caliber", "workflow_id", "version", "level", "graph_hash")}, @@ -136,8 +148,9 @@ async def run_view(run_id: str) -> dict[str, Any]: "reports": _labels_view(content, hider, names), "primary": content.get("primary"), "whole": _whole_view(content.get("whole") or {}, hider, names), - # 能做的动作（按口径正式出具、对账）在第 3 波；这一波照实给空的 - "actions": [], + "actions": [*(p.action() for p in plans), *ledger_reconcile.actions_of(content, names, hider.hide)], + "issued": issued, + "reconciled": reconciled, } diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py index 885c2579..24cc969a 100644 --- a/backend/tests/test_ledger_api.py +++ b/backend/tests/test_ledger_api.py @@ -61,7 +61,8 @@ async def test_a_matched_run_gives_labels_and_the_whole_judgement(client, monkey body = await ledger_of(client, row.id) assert body["state"] == "matched" and body["note"] == "封存后比对，不在封存范围内" and body["post_seal"] is True assert body["trigger"] == "sealed" and body["retro"] is False and "retro_note" not in body - assert body["actions"] == [] and body["primary"] == "write" + # 第 3 波起整题等价时有「按口径出具」这一项（详细的在 tests/test_ledger_issue.py） + assert [a["kind"] for a in body["actions"]] == ["issue"] and body["primary"] == "write" [caliber] = body["calibers"] assert caliber["workflow_name"] == "周销售" and caliber["level_text"] == "受管" and caliber["caliber"] == f"{wf}@1" [label] = body["reports"]["write"]["labels"].values() diff --git a/backend/tests/test_ledger_issue.py b/backend/tests/test_ledger_issue.py new file mode 100644 index 00000000..9451e27c --- /dev/null +++ b/backend/tests/test_ledger_issue.py @@ -0,0 +1,306 @@ +"""按口径正式出具（next-round 02 做法 5，拍板 L1、L5，app/ledger/issue.py）。 + +真的跑探索运行（虚构店铺库、mock 模型写报告），等后台比对做完，看自动出具、一键出具、拒绝原因、去重： +- 整题独立等价于受管口径：比对完自动发起一次正式运行（来源 ledger_issue），origin 记来源运行、口径、绑定哈希，参数是 + 绑定 + 缺省值，按 graph_hash 钉住；GET 账本的 actions 写能自动、issued 列出它；再点一次交回同一次； +- 已发布级别、助手选用的：不自动，只给一键（写明为什么），一键照常发起；设置关了自动出具也不自动； +- 拒绝写明原因：整题不等价、点名别的口径、版本退役、图里有人工节点、缺入口字段、超过每日上限、使用者没有应用授权； + 使用者有授权（应用固定着这一版）就能出具，origin 记靠的是哪个应用； +- 设置 ledger 一组：缺省值、写坏了拒收。 +数据是虚构店铺库。 +""" +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.api.deps import Principal +from app.db.base import SessionLocal +from app.db.models import Run, Setting, WorkflowVersion +from app.engine.runner import run_manager +from app.ledger import issue as ledger_issue +from app.ledger import match as ledger_match +from app.main import app +from tests import ledger_samples as S + +ADMIN = Principal(username=None, role="admin", mode="off", via="off") + + +@pytest.fixture(autouse=True) +async def engine_up(tmp_path): + await run_manager.setup() + await S.fresh_source(tmp_path / "shop.db") + # 测试库按进程共用：每日上限数的是今天全部的按口径出具，别的用例发起的也算。每个用例从缺省设置起步、上限放宽， + # 要测上限的用例自己改 + await reset_ledger(issue_daily_limit=100_000) + yield + 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 reset_ledger(**values) -> None: + async with SessionLocal() as session: + row = await session.get(Setting, "ledger") + if row is not None: + await session.delete(row) + if values: + session.add(Setting(key="ledger", value=values)) + await session.commit() + + +async def set_ledger(**values) -> None: + async with SessionLocal() as session: + row = await session.get(Setting, "ledger") + merged = {**(row.value if row else {}), **values} + if row: + row.value = merged + else: + session.add(Setting(key="ledger", value=merged)) + await session.commit() + + +async def issued(run_id: str) -> list[Run]: + async with SessionLocal() as session: + return await ledger_issue.issued_runs(session, run_id) + + +async def settle(run_id: str) -> list[Run]: + """等自动出具发起的正式运行跑完。""" + # 自动出具在比对里就发起了（ledger/match.match_run 等它），drain 之后有没有已经定了：只等发起了的跑完 + for _ in range(200): + found = await issued(run_id) + if all(r.status in ("succeeded", "failed", "cancelled") for r in found): + return found + await asyncio.sleep(0.05) + return await issued(run_id) + + +def two_writers(monkeypatch, answer: str = "本周销售额 [[v:Q1.r0.销售额]]。") -> None: + """mock 模型：口径那张图的报告（指令「写周销售」）引用口径卡的指标，问数据那张图的报告照 answer 写。""" + from langchain_core.messages import AIMessage + + from app.providers import mock_model + + def decide(self, messages): + text = "\n".join(str(getattr(m, "content", "")) for m in messages) + return AIMessage(content="周销售 [[m:gmv]]。" if "写周销售" in text else answer) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + +async def equivalent_run(monkeypatch): + from app.engine.runner import run_manager as manager + + two_writers(monkeypatch) + run = await manager.start(graph=S.explore_graph(S.EQUIVALENT), input_payload={}, origin_kind="canvas") + row = await S.wait(run.id) + assert row.status == "succeeded", row.error + await ledger_match.drain() + return row + + +# -------------------------------------------------------------------------- +# 自动出具 +# -------------------------------------------------------------------------- + + +async def test_an_independent_equivalent_answer_to_a_governed_caliber_is_issued_automatically(client, monkeypatch): + wf = await S.publish() + row = await equivalent_run(monkeypatch) + [formal] = await settle(row.id) + assert formal.run_class == "formal" and formal.origin_kind == "ledger_issue" and formal.status == "succeeded" + assert (formal.workflow_id, formal.version) == (wf, 1) + assert formal.input == {"start": "2026-09-01", "end": "2026-09-08", "store": "all"} + async with SessionLocal() as session: + version = (await session.execute(select(WorkflowVersion).where( + WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one() + assert formal.version_hash == version.graph_hash + origin = formal.origin + assert origin["source_run"] == row.id and origin["caliber"] == f"{wf}@1" and origin["how"] == "independent" + assert origin["auto"] is True and origin["binding_hash"] and origin["match"] + + body = (await client.get(f"/api/runs/{row.id}/ledger")).json() + [action] = [a for a in body["actions"] if a["kind"] == "issue"] + assert action["available"] is True and action["auto"] is True and action["issued_run"] == formal.id + assert action["level_text"] == "受管" and action["workflow_name"] == "周销售" + [item] = body["issued"] + assert item["run_id"] == formal.id and item["auto"] is True and item["status"] == "succeeded" + + # 再点一次「按口径出具」：同一运行、同一口径、同一组参数，交回已经发起的那一次 + again = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1}) + assert again.status_code == 200 and again.json()["created"] is False and again.json()["run_id"] == formal.id + assert len(await issued(row.id)) == 1 + + +async def test_published_level_calibers_only_get_one_click(client, monkeypatch): + wf = await S.publish(level="published") + row = await equivalent_run(monkeypatch) + assert await settle(row.id) == [] + body = (await client.get(f"/api/runs/{row.id}/ledger")).json() + [action] = [a for a in body["actions"] if a["kind"] == "issue"] + assert action["available"] is True and action["auto"] is False and action["auto_reason"] == ledger_issue.AUTO_LEVEL + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1}) + assert res.status_code == 200 and res.json()["created"] is True + [formal] = await settle(row.id) + assert formal.origin["auto"] is False and formal.status == "succeeded" + assert res.json()["issued"]["run_id"] == formal.id + + +async def test_switching_auto_issue_off_keeps_one_click(client, monkeypatch): + await S.publish() + await set_ledger(auto_issue=False) + row = await equivalent_run(monkeypatch) + assert await settle(row.id) == [] + [action] = (await client.get(f"/api/runs/{row.id}/ledger")).json()["actions"] + assert action["auto"] is False and action["auto_reason"] == ledger_issue.AUTO_OFF + + +async def test_composed_answers_are_never_issued_automatically(monkeypatch): + from app.ledger import compose + from app.look.calibers import CaliberKind + + wf = await S.publish() + async with SessionLocal() as session: + kind = await CaliberKind.load(session, compose=True) + made = compose.Composer(list(kind.options)).compose( + {"op": "use_caliber", "id": "c1", "workflow_id": wf, "version": 1, "metrics": ["gmv"], + "inputs": {"start": "2026-09-01", "end": "2026-09-08"}}, taken={"start"}) + graph = S.chain(S.node("start", "input"), *made.nodes, S.node("write", "report", instructions="写答案"), + S.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}])) + two_writers(monkeypatch, "本周销售额 [[m:gmv]]。") + row = await S.wait((await run_manager.start(graph=graph, input_payload={}, origin_kind="chat")).id) + await ledger_match.drain() + assert await settle(row.id) == [] + async with SessionLocal() as session: + recorded = await ledger_match.store.latest(row.id) + plan = await ledger_issue.plan_issue(session, await session.get(Run, row.id), recorded.content, wf, 1, + principal=ADMIN) + assert plan.available and plan.how == "composed" and plan.auto_reason == ledger_issue.AUTO_COMPOSED + + +# -------------------------------------------------------------------------- +# 拒绝 +# -------------------------------------------------------------------------- + + +async def test_refusals_say_why(client, monkeypatch): + wf = await S.publish(level="published") + other = await S.publish(name="另一份口径", level="published") + unrelated = await S.explore(monkeypatch, S.WITH_REFUNDS) + res = await client.post(f"/api/ledger/answers/{unrelated.id}/issue", json={"workflow_id": wf, "version": 1}) + assert res.status_code == 409 and res.json()["code"] == ledger_issue.REFUSED + assert res.json()["detail"].startswith("整题没有等价于这份口径") + + row = await equivalent_run(monkeypatch) + whole_on = (await ledger_match.store.latest(row.id)).content["whole"]["workflow_id"] + named = other if whole_on == wf else wf + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": named, "version": 1}) + assert res.status_code == 409 and "整题等价的是另一份口径" in res.json()["detail"] + + # 版本退役 + async with SessionLocal() as session: + version = (await session.execute(select(WorkflowVersion).where( + WorkflowVersion.workflow_id == whole_on, WorkflowVersion.version == 1))).scalar_one() + version.retired_at = datetime.now(timezone.utc) + await session.commit() + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": whole_on, "version": 1}) + assert res.status_code == 409 and "已退役" in res.json()["detail"] + + +async def test_a_caliber_with_a_human_step_is_not_issued_on_anyones_behalf(client, monkeypatch): + graph = S.caliber_graph() + graph["nodes"].insert(3, S.node("ok", "human", mode="approve", prompt="确认")) + graph["edges"] = [{"source": a["id"], "target": b["id"]} for a, b in zip(graph["nodes"], graph["nodes"][1:])] + wf = await S.publish(graph=graph) + row = await equivalent_run(monkeypatch) + assert await settle(row.id) == [] + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1}) + assert res.status_code == 409 and "人工节点" in res.json()["detail"] and "不代人做这些事" in res.json()["detail"] + + +async def test_missing_entry_fields_are_named(client, monkeypatch): + graph = S.caliber_graph() + graph["nodes"][0]["data"]["config"]["fields"].append({"name": "region", "required": True}) + wf = await S.publish(graph=graph) + row = await equivalent_run(monkeypatch) + [action] = (await client.get(f"/api/runs/{row.id}/ledger")).json()["actions"] + assert action["available"] is False and "请填写「region」" in action["reason"] + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1}) + assert res.status_code == 409 and "入口字段还不齐" in res.json()["detail"] + + +async def test_the_daily_limit_counts_every_issue(client, monkeypatch): + wf = await S.publish(level="published") + await set_ledger(issue_daily_limit=0) + row = await equivalent_run(monkeypatch) + res = await client.post(f"/api/ledger/answers/{row.id}/issue", json={"workflow_id": wf, "version": 1}) + assert res.status_code == 429 and res.json()["code"] == ledger_issue.DAILY_LIMIT and "上限（0 次）" in res.json()["detail"] + + +async def test_users_issue_only_versions_pinned_by_apps_they_can_use(monkeypatch): + from app.db.models_apps import App, AppGrant, AppVersion + + wf = await S.publish(level="published") + row = await equivalent_run(monkeypatch) + async with SessionLocal() as session: + run = await session.get(Run, row.id) + run.started_by = "carol" + await session.commit() + carol = Principal(username="carol", role="user", verified=True, mode="local") + with pytest.raises(ledger_issue.IssueRefused) as e: + await ledger_issue.issue(row.id, wf, 1, principal=carol) + assert e.value.status_code == 403 and e.value.message == ledger_issue.ASK_BUILDER + # 别人的运行读不到：先说读不到 + dave = Principal(username="dave", role="user", verified=True, mode="local") + with pytest.raises(ledger_issue.IssueRefused) as e: + await ledger_issue.issue(row.id, wf, 1, principal=dave) + assert e.value.status_code == 403 and e.value.message == ledger_issue.NO_RUN_ACCESS + + async with SessionLocal() as session: + board = App(name="周销售看板", published_version=1, created_by="bob") + session.add(board) + await session.flush() + session.add(AppVersion(app_id=board.id, version=1, spec={}, spec_hash="x", + pins=[{"alias": "week", "workflow_id": wf, "version": 1}])) + session.add(AppGrant(app_id=board.id, principal_kind="user", principal="carol", role="use")) + await session.commit() + board_id = board.id + formal, created = await ledger_issue.issue(row.id, wf, 1, principal=carol) + assert created and formal.started_by == "carol" and formal.origin["granted_by_app"] == board_id + + +async def test_concurrent_requests_issue_once(monkeypatch): + wf = await S.publish(level="published") + row = await equivalent_run(monkeypatch) + results = await asyncio.gather(*(ledger_issue.issue(row.id, wf, 1, principal=ADMIN) for _ in range(3))) + assert sorted(created for _, created in results) == [False, False, True] + assert len({r.id for r, _ in results}) == 1 + + +# -------------------------------------------------------------------------- +# 设置 +# -------------------------------------------------------------------------- + + +async def test_ledger_settings_have_defaults_and_refuse_bad_values(client): + await reset_ledger() + body = (await client.get("/api/settings")).json() + assert body["ledger"] == {"auto_issue": True, "issue_daily_limit": 20, "match_timeout_s": 8} + bad = await client.put("/api/settings", json={"values": {"ledger": {"auto_issue": "yes"}}}) + assert bad.status_code == 422 and "自动按口径正式出具" in bad.json()["detail"] + bad = await client.put("/api/settings", json={"values": {"ledger": {"match_timeout_s": 600}}}) + assert bad.status_code == 422 and "口径比对时限" in bad.json()["detail"] + ok = await client.put("/api/settings", json={"values": {"ledger": {"auto_issue": False, "issue_daily_limit": 5, + "match_timeout_s": 3}}}) + assert ok.status_code == 200 and ok.json()["ledger"] == {"auto_issue": False, "issue_daily_limit": 5, + "match_timeout_s": 3} diff --git a/backend/tests/test_ledger_reconcile.py b/backend/tests/test_ledger_reconcile.py new file mode 100644 index 00000000..bf2baf06 --- /dev/null +++ b/backend/tests/test_ledger_reconcile.py @@ -0,0 +1,216 @@ +"""口径对账：差多少（next-round 02 做法 6，app/ledger/reconcile.py、ledger/derive.reconcile_graph）。 + +真的跑探索运行（虚构店铺库、mock 模型写报告）、等后台比对做完，再点开对账： +- 动作：有相关口径的查询给「看差多少」；参数是按「同一列、同一运算符」猜的，要先核对（needs_check）； +- 不核对直接发起回 409（要核对参数）；核对过的参数发起：派生图是确定的、过 validate、出口契约 metrics_from 指向对账卡、 + 不加报告；跑出来的这次、口径、差值、相对差都对，差值的出处链点得开；GET 账本列出对过的账；统计 E2 数到它； +- 认不出哪一列对哪个指标（口径的指标式不是取数的一格）：要人选，选了照记进 origin； +- 多行结果：被引用的行按维度列经合并查询对齐，每行一组对账指标；维度是遮罩列、合并键不唯一时照实拒绝。 +数据是虚构店铺库。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.db.base import SessionLocal +from app.db.models import Run +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, validate_graph +from app.ledger import match as ledger_match +from app.ledger import reconcile as ledger_reconcile +from app.main import app +from tests import ledger_samples as S + +BY_REGION = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}' GROUP BY region ORDER BY region") +BY_REGION_ALL = ("SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE paid_at >= '2026-09-01' " + "AND paid_at < '2026-09-08' GROUP BY region ORDER BY region") +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 ledger_of(client, run_id: str) -> dict: + res = await client.get(f"/api/runs/{run_id}/ledger") + assert res.status_code == 200, res.text + return res.json() + + +async def card_of(run_id: str, node_id: str = "reconcile") -> dict: + from app.api.evidence import _sealed, _sealed_cards + + sealed = await _sealed(run_id) + [(_, card)] = [(a, c) for n, a, c in _sealed_cards(sealed) if n == node_id] + return {m["id"]: m for m in card["metrics"]} + + +async def reconcile(client, run_id: str, body: dict) -> Run: + res = await client.post(f"/api/ledger/answers/{run_id}/reconcile", json=body) + assert res.status_code == 201, res.text + return await S.wait(res.json()["run_id"]) + + +async def test_reconcile_the_refunds_answer_against_the_weekly_caliber(client, monkeypatch): + wf = await S.publish() + row = await S.explore(monkeypatch, S.WITH_REFUNDS) + body = await ledger_of(client, row.id) + [action] = [a for a in body["actions"] if a["kind"] == "reconcile"] + assert action["workflow_id"] == wf and action["version"] == 1 and action["node_id"] == "fetch" + assert action["needs_check"] is True and action["check_reason"] == ledger_reconcile.CHECK_REASON + assert action["binding"] == [{"field": "end", "value": "2026-09-08"}, {"field": "start", "value": "2026-09-01"}] + query = action["query"] + + # 参数是猜的：不核对就发起，照实拒绝 + res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json={"query": query, "workflow_id": wf, + "version": 1}) + assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.CHECK_INPUTS + assert res.json()["detail"] == ledger_reconcile.CONFIRM_FIRST + + done = await reconcile(client, row.id, {"query": query, "workflow_id": wf, "version": 1, "inputs": CHECKED, + "confirm": True}) + assert done.status == "succeeded", done.error + assert done.origin_kind == "reconcile" and done.run_class == "exploratory" and done.workflow_id is None + origin = done.origin + assert origin["source_run"] == row.id and origin["query"] == query and origin["inputs_from"] == "checked" + assert origin["pairs"] == [{"ours": "gmv", "metric": "gmv"}] and origin["caliber"] == f"{wf}@1" + assert done.input == {"start": "2026-09-01", "end": "2026-09-08", "store": "all"} + + graph = done.graph + ids = [n["id"] for n in graph["nodes"]] + assert ids == ["derived_in", "ours_q1", "fetch", "card", "reconcile", "derived_out"] + nodes = {n["id"]: n for n in graph["nodes"]} + assert nodes["ours_q1"]["data"]["config"]["args"]["sql"] == S.WITH_REFUNDS + assert "{{ input.start }}" in nodes["fetch"]["data"]["config"]["args"]["sql"], "口径的取数保留模板，运行时按入口渲染" + assert nodes["derived_out"]["data"]["config"]["contract"] == {"metrics_from": "reconcile"} + assert not [n for n in graph["nodes"] if n["type"] in ("report", "llm", "agent")] + assert not [i for i in validate_graph(GraphSpec.model_validate(graph)).issues if i.level == "error"] + + metrics = await card_of(done.id) + assert metrics["ours_0"]["value"] == 221.0 and metrics["cal_0"]["value"] == 170.5 + assert metrics["diff_0"]["value"] == 50.5 and metrics["rel_0"]["value"] == 0.2962 + assert metrics["rel_0"]["rendered"] == "29.62%" and metrics["diff_0"]["rendered"] == "50.50元" + # 三个数各自点得开：差值的出处链追到两边的查询 + res = await client.get(f"/api/runs/{done.id}/evidence/metrics/reconcile/diff_0") + assert res.status_code == 200, res.text + inputs = [s for s in res.json()["chain"] if s["step"] == "input"] + assert {i["node_id"] for i in inputs} == {"ours_q1", "card"} + + listed = (await ledger_of(client, row.id))["reconciled"] + assert [r["run_id"] for r in listed] == [done.id] and listed[0]["inputs_from"] == "checked" + + # 统计 E2：对账运行数到了（它不进使用统计，E2 专门数它） + stats = (await client.get("/api/ledger/stats")).json()["e2"] + assert stats["launched"] >= 1 and stats["succeeded"] >= 1 + + +async def test_the_derived_graph_is_deterministic(): + from app.ledger import derive + + wf = await S.publish() + async with SessionLocal() as session: + pinned = await derive.load_pinned(session, wf, 1, "card") + plan = ledger_reconcile.Plan(pairs=[ledger_reconcile.Pair(ours="gmv", metric="gmv", index=0, name="销售额", + unit="元", decimals=2)]) + a = ledger_reconcile.build(pinned, sql=S.WITH_REFUNDS, tool=S.tool(), plan=plan, inputs=CHECKED, fetch="fetch") + b = ledger_reconcile.build(pinned, sql=S.WITH_REFUNDS, tool=S.tool(), plan=plan, inputs=CHECKED, fetch="fetch") + assert a.graph == b.graph and a.kind == "reconcile" + assert a.origin["from"][0] == {"kind": "executed", "node_id": "ours_q1"} and a.origin["from"][1]["node_id"] == "card" + + +async def test_metrics_that_are_not_a_plain_cell_need_a_person_to_pick(client, monkeypatch): + wf = await S.publish(graph=S.caliber_graph(expression="cell(nodes.fetch, 0, 'gmv') * 1")) + row = await S.explore(monkeypatch, S.WITH_REFUNDS) + [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"] + body = {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, "confirm": True} + res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json=body) + assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.NEEDS_PAIRS + bad = await client.post(f"/api/ledger/answers/{row.id}/reconcile", + json={**body, "pairs": [{"ours": "nope", "metric": "gmv"}]}) + assert bad.status_code == 422 and "没有列「nope」" in bad.json()["detail"] + done = await reconcile(client, row.id, {**body, "pairs": [{"ours": "gmv", "metric": "gmv"}]}) + assert done.status == "succeeded", done.error + assert done.origin["pairs_chosen"] is True + assert (await card_of(done.id))["diff_0"]["value"] == 50.5 + + +async def test_multi_row_answers_are_aligned_by_their_dimensions(client, monkeypatch): + wf = await S.publish(graph=S.caliber_graph(BY_REGION)) + row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。") + [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"] + done = await reconcile(client, row.id, {"query": action["query"], "workflow_id": wf, "version": 1, + "inputs": CHECKED, "confirm": True}) + assert done.status == "succeeded", done.error + assert done.origin["rows"] == [0, 1] and done.origin["pairs"] == [{"ours": "gmv", "caliber": "gmv"}] + align = next(n for n in done.graph["nodes"] if n["id"] == "align") + assert align["type"] == "merge" and align["data"]["config"]["inputs"] == {"o": "ours_q1", "c": "fetch"} + metrics = await card_of(done.id) + assert metrics["ours_0_0"]["name"] == "区域=east · gmv · 这次" + assert (metrics["ours_0_0"]["value"], metrics["cal_0_0"]["value"], metrics["diff_0_0"]["value"]) == (163.0, 150.5, 12.5) + assert (metrics["ours_1_0"]["value"], metrics["cal_1_0"]["value"], metrics["diff_1_0"]["value"]) == (58.0, 20.0, 38.0) + + +async def test_masked_dimensions_and_duplicate_keys_are_refused(client, monkeypatch, engine_up): + wf = await S.publish(graph=S.caliber_graph(BY_REGION)) + duplicated = ("SELECT region AS 区域, amount AS gmv FROM orders WHERE paid_at >= '2026-09-01' " + "AND paid_at < '2026-09-08' ORDER BY id") + row = await S.explore(monkeypatch, duplicated, text="第一单 [[v:Q1.r0.gmv]]，第二单 [[v:Q1.r1.gmv]]。") + [action] = [a for a in (await ledger_of(client, row.id))["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 == 409 and "有重复的行" in res.json()["detail"] + + await S.set_source(engine_up, options={"mask_columns": ["region"]}) + row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。") + [action] = [a for a in (await ledger_of(client, row.id))["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 == 409 and "设置了遮罩" in res.json()["detail"] + + +def test_align_sql_keeps_the_referenced_rows_in_order(): + pairs = [ledger_reconcile.Pair(ours="gmv", column="销售额")] + sql = ledger_reconcile.align_sql([("区域", "区域")], pairs, [("west",), ("o'east",)]) + assert sql == ('SELECT o."区域" AS "区域", o."gmv" AS "ours_0", c."销售额" AS "cal_0" FROM o LEFT JOIN c ON ' + 'o."区域" = c."区域" WHERE (o."区域" = \'west\') OR (o."区域" = \'o\'\'east\') ' + 'ORDER BY CASE WHEN (o."区域" = \'west\') THEN 0 WHEN (o."区域" = \'o\'\'east\') THEN 1 END') + + +async def test_users_reconcile_their_own_answers_read_only_and_masked(client, monkeypatch): + """拍板 L5：对账（只读的探索运行）使用者都可以，带上问数据同一套 MASK_RESULTS / READ_ONLY；别人的答案读不到就不给对。""" + from app.api.deps import Principal + from app.tools.datasource import MASK_RESULTS_OPTION, READ_ONLY_OPTION + + wf = await S.publish() + row = await S.explore(monkeypatch, S.WITH_REFUNDS) + async with SessionLocal() as session: + (await session.get(Run, row.id)).started_by = "carol" + await session.commit() + [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"] + req = ledger_reconcile.Request(query=action["query"], workflow_id=wf, version=1, inputs=CHECKED, confirm=True) + dave = Principal(username="dave", role="user", verified=True, mode="local") + with pytest.raises(ledger_reconcile.ReconcileRefused) as e: + await ledger_reconcile.reconcile(row.id, req, principal=dave) + assert e.value.status_code == 403 + carol = Principal(username="carol", role="user", verified=True, mode="local") + run = await ledger_reconcile.reconcile(row.id, req, principal=carol) + assert run.started_by == "carol" + assert run.options[MASK_RESULTS_OPTION] is True and run.options[READ_ONLY_OPTION] is True + done = await S.wait(run.id) + assert done.status == "succeeded", done.error