diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py new file mode 100644 index 00000000..ccbbabfb --- /dev/null +++ b/backend/app/engine/dual_check_facts.py @@ -0,0 +1,308 @@ +"""双盲复算的选择写进数据目录：审核人在不一致的指标上选了原流程或盲算，而两边只差一个方面、又正好对得上数据目录 +里的一项事实时，把被采纳那一边的写法记成人工确认过的目录项。 + +对得上的事实只有三种（data/catalog.py 的表级项）： + +- 两边按不同的时间列筛时间 → 业务日期（business_date）：被采纳的一边按哪一列筛，业务日期就记哪一列 +- 一边多一个只涉及底层表列的筛选条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 +- 聚合只差去不去重 → 去重规则（dedup）：按被采纳那一边去不去重写一句 + +**为什么要这么严。** 目录是给助手、db_schema 工具和 SQL 检查看的数据事实，人工确认过的项还会让 SQL 检查报错误级。 +只差一个方面时，人选的那一边就是对这一处的判断；差了好几处、或者差异说明里有兜底（未能归纳的差异），人选的是 +「整体上哪边对」，拆不出是哪一处对，写进去就是替人编了一条事实。两条 SQL 规范化后一致却算出不同的值（两次取数之间 +数据变了、结果被截断），差在数据不在写法，更没有可记的事实。 + +**怎么写。** 照对话提案的规矩核对（plan_patch：列在不在表结构里、取值格式），记为人工填写、已确认，备注写明来自哪次 +运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖，照实交回原因；已经是同样的说法就什么都 +不写（再写一遍也不升版本）。整个写入由节点放在 ask() 之后的一个 @task 里：审批恢复重放、接着运行时不会再写一遍。 + +catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。 +""" +from __future__ import annotations + +import logging +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import select + +from app.data import catalog +from app.db.base import SessionLocal +from app.db.models import DataSource + +logger = logging.getLogger(__name__) + +#: 能写的三项（表级） +FACT_ITEMS = ("business_date", "valid_filter", "dedup") +#: 审核人采纳的一边在差异说明里的位置：a 是原流程、b 是盲算（节点调用 explain_sql_diff 时按这个顺序给两边） +SIDE_LABEL = {"a": "原流程", "b": "盲算"} +#: 差异条目的方面（sql_semantics.FacetDiff.aspect）在说明里的叫法 +ASPECT_LABEL = { + "time": "时间列", "time_range": "时间范围", "filter": "筛选条件", "aggregate": "聚合方式", "expression": "选出的列", + "join": "关联", "tables": "查询的表", "set_op": "几支查询的合并方式", "group_by": "分组", "having": "分组后的筛选", + "limit": "取哪几行", "order": "排序", "parse": "能否解析", "other": "未能归纳的地方", +} +#: 聚合函数的说法（去重规则那一句用） +_VERB = {"count": "计数", "sum": "求和", "avg": "求平均"} +#: 写入时版本对不上（别人刚改过这张表）最多重试几次 +_RETRIES = 2 + + +@dataclass(frozen=True) +class Fact: + """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明。""" + + source: str + table: str + item: str + value: Any + metric: str + choice: str + + +def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None) -> tuple[Fact | None, str | None]: + """一个候选（审核人在某个指标上选了 a 或 b）→ (要写的目录项, None) 或 (None, 为什么不写)。 + + candidate：{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}, diff: {equivalent, fallback, + items: [FacetDiff.as_dict()]}}。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 + """ + name = str(candidate.get("metric") or "") + side = candidate.get("choice") + diff = candidate.get("diff") if isinstance(candidate.get("diff"), Mapping) else {} + if side not in SIDE_LABEL: + return None, f"指标「{name}」没有采用任何一边的结果，不写入数据目录" + if diff.get("equivalent"): + return None, f"指标「{name}」两边的 SQL 规范化后一致，差异来自数据而不是写法，不写入数据目录" + items = [i for i in diff.get("items") or [] if isinstance(i, Mapping)] + if diff.get("fallback") or len(items) != 1: + return None, (f"指标「{name}」两边的 SQL 不止差一个方面，或有未能归纳的差异，" + "无法确定审核人认可的是哪一处，不写入数据目录") + item = items[0] + aspect = str(item.get("aspect") or "") + if aspect == "time": + fact, why = _business_date(item, side) + elif aspect == "filter": + fact, why = _valid_filter(item, side, candidate.get("dialect")) + elif aspect == "aggregate": + fact, why = _dedup(item, side, candidate, schema_cache) + else: + return None, (f"指标「{name}」两边差在{ASPECT_LABEL.get(aspect, '其他地方')}，" + "对不上数据目录里的业务日期、有效记录条件或去重规则，不写入数据目录") + if why is not None: + return None, f"指标「{name}」{why}，不写入数据目录" + table, key, value = fact + resolved = catalog.resolve_table_name(schema_cache, table) + if resolved is None: + return None, f"指标「{name}」涉及的表「{table}」不在数据源的表结构里，不写入数据目录" + return Fact(source=str(candidate.get("source") or ""), table=resolved, item=key, value=value, metric=name, + choice=str(side)), None + + +def _business_date(item: Mapping[str, Any], side: str) -> tuple[tuple[str, str, Any] | None, str | None]: + detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} + mine = [c for c in detail.get(side) or [] if isinstance(c, Mapping) and c.get("table") and c.get("column")] + if not mine: + return None, f"被采纳的{SIDE_LABEL[side]}没有按时间筛选，没有可记的业务日期" + if len(mine) > 1: + return None, f"被采纳的{SIDE_LABEL[side]}按好几列筛时间，无法确定业务日期按哪一列" + return (str(mine[0]["table"]), "business_date", {"column": str(mine[0]["column"])}), None + + +def _valid_filter(item: Mapping[str, Any], side: str, dialect: Any) -> tuple[tuple[str, str, Any] | None, str | None]: + other = "b" if side == "a" else "a" + condition, rival = item.get(side), item.get(other) + if rival is not None: + if condition is None: + return None, f"多出筛选条件的是{SIDE_LABEL[other]}，被采纳的{SIDE_LABEL[side]}没有这个条件" + return None, "两边各有不同的筛选条件，不是一边多出一个条件" + if not isinstance(condition, str) or not condition.strip(): + return None, "差异说明里没有被采纳一边的筛选条件" + detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} + tables = {str(c.get("table")) for c in detail.get("columns") or [] if isinstance(c, Mapping) and c.get("table")} + if len(tables) != 1: + return None, "筛选条件不是只涉及一张表的列" + (table,) = tables + return (table, "valid_filter", _unqualified(condition.strip(), dialect)), None + + +def _unqualified(condition: str, dialect: Any) -> str: + """条件里的列去掉表名：有效记录条件记在这张表自己名下，写成 status <> 'refunded' 而不是 orders.status <> …。 + 解析不了就原样。""" + import sqlglot + from sqlglot import exp + + from app.data.sqlcheck import dialect_of + + read = dialect_of(dialect) if isinstance(dialect, str) else None + try: + tree = sqlglot.parse_one(condition, read=read) + except Exception: # noqa: BLE001 - 写法特殊、解析不了：差异说明里给人看的就是这一句，原样记 + return condition + if not any(col.table for col in tree.find_all(exp.Column)): + return condition + for col in tree.find_all(exp.Column): + col.set("table", None) + try: + return tree.sql(dialect=read) + except Exception: # noqa: BLE001 + return condition + + +def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], + schema_cache: Mapping[str, Any] | None) -> tuple[tuple[str, str, Any] | None, str | None]: + """聚合只差去不去重（同一个函数、同一个参数，一边写了 DISTINCT）。表从被采纳那一边的 SQL 里认：参数列属于哪张实表。""" + from app.data.sql_semantics import facets + + pair = _aggregate_pair(item.get("a"), item.get("b"), candidate.get("dialect")) + if pair is None: + return None, "两边的聚合不只是去不去重的差别" + func, distinct = pair + adopted = distinct[0] if side == "a" else distinct[1] + sql = (candidate.get("sql") or {}).get(side) if isinstance(candidate.get("sql"), Mapping) else None + found = facets(str(sql or ""), candidate.get("dialect"), schema=schema_cache) + hits = [a for a in found.aggregates if a.func == func and a.distinct == adopted and not a.scope] + columns = {ref for a in hits for ref in a.arg.columns} + if len(columns) != 1: + return None, "无法确定去重的是哪一张表的哪一列" + ((table, column),) = columns + if table is None: + # 参数列没认出属于哪张表（没给表结构、列没写表名）：整条语句只查了一张表时就是它 + names = list(dict.fromkeys(t.text for t in found.tables)) + if len(names) != 1: + return None, "无法确定去重的列属于哪一张表" + table = names[0] + verb = _VERB.get(func, func) + text = f"对 {column} {verb}时去重" if adopted else f"对 {column} {verb}时不去重" + return (str(table), "dedup", text), None + + +def _aggregate_pair(a: Any, b: Any, dialect: Any) -> tuple[str, tuple[bool, bool]] | None: + """两边的聚合写法 → (函数名, (A 去重吗, B 去重吗))；不是「同一个函数、同一个参数、只差 DISTINCT」返回 None。""" + import sqlglot + from sqlglot import exp + + from app.data.sqlcheck import dialect_of + + read = dialect_of(dialect) if isinstance(dialect, str) else None + + def shape(text: Any) -> tuple[str, str, bool] | None: + if not isinstance(text, str) or not text.strip(): + return None + try: + node = sqlglot.parse_one(text, read=read) + except Exception: # noqa: BLE001 + return None + if not isinstance(node, exp.AggFunc): + return None + arg = node.this + distinct = isinstance(arg, exp.Distinct) + if distinct: + inner = arg.expressions + if len(inner) != 1: + return None + arg = inner[0] + if arg is None: + return None + return node.key, arg.sql(dialect=read), distinct + + left, right = shape(a), shape(b) + if left is None or right is None or left[:2] != right[:2] or left[2] == right[2]: + return None + return left[0], (left[2], right[2]) + + +# -------------------------------------------------------------------------- +# 写入 +# -------------------------------------------------------------------------- + + +def note_of(fact: Fact, *, run_id: str, node_title: str) -> str: + """目录项上的备注：来自哪次运行、哪个节点、哪个指标、采纳了哪一边。""" + return (f"双盲复算（运行 {run_id}，节点「{node_title}」）：审核人在指标「{fact.metric}」上" + f"采用了{SIDE_LABEL.get(fact.choice, fact.choice)}的写法") + + +def _value_text(item: str, value: Any) -> str: + if item == "business_date" and isinstance(value, Mapping): + return str(value.get("column") or "") + return str(value) + + +async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str, + actor: str | None) -> dict[str, list[dict[str, Any]]]: + """把审核人的选择写进数据目录：{written: [{source, table, item}], skipped: [{reason}]}。永远不抛：数据源没了、 + 目录写不进去，都落在 skipped 里照实说。""" + written: list[dict[str, Any]] = [] + skipped: list[dict[str, Any]] = [] + for candidate in candidates: + name = str(candidate.get("source") or "") + try: + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none() + if row is None: + skipped.append({"reason": f"数据源「{name}」已不存在，指标「{candidate.get('metric')}」的选择未写入数据目录"}) + continue + fact, why = derive(candidate, row.schema_cache) + if fact is None: + skipped.append({"reason": why}) + continue + done, why = await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title), + actor=actor) + except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记 + logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", name) + skipped.append({"reason": f"指标「{candidate.get('metric')}」的选择写入数据目录时出错，未写入"}) + continue + if done: + written.append({"source": fact.source, "table": fact.table, "item": fact.item}) + else: + skipped.append({"reason": why}) + return {"written": written, "skipped": skipped} + + +async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]: + """写一项。(写了没有, 没写的原因)。""" + label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) + tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None + for _ in range(_RETRIES): + entry = await catalog.read_entry(session, row.id, fact.table) + notes = entry.notes if entry else {} + version = entry.version if entry else 0 + live = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None + live = live if live is not None and live.get("status") != "rejected" else None + value = fact.value + if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \ + and live["value"].get("column") == value.get("column"): + # 业务日期按哪一列是这次选出来的事实；目录里同一列另写了规则、时区的，原样留着 + value = live["value"] + changes = [{"path": fact.item, "value": value, "reason": note}] + plan = catalog.plan_patch(notes, changes, table=fact.table, tables=tables) + if plan.problems: + return False, f"「{fact.table}」的{label}未写入数据目录：{plan.problems[0]}" + change = plan.changes[0] + if change.state == "same": + return False, (f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」，" + "无需重复写入") + if change.before_status == "confirmed": + return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」，" + f"与这次指标「{fact.metric}」采纳的「{_value_text(fact.item, value)}」不同，未覆盖。" + "如需更改，请到数据目录中修改") + at = catalog.now_iso() + try: + merged = catalog.apply_patch(notes, changes, table=fact.table, tables=tables, at=at) + except catalog.CatalogInvalid as e: + return False, f"「{fact.table}」的{label}未写入数据目录：{e}" + # 不论原来是谁提的：这一项现在出自人的选择，记为人工填写、已确认，备注写明出处 + merged[fact.item] = catalog.make_item(change.after, "human", "confirmed", note=note, at=at) + try: + await catalog.write_entry(session, row.id, fact.table, merged, if_version=version, actor=actor) + except catalog.CatalogConflict: + await session.rollback() + continue + except catalog.CatalogInvalid as e: + return False, f"「{fact.table}」的{label}未写入数据目录：{e}" + return True, None + return False, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认" + + +__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "record_choices"] diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 5174ffd4..83fa9cb1 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -1,21 +1,988 @@ -"""双盲复算节点：另起一路、不看原结果独立算一遍，再和原结果逐项比对。 +"""双盲复算节点：正式出具前，口径卡上输入来自查询的指标，另起一个看不到原 SQL 的只读 agent 独立算一遍再比对。 -目前只是占位：图里放了这个节点也能运行，执行器原样跳过、记一条日志，不改任何状态。 -真正的复算和比对由后续工作包补上；比不上的项通过节点产出里的 issuance_gaps 交给成果节点的 -出具契约降档（见 nodes/io.py 的 _upstream_issuance_gaps）。 +**按输入复算。** 指标由口径卡的原式算出，原式本身确定、可复算（metrics 的 recompute_ok）；会错的是喂给它的那几个数： +查哪张表、按哪天归属、排没排除退款……所以盲算的单位是「一个来自查询快照的输入」。盲算 agent（engine/verify_agent.py） +只看得到指标名、口径说明（节点配置的 definitions）、这个输入在计算式里是哪一项、数据源的数据目录，看不到原 SQL 和 +原结果；它自己写只读 SQL，最后交一个格子引用，由代码从它自己的查询快照里取值。盲值是拿原式代入盲算的输入算出来的 +（engine/recompute.py，取整同口径卡），和原值按相对误差比，原值为 0 时按绝对误差。两个指标用到同一格的只盲算一次。 + +**一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）， +完整对照落成一件工件（审批 payload 经 runner._safe 会截断，长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用 +原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有缺口；b、neither 写进产出的 issuance_gaps，成果节点的出具 +契约据此降档（on_reject 为 fail 时节点直接失败）。两边只差一个方面、又对得上数据目录事实的，选择写进数据目录 +（engine/dual_check_facts.py）。盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口。 + +**重放。** 审批恢复时节点从头重放。读设置（发起时没快照的才读）、盲算连同比对、差异说明连同工件、写数据目录各在一个 +@task 里，重放时直接取 checkpoint 里的结果：不再调用模型、不再计费、不再写目录，事件也都在 task 里发、不会再发一遍； +task 外面的日志经 replay.once 发。task 的先后只取决于这次运行固定下来的东西（图、运行选项、上游的产出）。设置读发起 +时的快照（launch_run 写进 run.options["verify"]）：审批停下期间改了设置，恢复后照旧按发起时的。 + +**不进写作目录。** 盲算的查询发 tool.end（role: verify），随运行封存、证据面板能下钻；台账只记一条 kind: dual_check， +绝不写 kind: query——报告的写作目录按台账里的 query 编号，盲算的查询混进去，写作者就会引用它们。 """ from __future__ import annotations +import ast +import asyncio +import copy +import math +import re +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass from typing import Any +from langgraph.func import task +from sqlalchemy import select + +from app.core import artifact_store +from app.core.errors import describe_exception from app.core.events import EventType -from app.engine.context import NodeContext +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.engine import recompute +from app.engine.approval import read_decision +from app.engine.context import NodeContext, NodeError +from app.engine.evidence import build_catalog, is_number, next_exec +from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \ + parse_expression +from app.engine.replay import _in_graph, ask, once +from app.engine.schema import DUAL_CHECK_FIELD_LABEL, DUAL_CHECK_ON_REJECT, NodeType, dual_check_card from app.engine.state import GraphState -#: 占位期间的跳过原因：节点产出和日志里都写这一句 -PENDING_REASON = "双盲复算功能开发中，本节点暂未生效，已跳过" +#: 节点产出的状态：跳过、无需裁决、人已裁决 +SKIPPED, AGREED, RESOLVED = "skipped", "agreed", "resolved" +#: 单个指标的结果：一致、不一致、无法复算 +SAME, DIFFERENT, UNCHECKED = "agreed", "disagreed", "unchecked" +#: 审核人的三种选择：采用原流程、采用盲算、两边都不对 +CHOICES = ("a", "b", "neither") +#: 盲算触顶的原因里交给界面认的几种（dual_check_limit 日志的 limit）；max_steps 只由核查本身的 verify_limit 日志说 +LIMIT_CODES = ("max_queries", "timeout_s", "max_cost_usd", "daily_max_usd") +#: 输入的出处：口径卡的 cell() 取的格、Agent 经 cite_fields 交来的字段，都落在一份查询快照上 +QUERY_VIAS = ("tool_cell", "agent_field") +#: 差异说明里两边的叫法，先原流程后盲算：dual_check_facts 按 a / b 认这个顺序 +SIDES = ("原流程", "盲算") +#: 审批卡摘要里每个指标最多列几条差异 +SUMMARY_DIFFS = 3 +#: 一次 SQL 比较最多等多久：解析是纯计算，放在线程里，超时就给兜底说明 +DIFF_TIMEOUT_S = 15.0 +#: 交给盲算 agent 的数据目录最多这么多字（表清单、各表的说明分开算），再多就请它自己用 db_schema 查 +_INDEX_CHARS = 3_000 +_NOTES_CHARS = 6_000 +#: 交给盲算 agent 的同一行其他列最多几列、每个值最多几个字 +_DIMS = 6 +_DIM_CHARS = 40 +#: 卡上给容器输入记的大概（metrics._brief）：「[12 项]」「{3 个字段}」，写进计算式时不当成字面量 +_BRIEF = re.compile(r"^(?:\[\d+ 项\]|\{\d+ 个字段\})$") + +SKIP_DISABLED = "双盲复算已在设置中关闭，本节点直接放行" +SKIP_NODE_OFF = "本节点已关闭双盲复算，直接放行" +SKIP_NO_INPUTS = "口径卡上没有输入来自查询的指标，无需双盲复算，直接放行" +#: 两条 SQL 规范化后一致、算出的值却不同：差在数据（两次取数之间数据变了、结果被截断），不在写法 +SAME_SQL = "两条 SQL 规范化后一致，差异可能来自两次取数之间数据有变化或取数结果被截断" +SAME_SQL_ASPECT = "data" +DIFF_TIMEOUT = "两条 SQL 比较超时，未能归纳成具体差异，请对照原文" +DIFF_FAILED = "两条 SQL 比较时出错，未能归纳成具体差异，请对照原文" +NO_SQL = "有一边没有可比较的 SQL，请对照原文" +#: 交给成果节点的缺口（不带节点名：出具契约统一加「双盲复算「节点名」：」） +GAP_BLIND = "指标「{name}」双盲复算不一致，审核人选择了盲算结果" +GAP_NEITHER = "指标「{name}」双盲复算不一致，审核人判断两边都不对" +GAP_UNCHECKED = "指标「{name}」无法复算：{reason}" + +BLIND_SYSTEM = ( + "你在做双盲复算：另一条流程已经算出了口径卡上的一个指标，你要在看不到它的 SQL 和结果的情况下，按业务定义独立算出" + "计算式里的一个数，交给人和原流程比对。\n" + "- 业务含义只按下面给的指标名称、口径说明和数据目录理解。数据目录里人工确认过的说明（业务日期按哪一列、有效记录" + "条件、去重规则）要遵守，标「推断，未确认」的谨慎使用\n" + "- 时间范围、要排除哪些行、要不要去重，都按口径说明来；口径说明没写到的，按数据目录和最常见的业务理解，并在理由里" + "写明你的取舍\n" + "- 尽量用一条 SQL 直接算出要交的那个数（聚合放在 SQL 里做），最后交出它所在的格子" +) +BLIND_TASK_PROMPT = ( + "指标：{name}{unit}\n" + "口径说明：{definition}\n" + "计算式：{formula}\n" + "你要算的是计算式里的 {label}{whole}：原流程里它是一次查询结果中「{column}」列的值{dims}。只交这一个数。\n" + "数据源：{source}\n\n" + "{catalog}" +) +BLIND_UNIT_PROMPT = "（单位：{unit}）" +BLIND_NO_DEFINITION_PROMPT = "（未提供，按指标名称和数据目录理解）" +BLIND_WHOLE_PROMPT = "（这个指标就是它本身）" +BLIND_DIMS_PROMPT = "，所在行的其他列：{dims}" +BLIND_INDEX_PROMPT = "数据源里的表（表名｜中文名｜粒度）：\n{index}" +BLIND_NOTES_PROMPT = "数据目录（标「推断，未确认」的项是推断，使用前请核实）：\n{notes}" +BLIND_NO_CATALOG_PROMPT = "这个数据源还没有数据目录，请用 db_schema 工具查看表结构。" +BLIND_MORE_PROMPT = "\n…（其余的请用 db_schema 工具查看）" +#: 盲算交回的结构：一个数（交给模型时换成格子引用，由代码取值），verify_agent 另加理由 +BLIND_SCHEMA: dict[str, Any] = { + "type": "object", + "properties": {"value": {"type": "number", "description": "要你算的那个数"}}, + "required": ["value"], +} + + +@dataclass(frozen=True) +class _Options: + """这次复算按什么来：节点配置优先，没写的取发起时的核查设置快照。""" + + tolerance: float + on_reject: str + metrics: list[str] | None + definitions: dict[str, str] + max_queries: int | None + timeout_s: float | None + daily_max_usd: float | None async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]: - ctx.emit(EventType.LOG, level="info", code="node_pending", message=PENDING_REASON) - return {"nodes": {ctx.node.id: {"skipped": True, "reason": PENDING_REASON}}} + settings = await _settings(ctx) + part = settings.get("dual_check") if isinstance(settings.get("dual_check"), dict) else {} + switch = ctx.config.get("enabled") + if switch is False or (switch is not True and not part.get("enabled", True)): + await once(ctx, EventType.LOG, level="info", code="dual_check_skipped", reason="disabled", + message=SKIP_NODE_OFF if switch is False else SKIP_DISABLED) + return _skipped(ctx, "disabled") + + card_id, card = _card(state, ctx) + options = _options(ctx, part, settings) + targets = _targets(card, options, ctx) if card is not None else [] + if not targets: + await once(ctx, EventType.LOG, level="info", code="dual_check_skipped", reason="no_inputs", + message=SKIP_NO_INPUTS) + return _skipped(ctx, "no_inputs") + + checked = await _step("dual_check_blind", lambda: _check(state, ctx, targets, options, settings)) + rows = checked["rows"] + disagreed = [r for r in rows if r["status"] == DIFFERENT] + if not disagreed: + return _finish(state, ctx, card_id, card, checked, status=AGREED, gaps=_unchecked_gaps(rows)) + + review = await _step("dual_check_review", lambda: _review(ctx, checked)) + response = await ask(ctx, review["payload"], **review["payload"]) + # 谁裁决的跟着事件走，和人工审批节点一样 + await once(ctx, EventType.HUMAN_RESOLVED, response=response, actor=ctx.actor()) + choices, note = _choices(ctx, response, disagreed) + gaps = _gaps(rows, choices) + await once(ctx, EventType.LOG, level="info", code="dual_check_resolved", choices=choices, gaps=gaps, + message=_resolved_text(disagreed, choices)) + candidates = [{**review["facts"][r["metric_id"]], "choice": choices[r["metric_id"]]} + for r in disagreed if choices[r["metric_id"]] in ("a", "b") and r["metric_id"] in review["facts"]] + catalog = await _step("dual_check_catalog", lambda: _catalog(ctx, candidates)) if candidates else None + rejected = [r for r in disagreed if choices[r["metric_id"]] != "a"] + if rejected and options.on_reject == "fail": + # 目录照样写了：审核人的选择是对数据的判断，和这次运行停不停无关 + names = "、".join(f"「{r['name']}」" for r in rejected[:5]) + raise NodeError(ctx.node.id, f"双盲复算不一致，审核人没有采用原流程的结果（指标{names}）。" + f"「{DUAL_CHECK_FIELD_LABEL['on_reject']}」设为「判为失败」，运行已停止；" + "请按审核意见修正取数或口径后重新运行") + return _finish(state, ctx, card_id, card, checked, status=RESOLVED, gaps=gaps, choices=choices, note=note, + artifact=review["card_artifact"], catalog=catalog) + + +# -------------------------------------------------------------------------- +# 设置、口径卡、要复算的指标 +# -------------------------------------------------------------------------- + + +async def _step(name: str, work: Callable[[], Awaitable[Any]]) -> Any: + """在一个 @task 里做 work：审批恢复、接着运行时节点从头重放，结果直接取 checkpoint 里的，不再做一遍；work 里发的 + 事件也就只发一次。不在图里执行（单独调用执行器）时直接做。""" + if not _in_graph(): + return await work() + + @task(name=name) + async def step() -> Any: + return await work() + + return await step() + + +async def _settings(ctx: NodeContext) -> dict[str, Any]: + """这次运行的核查设置：发起时的快照（run.options["verify"]）。发起时没快照的（不经 launch_run 发起、子工作流里的 + 节点）读当前设置，读一次记进 checkpoint：重放时不再读，审批停下期间改了设置也不会让重放走上另一条路。""" + from app.api.settings import VERIFY_GROUP, verify_settings, verify_values + from app.engine.runner import RUN_OPTIONS_KEY + + options = ctx.run.extra.get(RUN_OPTIONS_KEY) + snapshot = options.get(VERIFY_GROUP) if isinstance(options, dict) else None + if isinstance(snapshot, dict): + return verify_values(snapshot) + + async def read() -> dict[str, Any]: + async with SessionLocal() as session: + return await verify_settings(session) + + return await _step("dual_check_settings", read) + + +def _card(state: GraphState, ctx: NodeContext) -> tuple[str, dict[str, Any] | None]: + """要复算的口径卡：(节点 id, 它这次的产出)。配置不对（发起前的校验已经拦下，这里兜底）报节点错误；卡没执行到 + （在没走到的分支上）产出为 None。""" + card_id, problem = dual_check_card(ctx.run.spec, ctx.node) + if problem or not card_id: + raise NodeError(ctx.node.id, problem or f"「{DUAL_CHECK_FIELD_LABEL['card']}」需要选择一个口径卡节点") + node = ctx.run.spec.node_map().get(card_id) + if node is None or node.type != NodeType.METRICS: + raise NodeError(ctx.node.id, f"「{DUAL_CHECK_FIELD_LABEL['card']}」选择的「{card_id}」不是口径卡节点，" + "请重新选择") + output = (state.get("nodes") or {}).get(card_id) + return card_id, output if isinstance(output, dict) and output.get("kind") == "metric_set" else None + + +def _options(ctx: NodeContext, part: Mapping[str, Any], settings: Mapping[str, Any]) -> _Options: + cfg = ctx.config + tolerance = cfg.get("tolerance") + if isinstance(tolerance, bool) or not isinstance(tolerance, (int, float)) or not math.isfinite(tolerance) \ + or tolerance < 0: + tolerance = part.get("tolerance") + if isinstance(tolerance, bool) or not isinstance(tolerance, (int, float)) or tolerance < 0: + tolerance = 0.0 + wanted = cfg.get("metrics") + definitions = cfg.get("definitions") if isinstance(cfg.get("definitions"), dict) else {} + return _Options( + tolerance=float(tolerance), + on_reject=cfg.get("on_reject") if cfg.get("on_reject") in DUAL_CHECK_ON_REJECT else "degrade", + metrics=[str(m) for m in wanted if isinstance(m, str) and m.strip()] if isinstance(wanted, list) else None, + definitions={str(k): str(v) for k, v in definitions.items() if isinstance(v, str) and v.strip()}, + max_queries=part.get("max_queries"), timeout_s=part.get("timeout_s"), + daily_max_usd=settings.get("daily_max_usd"), + ) + + +def _from_query(item: Any) -> bool: + return isinstance(item, dict) and item.get("via") in QUERY_VIAS and isinstance(item.get("artifact"), str) \ + and bool(item["artifact"]) and isinstance(item.get("path"), str) + + +def _number(value: Any) -> bool: + if not is_number(value): + return False + try: + return math.isfinite(value) + except OverflowError: + return False + + +def _targets(card: dict[str, Any], options: _Options, ctx: NodeContext) -> list[dict[str, Any]]: + """要复算的指标，按配置的先后（没配就按卡上的先后）：原值是数、至少一个输入来自查询快照。配置里写了卡上没有的 + 指标 ID 报节点错误（口径卡引用自别的工作流时，画图时查不出来）。""" + metrics = [m for m in card.get("metrics") or [] if isinstance(m, dict) and m.get("id") is not None] + by_id = {str(m["id"]): m for m in metrics} + if options.metrics: + if unknown := [m for m in options.metrics if m not in by_id]: + raise NodeError(ctx.node.id, f"「{DUAL_CHECK_FIELD_LABEL['metrics']}」中的「{'、'.join(unknown[:5])}」" + f"不是口径卡上的指标 ID，可选的指标 ID：{'、'.join(list(by_id)[:12])}") + chosen = [by_id[m] for m in dict.fromkeys(options.metrics)] + else: + chosen = metrics + return [m for m in chosen if _number(m.get("value")) and any(_from_query(i) for i in m.get("inputs") or [])] + + +def _skipped(ctx: NodeContext, reason: str) -> dict[str, Any]: + return {"nodes": {ctx.node.id: {"status": SKIPPED, "reason": reason}}} + + +# -------------------------------------------------------------------------- +# 盲算与比对（一个 @task） +# -------------------------------------------------------------------------- + + +class _Sources: + """一次复算里要读的查询快照和数据源：同一份只读一次。""" + + def __init__(self) -> None: + self.snapshots: dict[str, tuple[Any, str | None]] = {} + self.rows: dict[str, DataSource | None] = {} + + def snapshot(self, artifact: str) -> tuple[dict[str, Any] | None, str | None]: + if artifact not in self.snapshots: + try: + content = artifact_store.load(artifact) + self.snapshots[artifact] = (content, None if content is not None else "查询快照无法读取") + except ValueError: + self.snapshots[artifact] = (None, "查询快照与哈希不一致，疑似被修改") + content, why = self.snapshots[artifact] + if why is None and not (isinstance(content, dict) and isinstance(content.get("rows"), list)): + return None, "查询快照无法读取" + return content, why + + async def source(self, name: str) -> DataSource | None: + if name not in self.rows: + async with SessionLocal() as session: + self.rows[name] = (await session.execute( + select(DataSource).where(DataSource.name == name))).scalar_one_or_none() + return self.rows[name] + + async def hidden(self, artifact: str) -> frozenset[str]: + """这份快照要遮的列（小写）：查询当时记下的，加上数据源现在设的。宁可多遮。""" + from app.data.engine import masked_columns + + snap, _ = self.snapshot(artifact) + recorded = (snap or {}).get("mask_columns") + out = {c.lower() for c in recorded if isinstance(c, str)} if isinstance(recorded, list) else set() + name = (snap or {}).get("source") + if isinstance(name, str) and name: + row = await self.source(name) + out |= {c.lower() for c in masked_columns(row.options if row is not None else None)} + return frozenset(out) + + +def _loader(state: GraphState) -> tuple[Callable[[str], Any], dict[str, str]]: + """重算要的取工件函数和节点产出：nodes.* 的容器输入从这次运行的状态里取（口径卡执行时上游的产出就是这些）。""" + prefix = "state:" + outputs = {str(nid): f"{prefix}{nid}" for nid in state.get("nodes") or {}} + + def load(key: str) -> Any: + if key.startswith(prefix): + return (state.get("nodes") or {}).get(key[len(prefix):]) + return artifact_store.load(key) + + return load, outputs + + +async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, Any]], options: _Options, + settings: Mapping[str, Any]) -> dict[str, Any]: + """逐个指标认出能复算的输入、按输入盲算、代回原式比对。返回 {rows, results, model, same_model, usage}。 + + rows 每个指标一项：{metric_id, name, unit, decimals, status, reason?, original: {value, rendered}, blind: {value, + rendered}, delta_pct, inputs: [{path, key, value, sql, source, alias, artifact, row, column, node_id}], differing}； + results 是每个盲算（按格子去重）的结果，见 _slim。 + """ + sources = _Sources() + load, outputs = _loader(state) + aliases = _original_aliases(state) + rows: list[dict[str, Any]] = [] + jobs: dict[str, dict[str, Any]] = {} + for metric in targets: + row, metric_jobs = await _plan_metric(state, metric, options, sources, aliases, load, outputs) + rows.append(row) + for key, job in metric_jobs.items(): + # 两个指标用到同一格只盲算一次，按先出现的那个指标说清要算什么；先出现的没写口径说明、后一个写了，用后一个 + if key not in jobs or (not jobs[key]["definition"] and job["definition"]): + jobs[key] = job + results, model, same, usage = await _blind(ctx, jobs, options, settings, _producer_models(state, rows)) + for row, metric in zip(rows, targets): + if row["status"] != UNCHECKED: + _compare(row, metric, results, options, load, outputs, row.pop("_masks")) + else: + row.pop("_masks", None) + agreed = sum(1 for r in rows if r["status"] == SAME) + if agreed and not any(r["status"] == DIFFERENT for r in rows): + ctx.emit(EventType.LOG, level="info", code="dual_check_agreed", metrics=agreed, + message=f"双盲复算一致：{agreed} 个指标两边的结果相同，已放行") + return {"rows": rows, "results": results, "model": model, "same_model": same, "usage": usage} + + +def _original_aliases(state: GraphState) -> dict[str, str]: + """原流程的查询快照在这次运行的证据台账里的编号（Q1、Q2…，和报告取全部上游查询时的编号一致）。""" + ledger = [e for e in state.get("evidence") or [] if isinstance(e, dict)] + catalog = build_catalog(nodes={}, ledger=ledger) + return {str(entry.get("artifact")): alias for alias, entry in catalog.items() + if entry.get("kind") == "query" and entry.get("artifact")} + + +async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Options, sources: _Sources, + aliases: dict[str, str], load: Callable[[str], Any], outputs: dict[str, str] + ) -> tuple[dict[str, Any], dict[str, dict[str, Any]]]: + """一个指标：认出来自查询的输入能不能单独复算，能的话给每个输入建一个盲算任务。""" + row: dict[str, Any] = { + "metric_id": str(metric["id"]), "name": str(metric.get("name") or metric["id"]), + "unit": str(metric.get("unit") or ""), "decimals": metric.get("decimals"), + "status": "pending", "original": {"value": metric.get("value"), "rendered": metric.get("rendered")}, + "blind": {"value": None, "rendered": None}, "delta_pct": None, "inputs": [], "differing": [], + } + tops = _tops(metric) + masks: dict[str, frozenset[str]] = {} + problem: str | None = None + for item in metric.get("inputs") or []: + if not _from_query(item): + continue + info, why = await _query_input(item, sources, aliases, tops) + masks[item["artifact"]] = await sources.hidden(item["artifact"]) + if why: + problem = problem or why + continue + row["inputs"].append(info) + row["_masks"] = masks + if problem is None: + # 什么都不改先重算一遍：和卡上记的对不上（输入还原不了、用到了遮罩列），换了盲算的输入也算不准 + check = recompute.recompute(metric, recompute.Changes(), recompute.Sources( + load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks)))) + if check["status"] != "ok" or check["changed"]: + problem = f"无法按原表达式重算这个指标（{check.get('reason') or '重算结果与口径卡不一致'}）" + if problem: + row.update(status=UNCHECKED, reason=problem) + return row, {} + labels = {inp["path"]: f"输入{i}" for i, inp in enumerate(row["inputs"], 1)} + formula = _formula(metric, labels) + jobs: dict[str, dict[str, Any]] = {} + for inp in row["inputs"]: + snap, _ = sources.snapshot(inp["artifact"]) + jobs[inp["key"]] = { + "key": inp["key"], "metric_id": row["metric_id"], "metric": row["name"], "unit": row["unit"], + "definition": options.definitions.get(row["metric_id"]) or "", "formula": formula, + "label": labels[inp["path"]], "column": inp["column"], "source": inp["source"], + "dims": _dims(snap or {}, inp["row"], inp["column"], masks.get(inp["artifact"]) or frozenset()), + } + return row, jobs + + +async def _query_input(item: dict[str, Any], sources: _Sources, aliases: dict[str, str], tops: list[str] + ) -> tuple[dict[str, Any], str | None]: + """一个来自查询快照的输入：(盲算要用的信息, None) 或 (_, 为什么不能单独复算)。""" + from app.engine.merge_query import MERGE_SOURCE + + artifact = item["artifact"] + snap, why = sources.snapshot(artifact) + if why: + return {}, f"输入 {item['path']} 的{why}" + source = snap.get("source") + if source == MERGE_SOURCE: + return {}, f"输入 {item['path']} 来自合并查询（跨数据源合并的结果），盲算只能在单个数据源里查询" + if not isinstance(source, str) or not source: + return {}, f"输入 {item['path']} 的查询快照没有记下数据源" + locator = item.get("locator") if isinstance(item.get("locator"), dict) else {} + row, column = locator.get("row"), locator.get("column") + if isinstance(row, bool) or not isinstance(row, int) or row < 0 or not isinstance(column, str): + return {}, f"输入 {item['path']} 取的是一整段查询结果，盲算只能复核单个格子" + if item["path"] not in tops: + return {}, f"输入 {item['path']} 决定取哪一格，不能单独复算" + value = cell_value(item.get("value")) + if not _number(value): + return {}, f"输入 {item['path']} 不是数值" + if column.lower() in await sources.hidden(artifact): + return {}, f"输入 {item['path']} 取自设置了遮罩的列，不能用于复算" + found = await sources.source(source) + if found is None or not found.enabled: + return {}, f"数据源「{source}」不存在或已停用" + return {"path": item["path"], "key": f"{artifact}|{row}|{column}", "value": value, + "sql": str(snap.get("sql") or ""), "source": source, "alias": aliases.get(artifact), + "artifact": artifact, "row": row, "column": column, "node_id": item.get("node_id"), + "via": item["via"]}, None + + +def _tops(metric: dict[str, Any]) -> list[str]: + """表达式最外层的引用：只有它们的值直接参与计算，嵌在下标、cell() 行列里的决定取哪一格。""" + try: + tree, _, _ = parse_expression(str(metric.get("expression") or "")) + except ExpressionError: + return [] + tops, _ = recompute._refs(tree) + return tops + + +class _Swap(ast.NodeTransformer): + """把最外层的引用换成给人（给盲算 agent）看的写法：取值链、整个 cell() 调用，认法同 recompute._Bind。""" + + def __init__(self, shown: Callable[[str], ast.AST | None]) -> None: + self.shown = shown + + def visit_Call(self, node: ast.Call) -> ast.AST: + if _cell_call(node) and (swapped := self.shown(ast.unparse(node))) is not None: + return ast.copy_location(swapped, node) + return self.generic_visit(node) + + def _chain(self, node: ast.AST) -> ast.AST: + if _chain_root(node) is None: + return self.generic_visit(node) + swapped = self.shown(ast.unparse(node)) + return ast.copy_location(swapped, node) if swapped is not None else node + + visit_Attribute = visit_Subscript = visit_Name = _chain + + +def _formula(metric: dict[str, Any], labels: dict[str, str]) -> str: + """交给盲算 agent 的计算式：来自查询的输入写成「输入1」「输入2」，别的输入写当时代进去的值（容器写「其他输入」）。 + 不露出取数节点的 id、cell() 的行列：那是原流程的取法，不是业务定义。""" + expr = str(metric.get("expression") or "") + try: + tree, _, _ = parse_expression(expr) + except ExpressionError: + return "、".join(labels.values()) + values = {i.get("path"): i.get("value") for i in metric.get("inputs") or [] if isinstance(i, dict)} + + def shown(path: str) -> ast.AST | None: + if path in labels: + return ast.Name(id=labels[path], ctx=ast.Load()) + value = values.get(path) + if value is None or isinstance(value, (bool, int, float)) or ( + isinstance(value, str) and len(value) <= _DIM_CHARS and not _BRIEF.match(value)): + return ast.Constant(value=value) + return ast.Name(id="其他输入", ctx=ast.Load()) + + swapped = _Swap(shown).visit(copy.deepcopy(tree)) + return ast.unparse(ast.fix_missing_locations(swapped)) + + +def _dims(snapshot: dict[str, Any], row: int, column: str, hidden: frozenset[str]) -> list[str]: + """原流程那一格所在行的其他列里不是数的那些（地区、日期这类维度）：结果分了组时，盲算 agent 要知道算的是哪一组。 + 数不给（同一行别的指标值会带偏它），遮罩列不给。""" + rows = snapshot.get("rows") or [] + if not 0 <= row < len(rows): + return [] + record = rows[row] + columns = [str(c) for c in snapshot.get("columns") or []] + pairs = list(record.items()) if isinstance(record, dict) else list(zip(columns, record)) + out: list[str] = [] + for name, value in pairs: + name = str(name) + if name == column or name.lower() in hidden or value is None: + continue + kind = column_kind(snapshot, name) + if kind == "number" or (kind is None and (is_number(value) or numeric_text(value) is not None)): + continue + text = str(value).replace("\n", " ") + out.append(f"{name} 为「{text[:_DIM_CHARS]}」") + if len(out) >= _DIMS: + break + return out + + +async def _catalog_text(sources: _Sources, name: str) -> str: + """交给盲算 agent 的数据目录：表清单（表名、中文名、粒度）和有目录的表的业务说明（业务日期、有效记录条件、去重、 + 字段含义、码值、关联关系）。太长的截断，其余请它用 db_schema 查。""" + from app.data import catalog + + row = await sources.source(name) + if row is None: + return BLIND_NO_CATALOG_PROMPT + async with SessionLocal() as session: + entries = await catalog.read_catalog(session, row.id) + cache = row.schema_cache if isinstance(row.schema_cache, dict) else {} + tables = cache.get("tables") or {} + notes = {table: entry.notes for table, entry in entries.items()} + parts: list[str] = [] + if index := catalog.render_table_index(cache, notes): + parts.append(BLIND_INDEX_PROMPT.format(index=_clip(index, _INDEX_CHARS))) + blocks = [] + for table, body in notes.items(): + block = catalog.render_table_notes(body, meta=tables.get(table), system_notes=catalog.system_notes_source(row)) + # 每张表的说明去掉标题行，整段只在前面说一次「推断，未确认」的意思 + lines = block.split("\n", 1)[1] if "\n" in block else "" + if lines: + blocks.append(f"表 {table}：\n{lines}") + if blocks: + parts.append(BLIND_NOTES_PROMPT.format(notes=_clip("\n".join(blocks), _NOTES_CHARS))) + return "\n\n".join(parts) or BLIND_NO_CATALOG_PROMPT + + +def _clip(text: str, limit: int) -> str: + return text if len(text) <= limit else text[:limit].rsplit("\n", 1)[0] + BLIND_MORE_PROMPT + + +def _task_text(job: dict[str, Any], catalog: str) -> str: + return BLIND_TASK_PROMPT.format( + name=job["metric"], unit=BLIND_UNIT_PROMPT.format(unit=job["unit"]) if job["unit"] else "", + definition=job["definition"] or BLIND_NO_DEFINITION_PROMPT, formula=job["formula"], label=job["label"], + whole=BLIND_WHOLE_PROMPT if job["formula"].strip() == job["label"] else "", column=job["column"], + dims=BLIND_DIMS_PROMPT.format(dims="、".join(job["dims"])) if job["dims"] else "", + source=job["source"], catalog=catalog) + + +async def _blind(ctx: NodeContext, jobs: dict[str, dict[str, Any]], options: _Options, settings: Mapping[str, Any], + producers: set[str]) -> tuple[dict[str, Any], str | None, bool, dict[str, Any]]: + """按先后逐个盲算：(各格子的结果, 盲算模型, 是否和产出原数据的模型相同, 用量)。producers 是产出原数据的模型。""" + from app.api.settings import verify_model_spec + from app.engine.verify_agent import run_verify_agent + from app.tools.datasource import run_versions + + usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0, "cost_usd": 0.0, "calls": 0} + if not jobs: + return {}, None, False, usage + async with SessionLocal() as session: + spec = await verify_model_spec(session, ctx.config, settings=dict(settings)) + sources = _Sources() + catalogs: dict[str, str] = {} + results: dict[str, Any] = {} + model: str | None = None + same = False + broken: dict[str, Any] | None = None + for key, job in jobs.items(): + if broken is not None: + # 核查模型本身用不了（没配好、接入出错）：后面的盲算一样用不了，照这一次的原因记，不再一个个去试 + results[key] = broken + continue + if job["source"] not in catalogs: + catalogs[job["source"]] = await _catalog_text(sources, job["source"]) + outcome = await run_verify_agent( + spec=spec, system=BLIND_SYSTEM, task=_task_text(job, catalogs[job["source"]]), sources=[job["source"]], + output_schema=BLIND_SCHEMA, max_queries=options.max_queries, timeout_s=options.timeout_s, + daily_max_usd=options.daily_max_usd, emit=ctx.emit, run_id=ctx.run.run_id, node_id=ctx.node.id, + data_versions=run_versions(ctx.run)) + results[key] = _slim(outcome) + if outcome.error and not outcome.model: + broken = results[key] + for k in usage: + usage[k] += outcome.usage.get(k) or 0 + if outcome.model and model is None: + model = outcome.model + if model in producers: + same = True + ctx.emit(EventType.LOG, level="warn", code="dual_check_same_model", model=model, + message=f"盲算模型与产出原数据的模型相同（「{model}」），难以发现原流程自身的错误。" + "请在「设置 → 偏好设置 → 核查」中选择另一个核查模型，" + f"或在节点的「{DUAL_CHECK_FIELD_LABEL['model']}」中指定") + if outcome.limit in LIMIT_CODES: + ctx.emit(EventType.LOG, level="warn", code="dual_check_limit", limit=outcome.limit, metric=job["metric_id"], + message=f"指标「{job['metric']}」的盲算{_limit_text(outcome.limit, options)}，按无法复算处理") + usage["cost_usd"] = round(usage["cost_usd"], 6) + return results, model, same, usage + + +def _producer_models(state: GraphState, rows: list[dict[str, Any]]) -> set[str]: + """产出原数据的模型：要复算的输入是 Agent 经 cite_fields 交来的，那个 Agent 用的模型（节点产出里的 model）。调用 + 工具节点写死的 SQL 没有模型参与，不算。""" + outputs = state.get("nodes") or {} + out: set[str] = set() + for row in rows: + for inp in row["inputs"]: + found = outputs.get(inp.get("node_id")) if inp.get("via") == "agent_field" else None + if isinstance(found, dict) and isinstance(found.get("model"), str) and found["model"]: + out.add(found["model"]) + return out + + +def _slim(outcome: Any) -> dict[str, Any]: + """一次盲算的结果里要留进 checkpoint 的部分。""" + cell = next((c for c in outcome.cells if c.get("path") == "value"), None) or {} + query = next((q for q in outcome.queries if q.get("alias") == cell.get("alias")), None) or {} + value = cell.get("value") if cell.get("status") == "resolved" else None + reason = None + if outcome.error: + reason = outcome.error + elif cell.get("status") == "missing": + reason = "盲算没有查到这个数" + elif cell.get("status") == "unresolved": + reason = f"盲算交回的格子引用不可用（{cell.get('reason') or '没有交这个数'}）" + elif value is None: + reason = "盲算取到的那一格是空值" + elif _cell_masked(cell): + value, reason = None, "盲算取的是设置了遮罩的列，不采用" + return { + "value": value, "reason": reason, "limit": outcome.limit, "model": outcome.model, + "alias": cell.get("alias"), "ref": cell.get("ref"), "artifact": cell.get("artifact"), + "row": cell.get("row"), "column": cell.get("column"), "eid": cell.get("eid"), + "sql": query.get("sql"), "source": query.get("source"), + "rationale": (outcome.result or {}).get("rationale") or "", + "queries": [{k: q.get(k) for k in ("alias", "artifact", "sql", "row_count", "truncated", "source")} + for q in outcome.queries], + "notes": list(outcome.notes), + } + + +def _cell_masked(cell: dict[str, Any]) -> bool: + """盲算交的那一格在遮罩列上（查询当时数据源设的遮罩记在它自己的快照里）：值会进审批卡和对照工件，不能用。""" + if not cell.get("artifact"): + return False + try: + snap = artifact_store.load(str(cell["artifact"])) + except (ValueError, OSError): + return True + recorded = snap.get("mask_columns") if isinstance(snap, dict) else None + column = str(cell.get("column") or "").lower() + return isinstance(recorded, list) and any(isinstance(c, str) and c.lower() == column for c in recorded) + + +def _limit_text(limit: str | None, options: _Options) -> str: + """盲算为什么停下，接在「盲算」后面读：「盲算已达每项最多查询次数（6 次）」。""" + if limit == "max_queries" and options.max_queries is not None: + return f"已达每项最多查询次数（{options.max_queries} 次）" + if limit == "timeout_s" and options.timeout_s is not None: + return f"已达每项时长上限（{options.timeout_s:g} 秒）" + if limit == "daily_max_usd" and options.daily_max_usd is not None: + return f"已达核查每日金额上限（${float(options.daily_max_usd):g}）" + if limit in LIMIT_CODES: + return "已达核查的上限" + return "轮数已用完" + + +def _agree(original: Any, blind: Any, tolerance: float) -> bool: + """按相对误差比（原值为 0 时按绝对误差）。""" + if original == blind: + return True + if not (_number(original) and _number(blind)): + return False + if original == 0: + return abs(blind) <= tolerance + return abs(blind - original) <= tolerance * abs(original) + + +def _delta_pct(original: Any, blind: Any) -> float | None: + """相差比例，口径同运行对比：(盲算值 − 原流程值) ÷ |原流程值| × 100，两位小数；原流程值为 0 时为 None。""" + if not (_number(original) and _number(blind)) or original == 0: + return None + return round((blind - original) / abs(original) * 100, 2) + + +def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any], options: _Options, + load: Callable[[str], Any], outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> None: + """盲算的输入代回原式，和原值比。row 就地改成一致 / 不一致 / 无法复算。""" + values: dict[str, Any] = {} + for inp in row["inputs"]: + found = results.get(inp["key"]) + if found is None: + row.update(status=UNCHECKED, reason="盲算没有执行") + return + if found.get("limit"): + row.update(status=UNCHECKED, reason="盲算" + _limit_text(found["limit"], options)) + return + if found.get("value") is None: + row.update(status=UNCHECKED, reason=found.get("reason") or "盲算没有交回可用的数值") + return + values[inp["path"]] = found["value"] + outcome = recompute.recompute(metric, recompute.Changes(inputs=values), recompute.Sources( + load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks)))) + if outcome["status"] != "ok": + row.update(status=UNCHECKED, reason=f"用盲算的输入按原表达式算不出结果（{outcome.get('reason') or '无法计算'}）") + return + original = row["original"]["value"] + blind = outcome["new"] + row["blind"] = {"value": blind, "rendered": outcome.get("rendered_new")} + row["delta_pct"] = _delta_pct(original, blind) + if _agree(original, blind, options.tolerance): + row["status"] = SAME + return + row["status"] = DIFFERENT + # 对不上的输入：超出误差的；都没超出（误差经算式放大了）就取值不同的，再不行就全部 + differing = [i["key"] for i in row["inputs"] if not _agree(i["value"], values[i["path"]], options.tolerance)] + differing = differing or [i["key"] for i in row["inputs"] if i["value"] != values[i["path"]]] + row["differing"] = differing or [i["key"] for i in row["inputs"]] + + +def _unchecked_gaps(rows: list[dict[str, Any]]) -> list[str]: + return [GAP_UNCHECKED.format(name=r["name"], reason=r.get("reason") or "原因不明") + for r in rows if r["status"] == UNCHECKED] + + +# -------------------------------------------------------------------------- +# 不一致：差异说明、对照工件、审批（一个 @task） +# -------------------------------------------------------------------------- + + +async def _review(ctx: NodeContext, checked: dict[str, Any]) -> dict[str, Any]: + """不一致的指标逐个说清差在哪，完整对照落成工件，组好审批的 payload。返回 {payload, facts, card_artifact}。 + + facts 是每个不一致指标写数据目录要用的东西（dual_check_facts.derive 的候选，少一个 choice）：只有一个输入对不上时是 + 那一处完整的差异条目（带两边的写法和涉及的表、列），几个输入都对不上时只说「不止一处」。 + """ + rows = [r for r in checked["rows"] if r["status"] == DIFFERENT] + results = checked["results"] + sources = _Sources() + infos: dict[str, dict[str, Any]] = {} + metrics: list[dict[str, Any]] = [] + summary: list[dict[str, Any]] = [] + facts: dict[str, dict[str, Any]] = {} + for row in rows: + per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]] = [] + for inp in row["inputs"]: + if inp["key"] not in row["differing"]: + continue + if inp["source"] not in infos: + infos[inp["source"]] = await _source_info(sources, inp["source"]) + blind = results.get(inp["key"]) or {} + diff = await _explain(inp["sql"], blind.get("sql"), infos[inp["source"]]) + per_input.append((inp, blind, diff)) + merged = _merge(per_input) + metrics.append({ + "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], "decimals": row["decimals"], + "original": {**row["original"], "inputs": [ + {"path": i["path"], "value": i["value"], "sql": i["sql"], "source": i["source"], "alias": i["alias"]} + for i in row["inputs"]]}, + "blind": {**row["blind"], "inputs": [ + {"path": i["path"], "value": (results.get(i["key"]) or {}).get("value"), + "sql": (results.get(i["key"]) or {}).get("sql"), "source": (results.get(i["key"]) or {}).get("source"), + "alias": (results.get(i["key"]) or {}).get("alias"), + "rationale": (results.get(i["key"]) or {}).get("rationale") or ""} + for i in row["inputs"]]}, + "diff": {"equivalent": merged["equivalent"], + "items": [{"aspect": i.get("aspect"), "text": i.get("text")} for i in merged["items"]], + "fallback": merged["fallback"], "text_diff": merged["text_diff"]}, + }) + summary.append({ + "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], + "original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"], + "delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]], + "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]), + }) + first = per_input[0][0] if per_input else (row["inputs"][0] if row["inputs"] else {}) + single = per_input[0] if len(per_input) == 1 else None + facts[row["metric_id"]] = { + "metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"), + "dialect": (infos.get(str(first.get("source"))) or {}).get("kind"), + "sql": {"a": single[0]["sql"], "b": single[1].get("sql")} if single else {}, + "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None, + "items": merged["items"]}, + } + content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))} + try: + artifact = await artifact_store.put_json(content, kind="dual_check", run_id=ctx.run.run_id, + node_id=ctx.node.id, meta={"metrics": len(metrics)}) + except Exception as e: # noqa: BLE001 - 对照存不下，审核人就看不到完整的两条 SQL：不能让人凭摘要裁决 + raise NodeError(ctx.node.id, f"双盲复算的对照结果保存失败（{describe_exception(e)}），请稍后接着运行") from e + payload = { + "kind": "dual_check", "node_id": ctx.node.id, "mode": "input", + "title": f"双盲复算：{len(rows)} 个指标两边结果不一致，请选择采用哪一边的结果", + "card_artifact": artifact, "summary": summary, + } + names = "、".join(f"「{r['name']}」" for r in rows[:5]) + (f"等 {len(rows)} 个" if len(rows) > 5 else "") + ctx.emit(EventType.LOG, level="warn", code="dual_check_disagreed", + metrics=[{"id": r["metric_id"], "name": r["name"]} for r in rows], card_artifact=artifact, + message=f"双盲复算不一致：指标{names}两边的结果对不上，已暂停等待审核") + return {"payload": payload, "facts": facts, "card_artifact": artifact} + + +async def _source_info(sources: _Sources, name: str) -> dict[str, Any]: + """比较两条 SQL 要用的：方言、表结构、列的中文名（数据目录里的）。数据源没了就都不给，照样比原文。""" + from app.data import catalog + from app.data.sql_semantics import catalog_labels + + row = await sources.source(name) + if row is None: + return {"kind": None, "schema": None, "names": {}} + async with SessionLocal() as session: + entries = await catalog.read_catalog(session, row.id) + return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None, + "names": catalog_labels({table: entry.notes for table, entry in entries.items()})} + + +async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]: + """两条 SQL 的差异说明（SqlDiff.as_dict 的形状，条目带两边的写法和 detail）。规范化后一致的换成「差在数据」的 + 一句：值不同，写法却一样，具体差异一条都说不出。""" + from app.data.sql_semantics import explain_sql_diff + + if not isinstance(original, str) or not original.strip() or not isinstance(blind, str) or not blind.strip(): + return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": NO_SQL}], + "fallback": True, "text_diff": []} + try: + diff = await asyncio.wait_for(asyncio.to_thread( + explain_sql_diff, original, blind, info.get("kind"), schema=info.get("schema"), names=info.get("names"), + labels=SIDES), timeout=DIFF_TIMEOUT_S) + except Exception as e: # noqa: BLE001 - 比较本身出了意外（超时、规范化的缺陷）：给兜底说明，审核人对照原文 + text = DIFF_TIMEOUT if isinstance(e, asyncio.TimeoutError) else DIFF_FAILED + return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": text}], + "fallback": True, "text_diff": [f"- {original.strip()}", f"+ {blind.strip()}"]} + out = diff.as_dict() + if out["equivalent"]: + return {"equivalent": True, "items": [{"aspect": SAME_SQL_ASPECT, "a": None, "b": None, "text": SAME_SQL}], + "fallback": False, "text_diff": []} + return out + + +def _merge(per_input: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]) -> dict[str, Any]: + """一个指标几个对不上的输入的差异合成一份：一个输入时就是它的；几个时每条前面写明是哪一列的输入。""" + if not per_input: + return {"equivalent": False, "items": [{"aspect": "other", "a": None, "b": None, "text": NO_SQL}], + "fallback": True, "text_diff": []} + if len(per_input) == 1: + diff = per_input[0][2] + return {"equivalent": bool(diff["equivalent"]), "items": list(diff["items"]), + "fallback": bool(diff["fallback"]), "text_diff": list(diff.get("text_diff") or [])} + items: list[dict[str, Any]] = [] + lines: list[str] = [] + for inp, _, diff in per_input: + items += [{**i, "text": f"输入「{inp['column']}」：{i.get('text')}"} for i in diff["items"]] + lines += list(diff.get("text_diff") or []) + return {"equivalent": all(d["equivalent"] for _, _, d in per_input), "items": items, + "fallback": any(d["fallback"] for _, _, d in per_input), "text_diff": lines[:40]} + + +# -------------------------------------------------------------------------- +# 恢复之后：审核人的选择、缺口、数据目录 +# -------------------------------------------------------------------------- + + +def _choices(ctx: NodeContext, response: Any, rows: list[dict[str, Any]]) -> tuple[dict[str, str], str]: + """审核人的回复 → ({指标 id: a | b | neither}, 备注)。回复的 value 是 {choices, note}；某个指标没选、选了认不出的 + 按「两边都不对」：不替人认可任何一边。明确驳回（approved 为否）的按人工驳回让节点失败，和人工审批节点的填写模式 + 一样：这一步没有驳回的出口。""" + decision = read_decision(response) + if isinstance(response, dict) and decision.rejected: + raise NodeError(ctx.node.id, f"人工驳回：{decision.note or '未说明原因'}") + value = decision.value if isinstance(decision.value, dict) else response if isinstance(response, dict) else {} + raw = value.get("choices") if isinstance(value.get("choices"), dict) else {} + choices = {r["metric_id"]: raw[r["metric_id"]] if raw.get(r["metric_id"]) in CHOICES else "neither" for r in rows} + note = value.get("note") if isinstance(value.get("note"), str) and value.get("note").strip() else decision.note + return choices, str(note or "").strip() + + +def _gaps(rows: list[dict[str, Any]], choices: dict[str, str]) -> list[str]: + out: list[str] = [] + for row in rows: + if row["status"] == UNCHECKED: + out.append(GAP_UNCHECKED.format(name=row["name"], reason=row.get("reason") or "原因不明")) + elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) == "b": + out.append(GAP_BLIND.format(name=row["name"])) + elif row["status"] == DIFFERENT and choices.get(row["metric_id"]) != "a": + out.append(GAP_NEITHER.format(name=row["name"])) + return out + + +_CHOICE_TEXT = {"a": "采用原流程", "b": "采用盲算", "neither": "两边都不对"} + + +def _resolved_text(rows: list[dict[str, Any]], choices: dict[str, str]) -> str: + parts = "；".join(f"「{r['name']}」{_CHOICE_TEXT[choices[r['metric_id']]]}" for r in rows[:8]) + return f"审核人已裁决 {len(rows)} 个不一致的指标：{parts}" + ("等" if len(rows) > 8 else "") + + +async def _catalog(ctx: NodeContext, candidates: list[dict[str, Any]]) -> dict[str, Any]: + from app.data.catalog import TABLE_FIELD_LABEL + from app.engine.dual_check_facts import record_choices + + result = await record_choices(candidates, run_id=ctx.run.run_id, node_title=ctx.node.title, actor=ctx.actor()) + written, skipped = result["written"], result["skipped"] + parts = [] + if written: + parts.append("已写入数据目录：" + "、".join( + f"「{w['table']}」的{TABLE_FIELD_LABEL.get(w['item'], w['item'])}" for w in written)) + if skipped: + parts.append("未写入数据目录：" + "；".join(s["reason"] for s in skipped[:3]) + + (f"等 {len(skipped)} 条" if len(skipped) > 3 else "")) + ctx.emit(EventType.LOG, level="info", code="dual_check_catalog", written=written, skipped=skipped, + message="。".join(parts)) + return result + + +# -------------------------------------------------------------------------- +# 产出 +# -------------------------------------------------------------------------- + + +def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, Any] | None, checked: dict[str, Any], + *, status: str, gaps: list[str], choices: dict[str, str] | None = None, note: str = "", + artifact: str | None = None, catalog: dict[str, Any] | None = None) -> dict[str, Any]: + rows = checked["rows"] + picked = choices or {} + metrics = [{ + "metric_id": r["metric_id"], "name": r["name"], "unit": r["unit"], "status": r["status"], + "original_rendered": r["original"]["rendered"], "blind_rendered": r["blind"]["rendered"], + "delta_pct": r["delta_pct"], + **({"reason": r["reason"]} if r.get("reason") else {}), + **({"choice": picked[r["metric_id"]]} if r["metric_id"] in picked else {}), + } for r in rows] + result: dict[str, Any] = {"status": status, "metrics": metrics, "model": checked.get("model"), + "same_model": bool(checked.get("same_model"))} + if artifact: + result["card_artifact"] = artifact + if choices is not None: + result.update(choices=choices, note=note) + if catalog is not None: + result["catalog"] = catalog + if gaps: + result["issuance_gaps"] = gaps + entry: dict[str, Any] = { + "kind": "dual_check", "node_id": ctx.node.id, "exec": next_exec(state, ctx.node.id), "card": card_id, + "status": status, + "metrics": [{"id": m["metric_id"], "status": m["status"], **({"choice": m["choice"]} if "choice" in m else {})} + for m in metrics], + **({"metric_set": card["artifact"]} if card and isinstance(card.get("artifact"), str) else {}), + **({"artifact": artifact} if artifact else {}), + } + updates: dict[str, Any] = {"nodes": {ctx.node.id: result}, "evidence": [entry]} + usage = checked.get("usage") or {} + if usage.get("calls"): + updates["usage"] = dict(usage) + return updates + + +__all__ = ["AGREED", "CHOICES", "RESOLVED", "SKIPPED", "run_dual_check"] diff --git a/backend/tests/fixtures/dual_check_shop.py b/backend/tests/fixtures/dual_check_shop.py new file mode 100644 index 00000000..215b09ac --- /dev/null +++ b/backend/tests/fixtures/dual_check_shop.py @@ -0,0 +1,145 @@ +"""双盲复算测试用的虚构零售库和盲算模型的剧本。表名、数字、地名都是编的。 + +订单表埋了两个陷阱：退款行（status = 'refunded'），以及下单日和支付日不在同一周的单。2026-09-01 至 2026-09-07 这一周： + +- 按支付日、排除退款（口径说明要的）：1、2、5 号单，330.0 +- 按下单日、排除退款：2、4、5 号单，310.0 +- 按支付日、不排除退款：1、2、3、5 号单，380.0 +- 按下单日、不排除退款：2、3、4、5 号单，360.0 +- 付款客户数（按支付日、排除退款）：按行计数 3，去重计数 2 +""" +from __future__ import annotations + +import json +import re +import sqlite3 +import uuid +from pathlib import Path + +from langchain_core.messages import AIMessage, ToolMessage + +from app.data import introspect +from app.db.base import SessionLocal +from app.db.models import DataSource + +ROWS = [ + (1, "华东", 101, 100.0, "paid", "2026-08-31", "2026-09-01"), + (2, "华东", 102, 200.0, "paid", "2026-09-02", "2026-09-02"), + (3, "华南", 101, 50.0, "refunded", "2026-09-03", "2026-09-03"), + (4, "华南", 103, 80.0, "paid", "2026-09-07", "2026-09-08"), + (5, "华北", 102, 30.0, "paid", "2026-09-05", "2026-09-05"), +] +WEEK = "BETWEEN '2026-09-01' AND '2026-09-07'" +#: 口径说明要的写法：按支付日、排除退款 +RIGHT = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}" +#: 只错在时间列：按下单日 +BY_ORDER_DATE = f"SELECT SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK}" +#: 只错在没排除退款 +WITH_REFUNDS = f"SELECT SUM(amount) AS gmv FROM orders WHERE paid_date {WEEK}" +#: 两处都错 +BOTH_WRONG = f"SELECT SUM(amount) AS gmv FROM orders WHERE order_date {WEEK}" +#: 付款客户数：按行计数 / 去重计数 +BUYERS = f"SELECT COUNT(customer_id) AS buyers FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}" +BUYERS_DISTINCT = (f"SELECT COUNT(DISTINCT customer_id) AS buyers FROM orders " + f"WHERE status <> 'refunded' AND paid_date {WEEK}") +#: 按地区分组：华东一行是 300.0（1、2 号单） +BY_REGION = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} " + "GROUP BY region ORDER BY region") +DEFINITION = "已支付订单的金额合计：按支付日期（paid_date）归属到当周，不含退款" + + +def build(path: Path) -> Path: + db = sqlite3.connect(path) + db.executescript( + "CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, customer_id INTEGER, amount REAL, status TEXT, " + "order_date TEXT, paid_date TEXT);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?)", ROWS) + db.commit() + db.close() + return path + + +async def add_source(path: Path, *, options: dict | None = None) -> tuple[str, str]: + """登记成只读的手工源，探查好表结构。返回 (数据源名, id)。""" + name = f"dc_shop_{uuid.uuid4().hex[:8]}" + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=dict(options or {})) + session.add(row) + await session.flush() + row.schema_cache = await introspect.introspect(row) + await session.commit() + return name, row.id + + +def _text(messages) -> str: + return "\n".join(m.content if isinstance(m.content, str) else json.dumps(m.content, ensure_ascii=False) + for m in messages) + + +class Blind: + """盲算模型的剧本，也管图里别的模型调用。 + + plans：{任务里出现的指标名: (SQL, 要交的列) 或 [SQL, …]、列}。盲算循环里依次查这几条 SQL，收尾时交最后一条查询的 + 第 0 行那一列（row 可改）。没对上指标名的盲算交 null。不是盲算的调用（报告撰写）返回 writer 的正文并记下收到的消息。 + calls 记下每次盲算调用（step / final），tasks 记下盲算收到的任务原文。 + """ + + def __init__(self, monkeypatch, source: str, plans: dict, *, writer: str = "本周销售额见口径卡。", row: int = 0): + from app.providers import mock_model + + self.source = source + self.plans = plans + self.writer = writer + self.row = row + self.calls: list[str] = [] + self.tasks: list[str] = [] + self.writer_prompts: list[str] = [] + blind = self + + def decide(model, messages): + text = _text(messages) + plan = blind.plan_of(text) + if model.response_format: + blind.calls.append("final") + done = len(re.findall(r"【证据 Q\d+】", text)) + if plan is None or not done: + return AIMessage(content=json.dumps({"value": None, "rationale": "没有查到。"}, ensure_ascii=False)) + _, column = plan + alias = f"Q{len(blind.sqls(plan))}" + body = {"value": {"alias": alias, "row": blind.row, "column": column}, + "rationale": f"按口径说明统计，结果是 [[v:{alias}.r{blind.row}.{column}]]。"} + return AIMessage(content=json.dumps(body, ensure_ascii=False)) + if model.tools: + blind.calls.append("step") + # 盲算的任务是第一条用户消息；同一次盲算的几轮收到的是同一条，只记一次 + task_text = next((str(m.content) for m in messages if m.type == "human"), "") + if not blind.tasks or blind.tasks[-1] != task_text: + blind.tasks.append(task_text) + done = sum(isinstance(m, ToolMessage) for m in messages) + sqls = blind.sqls(plan) if plan is not None else [] + if done < len(sqls): + call = {"name": f"db_query__{blind.source}", "args": {"sql": sqls[done]}, + "id": f"blind-{uuid.uuid4().hex[:6]}"} + return AIMessage(content="查一下。", tool_calls=[call]) + return AIMessage(content="查完了。") + blind.writer_prompts.append(text) + return AIMessage(content=blind.writer) + + monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide) + + def plan_of(self, text: str): + for name, plan in self.plans.items(): + if f"指标：{name}" in text: + return plan + return None + + @staticmethod + def sqls(plan) -> list[str]: + sql = plan[0] + return list(sql) if isinstance(sql, (list, tuple)) else [sql] + + @property + def runs(self) -> int: + """盲算 agent 一共执行了几次（每次收尾交一次结果）。""" + return self.calls.count("final") diff --git a/backend/tests/test_dual_check_executor.py b/backend/tests/test_dual_check_executor.py new file mode 100644 index 00000000..7a1f3736 --- /dev/null +++ b/backend/tests/test_dual_check_executor.py @@ -0,0 +1,160 @@ +"""单独调用双盲复算的执行器（不在图里执行：@task 直接执行，once 直接发），测图里不好凑的几种情形。 + +- 盲算模型和产出原数据的 Agent 是同一个模型时提醒；调用工具节点写死的 SQL 没有模型参与，不提醒 +- 合并查询、整段结果、遮罩列、文本这类输入不能单独复算：记缺口，一分钱不花 +- 口径卡没执行到的直接放行；配置里写了卡上没有的指标 ID 报节点错误 + +查询快照是现造的工件，表名、数字都是虚构的。 +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +import pytest +from sqlalchemy import delete + +from app.core import artifact_store +from app.db.base import SessionLocal +from app.db.models import Setting +from app.engine.context import NodeContext, NodeError, RunContext +from app.engine.evidence import cell_eid +from app.engine.nodes.dual_check import run_dual_check +from app.engine.runner import RUN_OPTIONS_KEY +from app.engine.schema import GraphSpec +from app.tools.datasource import RUN_VERSIONS_KEY +from tests.fixtures.dual_check_shop import RIGHT, Blind, add_source, build +from tests.fixtures.sources import drop_source + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +@dataclass +class Recorder(NodeContext): + """记下节点发的事件。不在图里执行：@task 直接执行，once 直接发。""" + + events: list[tuple[str, dict[str, Any]]] = field(default_factory=list) + + def emit(self, event_type, **data): + self.events.append((str(event_type), data)) + + def logs(self, code: str) -> list[dict[str, Any]]: + return [d for t, d in self.events if t == "log" and d.get("code") == code] + + +@pytest.fixture +async def shop(tmp_path): + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key == "verify")) + await session.commit() + name, source_id = await add_source(build(tmp_path / "shop.db"), options={"mask_columns": "secret_gmv"}) + yield name + await drop_source(source_id) + + +async def snapshot(source: str, *, columns=("gmv",), rows=((330.0,),), sql=RIGHT, **extra) -> str: + content = {"columns": list(columns), "rows": [list(r) for r in rows], "row_count": len(rows), "truncated": False, + "sql": sql, "source": source, **extra} + return await artifact_store.put_json(content, kind="query_snapshot", run_id="unit", node_id="agent") + + +def context(graph: dict, node_id: str = "dual") -> Recorder: + parsed = GraphSpec.model_validate(graph) + run = RunContext(run_id="unit-run", thread_id="unit-run", spec=parsed, + extra={RUN_OPTIONS_KEY: {"verify": {}}, RUN_VERSIONS_KEY: None}) + return Recorder(node=parsed.node_map()[node_id], run=run) + + +AGENT_GRAPH = {"nodes": [node("start", "input"), node("agent", "agent", prompt="查销售额"), + node("card", "metrics", metrics=[{"id": "gmv", "expression": "vars.kpi.gmv"}]), + node("dual", "dual_check", "复核"), node("out", "output", fields=[])], + "edges": [{"source": a, "target": b} for a, b in + (("start", "agent"), ("agent", "card"), ("card", "dual"), ("dual", "out"))]} + + +def card_of(*metrics: dict) -> dict: + return {"kind": "metric_set", "caliber": "销售口径", "caliber_version": "v1", "metrics": list(metrics), + "artifact": "card-artifact"} + + +def metric(artifact: str, *, path: str = "vars.kpi.gmv", locator: dict | None = None, value: Any = 330.0, + via: str = "agent_field", node_id: str = "agent", expression: str | None = None) -> dict: + where = locator or {"row": 0, "column": "gmv"} + return {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands", "value": 330.0, + "rendered": "330.0元", "expression": expression or path, "substituted": "330.0", "status": "ok", + "inputs": [{"path": path, "value": value, "node_id": node_id, "via": via, "field": "gmv", + "artifact": artifact, "locator": where, "status": "verified", + **({"eid": cell_eid(artifact, where["row"], where["column"])} if "row" in where else {})}]} + + +async def test_the_same_model_as_the_agent_that_fetched_the_data_is_flagged(monkeypatch, shop): + art = await snapshot(shop) + Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")}) + ctx = context(AGENT_GRAPH) + state = {"nodes": {"agent": {"model": "mock-fast", "data": {"gmv": 330.0}}, "card": card_of(metric(art))}, + "evidence": [], "trail": []} + out = (await run_dual_check(state, ctx))["nodes"]["dual"] + assert out["status"] == "agreed" and out["same_model"] is True and out["model"] == "mock-fast" + [warned] = ctx.logs("dual_check_same_model") + assert warned["model"] == "mock-fast" and "盲算模型与产出原数据的模型相同" in warned["message"] + assert "设置 → 偏好设置 → 核查" in warned["message"] + + +async def test_data_fetched_by_fixed_sql_raises_no_same_model_warning(monkeypatch, shop): + art = await snapshot(shop) + Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")}) + ctx = context(AGENT_GRAPH) + state = {"nodes": {"agent": {"model": "mock-fast"}, + "card": card_of(metric(art, path="cell(nodes.fetch, 0, 'gmv')", via="tool_cell", + node_id="fetch"))}, + "evidence": [], "trail": []} + out = (await run_dual_check(state, ctx))["nodes"]["dual"] + assert out["status"] == "agreed" and out["same_model"] is False and not ctx.logs("dual_check_same_model") + + +@pytest.mark.parametrize("case, reason", [ + ("merge", "来自合并查询"), + ("rows", "取的是一整段查询结果"), + ("masked", "设置了遮罩的列"), + ("text", "不是数值"), +]) +async def test_inputs_that_cannot_be_checked_alone_become_gaps(monkeypatch, shop, case, reason): + blind = Blind(monkeypatch, shop, {"销售额": (RIGHT, "gmv")}) + if case == "merge": + art = await snapshot("合并查询", sources=[shop]) + item = metric(art) + elif case == "rows": + art = await snapshot(shop, rows=((100.0,), (230.0,))) + item = metric(art, path="vars.kpi.weeks", expression="sum(vars.kpi.weeks)", + locator={"rows": [0, 1], "column": "gmv"}, value="[2 项]") + elif case == "masked": + art = await snapshot(shop, columns=("secret_gmv",)) + item = metric(art, locator={"row": 0, "column": "secret_gmv"}) + else: + art = await snapshot(shop, columns=("gmv",), rows=(("三百三",),)) + item = metric(art, value="三百三") # 指标值是数，输入却是文本（比如拿它做了判断） + ctx = context(AGENT_GRAPH) + state = {"nodes": {"agent": {"model": "other"}, "card": card_of(item)}, "evidence": [], "trail": []} + out = (await run_dual_check(state, ctx))["nodes"]["dual"] + [row] = out["metrics"] + assert row["status"] == "unchecked" and reason in row["reason"] + assert out["issuance_gaps"] == [f"指标「销售额」无法复算：{row['reason']}"] + assert blind.calls == [], "不能单独复算的输入也花了钱" + + +async def test_a_card_that_did_not_run_passes_through(shop): + ctx = context(AGENT_GRAPH) + out = await run_dual_check({"nodes": {}, "evidence": [], "trail": []}, ctx) + assert out == {"nodes": {"dual": {"status": "skipped", "reason": "no_inputs"}}} + assert [d["reason"] for d in ctx.logs("dual_check_skipped")] == ["no_inputs"] + + +async def test_unknown_metric_ids_fail_the_node(shop): + art = await snapshot(shop) + graph = {**AGENT_GRAPH, "nodes": [*AGENT_GRAPH["nodes"][:3], node("dual", "dual_check", "复核", metrics=["refund"]), + AGENT_GRAPH["nodes"][4]]} + ctx = context(graph) + with pytest.raises(NodeError, match="「复算的指标」中的「refund」不是口径卡上的指标 ID"): + await run_dual_check({"nodes": {"card": card_of(metric(art))}, "evidence": [], "trail": []}, ctx) diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py new file mode 100644 index 00000000..286ec05d --- /dev/null +++ b/backend/tests/test_dual_check_facts.py @@ -0,0 +1,198 @@ +"""双盲复算的选择写进数据目录（engine/dual_check_facts.py）。 + +- 只差一个方面、又对得上目录事实时才写：时间列 → 业务日期，一边多一个只涉及一张表的筛选条件 → 有效记录条件， + 聚合只差去不去重 → 去重规则；按被采纳那一边的写法记 +- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法：不写，照实说原因 +- 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖， + 推断的不同说法照改 + +差异条目一律由 sql_semantics.explain_sql_diff 真算出来，和节点交给审核人看的是同一份。表名、数字都是虚构的。 +""" +from __future__ import annotations + +import pytest + +from app.data import catalog +from app.data.sql_semantics import explain_sql_diff +from app.db.base import SessionLocal +from app.engine.dual_check_facts import derive, record_choices +from tests.fixtures.dual_check_shop import ( + BOTH_WRONG, + BUYERS, + BUYERS_DISTINCT, + BY_ORDER_DATE, + RIGHT, + WITH_REFUNDS, + add_source, + build, +) +from tests.fixtures.sources import drop_source + +SCHEMA = {"tables": {"orders": {"name": "orders", "columns": [ + {"name": "id", "type": "INTEGER"}, {"name": "region", "type": "TEXT"}, {"name": "customer_id", "type": "INTEGER"}, + {"name": "amount", "type": "REAL"}, {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "TEXT"}, + {"name": "paid_date", "type": "TEXT"}], "primary_key": ["id"]}}} + + +def candidate(original: str, blind: str, choice: str, *, metric: str = "销售额", source: str = "shop") -> dict: + diff = explain_sql_diff(original, blind, "sqlite", schema=SCHEMA, labels=("原流程", "盲算")).as_dict() + return {"metric": metric, "choice": choice, "source": source, "dialect": "sqlite", + "sql": {"a": original, "b": blind}, "diff": diff} + + +# -------------------------------------------------------------------------- +# 差异 → 目录事实 +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("choice, column", [("b", "paid_date"), ("a", "order_date")]) +def test_a_different_time_column_becomes_the_business_date(choice, column): + fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, choice), SCHEMA) + assert why is None + assert (fact.table, fact.item, fact.value, fact.choice) == ("orders", "business_date", {"column": column}, choice) + + +def test_an_extra_filter_on_the_adopted_side_becomes_the_valid_filter(): + fact, why = derive(candidate(WITH_REFUNDS, RIGHT, "b"), SCHEMA) + assert why is None and (fact.item, fact.value) == ("valid_filter", "status <> 'refunded'") + # 原流程多出这个条件、审核人采纳原流程：同样记这个条件 + fact, why = derive(candidate(RIGHT, WITH_REFUNDS, "a"), SCHEMA) + assert why is None and (fact.item, fact.value) == ("valid_filter", "status <> 'refunded'") + + +def test_the_adopted_side_without_the_extra_filter_has_nothing_to_record(): + fact, why = derive(candidate(RIGHT, WITH_REFUNDS, "b"), SCHEMA) + assert fact is None and "被采纳的盲算没有这个条件" in why and "销售额" in why + + +def test_table_names_are_dropped_from_the_recorded_filter(): + qualified = RIGHT.replace("FROM orders WHERE status", "FROM orders o WHERE o.status") \ + .replace("paid_date", "o.paid_date") + fact, why = derive(candidate(WITH_REFUNDS, qualified, "b"), SCHEMA) + assert why is None and fact.value == "status <> 'refunded'" + + +@pytest.mark.parametrize("choice, text", [("b", "对 customer_id 计数时去重"), ("a", "对 customer_id 计数时不去重")]) +def test_counting_with_or_without_distinct_becomes_the_dedup_rule(choice, text): + fact, why = derive(candidate(BUYERS, BUYERS_DISTINCT, choice, metric="付款客户数"), SCHEMA) + assert why is None and (fact.table, fact.item, fact.value) == ("orders", "dedup", text) + + +def test_more_than_one_difference_is_not_recorded(): + fact, why = derive(candidate(BOTH_WRONG, RIGHT, "b"), SCHEMA) + assert fact is None and "不止差一个方面" in why + + +def test_equivalent_sql_or_fallback_or_other_aspects_are_not_recorded(): + same = candidate(RIGHT, RIGHT.replace("SUM(amount) AS gmv", "SUM(amount) AS total"), "b") + assert same["diff"]["equivalent"] is True + assert "规范化后一致" in derive(same, SCHEMA)[1] + + fallback = candidate(BY_ORDER_DATE, RIGHT, "b") + fallback["diff"]["fallback"] = True + assert "未能归纳" in derive(fallback, SCHEMA)[1] + + joined = candidate(RIGHT, RIGHT.replace("SUM(amount)", "MAX(amount)"), "b") + assert [i["aspect"] for i in joined["diff"]["items"]] == ["aggregate"] + assert "不只是去不去重" in derive(joined, SCHEMA)[1] + + ranged = candidate(RIGHT, RIGHT.replace("'2026-09-07'", "'2026-09-06'"), "b") + assert [i["aspect"] for i in ranged["diff"]["items"]] == ["time_range"] + assert "时间范围" in derive(ranged, SCHEMA)[1] + + assert "没有采用任何一边" in derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "choice": "neither"}, SCHEMA)[1] + + +def test_a_table_missing_from_the_schema_is_not_recorded(): + fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), {"tables": {"refunds": {"columns": []}}}) + assert fact is None and "不在数据源的表结构里" in why + + +# -------------------------------------------------------------------------- +# 写进目录 +# -------------------------------------------------------------------------- + + +@pytest.fixture +async def shop(tmp_path): + name, source_id = await add_source(build(tmp_path / "shop.db")) + yield name, source_id + await drop_source(source_id) + + +async def entry_of(source_id: str): + async with SessionLocal() as session: + return await catalog.read_entry(session, source_id, "orders") + + +async def seed(source_id: str, notes: dict) -> None: + async with SessionLocal() as session: + await catalog.write_entry(session, source_id, "orders", notes, if_version=0, actor="数据组") + + +async def test_a_choice_is_written_as_a_confirmed_human_item_with_its_origin(shop): + name, source_id = shop + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-42", + node_title="复核销售额", actor="张三") + assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []} + entry = await entry_of(source_id) + item = entry.notes["business_date"] + assert item["value"] == {"column": "paid_date"} and (item["source"], item["status"]) == ("human", "confirmed") + assert "run-42" in item["note"] and "「复核销售额」" in item["note"] and "「销售额」" in item["note"] + assert "盲算" in item["note"] and entry.version == 1 and entry.updated_by == "张三" + + +async def test_writing_the_same_choice_again_changes_nothing(shop): + name, source_id = shop + choice = [candidate(BY_ORDER_DATE, RIGHT, "b", source=name)] + await record_choices(choice, run_id="run-1", node_title="复核", actor=None) + again = await record_choices(choice, run_id="run-2", node_title="复核", actor=None) + assert again["written"] == [] and "无需重复写入" in again["skipped"][0]["reason"] + entry = await entry_of(source_id) + assert entry.version == 1 and "run-1" in entry.notes["business_date"]["note"] + + +async def test_a_confirmed_different_item_is_kept(shop): + name, source_id = shop + await seed(source_id, {"business_date": catalog.make_item({"column": "order_date", "rule": "按下单日"}, "human", + "confirmed")}) + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1", + node_title="复核", actor=None) + assert out["written"] == [] + reason = out["skipped"][0]["reason"] + assert "已有人工确认过的说法「order_date」" in reason and "paid_date" in reason and "未覆盖" in reason + entry = await entry_of(source_id) + assert entry.version == 1 and entry.notes["business_date"]["value"]["column"] == "order_date" + + +async def test_an_inferred_different_item_is_replaced(shop): + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("status = 'paid'", "llm")}) + out = await record_choices([candidate(WITH_REFUNDS, RIGHT, "b", source=name)], run_id="run-1", + node_title="复核", actor=None) + assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}] + item = (await entry_of(source_id)).notes["valid_filter"] + assert item["value"] == "status <> 'refunded'" and (item["source"], item["status"]) == ("human", "confirmed") + + +async def test_the_same_column_inferred_is_confirmed_and_keeps_its_rule(shop): + """目录里推断的业务日期正是这一列（还写了规则、时区）：确认它，规则、时区原样留着。""" + name, source_id = shop + await seed(source_id, {"business_date": catalog.make_item( + {"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}, "profile")}) + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1", + node_title="复核", actor=None) + assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}] + item = (await entry_of(source_id)).notes["business_date"] + assert item["value"] == {"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"} + assert (item["source"], item["status"]) == ("human", "confirmed") + + +async def test_a_missing_source_or_an_unrecordable_choice_is_reported(shop): + name, _ = shop + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source="no_such_source"), + candidate(BOTH_WRONG, RIGHT, "b", source=name)], run_id="run-1", + node_title="复核", actor=None) + assert out["written"] == [] + assert "数据源「no_such_source」已不存在" in out["skipped"][0]["reason"] + assert "不止差一个方面" in out["skipped"][1]["reason"] diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py new file mode 100644 index 00000000..7f3bc783 --- /dev/null +++ b/backend/tests/test_dual_check_node.py @@ -0,0 +1,571 @@ +"""双盲复算节点（engine/nodes/dual_check.py）：真跑一遍图。虚构的 SQLite 订单库埋了退款行、下单日和支付日两个陷阱 +（tests/fixtures/dual_check_shop.py），盲算模型是 MockChatModel 剧本。 + +- 一致：不挂起，产出 agreed，日志 dual_check_agreed；盲算的查询发 tool.end（role: verify），台账只有一条 dual_check， + 用量计入运行；盲算 agent 的任务里没有原 SQL +- 不一致：挂起，审批 payload 和对照工件按约定；a / b / neither 三种选择的产出和缺口，on_reject 为 fail 时失败 +- 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖 +- 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询 +- 触顶、分组结果的维度、两个指标用到同一格只盲算一次 +""" +from __future__ import annotations + +import asyncio + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, select + +from app.core import artifact_store +from app.data import catalog +from app.db.base import SessionLocal +from app.db.models import Approval, Run, RunEvent, Setting +from app.engine import spend +from app.engine.launch import launch_run +from app.engine.runner import run_manager +from app.main import app +from tests.fixtures.dual_check_shop import ( + BOTH_WRONG, + BY_ORDER_DATE, + BY_REGION, + DEFINITION, + RIGHT, + WEEK, + WITH_REFUNDS, + Blind, + add_source, + build, +) +from tests.fixtures.sources import drop_source + +#: 和 RIGHT 等价、写法不同（别名、列名都换了），值也是 330.0：报告的写作目录里若混进了盲算的查询，认得出来 +BLIND_TOTAL = f"SELECT SUM(o.amount) AS blind_total FROM orders o WHERE o.status <> 'refunded' AND o.paid_date {WEEK}" +GMV = {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands", + "expression": "cell(nodes.fetch, 0, 'gmv')"} + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture(autouse=True) +async def clean_settings(): + async def clear() -> None: + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_(["verify", spend.SPEND_KEY]))) + await session.commit() + + await clear() + yield + await clear() + + +@pytest.fixture +async def shop(tmp_path): + name, source_id = await add_source(build(tmp_path / "shop.db")) + yield name, source_id + await drop_source(source_id) + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def graph(source: str, sql: str, *, dual: dict | None = None, metrics: list | None = None, report: bool = False): + """取数 → 口径卡 → 双盲复算 →（报告）→ 成果。成果的叙述里没有数字：不加双盲复算时是完整出具。""" + nodes = [ + node("start", "input"), + node("fetch", "tool", tool=f"db_query__{source}", args={"sql": sql}), + node("card", "metrics", caliber="销售口径", caliber_version="v1", metrics=metrics or [GMV]), + node("dual", "dual_check", "复核销售额", **{"definitions": {"gmv": DEFINITION}, **(dual or {})}), + ] + if report: + nodes.append(node("write", "report", instructions="写一句周报", on_violation="flag")) + nodes.append(node("out", "output", fields=[{"name": "结论", "value": "本周销售额见口径卡"}], + contract={"metrics_from": "card", "narrative": "本周销售额见口径卡"})) + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def wait(run_id: str, statuses=("succeeded", "failed", "interrupted")) -> Run: + for _ in range(600): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None): + await run_manager.wait_idle(run_id) + return row + raise AssertionError(f"运行 {run_id} 停在 {row.status}：{row.error}") + + +async def start(spec: dict, **kwargs) -> Run: + run = await run_manager.start(graph=spec, input_payload={}, **kwargs) + return await wait(run.id) + + +async def events(run_id: str) -> list[RunEvent]: + async with SessionLocal() as session: + return list((await session.execute( + select(RunEvent).where(RunEvent.run_id == run_id).order_by(RunEvent.seq))).scalars()) + + +def logs(evs: list[RunEvent], code: str) -> list[dict]: + return [e.data for e in evs if e.type == "log" and e.data.get("code") == code] + + +async def pending(run_id: str) -> Approval: + async with SessionLocal() as session: + return (await session.execute(select(Approval).where( + Approval.run_id == run_id, Approval.status == "pending"))).scalar_one() + + +async def decide(client, run_id: str, choices: dict, note: str = "") -> Run: + approval = await pending(run_id) + res = await client.post(f"/api/approvals/{approval.id}/decide", + json={"approved": True, "value": {"choices": choices, "note": note}}) + assert res.status_code == 200, res.text + return await wait(run_id, ("succeeded", "failed")) + + +def dual_output(run: Run, evs: list[RunEvent]) -> dict: + finished = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"] + assert finished, "双盲复算节点没有跑完" + return artifact_store.load(finished[-1].data["artifact"]) + + +async def notes_of(source_id: str, table: str = "orders"): + async with SessionLocal() as session: + return await catalog.read_entry(session, source_id, table) + + +# -------------------------------------------------------------------------- +# 一致：不挂起 +# -------------------------------------------------------------------------- + + +async def test_agreeing_check_passes_without_asking(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")}) + run = await start(graph(name, RIGHT)) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + assert out["status"] == "agreed" and "issuance_gaps" not in out + [metric] = out["metrics"] + assert metric["metric_id"] == "gmv" and metric["status"] == "agreed" and metric["delta_pct"] == 0.0 + assert metric["original_rendered"] == metric["blind_rendered"] + assert [d["metrics"] for d in logs(evs, "dual_check_agreed")] == [1] + assert not logs(evs, "dual_check_disagreed") and not [e for e in evs if e.type == "human.requested"] + # 盲算的查询发 tool.end（role: verify），带查询快照，随运行封存 + ends = [e.data for e in evs if e.type == "tool.end" and e.node_id == "dual"] + assert ends and all(d.get("role") == "verify" for d in ends) + assert any(d.get("query_artifact") for d in ends) + # 台账只记一条 dual_check，绝不写 query：报告的写作目录按台账里的 query 编号 + [finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"] + kinds = [entry["kind"] for entry in finished.data.get("evidence") or []] + assert kinds == ["dual_check"] + entry = finished.data["evidence"][0] + assert entry["card"] == "card" and entry["status"] == "agreed" + assert entry["metrics"] == [{"id": "gmv", "status": "agreed"}] + # 用量计入运行：图里只有盲算调用了模型 + assert blind.calls and run.usage["calls"] == len(blind.calls) + assert run.output["_issuance"]["tier"] == "formal" and run.output["_issuance"]["gaps"] == [] + # 盲算 agent 只看得到指标名、口径说明、计算式里是哪一项：看不到原 SQL、取数节点 + [task] = blind.tasks + assert "指标：销售额" in task and DEFINITION in task and "输入1" in task + assert "status <> 'refunded'" not in task and "nodes.fetch" not in task and "SUM(" not in task + + +async def test_grouped_rows_tell_the_blind_agent_which_group(monkeypatch, shop): + """结果分了组：盲算 agent 要知道算的是哪一组（同一行的维度列），但看不到同一行别的数。""" + name, _ = shop + metric = {**GMV, "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + blind = Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")}) + run = await start(graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}})) + assert run.status == "succeeded", run.error + [task] = blind.tasks + assert "region 为「华东」" in task and "（未提供" in task + out = dual_output(run, await events(run.id)) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed" + + +async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop): + name, _ = shop + share = {"id": "gmv_k", "name": "销售额（千元）", "unit": "千元", "decimals": 2, "format": "thousands", + "expression": "round(cell(nodes.fetch, 0, 'gmv') / 1000, 2)"} + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, RIGHT, metrics=[GMV, share])) + assert run.status == "succeeded", run.error + assert blind.runs == 1, "同一格盲算了不止一次" + out = dual_output(run, await events(run.id)) + assert [m["status"] for m in out["metrics"]] == ["agreed", "agreed"] + + +async def test_hitting_a_limit_counts_as_unchecked(monkeypatch, shop): + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, RIGHT), options={"verify": {"dual_check": {"max_queries": 1}}}) + assert run.status == "succeeded", run.error + evs = await events(run.id) + assert [(d["limit"], d["metric"]) for d in logs(evs, "dual_check_limit")] == [("max_queries", "gmv")] + out = dual_output(run, evs) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "unchecked" + assert out["issuance_gaps"] == ["指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"] + assert not [e for e in evs if e.type == "human.requested"] + assert run.output["_issuance"]["tier"] == "degraded" + assert run.output["_issuance"]["gaps"] == ["双盲复算「复核销售额」：指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"] + + +async def test_an_unusable_verify_model_is_tried_once(monkeypatch, shop): + """核查模型没配好：每个指标都记无法复算、原因照实，后面的盲算不再一个个去试。""" + from app.engine import verify_agent + from app.providers.factory import ProviderNotConfigured + + name, _ = shop + tried: list[str] = [] + + async def broken(session, spec): + tried.append(spec.model or "") + raise ProviderNotConfigured("核查用的模型接入已停用", "请到「设置 → 模型接入」中启用") + + monkeypatch.setattr(verify_agent, "get_chat_model", broken) + east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"} + run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}})) + assert run.status == "succeeded", run.error + out = dual_output(run, await events(run.id)) + assert [m["status"] for m in out["metrics"]] == ["unchecked", "unchecked"] + assert all("核查模型不可用" in m["reason"] and "已停用" in m["reason"] for m in out["metrics"]) + assert len(out["issuance_gaps"]) == 2 and len(tried) == 1 + + +# -------------------------------------------------------------------------- +# 不一致：挂起，按约定的 payload 和工件 +# -------------------------------------------------------------------------- + + +async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch, shop): + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + approval = await pending(run.id) + assert approval.mode == "input" and approval.node_id == "dual" + payload = approval.payload + assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary"} + assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input" + [item] = payload["summary"] + assert set(item) == {"metric_id", "name", "unit", "original_rendered", "blind_rendered", "delta_pct", "diff", + "fallback"} + assert (item["metric_id"], item["name"], item["unit"]) == ("gmv", "销售额", "元") + assert item["original_rendered"].startswith("310") and item["blind_rendered"].startswith("330") + assert item["delta_pct"] == round((330 - 310) / 310 * 100, 2) + assert item["fallback"] is False and len(item["diff"]) == 1 + assert "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] and "原流程" in item["diff"][0] + + card = artifact_store.load(payload["card_artifact"]) + assert set(card) == {"metrics", "model", "same_model"} and card["same_model"] is False + [metric] = card["metrics"] + assert set(metric) == {"metric_id", "name", "unit", "decimals", "original", "blind", "diff"} + assert metric["decimals"] == 1 + [orig] = metric["original"]["inputs"] + assert set(orig) == {"path", "value", "sql", "source", "alias"} + assert (orig["sql"], orig["value"], orig["source"], orig["alias"]) == (BY_ORDER_DATE, 310.0, name, "Q1") + [mine] = metric["blind"]["inputs"] + assert set(mine) == {"path", "value", "sql", "source", "alias", "rationale"} + assert (mine["sql"], mine["value"], mine["source"], mine["alias"]) == (RIGHT, 330.0, name, "Q1") + assert mine["path"] == orig["path"] == "cell(nodes.fetch, 0, 'gmv')" and "330" in mine["rationale"] + assert metric["original"]["value"] == 310.0 and metric["blind"]["value"] == 330.0 + assert set(metric["diff"]) == {"equivalent", "items", "fallback", "text_diff"} + assert metric["diff"]["equivalent"] is False and metric["diff"]["fallback"] is False + assert [set(i) for i in metric["diff"]["items"]] == [{"aspect", "text"}] + assert metric["diff"]["items"][0]["aspect"] == "time" and metric["diff"]["text_diff"] + + evs = await events(run.id) + [said] = logs(evs, "dual_check_disagreed") + assert said["metrics"] == [{"id": "gmv", "name": "销售额"}] and said["card_artifact"] == payload["card_artifact"] + assert len([e for e in evs if e.type == "human.requested"]) == 1 + + +async def test_choosing_the_original_leaves_no_gap(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + done = await decide(client, run.id, {"gmv": "a"}, note="周报一直按下单日统计") + assert done.status == "succeeded", done.error + evs = await events(run.id) + out = dual_output(done, evs) + assert out["status"] == "resolved" and out["choices"] == {"gmv": "a"} and "issuance_gaps" not in out + assert out["note"] == "周报一直按下单日统计" and out["metrics"][0]["choice"] == "a" + assert [d["gaps"] for d in logs(evs, "dual_check_resolved")] == [[]] + assert logs(evs, "dual_check_resolved")[0]["choices"] == {"gmv": "a"} + assert done.output["_issuance"]["tier"] == "formal" + assert len([e for e in evs if e.type == "human.resolved"]) == 1 + # 只差时间列一处：采纳原流程，业务日期记原流程那一列 + entry = await notes_of(source_id) + item = entry.notes["business_date"] + assert item["value"] == {"column": "order_date"} and (item["source"], item["status"]) == ("human", "confirmed") + assert run.id in item["note"] and "复核销售额" in item["note"] and "销售额" in item["note"] + [wrote] = logs(evs, "dual_check_catalog") + assert wrote["written"] == [{"source": name, "table": "orders", "item": "business_date"}] and wrote["skipped"] == [] + + +async def test_choosing_the_blind_result_degrades_and_writes_the_catalog(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + done = await decide(client, run.id, {"gmv": "b"}) + assert done.status == "succeeded", done.error + out = dual_output(done, await events(run.id)) + assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致，审核人选择了盲算结果"] + issuance = done.output["_issuance"] + assert issuance["tier"] == "degraded" + assert issuance["gaps"] == ["双盲复算「复核销售额」：指标「销售额」双盲复算不一致，审核人选择了盲算结果"] + assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"} + + +async def test_neither_degrades_and_writes_nothing(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + done = await decide(client, run.id, {"gmv": "neither"}, note="两边都没排除测试单") + assert done.status == "succeeded", done.error + evs = await events(run.id) + out = dual_output(done, evs) + assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致，审核人判断两边都不对"] + assert done.output["_issuance"]["tier"] == "degraded" + assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None + + +async def test_a_missing_choice_counts_as_neither(monkeypatch, shop, client): + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + done = await decide(client, run.id, {"gmv": "maybe"}) + out = dual_output(done, await events(run.id)) + assert out["choices"] == {"gmv": "neither"} + assert out["issuance_gaps"] == ["指标「销售额」双盲复算不一致，审核人判断两边都不对"] + + +async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"})) + done = await decide(client, run.id, {"gmv": "b"}) + assert done.status == "failed" and "审核人没有采用原流程的结果" in (done.error or "") and "销售额" in done.error + evs = await events(run.id) + assert [d["gaps"] for d in logs(evs, "dual_check_resolved")] == [ + ["指标「销售额」双盲复算不一致，审核人选择了盲算结果"]] + # 审核人的选择照样写进了目录：那是对数据的判断，和运行停不停无关 + entry = await notes_of(source_id) + assert entry.notes["business_date"]["value"] == {"column": "paid_date"} and entry.version == 1 + + +async def test_on_reject_fail_lets_the_original_through(monkeypatch, shop, client): + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"})) + done = await decide(client, run.id, {"gmv": "a"}) + assert done.status == "succeeded", done.error + + +async def test_the_same_choice_twice_writes_the_catalog_once(monkeypatch, shop, client): + """同样的不一致、同样的选择再来一次：目录里已经是这个说法，不再写，也不升版本。""" + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + for _ in range(2): + run = await start(graph(name, BY_ORDER_DATE)) + await decide(client, run.id, {"gmv": "b"}) + [wrote] = logs(await events(run.id), "dual_check_catalog") + assert wrote["written"] == [] and "无需重复写入" in wrote["skipped"][0]["reason"] + assert (await notes_of(source_id)).version == 1 + + +# -------------------------------------------------------------------------- +# 写数据目录：只差一个方面才写，确认过的不覆盖 +# -------------------------------------------------------------------------- + + +async def test_a_refund_filter_becomes_the_valid_row_filter(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, WITH_REFUNDS)) + payload = (await pending(run.id)).payload + assert "status = 'refunded'" in payload["summary"][0]["diff"][0] + await decide(client, run.id, {"gmv": "b"}) + item = (await notes_of(source_id)).notes["valid_filter"] + assert item["value"] == "status <> 'refunded'" and item["status"] == "confirmed" and item["source"] == "human" + + +async def test_the_adopted_side_without_the_filter_writes_nothing(monkeypatch, shop, client): + """多出条件的是原流程，审核人却采纳了没有这个条件的盲算：目录里没有对应的写法，照实说。""" + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (WITH_REFUNDS, "gmv")}) + run = await start(graph(name, RIGHT)) + done = await decide(client, run.id, {"gmv": "b"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "被采纳的盲算没有这个条件" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_two_differences_are_not_written(monkeypatch, shop, client): + name, source_id = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BOTH_WRONG)) + assert len((await pending(run.id)).payload["summary"][0]["diff"]) == 2 + done = await decide(client, run.id, {"gmv": "b"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, client): + name, source_id = shop + async with SessionLocal() as session: + await catalog.write_entry(session, source_id, "orders", { + "business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}, if_version=0, + actor="数据组") + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + done = await decide(client, run.id, {"gmv": "b"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] + assert "已有人工确认过的说法「order_date」" in wrote["skipped"][0]["reason"] and "paid_date" in wrote["skipped"][0]["reason"] + entry = await notes_of(source_id) + assert entry.version == 1 and entry.notes["business_date"]["value"] == {"column": "order_date"} + + +# -------------------------------------------------------------------------- +# 设置：全局开关、节点覆盖、发起时的快照 +# -------------------------------------------------------------------------- + + +async def save_verify(values: dict) -> None: + async with SessionLocal() as session: + row = await session.get(Setting, "verify") + if row is None: + session.add(Setting(key="verify", value=values)) + else: + row.value = values + await session.commit() + + +async def launch(spec: dict) -> Run: + async with SessionLocal() as session: + run = await launch_run(session, graph=spec, actor="测试") + return await wait(run.id) + + +async def test_globally_disabled_passes_through(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + await save_verify({"dual_check": {"enabled": False}}) + run = await launch(graph(name, BY_ORDER_DATE)) + assert run.status == "succeeded", run.error + assert run.options["verify"]["dual_check"]["enabled"] is False + evs = await events(run.id) + assert dual_output(run, evs) == {"status": "skipped", "reason": "disabled"} + assert [d["reason"] for d in logs(evs, "dual_check_skipped")] == ["disabled"] + assert blind.calls == [] and run.output["_issuance"]["tier"] == "formal" + + +async def test_the_node_can_override_the_global_switch(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + await save_verify({"dual_check": {"enabled": False}}) + run = await launch(graph(name, RIGHT, dual={"enabled": True})) + assert run.status == "succeeded" and dual_output(run, await events(run.id))["status"] == "agreed" + assert blind.runs == 1 + + await save_verify({"dual_check": {"enabled": True}}) + off = await launch(graph(name, RIGHT, dual={"enabled": False})) + assert dual_output(off, await events(off.id)) == {"status": "skipped", "reason": "disabled"} + assert blind.runs == 1 + + +async def test_no_query_inputs_pass_through(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {}) + constant = {"id": "target", "name": "目标", "unit": "元", "expression": "300"} + run = await start(graph(name, RIGHT, metrics=[constant], dual={"definitions": {}})) + evs = await events(run.id) + assert dual_output(run, evs) == {"status": "skipped", "reason": "no_inputs"} + assert [d["reason"] for d in logs(evs, "dual_check_skipped")] == ["no_inputs"] and blind.calls == [] + assert run.output["_issuance"]["tier"] == "formal" and run.output["_issuance"]["gaps"] == [] + + +async def test_settings_changed_while_waiting_do_not_apply(monkeypatch, shop, client): + """发起时是开着的：停在审批上的时候把双盲复算关了，恢复后照旧按发起时的设置裁决，不会走成直接放行。""" + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await launch(graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted" and run.options["verify"]["dual_check"]["enabled"] is True + await save_verify({"dual_check": {"enabled": False}}) + done = await decide(client, run.id, {"gmv": "b"}) + assert done.status == "succeeded", done.error + evs = await events(run.id) + out = dual_output(done, evs) + assert out["status"] == "resolved" and out["issuance_gaps"] + assert not logs(evs, "dual_check_skipped") + started = [e.data for e in evs if e.type == "run.started"] + assert len(started) == 2 and all(s["options"]["verify"]["dual_check"]["enabled"] is True for s in started) + + +async def test_the_node_reads_the_snapshot_not_the_current_settings(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE), options={"verify": {"dual_check": {"enabled": False}}}) + assert dual_output(run, await events(run.id)) == {"status": "skipped", "reason": "disabled"} + assert blind.calls == [] + + +# -------------------------------------------------------------------------- +# 重放、写作目录 +# -------------------------------------------------------------------------- + + +async def test_resuming_does_not_run_the_blind_agent_again(monkeypatch, shop, client): + """照 test_replay_side_effects：恢复后节点从头重放，盲算 agent 不再执行，事件不重复。""" + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + calls_before = list(blind.calls) + before = await events(run.id) + verify_starts = [e for e in before if e.type == "tool.start" and e.node_id == "dual"] + llm_ends = [e for e in before if e.type == "llm.end" and e.node_id == "dual"] + assert verify_starts and llm_ends + + done = await decide(client, run.id, {"gmv": "a"}) + assert done.status == "succeeded", done.error + assert blind.calls == calls_before, f"恢复之后盲算又调用了模型：{blind.calls[len(calls_before):]}" + evs = await events(run.id) + assert [e.seq for e in evs if e.type == "tool.start" and e.node_id == "dual"] == [e.seq for e in verify_starts] + assert [e.seq for e in evs if e.type == "llm.end" and e.node_id == "dual"] == [e.seq for e in llm_ends] + for code in ("dual_check_disagreed", "dual_check_resolved", "dual_check_catalog"): + assert len(logs(evs, code)) == 1, code + assert len([e for e in evs if e.type == "human.requested"]) == 1 + assert len([e for e in evs if e.type == "human.resolved"]) == 1 + assert len([e for e in evs if e.type == "node.finished" and e.node_id == "dual"]) == 1 + + +async def test_the_report_catalog_leaves_out_the_blind_queries(monkeypatch, shop): + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")}, writer="本周销售额 [[m:gmv]]。") + run = await start(graph(name, RIGHT, report=True)) + assert run.status == "succeeded", run.error + [prompt] = blind.writer_prompts + assert "blind_total" not in prompt and "Q2" not in prompt, "盲算的查询进了报告的写作目录" + assert "Q1" in prompt and "m:gmv" in prompt + evs = await events(run.id) + blind_queries = [e.data["query_artifact"] for e in evs + if e.type == "tool.end" and e.node_id == "dual" and e.data.get("query_artifact")] + ledger = [entry for e in evs if e.type == "node.finished" for entry in e.data.get("evidence") or []] + assert blind_queries and not any(entry.get("artifact") in blind_queries for entry in ledger) diff --git a/backend/tests/test_issuance_upstream_gaps.py b/backend/tests/test_issuance_upstream_gaps.py index 3427336d..cc6929c8 100644 --- a/backend/tests/test_issuance_upstream_gaps.py +++ b/backend/tests/test_issuance_upstream_gaps.py @@ -86,8 +86,9 @@ async def test_upstream_gap_degrades_issuance_and_names_the_node(): async def test_stub_gate_nodes_pass_through_without_touching_issuance(): - """双盲复算、反方审稿目前是占位：放在图里照样能运行，出具不受影响""" - for kind in ("dual_check", "challenge"): + """反方审稿目前是占位：放在图里照样能运行，出具不受影响。双盲复算已经实现，没有可复算的指标时直接放行、 + 不碰出具，见 test_dual_check_node.py""" + for kind in ("challenge",): row = await finish(graph(node("gate", kind))) issuance = row.output["_issuance"] assert issuance["tier"] == "formal", (kind, issuance)