From 32dbfcfb522707a56b241c286c4629e08a3ad211 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 18 Oct 2026 18:15:37 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=8D=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=88?= =?UTF-8?q?=E5=90=8E=E7=AB=AF=EF=BC=89=EF=BC=9A=E5=AD=90=E5=9B=BE=E9=87=8C?= =?UTF-8?q?=E7=9A=84=E6=8A=A5=E5=91=8A=E4=B8=8D=E5=86=8D=E8=A2=AB=E5=88=A4?= =?UTF-8?q?=E3=80=8C=E7=96=91=E4=BC=BC=E8=A2=AB=E4=BF=AE=E6=94=B9=E3=80=8D?= =?UTF-8?q?=E3=80=81=E9=A9=B3=E5=9B=9E=E5=87=BA=E5=8F=A3=E7=9A=84=E5=8F=8C?= =?UTF-8?q?=E7=9B=B2=E8=A3=81=E5=86=B3=E7=85=A7=E6=A0=B7=E9=99=8D=E6=A1=A3?= =?UTF-8?q?=E3=80=81=E5=8F=AF=E9=AA=8C=E8=AF=81=E6=8A=A5=E5=91=8A=E4=B8=8E?= =?UTF-8?q?=E6=8E=A8=E6=BC=94=E6=8C=89=E5=8D=A1=E6=89=80=E5=9C=A8=E7=9A=84?= =?UTF-8?q?=E5=B1=82=E6=89=BE=E4=BA=A7=E5=87=BA=E3=80=81=E5=AD=90=E5=9B=BE?= =?UTF-8?q?=E9=82=A3=E4=B8=80=E6=AD=A5=E5=86=99=E5=90=8D=E5=AD=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - B1 子工作流里有报告撰写节点时：它的 report.checked 进了父运行（节点 id「sub/write」），报告文档却记子图里的 id「write」， 证据接口、可验证报告、质询拿两边一比判「疑似被修改」，评测读答案也说答案不能当证据。 - nodes/report：报告文档、产出工件记这次运行里的节点 id（ctx.run.global_id），和事件同一个；nodes/io 子图自己的出具契约 按同一个 id 核对，_issuance.report.node_id 同样写这次运行里的 id - api/evidence：_reports 默认只认父图上的报告（子图的报告是子工作流那一步内部的东西，父图的答案是子工作流出口的值）， nested=True 连子图里的一起给；点名要子图报告的（?report=sub/write）照样点得开（_named_reports）；按成果挑报告的兜底 不再挑中子图的报告。成果上没有 _issuance 时退回最后一条 issuance 事件，子图成果节点的那条不认 - bundle：出具档位（_tier）同样不认子图成果节点的 issuance，verifier.js 同一个认法；子图报告的文档在引用链上照常核对 - B2 双盲裁决被驳回、走驳回出口接到出具成果时：节点产出换成 exit_payload，出具缺口和 dual_check 台账条目跟着丢了，审核人驳回的 那组数照样完整出具（比选「两边都不对」还宽）。NodeRejected 加 exit_updates（节点在驳回之前已算出的产出字段、台账、用量）， compiler._rejected_exit 并进去；双盲复算驳回时照记缺口「指标「X」两路结果不一致，审核人驳回了裁决」和一条结论为驳回的 dual_check。核对过别的关卡：人工节点、工具调用审批、动作确认、代码执行审批都在产出缺口、台账之前就被驳回，没有可丢的 - B4 子图里的卡转抄同一张子图里的卡、按位置取格：可验证报告（bundle/build._outputs_of）和推演（api/whatif._sources）按这次 运行里的 id 查产出，子图卡上记的是子图里的 id，查不到就整张遮住、推演不了；父图有同名节点时还拿父图那个当依据。改成按卡 所在那一层的 id 查（engine/lineage.in_scope / scoped_id，证据面板 _InputMasks 同一个认法），转抄的卡认不出时的兜底同样换层 - B6 子图里的一步出错、被驳回后 Run.error_node_id 是「sub/x」：实例页（apps/resolve）、通知（notify/service）照终态事件上记的 名字写「子工作流 › x」（run_node_label），不露出内部 id 测试：test_subgraph_evidence（子图含报告：证据图按父图、评测读答案不报被改、子图报告点得开、可验证报告核对通过且不写子图的 档位、子图自己的契约照常；子图卡在可验证报告里不被遮、推演还原得出、父图同名节点不当依据；子图那一步的名字）、 test_dual_check_node（驳回出口接到出具成果照样降档、台账照记、证据面板认得）。每条都去掉修复回跑过，确实失败 --- backend/app/api/evidence.py | 48 ++++--- backend/app/api/whatif.py | 19 ++- backend/app/apps/resolve.py | 20 ++- backend/app/bundle/build.py | 22 ++- backend/app/bundle/verifier.js | 5 +- backend/app/engine/compiler.py | 13 +- backend/app/engine/context.py | 3 + backend/app/engine/lineage.py | 26 ++++ backend/app/engine/nodes/dual_check.py | 24 +++- backend/app/engine/nodes/io.py | 6 +- backend/app/engine/nodes/report.py | 7 +- backend/app/notify/service.py | 8 +- backend/tests/test_dual_check_node.py | 34 +++++ backend/tests/test_subgraph_evidence.py | 178 ++++++++++++++++++++++++ 14 files changed, 368 insertions(+), 45 deletions(-) diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 6c42552c..82f491a0 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -323,11 +323,17 @@ class _Report: claims: str | None = None -def _reports(sealed: _Sealed) -> list[_Report]: - """封存范围内每个报告节点最后一次的自查结果，按第一次出现的顺序。""" +def _reports(sealed: _Sealed, *, nested: bool = False) -> list[_Report]: + """封存范围内每个报告节点最后一次的自查结果，按第一次出现的顺序。 + + 默认只认这次运行的图上的报告节点：子工作流里的报告（节点 id「子工作流节点 id/…」，子图的事件和父运行在同一个流里） + 是子工作流那一步内部的东西，父图的答案是子工作流出口的值，不是那份报告本身——证据图、取答案的评测、按成果挑报告 + （_choose）都按父图自己的认。nested 为真时连子工作流里的一起给（按 ?report=sub/write 点开它们）。可验证报告另按全部 + report.checked 核对（bundle/build._report_events：自查过的文档一份都不能少）。""" latest: dict[str, dict[str, Any]] = {} for _, etype, node_id, data in sealed.events: - if etype == "report.checked" and node_id and isinstance(data, dict) and data.get("doc_artifact"): + if etype == "report.checked" and node_id and isinstance(data, dict) and data.get("doc_artifact") \ + and (nested or "/" not in node_id): latest[node_id] = data evidence = _output_evidence(sealed) out = [] @@ -579,11 +585,13 @@ def _metric_in(card: dict[str, Any] | None, metric_id: Any) -> dict[str, Any] | def _issuance_of(sealed: _Sealed) -> dict[str, Any] | None: - """成果上的 _issuance；成果里取不到时退回封存范围内最后一条 issuance 事件。""" + """成果上的 _issuance；成果里取不到时退回封存范围内最后一条 issuance 事件。子工作流里成果节点的出具判定（节点 id + 「子工作流节点 id/…」，子图的事件和父运行在同一个流里）说的是子工作流交出的那份，不是这次运行的，不认。""" output = sealed.payload_of(sealed.nodes_of("output")) or {} issuance = output.get("_issuance") if not isinstance(issuance, dict): - issuance = next((data for _, etype, _, data in reversed(sealed.events) if etype == "issuance"), None) + issuance = next((data for _, etype, node_id, data in reversed(sealed.events) + if etype == "issuance" and "/" not in str(node_id or "")), None) return issuance if isinstance(issuance, dict) else None @@ -873,7 +881,7 @@ async def evidence_metric(run_id: str, node: str, metric: str) -> dict[str, Any] def _open_segment(sealed: _Sealed, wanted: str | None, segment_id: str ) -> tuple[_Report, tuple[dict[str, Any], dict[str, Any], dict[str, Any]]]: """选报告、确认文档完好、找到片段：(报告, (块, 句子, 片段))。片段接口和推断来源接口共用同一套错误码。""" - reports = _reports(sealed) + reports = _named_reports(sealed, wanted) if not reports: raise CodedHTTPException(404, "本次运行没有封存的报告文档，没有可查看的片段", REPORT_NOT_FOUND) chosen = _choose(reports, wanted, sealed) @@ -887,6 +895,12 @@ def _open_segment(sealed: _Sealed, wanted: str | None, segment_id: str return chosen, found +def _named_reports(sealed: _Sealed, wanted: str | None) -> list[_Report]: + """点名要某一份报告时从哪些里挑：点的是子工作流里的那份（「sub/write」）连子工作流里的一起给，否则只给父图上的 + （_reports）。""" + return _reports(sealed, nested=bool(wanted) and "/" in str(wanted)) + + def _choose(reports: list[_Report], wanted: str | None, sealed: _Sealed) -> _Report: if wanted: chosen = next((r for r in reports if r.node_id == wanted), None) @@ -1097,13 +1111,10 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask def _scoped(card_node: Any, node_id: Any) -> Any: - """口径卡上记的节点 id（输入来自哪个节点、穿过的整形节点、转抄的卡）是卡所在那一层图里的 id：子工作流里的卡 - （节点 id「子工作流节点 id/卡」）记的是子图里的 id，换成这次运行里的（接上同一个前缀）。最外层的卡原样；经子工作流 - 出口接上的出处（multi.run_subgraph 记的）已经是卡这一层的 id，同样接上卡的前缀就对了。""" - card = str(card_node or "") - if not isinstance(node_id, str) or not node_id or "/" not in card: - return node_id - return f"{card.rsplit('/', 1)[0]}/{node_id}" + """口径卡上记的节点 id 换成这次运行里的（engine/lineage.scoped_id：子工作流里的卡记的是子图里的 id）。""" + from app.engine.lineage import scoped_id + + return scoped_id(card_node, node_id) def _scoped_ids(item: dict[str, Any], card_node: str, sealed: _Sealed) -> dict[str, Any]: @@ -1335,10 +1346,11 @@ class _CopyGraph: upstream: list[Any] = [] # 转抄的是哪张卡：按读到的产出上记着的指标集认（经数据整形赋给变量、转了一道的也认得出是哪张卡）， # 认不出的按取值链上的那个节点 - owner = copied.node_id + # 取值链上写的是卡所在那一层图里的 id（_scoped） + owner = _scoped(node, copied.node_id) if copied.card is not None: up_art = str(copied.card.get("artifact") or "") - owner = self._owner(up_art) or copied.node_id + owner = self._owner(up_art) or owner for index in copied.picked: up_metric = copied.card["metrics"][index] if isinstance(up_metric, dict) and up_metric.get("inputs"): @@ -2943,7 +2955,7 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response ) sealed = await _sealed(run_id) - reports = _reports(sealed) + reports = _named_reports(sealed, report) if not reports: raise CodedHTTPException(404, "本次运行没有报告文档，没有可导出的报告", REPORT_NOT_FOUND) chosen = _choose(reports, report, sealed) @@ -3896,7 +3908,7 @@ def _formal_message(doc: dict[str, Any] | None, claims: str | None, verdicts: li def _formal_refusal(sealed: _Sealed, wanted: str | None, units: list[str] | None = None) -> str: """按需裁判接口拒绝正式运行时的那句话，和片段接口说的一致：按请求要的那份报告（没指定取成果里标注的那一份）、 点的那几句结论句（不是结论句的不算）说。选不出报告、文档读不出来的，只说正式运行不能按需裁判（FORMAL_ONLY）。""" - reports = _reports(sealed) + reports = _named_reports(sealed, wanted) try: chosen = _choose(reports, wanted, sealed) if reports else None except CodedHTTPException: @@ -4070,7 +4082,7 @@ async def judge_units(run_id: str, payload: Any = Body(None), report: str | None raise CodedHTTPException(409, UNSEALED_JUDGE, JUDGE_UNSEALED) if not sealed.trusted: raise CodedHTTPException(409, BROKEN_JUDGE, SEAL_BROKEN_CODE) - reports = _reports(sealed) + reports = _named_reports(sealed, wanted) if not reports: raise CodedHTTPException(404, "这次运行没有封存的报告文档，没有可以判断的句子", REPORT_NOT_FOUND) chosen = _choose(reports, wanted, sealed) diff --git a/backend/app/api/whatif.py b/backend/app/api/whatif.py index a9275906..9ebcb51c 100644 --- a/backend/app/api/whatif.py +++ b/backend/app/api/whatif.py @@ -86,7 +86,7 @@ async def evidence_whatif(run_id: str, payload: Any = Body(None)) -> dict[str, A # 转抄上游口径卡的输入给不给值：和证据面板同一张指标依赖图（copied_inputs → recompute.masked_closure），在事件循环里 # 认好交进去，重算的线程里不再查库 copied = await copied_inputs(sealed, card_node, artifact, card) - sources = _sources(sealed, cards, started, asyncio.get_running_loop(), copied=copied) + sources = _sources(sealed, cards, started, asyncio.get_running_loop(), copied=copied, card_node=card_node) docs = [(r.node_id, r.doc) for r in reports if r.doc is not None and r.hash_ok is True and (report is None or r.node_id == report)] try: @@ -246,15 +246,22 @@ def _card_of(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]], node def _sources(sealed: _Sealed, cards: list[tuple[str, str, dict[str, Any]]], started: int, - loop: asyncio.AbstractEventLoop, *, copied: Mapping[int, Mapping[str, Any]] | None = None - ) -> recompute.Sources: + loop: asyncio.AbstractEventLoop, *, copied: Mapping[int, Mapping[str, Any]] | None = None, + card_node: str = "") -> recompute.Sources: """重算要用的东西（recompute.Sources）：封存范围内的工件、口径卡那次执行之前的节点产出和跑完的时刻、快照交回的 时刻、用到哪份才算哪份的遮罩、转抄上游口径卡的输入给不给值（copied）。拿到的 Sources 要在工作线程里用（_Hidden - 在线程里等事件循环查库）。""" + 在线程里等事件循环查库）。 + + 节点产出、跑完的时刻按口径卡所在那一层的 id 交进去（engine/lineage.in_scope）：子工作流里的卡（card_node「子工作流 + 节点 id/卡」）的取值链写的是子图里的 id，只看同一张子图里的节点，父图里同名的不算。""" + from app.engine.lineage import in_scope + loader = _loader(sealed, cards) hidden = _Hidden(sealed, loader, loop) - return recompute.Sources(load=loader, queries=tuple(sealed.queries), outputs=_outputs_before(sealed, started), - masked=hidden, returned=_returned(sealed), finished=_finished_before(sealed, started), + return recompute.Sources(load=loader, queries=tuple(sealed.queries), + outputs=in_scope(_outputs_before(sealed, started), card_node), + masked=hidden, returned=_returned(sealed), + finished=in_scope(_finished_before(sealed, started), card_node), secrets=hidden.secrets, copied=dict(copied or {})) diff --git a/backend/app/apps/resolve.py b/backend/app/apps/resolve.py index ea7abd3c..c7b0ccb4 100644 --- a/backend/app/apps/resolve.py +++ b/backend/app/apps/resolve.py @@ -1506,6 +1506,9 @@ class _Trail: #: 走了「驳回」出口、而且最后一次执行就是走的驳回出口的关卡：节点 id → 那条 node.finished 的数据（outcome=rejected， #: 带 to：转到哪几步。engine/compiler._rejected_exit）。退回修改之后又批准了的不算：结局是批准 rejected_out: dict[str, dict[str, Any]] = field(default_factory=dict) + #: 终态事件（run.failed / run.cancelled）上记的出错、被驳回的那一步：节点 id → 名字。子工作流里的那一步（「sub/x」） + #: 不在运行的图上，名字只有这里有（runner._label_of 接好的「子工作流 › x」） + end_labels: dict[str, str] = field(default_factory=dict) async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]: @@ -1532,9 +1535,24 @@ async def _trails(session: AsyncSession, runs: list[Run]) -> dict[str, _Trail]: if node_id: out[run_id].events.append((etype, node_id)) await _rejected_exits(session, [r for r in runs if r.status == "succeeded"], out) + ended = [r.id for r in runs if r.status in ("failed", "cancelled")] + if ended: + for run_id, data in (await session.execute(select(RunEvent.run_id, RunEvent.data).where( + RunEvent.run_id.in_(ended), RunEvent.type.in_(("run.failed", "run.cancelled"))))).all(): + if isinstance(data, dict) and data.get("node_id") and isinstance(data.get("label"), str) and data["label"]: + out[run_id].end_labels[str(data["node_id"])] = data["label"] return out +def run_node_label(run: Run, trail: _Trail, node_id: str) -> str: + """运行里一个节点给人看的名字：运行的图上的标题；图上没有的（子工作流里的那一步「sub/x」）用终态事件上记的名字 + （「子工作流 › x」）；都没有的照旧写 id。实例页、通知（notify/service）说同一个名字。""" + from app.apps.tree import node_map + + found = (node_map(run.graph).get(node_id) or {}).get("title") + return found or trail.end_labels.get(node_id) or node_id + + async def _rejected_exits(session: AsyncSession, runs: list[Run], out: dict[str, _Trail]) -> None: """成功结束、途经驳回出口的运行（run.finished 带 rejected_via，engine/runner._rejected_via）：每个走过驳回出口的关卡， 最后一次执行是不是走的驳回出口（退回修改之后又批准的，结局是批准）。只翻途经驳回的运行里那几个关卡的 node.finished。""" @@ -1651,7 +1669,7 @@ def instance_state(run: Run, trail: _Trail) -> dict[str, str]: nodes = node_map(run.graph) def title(node_id: str) -> str: - return (nodes.get(node_id) or {}).get("title") or node_id + return (nodes.get(node_id) or {}).get("title") or trail.end_labels.get(node_id) or node_id if run.status == "interrupted": if any(a.state in ("sent", "unknown") for a in trail.actions): diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index b088980a..f742a610 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -58,6 +58,7 @@ from urllib.parse import quote from app.core.artifact_store import canonical_json, content_hash from app.core.events import EventType from app.engine.evidence import iter_units, query_entry_fields +from app.engine.lineage import in_scope, scoped_id from app.engine.masking import OPAQUE_SQL, SYSTEM_SCHEMAS, indirect_reads from app.engine.merge_query import MERGE_SOURCE from app.engine.recompute import ( @@ -300,10 +301,11 @@ class _Store: if copied is not None: upstream: list[Any] = [] # 转抄的是哪张卡：按读到的产出上记着的指标集认（经数据整形赋给变量的也认得出），认不出的按取值链上的节点 - owner = copied.node_id + # 取值链上写的是卡所在那一层图里的 id（engine/lineage.scoped_id） + owner = scoped_id(self._owner(art), copied.node_id) if copied.card is not None: up_card = str(copied.card.get("artifact") or "") - owner = self._owner(up_card) or copied.node_id + owner = self._owner(up_card) or owner for index in copied.picked: up_metric = copied.card["metrics"][index] if isinstance(up_metric, dict) and up_metric.get("inputs"): @@ -353,9 +355,13 @@ class _Store: def _outputs_of(self, card: str, events: Sequence[tuple[int, str, str | None, Any]]) -> Callable[[Any], Any]: """口径卡 card 那一次执行之前每个节点的产出（_card_outputs）：节点 id → 读出来的内容，读不出来、和 id 对不上的是 - None。同一张卡只找一次。""" + None。同一张卡只找一次。 + + 卡上的取值链按卡所在那一层图里的 id 写（engine/lineage.in_scope）：子工作流里的卡（台账记在「子工作流节点 id/卡」 + 上）只看同一张子图里的节点，按去掉前缀的 id 查，和证据面板 api/evidence._InputMasks._output 同一个认法。以前按这次 + 运行里的 id 查，子图里的卡转抄、按位置取的格一律查不到、当成核对不了整张遮住；父图里有同名的节点时还拿它当依据。""" if card not in self._card_outputs: - self._card_outputs[card] = _card_outputs(events, card) + self._card_outputs[card] = in_scope(_card_outputs(events, card), self._owner(card)) outputs = self._card_outputs[card] def output(node_id: Any) -> Any: @@ -1207,9 +1213,11 @@ def _summary(etype: str, data: Any) -> str: def _tier(events: Sequence[tuple[int, str, str | None, Any]]) -> dict[str, Any] | None: - """出具档位：封存范围内最后一条出具判定事件里的 tier，带上序号（浏览器拿那条已核对的事件对照）。""" - for seq, etype, _, data in reversed(events): - if etype == EventType.ISSUANCE and isinstance(data, dict) and isinstance(data.get("tier"), str): + """出具档位：封存范围内最后一条出具判定事件里的 tier，带上序号（浏览器拿那条已核对的事件对照）。子工作流里成果节点 + 的出具判定（节点 id「子工作流节点 id/…」）说的是子工作流交出的那份，不是这次运行的，不认（verifier.js 同一个认法）。""" + for seq, etype, node_id, data in reversed(events): + if etype == EventType.ISSUANCE and "/" not in str(node_id or "") and isinstance(data, dict) \ + and isinstance(data.get("tier"), str): return {"value": data["tier"], "seq": seq} return None diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index 24bf3a60..ffca5f15 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -1568,11 +1568,12 @@ } } - // 出具档位：封存范围内最后一条记下了档位的出具判定（和服务端 build._tier 同一个认法），还要是元信息指的那一条 + // 出具档位：封存范围内最后一条记下了档位的出具判定（和服务端 build._tier 同一个认法），还要是元信息指的那一条。 + // 子工作流里成果节点的出具判定（节点 id「子工作流节点 id/…」）说的是子工作流交出的那份，不是这次运行的，不认 const tierClaim = isObj(meta.tier) ? label(TIER_LABEL, meta.tier.value) : '—' let found = null for (const ev of [...order].reverse()) { - if (ev.type !== 'issuance') continue + if (ev.type !== 'issuance' || String(ev.node_id ?? '').includes('/')) continue if (typeof ev.canonical !== 'string') found = { result: FAIL, reason: T.tierAbsent(ev.seq) } else if (!out.checked.has(ev)) found = { result: NA, reason: T.tierUnchecked(ev.seq) } else if (typeof own(out.checked.get(ev), 'tier') === 'string') found = { ev, value: out.checked.get(ev).tier } diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 9e3a1f72..bfc48df8 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -218,13 +218,17 @@ async def _rejected_exit(node: GraphNode, ctx: NodeContext, error: NodeRejected, """被驳回、接了「驳回」出口：这一步正常结束（node.finished 带 outcome=rejected 和哪种关卡、结论、谁、备注、转到哪几步）， 产出 NodeRejected.exit_payload()，路由按它的 __decision__ 走驳回那条边。 + 节点在驳回之前已经算出的（NodeRejected.exit_updates）并进产出、台账和用量：双盲复算的裁决被驳回时，复算比出的不一致 + 照样记成出具缺口、台账照记一条结论为驳回的 dual_check，下游的出具不会把审核人驳回的那组数完整出具。 动作确认被驳回时 NodeRejected 抛在发送之前，走到这里时什么都没发（照旧「未确认（被驳回）」）。trail 上这一步记 outcome=rejected：运行收尾据此在 run.finished 上照实写「途经驳回」（runner._rejected_via）。审批模式写了「结果存为 变量」的照旧写 false（以前接了驳回边的审批就是这么写的）；别的关卡这一步没有产出值，变量不写。""" from app.core.artifact_store import put_json run_ctx = ctx.run - payload = error.exit_payload() + extra = error.exit_updates or {} + # 节点在驳回之前已经算出的（NodeRejected.exit_updates：双盲复算的出具缺口、台账条目）照样交出；驳回的那几个键在上面 + payload = {**(extra.get("payload") or {}), **error.exit_payload()} nodes = run_ctx.spec.node_map() to = [{"node_id": run_ctx.global_id(t), "label": nodes[t].title if t in nodes else t} for t in dict.fromkeys(targets)] try: @@ -234,12 +238,17 @@ async def _rejected_exit(node: GraphNode, ctx: NodeContext, error: NodeRejected, except Exception: # noqa: BLE001 - 工件写失败不该毁掉运行本身 artifact_id = None said = {"kind": error.kind, "decision": error.decision, "actor": error.actor, "note": error.note} + ledger = [e for e in extra.get("evidence") or [] if isinstance(e, dict)] ctx.emit(EventType.NODE_FINISHED, duration_ms=elapsed, attempt=attempt, preview=_preview({"nodes": {node.id: payload}}, node.id), artifact=artifact_id, - outcome="rejected", **said, to=to) + outcome="rejected", **said, to=to, **({"evidence": ledger} if ledger else {})) updates: dict[str, Any] = {"nodes": {node.id: payload}, "trail": [{ "node_id": node.id, "type": str(node.type), "duration_ms": elapsed, "attempt": attempt, "ts": time.time(), "outcome": "rejected", "label": node.title, **said, "to": to}]} + if ledger: + updates["evidence"] = ledger + if extra.get("usage"): + updates["usage"] = dict(extra["usage"]) var_name = str(node.config.get("assign_to") or "").strip() if var_name and error.kind == "approve": updates["vars"] = {var_name: False} diff --git a/backend/app/engine/context.py b/backend/app/engine/context.py index 26dee96b..fb85a203 100644 --- a/backend/app/engine/context.py +++ b/backend/app/engine/context.py @@ -167,6 +167,9 @@ class NodeRejected(NodeError): self.actor = actor self.note = note self.stop = stop + #: 走驳回出口时这一步另外交出的：{payload: 产出里的其余字段, evidence: 台账条目, usage: 用量}。节点在抛之前按需填： + #: 驳回之前已经算出、下游要据此判档的东西不能跟着驳回丢掉（双盲复算的出具缺口和台账条目）。没接出口时用不上 + self.exit_updates: dict[str, Any] | None = None @classmethod def of(cls, ctx: Any, message: str, *, kind: str, decision: Any, stop: bool = False) -> NodeRejected: diff --git a/backend/app/engine/lineage.py b/backend/app/engine/lineage.py index 93630579..e1e01b2c 100644 --- a/backend/app/engine/lineage.py +++ b/backend/app/engine/lineage.py @@ -137,6 +137,32 @@ def untraced(inputs: Iterable[Any]) -> list[dict[str, Any]]: if isinstance(item, dict) and item.get("value") is not None and not traced(item)] +# -------------------------------------------------------------------------- +# 卡上记的节点 id 在哪一层 +# -------------------------------------------------------------------------- + + +def scoped_id(card_node: Any, node_id: Any) -> Any: + """口径卡上记的节点 id（输入来自哪个节点、穿过的整形节点、转抄的卡、取值链里写的 nodes.X）是卡所在那一层图里的 id： + 子工作流里的卡（这次运行里的 id「子工作流节点 id/卡」）记的是子图里的 id，换成这次运行里的（接上同一个前缀）。最外层 + 的卡原样；经子工作流出口接上的出处（multi.run_subgraph 记的「子工作流节点 id/…」）已经是卡这一层的 id，同样接上卡的 + 前缀就对了。证据面板、可验证报告、推演都按这一个认。""" + card = str(card_node or "") + if not isinstance(node_id, str) or not node_id or "/" not in card: + return node_id + return f"{card.rsplit('/', 1)[0]}/{node_id}" + + +def in_scope(table: Mapping[str, Any], card_node: Any) -> dict[str, Any]: + """按这次运行里的节点 id 记的表（每个节点那一刻的产出、跑完的时刻）→ 按口径卡所在那一层的 id 记的：子工作流里的卡只看 + 同一张子图里的（去掉前缀；父图里同名的节点不算），最外层的卡原样。卡上的取值链按它这一层的 id 查（scoped_id 的反面）。""" + card = str(card_node or "") + if "/" not in card: + return dict(table) + prefix = card.rsplit("/", 1)[0] + "/" + return {key[len(prefix):]: value for key, value in table.items() if key.startswith(prefix)} + + # -------------------------------------------------------------------------- # 静态地认（只有图） # -------------------------------------------------------------------------- diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index b334e5cb..642cd847 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -86,6 +86,8 @@ from app.engine.tolerance import agree as _agree, delta_pct as _delta_pct #: 节点产出的状态：跳过、全部一致、核成了一部分（其余无法复算）、一个都没核成、人已裁决 SKIPPED, AGREED, PARTIAL, NONE_CHECKED, RESOLVED = "skipped", "agreed", "partial", "unchecked", "resolved" +#: 审核人驳回了裁决、接了驳回出口沿它往下走（docs/design/orchestration-round/00-总方案.md §3）：这一步的结论 +REJECTED = "rejected" #: 单个指标的结果：一致、不一致、无法复算 SAME, DIFFERENT, UNCHECKED = "agreed", "disagreed", "unchecked" #: 审核人的三种选择：采用原流程、采用盲算、两边都不对 @@ -133,6 +135,7 @@ NO_SQL = "有一边没有可比较的 SQL，请对照原文" GAP_BLIND = "指标「{name}」两路结果不一致，审核人选择了盲算结果" GAP_NEITHER = "指标「{name}」两路结果不一致，审核人判断两边都不对" GAP_UNANSWERED = "指标「{name}」两路结果不一致，审核人没有作出选择" +GAP_REJECTED = "指标「{name}」两路结果不一致，审核人驳回了裁决" GAP_UNCHECKED = "指标「{name}」无法复算：{reason}" #: 运行时长上限将至（engine/deadline.py）：没来得及开始的盲算、时长按剩余时间收紧后到了点的盲算，都按无法复算记 LATE = "本次运行即将达到 {seconds:g} 秒的时长上限，为保证运行按时完成，未进行盲算" @@ -222,7 +225,17 @@ async def run_dual_check(state: GraphState, ctx: NodeContext) -> dict[str, Any]: response = await ask(ctx, review["payload"], **review["payload"]) # 谁裁决的跟着事件走，和人工审批节点一样 await once(ctx, EventType.HUMAN_RESOLVED, response=response, actor=ctx.actor()) - choices, note = _choices(ctx, response, disagreed) + try: + choices, note = _choices(ctx, response, disagreed) + except NodeRejected as e: + # 审核人驳回了裁决。接了驳回出口时（engine/compiler._rejected_exit）这一步照常结束：两路不一致的那几个指标照样记成 + # 出具缺口，台账照记一条结论为驳回的 dual_check——驳回不是认可，下游的出具不能把这组数完整出具（不比选「两边 + # 都不对」宽）。没接出口的运行收成被驳回，这些用不上 + done = _finish(state, ctx, card_id, card, checked, status=REJECTED, gaps=_rejected_gaps(rows), choices={}, + note=e.note, artifact=review["card_artifact"]) + e.exit_updates = {"payload": done["nodes"][ctx.node.id], "evidence": done.get("evidence"), + "usage": done.get("usage")} + raise gaps = _gaps(rows, choices) await once(ctx, EventType.LOG, level="info", code="dual_check_resolved", choices=choices, gaps=gaps, message=_resolved_text(disagreed, choices)) @@ -1040,6 +1053,13 @@ def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any row["differing"] = differing or [i["key"] for i in row["inputs"]] +def _rejected_gaps(rows: list[dict[str, Any]]) -> list[str]: + """裁决被驳回时的出具缺口：不一致的写审核人驳回了裁决，无法复算的照旧写原因。""" + return [GAP_REJECTED.format(name=r["name"]) if r["status"] == DIFFERENT + else GAP_UNCHECKED.format(name=r["name"], reason=r.get("reason") or "原因不明") + for r in rows if r["status"] in (DIFFERENT, UNCHECKED)] + + def _unchecked_gaps(rows: list[dict[str, Any]]) -> list[str]: return [GAP_UNCHECKED.format(name=r["name"], reason=r.get("reason") or "原因不明") for r in rows if r["status"] == UNCHECKED] @@ -1474,4 +1494,4 @@ def _finish(state: GraphState, ctx: NodeContext, card_id: str, card: dict[str, A return updates -__all__ = ["AGREED", "CHOICES", "NONE_CHECKED", "PARTIAL", "RESOLVED", "SKIPPED", "run_dual_check"] +__all__ = ["AGREED", "CHOICES", "NONE_CHECKED", "PARTIAL", "REJECTED", "RESOLVED", "SKIPPED", "run_dual_check"] diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index ff53d3ca..c642a0fd 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -443,7 +443,8 @@ def _check_citations( if not isinstance(doc, dict): gaps.append(f"无法读取报告「{node.title}」的文档，未执行引用核对") return out - if doc.get("node_id") != report_from or doc.get("run_id") != ctx.run.run_id \ + # 文档记的是这次运行里的节点 id（子工作流里的报告带前缀，nodes/report 的 compose） + if doc.get("node_id") != ctx.run.global_id(str(report_from)) or doc.get("run_id") != ctx.run.run_id \ or doc.get("markdown") != payload.get("text"): gaps.append(f"报告「{node.title}」的文档与本次运行中的产出不一致，未执行引用核对") return out @@ -899,7 +900,8 @@ def _declare_citations( declared_at: Any, ctx: NodeContext, sql_checks: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """引用模式的 _issuance。旧字段一个不少（老前端照常显示），另加 mode、report、unresolved。""" - report = {"node_id": report_from, "doc_artifact": cited["doc_artifact"]} + # 节点 id 和报告文档、report.checked 事件上记的同一个（这次运行里的，子工作流里的带前缀） + report = {"node_id": ctx.run.global_id(report_from), "doc_artifact": cited["doc_artifact"]} issuance = { "mode": "citations", "tier": tier, diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index b2596b6f..ff5ed473 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -620,7 +620,7 @@ async def _store(doc: dict[str, Any], ctx: NodeContext) -> str | None: from app.core.artifact_store import put_json try: - return await put_json(doc, kind="report_doc", run_id=ctx.run.run_id, node_id=ctx.node.id, + return await put_json(doc, kind="report_doc", run_id=ctx.run.run_id, node_id=ctx.run.global_id(ctx.node.id), meta={"stats": doc.get("stats") or {}}) except Exception as e: # noqa: BLE001 - 工件写失败不毁掉运行，只是这份报告点不开证据 ctx.emit(EventType.LOG, level="warn", code="evidence_store_failed", @@ -710,7 +710,10 @@ async def run_report(state: GraphState, ctx: NodeContext) -> dict[str, Any]: # 第一稿、重写稿、裁判退回的改写稿都从这里过：整篇包在代码块里的先去掉外层围栏。文档的 source 记的是去掉 # 围栏之后的正文——markdown 由它渲染，出口契约复核取回的也是这份文档。引用了遮罩列的格子组装好之后再认 # （_MaskedCells）：只读正文里引用到的那几份快照 - return await masked(compose_doc(_unfenced(raw), catalog, node_id=ctx.node.id, run_id=ctx.run.run_id, + # 文档记这次运行里的节点 id（子工作流里的报告是「子工作流节点 id/…」）：和它的 report.checked 事件、产出工件同一个 + # id。证据接口、可验证报告、质询都拿文档上的 id 和事件上的比，对不上就判「疑似被修改」 + return await masked(compose_doc(_unfenced(raw), catalog, node_id=ctx.run.global_id(ctx.node.id), + run_id=ctx.run.run_id, allow_numbers=allow, cells_allowed=cells, entities=entities, absence=writer.trimmed)) diff --git a/backend/app/notify/service.py b/backend/app/notify/service.py index 462f4643..4c401397 100644 --- a/backend/app/notify/service.py +++ b/backend/app/notify/service.py @@ -436,7 +436,7 @@ async def notify_run(run_id: str, event: str) -> int: 发起人是认证过的身份（run.origin.actor_verified）时，他自己订阅了这件事的个人渠道也收得到（「我发起的实例完成 / 失败」， 04-对外集成与触达 B1）。以前这里给 _channels 的 users 是空的，个人渠道勾了这两项永远收不到。自报的署名不算：关登录 时谁都能填别人的名字。""" - from app.apps.resolve import _trails, instance_state + from app.apps.resolve import _trails, instance_state, run_node_label async with SessionLocal() as session: run = await session.get(Run, run_id) @@ -448,12 +448,14 @@ async def notify_run(run_id: str, event: str) -> int: return 0 base = await _base_url(session) where = await _where(session, run) - state = instance_state(run, (await _trails(session, [run]))[run.id]) + trail = (await _trails(session, [run]))[run.id] + state = instance_state(run, trail) finished = event == "run.finished" ok = state["code"] == "succeeded" lines = [("状态", state["text"]), ("实例", f"#{run_id[:6]}")] if not finished and run.error_node_id and state["code"] == "failed": - lines.append(("失败节点", f"「{_label(run.graph, run.error_node_id)}」")) + # 子工作流里的那一步（「sub/x」）图上没有：照终态事件上记的名字写，和实例页同一个（apps/resolve.run_node_label） + lines.append(("失败节点", f"「{run_node_label(run, trail, run.error_node_id)[:80]}」")) head = ("已完成" if ok else "已结束") if finished else ("已结束" if state["code"] == "cancelled" else "失败") message = {"event": event, "status": run.status, "ok": ok, "title": f"{where or '应用'}的实例{head}", "lines": lines, diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 67e08954..01b207b7 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -555,6 +555,40 @@ async def test_rejecting_the_approval_with_a_reject_exit_goes_on_along_it(monkey assert not [e for e in evs if e.type == "node.finished" and e.node_id == "out"], "普通出口没有走" +async def test_a_rejected_arbitration_through_the_exit_still_degrades_the_issuance(monkeypatch, shop, client): + """驳回出口直接接到出具的成果节点：审核人驳回了裁决，那组数不能照样完整出具。走出口时这一步的产出照样带出具缺口 + （「审核人驳回了裁决」），台账照记一条结论为驳回的 dual_check，证据面板的指标步骤据此认出复核过这份指标集； + 不比选「两边都不对」宽。""" + from app.api.evidence import _dual_check_of, _sealed + + name, _ = shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + spec = graph(name, BY_ORDER_DATE) + spec["edges"].append({"source": "dual", "target": "out", "sourceHandle": "rejected"}) + run = await start(spec) + approval = await pending(run.id) + res = await client.post(f"/api/approvals/{approval.id}/decide", json={"approved": False, "note": "口径要重新对"}) + assert res.status_code == 200, res.text + done = await wait(run.id, ("succeeded", "failed", "cancelled")) + assert done.status == "succeeded", done.error + issuance = done.output["_issuance"] + assert issuance["tier"] == "degraded", issuance + assert "双盲复算「复核销售额」：指标「销售额」两路结果不一致，审核人驳回了裁决" in issuance["gaps"], issuance["gaps"] + evs = await events(run.id) + [finished] = [e for e in evs if e.type == "node.finished" and e.node_id == "dual"] + assert finished.data["outcome"] == "rejected" + [entry] = [e for e in finished.data.get("evidence") or [] if e.get("kind") == "dual_check"] + assert (entry["status"], entry["card"]) == ("rejected", "card") and entry["metrics"][0]["status"] == "disagreed" + assert finished.data["preview"]["status"] == "rejected" and finished.data["preview"]["rejected"] is True + sealed = await _sealed(run.id) + [(_, artifact, _)] = [c for c in __import__("app.api.evidence", fromlist=["_sealed_cards"])._sealed_cards(sealed)] + checked = _dual_check_of(sealed, "card", artifact, "gmv") + assert checked is not None and checked["status"] == "disagreed" and "choice" not in checked, checked + # 对照：同一处选「两边都不对」也是降档 + other = await start(graph(name, BY_ORDER_DATE)) + assert (await decide(client, other.id, {"gmv": "neither"})).output["_issuance"]["tier"] == "degraded" + + async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) diff --git a/backend/tests/test_subgraph_evidence.py b/backend/tests/test_subgraph_evidence.py index 572d1634..f03dcc21 100644 --- a/backend/tests/test_subgraph_evidence.py +++ b/backend/tests/test_subgraph_evidence.py @@ -119,6 +119,49 @@ async def test_the_subgraph_steps_are_in_the_run_with_prefixed_ids(): assert row.output == {"数": "246.5"} +async def test_a_step_inside_the_subgraph_is_named_by_its_label_not_its_id(): + """子图里的一步被驳回、出错时，运行记下的出错节点是「sub/x」：实例页、通知照终态事件里的名字写「子工作流 › x」， + 不露出内部 id。""" + from app.apps.resolve import _trails, instance_state, run_node_label + from app.db.models import Run + + sub = S.chain(node("s", "input"), + node("gate", "tool", "算一下", tool="calculator", args={"expression": "1+1"}, approval="always"), + node("o", "output", fields=[{"name": "x", "value": "{{ nodes.gate }}"}])) + sub_id = await publish(sub) + graph = S.chain(node("start", "input"), node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", fields=[{"name": "x", "value": "{{ nodes.sub.output.x }}"}])) + started = await run_manager.start(graph=graph, input_payload={}) + for _ in range(200): + async with SessionLocal() as session: + if (await session.get(Run, started.id)).status == "interrupted" and not run_manager.is_active(started.id): + break + await __import__("asyncio").sleep(0.05) + await run_manager.resume(started.id, {"approved": False, "note": "不对"}, actor="张工") + for _ in range(200): + async with SessionLocal() as session: + row = await session.get(Run, started.id) + if row.status == "cancelled" and not run_manager.is_active(started.id): + break + await __import__("asyncio").sleep(0.05) + assert (row.status, row.end_reason, row.error_node_id) == ("cancelled", "rejected", "sub/gate") + async with SessionLocal() as session: + trail = (await _trails(session, [row]))[row.id] + assert instance_state(row, trail) == {"code": "cancelled", "text": "「取数子流程 › 算一下」被驳回"} + assert run_node_label(row, trail, "sub/gate") == "取数子流程 › 算一下" + + failing = await publish(fetch_card_graph(sql="SELECT nope FROM nowhere"), "出错的") + started = await run_manager.start(graph=parent(failing, contract_from=None, + out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}]), + input_payload={}) + row = await S.wait(started.id) + assert row.status == "failed" and row.error_node_id == "sub/fetch" + async with SessionLocal() as session: + trail = (await _trails(session, [row]))[row.id] + assert run_node_label(row, trail, row.error_node_id) == "取数子流程 › fetch" + assert run_node_label(row, trail, "sub") == "取数子流程" + + async def test_the_instance_timeline_counts_the_subgraph_as_one_step(): """应用的实例进度只说这张图上的步骤：子图里的那几步（「sub/…」）不单列，子工作流算一步。""" from types import SimpleNamespace @@ -356,6 +399,141 @@ async def test_the_parent_report_cites_the_subgraph_query_and_card(client, monke assert res.status_code == 200 and "246.5" in res.text +#: 载入可验证报告的核对脚本，交回引用链上每份报告文档的核对结果（同 tests/test_bundle_rejected.py 的做法） +VERIFY_CHAIN = r""" +const fs = require('fs') +const { createHash } = require('crypto') +const input = JSON.parse(fs.readFileSync(0, 'utf8')) +const mod = { exports: {} } +new Function('module', fs.readFileSync(process.argv[1], 'utf8'))(mod) +const digest = async (bytes) => createHash('sha256').update(bytes).digest('hex') +mod.exports.verify(input.data, digest, input.data.meta.manifest_hash).then((out) => { + process.stdout.write(JSON.stringify(out.chain.filter((c) => c.kind === 'report_doc') + .map((c) => ({ node: c.report && c.report.node_id, result: c.result, reason: c.reason || '' })))) +}, (e) => { console.error((e && e.stack) || String(e)); process.exit(1) }) +""" + + +async def test_a_report_inside_the_subgraph_is_not_tampered_and_not_the_parents_answer(client, monkeypatch): + """子图里有报告撰写节点：它的 report.checked 进了父运行（节点 id「sub/write」），报告文档记的也是这次运行里的 id， + 不被认成「疑似被修改」。父图自己没有报告：父图的答案是子图出口的值，不是子图那份报告——证据图、评测读答案都按父图 + 自己认（和子图事件进运行之前一样：没有报告），子图的报告按 ?report=sub/write 点得开，可验证报告里核对通过。""" + import json + import re + import shutil + import subprocess + from pathlib import Path + + import app.bundle + from app.api.evidence import _reports, _sealed + from app.eval.score import read_answer + + sub = S.chain(node("s", "input"), node("fetch", "tool", tool=S.tool(), args={"sql": PAID}), + node("write", "report", instructions="写答案"), + node("o", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write"})) + sub_id = await publish(sub) + graph = S.chain(node("start", "input"), node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1), + node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.sub.output.答案 }}"}])) + S.script(monkeypatch, "子流程里查到的销售额 [[v:Q1.r0.gmv]]。") + started = await run_manager.start(graph=graph, input_payload={}) + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + + body = (await client.get(f"/api/runs/{row.id}/evidence")).json() + # 父图没有报告、没有契约：照没有契约的答案按数值猜候选（legacy_text，写明是猜的）。子图查过的数如今在封存范围里， + # 猜得到它；不再把子图的报告当成父图的、更不说它被改过 + assert (body["mode"], body["reports"]) == ("legacy_text", []), body + [guessed] = [c for f in body["legacy"]["fields"] for seg in f["segments"] for c in seg.get("candidates") or []] + assert (guessed["node_id"], guessed["value"]) == ("sub/fetch", 246.5) + # 子图自己的出具契约照常核对它那份报告（文档上记的 id 和这一层的报告节点对得上） + from app.core.artifact_store import load + + inner_issuance = load((await _sealed(row.id)).outputs["sub"])["output"]["_issuance"] + assert inner_issuance["tier"] == "formal" and inner_issuance["report"]["node_id"] == "sub/write", inner_issuance + answer = await read_answer(row.id) + assert answer.problem is None and answer.report is None and "246.5" in row.output["答案"], answer + # 子图的报告本身：这次运行里的 id、哈希核对通过、点得开 + [inner] = _reports(await _sealed(row.id), nested=True) + assert (inner.node_id, inner.hash_ok, (inner.doc or {}).get("node_id")) == ("sub/write", True, "sub/write") + seg = next(s["id"] for b in inner.doc["blocks"] for u in b["units"] for s in u["segments"] + if (s.get("cite") or {}).get("kind") == "cell") + res = await client.get(f"/api/runs/{row.id}/evidence/segments/{seg}", params={"report": "sub/write"}) + assert res.status_code == 200, res.text + [query] = [st for st in res.json()["chain"] if st["step"] == "query"] + assert query["node_id"] == "sub/fetch" and query["sealed"] is True + # 可验证报告：子图那份报告的文档在引用链上核对通过（文档记的节点 id 和自查事件对得上） + node_bin = shutil.which("node") + if node_bin: + html = (await client.get(f"/api/runs/{row.id}/bundle.html")).text + data = json.loads(re.search(r'', html, re.S).group(1)) + verifier = Path(app.bundle.__file__).parent / "verifier.js" + done = subprocess.run([node_bin, "-e", VERIFY_CHAIN, str(verifier)], input=json.dumps({"data": data}), + capture_output=True, text=True, timeout=120) + assert done.returncode == 0, done.stderr[-2000:] + assert json.loads(done.stdout) == [{"node": "sub/write", "result": "pass", "reason": ""}] + # 这次运行没有出具判定（子图成果节点的那条是子图自己的）：元信息不写档位 + assert data["meta"]["tier"] is None, data["meta"]["tier"] + + +@pytest.mark.parametrize("namesake", [False, True]) +async def test_the_bundle_and_whatif_read_a_subgraph_card_in_its_own_layer(client, monkeypatch, namesake): + """子图里一张卡转抄同一张子图里的另一张（card1 ← card0），父图报告引用 [[m:g1]]，全程没设遮罩：卡上记的是子图里的 id + （nodes.card0…），可验证报告、推演都按卡所在的那一层找产出——找得到，不当成「核对不了」整张遮住；父图里另有一个同名的 + card0（namesake）时，也不拿它当依据（推演还原转抄来的值，还原成子图那张的 246.5，不是父图那张的 1）。""" + import json + import re + + # card0 另有一个按位置取的格（vars.rows.rows[0][0]，经原样解析查询结果的 parse）：可验证报告、推演都要顺着取值链从 + # parse 那一次的产出走 + sub = S.chain(node("s", "input"), node("fetch", "tool", tool=S.tool(), args={"sql": PAID}), + node("parse", "transform", mode="json", template="{{ nodes.fetch }}", assign_to="rows"), + node("card0", "metrics", caliber="子口径", caliber_version="v1", + metrics=[GMV, metric("first", "vars.rows.rows[0][0]"), + metric("cnt", "len(nodes.parse.rows)")]), + node("card1", "metrics", caliber="转抄", caliber_version="v1", + metrics=[metric("g1", "nodes.card0.metrics[0].value")]), + node("o", "output", fields=[{"name": "gmv", "value": "{{ nodes.card1.metrics[0].value }}"}])) + sub_id = await publish(sub) + parts = [node("start", "input")] + if namesake: + parts += [node("card0", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("one", "1")]), + node("parse", "transform", mode="json", template='{"rows": [[1], [2]]}', assign_to="other")] + parts += [node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1), + node("write", "report", instructions="写答案"), + node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}])] + S.script(monkeypatch, "转抄的口径 [[m:g1]]，第一行 [[m:first]]。") + started = await run_manager.start(graph=S.chain(*parts), input_payload={}) + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + card1 = next(e["artifact"] for e in (await card_entries(row.id)) if e["node_id"] == "sub/card1") + html = (await client.get(f"/api/runs/{row.id}/bundle.html")).text + data = json.loads(re.search(r'', html, re.S).group(1)) + entry = next(a for a in data["artifacts"] if a["id"] == card1) + assert "masked" not in entry and entry.get("b64"), entry + # 推演：转抄的那个输入按子图里 card0 那次的产出还原，什么都不改时值不变、没有推演不了的 + res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "sub/card1"}) + assert res.status_code == 200, res.text + body = res.json() + assert body["unsupported"] == [], body["unsupported"] + [g1] = [m for m in body["metrics"] if m.get("id") == "g1"] + assert g1.get("changed") in (False, None) and g1.get("new") == g1.get("old") == 246.5, g1 + card0 = next(e["artifact"] for e in (await card_entries(row.id)) if e["node_id"] == "sub/card0") + entry = next(a for a in data["artifacts"] if a["id"] == card0) + assert "masked" not in entry and entry.get("b64"), entry + res = await client.post(f"/api/runs/{row.id}/evidence/whatif", json={"card": "sub/card0"}) + assert res.status_code == 200 and res.json()["unsupported"] == [], res.json() + # nodes.parse.rows 这种整份产出的输入从 parse 那一次的产出还原：子图里的 parse（1 行），不是父图里同名的那个（2 行） + [cnt] = [m for m in res.json()["metrics"] if m.get("id") == "cnt"] + assert cnt.get("new") == cnt.get("old") == 1, cnt + + +async def card_entries(run_id: str) -> list[dict]: + from app.api.evidence import _sealed + + return [e for e in (await _sealed(run_id)).entries if e.get("kind") == "metric_set"] + + async def test_masks_inside_the_subgraph_follow_the_parent_run(client, monkeypatch, tmp_path): """数据源遮了 phone：子图里查出来的号码在父图报告的证据面板里照样写「已遮罩」。""" await S.set_source(tmp_path / "shop.db", options={"mask_columns": "phone"})