mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:11:59 +00:00
复核修正(后端):子图里的报告不再被判「疑似被修改」、驳回出口的双盲裁决照样降档、可验证报告与推演按卡所在的层找产出、子图那一步写名字
- 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(驳回出口接到出具成果照样降档、台账照记、证据面板认得)。每条都去掉修复回跑过,确实失败
This commit is contained in:
+30
-18
@@ -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)
|
||||
|
||||
@@ -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 {}))
|
||||
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 静态地认(只有图)
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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))
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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")})
|
||||
|
||||
@@ -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'<script type="application/json" id="bundle-data">(.*?)</script>', 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'<script type="application/json" id="bundle-data">(.*?)</script>', 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"})
|
||||
|
||||
Reference in New Issue
Block a user