From be164bd5c13134339a7e1a41bec47a485bebf017 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 09:20:09 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AF=81=E6=8D=AE=E6=8E=A5=E5=8F=A3=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=E6=95=B4=E6=94=B9=EF=BC=88=E5=90=8E=E7=AB=AF=EF=BC=89?= =?UTF-8?q?=EF=BC=9A=E5=8F=8D=E6=96=B9=E5=AE=A1=E7=A8=BF=E5=B8=A6=E6=95=B4?= =?UTF-8?q?=E4=BB=BD=E6=8A=A5=E5=91=8A=E7=9A=84=E7=9C=9F=E5=AE=9E=E8=AE=A1?= =?UTF-8?q?=E6=95=B0=E3=80=81=E8=B7=91=E5=81=9C=E8=BF=90=E8=A1=8C=E7=9A=84?= =?UTF-8?q?=E5=AE=A1=E7=A8=BF=E7=85=A7=E6=A0=B7=E8=AE=A4=E5=BE=97=EF=BC=8C?= =?UTF-8?q?=E5=8F=8C=E7=9B=B2=E5=A4=8D=E7=AE=97=E7=9A=84=E8=B7=B3=E8=BF=87?= =?UTF-8?q?=E5=8E=9F=E5=9B=A0=E6=8C=89=E6=8C=87=E6=A0=87=E8=AF=B4=EF=BC=8C?= =?UTF-8?q?=E9=81=AE=E7=BD=A9=E6=A0=B8=E5=AF=B9=E4=B8=8D=E4=BA=86=E7=9A=84?= =?UTF-8?q?=E5=AE=81=E5=8F=AF=E5=A4=9A=E9=81=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 反方审稿：unit.challenge 带上节点产出里整份报告的计数（counts）和没能反查的句数（unreviewed，产出里有才带）。证据面板借审批卡的 展示画这一句，以前凭这一句拼出「被推翻 1 句、其余 0 句」，读着像整份报告的结果。节点设为「报错并停止运行」、句子被推翻时节点报错、 没有 node.finished，以前证据接口认不出这次审稿；现在按审完那一刻的 challenge_done 日志认（只认审稿记录存上了的，处置写 fail）， 运行因此失败了也看得到被推翻的是哪几句。 双盲复算：节点以 no_inputs 直通时，那句说的是整张卡。以前原样套到卡上每个指标，配了「复算的指标」子集时，明明取自查询、只是没被 选上的指标也写成「口径卡上没有来自查询的输入」。现在按这个指标自己说：不在子集里、值不是数的是 not_targeted，自己没有来自查询的 输入的才是 no_inputs，关闭了的照旧 disabled；节点跑了、没复算的指标同样这样分。 遮罩：认哪个输入取自哪份查询结果的哪几列抽成 recompute.input_origin，证据面板、推演（之后还有可验证报告）同一个认法。核对不了的 宁可多遮，和推演、可验证报告的原则一致：cell()、Agent 字段落在的快照读不出来；按位置取的一格（以 .rows[行][列] 结尾）取值链上的 产出读不出来；那份结果没存快照、只带着数据源名。证据面板不给值，mask_unverified 写原因、不进遮罩列；推演按读不出来处理，不给值、 不推演。既没带快照 id 也没带数据源名的结果（数据整形改过形状）认不出出处，照常显示，写进交付说明。 recompute.recompute 给双盲复算的预检和盲值用，整个指标重算不了时原因不再说「推演」（_Plan.stuck，判断同 blocked，只换措辞）。 按位置的遮罩接进 recompute 之后，双盲复算的预检对「cell() 输入 + 按位置取自遮罩列的一格」的指标判无法复算（以前照常盲算，遮着的 号码还会作为字面量进盲算模型的计算式），运行对比里这种输入两边都不给值：各补一个用例钉住。 一个指标里同一件工件只读一次：按位置取了好几格时，写变量的节点那份可能几 MB 的产出以前每格重读一遍。 补测：遮罩列上 Agent 字段、cell() 对不上时 model_value / snapshot_value 不在答复里；代入式重新代不出来时整条不给。 test_recompute 的夹具补上数据整形节点 parse 的产出：以前缺着，按位置取的那一格现在会因此核对不了。 去掉实现后新用例各自变红（计数、跑停的审稿、子集的跳过原因、两处宁可多遮、推演、预检措辞、按位置遮罩、只读一次）。 --- backend/app/api/evidence.py | 200 ++++++++++++++---- backend/app/engine/recompute.py | 105 +++++++-- backend/tests/test_evidence_metric_masks.py | 183 ++++++++++++++++ backend/tests/test_evidence_verify_status.py | 106 ++++++++-- .../tests/test_positional_masks_elsewhere.py | 113 ++++++++++ backend/tests/test_recompute.py | 6 +- 6 files changed, 641 insertions(+), 72 deletions(-) create mode 100644 backend/tests/test_positional_masks_elsewhere.py diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 44cd639d..1ae739d8 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -865,7 +865,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s "caliber_upgrade": _caliber_upgrade(sealed.upgrades.get(str(entry.get("node_id") or ""))), } # 期 4：这个指标做没做过双盲复算、结果如何。没有双盲复算复核过这份指标集时没有这个键，老运行一字不差 - if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id): + if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id, metric): step["dual_check"] = checked catalog = doc.get("catalog") or {} inputs: list[dict[str, Any]] = [] @@ -881,8 +881,8 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s if cell := _input_cell(item, catalog): one["cell"] = cell wanted.setdefault(str(artifact_of), []).extend(_located(item.get("locator") or {})) - if columns := await hide.columns(item): - one = _withheld(one, columns) + if (hidden := await hide.check(item)) is not None: + one = _withheld(one, *hidden) withheld.add(str(item.get("path"))) inputs.append(one) if withheld: @@ -892,23 +892,38 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s return [step, *inputs, *queries] -def _withheld(step: dict[str, Any], columns: list[str]) -> dict[str, Any]: - """取自遮罩列的输入：值写「已遮罩」（本来就没有值的照旧是空的，缺输入还得看得出来），模型报的值、快照里的值一并 - 拿掉，masked 写明是哪几列。""" +def _withheld(step: dict[str, Any], columns: list[str], unverified: str | None) -> dict[str, Any]: + """不给值的输入：值写「已遮罩」（本来就没有值的照旧是空的，缺输入还得看得出来），模型报的值、快照里的值一并拿掉。 + 取自遮罩列的，masked 写明是哪几列；遮没遮核对不了的（_InputMasks），mask_unverified 写原因、不进 masked—— + 面板底下那句「这些列设置了遮罩」不能把它们也算进去。""" out = {k: v for k, v in step.items() if k not in ("model_value", "snapshot_value")} if out.get("value") is not None: out["value"] = MASKED - out["masked"] = columns + if unverified is None: + out["masked"] = columns + else: + out["mask_unverified"] = unverified return out -class _InputMasks: - """一个指标的输入里哪些取自遮罩列。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询 - 按它合并的那几个数据源算；快照读不出来时只能按数据源现在设的算。 +#: tool_cell、Agent 字段落在的查询快照读不出来：只知道数据源现在设的遮罩，查询当时记下的在快照里，遮没遮核对不了 +SNAPSHOT_UNVERIFIED = "查询快照无法读取，核对不了这一格是否取自设置了遮罩的列" - cell() 的格、Agent 交来的字段按定位里的列认。vars.res.rows[0][2] 这种按位置取自查询结果的标量卡上没记定位：顺着 - 取值链找到那份结果（recompute.positional_cell，推演用同一个认法），结果里带着快照的 id 和数据源。取值链从写变量的 - 节点在口径卡算这一次之前的产出起走（_card_outputs）。认不出来的不遮——卡上本来就只记着路径和值。 + +class _InputMasks: + """一个指标的输入里哪些不能给值。遮的列和查询步骤同一套（_query_step）：查询当时记下的加数据源现在设的，合并查询按它 + 合并的那几个数据源算。 + + 取自哪份查询结果的哪几列按 recompute.input_origin 认，推演、可验证报告是同一个认法：cell() 的格、Agent 交来的字段按 + 定位里的列认；vars.res.rows[0][2] 这种按位置取的一格卡上没记定位，顺着取值链认，从写变量的节点在口径卡算这一次之前 + 的产出起走（_card_outputs）。 + + 遮没遮核对不了的宁可多遮，和推演一致：快照读不出来（数据源现在没遮这一列，可查询当时记下的遮罩在快照里）、按位置取的 + 一格取值链上的产出读不出来、那份结果没存快照只带着数据源名。认得出、可结果既没带快照的 id 也没带数据源名的（数据整形 + 改过形状）不遮：认不出出处，卡上本来就只记着路径和值。 + + 一个指标里同一件工件只读一次（_fetch 记在 _read）：按位置取了十几格的指标，不必把写变量的节点那几 MB 的产出重读 + 十几遍。查询步骤（_query_step）另读一遍快照：它的接口好几条链都在用，读两遍比读十几遍是小事。 """ def __init__(self, sealed: _Sealed, masks: _Masks, card_node: str, artifact: str) -> None: @@ -916,34 +931,45 @@ class _InputMasks: self._card, self._artifact = card_node, artifact self._outputs: dict[str, str] | None = None self._memo: dict[tuple[str | None, str | None], set[str]] = {} + self._read: dict[str, tuple[Any, bool | None]] = {} - async def columns(self, item: dict[str, Any]) -> list[str]: - """这个输入用到的遮罩列（列名照原样，按出现的先后）；没用到的交回空列表。""" - if item.get("via") in ("tool_cell", "agent_field") and item.get("artifact"): - artifact, source = str(item["artifact"]), None - used = [c for _, c in _located(item.get("locator") or {}) if isinstance(c, str)] - else: - found = recompute.positional_cell(str(item.get("path") or ""), item.get("node_id"), self._output) - if found is None: - return [] - artifact, source, used = found["artifact"], found["source"], [found["column"]] + def _fetch(self, artifact: str) -> tuple[Any, bool | None]: + if artifact not in self._read: + self._read[artifact] = _fetch(artifact) + return self._read[artifact] + + async def check(self, item: dict[str, Any]) -> tuple[list[str], str | None] | None: + """None：给值。(遮罩列, None)：取自这几列遮罩列（列名照原样，按出现的先后）。(用到的列, 原因)：遮没遮核对不了， + 原因给人看（认不出是哪一列时列为空）。""" + origin = recompute.input_origin(item, self._output) + if origin is None: + return None + if origin.unverified is not None: + return list(origin.columns), origin.unverified + used = list(dict.fromkeys(origin.columns)) if not used: - return [] - hidden = await self._hidden(artifact, source) - return list(dict.fromkeys(c for c in used if c.lower() in hidden)) + return None + snap, ok = self._fetch(origin.artifact) if origin.artifact else (None, None) + readable = ok is True and isinstance(snap, dict) + hidden = await self._hidden(origin.artifact, origin.source, snap if readable else None) + if masked := [c for c in used if c.lower() in hidden]: + return masked, None + if origin.artifact and not readable: + return used, SNAPSHOT_UNVERIFIED + if not origin.artifact and origin.source: + return used, recompute.POSITION_NO_SNAPSHOT + return None def _output(self, node_id: Any) -> Any: if self._outputs is None: self._outputs = _card_outputs(self._sealed, self._card, self._artifact) artifact = self._outputs.get(str(node_id or "")) - content, ok = _fetch(artifact) if artifact else (None, None) + content, ok = self._fetch(artifact) if artifact else (None, None) return content if ok is True else None - async def _hidden(self, artifact: str | None, source: str | None) -> set[str]: + async def _hidden(self, artifact: str | None, source: str | None, snap: dict[str, Any] | None) -> set[str]: key = (artifact, source) if key not in self._memo: - snap, ok = _fetch(artifact) if artifact else (None, None) - snap = snap if ok is True and isinstance(snap, dict) else None if snap is not None and snap.get("source") == MERGE_SOURCE: hidden, _ = await _merge_masks(snap, self._masks) else: @@ -1013,8 +1039,10 @@ def _spec_of(sealed: _Sealed) -> GraphSpec | None: return None -def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: Any) -> dict[str, Any] | None: +def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: Any, + metric: dict[str, Any] | None = None) -> dict[str, Any] | None: """这个指标的双盲复算状态：{status, node_id, choice?, blind_rendered?, reason?}。没有双盲复算复核过这份指标集的是 None。 + metric 是卡上的这个指标（认它为什么没被复算用），读不到卡时为 None。 认的是复核了口径卡的哪一次结果：台账条目（kind dual_check）记着它复算的指标集工件，和指标步骤的工件对上才算——口径卡 在循环里算了几次时，报告引用的那一次才是。直通了的（设置里关了、口径卡上没有来自查询的输入）没有台账条目：按画布上 @@ -1022,8 +1050,9 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An - agreed / disagreed / unchecked：节点产出里这个指标的结果。不一致的带审核人的选择（a / b / neither，没选的不带）； 一致、不一致的带盲算值（渲染好的）；无法复算的带原因 - - skipped：节点直通，reason 同节点产出（disabled / no_inputs）；或者节点跑了、这个指标不在复算范围里（没选它、值不是 - 数、没有来自查询的输入），reason 为 not_targeted + - skipped：节点没复算这个指标，reason 说为什么（_not_checked）：关闭了（disabled，节点直通）；不在节点配置的 + 「复算的指标」里、或者值不是数（not_targeted）；这个指标没有一个输入来自查询快照（no_inputs）。节点直通时产出里的 + no_inputs 说的是整张卡，不原样套到每个指标上：卡上确有来自查询的指标、只是没被选上的，写它的说法是 not_targeted 产出读不出来（文件不在、与哈希不一致）时按台账条目说状态和选择，不带盲算值和原因。 """ if not card_node or not artifact or metric_id is None: @@ -1032,6 +1061,7 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An if not checkers: return None spec: GraphSpec | None = None + spec_read = False #: 口径卡节点 → 到这一刻为止它最后一次的指标集 latest: dict[str, str] = {} found: dict[str, Any] | None = None @@ -1047,25 +1077,54 @@ def _dual_check_of(sealed: _Sealed, card_node: str, artifact: str, metric_id: An output, ok = _fetch(str(data.get("artifact") or "")) output = output if ok is True and isinstance(output, dict) else {} entry = next((e for e in ledger if e.get("kind") == "dual_check"), None) + if not spec_read: + spec, spec_read = _spec_of(sealed), True + node = spec.node_map().get(node_id) if spec else None if entry is not None: card, checked = str(entry.get("card") or ""), str(entry.get("metric_set") or "") elif output.get("status") == "skipped": - spec = spec or _spec_of(sealed) - node = spec.node_map().get(node_id) if spec else None card = (dual_check_card(spec, node)[0] or "") if spec and node else "" checked = latest.get(card, "") else: continue if card != card_node or checked != artifact: continue - found = _dual_status(node_id, output, entry, str(metric_id)) + found = _dual_status(node_id, output, entry, str(metric_id), metric, _picked(node)) return found -def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | None, metric_id: str) -> dict[str, Any]: +def _picked(node: Any) -> list[str] | None: + """画布上这个双盲复算节点配的「复算的指标」，认法同 dual_check._options：没配、配了空列表都是 None（复算全部）。""" + wanted = node.config.get("metrics") if node is not None else None + picked = [str(m) for m in wanted if isinstance(m, str) and m.strip()] if isinstance(wanted, list) else [] + return picked or None + + +def _query_input(item: Any) -> bool: + """来自查询快照的输入（同 dual_check._from_query）：双盲复算只复算有这种输入的指标。""" + return isinstance(item, dict) and item.get("via") in ("tool_cell", "agent_field") \ + and isinstance(item.get("artifact"), str) and bool(item["artifact"]) + + +def _not_checked(node_reason: str, metric: dict[str, Any] | None, picked: list[str] | None, metric_id: str) -> str: + """节点没复算这个指标的原因。关闭了（disabled）照节点的说；别的按这个指标自己说：不在配置的「复算的指标」里是 + not_targeted；卡上读得到它、它没有一个输入来自查询快照是 no_inputs，有（那就是值不是数）是 not_targeted。读不到卡时 + 照节点直通的原因说，节点跑了的说 not_targeted。""" + if node_reason == "disabled": + return node_reason + if picked is not None and metric_id not in picked: + return "not_targeted" + if metric is not None: + return "not_targeted" if any(_query_input(i) for i in metric.get("inputs") or []) else "no_inputs" + return node_reason or "not_targeted" + + +def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | None, metric_id: str, + metric: dict[str, Any] | None, picked: list[str] | None) -> dict[str, Any]: """一次双盲复算里这个指标的状态（_dual_check_of 的答复）。""" if output.get("status") == "skipped": - return {"status": "skipped", "node_id": node_id, "reason": str(output.get("reason") or "")} + reason = _not_checked(str(output.get("reason") or ""), metric, picked, metric_id) + return {"status": "skipped", "node_id": node_id, "reason": reason} row = next((m for m in output.get("metrics") or [] if isinstance(m, dict) and str(m.get("metric_id")) == metric_id), None) if row is None and not output and entry is not None: @@ -1075,7 +1134,7 @@ def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | N row = {"status": brief.get("status"), **({"choice": brief["choice"]} if "choice" in brief else {})} \ if brief is not None else None if row is None: - return {"status": "skipped", "node_id": node_id, "reason": "not_targeted"} + return {"status": "skipped", "node_id": node_id, "reason": _not_checked("", metric, picked, metric_id)} status = row.get("status") if row.get("status") in _DUAL_RESULTS else "unchecked" out: dict[str, Any] = {"status": status, "node_id": node_id} if status == "disagreed" and row.get("choice") in ("a", "b", "neither"): @@ -1088,14 +1147,26 @@ def _dual_status(node_id: str, output: dict[str, Any], entry: dict[str, Any] | N return out +#: 反方审稿的四种判断（engine/nodes/challenge.py 的 VERDICTS）：审稿概况的计数照这个顺序 +_CHALLENGE_VERDICTS = ("stands", "weakened", "refuted", "inconclusive") + + def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, Any] | None: - """这一句的反方审稿判断：{verdict, node_id, artifact, on_refuted, report_label}。没审到这一句的是 None。 + """这一句的反方审稿判断：{verdict, node_id, artifact, on_refuted, report_label, counts, unreviewed?}。没审到这一句的 + 是 None。 只认审的就是正文这一份报告文档的：审稿记录（challenge_artifact）里记着它审的报告节点和文档工件，和这份对上才算—— 报告在审稿之后又重写了（循环、接着跑），同一个句子编号在新的一稿里说的可能是另一句话。记录没存上（产出里的 challenge_artifact 为空）或读不出来时，按画布上它审的报告、和它跑完之前那份报告最后一次自查记下的文档认，判断取 产出里列的那几句（最多 10 句，被推翻、被削弱的排在前面），artifact 照产出写（没存上的为空）。几个反方审稿审了同一 - 份文档时取最后跑完、审到了这一句的那个。on_refuted 是实际生效的处置（degrade / flag），report_label 是报告节点的名字。 + 份文档时取最后跑完、审到了这一句的那个。on_refuted 是实际生效的处置，report_label 是报告节点的名字。 + + counts 是这次审稿四种判断各几句——整份报告的，取节点产出里的那一份；unreviewed 是没能反查的句数，产出里有才带。 + 证据面板借审批卡上的展示画这一句，那一块的概况得是这份报告真实的数，不能凭这一句拼出一个「被推翻 0 句」。 + + 节点设为「报错并停止运行」、又有句子被推翻时，节点报错、没有 node.finished：审完那一刻的 challenge_done 日志记着 + 审稿记录和计数，按它认（只认审稿记录存上了的），on_refuted 为 fail。运行因此失败了，被推翻的是哪几句、反方凭什么， + 证据面板里照样看得到。 """ reviewers = set(sealed.nodes_of("challenge")) if not reviewers or not unit_id: @@ -1103,6 +1174,8 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A spec: GraphSpec | None = None #: 报告节点 → 到这一刻为止它最后一次自查记下的文档 drafts: dict[str, str] = {} + #: 反方审稿节点 → 这一次执行里审完时的日志：节点报错时它就是审稿结果 + done: dict[str, dict[str, Any]] = {} found: dict[str, Any] | None = None for _, etype, node_id, data in sealed.events: if not node_id or not isinstance(data, dict): @@ -1110,17 +1183,34 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A if etype == "report.checked" and data.get("doc_artifact"): drafts[node_id] = str(data["doc_artifact"]) continue - if etype != "node.finished" or node_id not in reviewers: + if node_id not in reviewers: + continue + if etype == "node.started": + done.pop(node_id, None) + continue + if etype == "log" and data.get("code") == "challenge_done": + done[node_id] = data + continue + if etype == "node.finished": + done.pop(node_id, None) + output, ok = _fetch(str(data.get("artifact") or "")) + if ok is not True or not isinstance(output, dict) or not isinstance(output.get("counts"), dict): + continue # 直通了（关闭、没有可审的句子）或者产出读不出来 + elif etype == "node.failed" and node_id in done: + log = done.pop(node_id) + if not isinstance(log.get("artifact"), str) or not log["artifact"]: + continue + output = {"challenge_artifact": log["artifact"], "counts": log.get("counts"), "on_refuted": "fail"} + else: continue - output, ok = _fetch(str(data.get("artifact") or "")) - if ok is not True or not isinstance(output, dict) or not isinstance(output.get("counts"), dict): - continue # 直通了（关闭、没有可审的句子）或者产出读不出来 artifact = output.get("challenge_artifact") if isinstance(output.get("challenge_artifact"), str) else None record, saved = _fetch(artifact) if artifact else (None, None) if saved is True and isinstance(record, dict): if record.get("report") != report.node_id or record.get("doc_artifact") != report.doc_artifact: continue claims = record.get("claims") + elif etype == "node.failed": + continue # 节点报错时没有产出可退：认不出审的是哪一稿 else: spec = spec or _spec_of(sealed) node = spec.node_map().get(node_id) if spec else None @@ -1128,17 +1218,33 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A if target != report.node_id or drafts.get(target) != report.doc_artifact: continue claims = output.get("items") - verdict = next((c.get("verdict") for c in claims or [] if isinstance(c, dict) and c.get("unit") == unit_id), None) + claims = [c for c in claims or [] if isinstance(c, dict)] + verdict = next((c.get("verdict") for c in claims if c.get("unit") == unit_id), None) if not isinstance(verdict, str) or not verdict: continue spec = spec or _spec_of(sealed) writer = spec.node_map().get(report.node_id) if spec else None + unreviewed = output.get("unreviewed") found = {"verdict": verdict, "node_id": node_id, "artifact": artifact, "on_refuted": str(output.get("on_refuted") or "degrade"), - "report_label": writer.title if writer else report.node_id} + "report_label": writer.title if writer else report.node_id, + "counts": _challenge_counts(output.get("counts"), claims), + **({"unreviewed": unreviewed} if isinstance(unreviewed, int) and not isinstance(unreviewed, bool) + and unreviewed >= 0 else {})} return found +def _challenge_counts(raw: Any, claims: list[dict[str, Any]]) -> dict[str, int]: + """四种判断各几句：取节点产出（或审完时的日志）里的计数；一个数都认不出来时按认得的句子数（审稿记录里的是全部， + 记录没存上时只有产出里列的那几句）。""" + given = raw if isinstance(raw, dict) else {} + known = {v: given[v] for v in _CHALLENGE_VERDICTS + if isinstance(given.get(v), int) and not isinstance(given[v], bool) and given[v] >= 0} + if known: + return {v: known.get(v, 0) for v in _CHALLENGE_VERDICTS} + return {v: sum(1 for c in claims if c.get("verdict") == v) for v in _CHALLENGE_VERDICTS} + + class _Masks: """一次请求里查过的数据源遮罩：数据源名 → (要遮的列（小写）, 数据源现在还在不在)。""" diff --git a/backend/app/engine/recompute.py b/backend/app/engine/recompute.py index f8590d4d..fa3b86d9 100644 --- a/backend/app/engine/recompute.py +++ b/backend/app/engine/recompute.py @@ -61,6 +61,13 @@ NOT_RESTORABLE = "这类输入当时没有保存完整值" MASKED_INPUT = "这一列在数据源中设置了遮罩，不能用于推演" MASKED_TABLE = "这份查询结果里有设置了遮罩的列，不能用于推演" MASKED_METRIC = "这个指标用到了设置了遮罩的列，不能推演" +#: 按位置取自查询结果的一格（vars.res.rows[0][2]）遮没遮核对不了：取值链上节点的产出读不出来，或者那份结果没存快照、 +#: 只带着数据源名（查询当时记下的遮罩在快照里）。和读不出来的快照一样宁可多遮（_positional） +POSITION_UNREADABLE = "查询结果无法读取，核对不了这一格是否取自设置了遮罩的列" +POSITION_NO_SNAPSHOT = "查询结果没有保存快照，核对不了这一格是否取自设置了遮罩的列" +#: recompute（双盲复算的预检、盲值）里整个指标重算不了的说法：同 _Plan.blocked，只是不说「推演」——那里不是推演 +NO_LINEAGE_PLAIN = "这个指标生成于系统升级前，口径卡没有记录输入" +MASKED_PLAIN = "这个指标用到了设置了遮罩的列" SELECTOR = "这个输入决定取哪一行、哪一项，修改它需要重新取数，不能推演" EXCLUDED = "排除后无法计算" UNREADABLE = "查询快照无法读取" @@ -170,6 +177,8 @@ class _Plan: tops: list[str] = field(default_factory=list) inputs: dict[str, _Input] = field(default_factory=dict) problem: str | None = None + #: problem 不说「推演」的说法（recompute 给双盲复算用）；和 problem 一样时不填 + plain: str | None = None #: 嵌在别的引用里的输入 → 包着它的最外层引用（_refs） within: dict[str, set[str]] = field(default_factory=dict) @@ -200,6 +209,20 @@ class _Plan: return "unsupported", f"无法推演：{path}（{problem}）" return None + @property + def stuck(self) -> str | None: + """blocked 的原因换成不说「推演」的说法：recompute 交给双盲复算的预检、盲值，那里说「不能推演」驴唇不对马嘴。 + 判断和先后与 blocked 一字不差，只换措辞；推演得了的是 None。""" + blocked = self.blocked + if blocked is None: + return None + if self.problem: + return self.plain or self.problem + if blocked[0] == "masked": + return MASKED_PLAIN + path = next(p for p in self.tops if self.inputs[p].problem) + return f"输入 {path} 无法还原（{self.inputs[path].problem}）" + # -------------------------------------------------------------------------- # 入口 @@ -232,8 +255,14 @@ def whatif(card: dict[str, Any], changes: Changes, sources: Sources, *, metric: def recompute(metric: dict[str, Any], changes: Changes, sources: Sources) -> dict[str, Any]: - """重算一个指标，不管改动碰没碰到它，也不校验改动。核对还原是否忠实时用：什么都不改，重算出来应当和卡上记的一样。""" - return _outcome(_plan(metric, _Store(sources)), changes, sources, force=True)[0] + """重算一个指标，不管改动碰没碰到它，也不校验改动。核对还原是否忠实时用：什么都不改，重算出来应当和卡上记的一样。 + + 调用方是双盲复算（预检、按盲算的输入代回原式），不是推演：整个指标重算不了时原因不说「推演」（_Plan.stuck）。""" + plan = _plan(metric, _Store(sources)) + result = _outcome(plan, changes, sources, force=True)[0] + if (plain := plan.stuck) is not None: + result["reason"] = plain + return result # -------------------------------------------------------------------------- @@ -245,7 +274,7 @@ def _plan(metric: dict[str, Any], store: _Store) -> _Plan: expr = str(metric.get("expression") or "").strip() items = metric.get("inputs") if not expr or not isinstance(items, list): - return _Plan(metric, problem=NO_LINEAGE) + return _Plan(metric, problem=NO_LINEAGE, plain=NO_LINEAGE_PLAIN) try: tree, _, _ = parse_expression(expr) except ExpressionError as e: @@ -258,7 +287,8 @@ def _plan(metric: dict[str, Any], store: _Store) -> _Plan: inputs[item["path"]] = _restore(item, remaining, set(within), store) plan = _Plan(metric, tree, tops, inputs, within=within) if lost := [p for p in tops if p not in inputs]: - plan.problem = f"口径卡缺少输入 {'、'.join(lost[:3])} 的记录，不能推演" + plan.plain = f"口径卡缺少输入 {'、'.join(lost[:3])} 的记录" + plan.problem = f"{plan.plain}，不能推演" return plan @@ -335,7 +365,9 @@ def _restore(item: dict[str, Any], remaining: set[str], nested: set[str], store: # 也不拿来推演。数据源现在设的遮罩照样认（_masked），认出来了就按遮罩说 inp.unreadable = True inp.problem = inp.problem or (TAMPERED if trouble == "tampered" else UNREADABLE) - inp.masked = _masked(inp, locator, store) or _positional_masked(inp, store) + inp.masked = _masked(inp, locator, store) + if not inp.masked: + _positional(inp, store) return inp @@ -417,14 +449,56 @@ def _masked(inp: _Input, locator: dict[str, Any], store: _Store) -> bool: return any(isinstance(c, str) and c.lower() in hidden for c in columns) -def _positional_masked(inp: _Input, store: _Store) -> bool: - """按位置取自查询结果的一格（vars.res.rows[0][2]）落在遮罩列上：卡上没记定位，顺着取值链找到那份结果、认出是哪一列。 - 认不出来的（结果里没带快照的 id、取值链走不通）不算：和以前一样照常推演。""" +def _positional(inp: _Input, store: _Store) -> None: + """按位置取自查询结果的一格（vars.res.rows[0][2]）：卡上没记定位，顺着取值链找到那份结果、认出是哪一列（input_origin）。 + + 落在遮罩列上的按遮罩（masked）。遮没遮核对不了的——取值链上节点的产出读不出来，或者那份结果没存快照、只带着数据源名 + ——和读不出来的快照同一个处理：值不给、也不拿来推演（unreadable，原因记在 problem），宁可多遮。认得出、可那份结果 + 既没带快照的 id 也没带数据源名的（数据整形改过形状的结果）认不出出处，照常推演，见模块说明。""" if inp.snapshot is not None or inp.shape != "scalar" or inp.item.get("via") in ("tool_cell", "agent_field"): - return False - found = positional_cell(inp.path, inp.item.get("node_id"), lambda node_id: store.output(node_id)[0]) - hidden = store.sources.masked.get(found["artifact"]) if found and found["artifact"] else None - return bool(hidden) and isinstance(found["column"], str) and found["column"].lower() in hidden + return + origin = input_origin(inp.item, lambda node_id: store.output(node_id)[0]) + if origin is None: + return + if origin.unverified is None and origin.artifact: + hidden = store.sources.masked.get(origin.artifact) + inp.masked = bool(hidden) and any(c.lower() in hidden for c in origin.columns) + return + if origin.unverified is not None or origin.source: + inp.unreadable = True + inp.problem = inp.problem or origin.unverified or POSITION_NO_SNAPSHOT + + +@dataclass(frozen=True) +class Origin: + """一个输入取自哪份查询结果的哪几列（input_origin）。 + + - artifact、source：那份查询结果的快照 id 和数据源名，结果里没带的是 None + - columns：用到的列（列名照原样）。按位置取的一格、取值链上的产出读不出来时认不出是哪一列，为空 + - unverified：遮没遮核对不了的原因（取值链上节点的产出读不出来）；核对得了的是 None + """ + + artifact: str | None + source: str | None + columns: tuple[str, ...] + unverified: str | None = None + + +def input_origin(item: dict[str, Any], output: Callable[[Any], Any]) -> Origin | None: + """卡上的一个输入取自哪份查询结果的哪几列，遮罩按列认时用（推演、证据面板的指标步骤、可验证报告是同一个认法）。 + + cell() 的格、Agent 交来的字段按定位里的列认；vars.res.rows[0][2] 这种按位置取的一格顺着取值链认（positional_cell， + output 同它）。不是取自查询结果的格（常量、运行输入、Agent 没按出处交的字段、加工过的数）是 None。""" + if item.get("via") in ("tool_cell", "agent_field") and isinstance(item.get("artifact"), str) and item["artifact"]: + locator = item.get("locator") if isinstance(item.get("locator"), dict) else {} + columns = [locator.get("column")] if locator.get("column") else list((locator.get("columns") or {}).values()) + return Origin(item["artifact"], None, tuple(c for c in columns if isinstance(c, str) and c)) + found = positional_cell(str(item.get("path") or ""), item.get("node_id"), output) + if found is None: + return None + if found.get("unreadable"): + return Origin(None, None, (), POSITION_UNREADABLE) + return Origin(found["artifact"], found["source"], (found["column"],)) #: 取值链没走通 @@ -441,6 +515,10 @@ def positional_cell(path: str, node_id: Any, output: Callable[[Any], Any]) -> di output(节点 id) 交回那个节点当时的产出（读不出来给 None）。nodes.X 的根就是 X 的产出；vars.X 是写它的节点（输入记着 的 node_id）交给变量的那个值——数据整形、合并查询交的是产出本身，Agent、LLM、代码节点交的是产出里的 data，工具节点 交的是 JSON 文本，都试一遍，取值链走得通的那个就是。下标得是字面量：rows[vars.i] 要另取一个值才知道是哪一行，不认。 + + 路径就是按位置取一格的样子（以 .rows[行号][列] 结尾），可那个节点的产出读不出来（文件不在、与哈希不一致、不知道是 + 谁写的）时返回 {"unreadable": True}：这一格取自哪一列核对不了，调用方宁可多遮。卡上记着这个输入的值，算的时候 + 取值链是走得通的，现在读不出来只能是产出丢了或被改了。 """ try: tree = ast.parse(path, mode="eval").body @@ -461,6 +539,9 @@ def positional_cell(path: str, node_id: Any, output: Callable[[Any], Any]) -> di or not isinstance(keys[0], str): return None base = output(keys[0] if tree.id == "nodes" else node_id) + if base is None: + positional = keys[-3] == "rows" and isinstance(keys[-2], int) and isinstance(keys[-1], (int, str)) + return {"unreadable": True} if positional else None candidates = [base] if isinstance(base, dict) and "data" in base: candidates.append(base["data"]) diff --git a/backend/tests/test_evidence_metric_masks.py b/backend/tests/test_evidence_metric_masks.py index 586c9fac..f3e72094 100644 --- a/backend/tests/test_evidence_metric_masks.py +++ b/backend/tests/test_evidence_metric_masks.py @@ -231,3 +231,186 @@ async def test_the_whatif_withholds_a_masked_scalar_taken_by_position(client, mo res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "card", "metric": "plen2", "overrides": {"inputs": {FIRST_PHONE: 1}}}) assert res.status_code == 422 and "遮罩" in res.json()["detail"] + + +# -------------------------------------------------------------------------- +# 手搭的封存范围：模型报的值、快照里的值，以及遮没遮核对不了的几种 +# -------------------------------------------------------------------------- + +#: 一份带遮罩的快照：phone 在查询当时就设了遮罩（记在快照里）。数据源名是编的，库里没有这个数据源（现在的遮罩为空） +GHOST = "ghost_mask_src" +PHONE_SNAP = {"columns": ["id", "phone"], "rows": [[1, "139-0000-0009"]], "source": GHOST, + "sql": "SELECT id, phone FROM members", "mask_columns": ["phone"]} +#: 这几条用例里出现过的号码，答复里一个都不能有 +SECRETS = ["139-0000-0009", "139-0000-0008", "139-0000-0007", "139-0000-0006", "139-0000-0005", "139-0000-0004"] +CELL = "cell(nodes.fetch, 0, 'phone')" + + +def card_metric(mid, expression, substituted, inputs, value=13): + return {"id": mid, "name": mid, "unit": "位", "value": value, "rendered": f"{value}位", "format": "thousands", + "expression": expression, "substituted": substituted, "recompute_ok": True, "inputs": inputs, + "status": "ok"} + + +async def hand_chain(metrics, *, queries=(), outputs=None, metric_id=None): + """手搭一次封存了的运行（取数 → 数据整形 → 口径卡），直接走指标步骤（evidence._metric_chain）。queries 是封存范围内 + 交回过的查询快照，outputs 是口径卡算之前各节点的产出工件。""" + from app.api import evidence + from app.core.artifact_store import put_json + from app.engine.evidence import make_eid + + card = await put_json({"kind": "metric_set", "caliber": "遮罩口径", "caliber_version": "v1", "metrics": metrics}, + kind="metric_set") + events = [(i + 1, "node.finished", nid, {"artifact": art}) for i, (nid, art) in enumerate((outputs or {}).items())] + events += [(len(events) + 1, "node.started", "card", {}), + (len(events) + 2, "node.finished", "card", {"evidence": [ + {"kind": "metric_set", "node_id": "card", "artifact": card}]})] + sealed = evidence._Sealed(run_id="syn-mask", graph=graph(), events=events, + seal={"sealed": True, "ok": True, "manifest_seq": len(events), "legacy": False}) + sealed.queries = {a: {"node_id": "fetch", "tool": f"db_query__{GHOST}", "source": GHOST} for a in queries} + sealed.ledger = {("card", card)} + mid = metric_id or metrics[0]["id"] + eid = make_eid("metric", card, {"metric": mid}) + cite = {"alias": f"m:{mid}", "kind": "metric", "locator": {"metric": mid}, "status": "resolved", "eid": eid} + entry = {"alias": f"m:{mid}", "kind": "metric", "artifact": card, "node_id": "card", "locator": {"metric": mid}, + "eid": eid} + seg = {"id": "s1", "kind": "number", "text": "13位", "state": "deterministic", "cite": cite} + return await evidence._metric_chain(seg, cite, entry, {"catalog": {f"m:{mid}": entry}}, sealed, evidence._Masks()) + + +def said(chain) -> list[str]: + raw = json.dumps(chain, ensure_ascii=False) + return [p for p in SECRETS if p in raw] + + +async def test_what_the_model_reported_and_what_the_snapshot_held_are_withheld_too(): + """对不上的输入另记着两边的值：Agent 字段的 model_value（模型报的）、cell() 的 snapshot_value（快照里的）。取自遮罩列 + 时一并拿掉——界面上不显示它们，接口答复里同样不能带着。""" + from app.core.artifact_store import put_json + + snap = await put_json(PHONE_SNAP, kind="query") + cell = card_metric("plen", f"len({CELL})", "len('139-0000-0008')", [ + {"path": CELL, "value": "139-0000-0008", "node_id": "fetch", "via": "tool_cell", "artifact": snap, + "locator": {"row": 0, "column": "phone"}, "status": "mismatch", "snapshot_value": "139-0000-0009"}]) + agent = card_metric("alen", "len(vars.kpi.phone)", "len('139-0000-0009')", [ + {"path": "vars.kpi.phone", "value": "139-0000-0009", "node_id": "bot", "via": "agent_field", "field": "phone", + "artifact": snap, "locator": {"row": 0, "column": "phone"}, "status": "mismatch", + "model_value": "139-0000-0007"}]) + for metric in (cell, agent): + chain = await hand_chain([metric], queries=[snap]) + [step] = [s for s in chain if s["step"] == "input"] + assert step["value"] == "已遮罩" and step["masked"] == ["phone"], metric["id"] + assert "model_value" not in step and "snapshot_value" not in step, metric["id"] + assert chain[0]["substituted"] == "len(已遮罩)" and said(chain) == [], (metric["id"], said(chain)) + + +async def test_a_substituted_expression_that_does_not_reproduce_is_not_shown(): + """卡上记的代入式和按卡上的输入重新代出来的对不上（老卡、卡上记的不是原值）：整条代入式不给，宁可不显示，也不能让 + 遮着的值从对不上的那一处漏出来。""" + from app.core.artifact_store import put_json + from app.engine import recompute + + snap = await put_json(PHONE_SNAP, kind="query") + odd = card_metric("odd", f"len({CELL})", "len('139-0000-0006')", [ + {"path": CELL, "value": "139-0000-0005", "node_id": "fetch", "via": "tool_cell", "artifact": snap, + "locator": {"row": 0, "column": "phone"}, "status": "verified"}]) + assert recompute.substituted_without(odd, {CELL}, "已遮罩") is None + chain = await hand_chain([odd], queries=[snap]) + assert chain[0]["substituted"] is None and said(chain) == [], said(chain) + # 对得上的照常换掉那一处 + assert recompute.substituted_without({**odd, "substituted": "len('139-0000-0005')"}, {CELL}, "已遮罩") == "len(已遮罩)" + + +async def test_a_cell_on_a_snapshot_that_cannot_be_read_is_withheld(): + """cell() 落在的快照读不出来（文件不在了）：数据源现在没遮这一列，可查询当时记下的遮罩在快照里，遮没遮核对不了—— + 宁可多遮，和推演一致。说法另写，不说「这一列设置了遮罩」（不进 masked）。""" + gone = "dd" * 32 + metric = card_metric("plen", f"len({CELL})", "len('139-0000-0004')", [ + {"path": CELL, "value": "139-0000-0004", "node_id": "fetch", "via": "tool_cell", "artifact": gone, + "locator": {"row": 0, "column": "phone"}, "status": "verified"}]) + chain = await hand_chain([metric], queries=[gone]) + [step] = [s for s in chain if s["step"] == "input"] + assert step["value"] == "已遮罩" and "masked" not in step + assert step["mask_unverified"] == "查询快照无法读取，核对不了这一格是否取自设置了遮罩的列" + assert chain[0]["substituted"] == "len(已遮罩)" and said(chain) == [] + + +PARSED = {"columns": ["id", "phone"], "rows": [[1, "139-0000-0004"]]} +POSITIONAL = "vars.res.rows[0][1]" + + +@pytest.mark.parametrize("parsed, why", [ + # 写变量的节点的产出读不出来：取值链走不通，那一格取自哪一列核对不了 + (None, "查询结果无法读取，核对不了这一格是否取自设置了遮罩的列"), + # 结果没存快照、只带着数据源名：查询当时记下的遮罩在快照里，核对不了 + ({**PARSED, "source": GHOST}, "查询结果没有保存快照，核对不了这一格是否取自设置了遮罩的列"), +]) +async def test_a_positional_cell_that_cannot_be_traced_is_withheld(parsed, why): + from app.core.artifact_store import put_json + + out = await put_json(parsed, kind="node_output") if parsed is not None else "ee" * 32 + metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [ + {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]", + "status": "ok"}], value=26) + chain = await hand_chain([metric], outputs={"parse": out}) + [step] = [s for s in chain if s["step"] == "input"] + assert step["value"] == "已遮罩" and step["mask_unverified"] == why and "masked" not in step + assert chain[0]["substituted"] == "len(已遮罩) * 2" and said(chain) == [] + + +async def test_a_reshaped_result_without_a_snapshot_or_a_source_is_shown_as_is(): + """数据整形改过形状的结果（既没带快照的 id，也没带数据源名）认不出出处：照常显示。已知的缺口，写在交付说明里。""" + from app.core.artifact_store import put_json + + out = await put_json(PARSED, kind="node_output") + metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [ + {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]", + "status": "ok"}], value=26) + chain = await hand_chain([metric], outputs={"parse": out}) + [step] = [s for s in chain if s["step"] == "input"] + assert step["value"] == "139-0000-0004" and "mask_unverified" not in step and "masked" not in step + + +@pytest.mark.parametrize("parsed, withheld", [ + (None, True), + ({**PARSED, "source": GHOST}, True), + (PARSED, False), +]) +def test_the_whatif_withholds_the_same_untraceable_cells(parsed, withheld): + """推演同一个认法：取值链走不通、结果没存快照的那一格不给值、不推演；认不出出处的照常推演。""" + from app.engine import recompute + + store = {"out": parsed} + metric = card_metric("plen2", f"len({POSITIONAL}) * 2", "len('139-0000-0004') * 2", [ + {"path": POSITIONAL, "value": "139-0000-0004", "node_id": "parse", "via": "transform", "field": "rows[0][1]", + "status": "ok"}], value=26) + sources = recompute.Sources(load=store.get, outputs={"parse": "out"} if parsed is not None else {}) + [result] = recompute.whatif({"metrics": [metric]}, recompute.Changes(), sources)["metrics"] + [inp] = result["inputs"] + if withheld: + assert inp["old"] is None and inp.get("withheld") is True and result["status"] == "unsupported" + assert result["substituted"] is None and "139-0000-0004" not in json.dumps(result, ensure_ascii=False) + else: + assert inp["old"] == "139-0000-0004" and result["status"] == "unchanged" + + +async def test_the_writer_output_is_read_once_for_all_cells_taken_by_position(monkeypatch): + """一个指标按位置取了好几格：写变量的节点的产出（可能是几 MB 的查询结果）一次请求里只读一次，快照也只读一次。""" + from app.api import evidence + from app.core.artifact_store import put_json + + snap = await put_json({**PHONE_SNAP, "rows": [[1, "139-0000-0009"], [2, "139-0000-0008"]]}, kind="query") + out = await put_json({"columns": ["id", "phone"], "rows": [[1, "139-0000-0009"], [2, "139-0000-0008"]], + "artifact": snap, "source": GHOST}, kind="node_output") + paths = ["vars.res.rows[0][1]", "vars.res.rows[1][1]", "vars.res.rows[0][0]"] + metric = card_metric("many", " + ".join(f"len(str({p}))" for p in paths), "len('139-0000-0009') + 0", [ + {"path": p, "value": v, "node_id": "parse", "via": "transform", "field": p[9:], "status": "ok"} + for p, v in zip(paths, ["139-0000-0009", "139-0000-0008", 1])], value=27) + reads: list[str] = [] + real = evidence.load + monkeypatch.setattr(evidence, "load", lambda artifact: (reads.append(artifact), real(artifact))[1]) + chain = await hand_chain([metric], outputs={"parse": out}) + inputs = {s["path"]: s for s in chain if s["step"] == "input"} + assert inputs[paths[0]]["masked"] == ["phone"] and inputs[paths[1]]["masked"] == ["phone"] + assert inputs[paths[2]]["value"] == 1 and "masked" not in inputs[paths[2]] + assert reads.count(out) == 1 and reads.count(snap) == 1, reads diff --git a/backend/tests/test_evidence_verify_status.py b/backend/tests/test_evidence_verify_status.py index e413faa8..280967da 100644 --- a/backend/tests/test_evidence_verify_status.py +++ b/backend/tests/test_evidence_verify_status.py @@ -1,11 +1,15 @@ """证据接口的核查状态（期 4 I2a）：口径卡指标那一步带双盲复算的结果，句子带反方审稿的判断。 - 指标步骤 dual_check：{status, node_id, choice?, blind_rendered?, reason?}。status 是一致（agreed）、不一致（disagreed， - 带审核人的选择，没选的不带）、无法复算（unchecked，带原因）、跳过（skipped：节点直通时 reason 是 disabled / no_inputs， - 节点跑了、这个指标不在复算范围里时是 not_targeted）。读双盲复算节点封存的产出和它的台账条目，按台账里记的指标集工件 - 认它复算的是口径卡的哪一次结果；直通的没有台账条目，按画布上它复算的卡、和它跑完之前那张卡最后一次的结果认 -- 句子 unit.challenge：{verdict, node_id, artifact, on_refuted, report_label}。读反方审稿节点的产出和审稿记录，只认审的 - 就是这一份报告文档的；记录没存上时按画布上审的报告、和它跑完之前那份报告最后一次自查的文档认，判断取产出里列的那几句 + 带审核人的选择，没选的不带）、无法复算（unchecked，带原因）、跳过（skipped）。跳过的原因按这个指标自己说：关闭了 + （disabled）；不在节点配置的「复算的指标」里、或值不是数（not_targeted）；这个指标没有来自查询的输入（no_inputs）—— + 节点直通时产出里的 no_inputs 说的是整张卡，卡上确有来自查询的指标、只是没被选上的不能也说成没有。读双盲复算节点封存的 + 产出和它的台账条目，按台账里记的指标集工件认它复算的是口径卡的哪一次结果；直通的没有台账条目，按画布上它复算的卡、 + 和它跑完之前那张卡最后一次的结果认 +- 句子 unit.challenge：{verdict, node_id, artifact, on_refuted, report_label, counts, unreviewed?}。读反方审稿节点的产出 + 和审稿记录，只认审的就是这一份报告文档的；记录没存上时按画布上审的报告、和它跑完之前那份报告最后一次自查的文档认， + 判断取产出里列的那几句。counts 是整份报告的审稿计数（节点产出里的那一份），证据面板借审批卡的展示画这一句时，概况那 + 一行不能凭这一句拼。节点设为「报错并停止运行」、句子被推翻、运行因此失败时，按审完那一刻的日志认 - 没有双盲复算、反方审稿的运行，答复里没有这两个键（老运行和以前一字不差） 双盲复算、反方审稿各真跑一遍图（虚构的 SQLite 库，模型是 MockChatModel 剧本）；认「哪一次」的边界情形用手写的封存事件。 @@ -161,9 +165,21 @@ async def test_an_agreeing_dual_check_shows_on_the_metric_step(client, monkeypat step = (await segment(client, run.id, "330.0元"))["chain"][0] assert step["step"] == "metric" and step["metric"] == "gmv" assert step["dual_check"] == {"status": "agreed", "node_id": "dual", "blind_rendered": "330.0元"} - # 节点跑了、这个指标不在复算范围里（没有来自查询的输入）：照实说没复算，不是缺省 + # 节点跑了、这个指标没复算（它没有来自查询的输入）：照实说没复算、为什么，不是缺省 days = (await segment(client, run.id, "7天"))["chain"][0] - assert days["dual_check"] == {"status": "skipped", "node_id": "dual", "reason": "not_targeted"} + assert days["dual_check"] == {"status": "skipped", "node_id": "dual", "reason": "no_inputs"} + + +async def test_a_metric_left_out_of_the_subset_is_not_said_to_lack_query_inputs(client, monkeypatch, shop): + """「复算的指标」只选了统计天数：它没有来自查询的输入，节点以 no_inputs 直通。销售额明明取自查询，只是没被选上—— + 它的说法是不在复算范围内，不能套用节点那句「没有来自查询的输入」。""" + Blind(monkeypatch, shop, {}, writer=WRITER) + run = await start(dual_graph(shop, metrics=["days"])) + assert run.status == "succeeded", run.error + gmv = (await segment(client, run.id, "330.0元"))["chain"][0]["dual_check"] + assert gmv == {"status": "skipped", "node_id": "dual", "reason": "not_targeted"} + days = (await segment(client, run.id, "7天"))["chain"][0]["dual_check"] + assert days == {"status": "skipped", "node_id": "dual", "reason": "no_inputs"} @pytest.mark.parametrize("choice", ["b", None]) @@ -291,13 +307,16 @@ async def test_a_challenged_sentence_carries_the_verdict(client, monkeypatch, sa reviewer(monkeypatch, sales) run = await start(challenge_graph(sales), run_class="formal") assert run.status == "succeeded", run.error - record = (await challenge_output(run.id))["challenge_artifact"] + output = await challenge_output(run.id) + record = output["challenge_artifact"] south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"] + # counts 是整份报告的审稿计数（节点产出里的那一份）：一句被推翻、一句被削弱，不是只数这一句 assert south == {"verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "degrade", - "report_label": "写周报"} + "report_label": "写周报", "counts": {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0}} + assert south["counts"] == output["counts"] # 点开句子里的数字，片段接口答的这一句同样带着判断 east = (await segment(client, run.id, "1,500"))["unit"]["challenge"] - assert east["verdict"] == "weakened" and east["artifact"] == record + assert east["verdict"] == "weakened" and east["artifact"] == record and east["counts"] == output["counts"] # 反方没审的（标题不是结论）：没有这个键 heading = await sentence(client, run.id, kind="heading") assert heading["unit"]["kind"] == "heading" and "challenge" not in heading["unit"] @@ -315,7 +334,22 @@ async def test_a_lost_review_record_still_shows_the_listed_verdicts(client, monk assert (await challenge_output(run.id))["challenge_artifact"] is None south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"] assert south == {"verdict": "refuted", "node_id": "chal", "artifact": None, "on_refuted": "degrade", - "report_label": "写周报"} + "report_label": "写周报", "counts": {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0}} + + +async def test_a_review_that_stopped_the_run_still_marks_its_sentences(client, monkeypatch, sales): + """「结论被推翻时」设为报错并停止运行：节点报错、没有 node.finished，运行失败。被推翻的是哪几句、反方凭什么， + 证据面板里照样要看得到——按审完那一刻的日志（challenge_done）认审稿记录，处置写 fail。""" + reviewer(monkeypatch, sales) + graph = challenge_graph(sales) + next(n for n in graph["nodes"] if n["id"] == "chal")["data"]["config"]["on_refuted"] = "fail" + run = await start(graph, run_class="formal") + assert run.status == "failed" and "推翻" in (run.error or ""), run.error + south = (await sentence(client, run.id, SOUTH))["unit"]["challenge"] + assert south["verdict"] == "refuted" and south["on_refuted"] == "fail" and south["node_id"] == "chal" + assert south["artifact"] and south["counts"] == {"stands": 0, "weakened": 1, "refuted": 1, "inconclusive": 0} + east = (await segment(client, run.id, "1,500"))["unit"]["challenge"] + assert east["verdict"] == "weakened" and east["artifact"] == south["artifact"] async def test_reports_without_a_challenge_have_no_verdicts(client, monkeypatch, sales): @@ -417,6 +451,54 @@ async def test_a_review_of_an_earlier_draft_does_not_mark_the_rewritten_report(s ] sealed = sealed_of(graph, events) assert evidence._challenge_of(sealed, report(draft), "u1") == { - "verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "flag", "report_label": "写周报"} + "verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "flag", "report_label": "写周报", + "counts": {"stands": 0, "weakened": 0, "refuted": 1, "inconclusive": 0}} assert evidence._challenge_of(sealed, report(final), "u1") is None assert evidence._challenge_of(sealed, report(draft), "u2") is None + + +async def test_the_counts_are_the_whole_reviews_and_unreviewed_rides_along(): + """counts 照产出（整份报告审了 12 句），没能反查的句数产出里有就带上；产出里的计数认不出来时按审稿记录数。""" + doc = await put_json({"markdown": "周报", "n": 3}, kind="report_doc") + claims = [{"unit": "u1", "text": "华南销售额超过华东。", "verdict": "weakened"}, + {"unit": "u2", "text": "华北持平。", "verdict": "stands"}] + record = await put_json({"report": "write", "doc_artifact": doc, "model": "mock", + "claims": [{**c, "rationale": "", "rationale_doc": None, "queries": []} for c in claims]}, + kind="challenge") + graph = chain(node("write", "report", "写周报"), node("chal", "challenge", "审稿", report="write")) + + async def answer(output: dict) -> dict: + out = await put_json({"challenge_artifact": record, "items": claims, "on_refuted": "degrade", **output}, + kind="node_output") + events = [(1, "report.checked", "write", {"doc_artifact": doc, "ok": True}), + (2, "node.finished", "chal", {"artifact": out})] + return evidence._challenge_of(sealed_of(graph, events), report(doc), "u1") + + got = await answer({"counts": {"stands": 7, "weakened": 3, "refuted": 2, "inconclusive": 0}, "unreviewed": 1}) + assert got["counts"] == {"stands": 7, "weakened": 3, "refuted": 2, "inconclusive": 0} and got["unreviewed"] == 1 + # 计数格式不对（一个数都认不出来）：按审稿记录里的句子数，不写四个 0 + got = await answer({"counts": {"stands": "7"}, "unreviewed": True}) + assert got["counts"] == {"stands": 1, "weakened": 1, "refuted": 0, "inconclusive": 0} and "unreviewed" not in got + + +async def test_a_review_that_failed_the_node_needs_its_record(): + """节点报错：认审完那一刻的日志。重跑（node.started）之后没再审完的不算；日志里没有审稿记录（保存失败）认不出审的 + 是哪一稿，也不算。""" + doc = await put_json({"markdown": "周报", "n": 4}, kind="report_doc") + claims = [{"unit": "u1", "text": "华南销售额超过华东。", "verdict": "refuted", "rationale": "", "rationale_doc": None, + "queries": []}] + record = await put_json({"report": "write", "doc_artifact": doc, "model": "mock", "claims": claims}, + kind="challenge") + graph = chain(node("write", "report", "写周报"), node("chal", "challenge", "审稿", report="write", + on_refuted="fail")) + counts = {"stands": 0, "weakened": 0, "refuted": 1, "inconclusive": 0} + head = [(1, "report.checked", "write", {"doc_artifact": doc, "ok": True}), (2, "node.started", "chal", {})] + done = (3, "log", "chal", {"code": "challenge_done", "counts": counts, "artifact": record}) + failed = (4, "node.failed", "chal", {"error": "反方审稿推翻了 1 句结论"}) + got = evidence._challenge_of(sealed_of(graph, [*head, done, failed]), report(doc), "u1") + assert got == {"verdict": "refuted", "node_id": "chal", "artifact": record, "on_refuted": "fail", + "report_label": "写周报", "counts": counts} + retried = [*head, done, (4, "node.started", "chal", {}), (5, "node.failed", "chal", {"error": "模型不可用"})] + assert evidence._challenge_of(sealed_of(graph, retried), report(doc), "u1") is None + lost = [*head, (3, "log", "chal", {"code": "challenge_done", "counts": counts, "artifact": None}), failed] + assert evidence._challenge_of(sealed_of(graph, lost), report(doc), "u1") is None diff --git a/backend/tests/test_positional_masks_elsewhere.py b/backend/tests/test_positional_masks_elsewhere.py new file mode 100644 index 00000000..654f822b --- /dev/null +++ b/backend/tests/test_positional_masks_elsewhere.py @@ -0,0 +1,113 @@ +"""按位置取自遮罩列的一格（vars.res.rows[1][2]、nodes.fetch.rows[0][0]）在推演之外的两处：期 4 I2a 把它接进了 +recompute 的还原（_positional），用 recompute 的地方都跟着变，这里把变化钉住。 + +- 双盲复算的预检（dual_check._plan_metric 什么都不改先重算一遍）：以前这种指标照常盲算，遮着的那一格还会作为字面量 + 写进给盲算模型的计算式；现在预检判它用到了遮罩列，按无法复算记，原因不说「推演」（这里不是推演） +- 运行对比（engine/compare.py）：这种输入两边的值都不给，标 masked，和 cell() 取的遮罩格一样 + +库和数字都是编的：号码写成带横线的文本，区域名是「东区」这类。 +""" +from __future__ import annotations + +import json + +from app.engine import compare +from app.engine.runner import run_manager +from tests.fixtures.compare_runs import BASE_ROWS, NEXT_ROWS, metric, scalar_in, week +from tests.test_evidence_metric_masks import ( # noqa: F401 - 夹具按名字注入（engine_up 自动生效） + METRICS, + PHONES, + SQL, + client, + engine_up, + node, + set_source, + shop, +) + + +def dual_graph(): + """取数 → 数据整形（vars.res）→ 口径卡 → 双盲复算（只复算 mix）→ 成果。""" + nodes = [ + node("start", "input"), + node("fetch", "tool", tool="db_query__shop", args={"sql": SQL}), + node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="res"), + node("card", "metrics", caliber="遮罩口径", caliber_version="v1", metrics=METRICS), + node("dual", "dual_check", metrics=["mix"]), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.card.text }}"}]), + ] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def dual_output(run_id: str) -> dict: + from sqlalchemy import select + + from app.core import artifact_store + from app.db.base import SessionLocal + from app.db.models import RunEvent + + async with SessionLocal() as session: + [finished] = (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "node.finished", RunEvent.node_id == "dual"))).scalars() + return artifact_store.load(finished.data["artifact"]) + + +async def test_the_dual_check_does_not_blind_compute_a_metric_reading_a_masked_cell_by_position(shop): + """mix = round(cell(nodes.fetch, 1, 'amount') + len(vars.res.rows[1][2]), 1)：cell() 那一格不在遮罩列上，按位置取的 + 号码在。预检判它用到了遮罩列，按无法复算记缺口，原因不说「推演」；没遮罩时预检照常通过、指标进入盲算。""" + import asyncio + + from app.db.base import SessionLocal + from app.db.models import Run + + async def run(options): + await set_source(shop, options) + run_id = (await run_manager.start(graph=dual_graph(), input_payload={})).id + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + await run_manager.wait_idle(run_id) + assert row.status == "succeeded", row.error + return await dual_output(run_id) + raise AssertionError(f"运行 {run_id} 没跑完：{row.status}") + + masked = await run({"mask_columns": ["phone"]}) + [mix] = masked["metrics"] + assert mix["metric_id"] == "mix" and mix["status"] == "unchecked" + assert mix["reason"] == "无法按原表达式重算这个指标（这个指标用到了设置了遮罩的列）" + assert "推演" not in json.dumps(masked, ensure_ascii=False) + assert not [p for p in PHONES if p in json.dumps(masked, ensure_ascii=False)] + assert any("遮罩" in gap for gap in masked["issuance_gaps"]) + + # 没遮罩：预检通过，原因不再是遮罩（盲算模型没配，按盲算没交回数记，但已经走到了盲算这一步） + plain = await run({}) + [mix] = plain["metrics"] + assert "遮罩" not in (mix.get("reason") or "") and "无法按原表达式重算" not in (mix.get("reason") or "") + + +def region_len(fetch, bot, rows, bot_rows): + """东区名称的字数：按位置取自取数结果的第 0 行第 0 列（region）。""" + return [metric("rlen", "len(nodes.fetch.rows[0][0])", len(rows[0][0]), + [scalar_in("nodes.fetch.rows[0][0]", rows[0][0], "fetch", "tool")], name="首个区域名字数", + unit="字", decimals=0)] + + +def test_the_compare_page_withholds_a_masked_cell_read_by_position(): + """运行对比：按位置取自遮罩列（region）的一格两边都不给值、标 masked，和 cell() 取的遮罩格一样；没遮罩时照给。""" + def masked(run): + run.masked = {a: frozenset({"region"}) for a in run.queries} + return run + + result = compare.compare(masked(week("r1", metrics=region_len)).side(), + masked(week("r2", NEXT_ROWS, metrics=region_len)).side()) + [rlen] = [m for m in result["metrics"] if m["metric_id"] == "rlen"] + [inp] = rlen["inputs"] + assert inp["masked"] is True and inp["base"] is None and inp["target"] is None + assert "东区" not in json.dumps(result, ensure_ascii=False) + + shown = compare.compare(week("r1", metrics=region_len).side(), week("r2", NEXT_ROWS, metrics=region_len).side()) + [rlen] = [m for m in shown["metrics"] if m["metric_id"] == "rlen"] + [inp] = rlen["inputs"] + assert not inp.get("masked") and inp["base"] == BASE_ROWS[0][0] and inp["target"] == NEXT_ROWS[0][0] diff --git a/backend/tests/test_recompute.py b/backend/tests/test_recompute.py index 1f5c4bcf..cb21fb40 100644 --- a/backend/tests/test_recompute.py +++ b/backend/tests/test_recompute.py @@ -29,6 +29,9 @@ BOT_OUT = "d4" * 32 FETCH_OUT = "e5" * 32 ITEMS_VALUE = "f6" * 32 RES_ROWS = "a7" * 32 +#: 数据整形节点 parse 解析好的 S1（数据源工具交回的结果原样解析，带着快照的 id）：vars.res 就是它。按位置取的 +#: vars.res.rows[0][1] 顺着它认得出取自 S1 的 amount 列；没有这份产出时那一格遮没遮核对不了，推演不给值 +PARSE_OUT = "c9" * 32 STORE = { **SNAPSHOTS, BOT_OUT: {"text": "四个区域合计 340", "data": {"amounts": [120.5, 80.0, 99.5, 40.0]}, @@ -37,6 +40,7 @@ STORE = { FETCH_OUT: {"items": ["甲", "乙", "丙"], "k": 2}, ITEMS_VALUE: {"path": "vars.kpi.items", "value": ["甲", "乙", "丙"]}, RES_ROWS: {"path": "vars.res.rows", "value": SNAPSHOTS[S1]["rows"]}, + PARSE_OUT: {**SNAPSHOTS[S1], "artifact": S1}, } @@ -47,7 +51,7 @@ def load(artifact): def sources(**kw): - base = {"load": load, "queries": (S1, S2, BAD), "outputs": {"bot": BOT_OUT, "fetch": FETCH_OUT}, + base = {"load": load, "queries": (S1, S2, BAD), "outputs": {"bot": BOT_OUT, "fetch": FETCH_OUT, "parse": PARSE_OUT}, "masked": {S2: frozenset({"phone"})}} return Sources(**{**base, **kw})