diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 552e7a18..6c42552c 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -209,6 +209,9 @@ class _Sealed: #: 按需裁判的判定(evidence.judged 事件),按先后:(seq, 节点, 载荷)。不管落在封存前后都收—— #: 失败后接着跑的运行会把早先追加的判定封进新的清单里,它仍然是按需裁判的批注 judged: list[tuple[int, str | None, dict[str, Any]]] = field(default_factory=list) + #: 节点 id → node.started 上记的名字、类型。子工作流里的节点(「子工作流节点 id/…」)不在这次运行的图上,名字从这里取 + labels: dict[str, str] = field(default_factory=dict) + types: dict[str, str] = field(default_factory=dict) @property def sealed(self) -> bool: @@ -273,6 +276,11 @@ async def _sealed(run_id: str) -> _Sealed: out.retrievals.setdefault(str(data["artifact"]), {"node_id": node_id, "source": data.get("collection")}) elif etype == "caliber.upgrade": out.upgrades[node_id] = data + elif etype == "node.started": + if isinstance(data.get("label"), str) and data["label"]: + out.labels.setdefault(node_id, data["label"]) + if isinstance(data.get("node_type"), str) and data["node_type"]: + out.types.setdefault(node_id, data["node_type"]) if etype != "node.finished": continue if data.get("artifact"): @@ -280,7 +288,9 @@ async def _sealed(run_id: str) -> _Sealed: for entry in data.get("evidence") or []: if isinstance(entry, dict) and entry.get("artifact"): out.ledger.add((node_id, entry["artifact"])) - out.entries.append({**entry, "node_id": node_id}) + # 条目里另记的节点 id(双盲复算复算的口径卡)是那个节点所在那一层图里的 id:换成这次运行里的 + out.entries.append({**entry, "node_id": node_id, + **({"card": _scoped(node_id, entry["card"])} if entry.get("card") else {})}) artifact = str(entry["artifact"]) if entry.get("kind") == "query": out.queries[artifact] = { @@ -975,20 +985,21 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s # 期 4:这个指标做没做过双盲复算、结果如何。没有双盲复算复核过这份指标集时没有这个键,老运行一字不差 if checked := _dual_check_of(sealed, str(entry.get("node_id") or ""), artifact, metric_id, metric): step["dual_check"] = checked + card_node = str(entry.get("node_id") or "") if metric is not None and metric.get("inputs"): # 输入追没追到原始数据(engine/lineage.py):和「有出处」(确定性复算一致)是两件事,面板分开说。追不到的照实说 # 出处追到哪个节点为止;卡上没有输入(常量)的没有可追的,不给这个键 - step["untraced"] = _untraced_steps(metric, sealed) + step["untraced"] = _untraced_steps(metric, sealed, card_node) catalog = doc.get("catalog") or {} inputs: list[dict[str, Any]] = [] #: 输入指到的查询快照 → 被引用的格;同一份快照只给一个查询步骤,格合在一起高亮 wanted: dict[str, list[tuple[Any, Any]]] = {} held, word = await _metric_masks(metric or {}, str(entry.get("node_id") or ""), artifact, sealed, masks, card=card) for item in (metric or {}).get("inputs") or []: - one = {"step": "input", **item} + one = {"step": "input", **item, **_scoped_ids(item, card_node, sealed)} if item.get("through"): # 穿过原样解析查询结果的整形节点按位置取的格:面板写明经过的是哪个节点(图里的名字) - one["through_label"] = _node_label(sealed, item["through"]) + one["through_label"] = _node_label(sealed, one["through"]) artifact_of = item.get("artifact") if artifact_of and item.get("via") in ("tool_cell", "agent_field"): one["query"] = _query_alias(catalog, artifact_of) @@ -999,7 +1010,8 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s one = _withheld(one, found) inputs.append(one) # 转抄另一张口径卡的指标(对账卡、召回卡的「口径」那一项):跨卡往下追,接上那个指标自己的一段出处,一直到查询格子 - upstream = await _upstream_steps(metric or {}, sealed, masks, catalog, wanted, {(artifact, str(metric_id))}) + upstream = await _upstream_steps(metric or {}, sealed, masks, catalog, wanted, {(artifact, str(metric_id))}, + card_node=card_node) if withheld := {path: _held_word(found) for path, found in held.items() if found.kind != "table"}: # 代入式里那几处同样写「已遮罩」(遮没遮核对不了的写「无法核对」):不然标签里遮着、代入式里亮着,同一个值换个 # 地方照样看得到。整份结果当输入的不换:代入式里本来就只留着它的路径 @@ -1015,7 +1027,7 @@ async def _metric_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[s async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Masks", catalog: dict[str, Any], wanted: dict[str, list[tuple[Any, Any]]], seen: set[tuple[str, str]], - depth: int = 1) -> list[dict[str, Any]]: + depth: int = 1, *, card_node: str = "") -> list[dict[str, Any]]: """指标的输入里转抄另一张口径卡的指标的(nodes.<卡>.metrics[i].value,卡上记着 upstream),每个接一步 upstream:那张 卡的那个指标(名字、值、式子、代入式、核对状态,取值按同一套遮罩),它自己的输入(同指标步骤的输入;落在查询快照上的 并进 wanted,后面照常给查询步骤),追没追到原始数据。转抄的转抄一层层接着追(防环:同一个指标只接一次;最多 @@ -1033,7 +1045,8 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask if (artifact, metric_id) in seen: continue seen.add((artifact, metric_id)) - node = str(item.get("node_id") or "") + # 卡上记的是卡所在那一层图里的 id(子工作流里的卡转抄同一张子图里的卡;经子工作流出口转抄的已经是这一层的) + node = str(_scoped(card_node, item.get("node_id")) or "") card, hash_ok = _load_card(artifact) found = _metric_in(card, metric_id) step: dict[str, Any] = { @@ -1060,7 +1073,7 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask if found.get("incomplete") else {}), **({"sql_check_failed": True, "sql_check_reason": found.get("sql_check_reason") or ""} if found.get("sql_check_failed") else {}), - "untraced": _untraced_steps(found, sealed), + "untraced": _untraced_steps(found, sealed, node), }) if withheld := {path: _held_word(h) for path, h in held.items() if h.kind != "table"}: step["substituted"] = recompute.substituted_without(found, withheld) @@ -1068,7 +1081,7 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask for sub in found.get("inputs") or []: if not isinstance(sub, dict): continue - one = {"step": "input", **sub} + one = {"step": "input", **sub, **_scoped_ids(sub, node, sealed)} if sub.get("artifact") and sub.get("via") in SNAPSHOT_VIAS: one["query"] = _query_alias(catalog, sub["artifact"]) if cell := _input_cell(sub, catalog): @@ -1079,27 +1092,60 @@ async def _upstream_steps(metric: dict[str, Any], sealed: _Sealed, masks: "_Mask inputs.append(one) step["inputs"] = inputs out.append(step) - out += await _upstream_steps(found, sealed, masks, catalog, wanted, seen, depth + 1) + out += await _upstream_steps(found, sealed, masks, catalog, wanted, seen, depth + 1, card_node=node) + return out + + +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}" + + +def _scoped_ids(item: dict[str, Any], card_node: str, sealed: _Sealed) -> dict[str, Any]: + """输入步骤上的节点 id 换成这次运行里的(_scoped);落在子工作流里的另给图上的名字(node_label、subgraph_label: + 「子工作流 › 节点」),画布、记录页认不出这个 id 时照它写。最外层的输入不多带键,和以前一字不差。""" + out: dict[str, Any] = {} + for key in ("node_id", "through", "subgraph"): + if isinstance(item.get(key), str) and item[key]: + out[key] = _scoped(card_node, item[key]) + if "/" in str(out.get("node_id") or ""): + out["node_label"] = _node_label(sealed, out["node_id"]) + if out.get("subgraph"): + out["subgraph_label"] = _node_label(sealed, out["subgraph"]) return out def _node_label(sealed: _Sealed, node_id: Any) -> str: - """封存的图里这个节点的名字,没有名字的写 id。""" + """封存的图里这个节点的名字,没有名字的写 id。子工作流里的节点(「子工作流节点 id/…」)不在图上,按 node.started 记的 + 名字一层层接上:「子工作流 › 节点」。""" node = next((n for n in (sealed.graph or {}).get("nodes") or [] if isinstance(n, dict) and n.get("id") == node_id), {}) - return str((node.get("data") or {}).get("label") or node_id or "") + if node or not isinstance(node_id, str) or "/" not in node_id: + return str((node.get("data") or {}).get("label") or node_id or "") + parts = node_id.split("/") + names = [_node_label(sealed, parts[0])] + for i in range(2, len(parts) + 1): + here = "/".join(parts[:i]) + names.append(sealed.labels.get(here) or parts[i - 1]) + return " › ".join(names) -def _untraced_steps(metric: dict[str, Any], sealed: _Sealed) -> list[dict[str, Any]]: +def _untraced_steps(metric: dict[str, Any], sealed: _Sealed, card_node: str = "") -> list[dict[str, Any]]: """指标的输入里追不到原始数据的:[{path, node_id, label, type}],label 是封存的图里那个节点的名字、type 是节点类型 - (面板按它写「数据整形」「沙箱代码」)。认不出是哪个节点的,node_id、label 为 None。""" + (面板按它写「数据整形」「沙箱代码」)。认不出是哪个节点的,node_id、label 为 None。节点 id 按卡所在那一层认 + (_scoped)。""" nodes = {n.get("id"): n for n in (sealed.graph or {}).get("nodes") or [] if isinstance(n, dict)} out = [] for item in untraced(metric.get("inputs") or []): - node_id = item.get("node_id") + node_id = _scoped(card_node, item.get("node_id")) node = nodes.get(node_id) or {} out.append({"path": item.get("path"), "node_id": node_id, "label": _node_label(sealed, node_id) if node_id else None, - "type": node.get("type") or None}) + "type": node.get("type") or sealed.types.get(str(node_id or "")) or None}) return out @@ -1495,6 +1541,8 @@ class _InputMasks: return None, True def _output(self, node_id: Any) -> Any: + # 卡的取值链里写的是卡所在那一层图里的 id(子工作流里的卡写子图里的 id),换成这次运行里的再找 + node_id = _scoped(self._card, node_id) if self._given is not None: return self._given(node_id) if self._outputs is None: diff --git a/backend/app/engine/lineage.py b/backend/app/engine/lineage.py index a7eec4fd..93630579 100644 --- a/backend/app/engine/lineage.py +++ b/backend/app/engine/lineage.py @@ -13,6 +13,16 @@ - 应用检查(apps/lint)、受管级发布检查(governance):按 static_stops 静态地认(只有图、没有运行),追不到的给提示。 转抄同一张图里另一张口径卡的指标,和运行时一样跨卡按那个指标的式子接着认(最多 MAX_COPY_DEPTH 层);被抄的卡按 caliber_from 固定在别处的版本时,静态看不到那一版的图,照旧停在那张卡。 + +读子工作流出口的字段(nodes.<子图>.output.<字段>,docs/design/orchestration-round/00-总方案.md §2)跨子图往下追,三处同一套: +- 运行时:子图跑完时按出口模板认每个字段的出处(multi.run_subgraph 的 provenance:模板整串就是一个取值链的字段,在子图 + 的状态里按口径卡认输入的同一套认;出自子图运行输入的再按子工作流节点的「输入」模板往父图追一层),父图口径卡读这个 + 字段时接上它(metrics._subgraph_field),卡上记的就是子图里那一格的出处(via、artifact、locator、status),另记 + subgraph、field——traced / untraced 照常认;子图里追不到原始数据的,父图照样追不到、照样降档; +- 证据面板:卡上记的节点 id 按卡所在的那一层认(api/evidence 的 _scoped),一路给到子图里的那一格; +- 静态:按子工作流节点固定的那一版图(lint_process.subgraph_spec,调用方用 with_subgraphs 放进来)认出口字段的模板,接着 + 在子图里认;子图的运行输入按子工作流节点的「输入」模板回到父图认。最多 MAX_SUB_DEPTH 层(和运行时的嵌套上限一样), + 看不到子图的(没放进来、读不到)照旧停在子工作流节点。 """ from __future__ import annotations @@ -140,20 +150,52 @@ class Stop: node_id: str | None +#: 出口字段的模板整串就是一个取值链:{{ nodes.card.metrics[0].value }}(不带过滤器、不调函数、前后不带别的字) +_WHOLE_REF = re.compile(r"^\s*\{\{\s*((?:nodes|vars|input)\.[^{}|()]+?)\s*\}\}\s*$", re.S) + + +def whole_ref(template: Any) -> str | None: + """模板整串就是一个取值链时交回它(成果字段的值就是那个值本身,渲染只是转成文字);别的写法为 None:拼了别的字、 + 过了过滤器的值,对不上那一格的值。子工作流出口的出处,运行时(multi._output_provenance)和静态(_exit_template) + 都按这一个判断认。""" + m = _WHOLE_REF.match(template) if isinstance(template, str) else None + return m.group(1).strip() if m else None + + +#: 静态地跨子工作流往下追的层数上限:和运行时子工作流的嵌套上限(multi._MAX_DEPTH)一样 +MAX_SUB_DEPTH = 3 + + +@dataclass(frozen=True) +class _Scope: + """静态地认时所在的那一层图:节点 id → {type, title?, config}。是子工作流时记着父图那一层和它在父图里的节点 id(子图的 + 运行输入按那个子工作流节点的「输入」模板回到父图认)。""" + + nodes: Mapping[str, Mapping[str, Any]] + parent: "_Scope | None" = None + via: str | None = None + level: int = 0 + + def static_stops(nodes: Mapping[str, Mapping[str, Any]], expression: str, *, _depth: int = 0) -> list[Stop]: """静态地认一个指标表达式里追不到原始数据的输入。nodes 是节点 id → {type, config}(apps/tree.node_map 的形状)。 和运行时同一套认法,只是不看值:cell() 的第一个参数是查询结果(调用工具、合并查询的产出,或原样解析它的整形节点)、 按位置取的格穿过原样解析的整形节点、运行输入、开了按出处核对字段的 Agent 交来的字段,算追到了;转抄同一张图里 - 另一张口径卡的指标,按那个指标的式子接着认(它的输入都追到了才算);别的(表达式整形、沙箱代码、模型调用、转抄 - 固定在别处版本的口径卡……)停在写它的那个节点。表达式写错了的不在这里报(画布的校验报)。""" + 另一张口径卡的指标,按那个指标的式子接着认(它的输入都追到了才算);读子工作流出口的字段,按那一版子图的出口模板 + 接着认(_through_sub);别的(表达式整形、沙箱代码、模型调用、转抄固定在别处版本的口径卡、看不到的子图……)停在 + 写它的那个节点。表达式写错了的不在这里报(画布的校验报)。""" + return _stops(_Scope(nodes), expression, _depth) + + +def _stops(scope: _Scope, expression: str, depth: int) -> list[Stop]: try: tree = parse_expression(expression)[0] except ExpressionError: return [] out: list[Stop] = [] for path in leaf_refs(tree): - stop = _static_stop(nodes, path, _depth) + stop = _static_stop(scope, path, depth) if stop is not None: out.append(stop) return out @@ -179,16 +221,17 @@ def _copied_metric(nodes: Mapping[str, Mapping[str, Any]], path: str) -> tuple[s return writers[0], str(metrics[index]["expression"]) -def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int = 0) -> Stop | None: +def _static_stop(scope: _Scope, path: str, depth: int = 0) -> Stop | None: + nodes = scope.nodes type_of = lambda node_id: (nodes.get(node_id) or {}).get("type") # noqa: E731 if (copied := _copied_metric(nodes, path)) is not None: card_id, expression = copied - if depth < MAX_COPY_DEPTH and not static_stops(nodes, expression, _depth=depth + 1): + if depth < MAX_COPY_DEPTH and not _stops(scope, expression, depth + 1): return None return Stop(path, card_id) if (parts := cell_parts(path)) is not None: writers = _writers(nodes, parts[0]) - bad = [w for w in writers if not _table(nodes, w, type_of)] + bad = [w for w in writers if not _table(nodes, w, type_of) and not _sub_table(scope, w, parts[0])] return Stop(path, bad[0]) if bad or not writers else None if (at := row_cell(path)) is not None: writers = _writers(nodes, at.table) @@ -197,7 +240,7 @@ def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int = return None head = _REF_HEAD.match(path) if head and head.group(1) == "input": - return None + return None if _from_parent(scope, head.group(2), depth) else Stop(path, scope.via) writers = _writers(nodes, path) if not writers: return Stop(path, None) @@ -205,13 +248,84 @@ def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int = for writer in writers: node = nodes[writer] if node.get("type") == "input": + if head and head.group(1) == "vars" and not _from_parent(scope, head.group(2), depth): + return Stop(path, scope.via) continue if node.get("type") == "agent" and _cited(node, head.group(1) if head else "", field): continue + if node.get("type") == "subgraph" and head and _through_sub(scope, writer, head, depth): + continue return Stop(path, writer) return None +def _from_parent(scope: _Scope, name: str, depth: int) -> bool: + """运行输入的一个字段算不算原始数据:最外层的是(发起运行的人填的);子工作流的运行输入是父图按子工作流节点的「输入」 + 模板渲染出来的,那个模板整串就是一个取值链、在父图里追得到原始数据,才算(同运行时 multi._lifted)。""" + if scope.parent is None: + return True + templates = ((scope.parent.nodes.get(scope.via or "") or {}).get("config") or {}).get("input") + path = whole_ref(templates.get(name)) if isinstance(templates, dict) and name else None + return path is not None and _static_stop(scope.parent, path, depth + 1) is None + + +def _sub_scope(scope: _Scope, node_id: str) -> _Scope | None: + """子工作流节点 node_id 引用的那一版图(lint_process.subgraph_spec 读好放进来的),看不到、超过层数上限的为 None。""" + if scope.level >= MAX_SUB_DEPTH: + return None + from app.engine.lint_process import subgraph_spec + + sub = subgraph_spec((scope.nodes.get(node_id) or {}).get("config") or {}) + if sub is None: + return None + nodes = {n.id: {"type": str(n.type), "title": n.title, "config": n.config} for n in sub.nodes} + return _Scope(nodes, parent=scope, via=node_id, level=scope.level + 1) + + +def _exit_template(sub: _Scope, key: str) -> str | None: + """子图出口字段 key 的模板整串所写的取值链(whole_ref);没有这个字段、几个成果节点写法不一样、不是整串一个 + 取值链的为 None。""" + found = {whole_ref(f.get("value", "")) for node in sub.nodes.values() if node.get("type") == "output" + for f in (node.get("config") or {}).get("fields") or [] if isinstance(f, dict) and f.get("name") == key} + return found.pop() if len(found) == 1 else None + + +def _sub_key(head: re.Match[str], node: Mapping[str, Any]) -> str | None: + """nodes.<子图>.output.<字段> / vars.<子图的 assign_to>.<字段> 读的出口字段名;不是正好一个字段的为 None。""" + root, name, rest = head.group(1), head.group(2), head.group(3) + key = re.sub(r"\[\s*['\"]([^'\"\]]+)['\"]\s*\]", r".\1", rest).strip(".") + if root == "nodes": + if not key.startswith("output."): + return None + key = key[len("output."):] + elif root != "vars" or str((node.get("config") or {}).get("assign_to") or "").strip() != name: + return None + return key if key and "." not in key and "[" not in key else None + + +def _through_sub(scope: _Scope, writer: str, head: re.Match[str], depth: int) -> bool: + """读子工作流出口的一个字段:那个字段在子图里追得到原始数据(出口模板整串就是一个取值链,接着在子图里认)。""" + key = _sub_key(head, scope.nodes.get(writer) or {}) + sub = _sub_scope(scope, writer) if key else None + inner = _exit_template(sub, key) if sub is not None and key else None + return inner is not None and _static_stop(sub, inner, depth + 1) is None + + +def _sub_table(scope: _Scope, writer: str, source: str) -> bool: + """cell() 的第一个参数是子工作流出口原样交出的一份查询结果(出口字段的模板整串就是 {{ nodes.<查询节点> }})。""" + head = _REF_HEAD.match(source) + if not head or (scope.nodes.get(writer) or {}).get("type") != "subgraph": + return False + key = _sub_key(head, scope.nodes.get(writer) or {}) + sub = _sub_scope(scope, writer) if key else None + inner = _exit_template(sub, key) if sub is not None and key else None + if inner is None: + return False + writers = _writers(sub.nodes, inner) + type_of = lambda node_id: (sub.nodes.get(node_id) or {}).get("type") # noqa: E731 + return bool(writers) and all(_table(sub.nodes, w, type_of) or _sub_table(sub, w, inner) for w in writers) + + def _writers(nodes: Mapping[str, Mapping[str, Any]], source: str) -> list[str]: """取值链的根是谁写的:nodes.X 是 X;vars.X 是 assign_to 为 X 的节点(入口节点声明了字段 X 的也算)。""" head = _REF_HEAD.match(source) diff --git a/backend/app/engine/lint_process.py b/backend/app/engine/lint_process.py index e7ccac9d..58cb9f45 100644 --- a/backend/app/engine/lint_process.py +++ b/backend/app/engine/lint_process.py @@ -147,9 +147,17 @@ class Subgraphs(dict): self.pin_problems: list[Any] = [] +#: 嵌套的子工作流往下读几层:和运行时的嵌套上限(multi._MAX_DEPTH)一样,再深的运行时本来就拒 +_LOAD_DEPTH = 3 + + async def load_subgraphs(session: Any, spec: GraphSpec) -> Subgraphs: - """读出图里子工作流节点引用的图(固定了版本的读那一版,没固定的读工作流现在的图)。读不到、不成形的跳过。 - 固定的版本能不能用也一并读好(Subgraphs.pin_problems)。""" + """读出图里子工作流节点引用的图(固定了版本的读那一版,没固定的读工作流现在的图),子图里再嵌套的一层层往下读(最多 + _LOAD_DEPTH 层,同一张图只读一次,互相引用的不会转圈)。读不到、不成形的跳过。固定的版本能不能用也一并读好 + (Subgraphs.pin_problems,只看这张图自己的子工作流节点)。 + + 拿去用的:人工节点在子工作流里(_subgraph_humans)、口径卡的输入跨子工作流追到原始数据(engine/lineage 的静态认法, + subgraph_spec)。""" from sqlalchemy import select from app.apps.versions import subgraph_pin_problems @@ -157,24 +165,39 @@ async def load_subgraphs(session: Any, spec: GraphSpec) -> Subgraphs: out = Subgraphs() out.pin_problems = await subgraph_pin_problems(session, spec) - for node in spec.nodes: - if node.type != NodeType.SUBGRAPH: - continue - wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version") - version = int(pinned) if str(pinned or "").isdigit() else None - if not wid or (str(wid), version) in out: - continue - query = (select(WorkflowVersion.graph).where(WorkflowVersion.workflow_id == str(wid), - WorkflowVersion.version == version) - if version is not None else select(Workflow.graph).where(Workflow.id == str(wid))) - graph = (await session.execute(query)).scalar_one_or_none() - try: - out[(str(wid), version)] = GraphSpec.model_validate(graph) - except Exception: # noqa: BLE001 - 读不到、不成形:子工作流节点运行时自己会说清楚 - continue + todo = [(spec, 0)] + while todo: + current, level = todo.pop(0) + for node in current.nodes: + if node.type != NodeType.SUBGRAPH: + continue + wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version") + version = int(pinned) if str(pinned or "").isdigit() else None + if not wid or (str(wid), version) in out: + continue + query = (select(WorkflowVersion.graph).where(WorkflowVersion.workflow_id == str(wid), + WorkflowVersion.version == version) + if version is not None else select(Workflow.graph).where(Workflow.id == str(wid))) + graph = (await session.execute(query)).scalar_one_or_none() + try: + out[(str(wid), version)] = GraphSpec.model_validate(graph) + except Exception: # noqa: BLE001 - 读不到、不成形:子工作流节点运行时自己会说清楚 + continue + if level + 1 < _LOAD_DEPTH: + todo.append((out[(str(wid), version)], level + 1)) return out +def subgraph_spec(config: Mapping[str, Any]) -> GraphSpec | None: + """子工作流节点(配置 config)引用的那一版图:load_subgraphs 读好、with_subgraphs 放进来的;没放进来、没读到的为 None。""" + graphs = _SUBGRAPHS.get() + if graphs is None: + return None + wid, pinned = config.get("workflow_id"), config.get("workflow_version") + version = int(pinned) if str(pinned or "").isdigit() else None + return graphs.get((str(wid or ""), version)) + + @contextlib.contextmanager def with_subgraphs(graphs: Mapping[tuple[str, int | None], GraphSpec]) -> Iterator[None]: token = _SUBGRAPHS.set(graphs) @@ -222,12 +245,7 @@ def _agent_tool_sets(node: GraphNode) -> list[tuple[str, Mapping[str, Any], list def _subgraph_humans(spec: GraphSpec, node: GraphNode) -> list[str] | None: - graphs = _SUBGRAPHS.get() - if graphs is None: - return None - wid, pinned = node.config.get("workflow_id"), node.config.get("workflow_version") - version = int(pinned) if str(pinned or "").isdigit() else None - sub = graphs.get((str(wid or ""), version)) + sub = subgraph_spec(node.config) if sub is None: return None return [n.title for n in sub.nodes if n.type == NodeType.HUMAN] diff --git a/backend/app/engine/nodes/io.py b/backend/app/engine/nodes/io.py index a1b956af..ff53d3ca 100644 --- a/backend/app/engine/nodes/io.py +++ b/backend/app/engine/nodes/io.py @@ -351,8 +351,10 @@ def _untraced_gaps(metrics: list[dict[str, Any]], ctx: NodeContext) -> list[str] stops = untraced(m.get("inputs") or []) if not stops: continue - where = "、".join(dict.fromkeys(f"{s.get('path')}(停在「{titles.get(s.get('node_id') or '', s.get('node_id') or '未知来源')}」)" - for s in stops[:3])) + # 停在子工作流里的(节点 id「子工作流节点 id/…」)说停在那个子工作流:这张图上看得到的是它,静态检查也这么说 + where = "、".join(dict.fromkeys( + f"{s.get('path')}(停在「{titles.get(at, at or '未知来源')}」)" + for s in stops[:3] for at in [str(s.get('node_id') or '').split('/', 1)[0]])) out.append(f"指标「{m.get('name') or m.get('id')}」的输入没追到原始数据:{where}") return out diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 740d545b..85a9aa6d 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -624,6 +624,9 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext, - vars.q.rows[0][0],q 由「只是把调用工具节点的结果原样按 JSON 解析」的整形节点写:同 cell(),via tool_cell, 另记 through(穿过的整形节点) - agent 开了 cite_fields 的字段:via agent_field,接上它的 data_evidence(eid、核对状态) + + 读子工作流出口的一个字段(nodes.<子图>.output.<字段>、vars.<子图的 assign_to>.<字段>):接上子图记下的那个字段的出处 + (_subgraph_field),via 等照子图里认出的写,另记 subgraph、field;认不出的照旧记 via subgraph,停在子工作流。 """ if (parts := cell_parts(path)) is not None: return _cell_input(path, value, parts, state, ctx, tctx or template_context(state)) @@ -644,6 +647,9 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext, else: node_id = _writer_of(name, state, ctx) producer = nodes.get(node_id) if node_id else None + if producer is not None and str(producer.type) == "subgraph" and head \ + and (linked := _subgraph_field(path, value, root, field, producer.id, state)) is not None: + return linked entry: dict[str, Any] = { "path": path, "value": value if value is None or isinstance(value, (int, float, str, bool)) else _brief(value), @@ -667,6 +673,32 @@ def _input_of(path: str, value: Any, state: GraphState, ctx: NodeContext, return entry +def _subgraph_field(path: str, value: Any, root: str, field: str, node_id: str, + state: GraphState) -> dict[str, Any] | None: + """读子工作流出口的一个字段:接上子图记下的那个字段的出处(multi.run_subgraph 的 provenance,节点 id 已经是这一层图里 + 的「子工作流节点 id/…」)。只认整个字段(nodes.<子图>.output.<字段>、vars..<字段>,不再往里取),而且这次 + 读到的值就是子图交出的那个;别的为 None,照一般的来历记(via subgraph,停在子工作流)。 + + 接上的出处照子图里认出的写(via、node_id、artifact、locator、eid、status、upstream),path、value 换成这里读的,另记 + subgraph(经过的子工作流节点)、field(出口字段):engine/lineage.traced 照常认,证据面板照常一路给到子图里的那一格。""" + result = (state.get("nodes") or {}).get(node_id) + if not isinstance(result, dict) or not isinstance(result.get("provenance"), dict): + return None + key = _field_key(field) + if root == "nodes": + if not key.startswith("output."): + return None + key = key[len("output."):] + elif root != "vars": + return None + output = result.get("output") if isinstance(result.get("output"), dict) else {} + item = result["provenance"].get(key) + if not key or not isinstance(item, dict) or key not in output or output[key] != value: + return None + return {**item, "path": path, "value": value if value is None or isinstance(value, (int, float, str, bool)) + else _brief(value), "subgraph": node_id, "field": key} + + def _copied_card(path: str, tctx: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]] | None: """转抄另一张口径卡一个指标的输入(engine/lineage.metric_copy 认的写法):(那张卡这一次的产出, 那个指标)。不是这种 写法、产出不是指标集、下标越界的为 None。""" @@ -702,9 +734,17 @@ def _upstream(path: str, tctx: dict[str, Any]) -> dict[str, Any] | None: def _copied_metric(item: dict[str, Any], tctx: dict[str, Any]) -> dict[str, Any] | None: - """卡上的一个输入转抄了另一张口径卡的哪个指标(这一次的产出里那一项);不是转抄的为 None。""" + """卡上的一个输入转抄了另一张口径卡的哪个指标(这一次的产出里那一项);不是转抄的为 None。 + + 经子工作流出口转抄的(subgraph,子图出口字段就是子图里一张卡的指标):那张卡以「子工作流节点 id/卡」进了这一层的 + nodes(multi.run_subgraph),按 node_id 和 upstream 记的指标 id 找。""" if item.get("via") != "metric" or not isinstance(item.get("upstream"), dict): return None + if item.get("subgraph"): + card = (tctx.get("nodes") or {}).get(str(item.get("node_id") or "")) + metrics = card.get("metrics") if isinstance(card, dict) and card.get("kind") == "metric_set" else None + wanted = item["upstream"].get("metric") + return next((m for m in metrics or [] if isinstance(m, dict) and m.get("id") == wanted), None) found = _copied_card(str(item.get("path") or ""), tctx) return found[1] if found is not None else None diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index c83efada..80b20064 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -14,7 +14,8 @@ from app.core.errors import describe_exception, raw_detail from app.core.events import EventType from app.db.base import SessionLocal from app.db.models import Workflow -from app.engine.context import NodeContext, NodeError +from app.engine.context import PREFIX_KEY, STREAM_KEY, NodeContext, NodeError, stream_writer +from app.engine.lineage import whole_ref from app.engine.state import GraphState, message_text, template_context from app.engine.toolcalls import ( TOOL_MARKUP_NUDGE, @@ -616,8 +617,17 @@ async def run_supervisor(state: GraphState, ctx: NodeContext) -> dict[str, Any]: async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: """嵌套执行另一个工作流。 - 子图共享同一个 run 的事件流(节点 id 会加前缀),所以在时间线上能看到它内部 - 每一步;但它有独立的状态,不会污染父图的变量池。 + 子图共享同一个 run 的事件流:节点 id 加前缀「子工作流节点 id/」(嵌套的一层层接上),事件经父运行交下来的写入口发 + (engine/context.STREAM_KEY;以前只设了前缀没用上,子图里每一步在运行的事件里一条都没有),时间线上看得到它内部每一步, + 封存也就覆盖了它查过的数。它有独立的状态,不会污染父图的变量池。 + + 证据交回父图(docs/design/orchestration-round/00-总方案.md §2),三样: + - 台账:子图记下的条目(查询快照、口径卡的指标集……)节点 id 加上前缀并进父图的 evidence,父图报告撰写的目录据此收进 + 子图里的查询和口径卡(report.report_catalog),cell(nodes.<子图>.output.<原样交出的查询结果>, …) 也按台账认得出快照; + - 口径卡:子图里口径卡的产出以「子工作流节点 id/卡」为 id 进父图的 nodes(目录要卡上的指标值;父图转抄它的指标时认得 + 出那张卡、带上不完整和存疑); + - 出处:出口的每个字段按出口模板认它来自哪里(_output_provenance),记在产出的 provenance 上;父图里读 + nodes.<子图>.output.<字段> 的口径卡据此接上子图里的那一格(metrics._subgraph_field),engine/lineage 同一套认法。 """ from app.engine.compiler import compile_graph # 延迟导入,避免循环依赖 from app.engine.schema import GraphSpec, loop_steps @@ -678,9 +688,10 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: approval_default=ctx.run.approval_default, tool_trust=ctx.run.tool_trust, agent_limits=ctx.run.agent_limits, - # 恢复重放的标记和签批人只属于父图的节点,不能漏进子图(节点 id 可能重名) + # 恢复重放的标记和签批人只属于父图的节点,不能漏进子图(节点 id 可能重名)。前缀和写入口见 engine/context: + # 子图里的节点经父运行的写入口发事件、节点 id 带前缀;嵌套的子图拿到的是同一个写入口、更长的前缀 extra={**{k: v for k, v in ctx.run.extra.items() if k not in ("resumed", "actors")}, - "node_prefix": f"{ctx.node.id}/"}, + PREFIX_KEY: f"{ctx.run.global_id(ctx.node.id)}/", STREAM_KEY: stream_writer(ctx.run)}, ) ctx.emit(EventType.LOG, level="info", @@ -711,6 +722,13 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: f"子工作流「{workflow.name}」已执行 {limit} 步仍未结束,已中止。可能存在没有轮数上限的环路," "请在子工作流中用「循环」节点限制轮数", ) from e + except NodeError as e: + # 子图里的节点报的错(含被驳回):节点 id 换成这一层图里的 id(「子工作流节点 id/…」),名字接上子工作流的名字。 + # 以前带的是子图里的 id,运行记下的出错节点会和父图里同名的节点混在一起 + titles = sub_spec.node_map() + inner = getattr(e, "label", None) or (titles[e.node_id].title if e.node_id in titles else e.node_id) + e.node_id, e.label = f"{ctx.node.id}/{e.node_id}", f"{ctx.node.title} › {inner}" + raise output = result_state.get("output") or {} result = { @@ -721,13 +739,94 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: "pinned": bool(pinned), "text": json.dumps(output, ensure_ascii=False) if output else "", } + if provenance := _output_provenance(sub_spec, sub_run, result_state, output, state, ctx): + # 出口每个字段的出处(没有能认的字段时不带这个键,产出和以前一字不差) + result["provenance"] = provenance ctx.emit(EventType.LOG, level="info", message=f"子工作流「{workflow.name}」完成") + sub_nodes = result_state.get("nodes") or {} + cards = {f"{ctx.node.id}/{nid}": payload for nid, payload in sub_nodes.items() + if isinstance(payload, dict) and payload.get("kind") == "metric_set"} updates: dict[str, Any] = { - "nodes": {ctx.node.id: result}, + "nodes": {ctx.node.id: result, **cards}, "usage": result_state.get("usage") or {}, } + if ledger := [_forwarded(e, ctx.node.id) for e in result_state.get("evidence") or [] if isinstance(e, dict)]: + updates["evidence"] = ledger var_name = ctx.cfg("assign_to", "") if var_name: updates["vars"] = {var_name: output} return updates + + +#: 台账条目里写着节点 id 的键:交回父图时一并加前缀(双盲复算的条目记着它复算的口径卡) +_LEDGER_NODE_KEYS = ("node_id", "card") + + +def _forwarded(entry: dict[str, Any], node_id: str) -> dict[str, Any]: + """子图台账里的一条交回父图:节点 id 加上「子工作流节点 id/」前缀,别的照抄。""" + out = dict(entry) + for key in _LEDGER_NODE_KEYS: + if isinstance(out.get(key), str) and out[key]: + out[key] = f"{node_id}/{out[key]}" + return out + + +def _output_provenance(sub_spec: Any, sub_run: Any, sub_state: dict[str, Any], output: dict[str, Any], + state: GraphState, ctx: NodeContext) -> dict[str, dict[str, Any]]: + """子图出口每个字段的出处:{字段: 口径卡上一个输入的样子(via、node_id、artifact、locator、status、upstream……)}。 + + 认法和口径卡认它的输入同一个(metrics._input_of),在子图的状态里认:字段的模板整串就是一个取值链(whole_ref),而且 + 渲染出来的值正是那个值转成的文字,才算这个字段就是那个值;拼了别的字、过了过滤器的不认(不在结果里)。几个成果节点 + 写同一个字段时按最后跑的那个。认出来的节点 id 换成父图这一层的(加「子工作流节点 id/」前缀)。 + + 出自子图运行输入的(via input)再往父图追一层:子图的输入是父图按子工作流节点的「输入」模板渲染出来的,那个模板整串 + 也是一个取值链时,按父图的状态接着认(追到父图的运行输入、查询格……);否则记 via subgraph_input,停在子工作流的 + 入口——子图的运行输入对父图不是原始数据。""" + from app.engine.expressions import _stringify + from app.engine.nodes.metrics import _input_of, _value_of + from app.engine.schema import NodeType + + outputs = {n.id: n for n in sub_spec.nodes if n.type == NodeType.OUTPUT} + fields: dict[str, tuple[Any, str]] = {} + for step in sub_state.get("trail") or []: + exit_node = outputs.get(step.get("node_id")) if "error" not in step else None + for field in (exit_node.config.get("fields") or []) if exit_node is not None else []: + if isinstance(field, dict) and field.get("name"): + fields[str(field["name"])] = (exit_node, field.get("value", "")) + tctx = template_context(sub_state) + out: dict[str, dict[str, Any]] = {} + for name, (exit_node, template) in fields.items(): + path = whole_ref(template) + if path is None or name not in output: + continue + value = _value_of(path, tctx) + if _stringify(value) != output[name]: + continue + item = _input_of(path, value, sub_state, NodeContext(node=exit_node, run=sub_run), tctx) + out[name] = _lifted(item, state, ctx) + return out + + +def _lifted(item: dict[str, Any], state: GraphState, ctx: NodeContext) -> dict[str, Any]: + """子图里认出的一个出处换到父图这一层:节点 id 加前缀;出自子图运行输入的,按子工作流节点的「输入」模板往父图追。""" + from app.engine.expressions import _stringify + from app.engine.nodes.metrics import _REF_HEAD, _input_of, _value_of + + if item.get("via") == "input": + head = _REF_HEAD.match(str(item.get("path") or "")) + name = head.group(2) if head else "" + templates = ctx.cfg("input", {}) or {} + path = whole_ref(templates.get(name)) if isinstance(templates, dict) and name else None + if path is not None: + parent = template_context(state) + value = _value_of(path, parent) + if _stringify(value) == _stringify(item.get("value")): + return _input_of(path, value, state, ctx, parent) + return {"path": item.get("path"), "value": item.get("value"), "node_id": ctx.node.id, "via": "subgraph_input", + **({"field": name} if name else {}), "status": item.get("status") or "ok"} + out = dict(item) + for key in ("node_id", "through", "subgraph"): + if isinstance(out.get(key), str) and out[key]: + out[key] = f"{ctx.node.id}/{out[key]}" + return out diff --git a/backend/app/engine/nodes/report.py b/backend/app/engine/nodes/report.py index 5a33171f..b2596b6f 100644 --- a/backend/app/engine/nodes/report.py +++ b/backend/app/engine/nodes/report.py @@ -37,6 +37,7 @@ import asyncio import re import time from collections import Counter +from collections.abc import Iterable from typing import Any from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, SystemMessage @@ -140,16 +141,18 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[ if isinstance(sources, str) and sources != "ancestors": sources = [sources] if isinstance(sources, list): - # evidence_from 只管查询、检索这类证据(连同查询当时的表结构)从哪几个节点来;指标从哪来由 metrics_from 管 + # evidence_from 只管查询、检索这类证据(连同查询当时的表结构)从哪几个节点来;指标从哪来由 metrics_from 管。 + # 写了子工作流节点的,它交回来的(子图里的查询,节点 id「子工作流节点 id/…」)一并算 keep = {str(s) for s in sources} ledger = [e for e in ledger - if e.get("kind") not in ("query", "retrieval", "schema") or e.get("node_id") in keep] + if e.get("kind") not in ("query", "retrieval", "schema") or _outer(e.get("node_id")) in keep] + nodes = state.get("nodes") or {} catalog = build_catalog( - nodes=state.get("nodes") or {}, + nodes=nodes, ledger=ledger, inputs=state.get("input") or {}, metrics_from=_setting(spec, node, "metrics_from") or None, - allowed=_ancestors(spec, node.id), + allowed=with_subgraphs(_ancestors(spec, node.id), [*nodes, *(e.get("node_id") for e in ledger)]), ) if report_entities(spec, node) == "off": # 目录里没有表和字段,写作目录也就不列;compose / verify 另传 entities=False,解析不了的原因照实说 @@ -161,6 +164,20 @@ def report_catalog(state: GraphState, spec: GraphSpec, node: GraphNode) -> dict[ return catalog +def _outer(node_id: Any) -> str: + """节点 id 在这一层图上是哪个节点:子工作流交回来的「子工作流节点 id/…」是那个子工作流节点,别的原样。""" + return str(node_id or "").split("/", 1)[0] + + +def with_subgraphs(allowed: set[str], ids: Iterable[Any]) -> set[str]: + """报告收证据的范围(祖先节点)加上范围里的子工作流交回来的那些(multi.run_subgraph:台账条目、口径卡的节点 id 是 + 「子工作流节点 id/…」,嵌套的一层层接上):子工作流在报告上游,它里面查过的数、算过的口径卡也在上游 + (docs/design/orchestration-round/00-总方案.md §2)。""" + out = set(allowed) + out.update(i for i in ids if isinstance(i, str) and "/" in i and _outer(i) in allowed) + return out + + def _masked_word() -> str: """「已遮罩」:证据面板、推演同一个说法。api.evidence 导入了引擎里的模块,用到时再导入。""" from app.api.evidence import MASKED diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 27fea92e..e10d2d0a 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -1360,7 +1360,7 @@ def rejected_data(exc: NodeRejected, spec: GraphSpec) -> dict[str, Any]: """被驳回的 run.cancelled 带什么:原因(rejected)、在哪一步(节点 id 和图上的名字)、哪种关卡、结论(驳回 / 超时 驳回)、谁、备注。进封存:谁在哪一步驳回、为什么,事后查得到。""" node = spec.node_map().get(exc.node_id) - return {"reason": END_REJECTED, "node_id": exc.node_id, "label": node.title if node else exc.node_id, + return {"reason": END_REJECTED, "node_id": exc.node_id, "label": _label_of(exc, node), "kind": exc.kind, "decision": exc.decision, "actor": exc.actor, "note": exc.note} @@ -1390,10 +1390,18 @@ def _failure(exc: BaseException, spec: GraphSpec) -> tuple[str, tuple[EventType, data["detail"] = detail if node_id: node = spec.node_map().get(node_id) - data.update(node_id=node_id, label=node.title if node else node_id) + data.update(node_id=node_id, label=_label_of(exc, node)) return message, (EventType.RUN_FAILED, data) +def _label_of(exc: BaseException, node: Any) -> str: + """出错、被驳回的那一步在图上叫什么。子工作流里的那一步(节点 id「子工作流节点 id/…」)不在这张图上,名字由 + multi.run_subgraph 接好记在异常上(「子工作流 › 那一步」)。""" + if node is not None: + return node.title + return str(getattr(exc, "label", None) or getattr(exc, "node_id", "") or "") + + async def _approval_default(*, run_class: str) -> str: """工具审批策略的全局默认,取自设置里的「危险工具默认需要人工确认」。 diff --git a/backend/tests/test_subgraph_evidence.py b/backend/tests/test_subgraph_evidence.py new file mode 100644 index 00000000..572d1634 --- /dev/null +++ b/backend/tests/test_subgraph_evidence.py @@ -0,0 +1,375 @@ +"""子工作流的证据交回父图(docs/design/orchestration-round/00-总方案.md §2)。 + +以前子图只把出口的值交回父图:子图里的查询快照、口径卡留在子图自己的状态里,子图里每一步在运行的事件里一条都没有(只设了 +节点 id 前缀没用上,LangGraph 的 ainvoke 不把子图的 custom 事件交给父图的流),父图报告引用不到子图里的数,父图口径卡读 +子图出口的字段一律停在子工作流节点。现在: +- 子图里每一步进同一个运行的事件流,节点 id「子工作流节点 id/…」,封存覆盖它们; +- 子图的台账(查询快照、口径卡)交回父图:父图报告的目录收进它们,引用标记、证据面板一路点到子图里的那一格; +- 子图出口的每个字段带出处,父图口径卡读它时跨子图往下追:追到原始数据的完整出具,子图里追不到的父图照样降档; + 静态检查(发布前、应用检查)和运行时同一套认法;嵌套两层照样; +- 子图里的遮罩照父图的运行。 +数据是虚构店铺库(tests/ledger_samples.py)。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.db.base import SessionLocal +from app.db.models import RunEvent +from app.engine.lineage import traced, untraced +from app.engine.runner import run_manager +from tests import ledger_samples as S + +PAID = "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid'" +GMV = {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "expression": "cell(nodes.fetch, 0, 'gmv')"} + + +@pytest.fixture(autouse=True) +async def engine_up(tmp_path): + await run_manager.setup() + await S.fresh_source(tmp_path / "shop.db") + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def node(nid: str, ntype: str, label: str | None = None, **config) -> dict: + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label or nid, "config": config}} + + +def metric(mid: str, expression: str, name: str = "") -> dict: + return {"id": mid, "name": name or mid, "unit": "元", "decimals": 2, "expression": expression} + + +def fetch_card_graph(*, card: dict | None = None, out: list[dict] | None = None, sql: str = PAID) -> dict: + """子工作流:取数 → 口径卡 → 出口(gmv 原样交出口径卡的指标,table 原样交出查询结果)。""" + return S.chain( + node("s", "input"), + node("fetch", "tool", tool=S.tool(), args={"sql": sql}), + card or node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[GMV]), + node("o", "output", fields=out or [{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"}, + {"name": "table", "value": "{{ nodes.fetch }}"}])) + + +async def publish(graph: dict, name: str = "子流程") -> str: + return await S.publish(name, graph=graph) + + +def parent(sub_id: str, *middle: dict, contract_from: str | None = "top", out: list[dict] | None = None, + sub_input: dict | None = None) -> dict: + return S.chain( + node("start", "input", fields=[{"name": "region"}]), + node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1, + **({"input": sub_input} if sub_input else {})), + *middle, + node("out", "output", fields=out or [{"name": "数", "value": "{{ nodes.top.metrics[0].value }}"}], + **({"contract": {"metrics_from": contract_from}} if contract_from else {}))) + + +async def run(graph: dict, inputs: dict | None = None): + started = await run_manager.start(graph=graph, input_payload=inputs or {}) + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + return row + + +async def events(run_id: str) -> list[RunEvent]: + async with SessionLocal() as session: + return list((await session.execute(select(RunEvent).where(RunEvent.run_id == run_id) + .order_by(RunEvent.seq))).scalars()) + + +async def card(run_id: str, node_id: str) -> dict: + from app.api.evidence import _sealed, _sealed_cards + + sealed = await _sealed(run_id) + [found] = [c for n, _, c in _sealed_cards(sealed) if n == node_id] + return {m["id"]: m for m in found["metrics"]} + + +# -------------------------------------------------------------------------- +# 子图里每一步进同一个运行的事件流 +# -------------------------------------------------------------------------- + + +async def test_the_subgraph_steps_are_in_the_run_with_prefixed_ids(): + sub_id = await publish(fetch_card_graph()) + row = await run(parent(sub_id, contract_from=None, out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}])) + evs = await events(row.id) + started = [(e.node_id, e.data.get("label")) for e in evs if e.type == "node.started"] + assert [n for n, _ in started] == ["start", "sub", "sub/s", "sub/fetch", "sub/card", "sub/o", "out"] + assert dict(started)["sub/fetch"] == "fetch" + [end] = [e for e in evs if e.type == "tool.end" and e.node_id == "sub/fetch"] + assert end.data.get("query_artifact") + # 子图里的台账条目在子图那一步的 node.finished 上(节点 id 带前缀),子工作流节点自己的 node.finished 不再记一遍 + finished = {e.node_id: e for e in evs if e.type == "node.finished"} + assert [x["kind"] for x in finished["sub/fetch"].data["evidence"]][:1] == ["tool"] + assert "evidence" not in finished["sub"].data + # 封存覆盖它们 + assert row.manifest_seq == evs[-1].seq + assert row.output == {"数": "246.5"} + + +async def test_the_instance_timeline_counts_the_subgraph_as_one_step(): + """应用的实例进度只说这张图上的步骤:子图里的那几步(「sub/…」)不单列,子工作流算一步。""" + from types import SimpleNamespace + + from app.apps.resolve import _timeline + from app.db.models import Run + + sub_id = await publish(fetch_card_graph()) + row = await run(parent(sub_id, contract_from=None, out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}])) + async with SessionLocal() as session: + steps = (await _timeline(session, SimpleNamespace(instance=await session.get(Run, row.id))))["steps"] + assert [s["node"] for s in steps] == ["start", "sub", "out"] + + +async def test_an_error_inside_the_subgraph_names_the_step_inside_it(): + sub_id = await publish(fetch_card_graph(sql="SELECT nope FROM nowhere")) + started = await run_manager.start(graph=parent(sub_id, 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" + [failed] = [e for e in await events(row.id) if e.type == "run.failed"] + assert (failed.data["node_id"], failed.data["label"]) == ("sub/fetch", "取数子流程 › fetch") + + +# -------------------------------------------------------------------------- +# 父图口径卡读子图出口的字段:跨子图往下追 +# -------------------------------------------------------------------------- + + +async def test_a_parent_card_reading_a_subgraph_field_traces_to_the_query_cell_and_issues_formally(client): + sub_id = await publish(fetch_card_graph()) + top = node("top", "metrics", caliber="父口径", caliber_version="v1", + metrics=[metric("gmv", "float(nodes.sub.output.gmv)"), metric("raw", "cell(nodes.sub.output.table, 0, 'gmv')")]) + row = await run(parent(sub_id, top)) + metrics = await card(row.id, "top") + [copy] = metrics["gmv"]["inputs"] + # 子图出口 gmv 原样交出子图口径卡的指标:接上那个指标自己的出处,一路追到取数查询 + assert (copy["via"], copy["node_id"], copy["subgraph"], copy["field"]) == ("metric", "sub/card", "sub", "gmv") + assert copy["path"] == "nodes.sub.output.gmv" and copy["value"] == "246.5" + assert copy["upstream"]["traced"] is True and traced(copy) + # 子图出口 table 原样交出查询结果:cell() 按台账认出是子图里的那份快照 + [cell] = metrics["raw"]["inputs"] + assert (cell["via"], cell["node_id"], cell["status"]) == ("tool_cell", "sub/fetch", "verified") + assert untraced(metrics["gmv"]["inputs"] + metrics["raw"]["inputs"]) == [] + assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"] + + # 证据面板:转抄的那一项后面接一步 upstream(子图里的卡),一直到子图里的取数查询 + body = (await client.get(f"/api/runs/{row.id}/evidence/metrics/top/gmv")).json() + chain = body["chain"] + assert chain[0]["untraced"] == [] + [inp] = [s for s in chain if s["step"] == "input"] + assert inp["subgraph"] == "sub" and inp["subgraph_label"] == "取数子流程" + assert inp["node_id"] == "sub/card" and inp["node_label"] == "取数子流程 › card" + [up] = [s for s in chain if s["step"] == "upstream"] + assert (up["node_id"], up["label"], up["sealed"], up["traced"]) == ("sub/card", "取数子流程 › card", True, True) + [read] = up["inputs"] + assert read["node_id"] == "sub/fetch" and read["node_label"] == "取数子流程 › fetch" + [query] = [s for s in chain if s["step"] == "query"] + assert query["artifact"] == read["artifact"] and query["node_id"] == "sub/fetch" and query["sealed"] is True + + +async def test_an_untraced_field_inside_the_subgraph_degrades_the_parent(): + inner = node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[metric("gmv", "vars.calc.v")]) + sub = S.chain(node("s", "input"), + node("calc", "transform", mode="json", template='{"v": 5}', assign_to="calc"), inner, + node("o", "output", fields=[{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"}])) + sub_id = await publish(sub) + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")]) + row = await run(parent(sub_id, top)) + [copy] = (await card(row.id, "top"))["gmv"]["inputs"] + assert copy["upstream"]["traced"] is False and not traced(copy) + issuance = row.output["_issuance"] + assert issuance["tier"] == "degraded" + assert any("指标「gmv」的输入没追到原始数据:nodes.sub.output.gmv(停在「取数子流程」)" in g + for g in issuance["gaps"]), issuance["gaps"] + + +async def test_a_field_that_is_not_a_whole_reference_stops_at_the_subgraph(): + sub_id = await publish(fetch_card_graph(out=[{"name": "gmv", "value": "合计 {{ nodes.card.metrics[0].value }}"}])) + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("n", "len(nodes.sub.output.gmv)")]) + row = await run(parent(sub_id, top)) + [item] = (await card(row.id, "top"))["n"]["inputs"] + assert (item["via"], item["node_id"]) == ("subgraph", "sub") and not traced(item) + assert row.output["_issuance"]["tier"] == "degraded" + + +async def test_the_subgraph_run_input_is_traced_back_through_the_parent_input_template(): + """子图出口交出它的运行输入:子图的输入是父图按「输入」模板渲染的。模板整串就是父图的运行输入时追到原始数据;拼了 + 别的字的停在子工作流的入口。""" + sub = S.chain(node("s", "input", fields=[{"name": "region"}]), + node("o", "output", fields=[{"name": "region", "value": "{{ input.region }}"}])) + sub_id = await publish(sub) + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("r", "nodes.sub.output.region")]) + row = await run(parent(sub_id, top, sub_input={"region": "{{ input.region }}"}), {"region": "east"}) + [item] = (await card(row.id, "top"))["r"]["inputs"] + assert (item["via"], item["path"], item["subgraph"]) == ("input", "nodes.sub.output.region", "sub") and traced(item) + assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"] + + row = await run(parent(sub_id, top, sub_input={"region": "{{ input.region }}区"}), {"region": "east"}) + [item] = (await card(row.id, "top"))["r"]["inputs"] + assert (item["via"], item["node_id"], item["field"]) == ("subgraph_input", "sub", "region") and not traced(item) + + +async def test_two_levels_of_nesting_trace_all_the_way_down(client): + inner_id = await publish(fetch_card_graph(), "内层") + middle = S.chain(node("m_in", "input"), + node("inner", "subgraph", "内层", workflow_id=inner_id, workflow_version=1), + node("m_out", "output", fields=[{"name": "gmv", "value": "{{ nodes.inner.output.gmv }}"}])) + middle_id = await publish(middle, "中层") + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")]) + row = await run(parent(middle_id, top)) + evs = await events(row.id) + assert "sub/inner/fetch" in {e.node_id for e in evs if e.type == "node.finished"} + [copy] = (await card(row.id, "top"))["gmv"]["inputs"] + assert (copy["via"], copy["node_id"], copy["subgraph"]) == ("metric", "sub/inner/card", "sub") + assert copy["upstream"]["traced"] is True + assert row.output["_issuance"]["tier"] == "formal", row.output["_issuance"]["gaps"] + chain = (await client.get(f"/api/runs/{row.id}/evidence/metrics/top/gmv")).json()["chain"] + [up] = [s for s in chain if s["step"] == "upstream"] + assert up["node_id"] == "sub/inner/card" and up["sealed"] is True and up["label"] == "取数子流程 › 内层 › card" + [query] = [s for s in chain if s["step"] == "query"] + assert query["node_id"] == "sub/inner/fetch" and query["sealed"] is True + + +# -------------------------------------------------------------------------- +# 静态检查:和运行时同一套认法 +# -------------------------------------------------------------------------- + + +async def _static(graph: dict) -> list[str]: + """受管级按指标出具的发布检查里「指标追不到原始数据」的那几条(governed.metric_untraced),子工作流读好放进来。""" + from app.engine.governance import lint_for_publish + from app.engine.lint_process import load_subgraphs, with_subgraphs + from app.engine.schema import GraphSpec + + spec = GraphSpec.model_validate(graph) + async with SessionLocal() as session: + graphs = await load_subgraphs(session, spec) + with with_subgraphs(graphs): + issues = lint_for_publish(spec, level="governed").issues + return [i.message for i in issues if i.code == "governed.metric_untraced"] + + +async def test_the_static_check_agrees_with_the_run(): + from app.engine.governance import lint_for_publish + from app.engine.schema import GraphSpec + + traced_sub = await publish(fetch_card_graph()) + inner = node("card", "metrics", caliber="子口径", caliber_version="v1", metrics=[metric("gmv", "vars.calc.v")]) + loose_sub = await publish(S.chain( + node("s", "input"), node("calc", "transform", mode="json", template='{"v": 5}', assign_to="calc"), inner, + node("o", "output", fields=[{"name": "gmv", "value": "{{ nodes.card.metrics[0].value }}"}])), "松的") + top = node("top", "metrics", caliber="父口径", caliber_version="v1", + metrics=[metric("gmv", "float(nodes.sub.output.gmv)"), metric("raw", "cell(nodes.sub.output.table, 0, 'gmv')")]) + good = parent(traced_sub, top) + bad = parent(loose_sub, node("top", "metrics", caliber="父口径", caliber_version="v1", + metrics=[metric("gmv", "float(nodes.sub.output.gmv)")])) + assert await _static(good) == [] + [said] = await _static(bad) + assert "停在「取数子流程」" in said + # 运行时同一个结论 + assert (await run(good)).output["_issuance"]["tier"] == "formal" + assert (await run(bad)).output["_issuance"]["tier"] == "degraded" + # 没把子工作流读好放进来(只看图本身的地方):看不到子图,照旧停在子工作流节点 + plain = [i for i in lint_for_publish(GraphSpec.model_validate(good), level="governed").issues + if i.code == "governed.metric_untraced"] + assert plain and "停在「取数子流程」" in plain[0].message + + +async def test_the_static_check_follows_two_levels_and_the_input_template(): + inner_id = await publish(fetch_card_graph(), "内层") + middle_id = await publish(S.chain( + node("m_in", "input"), node("inner", "subgraph", "内层", workflow_id=inner_id, workflow_version=1), + node("m_out", "output", fields=[{"name": "gmv", "value": "{{ nodes.inner.output.gmv }}"}])), "中层") + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("gmv", "float(nodes.sub.output.gmv)")]) + assert await _static(parent(middle_id, top)) == [] + + echo_id = await publish(S.chain(node("s", "input", fields=[{"name": "region"}]), + node("o", "output", fields=[{"name": "region", "value": "{{ input.region }}"}])), + "回声") + top = node("top", "metrics", caliber="父口径", caliber_version="v1", metrics=[metric("r", "nodes.sub.output.region")]) + assert await _static(parent(echo_id, top, sub_input={"region": "{{ input.region }}"})) == [] + assert await _static(parent(echo_id, top, sub_input={"region": "{{ input.region }}区"})) + + +# -------------------------------------------------------------------------- +# 父图报告直接引用子图里的查询和口径卡 +# -------------------------------------------------------------------------- + + +def report_parent(sub_id: str) -> dict: + return S.chain( + node("start", "input"), + node("sub", "subgraph", "取数子流程", workflow_id=sub_id, workflow_version=1), + node("write", "report", instructions="写答案"), + node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}], + contract={"report_from": "write"})) + + +async def _segments(client: AsyncClient, run_id: str) -> dict[str, dict]: + body = (await client.get(f"/api/runs/{run_id}/evidence")).json() + assert body["mode"] == "cited", body + return body + + +async def test_the_parent_report_cites_the_subgraph_query_and_card(client, monkeypatch): + sub_id = await publish(fetch_card_graph()) + S.script(monkeypatch, "子流程里查到的销售额 [[v:Q1.r0.gmv]],口径算的是 [[m:gmv]]。") + started = await run_manager.start(graph=report_parent(sub_id), input_payload={}) + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + body = await _segments(client, row.id) + items = {e["alias"]: e for e in body["evidence"]} + assert (items["Q1"]["node_id"], items["Q1"]["sealed"]) == ("sub/fetch", True) + assert (items["m:gmv"]["node_id"], items["m:gmv"]["sealed"]) == ("sub/card", True) + cited = {e["to"]: e["from"] for e in body["edges"] if e["rel"] == "cites"} + # 单元格引用一路点到子图里的那一格 + cell = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['Q1']}")).json()["chain"] + [query] = [s for s in cell if s["step"] == "query"] + assert query["node_id"] == "sub/fetch" and query["sealed"] is True and query["artifact"] == items["Q1"]["artifact"] + # 指标引用:子图里的卡、它的输入、子图里的查询 + chain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['m:gmv']}")).json()["chain"] + step = chain[0] + assert (step["step"], step["node_id"], step["sealed"], step["untraced"]) == ("metric", "sub/card", True, []) + [inp] = [s for s in chain if s["step"] == "input"] + assert (inp["node_id"], inp["node_label"], inp["via"]) == ("sub/fetch", "取数子流程 › fetch", "tool_cell") + [q] = [s for s in chain if s["step"] == "query"] + assert q["node_id"] == "sub/fetch" and q["sealed"] is True + # 封存核对通过,可验证报告照常(事件本来就在同一个运行里) + from app.engine.runner import verify_manifest + + assert (await verify_manifest(row.id))["ok"] is True + res = await client.get(f"/api/runs/{row.id}/report.html") + assert res.status_code == 200 and "246.5" in res.text + + +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"}) + amount = node("card", "metrics", caliber="子口径", caliber_version="v1", + metrics=[metric("gmv", "cell(nodes.fetch, 0, 'amount')")]) + sub_id = await publish(fetch_card_graph(card=amount, sql="SELECT phone, amount FROM orders WHERE id = 1")) + S.script(monkeypatch, "第一单的电话 [[v:Q1.r0.phone]]。") + started = await run_manager.start(graph=report_parent(sub_id), input_payload={}) + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + body = await _segments(client, row.id) + cited = {e["to"]: e["from"] for e in body["edges"] if e["rel"] == "cites"} + chain = (await client.get(f"/api/runs/{row.id}/evidence/segments/{cited['Q1']}")).json()["chain"] + [query] = [s for s in chain if s["step"] == "query"] + assert query["node_id"] == "sub/fetch" + assert "13800000001" not in str(query), "子图里的遮罩列照父图的运行遮" + assert "已遮罩" in str(query) and query["sealed"] is True