yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 05:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
5d5a29cfa69a252a451c1d1e55c56ba2caa07ea2
AgentLab/backend/tests/test_dual_check_node.py
T

1578 lines
96 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""双盲复算节点（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，也没有原流程结果的列名
- 一个都没核成是 unchecked、只核成一部分是 partial，日志 dual_check_unchecked 说清哪几个、为什么
- 不一致：挂起，审批 payload 和对照工件按约定，payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往
数据目录里写什么的预告；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject 为 fail 时失败
- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比，HAVING 一并去掉）；分组的列设了
遮罩时（改了名再分组的、在子查询或 WITH 里改了名的也算）不收窄、照整条比，遮罩列的取值不进审批卡、对照工件、产出、
事件和盲算的任务
- 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的
筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写；人工驳回过的条件不会借着和别的
指标并成一项写回去，真写的和审批卡上的预告一致；时间列在表结构里声明成文本的不记成业务日期，预告照实说（上传表格
建的数据源说清日期是按文字保存的）
- 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询
- 触顶、分组结果的维度（只给取值、取自遮罩列的不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列
上的盲算结果不采用；按位置取自另一份快照遮罩列的输入（cell() 取 A、vars.res.rows[0][0] 取 B）按无法复算记、原因点名
是哪个输入，不盲算；认不出取自哪一列的按位置取的格不写进计算式；按位置取的格不改变容器输入的判法
"""
from __future__ import annotations
import asyncio
import json
import sqlite3
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, DataSource, Run, RunEvent, Setting
from app.engine import spend
from app.engine.launch import launch_run
from app.engine.nodes.dual_check import GAP_UNCHECKED
from app.engine.runner import run_manager
from app.main import app
from tests.fixtures.dual_check_shop import (
BOTH_WRONG,
BUYERS,
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 masked_shop(tmp_path):
"""遮了 region、secret_gmv 两列的同一个库。"""
name, source_id = await add_source(build(tmp_path / "masked.db"), options={"mask_columns": "region, secret_gmv"})
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, inputs: dict | None = None, **kwargs) -> Run:
run = await run_manager.start(graph=spec, input_payload=inputs or {}, **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
assert out["counts"] == {"agreed": 1, "disagreed": 0, "unchecked": 0}
[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"}] and entry["counts"] == out["counts"]
assert not logs(evs, "dual_check_unchecked")
# 用量计入运行：图里只有盲算调用了模型
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
assert "gmv" 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")})
spec = graph(name, BY_REGION, metrics=[metric], dual={"definitions": {}})
spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "week_start"}, {"name": "week_end"}]
run = await start(spec, {"week_start": "2026-09-01", "week_end": "2026-09-07"})
assert run.status == "succeeded", run.error
[task] = blind.tasks
assert "「华东」" in task and "（未提供" in task and "300" not in task
assert "region" not in task and "gmv" not in task
assert "week_start = 2026-09-01" in task and "week_end = 2026-09-07" in task
out = dual_output(run, await events(run.id))
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"
async def test_the_blind_agent_is_not_told_which_date_column_the_original_grouped_by(monkeypatch, shop):
"""原流程按下单日分组、口径卡取其中一天：盲算只知道是哪一天，按哪一列归属要它自己按口径说明判断。告诉它
「order_date 为 2026-09-02」，它照着算出同一个数，下单日和支付日的陷阱就漏过去了。"""
name, _ = shop
by_day = (f"SELECT order_date, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} "
"GROUP BY order_date ORDER BY order_date")
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, by_day))
assert run.status in ("succeeded", "interrupted"), run.error
[task] = blind.tasks
assert "2026-09-02 这一天" in task
assert "order_date" not in task and "gmv" not in task
async def test_column_names_carrying_a_choice_are_not_given_as_hints(monkeypatch, shop):
"""几项输入都来自查询、又没写口径说明时，列名才交给盲算区分计算式里的几项；列名里带着原流程的取舍的（按下单日、
已支付）不给：「gmv_by_order_date」等于告诉它按下单日归属。看不出取舍的（total）照给。"""
name, _ = shop
wide = (f"SELECT SUM(amount) AS gmv_by_order_date, COUNT(*) AS paid_orders, SUM(amount) AS total FROM orders "
f"WHERE status <> 'refunded' AND order_date {WEEK}")
aov = {"id": "aov", "name": "客单价", "unit": "元", "decimals": 2, "format": "plain",
"expression": "(cell(nodes.fetch, 0, 'gmv_by_order_date') + cell(nodes.fetch, 0, 'total')) "
"/ cell(nodes.fetch, 0, 'paid_orders')"}
blind = Blind(monkeypatch, name, {}) # 盲算都交不出数：这里只看任务原文
await start(graph(name, wide, metrics=[aov], dual={"definitions": {}}))
assert len(blind.tasks) == 3
said = "\n".join(blind.tasks)
assert "gmv_by_order_date" not in said and "paid_orders" not in said
assert "「total」" in said
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):
"""查询次数用完了还没查到要交的那一格：按无法复算计入缺口；一个指标都没核成，产出是 unchecked，不是 agreed。"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": ([BLIND_TOTAL, 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"] == "unchecked" and out["metrics"][0]["status"] == "unchecked"
assert out["counts"] == {"agreed": 0, "disagreed": 0, "unchecked": 1}
assert out["issuance_gaps"] == ["指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"]
[said] = logs(evs, "dual_check_unchecked")
assert said["metrics"] == [{"id": "gmv", "name": "销售额", "reason": "盲算已达每项最多查询次数（1 次）"}]
assert not logs(evs, "dual_check_agreed")
assert not [e for e in evs if e.type == "human.requested"]
[finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"]
assert finished.data["evidence"][0]["status"] == "unchecked"
assert run.output["_issuance"]["tier"] == "degraded"
assert run.output["_issuance"]["gaps"] == ["双盲复算「复核销售额」：指标「销售额」无法复算：盲算已达每项最多查询次数（1 次）"]
async def test_an_answer_on_the_last_allowed_query_still_counts(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)
out = dual_output(run, evs)
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed" and "issuance_gaps" not in out
assert not logs(evs, "dual_check_limit") and not logs(evs, "dual_check_unchecked")
assert run.output["_issuance"]["tier"] == "formal"
async def test_every_metric_sharing_a_cell_reports_the_limit(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, {"销售额": ([BLIND_TOTAL, RIGHT], "gmv")})
run = await start(graph(name, RIGHT, metrics=[GMV, share]), options={"verify": {"dual_check": {"max_queries": 1}}})
assert run.status == "succeeded", run.error
assert blind.runs == 1
evs = await events(run.id)
assert [(d["limit"], d["metric"]) for d in logs(evs, "dual_check_limit")] == [
("max_queries", "gmv"), ("max_queries", "gmv_k")]
[said] = logs(evs, "dual_check_unchecked")
assert [m["id"] for m in said["metrics"]] == ["gmv", "gmv_k"]
async def test_some_unchecked_metrics_make_a_partial_result(monkeypatch, shop):
name, _ = shop
east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"}
Blind(monkeypatch, name, {"华东销售额": (BY_REGION, "gmv")})
run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}}))
assert run.status == "succeeded", run.error
evs = await events(run.id)
out = dual_output(run, evs)
assert out["status"] == "partial" and [m["status"] for m in out["metrics"]] == ["agreed", "unchecked"]
assert out["counts"] == {"agreed": 1, "disagreed": 0, "unchecked": 1}
assert [d["metrics"] for d in logs(evs, "dual_check_agreed")] == [1]
[said] = logs(evs, "dual_check_unchecked")
assert [(m["id"], m["name"]) for m in said["metrics"]] == [("north", "华北销售额")] and said["metrics"][0]["reason"]
assert out["issuance_gaps"] == [f"指标「华北销售额」无法复算：{said['metrics'][0]['reason']}"]
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
evs = await events(run.id)
out = dual_output(run, evs)
assert out["status"] == "unchecked" and [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
[said] = logs(evs, "dual_check_unchecked")
assert [m["id"] for m in said["metrics"]] == ["east", "north"] and "核查模型不可用" in said["metrics"][0]["reason"]
# --------------------------------------------------------------------------
# 不一致：挂起，按约定的 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
# 约定的六个键，另加 on_reject（实际生效的处置）：审核人在卡上要看得出没采用原流程会不会让运行失败；reject_route
# （驳回之后去哪，engine/replay.reject_route）：没接驳回出口，驳回后运行就此结束
assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary", "on_reject", "reject_route"}
assert payload["reject_route"] == {"ends": "run"}
assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input"
assert payload["on_reject"] == "degrade"
[item] = payload["summary"]
assert set(item) == {"metric_id", "name", "unit", "original_rendered", "blind_rendered", "delta_pct", "diff",
"fallback", "catalog"}
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]
# 选哪一边会往数据目录里写什么，按发起审批时的目录预告：只差时间列一处，选哪边就记哪一列
assert item["catalog"] == {
"a": {"write": True, "source": name, "table": "orders", "item": "business_date",
"value": {"column": "order_date"}},
"b": {"write": True, "source": name, "table": "orders", "item": "business_date",
"value": {"column": "paid_date"}},
}
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", "rationale_doc"}
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"]
# 理由组装成的文档：正文同 rationale，数字挂着盲算自己那次查询的格子
doc = mine["rationale_doc"]
assert doc["markdown"] == mine["rationale"] and doc["violations"] == []
[cited] = [seg for block in doc["blocks"] for unit in block["units"] for seg in unit["segments"] if seg.get("cite")]
assert cited["text"] == "330" and cited["cite"]["alias"] == "Q1"
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_a_bare_number_in_the_blind_rationale_is_marked(monkeypatch, shop):
"""盲算的理由里写了没有出处的数（「比上周多 20」）：对照工件里的理由文档把它标成裸数字，审批卡据此醒目标出"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}, rationale="按口径说明统计，结果是 {ref}，比上周多 20。")
run = await start(graph(name, BY_ORDER_DATE))
card = artifact_store.load((await pending(run.id)).payload["card_artifact"])
[mine] = card["metrics"][0]["blind"]["inputs"]
doc = mine["rationale_doc"]
bare = [seg["text"] for block in doc["blocks"] for unit in block["units"] for seg in unit["segments"]
if seg.get("issue") == "uncited_number"]
assert bare == ["20"] and mine["rationale"] == doc["markdown"]
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",
"value": {"column": "order_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_a_dual_check_named_after_its_type_is_named_once_in_the_gap(monkeypatch, shop, client):
"""拖进画布没改名的双盲复算：缺口读成「双盲复算：指标「销售额」两路结果不一致……」，不写两遍「双盲复算」"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
spec = graph(name, BY_ORDER_DATE)
next(n for n in spec["nodes"] if n["id"] == "dual")["data"]["label"] = "双盲复算"
run = await start(spec)
done = await decide(client, run.id, {"gmv": "neither"})
assert done.status == "succeeded", done.error
assert done.output["_issuance"]["gaps"] == ["双盲复算：指标「销售额」两路结果不一致，审核人判断两边都不对"]
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_is_recorded_as_no_choice(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": "maybe"})
assert done.status == "succeeded", done.error
evs = await events(run.id)
out = dual_output(done, evs)
assert out["choices"] == {} and "choice" not in out["metrics"][0]
assert out["issuance_gaps"] == ["指标「销售额」两路结果不一致，审核人没有作出选择"]
[resolved] = logs(evs, "dual_check_resolved")
assert resolved["choices"] == {} and resolved["gaps"] == out["issuance_gaps"]
assert "没有作出选择" in resolved["message"] and "两边都不对" not in resolved["message"]
assert done.output["_issuance"]["tier"] == "degraded"
assert not logs(evs, "dual_check_catalog") and await notes_of(source_id) is None
async def test_rejecting_the_approval_ends_the_run_as_rejected(monkeypatch, shop, client):
"""明确驳回裁决：人的决定，不是故障（05-已知遗留 (a)）。运行收成 cancelled + end_reason=rejected，原话照旧。"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
approval = await pending(run.id)
res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"})
assert res.status_code == 200, res.text
done = await wait(run.id, ("succeeded", "failed", "cancelled"))
assert (done.status, done.end_reason) == ("cancelled", "rejected"), done.error
assert "人工驳回" in (done.error or "") and "口径要重新对" in done.error
[ended] = [e for e in await events(run.id) if e.type == "run.cancelled"]
assert (ended.data["kind"], ended.data["note"]) == ("dual_check", "口径要重新对")
async def test_rejecting_the_approval_with_a_reject_exit_goes_on_along_it(monkeypatch, shop, client):
"""裁决接了「驳回」出口（docs/design/orchestration-round/00-总方案.md §3）：驳回时这一步正常结束、沿驳回边往下走，产出带
哪种关卡、结论、备注；运行照它的路径结束。"""
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
spec = graph(name, BY_ORDER_DATE)
spec["nodes"].append(node("redo", "output", "重新核对", fields=[{"name": "结论", "value": "{{ nodes.dual.note }}"}]))
spec["edges"].append({"source": "dual", "target": "redo", "sourceHandle": "rejected"})
run = await start(spec)
approval = await pending(run.id)
res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"})
assert res.status_code == 200, res.text
done = await wait(run.id, ("succeeded", "failed", "cancelled"))
assert (done.status, done.end_reason) == ("succeeded", None), done.error
assert done.output == {"结论": "口径要重新对"}
evs = await events(run.id)
[finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"]
assert (finished.data["outcome"], finished.data["kind"]) == ("rejected", "dual_check")
assert not [e for e in evs if e.type == "node.finished" and e.node_id == "out"], "普通出口没有走"
async def test_a_rejected_arbitration_through_the_exit_still_degrades_the_issuance(monkeypatch, shop, client):
"""驳回出口直接接到出具的成果节点：审核人驳回了裁决，那组数不能照样完整出具。走出口时这一步的产出照样带出具缺口
（「审核人驳回了裁决」），台账照记一条结论为驳回的 dual_check，证据面板的指标步骤据此认出复核过这份指标集；
不比选「两边都不对」宽。"""
from app.api.evidence import _dual_check_of, _sealed
name, _ = shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
spec = graph(name, BY_ORDER_DATE)
spec["edges"].append({"source": "dual", "target": "out", "sourceHandle": "rejected"})
run = await start(spec)
approval = await pending(run.id)
res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"})
assert res.status_code == 200, res.text
done = await wait(run.id, ("succeeded", "failed", "cancelled"))
assert done.status == "succeeded", done.error
issuance = done.output["_issuance"]
assert issuance["tier"] == "degraded", issuance
assert "双盲复算「复核销售额」：指标「销售额」两路结果不一致，审核人驳回了裁决" in issuance["gaps"], issuance["gaps"]
evs = await events(run.id)
[finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"]
assert finished.data["outcome"] == "rejected"
[entry] = [e for e in finished.data.get("evidence") or [] if e.get("kind") == "dual_check"]
assert (entry["status"], entry["card"]) == ("rejected", "card") and entry["metrics"][0]["status"] == "disagreed"
assert finished.data["preview"]["status"] == "rejected" and finished.data["preview"]["rejected"] is True
sealed = await _sealed(run.id)
[(_, artifact, _)] = [c for c in __import__("app.api.evidence", fromlist=["_sealed_cards"])._sealed_cards(sealed)]
checked = _dual_check_of(sealed, "card", artifact, "gmv")
assert checked is not None and checked["status"] == "disagreed" and "choice" not in checked, checked
# 对照：同一处选「两边都不对」也是降档
other = await start(graph(name, BY_ORDER_DATE))
assert (await decide(client, other.id, {"gmv": "neither"})).output["_issuance"]["tier"] == "degraded"
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"}))
# 审批卡上看得出：这一步没采用原流程，运行就会停下
assert (await pending(run.id)).payload["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
# --------------------------------------------------------------------------
# 写数据目录：只差一个方面才写，确认过的不覆盖
# --------------------------------------------------------------------------
@pytest.fixture
async def text_dated_shop(tmp_path):
"""同一个库，只是下单日、支付日两列声明成文本（SQLite 常这样存日期）。"""
name, source_id = await add_source(build(tmp_path / "text_dated.db", date_type="TEXT"))
yield name, source_id
await drop_source(source_id)
async def test_time_columns_declared_as_text_are_not_written_as_the_business_date(monkeypatch, text_dated_shop, client):
"""两边按不同的列筛时间，可这两列在表结构里声明成文本：认不准存的是不是日期，选哪边都不记成业务日期。审批卡上的
预告照实说（认不出可记的事实，不带写到哪），裁决后时间线上跳过的原因和预告一字不差。声明成 DATE 的同一个库照常记
（test_choosing_the_blind_result_degrades_and_writes_the_catalog）。"""
name, source_id = text_dated_shop
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
assert run.status == "interrupted", run.error
[item] = (await pending(run.id)).payload["summary"]
assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0]
said = item["catalog"]
# 两列都是文本：各说被采纳的那一边的列
for side, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")):
assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「orders.{column}」"
"在表结构里的类型是 TEXT，无法确认存的是日期，不写入数据目录"}
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "succeeded", done.error
[wrote] = logs(await events(run.id), "dual_check_catalog")
assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [said["b"]["reason"]]
assert await notes_of(source_id) is None
async def test_an_uploaded_source_says_its_dates_are_stored_as_text(monkeypatch, text_dated_shop, client):
"""上传表格建的数据源（origin=upload），导入时日期一律存成文字：审批卡的预告和裁决后时间线上跳过的原因，都说上传的
表格按文字保存日期，不说「类型是 TEXT」（那不是用户声明的，导入确认时这一列标的是「日期」）。上传源的查询要走版本
快照，这里只把节点认数据源来历的那一步（_source_info）换成上传源，查询照手工源走；_source_info 自己认 origin 的那一步
另见下一个用例。"""
from app.engine.nodes import dual_check
name, source_id = text_dated_shop
real = dual_check._source_info
async def uploaded(sources, source):
return {**await real(sources, source), "uploaded": True}
monkeypatch.setattr(dual_check, "_source_info", uploaded)
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
assert run.status == "interrupted", run.error
[item] = (await pending(run.id)).payload["summary"]
said = item["catalog"]
for side, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")):
assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「orders.{column}」"
"来自上传的表格，表格里的日期一律按文字保存，无法确认存的是"
"日期，不写入数据目录"}
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "succeeded", done.error
[wrote] = logs(await events(run.id), "dual_check_catalog")
assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [said["b"]["reason"]]
assert await notes_of(source_id) is None
async def test_the_source_info_knows_an_uploaded_source(shop):
from app.engine.nodes.dual_check import _source_info, _Sources
name, source_id = shop
assert (await _source_info(_Sources(), name))["uploaded"] is False
async with SessionLocal() as session:
(await session.get(DataSource, source_id)).origin = "upload"
await session.commit()
assert (await _source_info(_Sources(), name))["uploaded"] is True
assert (await _source_info(_Sources(), "dc_no_such_source"))["uploaded"] is False
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))
# 预告按目录里已有的说法算：选原流程是同一个说法、不用再写，选盲算不会覆盖已确认的
preview = (await pending(run.id)).payload["summary"][0]["catalog"]
assert preview["a"]["write"] is False and "无需重复写入" in preview["a"]["reason"]
assert preview["b"]["write"] is False and "已有人工确认过的说法「order_date」" in preview["b"]["reason"]
# 认出了事实、只是被目录挡下：照样说写到哪张表的哪一项、本来要写什么
assert {k: preview["b"][k] for k in ("source", "table", "item", "value")} == \
{"source": name, "table": "orders", "item": "business_date", "value": {"column": "paid_date"}}
done = await decide(client, run.id, {"gmv": "b"})
[wrote] = logs(await events(done.id), "dual_check_catalog")
assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [preview["b"]["reason"]]
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 test_a_rejected_filter_is_not_written_back_along_with_another_metric(monkeypatch, shop, client, tmp_path):
"""目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」采纳了排除退款的盲算、「订单金额」采纳了排除
测试单的盲算：审批卡上预告前者不写、后者写「is_test = 0」。真写的和预告一字不差——驳回过的条件不会借着两个指标
并成一项又写成人工确认。"""
name, source_id = shop
db = sqlite3.connect(tmp_path / "shop.db")
db.execute("INSERT INTO orders VALUES (6, '华东', 105, 40.0, 'paid', '2026-09-03', '2026-09-03', 'online', 1)")
db.commit()
db.close()
async with SessionLocal() as session:
await catalog.write_entry(session, source_id, "orders", {
"valid_filter": catalog.make_item("status <> 'refunded'", "human", "rejected")}, if_version=0, actor="数据组")
original = f"SELECT SUM(amount) AS gmv, SUM(amount) AS total FROM orders WHERE paid_date {WEEK}"
no_test = f"SELECT SUM(amount) AS total FROM orders WHERE is_test = 0 AND paid_date {WEEK}"
total = {**GMV, "id": "total", "name": "订单金额", "expression": "cell(nodes.fetch, 0, 'total')"}
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单金额": (no_test, "total")})
run = await start(graph(name, original, metrics=[GMV, total], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
said = {s["metric_id"]: s["catalog"]["b"] for s in payload["summary"]}
assert said["gmv"]["write"] is False and "曾被人工驳回" in said["gmv"]["reason"]
assert said["total"]["write"] is True and said["total"]["value"] == "is_test = 0"
# 两个指标都落到有效记录条件上：合起来的结果另有预告，和真写一致
choices = {"gmv": "b", "total": "b"}
expected = card_said(payload, choices)
assert expected["gmv"]["write"] is False and expected["gmv"]["reason"] == said["gmv"]["reason"]
assert expected["total"] == {"write": True, "value": "is_test = 0", "item": "valid_filter"}
done = await decide(client, run.id, choices)
assert done.status == "succeeded", done.error
[wrote] = logs(await events(run.id), "dual_check_catalog")
assert wrote["written"] == [{"source": name, "table": "orders", "item": "valid_filter", "value": "is_test = 0"}]
assert [s["reason"] for s in wrote["skipped"]] == [said["gmv"]["reason"]]
await assert_written_as_said(expected, wrote, source_id)
item = (await notes_of(source_id)).notes["valid_filter"]
assert item["value"] == said["total"]["value"] and (item["source"], item["status"]) == ("human", "confirmed")
assert "「订单金额」" in item["note"] and "「销售额」" not in item["note"]
def card_said(payload: dict, choices: dict) -> dict:
"""审批卡按这组选择会怎么预告（DualCheckApproval 的认法）：选的那一边认出了写到哪，就看按当前的选择还有哪些指标
落到同一项上——只有它自己时照摘要里它的预告，几个时照对照工件 catalog_joint 里这几个成员的那一条。"""
summary = {s["metric_id"]: s["catalog"] for s in payload["summary"]}
joint = artifact_store.load(payload["card_artifact"]).get("catalog_joint") or []
def where(metric_id: str, choice: str):
side = summary[metric_id][choice]
return (side["source"], side["table"], side["item"]) if "table" in side else None
out = {}
for metric_id, choice in choices.items():
if choice not in ("a", "b"):
continue
target = where(metric_id, choice)
peers = sorted((m, c) for m, c in choices.items() if c in ("a", "b") and target and where(m, c) == target)
if len(peers) < 2:
out[metric_id] = summary[metric_id][choice]
continue
[entry] = [e for e in joint if sorted((m["metric_id"], m["choice"]) for m in e["members"]) == peers]
out[metric_id] = {**entry["outcomes"][metric_id], "item": entry["item"]}
return out
async def assert_written_as_said(said: dict, wrote: dict, source_id: str) -> None:
"""预告和真写逐字一致：说会记为什么的，目录里就是那个值、日志的 written 里有它；说不记的，原因就在日志的跳过原因里"""
entry = await notes_of(source_id)
for metric_id, side in said.items():
if side["write"]:
assert any(w["item"] == side["item"] and w["value"] == side["value"] for w in wrote["written"]), \
(metric_id, side, wrote)
assert entry.notes[side["item"]]["value"] == side["value"]
else:
assert side["reason"] in [s["reason"] for s in wrote["skipped"]], (metric_id, side, wrote)
#: 两个指标取自同一条查询的两列，都是按下单日筛的：盲算按支付日算，两个指标都只差时间列，落到 orders 的业务日期上
TWO_BY_ORDER_DATE = (f"SELECT SUM(amount) AS gmv, SUM(amount) AS total FROM orders "
f"WHERE status <> 'refunded' AND order_date {WEEK}")
TOTAL = {**GMV, "id": "total", "name": "订单金额", "expression": "cell(nodes.fetch, 0, 'total')"}
@pytest.mark.parametrize("choices", [
{"gmv": "a", "total": "b"}, # 说法不同：两个都不写
{"gmv": "b", "total": "b"}, # 说法一样：写一次
{"gmv": "a", "total": "a"},
{"gmv": "b", "total": "neither"}, # 只有一个落到这一项上：照它自己的预告
])
async def test_the_card_preview_is_what_the_decision_writes(monkeypatch, shop, client, choices):
"""图级：两个指标都只差时间列、落到数据目录的同一项上。审批卡按当前的选择预告的（摘要里各自的预告、对照工件里
合起来的预告），和裁决后真写进目录的逐字一致：写的值一样，不写的原因一句不差"""
name, source_id = shop
right_total = f"SELECT SUM(amount) AS total FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}"
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单金额": (right_total, "total")})
run = await start(graph(name, TWO_BY_ORDER_DATE, metrics=[GMV, TOTAL], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
card = artifact_store.load(payload["card_artifact"])
assert len(card["catalog_joint"]) == 4, "两个指标两边都落在业务日期上：四种组合各预告一次"
said = card_said(payload, choices)
done = await decide(client, run.id, choices)
assert done.status == "succeeded", done.error
[wrote] = logs(await events(run.id), "dual_check_catalog")
await assert_written_as_said(said, wrote, source_id)
if choices == {"gmv": "a", "total": "b"}:
assert not said["gmv"]["write"] and said["gmv"] == {**said["total"], "item": "business_date"}
assert wrote["written"] == [] and "给出了不同的说法" in said["gmv"]["reason"]
async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatch, shop, client):
"""「华东销售额」原流程按指标名只看华东，盲算漏了地区：审核人选原流程，但 region = '华东' 是这个指标自己的筛选
范围，写成表的有效记录条件，以后所有查订单表的工作流（包括下一次盲算）都会只看华东。"""
name, source_id = shop
east = {**GMV, "id": "east", "name": "华东销售额"}
Blind(monkeypatch, name, {"华东销售额": (RIGHT, "gmv")})
original = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ")
run = await start(graph(name, original, metrics=[east], dual={"definitions": {"east": "华东地区已支付订单的金额"}}))
[item] = (await pending(run.id)).payload["summary"]
assert len(item["diff"]) == 1 and "region = '华东'" in item["diff"][0]
# 预告和真写的是同一套判断：两边选哪个都不写，原因照实
assert item["catalog"]["a"]["write"] is False and "可能是指标自己的筛选范围" in item["catalog"]["a"]["reason"]
assert item["catalog"]["b"]["write"] is False and "被采纳的盲算没有这个条件" in item["catalog"]["b"]["reason"]
done = await decide(client, run.id, {"east": "a"})
assert done.status == "succeeded", done.error
[wrote] = logs(await events(run.id), "dual_check_catalog")
assert wrote["written"] == [] and "可能是指标自己的筛选范围" in wrote["skipped"][0]["reason"]
assert await notes_of(source_id) is None
async def test_equivalent_sql_with_different_values_points_at_the_data(monkeypatch, shop, client, tmp_path):
"""原流程查完之后库里多了一单：两条 SQL 规范化后一致、值却不同，差异说明写「差在数据」，不写数据目录。"""
from app.engine.nodes.dual_check import SAME_SQL
name, source_id = shop
def late_order() -> None:
db = sqlite3.connect(tmp_path / "shop.db")
db.execute("INSERT INTO orders VALUES (6, '华东', 104, 70.0, 'paid', '2026-09-04', '2026-09-04', 'online', 0)")
db.commit()
db.close()
Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")}, on_query=late_order)
run = await start(graph(name, RIGHT))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
[item] = payload["summary"]
assert item["diff"] == [SAME_SQL] and item["fallback"] is False
assert item["original_rendered"].startswith("330") and item["blind_rendered"].startswith("400")
[metric] = artifact_store.load(payload["card_artifact"])["metrics"]
assert metric["diff"]["equivalent"] is True and metric["diff"]["items"] == [{"aspect": "data", "text": SAME_SQL}]
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_multi_column_query_is_compared_on_its_own_column(monkeypatch, shop, client):
"""原流程一条查询算了好几列，口径卡取其中一列：差异只比这一列是怎么算的，不混进「选出的列」这类差别。"""
name, source_id = shop
wide = f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE status <> 'refunded' AND order_date {WEEK}"
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, wide))
payload = (await pending(run.id)).payload
[item] = payload["summary"]
assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0]
[metric] = artifact_store.load(payload["card_artifact"])["metrics"]
assert [i["aspect"] for i in metric["diff"]["items"]] == ["time"]
# 对照里的原 SQL 还是原流程整条查询：给人看的是原文
assert metric["original"]["inputs"][0]["sql"] == wide
await decide(client, run.id, {"gmv": "b"})
assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
async def test_a_grouped_query_is_compared_on_the_row_it_was_taken_from(monkeypatch, shop, client):
"""原流程按地区分组、取华东那一行：收窄成「华东这一组」再比，差异里不出现分组、地区条件这类和这一格无关的差别。"""
name, source_id = shop
by_region = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} "
"GROUP BY region ORDER BY region")
east_paid = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ")
east = {**GMV, "id": "east", "name": "华东销售额"}
Blind(monkeypatch, name, {"华东销售额": (east_paid, "gmv")})
run = await start(graph(name, by_region, metrics=[east], dual={"definitions": {}}))
[item] = (await pending(run.id)).payload["summary"]
assert item["original_rendered"].startswith("200") and item["blind_rendered"].startswith("300")
assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0]
await decide(client, run.id, {"east": "b"})
assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
async def test_a_having_clause_is_dropped_when_narrowing_to_one_row(monkeypatch, shop, client):
"""原流程分组后另有 HAVING：取的这一行既然在结果里，它所在的组就满足 HAVING，对这一格是噪声。收窄时连 HAVING
一起去掉，差异只剩真正的那一处（时间列），写得进目录。"""
name, source_id = shop
by_region = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} "
"GROUP BY region HAVING SUM(amount) > 10 ORDER BY region")
east_paid = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ")
east = {**GMV, "id": "east", "name": "华东销售额"}
Blind(monkeypatch, name, {"华东销售额": (east_paid, "gmv")})
run = await start(graph(name, by_region, metrics=[east], dual={"definitions": {}}))
[item] = (await pending(run.id)).payload["summary"]
assert item["original_rendered"].startswith("200") and item["blind_rendered"].startswith("300")
assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0]
await decide(client, run.id, {"east": "b"})
assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"}
async def test_a_condition_on_a_grouped_column_is_not_written(monkeypatch, shop, client, tmp_path):
"""原流程按 is_test 分组、口径卡取 is_test = 0 那一行：收窄后原流程多出「is_test = 0」，看着像排除测试单的有效
记录条件，可它用的是原流程分组的列——是这个指标按组取的范围，不写目录。取值 0 太短、不在任何文字里，只有分组的列
拦得下它。"""
name, source_id = shop
db = sqlite3.connect(tmp_path / "shop.db")
db.execute("INSERT INTO orders VALUES (6, '华东', 105, 40.0, 'paid', '2026-09-03', '2026-09-03', 'online', 1)")
db.commit()
db.close()
by_flag = (f"SELECT is_test, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} "
"GROUP BY is_test ORDER BY is_test")
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
run = await start(graph(name, by_flag, dual={"definitions": {}}))
[item] = (await pending(run.id)).payload["summary"]
assert item["original_rendered"].startswith("330") and item["blind_rendered"].startswith("370")
assert len(item["diff"]) == 1 and "is_test = 0" in item["diff"][0]
done = await decide(client, run.id, {"gmv": "a"})
[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_condition_matching_a_run_input_is_not_written(monkeypatch, shop, client):
"""排除退款的条件看着像有效记录条件，可取值正是本次运行的输入（这一期要排除的状态）：是这个指标这一期的筛选范围，
不是整张表的事实，不写目录。没有这个输入时同样的选择会写（test_a_refund_filter_becomes_the_valid_row_filter）。"""
name, source_id = shop
spec = graph(name, RIGHT, dual={"definitions": {}})
spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "exclude_status"}]
Blind(monkeypatch, name, {"销售额": (WITH_REFUNDS, "gmv")})
run = await start(spec, {"exclude_status": "refunded"})
[item] = (await pending(run.id)).payload["summary"]
assert len(item["diff"]) == 1 and "refunded" in item["diff"][0]
done = await decide(client, run.id, {"gmv": "a"})
[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_differing_inputs_do_not_write_the_catalog(monkeypatch, shop, client):
"""一个指标的两个输入都对不上：审核人选的是整个指标哪边对，拆不出是哪一处，不写目录。"""
name, source_id = shop
share = (f"SELECT SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END) AS east, SUM(amount) AS total FROM orders "
f"WHERE status <> 'refunded' AND order_date {WEEK}")
east_paid = (f"SELECT SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END) AS east FROM orders "
f"WHERE status <> 'refunded' AND paid_date {WEEK}")
ratio = {"id": "east_share", "name": "华东占比", "unit": "", "decimals": 3, "format": "plain",
"expression": "cell(nodes.fetch, 0, 'east') / cell(nodes.fetch, 0, 'total')"}
blind = Blind(monkeypatch, name, {"华东占比|输入1": (east_paid, "east"), "华东占比|输入2": (RIGHT, "gmv")})
run = await start(graph(name, share, metrics=[ratio], dual={"definitions": {}}))
assert blind.runs == 2
[item] = (await pending(run.id)).payload["summary"]
assert len(item["diff"]) == 2 and all("输入「" in d for d in item["diff"])
done = await decide(client, run.id, {"east_share": "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_only_the_configured_metrics_are_checked(monkeypatch, shop):
name, _ = shop
wide = f"SELECT SUM(amount) AS gmv, COUNT(*) AS n FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}"
count = {"id": "orders", "name": "订单数", "unit": "单", "decimals": 0, "expression": "cell(nodes.fetch, 0, 'n')"}
blind = Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单数": (BUYERS, "buyers")})
run = await start(graph(name, wide, metrics=[GMV, count], dual={"metrics": ["gmv"]}))
assert run.status == "succeeded", run.error
out = dual_output(run, await events(run.id))
assert [m["metric_id"] for m in out["metrics"]] == ["gmv"] and out["status"] == "agreed"
assert blind.runs == 1 and all("指标：订单数" not in t for t in blind.tasks)
async def test_a_zero_original_compares_by_absolute_error_and_has_no_delta_pct(monkeypatch, shop):
name, _ = shop
voids = {"id": "voids", "name": "作废单数", "unit": "单", "decimals": 0, "expression": "cell(nodes.fetch, 0, 'n')"}
none_void = f"SELECT COUNT(*) AS n FROM orders WHERE status = 'void' AND paid_date {WEEK}"
refunded = f"SELECT COUNT(*) AS n FROM orders WHERE status = 'refunded' AND paid_date {WEEK}"
Blind(monkeypatch, name, {"作废单数": (refunded, "n")})
run = await start(graph(name, none_void, metrics=[voids], dual={"definitions": {}}))
[item] = (await pending(run.id)).payload["summary"]
assert item["original_rendered"].startswith("0") and item["blind_rendered"].startswith("1")
assert item["delta_pct"] is None
def test_zero_originals_use_absolute_error():
from app.engine.nodes.dual_check import _agree, _delta_pct
assert _agree(0, 1e-7, 1e-6) and not _agree(0, 0.5, 1e-6)
assert _agree(100.0, 100.00005, 1e-6) and not _agree(100.0, 100.01, 1e-6)
assert _delta_pct(0, 5) is None and _delta_pct(310.0, 330.0) == 6.45 and _delta_pct(-10, -5) == 50.0
async def test_the_summary_lists_at_most_three_differences(monkeypatch, shop):
name, _ = shop
blind_sql = RIGHT + " AND region <> '华北' AND customer_id <> 999"
Blind(monkeypatch, name, {"销售额": (blind_sql, "gmv")})
run = await start(graph(name, BOTH_WRONG))
payload = (await pending(run.id)).payload
[item] = payload["summary"]
[metric] = artifact_store.load(payload["card_artifact"])["metrics"]
assert len(metric["diff"]["items"]) == 4 and len(item["diff"]) == 3
assert item["diff"] == [i["text"] for i in metric["diff"]["items"][:3]]
# --------------------------------------------------------------------------
# 遮罩
# --------------------------------------------------------------------------
async def test_a_blind_cell_on_a_masked_column_is_not_used(monkeypatch, masked_shop, client):
"""盲算交的那一格在遮罩列上：值不采用（不进审批卡、对照工件、产出），这个指标按无法复算记。"""
name, _ = masked_shop
east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
north = {**GMV, "id": "north", "name": "华北销售额", "expression": "cell(nodes.fetch, 1, 'gmv')"}
east_by_order = BY_ORDER_DATE.replace("WHERE ", "WHERE region = '华东' AND ")
north_masked = (f"SELECT SUM(amount) + 0.25 AS secret_gmv FROM orders WHERE status <> 'refunded' "
f"AND paid_date {WEEK} AND region = '华北'")
Blind(monkeypatch, name, {"华东销售额": (east_by_order, "gmv"), "华北销售额": (north_masked, "secret_gmv")})
run = await start(graph(name, BY_REGION, metrics=[east, north], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
card = artifact_store.load(payload["card_artifact"])
assert [m["metric_id"] for m in payload["summary"]] == ["east"]
assert [m["metric_id"] for m in card["metrics"]] == ["east"]
assert "30.25" not in json.dumps(payload, ensure_ascii=False)
assert "30.25" not in json.dumps(card, ensure_ascii=False)
done = await decide(client, run.id, {"east": "a"})
out = dual_output(done, await events(run.id))
[_, row] = out["metrics"]
assert row["status"] == "unchecked" and "遮罩" in row["reason"] and row["blind_rendered"] is None
assert "30.25" not in json.dumps(out, ensure_ascii=False)
async def test_a_renamed_masked_column_in_the_blind_query_is_masked_before_the_blind_agent_sees_it(
monkeypatch, masked_shop):
"""盲算的查询选出了改了名字的遮罩列（region AS 区域）：交给盲算模型、落进盲算快照之前就是「已遮罩」，快照记下查询
时遮了哪一列；没碰遮罩列的那一列（销售额）照常取值、照常比对。"""
name, _ = masked_shop
east = {**GMV, "id": "east", "name": "华东销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
renamed = (f"SELECT region AS 区域, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} "
"GROUP BY region ORDER BY region")
Blind(monkeypatch, name, {"华东销售额": (renamed, "gmv")})
run = await start(graph(name, BY_REGION, metrics=[east], dual={"definitions": {}}))
assert run.status == "succeeded", run.error
evs = await events(run.id)
out = dual_output(run, evs)
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed"
[end] = [e.data for e in evs if e.type == "tool.end" and e.node_id == "dual" and e.data.get("query_artifact")]
snapshot = artifact_store.load(end["query_artifact"])
assert {row[0] for row in snapshot["rows"]} == {"已遮罩"} and snapshot["masked_at_query"] == ["区域"]
shown = json.loads(end["preview"].split("\n", 1)[1])
assert {row[0] for row in shown["rows"]} == {"已遮罩"}, "交给盲算模型的结果里也是「已遮罩」"
async def test_a_metric_summed_over_a_masked_column_is_not_blind_computed(monkeypatch, tmp_path):
"""复核第 1 轮：金额这一列设了遮罩，指标是 SUM(amount)。核查 agent 的查询结果在工具边界遮好，汇总遮罩列的值也遮
（只有一行的汇总就是那一行的原值），盲算交回的格子注定取不到。所以派出盲算之前按 SQL 认出这一列由遮罩列算出，
直接按无法复算记、说清原因，不白花一次盲算的钱。以前只按列名认，gmv 认不出，盲算照样派出去，最后才在格子引用上
被拒，原因说成「盲算交回的格子引用不可用」。"""
name, source_id = await add_source(build(tmp_path / "amount_masked.db"), options={"mask_columns": "amount"})
try:
blind = Blind(monkeypatch, name, {"销售额": (BLIND_TOTAL, "blind_total")})
run = await start(graph(name, RIGHT))
evs = await events(run.id)
out = dual_output(run, evs)
[row] = out["metrics"]
assert row["status"] == "unchecked" and blind.runs == 0, "没派盲算"
assert row["reason"] == ("输入 cell(nodes.fetch, 0, 'gmv') 由设置了遮罩的列算出（经改名、函数或汇总），"
"盲算时会被遮罩，不能用于复算")
assert not [e for e in evs if e.type == "tool.end" and e.node_id == "dual"]
finally:
await drop_source(source_id)
async def test_a_metric_filtered_by_a_masked_column_inside_its_sum_is_still_blind_computed(monkeypatch, masked_shop):
"""遮罩列（区域）只在汇总的 CASE 条件里（SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END)）：同写在 WHERE 里，
值全部来自金额，原 SQL 和盲算的查询都不遮，照常比对。"""
name, _ = masked_shop
east = (f"SELECT SUM(CASE WHEN region = '华东' THEN amount ELSE 0 END) AS gmv FROM orders "
f"WHERE status <> 'refunded' AND paid_date {WEEK}")
blind = Blind(monkeypatch, name, {"销售额": (east, "gmv")})
run = await start(graph(name, east))
out = dual_output(run, await events(run.id))
assert blind.runs == 1 and out["metrics"][0]["status"] == "agreed", out["metrics"][0]
async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch, masked_shop):
name, _ = masked_shop
first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
blind = Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")})
await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}}))
[task] = blind.tasks
assert "华东" not in task and "region" not in task
async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monkeypatch, masked_shop, client):
"""原流程按遮罩列（region）分组、口径卡取华东那一行：差异说明收窄到这一行时，不能拿这一行的分组值补等值条件——
那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄，照整条比。
盲算没分组、按下单日筛：收窄了的话，两边差在时间列和原流程多出来的「region = '华东'」，说明和对照里都带着这个取值；
照整条比，时间列照样说得出，原流程的分组、多选的列归纳不了（_compared_whole）。"""
name, _ = masked_shop
first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")})
run = await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
[item] = payload["summary"]
assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310")
card = artifact_store.load(payload["card_artifact"])
done = await decide(client, run.id, {"first": "a"})
assert done.status == "succeeded", done.error
evs = await events(run.id)
_no_masked_value(payload, card, done, evs)
_compared_whole(payload, card)
# 差异说明里有未能归纳的部分：选了也不写目录
[wrote] = logs(evs, "dual_check_catalog")
assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"]
async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, masked_shop, client):
"""原流程把遮罩列改了名再分组（region AS area）：结果列名 area 不在遮罩里，分组值却还是 region 的取值。收窄时看的
是分组键用到的列，一样不拿这一行的分组值补等值条件，「华东」不进审批卡、对照工件、产出和事件，照整条比。"""
name, _ = masked_shop
renamed = (f"SELECT region AS area, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' "
f"AND paid_date {WEEK} GROUP BY region ORDER BY region")
first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
blind = Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")})
run = await start(graph(name, renamed, metrics=[first], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
[item] = payload["summary"]
assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310")
card = artifact_store.load(payload["card_artifact"])
done = await decide(client, run.id, {"first": "a"})
assert done.status == "succeeded", done.error
_no_masked_value(payload, card, done, await events(run.id))
_compared_whole(payload, card)
# 交给盲算的这一行的分组取值也不给：结果列 area 的值取自遮罩列 region（以前 _dims 只按结果列名认遮罩）
[task] = blind.tasks
assert "华东" not in task, task
#: 外层按 area 分组、取每组的销售额
_BY_AREA = f"SELECT area, SUM(amount) AS gmv FROM {{}} WHERE status <> 'refunded' AND paid_date {WEEK} GROUP BY area ORDER BY area"
#: 遮罩列在子查询、WITH 里改了名（直接起别名、套函数、写在列名表里、隔了两层），外层再按新名字分组：外层的分组键 area
#: 看不出取自遮罩列 region
RENAMED_INSIDE = {
"subquery": _BY_AREA.format("(SELECT region AS area, amount, status, paid_date FROM orders) t"),
"cte": "WITH t AS (SELECT region AS area, amount, status, paid_date FROM orders) " + _BY_AREA.format("t"),
"function": "WITH t AS (SELECT UPPER(region) AS area, amount, status, paid_date FROM orders) " + _BY_AREA.format("t"),
"column_list": ("WITH t(area, amount, status, paid_date) AS (SELECT region, amount, status, paid_date FROM orders) "
+ _BY_AREA.format("t")),
"nested": _BY_AREA.format("(SELECT area, amount, status, paid_date FROM (SELECT region AS area, amount, status, "
"paid_date FROM orders) a) b"),
}
@pytest.mark.parametrize("shape", list(RENAMED_INSIDE))
async def test_a_masked_group_column_renamed_inside_is_not_narrowed_or_shown(monkeypatch, masked_shop, client, shape):
"""遮罩列在子查询或 WITH 里改了名（region AS area），外层按 area 分组：外层只看得到 area，以前照常拿这一行的分组值
补上「area = '华东'」，差异说明把别名还原成「region = '华东'」，进了审批卡、对照工件和目录日志；交给盲算的这一行的
分组取值里也有「华东」。现在按 SQL 认出取值来自遮罩列的名字（同审批质询认改了名的遮罩列）：不收窄、照整条比，
盲算任务里也不给这一组的取值。"""
name, _ = masked_shop
first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
blind = Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")})
run = await start(graph(name, RENAMED_INSIDE[shape], metrics=[first], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
[item] = payload["summary"]
assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310")
card = artifact_store.load(payload["card_artifact"])
done = await decide(client, run.id, {"first": "a"})
assert done.status == "succeeded", done.error
evs = await events(run.id)
_no_masked_value(payload, card, done, evs)
[task] = blind.tasks
assert "华东" not in task, task
def _compared_whole(payload: dict, card: dict) -> None:
"""差异说明照整条比的痕迹（原流程按地区分组取一行，盲算没分组、按下单日筛）：时间列照样说得出；原流程的分组、多选的
地区列归纳不了，另起一条「另有未能归纳的差异」；规范化后的对照里还留着原流程的 GROUP BY。收窄了的话分组去掉、换成
等值条件，两边都是简单形状，说明里就是时间列和多出来的那个条件，这几样都不会有。"""
[metric] = card["metrics"]
diff = metric["diff"]
assert [i["aspect"] for i in diff["items"]] == ["time", "other"], diff["items"]
assert diff["fallback"] is True and any("GROUP BY" in line for line in diff["text_diff"]), diff["text_diff"]
[item] = payload["summary"]
assert len(item["diff"]) == 2 and "order_date" in item["diff"][0] and "另有未能归纳的差异" in item["diff"][1], item
assert item["fallback"] is False
async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client):
"""盲算自己按遮罩列分组、交了华东那一行：同样不拿这一行的分组值改写盲算的 SQL。"""
name, _ = masked_shop
Blind(monkeypatch, name, {"销售额": (BY_REGION, "gmv")})
run = await start(graph(name, BY_ORDER_DATE))
assert run.status == "interrupted", run.error
payload = (await pending(run.id)).payload
assert payload["summary"][0]["blind_rendered"].startswith("300")
card = artifact_store.load(payload["card_artifact"])
done = await decide(client, run.id, {"gmv": "b"})
assert done.status == "succeeded", done.error
_no_masked_value(payload, card, done, await events(run.id))
async def test_a_with_clause_that_does_not_touch_masked_columns_is_still_narrowed(monkeypatch, masked_shop):
"""数据源设了遮罩，原流程的 WITH 里却没用到遮罩列、按渠道分组：照常收窄到这一行（补上 channel = 'offline'），差异
说明说得出这个条件。认改了名的遮罩列是按 SQL 追到底的，不是见到子查询、WITH 就一律不收窄。"""
name, _ = masked_shop
by_channel = (f"WITH t AS (SELECT channel, amount, status, paid_date FROM orders) SELECT channel, SUM(amount) AS gmv "
f"FROM t WHERE status <> 'refunded' AND paid_date {WEEK} GROUP BY channel ORDER BY channel")
first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}
blind = Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")})
run = await start(graph(name, by_channel, metrics=[first], dual={"definitions": {}}))
assert run.status == "interrupted", run.error
[item] = (await pending(run.id)).payload["summary"]
assert item["original_rendered"].startswith("230") and item["blind_rendered"].startswith("330")
assert any("channel = 'offline'" in d for d in item["diff"]), item["diff"]
# 渠道不是遮罩列：交给盲算的这一行的分组取值照给
[task] = blind.tasks
assert "「offline」" in task, task
#: 销售额（第一份快照的一格）加上首组地区名的字数（第二份快照按位置取的一格）
MIX = {"id": "mix", "name": "销售额加首组名称字数", "unit": "元", "decimals": 1, "format": "thousands",
"expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows[0][0])"}
#: 数据整形解析第二份查询结果的两种写法。原样解析（模板整串就是 {{ nodes.regions }}）的，vars.res.rows[0][0] 确定地对到
#: 那份快照的第 0 行第 0 列，卡上记成 cell() 一样的单元格取数（engine/lineage.py）；模板带了过滤器的不算原样解析，按位置
#: 取的格卡上只记路径和值，遮罩顺着取值链认（_positional_masks）——下面几个测试守的是后一条路
PARSED = "{{ nodes.regions }}"
FILTERED = "{{ nodes.regions | trim }}"
def with_regions(spec: dict, source: str, template: str = FILTERED) -> dict:
"""取数之后另查一次按地区分组的结果（第二份快照），数据整形把它解析进 vars.res：口径卡按位置取它的格。"""
nodes = list(spec["nodes"])
at = next(i for i, n in enumerate(nodes) if n["id"] == "fetch") + 1
nodes[at:at] = [node("regions", "tool", tool=f"db_query__{source}", args={"sql": BY_REGION}),
node("parse", "transform", mode="json", template=template, assign_to="res")]
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
async def test_a_masked_cell_read_by_position_from_another_snapshot_is_not_blind_computed(monkeypatch, masked_shop):
"""口径卡上一个指标：cell() 取第一次查询（A，不碰遮罩列）的销售额，另按位置取第二次查询（B，按遮罩列 region 分组）的
首组名称（vars.res.rows[0][0]）。B 不是 cell() 用到的快照：以前预检只为 A 算遮罩，认不出按位置取的那一格在遮罩列上，
指标照常盲算，「华东」作为字面量写进了交给盲算模型的计算式。现在按需算 B 的遮罩：这个指标按无法复算记缺口、不盲算，
遮罩列的取值不进盲算的任务、节点产出和节点发的事件。"""
name, _ = masked_shop
blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")})
run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name))
assert run.status == "succeeded", run.error
evs = await events(run.id)
out = dual_output(run, evs)
[row] = out["metrics"]
assert row["metric_id"] == "mix" and row["status"] == "unchecked", row
# 和 cell() 取自遮罩列的说法一致，点名是哪个输入：指标有好几个输入时审核人才找得到是哪一格
assert row["reason"] == "输入 vars.res.rows[0][0] 取自设置了遮罩的列，不能用于复算", row["reason"]
assert out["status"] == "unchecked" and out["issuance_gaps"] == [GAP_UNCHECKED.format(name=MIX["name"],
reason=row["reason"])]
assert blind.tasks == [] and blind.runs == 0, blind.tasks
assert "华东" not in json.dumps(out, ensure_ascii=False)
assert not [e.type for e in evs if e.node_id == "dual" and not e.type.startswith(("tool.", "llm."))
and "华东" in json.dumps(e.data, ensure_ascii=False)]
async def test_an_unmasked_cell_read_by_position_from_another_snapshot_is_blind_computed(monkeypatch, shop):
"""同样的图、数据源没设遮罩：预检照常通过，指标进入盲算（按位置取的那一格作为字面量写进计算式），两边一致。拦下上面
那个指标的确实是遮罩。"""
name, _ = shop
blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")})
run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name))
assert run.status == "succeeded", run.error
out = dual_output(run, await events(run.id))
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed", out
[task] = blind.tasks
assert "len('华东')" in task
async def test_a_cell_read_through_a_plain_parse_is_judged_like_cell(monkeypatch, masked_shop, shop):
"""数据整形只是原样解析第二次查询的结果：vars.res.rows[0][0] 卡上记成单元格取数（落在第二份快照的第 0 行 region 列），
和 cell(nodes.regions, 0, 'region') 一个说法——遮罩列上的照实说是遮罩、不盲算、取值不进任何产出；没遮罩时它是查询来
的文本，和 cell() 取的文本格一样不能单独复算，照实说不是数值。"""
for (name, _), reason in ((masked_shop, "输入 vars.res.rows[0][0] 取自设置了遮罩的列，不能用于复算"),
(shop, "输入 vars.res.rows[0][0] 不是数值")):
blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")})
run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name, PARSED))
assert run.status == "succeeded", run.error
evs = await events(run.id)
[card] = [e.data for e in evs if e.type == "node.finished" and e.node_id == "card"]
item = next(i for i in artifact_store.load(card["artifact"])["metrics"][0]["inputs"]
if i["path"] == "vars.res.rows[0][0]")
assert (item["via"], item["through"], item["node_id"], item["locator"]) == (
"tool_cell", "parse", "regions", {"row": 0, "column": "region"}), item
out = dual_output(run, evs)
[row] = out["metrics"]
assert row["status"] == "unchecked" and row["reason"] == reason, row
assert blind.tasks == [] and blind.runs == 0, blind.tasks
assert "华东" not in json.dumps(out, ensure_ascii=False)
#: 按位置取的那一格：数据整形解析好的第二份查询结果（首组地区名），快照 id 是编的
POSITIONAL = {"path": "vars.res.rows[0][0]", "value": "华东", "node_id": "parse", "via": "transform", "status": "ok"}
LOST = "cd" * 32
def parsed_regions(source: str) -> dict:
return {"columns": ["region", "gmv"], "rows": [["华东", 300.0], ["华北", 30.0]], "artifact": LOST, "source": source}
async def test_a_cell_read_by_position_from_an_unreadable_snapshot_is_not_trusted(masked_shop):
"""按位置取的那一格所在的快照读不出来（被改过、文件不在了）：查询当时记下的遮罩无从核对。数据源现在设的遮罩遮着这一列
的，照遮罩算（交给重算的预检按遮罩说）；没遮的也不当成没遮，按无法复算记、原因说清。"""
from app.engine.nodes.dual_check import _positional_masks, _Sources
name, _ = masked_shop
for source, expected in ((name, "输入 vars.res.rows[0][0] 取自设置了遮罩的列，不能用于复算"),
("dc_no_such_source", "输入 vars.res.rows[0][0] 所在的查询快照与哈希不一致，疑似被修改，"
"无法核对这一格是否取自设置了遮罩的列")):
sources = _Sources()
sources.snapshots[LOST] = (None, "查询快照与哈希不一致，疑似被修改")
masks: dict = {}
state = {"state:parse": parsed_regions(source)}
why, shown = await _positional_masks(POSITIONAL, sources, state.get, {"parse": "state:parse"}, masks)
assert (why, shown) == (expected, False), source
assert ("region" in masks[LOST]) is (source == name), masks
async def test_a_cell_whose_origin_cannot_be_told_is_not_written_into_the_formula(masked_shop):
"""按位置取的一格认不出取自哪份结果的哪一列——下标不是字面量（rows[vars.i]）、数据整形改过形状的结果既没带快照的 id
也没带数据源名——遮没遮无从核对：计算式里写成「其他输入」，不把卡上记的值交给盲算模型。认得出、又没遮的照常写值；
不伸进查询结果的行的输入（运行输入、常量）也照常写。"""
from app.engine.nodes.dual_check import _formula, _positional_masks, _Sources
name, _ = masked_shop
reshaped = {"columns": ["region", "gmv"], "rows": [["华东", 300.0], ["华北", 30.0]]}
cases = [
({"path": "vars.res.rows[vars.i][0]", "value": "华东", "node_id": "parse", "via": "transform"},
{**reshaped, "artifact": LOST, "source": name}, False),
({**POSITIONAL}, reshaped, False),
({"path": "vars.limit", "value": 3, "node_id": "parse", "via": "transform"}, reshaped, True),
]
for item, parsed, expected in cases:
state = {"state:parse": parsed}
why, shown = await _positional_masks(item, _Sources(), state.get, {"parse": "state:parse"}, {})
assert (why, shown) == (None, expected), item["path"]
metric = {"expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows[vars.i][0]) + vars.limit",
"inputs": [{"path": "cell(nodes.fetch, 0, 'gmv')", "value": 330.0},
{"path": "vars.res.rows[vars.i][0]", "value": "华东"}, {"path": "vars.i", "value": 0},
{"path": "vars.limit", "value": 3}]}
formula = _formula(metric, {"cell(nodes.fetch, 0, 'gmv')": "输入1"}, {"vars.res.rows[vars.i][0]"})
assert formula == "输入1 + len(其他输入) + 3", formula
#: 销售额，加上第二份快照的行数（容器）和首组的销售额（按位置取的一格，不在遮罩列上）
MIX_COUNTS = {"id": "mix", "name": "销售额加地区数和首组销售额", "unit": "元", "decimals": 1, "format": "thousands",
"expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows) + vars.res.rows[0][1]"}
async def test_reading_a_cell_by_position_does_not_change_how_a_container_input_is_judged(monkeypatch, masked_shop):
"""按位置取的那一格在第二份快照（B）的没遮的列上，同一个指标另用了 B 的行数（容器，计算式里写成「其他输入」，值不交给
盲算）。按位置取的那一格只为认遮罩算 B 的遮罩，不把 B 当成 cell() 用到的快照：以前 B 因此进了重算的快照清单，容器的
完整值对上了 B，B 有一列遮罩就整个按「用到了遮罩列」挡下——只写 len(vars.res.rows) 能核、加上这一格就不能核了。"""
name, _ = masked_shop
blind = Blind(monkeypatch, name, {MIX_COUNTS["name"]: (RIGHT, "gmv")})
run = await start(with_regions(graph(name, RIGHT, metrics=[MIX_COUNTS], dual={"definitions": {}}), name))
assert run.status == "succeeded", run.error
out = dual_output(run, await events(run.id))
assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed", out
[task] = blind.tasks
assert "len(其他输入) + 300.0" in task and "华东" not in task, task
def _no_masked_value(payload: dict, card: dict, done: Run, evs: list[RunEvent], value: str = "华东") -> None:
"""遮罩列上的取值不出现在审批 payload、对照工件、节点产出和节点发的事件里。查询本身的 tool.end、模型的往来按
设计带着原始结果（证据面板显示时再遮），不在此列。"""
seen = {"payload": payload, "card": card, "output": dual_output(done, evs),
"events": [e.data for e in evs if (e.node_id == "dual" or e.type.startswith("human.") or e.type == "log")
and not e.type.startswith(("tool.", "llm."))]}
for where, content in seen.items():
assert value not in json.dumps(content, ensure_ascii=False), where
# --------------------------------------------------------------------------
# 设置：全局开关、节点覆盖、发起时的快照
# --------------------------------------------------------------------------
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)
Reference in New Issue View Git Blame Copy Permalink