静态地认指标出处也跨卡往下追:转抄同一张图里另一张口径卡的指标,按那个指标的式子接着认(最多 MAX_COPY_DEPTH 层),受管级发布检查、应用检查和运行时、证据面板同一套说法;被抄的卡按 caliber_from 固定在别处版本的照旧停在那张卡

This commit is contained in:
ThearchyHelios
2026-10-17 12:31:13 +08:00
parent 124e1c0f49
commit f520a88891
2 changed files with 81 additions and 6 deletions
+34 -6
View File
@@ -10,7 +10,9 @@
这么写的)跨卡往下追——接上那个指标自己的出处,那个指标的输入都追到了原始数据,这一项才算追到(upstream.traced);
转抄的转抄一层层接着追,最多 MAX_COPY_DEPTH 层(循环里抄自己上一轮的卡每一轮多一层,到了上限就算没追到);
- 证据面板(api/evidence 的指标步骤):卡上记的输入里追不到原始数据的(untraced),照实说出处追到哪个节点为止;
- 应用检查(apps/lint):受管级应用里绑定的指标,按 static_stops 静态地认(只有图、没有运行),追不到的给提示。
- 应用检查(apps/lint)、受管级发布检查(governance):按 static_stops 静态地认(只有图、没有运行),追不到的给提示。
转抄同一张图里另一张口径卡的指标,和运行时一样跨卡按那个指标的式子接着认(最多 MAX_COPY_DEPTH 层);被抄的卡按
caliber_from 固定在别处的版本时,静态看不到那一版的图,照旧停在那张卡。
"""
from __future__ import annotations
@@ -138,26 +140,52 @@ class Stop:
node_id: str | None
def static_stops(nodes: Mapping[str, Mapping[str, Any]], expression: str) -> list[Stop]:
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 交来的字段,算追到了;别的(表达式整形、
沙箱代码、模型调用、转抄别的口径卡……)停在写它的那个节点。表达式写错了的不在这里报(画布的校验报)。"""
按位置取的格穿过原样解析的整形节点、运行输入、开了按出处核对字段的 Agent 交来的字段,算追到了;转抄同一张图里
另一张口径卡的指标,按那个指标的式子接着认(它的输入都追到了才算);别的(表达式整形、沙箱代码、模型调用、转抄
固定在别处版本的口径卡……)停在写它的那个节点。表达式写错了的不在这里报(画布的校验报)。"""
try:
tree = parse_expression(expression)[0]
except ExpressionError:
return []
out: list[Stop] = []
for path in leaf_refs(tree):
stop = _static_stop(nodes, path)
stop = _static_stop(nodes, path, _depth)
if stop is not None:
out.append(stop)
return out
def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str) -> Stop | None:
def _copied_metric(nodes: Mapping[str, Mapping[str, Any]], path: str) -> tuple[str, str] | None:
"""path 转抄同一张图里一张口径卡的第 i 个指标时:(写它的卡, 那个指标的式子)。卡按 caliber_from 固定在别处版本的、
认不出是哪张卡的、下标越界的为 None。"""
copy = metric_copy(path)
if copy is None:
return None
root, name, index = copy
writers = _writers(nodes, f"{root}.{name}")
if len(writers) != 1:
return None
card = nodes.get(writers[0]) or {}
cfg = card.get("config") or {}
if card.get("type") != "metrics" or isinstance(cfg.get("caliber_from"), dict):
return None
metrics = cfg.get("metrics") or []
if index >= len(metrics) or not isinstance(metrics[index], dict) or not metrics[index].get("expression"):
return None
return writers[0], str(metrics[index]["expression"])
def _static_stop(nodes: Mapping[str, Mapping[str, Any]], path: str, depth: int = 0) -> Stop | None:
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):
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)]
+47
View File
@@ -0,0 +1,47 @@
"""静态地认(engine/lineage.static_stops)和运行时同一套:转抄同一张图里另一张口径卡的指标时跨卡接着认。
受管级发布检查(governance)和应用检查(apps/lint)都用它:被抄的那个指标追到了查询格,转抄的这一项就算追到;被抄的
指标停在表达式整形节点,转抄的这一项停在被抄的那张卡;被抄的卡按 caliber_from 固定在别处版本的,静态看不到那一版的
图,照旧停在那张卡;转抄成环的到层数上限算没追到。
"""
from __future__ import annotations
from app.engine.lineage import Stop, static_stops
def _nodes(**extra):
nodes = {
"q": {"type": "tool", "config": {"tool": "db_query__shop", "args": {"sql": "SELECT 1 AS gmv"}}},
"base": {"type": "metrics", "config": {"metrics": [
{"id": "gmv", "expression": "cell(nodes.q, 0, 'gmv')"}]}},
}
nodes.update(extra)
return nodes
def test_a_copied_metric_that_reaches_a_query_cell_is_traced():
nodes = _nodes(copy={"type": "metrics", "config": {"metrics": [
{"id": "x", "expression": "nodes.base.metrics[0].value"}]}})
assert static_stops(nodes, "nodes.base.metrics[0].value") == []
# 转抄的转抄也接着认
nodes["copy2"] = {"type": "metrics", "config": {"metrics": [{"id": "y", "expression": "nodes.copy.metrics[0].value"}]}}
assert static_stops(nodes, "nodes.copy2.metrics[0].value") == []
def test_a_copied_metric_whose_source_is_untraced_stops_at_that_card():
nodes = _nodes(shape={"type": "transform", "config": {"expression": "1"}},
loose={"type": "metrics", "config": {"metrics": [{"id": "z", "expression": "nodes.shape.value"}]}})
assert static_stops(nodes, "nodes.loose.metrics[0].value") == [Stop("nodes.loose.metrics[0].value", "loose")]
def test_a_card_pinned_elsewhere_still_stops_and_out_of_range_still_stops():
nodes = _nodes(pinned={"type": "metrics", "config": {"caliber_from": {"workflow_id": "w", "version": 1},
"metrics": [{"id": "gmv", "expression": "cell(nodes.q, 0, 'gmv')"}]}})
assert static_stops(nodes, "nodes.pinned.metrics[0].value") == [Stop("nodes.pinned.metrics[0].value", "pinned")]
assert static_stops(_nodes(), "nodes.base.metrics[3].value") == [Stop("nodes.base.metrics[3].value", "base")]
def test_a_copy_cycle_gives_up_at_the_depth_limit():
nodes = {"a": {"type": "metrics", "config": {"metrics": [{"id": "x", "expression": "nodes.b.metrics[0].value"}]}},
"b": {"type": "metrics", "config": {"metrics": [{"id": "y", "expression": "nodes.a.metrics[0].value"}]}}}
assert static_stops(nodes, "nodes.a.metrics[0].value") == [Stop("nodes.a.metrics[0].value", "a")]