AgentLab/backend/tests/test_dual_check_node.py

"""双盲复算节点（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")},
          rationale="只查到了不含退款的合计，还没来得及按支付日核对。")
    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}
    reason = "盲算已达每项最多查询次数（1 次）；盲算交回的理由：只查到了不含退款的合计，还没来得及按支付日核对。"
    assert out["issuance_gaps"] == [f"指标「销售额」无法复算：{reason}"]
    [said] = logs(evs, "dual_check_unchecked")
    assert said["metrics"] == [{"id": "gmv", "name": "销售额", "reason": reason}]
    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"] == [f"双盲复算「复核销售额」：指标「销售额」无法复算：{reason}"]


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)


# --------------------------------------------------------------------------
# 查询区间里这张表没有数据（data/coverage.py）：不判一致，按无法复算记
# --------------------------------------------------------------------------

#: 订单表只有 2026 年的数：2030 年那一周表里没有数据，COUNT 是 0
EMPTY_WEEK = "BETWEEN '2030-01-07' AND '2030-01-13'"
NO_ORDERS = f"SELECT COUNT(*) AS orders FROM orders WHERE status <> 'refunded' AND paid_date {EMPTY_WEEK}"
ORDERS = {"id": "orders", "name": "订单数", "unit": "单", "expression": "cell(nodes.fetch, 0, 'orders')"}


async def test_an_original_cell_without_data_is_not_checked(monkeypatch, shop):
    """原值所在的格子落在 no_data 的列上：那个 0 不代表真的是 0，盲算查到 0 也说明不了一致。不派盲算，按无法复算记。"""
    name, _ = shop
    blind = Blind(monkeypatch, name, {"订单数": (NO_ORDERS, "orders")})
    run = await start(graph(name, NO_ORDERS, metrics=[ORDERS], dual={"definitions": {"orders": "已支付订单数"}}))
    assert run.status == "succeeded", run.error
    evs = await events(run.id)
    out = dual_output(run, evs)
    [row] = out["metrics"]
    assert row["status"] == "unchecked" and row["reason"].startswith("查询区间里这张表没有数据（输入 cell(nodes.fetch")
    assert "2030-01-07 到 2030-01-13" in row["reason"]
    assert "step" not in blind.calls, "原值就没有数据：不白派一次盲算"
    assert out["counts"] == {"agreed": 0, "disagreed": 0, "unchecked": 1}


async def test_a_blind_cell_without_data_is_not_agreed(monkeypatch, shop):
    """盲算交回的格子落在 no_data 的列上（它查了一段表里没有数据的区间）：不判一致，按无法复算记。"""
    name, _ = shop
    buyers = {"id": "orders", "name": "订单数", "unit": "单", "expression": "cell(nodes.fetch, 0, 'orders')"}
    original = f"SELECT COUNT(*) AS orders FROM orders WHERE status <> 'refunded' AND paid_date {WEEK}"
    Blind(monkeypatch, name, {"订单数": (NO_ORDERS.replace("AS orders", "AS n"), "n")})
    run = await start(graph(name, original, metrics=[buyers], dual={"definitions": {"orders": "已支付订单数"}}))
    assert run.status == "succeeded", run.error
    [row] = dual_output(run, await events(run.id))["metrics"]
    assert row["status"] == "unchecked", row
    assert row["reason"].startswith("查询区间里这张表没有数据（盲算查到的那一格：「n」查的是 2030-01-07 到 2030-01-13")