yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 04:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

双盲复算写目录的预告和真写逐字一致：被目录挡下（已确认、驳回过、和已有条件合不起来）的那一边也带上写到哪张表哪一项、本来要写什么和原因；几个指标落到同一项上时，preview_joint 按真写的办法（归拢、拿掉驳回的、对目录）把每种选择组合算一遍放进对照工件的 catalog_joint，组合太多的那一项不预告；真写的日志 written 带上写进去的值。对照工件里盲算的输入另带 rationale_doc（核查 agent 组装的理由文档，裸数字已标出）；补图级测试，按审批卡的认法断言预告与实际写入一致

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-05 18:56:04 +08:00
parent 375f7165e5
commit 9eeb11d29e
5 changed files with 373 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/engine/dual_check_facts.py
+118 -35
View File
@@ -36,10 +36,11 @@ is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不
也不写。
**预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么（preview）：按发起审批时的目录，选原流程、选盲算各会
写什么、不写的为什么。和真写是同一套判断（derive 认事实、_plan 对目录、_revives 认驳回），所以只在两处可能和最后写
的不一样：审批停下期间目录被人改了；同一次裁决里几个指标落到同一项上（上一段）——写进去的是合起来的说法，预告说会
写的也可能因为说法互相矛盾、和已有条件的关系拿不准、合起来又成了驳回过的说法而不写。反过来不会：预告说不写的，
不会因为和别的指标并成一项被写进去。
写什么、不写的为什么。认出了事实、只是被目录挡下（已确认、驳回过、和已有条件合不起来）的那一边也带上写到哪张表的
哪一项、本来要写什么，卡上才认得出它和别的指标是不是落到同一项上。同一次裁决里几个指标落到同一项上时，写进去的是
合起来的说法（上一段），单个指标的预告说不准：preview_joint 把这几个指标各自选哪一边的每种组合都按真写的办法算一遍
（_merged 归拢、_screen 拿掉驳回的、_plan 对目录），卡上按各指标当前的选择取对应的那一种。和真写是同一套判断、同一句
原因，只有审批停下期间目录被人改了时才会和最后写的不一样。组合太多的（JOINT_LIMIT）那一项不预告，卡上有条件地说。
catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。
"""
@@ -73,6 +74,9 @@ ASPECT_LABEL = {
_VERB = {"count": "计数", "sum": "求和", "avg": "求平均"}
#: 写入时版本对不上（别人刚改过这张表）最多重试几次
_RETRIES = 2
#: 同一次裁决里几个指标落到同一项上时，最多预告多少种组合（preview_joint）。每个指标两边都落在这一项上时是
#: 3 的指标数次方：四个指标 72 种还在里面，再多的那一项不预告，审批卡回到按其余指标当前的选择有条件地说
JOINT_LIMIT = 128
#: 无效记录的取值：被排除的取值（或码值表里它的意思）里有这些词，才算在排除无效记录
_INVALID_WORDS = frozenset({
@@ -94,7 +98,7 @@ class Fact:
"""要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明；sql、dialect 是被采纳
那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。同一次裁决里几个指标落到同一项上
时并成一项：members 是并进来的每个指标各自的那一项（各带自己的说法，审批卡上各自的预告就是按它算的），value 是
合起来之后的说法；单个指标的 members 为空。"""
合起来之后的说法；单个指标的 members 为空。metric_id 是指标 ID：几个指标并成一项时，预告按它说清每个指标的结果。"""
source: str
table: str
@@ -105,6 +109,7 @@ class Fact:
sql: str = ""
dialect: str | None = None
members: tuple[Fact, ...] = ()
metric_id: str = ""
def _members(fact: Fact) -> tuple[Fact, ...]:
@@ -160,7 +165,8 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
dialect = candidate.get("dialect")
return Fact(source=str(candidate.get("source") or ""), table=resolved, item=key, value=value, metric=name,
choice=str(side), sql=str(sql.get(side) or ""),
dialect=dialect if isinstance(dialect, str) else None), None
dialect=dialect if isinstance(dialect, str) else None,
metric_id=str(candidate.get("metric_id") or "")), None
def _business_date(item: Mapping[str, Any], side: str) -> tuple[tuple[str, str, Any] | None, str | None]:
@@ -631,33 +637,35 @@ def _rejected_text(fact: Fact, rejected: Any, members: Sequence[Fact]) -> str:
"的选择相反，未写入。如需更改，请到数据目录中修改")
def _screen(notes: Mapping[str, Any], fact: Fact) -> tuple[Fact | None, list[str]]:
"""几个指标并成的一项，先拿掉自己采纳的说法曾被人工驳回的指标：(剩下的重新并成的一项, 拿掉的为什么不写)，一个
都不剩时前者为 None。单个指标的、这一项没被驳回过的原样返回，由 _plan 照常判断。
def _screen(notes: Mapping[str, Any], fact: Fact) -> tuple[Fact | None, list[Fact], list[str]]:
"""几个指标并成的一项，先拿掉自己采纳的说法曾被人工驳回的指标：(剩下的重新并成的一项, 拿掉的指标, 没写的原因)，
一个都不剩、或者剩下的又合不起来时前者为 None；原因先是拿掉的那几个的（一条），剩下的合不起来时再跟一条。单个指标
的、这一项没被驳回过的原样返回，由 _plan 照常判断。
审批卡上每个指标的预告是按它自己的说法单独算的（preview → _plan），在那里被驳回检查拦下、卡上写着「不写」的，
并进来以后也得拦下。只拿合起来的整句去比，「status <> 'refunded'」和别的条件 AND 在一起就不再等于驳回的那句，
会被写成人工确认，和卡上说的正相反。拿掉的指标照实说原因（和它的预告同一句话），其余的照写，也和各自的预告一致。"""
rejected, said = _rejected(notes, fact.item)
if not rejected or not fact.members:
return fact, []
return fact, [], []
out: list[Fact] = []
kept: list[Fact] = []
for member in fact.members:
(out if _revives(member, said) else kept).append(member)
if not out:
return fact, []
return fact, [], []
reasons = [_rejected_text(fact, said, out)]
if not kept:
return None, reasons
return None, out, reasons
narrowed, why = _merged(kept)
return narrowed, reasons if why is None else [*reasons, why]
return narrowed, out, reasons if why is None else [*reasons, why]
async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str,
actor: str | None) -> dict[str, list[dict[str, Any]]]:
"""把审核人的选择写进数据目录：{written: [{source, table, item}], skipped: [{reason}]}。永远不抛：数据源没了、
目录写不进去，都落在 skipped 里照实说。
"""把审核人的选择写进数据目录：{written: [{source, table, item, value}], skipped: [{reason}]}。value 是真写进去的
说法（和已有条件合起来的就是合起来的整句），审批卡上的预告说「会记为」的就是它。永远不抛：数据源没了、目录写不
进去，都落在 skipped 里照实说。
先把每个候选认成目录项，再按（数据源、表、项）归拢成一项（_merged：说法一样的、有效记录条件合起来写，说法
不同的整组不写），逐项写（_write：按读到的目录先拿掉各自带回驳回说法的指标，其余的照写）。"""
@@ -690,24 +698,25 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no
if fact is None:
skipped.append({"reason": why})
continue
done, reasons = await _record(fact, run_id=run_id, node_title=node_title, actor=actor)
if done:
written.append({"source": fact.source, "table": fact.table, "item": fact.item})
value, reasons = await _record(fact, run_id=run_id, node_title=node_title, actor=actor)
if value is not None:
written.append({"source": fact.source, "table": fact.table, "item": fact.item, "value": value})
skipped += [{"reason": r} for r in reasons]
return {"written": written, "skipped": skipped}
async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, list[str]]:
"""写一项：(写了没有, 没写的原因)。几个指标并成的一项可能写了、同时有几个指标被拿出来没写，原因照样交回。"""
async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[Any, list[str]]:
"""写一项：(写进去的说法，没写为 None, 没写的原因)。几个指标并成的一项可能写了、同时有几个指标被拿出来没写，原因
照样交回。"""
try:
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == fact.source))).scalar_one_or_none()
if row is None:
return False, [f"数据源「{fact.source}」已不存在，指标{_names(_members(fact))}的选择未写入数据目录"]
return None, [f"数据源「{fact.source}」已不存在，指标{_names(_members(fact))}的选择未写入数据目录"]
return await _write(session, row, fact, run_id=run_id, node_title=node_title, actor=actor)
except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记
logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", fact.source)
return False, [f"指标{_names(_members(fact))}的选择写入数据目录时出错，未写入"]
return None, [f"指标{_names(_members(fact))}的选择写入数据目录时出错，未写入"]
def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]:
@@ -767,7 +776,9 @@ def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None
candidate 同 derive（choice 不用给）；notes 是这个数据源的目录（{表名: notes}）。返回 {a: …, b: …}，每边是
{write: true, source, table, item, value}（value 是真要写进去的说法：和目录里已有的有效记录条件合起来的，是合起来
之后的整句）或 {write: false, reason}。和恢复后真写是同一套判断，只在模块说明里说的两处可能不一样。永远不抛。"""
之后的整句）或 {write: false, reason}；认出了事实、只是被目录挡下的不写，照样带上 source、table、item 和本来要写
的 value（这个指标自己的说法）——卡上据此认出几个指标落到同一项上。这个指标单独落在这一项上时，reason 和真写时
跳过的原因一字不差；和别的指标落到同一项上时看 preview_joint。永远不抛。"""
notes = notes or {}
tables = schema_cache.get("tables") if isinstance(schema_cache, Mapping) else None
out: dict[str, dict[str, Any]] = {}
@@ -780,15 +791,87 @@ def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None
except Exception: # noqa: BLE001 - 预告算不出来不能拦住审批：照实说，以写入后的结果为准
logger.exception("双盲复算预告数据目录的写入失败（指标 %s）", candidate.get("metric"))
fact, why = None, f"指标「{candidate.get('metric')}」的选择核对数据目录时出错，无法预告是否写入"
out[side] = {"write": True, "source": fact.source, "table": fact.table, "item": fact.item, "value": value} \
if fact is not None and why is None else {"write": False, "reason": why}
if fact is None:
out[side] = {"write": False, "reason": why}
continue
where = {"source": fact.source, "table": fact.table, "item": fact.item}
out[side] = {"write": True, **where, "value": value} if why is None else \
{"write": False, **where, "value": fact.value, "reason": why}
return out
def _joint_outcome(notes: Mapping[str, Any], members: list[Fact], tables: Mapping[str, Any] | None
) -> dict[str, dict[str, Any]]:
"""几个指标落到同一项上、各按自己的选择并成一项时，每个指标的结果：{指标 ID: {write: true, value} 或 {write: false,
reason}}。和 record_choices 逐步一样：_merged 归拢（说法不同的整组不写），_screen 拿掉带回驳回说法的（它们说驳回
那一句，剩下的合不起来时说合不起来的那一句），剩下的按 _plan 对目录（写的是同一个值，不写的是同一句原因）。"""
fact, why = _merged(members)
if fact is None:
return {m.metric_id: {"write": False, "reason": why} for m in members}
kept, out, reasons = _screen(notes, fact)
result = {m.metric_id: {"write": False, "reason": reasons[0]} for m in out}
rest = [m for m in members if m.metric_id not in result]
if kept is None:
result.update({m.metric_id: {"write": False, "reason": reasons[-1]} for m in rest})
return result
value, why = _plan(notes, kept, tables)
result.update({m.metric_id: {"write": False, "reason": why} if why is not None else {"write": True, "value": value}
for m in rest})
return result
def preview_joint(candidates: Sequence[Mapping[str, Any]],
catalogs: Mapping[str, tuple[Mapping[str, Any] | None, Mapping[str, Mapping[str, Any]]]]
) -> list[dict[str, Any]]:
"""同一次裁决里几个指标落到同一项上时的预告：[{source, table, item, members: [{metric_id, choice}], outcomes:
{指标 ID: {write: true, value} 或 {write: false, reason}}}]，每条是一种组合（members 是按各自的选择落到这一项上的
指标，照候选的先后，至少两个）。
candidates 照节点交给真写的先后（不一致的指标的先后），带 metric_id，不带 choice；catalogs 是 {数据源名: (表结构,
{表名: notes})}，同 preview 的两个参数。每个指标的两边先各按 derive 认出落到哪一项（同 preview），落到同一项上的
几个指标，每种「谁选了哪一边」的组合按 _joint_outcome 算一遍：真写时这几个指标就是这样被归拢、写或不写的。组合超过
JOINT_LIMIT 的那一项整项不预告。永远不抛：算不出来的那一项不预告。"""
from itertools import product
landed: dict[tuple[str, str, str], dict[int, list[tuple[str, Fact]]]] = {}
for index, candidate in enumerate(candidates):
schema, notes = catalogs.get(str(candidate.get("source") or ""), (None, {}))
for side in SIDE_LABEL:
try:
fact, _ = derive({**candidate, "choice": side}, schema, notes)
except Exception: # noqa: BLE001 - 认不出就当这一边落不到任何一项上，同 preview 照实不说
fact = None
if fact is not None:
landed.setdefault((fact.source, fact.table, fact.item), {}).setdefault(index, []).append((side, fact))
out: list[dict[str, Any]] = []
budget = JOINT_LIMIT
for (source, table, item), by_metric in landed.items():
if len(by_metric) < 2:
continue
options = [[None, *sides] for _, sides in sorted(by_metric.items())]
combos = [[c for c in combo if c is not None] for combo in product(*options)]
combos = [c for c in combos if len(c) >= 2]
if len(combos) > budget:
continue
budget -= len(combos)
schema, notes = catalogs.get(source, (None, {}))
tables = schema.get("tables") if isinstance(schema, Mapping) else None
try:
entries = [{"source": source, "table": table, "item": item,
"members": [{"metric_id": fact.metric_id, "choice": side} for side, fact in combo],
"outcomes": _joint_outcome((notes or {}).get(table) or {}, [fact for _, fact in combo], tables)}
for combo in combos]
except Exception: # noqa: BLE001 - 预告算不出来不能拦住审批：这一项不预告，卡上有条件地说
logger.exception("双盲复算预告几个指标落到同一项上的写入失败（%s 的 %s）", table, item)
continue
out.extend(entries)
return out
async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node_title: str,
actor: str | None) -> tuple[bool, list[str]]:
"""写一项：(写了没有, 没写的原因)。几个指标并成的一项先按读到的目录拿掉各自带回驳回说法的指标（_screen），写什么、
写不写再由 _plan 定；备注只列真写进去的指标。版本对不上（别人刚改过）重读再来。"""
"""写一项：(写进去的说法，没写为 None, 没写的原因)。几个指标并成的一项先按读到的目录拿掉各自带回驳回说法的指标
（_screen），写什么、写不写再由 _plan 定；备注只列真写进去的指标。版本对不上（别人刚改过）重读再来。"""
label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item)
tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None
reasons: list[str] = []
@@ -796,19 +879,19 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node
entry = await catalog.read_entry(session, row.id, fact.table)
notes = entry.notes if entry else {}
version = entry.version if entry else 0
kept, reasons = _screen(notes, fact)
kept, _, reasons = _screen(notes, fact)
if kept is None:
return False, reasons
return None, reasons
value, why = _plan(notes, kept, tables)
if why is not None:
return False, [*reasons, why]
return None, [*reasons, why]
note = note_of(kept, run_id=run_id, node_title=node_title)
at = catalog.now_iso()
try:
merged = catalog.apply_patch(notes, [{"path": kept.item, "value": value, "reason": note}],
table=kept.table, tables=tables, at=at)
except catalog.CatalogInvalid as e:
return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
return None, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
# 不论原来是谁提的：这一项现在出自人的选择，记为人工填写、已确认，备注写明出处
merged[kept.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at)
try:
@@ -817,9 +900,9 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node
await session.rollback()
continue
except catalog.CatalogInvalid as e:
return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
return True, reasons
return False, [*reasons, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"]
return None, [*reasons, f"「{kept.table}」的{label}未写入数据目录：{e}"]
return value, reasons
return None, [*reasons, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"]
__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"]
__all__ = ["FACT_ITEMS", "JOINT_LIMIT", "Fact", "derive", "note_of", "preview", "preview_joint", "record_choices"]
backend/app/engine/nodes/dual_check.py
+14 -3
View File
@@ -21,7 +21,8 @@ SQL，就跟着差异说明进了审批卡、对照工件和目录日志。完
缺口；b、neither、没选的写进产出的 issuance_gaps，成果节点的出具契约据此降档（on_reject 为 fail 时节点直接失败）。
两边只差一个方面、又对得上数据目录事实的，选择写进数据目录（engine/dual_check_facts.py，指标自己的筛选范围不写）。
选了会怎样要在卡上看得出：payload 在约定的几项之外带上实际生效的 on_reject，摘要里每个指标带 catalog——选 a、选 b
按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）。
按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）；几个指标落到目录
同一项上时每种选择组合合起来会怎样写（preview_joint），放在对照工件的 catalog_joint 里。
盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。
**产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial；
@@ -57,7 +58,7 @@ from app.db.models import DataSource
from app.engine import recompute
from app.engine.approval import read_decision
from app.engine.context import NodeContext, NodeError
from app.engine.dual_check_facts import preview, record_choices
from app.engine.dual_check_facts import preview, preview_joint, record_choices
from app.engine.evidence import build_catalog, is_number, next_exec
from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \
parse_expression
@@ -807,6 +808,8 @@ def _slim(outcome: Any) -> dict[str, Any]:
"row": cell.get("row"), "column": cell.get("column"), "eid": cell.get("eid"),
"sql": query.get("sql"), "source": query.get("source"),
"rationale": (outcome.result or {}).get("rationale") or "",
# 理由组装成的文档（evidence.compose_doc）：数字有出处的挂着盲算自己查询的格子，裸数字已标出。审批卡照它画理由
"rationale_doc": outcome.rationale if isinstance(outcome.rationale, dict) else None,
"queries": [{k: q.get(k) for k in ("alias", "artifact", "sql", "row_count", "truncated", "source")}
for q in outcome.queries],
"notes": list(outcome.notes),
@@ -966,7 +969,8 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
{"path": i["path"], "value": (results.get(i["key"]) or {}).get("value"),
"sql": (results.get(i["key"]) or {}).get("sql"), "source": (results.get(i["key"]) or {}).get("source"),
"alias": (results.get(i["key"]) or {}).get("alias"),
"rationale": (results.get(i["key"]) or {}).get("rationale") or ""}
"rationale": (results.get(i["key"]) or {}).get("rationale") or "",
"rationale_doc": (results.get(i["key"]) or {}).get("rationale_doc")}
for i in row["inputs"]]},
"diff": {"equivalent": merged["equivalent"],
"items": [{"aspect": i.get("aspect"), "text": i.get("text")} for i in merged["items"]],
@@ -991,6 +995,13 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
"catalog": preview(facts[row["metric_id"]], info.get("schema"), info.get("notes")),
})
content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))}
# 同一次裁决里几个指标落到数据目录的同一项上时，按各自的选择合起来写（dual_check_facts._merged）：每个指标单独的
# 预告说不准，把每种组合的结果都算好放进对照工件（payload 里放不下，审批卡打开时就取工件），卡上按当前的选择取
catalogs = {name: (info.get("schema"), info.get("notes") or {}) for name, info in infos.items()}
joint = await asyncio.to_thread(preview_joint, [facts[r["metric_id"]] for r in rows if r["metric_id"] in facts],
catalogs)
if joint:
content["catalog_joint"] = joint
try:
artifact = await artifact_store.put_json(content, kind="dual_check", run_id=ctx.run.run_id,
node_id=ctx.node.id, meta={"metrics": len(metrics)})
backend/tests/fixtures/dual_check_shop.py
Vendored
+4 -2
View File
@@ -87,10 +87,11 @@ class Blind:
没对上指标名的盲算交 null。不是盲算的调用（报告撰写）返回 writer 的正文并记下收到的消息。
calls 记下每次盲算调用（step / final），tasks 记下盲算收到的任务原文（几个盲算同时进行时也是每个任务一条）。
on_query：盲算第一次查库之前调用一次（比如往库里插一行，造出「两次取数之间数据变了」）。
rationale：盲算交的理由，{ref} 换成交的那一格的引用（缺省「按口径说明统计，结果是 {ref}。」）。
"""
def __init__(self, monkeypatch, source: str, plans: dict, *, writer: str = "本周销售额见口径卡。", row: int = 0,
on_query=None):
on_query=None, rationale: str = "按口径说明统计，结果是 {ref}。"):
from app.providers import mock_model
self.source = source
@@ -98,6 +99,7 @@ class Blind:
self.writer = writer
self.row = row
self.on_query = on_query
self.rationale = rationale
self.calls: list[str] = []
self.tasks: list[str] = []
self.writer_prompts: list[str] = []
@@ -114,7 +116,7 @@ class Blind:
_, column = plan
alias = f"Q{len(blind.sqls(plan))}"
body = {"value": {"alias": alias, "row": blind.row, "column": column},
"rationale": f"按口径说明统计，结果是 [[v:{alias}.r{blind.row}.{column}]]。"}
"rationale": blind.rationale.format(ref=f"[[v:{alias}.r{blind.row}.{column}]]")}
return AIMessage(content=json.dumps(body, ensure_ascii=False))
if model.tools:
blind.calls.append("step")
backend/tests/test_dual_check_facts.py
+133 -10
View File
@@ -252,7 +252,8 @@ async def test_a_choice_is_written_as_a_confirmed_human_item_with_its_origin(sho
name, source_id = shop
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-42",
node_title="复核销售额", actor="张三")
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []}
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date",
"value": {"column": "paid_date"}}], "skipped": []}
entry = await entry_of(source_id)
item = entry.notes["business_date"]
assert item["value"] == {"column": "paid_date"} and (item["source"], item["status"]) == ("human", "confirmed")
@@ -290,7 +291,8 @@ async def test_an_inferred_condition_both_sides_use_is_extended(shop):
base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}"
out = await record_choices([candidate(base, extra("status <> 'refunded'", base), "b", source=name)],
run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter",
"value": "is_test = 0 AND status <> 'refunded'"}], out
item = (await entry_of(source_id)).notes["valid_filter"]
assert item["value"] == "is_test = 0 AND status <> 'refunded'"
assert (item["source"], item["status"]) == ("human", "confirmed")
@@ -327,7 +329,8 @@ async def test_an_inferred_same_condition_is_confirmed(shop):
await seed(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "llm")})
out = await record_choices([candidate(WITH_REFUNDS, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}]
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter",
"value": "status <> 'refunded'"}]
item = (await entry_of(source_id)).notes["valid_filter"]
assert item["value"] == "status <> 'refunded'" and (item["source"], item["status"]) == ("human", "confirmed")
@@ -348,7 +351,8 @@ async def test_the_same_column_inferred_is_confirmed_and_keeps_its_rule(shop):
{"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}, "profile")})
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-1",
node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}]
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date",
"value": {"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}}]
item = (await entry_of(source_id)).notes["business_date"]
assert item["value"] == {"column": "paid_date", "rule": "按付款成功时间", "timezone": "Asia/Shanghai"}
assert (item["source"], item["status"]) == ("human", "confirmed")
@@ -365,7 +369,8 @@ async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_do
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", source=name)], run_id="run-2",
node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date"}]
assert out["written"] == [{"source": name, "table": "orders", "item": "business_date",
"value": {"column": "order_date"}}]
assert (await entry_of(source_id)).notes["business_date"]["value"] == {"column": "order_date"}
@@ -394,7 +399,8 @@ async def test_complementary_valid_filters_in_one_decision_are_combined(shop):
# 第三个指标采纳的是和第一个一样的条件：合并时不重复
candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name),
], run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter",
"value": "status <> 'refunded' AND is_test = 0"}], "skipped": []}
entry = await entry_of(source_id)
item = entry.notes["valid_filter"]
assert item["value"] == "status <> 'refunded' AND is_test = 0" and entry.version == 1
@@ -410,7 +416,8 @@ async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_
both = [candidate(base, extra("status <> 'refunded'", base), "b", metric="销售额", source=name),
candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
out = await record_choices(both, run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter",
"value": "channel <> 'test' AND status <> 'refunded' AND is_test = 0"}], out
assert (await entry_of(source_id)).notes["valid_filter"]["value"] == \
"channel <> 'test' AND status <> 'refunded' AND is_test = 0"
@@ -437,7 +444,8 @@ async def test_a_rejected_condition_is_not_written_back_through_a_combined_decis
assert preview(refunds, SCHEMA, notes)["b"]["write"] is False
assert preview(tests, SCHEMA, notes)["b"]["value"] == "is_test = 0"
out = await record_choices([refunds, tests, counted], run_id="run-1", node_title="复核", actor=None)
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out
assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter",
"value": "is_test = 0"}], out
[skipped] = out["skipped"]
assert "「status <> 'refunded'」曾被人工驳回" in skipped["reason"] and "「销售额」、「订单数」" in skipped["reason"]
assert "订单金额" not in skipped["reason"]
@@ -498,7 +506,8 @@ async def test_combined_filters_do_not_repeat_a_condition_already_in_the_catalog
candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)]
said = preview(picked[1], SCHEMA, {"orders": (await entry_of(source_id)).notes})
out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []}
assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter",
"value": "status <> 'refunded' AND is_test = 0"}], "skipped": []}
item = (await entry_of(source_id)).notes["valid_filter"]
assert item["value"] == said["b"]["value"] == "status <> 'refunded' AND is_test = 0"
@@ -532,7 +541,8 @@ async def test_the_same_choice_from_two_metrics_is_written_once(shop):
out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", metric="销售额", source=name),
candidate(BY_ORDER_DATE, RIGHT, "b", metric="订单金额", source=name)],
run_id="run-1", node_title="复核", actor=None)
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []}
assert out == {"written": [{"source": name, "table": "orders", "item": "business_date",
"value": {"column": "paid_date"}}], "skipped": []}
entry = await entry_of(source_id)
assert entry.version == 1 and "「销售额」" in entry.notes["business_date"]["note"]
assert "「订单金额」" in entry.notes["business_date"]["note"]
@@ -579,3 +589,116 @@ async def test_a_missing_source_or_an_unrecordable_choice_is_reported(shop):
assert out["written"] == []
assert "数据源「no_such_source」已不存在" in out["skipped"][0]["reason"]
assert "不止差一个方面" in out["skipped"][1]["reason"]
# --------------------------------------------------------------------------
# 预告和真写逐字一致：挡下的一边也说写到哪、为什么；几个指标落到同一项上按合并后的结果预告
# --------------------------------------------------------------------------
def test_a_side_the_catalog_blocks_still_says_where_and_why():
"""认出了事实、只是被目录挡下（已确认、驳回过、和已有条件合不起来）的一边：照样带上写到哪张表的哪一项、本来要写
什么，原因和真写时跳过的原因一字不差。认不出事实的一边只有原因"""
confirmed = {"orders": {"business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}}
said = preview(candidate(BY_ORDER_DATE, RIGHT, "a"), SCHEMA, confirmed)
target = {"source": "shop", "table": "orders", "item": "business_date"}
assert said["a"] == {"write": False, **target, "value": {"column": "order_date"},
"reason": "「orders」的业务日期已是人工确认过的「order_date」，无需重复写入"}
assert said["b"]["write"] is False and {k: said["b"][k] for k in target} == target
assert said["b"]["value"] == {"column": "paid_date"} and "已有人工确认过的说法「order_date」" in said["b"]["reason"]
# 认不出事实（采纳的一边没有这个条件）：没有写到哪，只有原因
said = preview(candidate(WITH_REFUNDS, RIGHT, "a"), SCHEMA, {})
assert set(said["a"]) == {"write", "reason"} and said["a"]["write"] is False
def test_the_joint_preview_covers_every_way_two_metrics_land_on_one_item():
"""两个指标都只差时间列：各自选原流程是 order_date、选盲算是 paid_date，落到同一张表的业务日期上。四种组合各预告
一次：说法一样的写，说法不同的两个都不写，原因是真写时那一句"""
from app.engine.dual_check_facts import preview_joint
picked = [{**candidate(BY_ORDER_DATE, RIGHT, "a", metric="销售额"), "metric_id": "gmv"},
{**candidate(BY_ORDER_DATE, RIGHT, "a", metric="订单金额"), "metric_id": "total"}]
joint = preview_joint(picked, {"shop": (SCHEMA, {})})
by = {tuple((m["metric_id"], m["choice"]) for m in e["members"]): e for e in joint}
assert set(by) == {(("gmv", a), ("total", b)) for a in "ab" for b in "ab"}
for entry in joint:
assert {k: entry[k] for k in ("source", "table", "item")} == \
{"source": "shop", "table": "orders", "item": "business_date"}
assert by[(("gmv", "b"), ("total", "b"))]["outcomes"] == {
"gmv": {"write": True, "value": {"column": "paid_date"}},
"total": {"write": True, "value": {"column": "paid_date"}}}
split = by[(("gmv", "a"), ("total", "b"))]["outcomes"]
assert split["gmv"] == split["total"] and split["gmv"]["write"] is False
assert "给出了不同的说法" in split["gmv"]["reason"] and "「销售额」" in split["gmv"]["reason"]
def test_the_joint_preview_screens_out_a_rejected_condition_like_the_write():
"""驳回过「status <> 'refunded'」：两个指标并成一项时，带回驳回说法的那个拿出来不写（原因同真写），另一个照写自己的"""
from app.engine.dual_check_facts import preview_joint
notes = {"orders": {"valid_filter": catalog.make_item("status <> 'refunded'", "llm", "rejected")}}
picked = [{**candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额"), "metric_id": "gmv"},
{**candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额"), "metric_id": "total"}]
[entry] = preview_joint(picked, {"shop": (SCHEMA, notes)})
assert [(m["metric_id"], m["choice"]) for m in entry["members"]] == [("gmv", "b"), ("total", "b")]
assert entry["outcomes"]["total"] == {"write": True, "value": "is_test = 0"}
assert entry["outcomes"]["gmv"]["write"] is False and "曾被人工驳回" in entry["outcomes"]["gmv"]["reason"]
def test_too_many_combinations_are_left_to_the_timeline():
"""几个指标落到同一项上的组合太多（每个两边都落在这一项上，3 的指标数次方）：这一项不预告，审批卡回到有条件地说"""
from app.engine import dual_check_facts as facts
picked = [{**candidate(BY_ORDER_DATE, RIGHT, "a", metric=f"指标{i}"), "metric_id": f"m{i}"} for i in range(6)]
assert facts.preview_joint(picked, {"shop": (SCHEMA, {})}) == []
assert len(facts.preview_joint(picked[:3], {"shop": (SCHEMA, {})})) == 3 ** 3 - 1 - 3 * 2
@pytest.mark.parametrize("choices", [("a", "b"), ("b", "b"), ("b", "a"), ("a", None)])
async def test_the_joint_preview_is_what_gets_written(shop, choices):
"""预告和真写逐字一致：按这组选择，各指标的预告（单独落一项的看 preview，几个落到同一项的看 preview_joint）说写的，
目录里就是那个值；说不写的，原因就在真写跳过的原因里"""
from app.engine.dual_check_facts import preview_joint
name, source_id = shop
await seed(source_id, {"valid_filter": catalog.make_item("is_test = 0", "llm")})
base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}"
picked = [{**candidate(base, extra("status <> 'refunded'", base), "a", metric="销售额", source=name),
"metric_id": "gmv"},
{**candidate(base, extra("channel <> 'test'", base), "a", metric="订单金额", source=name),
"metric_id": "total"}]
notes = {"orders": (await entry_of(source_id)).notes}
singles = {c["metric_id"]: preview(c, SCHEMA, notes) for c in picked}
joint = preview_joint(picked, {name: (SCHEMA, notes)})
chosen = {c["metric_id"]: choice for c, choice in zip(picked, choices) if choice}
expected = said_for(chosen, singles, joint)
out = await record_choices([{**c, "choice": chosen[c["metric_id"]]} for c in picked if c["metric_id"] in chosen],
run_id="run-1", node_title="复核", actor=None)
entry = await entry_of(source_id)
for metric_id, said in expected.items():
if said["write"]:
assert {"source": name, "table": "orders", "item": said["item"], "value": said["value"]} in out["written"]
assert entry.notes[said["item"]]["value"] == said["value"]
else:
assert said["reason"] in [s["reason"] for s in out["skipped"]], (metric_id, said, out)
def said_for(chosen: dict, singles: dict, joint: list) -> dict:
"""审批卡的认法：一个指标选的那一边认出了写到哪，就看按当前的选择还有谁落到同一项上——只有它自己时照单个指标的
预告，几个时照 preview_joint 里这几个成员的那一条"""
out = {}
for metric_id, choice in chosen.items():
side = singles[metric_id][choice]
if "table" not in side:
out[metric_id] = side
continue
where = (side["source"], side["table"], side["item"])
peers = sorted((m, c) for m, c in chosen.items()
if "table" in singles[m][c] and (singles[m][c]["source"], singles[m][c]["table"],
singles[m][c]["item"]) == where)
if len(peers) == 1:
out[metric_id] = side
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
backend/tests/test_dual_check_node.py
+104 -6
View File
@@ -401,9 +401,14 @@ async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch,
assert set(orig) == {"path", "value", "sql", "source", "alias"}
assert (orig["sql"], orig["value"], orig["source"], orig["alias"]) == (BY_ORDER_DATE, 310.0, name, "Q1")
[mine] = metric["blind"]["inputs"]
assert set(mine) == {"path", "value", "sql", "source", "alias", "rationale"}
assert 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
@@ -416,6 +421,19 @@ async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch,
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")})
@@ -436,7 +454,8 @@ async def test_choosing_the_original_leaves_no_gap(monkeypatch, shop, client):
assert item["value"] == {"column": "order_date"} and (item["source"], item["status"]) == ("human", "confirmed")
assert run.id in item["note"] and "复核销售额" in item["note"] and "销售额" in item["note"]
[wrote] = logs(evs, "dual_check_catalog")
assert wrote["written"] == [{"source": name, "table": "orders", "item": "business_date"}] and wrote["skipped"] == []
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):
@@ -593,9 +612,12 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c
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"] == []
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"}
@@ -619,19 +641,95 @@ async def test_a_rejected_filter_is_not_written_back_along_with_another_metric(m
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
said = {s["metric_id"]: s["catalog"]["b"] for s in (await pending(run.id)).payload["summary"]}
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"
done = await decide(client, run.id, {"gmv": "b", "total": "b"})
# 两个指标都落到有效记录条件上：合起来的结果另有预告，和真写一致
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"}]
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 = '华东' 是这个指标自己的筛选
范围，写成表的有效记录条件，以后所有查订单表的工作流（包括下一次盲算）都会只看华东。"""