diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 82f491a0..fb232c6e 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -223,9 +223,13 @@ class _Sealed: 当「已封存」展示——只看 sealed 的消费方会把一条断掉的封存链画成绿的。""" return self.sealed and self.seal["ok"] is True - def nodes_of(self, node_type: str) -> list[str]: - return [str(n.get("id")) for n in self.graph.get("nodes") or [] - if isinstance(n, dict) and n.get("type") == node_type] + def nodes_of(self, node_type: str, *, nested: bool = False) -> list[str]: + """这次运行的图上这一类的节点；nested 为真时连子工作流里跑过的一起给（「子工作流节点 id/…」，类型照 node.started 记的）。""" + out = [str(n.get("id")) for n in self.graph.get("nodes") or [] + if isinstance(n, dict) and n.get("type") == node_type] + if nested: + out += [nid for nid, kind in self.types.items() if "/" in nid and kind == node_type] + return out def payload_of(self, node_ids: list[str]) -> dict[str, Any] | None: """这几个节点里最后跑完的那个的产出，从封存的事件指向的工件里取。""" @@ -1764,7 +1768,8 @@ def _reviews(sealed: _Sealed, report: _Report 审稿记录和计数，按它认（只认审稿记录存上了的），产出拼成 {challenge_artifact, counts, on_refuted: fail}。运行因此 失败了，被推翻的是哪几句、反方凭什么，证据面板里照样看得到。 """ - reviewers = set(sealed.nodes_of("challenge")) + # 子工作流里的反方审稿也认（「sub/chal」）：它审的是子工作流里那份报告（审稿记录按全局 id 记，「sub/write」） + reviewers = set(sealed.nodes_of("challenge", nested=True)) if not reviewers: return spec: GraphSpec | None = None @@ -1843,7 +1848,8 @@ def _challenge_of(sealed: _Sealed, report: _Report, unit_id: str) -> dict[str, A 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, + # 子工作流里的报告（「sub/write」）不在运行的图上：写「子工作流 › 报告」，不写内部 id + "report_label": writer.title if writer else _node_label(sealed, 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 {})} diff --git a/backend/app/engine/nodes/challenge.py b/backend/app/engine/nodes/challenge.py index 4b8f21ce..f44c98ad 100644 --- a/backend/app/engine/nodes/challenge.py +++ b/backend/app/engine/nodes/challenge.py @@ -811,8 +811,10 @@ class _Review: # 没能反查的句数：没派出去、派出去却没查成的（unreviewed 里各有原因），运行时长上限将至没开始的，加上触顶时没能给出 # 结论的。都记为无法判断，可「无法判断 2 句」既可能是查了、数据不够，也可能是根本没查：审批卡要分开说 missed = len(unreviewed) + overdue + sum(skipped for _, skipped in hit.values()) - artifact = await self._store({"report": self.report_id, "doc_artifact": self.doc_artifact, "claims": claims, - "model": model}) + # 报告按这次运行里的 id 记（子工作流里的写「子工作流节点 id/报告」，同 report.checked、报告文档上的 node_id）： + # 证据面板、可验证报告按它认审的是哪份报告 + artifact = await self._store({"report": self.ctx.run.global_id(self.report_id), "doc_artifact": self.doc_artifact, + "claims": claims, "model": model}) for limit, (touched, skipped) in hit.items(): words = self._limit_words(limit) message = f"已达上限（{words}），{skipped} 句未反查" if limit == "daily_max_usd" else \ @@ -872,7 +874,8 @@ class _Review: async def _store(self, record: dict[str, Any]) -> str | None: try: return await artifact_store.put_json(record, kind="challenge", run_id=self.ctx.run.run_id, - node_id=self.ctx.node.id, meta={"report": self.report_id}) + node_id=self.ctx.run.global_id(self.ctx.node.id), + meta={"report": self.ctx.run.global_id(self.report_id)}) except Exception as e: # noqa: BLE001 - 记录写失败不毁掉运行：结论照常处置，只是理由和查询点不开 self.ctx.emit(EventType.LOG, level="warn", code="evidence_store_failed", message=f"反方审稿记录保存失败（{describe_exception(e)}），审稿的理由和查询将无法查看") diff --git a/backend/app/engine/nodes/human.py b/backend/app/engine/nodes/human.py index b112a39f..cf2639ee 100644 --- a/backend/app/engine/nodes/human.py +++ b/backend/app/engine/nodes/human.py @@ -227,7 +227,8 @@ def _reviewed_reports(state: GraphState, ctx: NodeContext) -> list[dict[str, Any or not isinstance(result.get("stats"), dict): continue item: dict[str, Any] = { - "node_id": rid, "label": node.title, + # 按这次运行里的 id 记（子工作流里的写「子工作流节点 id/报告」）：审批卡据此打开证据面板（?report=） + "node_id": ctx.run.global_id(rid), "label": node.title, "stats": {k: v for k, v in result["stats"].items() if isinstance(v, int) and not isinstance(v, bool)}, "claims": result.get("claims") or "off", } diff --git a/backend/tests/test_challenge_node.py b/backend/tests/test_challenge_node.py index 896b1dac..1b8768ea 100644 --- a/backend/tests/test_challenge_node.py +++ b/backend/tests/test_challenge_node.py @@ -1431,3 +1431,56 @@ def test_values_the_agent_found_are_masked_in_the_rationale_when_the_report_fill rationale=doc, usage={}, masked_values=seen.pooled()) masked = challenge_mod._masked_rationale(outcome, secrets, given=[]) assert masked.result["rationale"] == "号码 已遮罩 的客户在华南。" + + +# -------------------------------------------------------------------------- +# 子工作流里的反方审稿 +# -------------------------------------------------------------------------- + + +async def _in_subworkflow(spec: dict) -> dict: + """把 spec 发布成子工作流，父图：开始 → 子工作流 → 成果。""" + from tests import ledger_samples as S + + sub_id = await S.publish("周报子流程", graph=spec, level="draft") + return chain(node("start", "input"), node("sub", "subgraph", "周报子流程", workflow_id=sub_id, workflow_version=1), + node("top", "output", fields=[{"name": "周报", "value": "{{ nodes.sub.output.周报 }}"}])) + + +async def test_a_challenge_inside_a_subworkflow_names_the_report_by_its_run_id(monkeypatch, shop): + """子工作流里的报告在父运行里按全局 id（「sub/write」）记（report.checked、报告文档）：审稿记录也按它记，证据面板点开 + 子图里那份报告的句子时认得出审它的那次审稿，不再因为记的是子图里的 id（write）对不上、什么都不显示。""" + from httpx import ASGITransport, AsyncClient + + from app.main import app + + Models(monkeypatch, shop) + row = await finish(await _in_subworkflow(graph(shop, contract=False)), run_class="exploratory") + assert row.status == "succeeded", row.error + out = await output_of(row.id, "sub/chal") + record = artifact_store.load(out["challenge_artifact"]) + [checked] = await events(row.id, "report.checked", "sub/write") + assert (record["report"], record["doc_artifact"]) == ("sub/write", checked.data["doc_artifact"]) + doc = artifact_store.load(checked.data["doc_artifact"]) + south = unit_of(doc, SOUTH) + seg = next(s for _, u in iter_units(doc) if u["id"] == south + for s in u.get("segments") or [] if s.get("kind") != "structural") + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + res = await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}?report=sub/write") + assert res.status_code == 200, res.text + challenge = res.json()["unit"].get("challenge") or {} + assert (challenge.get("verdict"), challenge.get("node_id")) == ("refuted", "sub/chal") + + +async def test_an_approval_inside_a_subworkflow_points_at_the_report_by_its_run_id(monkeypatch, shop): + """子工作流里「报告 → 反方审稿 → 人工审批」：审批卡上的报告按这次运行里的 id 记（「sub/write」），审批卡据此打开证据面板， + 反方审稿的概况照样带上。""" + Models(monkeypatch, shop) + row = await finish(await _in_subworkflow(gated(shop)), run_class="exploratory", + statuses=("interrupted", "failed", "succeeded")) + assert row.status == "interrupted", row.error + async with SessionLocal() as session: + approval = (await session.execute(select(Approval).where(Approval.run_id == row.id))).scalar_one() + [report] = approval.payload["reports"] + assert report["node_id"] == "sub/write" + assert report["challenge"]["counts"]["refuted"] == 1