diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index 6bde1842..4f101d61 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -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"] diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index b1c47136..94a32f0f 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -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)}) diff --git a/backend/tests/fixtures/dual_check_shop.py b/backend/tests/fixtures/dual_check_shop.py index 7ac6bff7..565ca784 100644 --- a/backend/tests/fixtures/dual_check_shop.py +++ b/backend/tests/fixtures/dual_check_shop.py @@ -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") diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index 6f8067f8..9c6299ae 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -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 diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 9703144d..6ce5a013 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -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 = '华东' 是这个指标自己的筛选 范围,写成表的有效记录条件,以后所有查订单表的工作流(包括下一次盲算)都会只看华东。"""