"""运行对比的纯函数（engine/compare.py）：同一个工作流的两次运行，每个指标的变化拆成数据、口径、模板 / SQL。 两次运行都是照真实运行的事件形状手写的虚构数据（区域、金额、日期都是编的）：取数节点 fetch 查一条按区域汇总的 SQL，Agent 节点 bot 自己再查一次、把各区域的金额作为数组字段交给口径卡，口径卡算几个指标。工件放在字典里。守的是： - 只数据变（SQL 一字不差、结果变了）是 data_only；只改日期字面量是 params_only；SQL 结构变了是 sql_changed、不拆 - 口径变了（表达式、口径版本）是 caliber_and_data：旧表达式的叶子路径在这次的输入里都取得到时严格拆分， 口径效应 + 数据效应 = 总变化（逐位相等）；取不到就不拆，说明缺的是哪一个 - 维度归因：按维度键配对两边的行，按变化的绝对值取前 10；度量列之和等于输入的值（两边都等）才给占比，占比之和 为 1；对不上时占比为空并说明；截断的快照不归因，照实写 - 遮罩列的值绝不出现在结果里；用到遮罩列的指标不拆分 - 模板对比：新增、删除的节点，配置有变化的节点和配置项名 - 输入的出处：合并查询追到它的上游查询；Agent 按出处交的字段还要引用同一列、同一段行（行段相同、一边是整份结果 一边不是的不算；都不是整份结果的，一边引用了的区域另一边还在快照里却没引用的也不算）；没按出处交的数、上游数据 没变数却变了的、上游的 Agent 两次交出的结果不同的，说不清变化从哪来（incomparable），记不出写入者的只看口径卡 之前可能执行过的节点；图里写死的字面量改了、连线变了、读到的默认设置变了都算模板变化 """ from __future__ import annotations import ast import copy import importlib import inspect import json import pkgutil import textwrap from decimal import Decimal from types import SimpleNamespace import app.engine.nodes as nodes_package from app.engine import compare from app.engine.compiler import RUNNERS from app.engine.merge_query import MERGE_SOURCE from app.engine.schema import NodeType from tests.fixtures.compare_runs import ( BASE_ROWS, COLUMNS, EAST, NEXT_ROWS, REFUND, TOOL, WEEK37, Fake, amounts_in, cell_in, chain, default_metrics, metric, node, region_sql, scalar_in, week, weekly_graph, ) def run_compare(base, target): result = compare.compare(base.side(), target.side()) json.dumps(result, allow_nan=False) # 结果要能原样交给接口 return result def by_id(result): return {m["metric_id"]: m for m in result["metrics"]} def exact(value): return Decimal(repr(value)) if isinstance(value, float) else Decimal(value) # -------------------------------------------------------------------------- # 分类：只数据变、只参数变、SQL 结构变 # -------------------------------------------------------------------------- def test_same_sql_new_rows_is_data_only(): result = run_compare(week("r1"), week("r2", NEXT_ROWS)) metrics = by_id(result) east, total = metrics["east"], metrics["total"] assert (east["classification"], total["classification"]) == ("data_only", "data_only") assert (east["base"], east["target"], east["delta"]) == (1200.5, 1500.5, 300.0) assert (east["base_rendered"], east["target_rendered"], east["delta_rendered"]) == ("1,200.5元", "1,500.5元", "+300.0元") assert east["delta_pct"] == round(300 / 1200.5 * 100, 2) assert east["card_id"] == "card" and east["card_title"] == "周报口径卡" and east["name"] == "东区金额" assert east["decomposition"] is None and east["decomposition_note"] is None # 两个输入都落在查询快照上：直接说取数的 SQL 没变、查询结果变了 assert east["reasons"] == total["reasons"] == [compare.DATA_FROM_QUERIES] assert total["delta"] == 200.0 and total["delta_rendered"] == "+200.0元" assert result["summary"] == {"unchanged": 0, "data_only": 2, "caliber_and_data": 0, "sql_changed": 0, "params_only": 0, "added": 0, "removed": 0, "incomparable": 0} # 输入两边的值都列出来，带着各自的查询快照 [cell] = east["inputs"] assert (cell["path"], cell["base"], cell["target"]) == (EAST, 1200.5, 1500.5) assert cell["base_artifact"] != cell["target_artifact"] # 查询：节点 + 第几次 + 工具配上对，SQL 一字不差 assert [(q["node_id"], q["ordinal"], q["structure"]) for q in result["queries"]] == [ ("fetch", 1, "same"), ("bot", 1, "same")] fetch = result["queries"][0] assert fetch["node_title"] == "区域取数" and fetch["base_sql"] == fetch["target_sql"] == region_sql() assert fetch["literals"] == [] and (fetch["base_rows"], fetch["target_rows"]) == (3, 3) assert (fetch["base_truncated"], fetch["target_truncated"]) == (False, False) assert result["template"]["same_graph"] is True and result["template"]["nodes"] == [] assert (result["base"]["id"], result["target"]["id"]) == ("r1", "r2") def test_identical_runs_are_unchanged(): result = run_compare(week("r1"), week("r2")) assert {m["classification"] for m in result["metrics"]} == {"unchanged"} east = by_id(result)["east"] assert east["delta"] == 0 and east["delta_rendered"] is None and east["attribution"] is None assert result["summary"]["unchanged"] == 2 def test_only_date_literals_changed_is_params_only(): # 图里的 SQL 是模板（两次一样），执行时代进去的日期不同 graph = weekly_graph(region_sql(("{{ input.start }}", "{{ input.end }}"))) result = run_compare(week("r1", graph=graph), week("r2", NEXT_ROWS, sql=region_sql(WEEK37), graph=graph)) metrics = by_id(result) assert {m["classification"] for m in metrics.values()} == {"params_only"} assert any("参数" in r for r in metrics["east"]["reasons"]) fetch = result["queries"][0] assert fetch["structure"] == "params_only" assert fetch["literals"] == [{"base": "'2026-08-31'", "target": "'2026-09-07'"}, {"base": "'2026-09-06'", "target": "'2026-09-13'"}] # 定时运行换了日期属于参数变化：模板（图）一样 assert result["summary"]["params_only"] == 2 and result["template"]["same_graph"] is True def test_a_changed_sql_structure_is_flagged_not_split(): paid = region_sql(extra=" AND status = 'paid'") result = run_compare(week("r1"), week("r2", NEXT_ROWS, sql=paid)) metrics = by_id(result) assert {m["classification"] for m in metrics.values()} == {"sql_changed"} east = metrics["east"] assert east["decomposition"] is None and east["attribution"] is None assert any("区域取数" in r and "SQL" in r for r in east["reasons"]) fetch = result["queries"][0] assert fetch["structure"] == "changed" and fetch["literals"] == [] assert any("status = 'paid'" in d for d in fetch["differences"]) # 图里 fetch 的 SQL 也改了：模板对比列出它和改了的配置项 [changed] = result["template"]["nodes"] assert (changed["node_id"], changed["change"], changed["keys"]) == ("fetch", "config", ["args.sql"]) assert result["template"]["same_graph"] is False # -------------------------------------------------------------------------- # 口径变了：严格拆分 # -------------------------------------------------------------------------- def net_metrics(expression): """净额指标 net：表达式由调用方定（v1 只取金额，v2 减去退款）。""" def build(fetch, bot, rows, bot_rows): gmv, refund = rows[0][1], rows[0][2] inputs = [cell_in(0, "gmv", gmv, fetch)] value = gmv if REFUND in expression: inputs.append(cell_in(0, "refund", refund, fetch)) value = gmv - refund return [metric("net", expression, value, inputs, name="东区净额")] return build def test_a_caliber_change_is_split_into_caliber_and_data_effects(): base = week("r1", metrics=net_metrics(EAST), version="v1") target = week("r2", NEXT_ROWS, metrics=net_metrics(f"{EAST} - {REFUND}"), version="v2") net = by_id(run_compare(base, target))["net"] assert net["classification"] == "caliber_and_data" assert any("表达式" in r for r in net["reasons"]) and any("v1" in r and "v2" in r for r in net["reasons"]) assert (net["base"], net["target"], net["delta"]) == (1200.5, 1380.5, 180.0) split = net["decomposition"] assert net["decomposition_note"] is None # 口径效应 = f_new(x_new) − f_old(x_new) = 1380.5 − 1500.5；数据效应 = f_old(x_new) − f_old(x_old) = 1500.5 − 1200.5 assert (split["caliber_effect"], split["data_effect"]) == (-120.0, 300.0) assert (split["caliber_rendered"], split["data_rendered"]) == ("-120.0元", "+300.0元") assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(net["delta"]) def test_the_split_adds_up_exactly_with_awkward_decimals(): """小数位不同、浮点相减有误差的数：两效应之和逐位等于总变化。""" def build(expression, decimals): def make(fetch, bot, rows, bot_rows): gmv, refund = rows[0][1], rows[0][2] inputs = [cell_in(0, "gmv", gmv, fetch), cell_in(0, "refund", refund, fetch)] value = round(eval(expression.replace(EAST, repr(gmv)).replace(REFUND, repr(refund))), decimals) return [metric("ratio", expression, value, inputs, decimals=decimals, unit="")] return make base_rows = [["东区", 1000.1, 0.3], *BASE_ROWS[1:]] next_rows = [["东区", 1000.75, 0.2], *NEXT_ROWS[1:]] base = week("r1", base_rows, metrics=build(f"{EAST} * 0.1 - {REFUND}", 2)) target = week("r2", next_rows, metrics=build(f"{EAST} * 0.3 + {REFUND}", 3), version="v2") ratio = by_id(run_compare(base, target))["ratio"] split = ratio["decomposition"] assert split is not None, ratio["decomposition_note"] # f_old(x_new) 按旧口径的小数位取整：1000.75 * 0.1 - 0.2 = 99.875，取 2 位是 99.88（按新口径的 3 位是 99.875） assert exact(split["data_effect"]) == exact(99.88) - exact(ratio["base"]) assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(ratio["delta"]) assert exact(ratio["delta"]) == exact(ratio["target"]) - exact(ratio["base"]) def test_a_version_bump_with_the_same_formula_is_all_data(): base = week("r1", metrics=net_metrics(EAST), version="v1") target = week("r2", NEXT_ROWS, metrics=net_metrics(EAST), version="v2") net = by_id(run_compare(base, target))["net"] assert net["classification"] == "caliber_and_data" assert (net["decomposition"]["caliber_effect"], net["decomposition"]["data_effect"]) == (0, 300.0) assert net["decomposition"]["caliber_rendered"] == "0.0元" def test_no_split_when_an_old_leaf_is_not_recorded_this_time(): base = week("r1", metrics=net_metrics(f"{EAST} - {REFUND}"), version="v1") target = week("r2", NEXT_ROWS, metrics=net_metrics(EAST), version="v2") net = by_id(run_compare(base, target))["net"] assert net["classification"] == "caliber_and_data" and net["decomposition"] is None assert REFUND in net["decomposition_note"] and "无法严格拆分" in net["decomposition_note"] # 总变化照算 assert net["delta"] == 1500.5 - (1200.5 - 100.0) def test_no_split_when_the_sql_changed_too(): base = week("r1", metrics=net_metrics(EAST), version="v1") target = week("r2", NEXT_ROWS, sql=region_sql(extra=" AND status = 'paid'"), metrics=net_metrics(EAST), version="v2") net = by_id(run_compare(base, target))["net"] assert net["classification"] == "caliber_and_data" and net["decomposition"] is None assert "SQL" in net["decomposition_note"] # 两条不同的查询的行不拿来按维度配对 assert net["attribution"] is None def test_a_caliber_change_still_names_the_sql_change(): """口径变了、SQL 结构也变了、旧口径的叶子这次又没记录：分类按口径，SQL 的变化照样写进说明。""" base = week("r1", metrics=net_metrics(f"{EAST} - {REFUND}"), version="v1") paid = region_sql(extra=" AND status = 'paid'") target = week("r2", NEXT_ROWS, sql=paid, metrics=net_metrics(EAST), version="v2") net = by_id(run_compare(base, target))["net"] assert net["classification"] == "caliber_and_data" and net["decomposition"] is None assert any("区域取数" in r and "SQL" in r for r in net["reasons"]), net["reasons"] # 退款那一格只在旧口径里用到：改表达式带来的，不再另说一遍「只在一边有记录」 assert not any("只在一边" in r for r in net["reasons"]), net["reasons"] def test_a_unit_change_is_not_subtracted(): def build(unit): def make(fetch, bot, rows, bot_rows): return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], unit=unit)] return make east = by_id(run_compare(week("r1", metrics=build("元")), week("r2", NEXT_ROWS, metrics=build("万元"))))["east"] assert east["classification"] == "caliber_and_data" and any("单位" in r for r in east["reasons"]) assert east["delta"] is None and east["delta_rendered"] is None and east["decomposition"] is None assert "单位" in east["decomposition_note"] def test_display_settings_that_change_the_value_are_caliber_changes(): def build(**kw): def make(fetch, bot, rows, bot_rows): return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], **kw)] return make def east(before, after): return by_id(run_compare(week("r1", metrics=build(**before)), week("r2", NEXT_ROWS, metrics=build(**after)))) # 小数位改变取整 found = east({"decimals": 1}, {"decimals": 0}) assert found["east"]["classification"] == "caliber_and_data" and any("小数位" in r for r in found["east"]["reasons"]) # 比率与否改变数值的含义 found = east({"unit": "", "fmt": "plain"}, {"unit": "", "fmt": "percent_of_ratio"}) assert found["east"]["classification"] == "caliber_and_data" and found["east"]["delta"] is None # 千分位和不分组只影响显示：分类照数据算，变化写进说明 result = run_compare(week("r1", metrics=build(fmt="thousands")), week("r2", NEXT_ROWS, metrics=build(fmt="plain"))) assert by_id(result)["east"]["classification"] == "data_only" assert any("显示格式" in n for n in result["notes"]) def test_an_old_card_without_display_settings_is_not_a_caliber_change(): """升级前封存的运行：卡上的指标没记小数位、显示格式和输入（当时按定义里的小数位取过整）。""" def old(run_id, value, graph=None): run = Fake(run_id, graph=graph) run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.card([{"id": "east", "name": "东区金额", "unit": "元", "value": value, "expression": EAST}]) return run def new(run_id, value): run = Fake(run_id) fetch = run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.card([metric("east", EAST, value, [cell_in(0, "gmv", value, fetch)], name="东区金额")]) return run east = by_id(run_compare(old("r1", 1200.5), new("r2", 1200.5)))["east"] assert (east["classification"], east["reasons"]) == ("unchanged", []) east = by_id(run_compare(old("r1", 1200.5), new("r2", 1500.5)))["east"] assert east["classification"] == "incomparable" and east["reasons"] == [compare.NO_LINEAGE] # 那次运行的图里写着定义：按定义里的小数位核对 graph = weekly_graph() graph["nodes"][3]["data"]["config"]["metrics"] = [{"id": "east", "expression": EAST, "decimals": 2}] east = by_id(run_compare(old("r1", 1200.5, graph), new("r2", 1200.5)))["east"] assert east["classification"] == "caliber_and_data" and any("小数位由 2 改为 1" in r for r in east["reasons"]) def test_the_split_adds_up_in_json_without_fixed_decimals(): """没设小数位的指标：三个数各自转成浮点也要逐位相加（接口按 JSON 交回的就是浮点）。""" def build(expression): def make(fetch, bot, rows, bot_rows): gmv, refund = rows[0][1], rows[0][2] inputs = [cell_in(0, "gmv", gmv, fetch), cell_in(0, "refund", refund, fetch)] value = eval(expression.replace(EAST, repr(gmv)).replace(REFUND, repr(refund))) return [metric("m", expression, value, inputs, decimals=None, unit="")] return make base_rows = [["东区", 0.1, 0.2], *BASE_ROWS[1:]] next_rows = [["东区", 0.7, 0.1], *NEXT_ROWS[1:]] base = week("r1", base_rows, metrics=build(f"{EAST} + {REFUND}")) # 0.30000000000000004 target = week("r2", next_rows, metrics=build(f"{EAST} * 3 + {REFUND}"), version="v2") # 2.1999999999999997 m = by_id(run_compare(base, target))["m"] split = m["decomposition"] assert split is not None, m["decomposition_note"] assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(m["delta"]) assert abs(m["delta"] - (m["target"] - m["base"])) < 1e-12 def test_a_tiny_base_does_not_break_the_percentage(): """基准是浮点误差那么小的数、这次很大：百分比写得出来（或者为空），对比不能整个失败。""" def build(value): def make(fetch, bot, rows, bot_rows): m = metric("diff", EAST, 1.0, [cell_in(0, "gmv", rows[0][1], fetch)], decimals=None) m["value"], m["rendered"] = value, "—" return [m] return make diff = by_id(run_compare(week("r1", metrics=build(0.30000000000000004 - 0.3)), week("r2", NEXT_ROWS, metrics=build(1e8))))["diff"] assert diff["delta"] == 1e8 - (0.30000000000000004 - 0.3) assert diff["delta_pct"] is not None and diff["delta_pct"] > 1e25 # -------------------------------------------------------------------------- # 维度归因 # -------------------------------------------------------------------------- def test_attribution_pairs_keys_and_shares_add_up_to_one(): base, target = week("r1"), week("r2", NEXT_ROWS) total = by_id(run_compare(base, target))["total"] attribution = total["attribution"] assert attribution["columns"] == ["region"] and attribution["measure"] == "gmv" # 归因用的是哪个输入、两边各自哪份快照 assert (attribution["path"], attribution["base_artifact"], attribution["target_artifact"]) == ( "vars.kpi.amounts", base.queries[1], target.queries[1]) assert attribution["total_delta"] == 200.0 rows = attribution["rows"] assert [(r["key"], r["base"], r["target"], r["delta"]) for r in rows] == [ (["东区"], 1200.5, 1500.5, 300.0), (["南区"], 800.0, 700.0, -100.0), (["西区"], 600.0, 600.0, 0.0)] assert [r["share"] for r in rows] == [1.5, -0.5, 0.0] assert abs(sum(r["share"] for r in rows) - 1) < 1e-12 def test_attribution_pairs_by_key_not_by_row_position(): """这次多了一个区域、顺序也变了：按区域配对，只有一边有的那一边写空。""" next_rows = [["北区", 90.0, 0.0], ["东区", 1500.5, 120.0], ["西区", 600.0, 10.0], ["南区", 700.0, 40.0]] total = by_id(run_compare(week("r1"), week("r2", next_rows)))["total"] rows = {tuple(r["key"]): r for r in total["attribution"]["rows"]} assert rows[("北区",)]["base"] is None and rows[("北区",)]["target"] == 90.0 and rows[("北区",)]["delta"] == 90.0 assert rows[("东区",)]["delta"] == 300.0 and rows[("南区",)]["delta"] == -100.0 assert abs(sum(r["share"] for r in rows.values()) - 1) < 1e-12 def test_attribution_keeps_the_ten_largest_changes(): names = [f"{n}区" for n in "一二三四五六七八九十甲乙"] base_rows = [[name, 100.0, 0.0] for name in names] next_rows = [[name, 100.0 + (i + 1) * 10 * (-1) ** i, 0.0] for i, name in enumerate(names)] total = by_id(run_compare(week("r1", base_rows), week("r2", next_rows)))["total"] rows = total["attribution"]["rows"] assert len(rows) == 10 assert [abs(r["delta"]) for r in rows] == [120.0, 110.0, 100.0, 90.0, 80.0, 70.0, 60.0, 50.0, 40.0, 30.0] assert "12" in total["attribution"]["note"] # 占比按全部维度值的总变化算：前 10 个之和不必为 1，全部之和为 1 assert total["attribution"]["total_delta"] == sum((i + 1) * 10 * (-1) ** i for i in range(12)) def test_no_share_when_the_measure_does_not_add_up_to_the_input(): """东区金额只是快照里的一格：各区域之和不等于它，各键的变化照列，占比为空并说明。""" east = by_id(run_compare(week("r1"), week("r2", NEXT_ROWS)))["east"] attribution = east["attribution"] assert attribution["measure"] == "gmv" and [r["key"] for r in attribution["rows"]] == [["东区"], ["南区"], ["西区"]] assert all(r["share"] is None for r in attribution["rows"]) and attribution["note"] def test_no_share_when_only_one_side_adds_up(): """基准运行里东区就是全部（各区域之和等于它），这次南区也有了：只有一边对得上，不给占比。""" base_rows = [["东区", 100.0, 0.0], ["南区", 0.0, 0.0], ["西区", 0.0, 0.0]] next_rows = [["东区", 150.0, 0.0], ["南区", 20.0, 0.0], ["西区", 0.0, 0.0]] east = by_id(run_compare(week("r1", base_rows), week("r2", next_rows)))["east"] attribution = east["attribution"] assert east["classification"] == "data_only" and len(attribution["rows"]) == 3 assert all(r["share"] is None for r in attribution["rows"]) and compare.NO_SHARE in attribution["note"] def test_a_total_row_summed_in_floating_point_still_adds_up(): """合计行是库里按浮点加出来的（0.1 + 0.2 + 0.3 = 0.6000000000000001）：差在最后几位也算相等，照给占比。""" base_rows = [["东区", 0.1, 0.0], ["南区", 0.2, 0.0], ["西区", 0.3, 0.0], ["合计", 0.1 + 0.2 + 0.3, 0.0]] next_rows = [["东区", 0.4, 0.0], ["南区", 0.2, 0.0], ["西区", 0.3, 0.0], ["合计", 0.4 + 0.2 + 0.3, 0.0]] def build(fetch, bot, rows, bot_rows): return [metric("all", "cell(nodes.fetch, 3, 'gmv')", rows[3][1], [cell_in(3, "gmv", rows[3][1], fetch)], decimals=None)] found = by_id(run_compare(week("r1", base_rows, metrics=build), week("r2", next_rows, metrics=build)))["all"] rows = found["attribution"]["rows"] assert [r["key"] for r in rows] == [["东区"], ["南区"], ["西区"]] assert abs(sum(r["share"] for r in rows) - 1) < 1e-9 def test_date_typed_and_constant_columns_are_not_dimensions(): """日期类型的列（两边有相同的日子也不算）和两边都只有一个值的列（币种）不当维度。""" columns = ["day", "region", "currency", "gmv"] types = {"day": "date", "region": "text", "currency": "text", "gmv": "number"} def make(run_id, rows): run = Fake(run_id) sql = "SELECT day, region, currency, SUM(amount) AS gmv FROM orders GROUP BY day, region, currency" bot = run.agent("bot", sql, columns, rows, types=types) run.card([metric("total", "sum(vars.kpi.amounts)", sum(r[3] for r in rows), [amounts_in(bot, rows)])]) return run base_rows = [["2026-09-01", "东区", "CNY", 10.0], ["2026-09-02", "南区", "CNY", 20.0]] next_rows = [["2026-09-02", "东区", "CNY", 15.0], ["2026-09-03", "南区", "CNY", 18.0]] attribution = by_id(run_compare(make("r1", base_rows), make("r2", next_rows)))["total"]["attribution"] assert attribution["columns"] == ["region"] assert [(r["key"], r["delta"]) for r in attribution["rows"]] == [(["东区"], 5.0), (["南区"], -2.0)] def test_a_total_row_is_left_out_of_its_own_breakdown(): """合计行和各区域在同一份结果里：指标取合计那一格，按其余各行归因，占比之和为 1。""" base_rows = [*BASE_ROWS, ["合计", 2600.5, 150.0]] next_rows = [*NEXT_ROWS, ["合计", 2800.5, 170.0]] def build(fetch, bot, rows, bot_rows): return [metric("all", "cell(nodes.fetch, 3, 'gmv')", rows[3][1], [cell_in(3, "gmv", rows[3][1], fetch)])] total = by_id(run_compare(week("r1", base_rows, metrics=build), week("r2", next_rows, metrics=build)))["all"] rows = total["attribution"]["rows"] assert ["合计"] not in [r["key"] for r in rows] and len(rows) == 3 assert abs(sum(r["share"] for r in rows) - 1) < 1e-12 and total["attribution"]["total_delta"] == 200.0 def test_a_truncated_snapshot_is_not_attributed(): base, target = week("r1"), Fake("r2") sql = region_sql() fetch = target.tool("fetch", sql, rows=NEXT_ROWS) bot = target.agent("bot", sql, rows=NEXT_ROWS, truncated=True) target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS)) total = by_id(run_compare(base, target))["total"] attribution = total["attribution"] assert attribution["rows"] == [] and "截断" in attribution["note"] assert result_queries(run_compare(base, target), "bot")["target_truncated"] is True def test_a_truncated_base_snapshot_is_not_attributed_either(): base = Fake("r1") fetch = base.tool("fetch", region_sql()) bot = base.agent("bot", region_sql(), truncated=True) base.card(default_metrics(fetch, bot, BASE_ROWS, BASE_ROWS)) total = by_id(run_compare(base, week("r2", NEXT_ROWS)))["total"] assert total["classification"] == "data_only" assert total["attribution"]["rows"] == [] and "基准运行" in total["attribution"]["note"] def result_queries(result, node_id): return next(q for q in result["queries"] if q["node_id"] == node_id) def test_a_period_column_that_never_matches_is_not_a_dimension(): """每期都不同的列（这里是日期写成的文本）不拿来配对：按其余的维度列归因。""" columns = ["day", "region", "gmv"] types = {"day": "text", "region": "text", "gmv": "number"} def make(run_id, days, values): run = Fake(run_id) rows = [[day, region, value] for day in days for region, value in zip(("东区", "南区"), values)] sql = "SELECT day, region, SUM(amount) AS gmv FROM orders GROUP BY day, region" bot = run.agent("bot", sql, columns, rows, types=types) total = sum(r[2] for r in rows) run.card([metric("total", "sum(vars.kpi.amounts)", total, [amounts_in(bot, rows)])]) return run total = by_id(run_compare(make("r1", ("2026-09-05", "2026-09-06"), (10.0, 20.0)), make("r2", ("2026-09-12", "2026-09-13"), (15.0, 18.0))))["total"] attribution = total["attribution"] assert attribution["columns"] == ["region"] assert [(r["key"], r["delta"]) for r in attribution["rows"]] == [(["东区"], 10.0), (["南区"], -4.0)] assert "day" in attribution["note"] def test_odd_values_stay_json_safe(): """维度列里有非数、度量加起来超出浮点范围：结果照样写得进 JSON（接口按 allow_nan=False 写，写不进就是 500）。""" def make(run_id, rows): run = Fake(run_id) bot = run.agent("bot", region_sql(), ["region", "gmv"], rows, types={"region": "text", "gmv": "number"}) # 卡上的值随便取一个有限的数（这里不重算它），两次不同就行 run.card([metric("total", "sum(vars.kpi.amounts)", float(len(rows)), [amounts_in(bot, rows)])]) return run nan = float("nan") result = run_compare(make("r1", [["东区", 1.0], [nan, 2.0]]), make("r2", [["东区", 1e308], ["东区", 1e308], [nan, 1.0]])) rows = {tuple(r["key"]): r for r in by_id(result)["total"]["attribution"]["rows"]} assert rows[("东区",)]["target"] is None and rows[("nan",)]["delta"] == -1.0 # -------------------------------------------------------------------------- # 遮罩列 # -------------------------------------------------------------------------- def test_masked_columns_never_show_up(): """数据源遮了 manager（经理名）和 refund：经理名不当维度、不出现在结果里；用到 refund 的指标输入不给值、不拆分。""" columns = ["region", "manager", "gmv", "refund"] types = {"region": "text", "manager": "text", "gmv": "number", "refund": "number"} def make(run_id, rows, expression, version): run = Fake(run_id) sql = "SELECT region, manager, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders GROUP BY region, manager" fetch = run.tool("fetch", sql, columns, rows, types=types) bot = run.agent("bot", sql, columns, rows, types=types) run.masked = {fetch: frozenset({"manager", "refund"}), bot: frozenset({"manager", "refund"})} inputs = [cell_in(0, "gmv", rows[0][2], fetch)] value = rows[0][2] if "refund" in expression: inputs.append(cell_in(0, "refund", rows[0][3], fetch)) value = rows[0][2] - rows[0][3] run.card([metric("net", expression, value, inputs), metric("total", "sum(vars.kpi.amounts)", sum(r[2] for r in rows), [amounts_in(bot, rows)])], version=version) return run base_rows = [["东区", "经理甲", 1200.5, 101.25], ["南区", "经理乙", 800.0, 52.75]] next_rows = [["东区", "经理甲", 1500.5, 133.5], ["南区", "经理乙", 700.0, 47.25]] result = run_compare(make("r1", base_rows, EAST, "v1"), make("r2", next_rows, "cell(nodes.fetch, 0, 'gmv') - cell(nodes.fetch, 0, 'refund')", "v2")) text = json.dumps(result, ensure_ascii=False) for secret in ("经理甲", "经理乙", "101.25", "52.75", "133.5", "47.25"): assert secret not in text, secret metrics = by_id(result) net = metrics["net"] refund = next(i for i in net["inputs"] if "refund" in i["path"]) assert refund["masked"] is True and refund["base"] is None and refund["target"] is None assert net["decomposition"] is None and "遮罩" in net["decomposition_note"] # 维度只用没遮的 region；占比照给 attribution = metrics["total"]["attribution"] assert attribution["columns"] == ["region"] and abs(sum(r["share"] for r in attribution["rows"]) - 1) < 1e-12 assert any("遮罩" in n for n in result["notes"]) def test_a_masked_measure_is_not_attributed(): def masked(run): run.masked = {a: frozenset({"gmv"}) for a in run.queries} return run result = run_compare(masked(week("r1")), masked(week("r2", NEXT_ROWS))) total = by_id(result)["total"] assert total["attribution"]["rows"] == [] and "遮罩" in total["attribution"]["note"] east = by_id(result)["east"] assert east["inputs"][0]["base"] is None and east["inputs"][0]["masked"] is True assert "1200.5" not in json.dumps(east["inputs"]) and "800.0" not in json.dumps(result) # -------------------------------------------------------------------------- # 对不上具体查询的输入、运行输入、一边才有的指标 # -------------------------------------------------------------------------- def code_metrics(value): def build(fetch, bot, rows, bot_rows): return [metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, "calc", "code")])] return build def calc_node(code="print(1)"): return node("calc", "code", "调整", code=code, assign_to="adj") def test_an_input_from_a_code_node_follows_its_upstream(): """代码节点接在取数节点后面（不在 Agent 下游）：看它画布上的全部上游。""" def make(run_id, rows, value, sql=None): run = Fake(run_id, graph=weekly_graph(sql, after_fetch=(calc_node(),))) sql = sql or region_sql() run.tool("fetch", sql, rows=rows) run.agent("bot", sql, rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run # 上游什么都没变：变化来自数据 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12)))["calc"] assert calc["classification"] == "data_only" and calc["delta"] == 4 assert calc["reasons"] == [compare.DATA_FROM_UPSTREAM] # 上游的 SQL 结构变了：说不清它变没变，不能报成数据变化 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, sql=region_sql(extra=" AND status = 'paid'"))))["calc"] assert calc["classification"] == "incomparable" and any("调整" in r for r in calc["reasons"]) # 上游查到的数据两次一模一样，它交的数却变了：变化不来自数据 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", BASE_ROWS, 12)))["calc"] assert calc["classification"] == "incomparable" and any("调整" in r for r in calc["reasons"]) assert compare.DATA_FROM_UPSTREAM not in calc["reasons"] def test_a_code_node_after_an_agent_that_answered_differently_is_incomparable(): """代码节点在 Agent 下游，算的是 Agent 没按出处交的占比：Agent 两次交出的结果不同，变化可能来自数据，也可能来自 模型本身的差异，不能说变化来自上游数据。Agent 两次交出的一模一样时照常（见 test_swapped_queries_with_the_same_literals_pair_up_exactly）。""" def make(run_id, rows, share, value=None): value = share * 2 if value is None else value run = Fake(run_id, graph=weekly_graph(between=(calc_node("result = nodes.bot.data.kpi.share * 2"),))) run.tool("fetch", region_sql(), rows=rows) run.plain_agent("bot", [region_sql()], {"kpi": {"share": share}}, rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run calc = by_id(run_compare(make("r1", BASE_ROWS, 0.46), make("r2", NEXT_ROWS, 0.52)))["calc"] assert calc["classification"] == "incomparable" and calc["attribution"] is None assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「区域分析」（Agent）两次交出的结果不同，" "无法判断变化是否来自数据"] # Agent 交的不同，代码节点交的数两次一样：照常没变，没有可说不清的变化 calc = by_id(run_compare(make("r1", BASE_ROWS, 0.46, value=1.0), make("r2", NEXT_ROWS, 0.52, value=1.0)))["calc"] assert (calc["classification"], calc["reasons"]) == ("unchanged", []) # Agent 的产出两次都没落成工件（写工件失败）：一样不一样核对不了，同样不报成数据变化 def unstored(run): run.events = [(seq, etype, nid, {k: v for k, v in data.items() if k != "artifact"} if (etype, nid) == ("node.finished", "bot") else data) for seq, etype, nid, data in run.events] return run calc = by_id(run_compare(unstored(make("r1", BASE_ROWS, 0.46)), unstored(make("r2", NEXT_ROWS, 0.46, value=1.0))))["calc"] assert calc["classification"] == "incomparable" and "区域分析" in calc["reasons"][0] def test_a_reconfigured_node_upstream_of_a_code_node_is_named(): """代码节点上游的数据整形节点改了表达式（SQL 一字没改）：代码节点交的数变了说不清从哪来，说明里点出改的是哪个节点； 改了几个的，点名画布上最靠前的那个（节点 ID 的先后和画布上的先后不同）。""" def make(run_id, expressions, rows, value): shapes = [node(nid, "transform", label, mode="expression", expression=expression, assign_to=f"kept_{nid}") for (nid, label), expression in zip((("shape_m", "整理"), ("shape_z", "再整理"), ("shape_a", "汇总")), expressions)] run = Fake(run_id, graph=weekly_graph(after_fetch=(*shapes, calc_node()))) run.tool("fetch", region_sql(), rows=rows) run.agent("bot", region_sql(), rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run same = ["nodes.fetch.rows"] * 3 calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10), make("r2", ["nodes.fetch.rows[:2]", *same[1:]], NEXT_ROWS, 12)))["calc"] assert calc["classification"] == "incomparable" assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"] calc = by_id(run_compare(make("r1", same, BASE_ROWS, 10), make("r2", ["nodes.fetch.rows[:2]"] * 3, NEXT_ROWS, 12)))["calc"] assert calc["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「整理」的配置有变化，无法判断变化来源"] def test_a_report_upstream_of_a_code_node_that_wrote_differently_is_incomparable(): """代码节点上游有报告撰写节点（文字是模型写的），两次写得不同：代码节点交的数变了，可能来自数据，也可能来自模型 本身的差异；两次写得一字不差的（产出工件一样）照常是数据变化。""" def make(run_id, rows, text, value): run = Fake(run_id, graph=weekly_graph(after_fetch=(node("draft", "report", "初稿"), calc_node()))) run.tool("fetch", region_sql(), rows=rows) run.add("node.started", "draft", {"node_type": "report"}) run.add("node.finished", "draft", {"artifact": run.put({"text": text})}) run.agent("bot", region_sql(), rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run calc = by_id(run_compare(make("r1", BASE_ROWS, "上周初稿", 10), make("r2", NEXT_ROWS, "本周初稿", 12)))["calc"] assert calc["classification"] == "incomparable" assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游节点「初稿」（报告撰写）两次交出的结果不同，" "无法判断变化是否来自数据"] calc = by_id(run_compare(make("r1", BASE_ROWS, "初稿", 10), make("r2", NEXT_ROWS, "初稿", 12)))["calc"] assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM]) def test_an_unknown_writer_is_looked_for_only_among_nodes_before_the_card(): """两边都记不出 vars.adj 是哪个节点写的：只看口径卡之前可能执行过的节点。口径卡自己改了配置、它下游的写周报两次写得 不同，都写不了它读到的值，不能点名成它的上游；它上游的节点改了配置照常说不清，并行支路上的节点也可能先写了变量， 同样说不清，只是不说成「上游」。""" def make(run_id, rows, value, *, other="1", report="上周", reshape="nodes.fetch.rows", aside="nodes.fetch.rows"): graph = chain(node("start", "input", "开始"), node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": region_sql()}), node("shape", "transform", "整理", mode="expression", expression=reshape, assign_to="kept"), node("calc", "code", "调整", code="x", assign_to="adj"), node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1", metrics=[{"id": "other", "expression": other}]), node("write", "report", "写周报"), node("out", "output", "成果")) # 并行支路：开始 → 旁路整理 → 成果，和口径卡不在一条线上 graph["nodes"].append(node("aside", "transform", "旁路整理", mode="expression", expression=aside, assign_to="extra")) graph["edges"] += [{"source": "start", "target": "aside"}, {"source": "aside", "target": "out"}] run = Fake(run_id, graph=graph) run.tool("fetch", region_sql(), rows=rows) run.code("calc", value, "adj") run.card([metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, None, "state")])]) run.add("node.started", "write", {"node_type": "report"}) run.add("node.finished", "write", {"artifact": run.put({"text": report})}) return run # 口径卡自己改了另一个指标的定义，写周报两次写得不同：都在它之后执行，变化照常来自上游数据 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, other="2", report="本周")))["calc"] assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM_SCOPED]) calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, report="本周")))["calc"] assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM]) # 口径卡上游的整理节点改了表达式 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, reshape="nodes.fetch.rows[:2]")))["calc"] assert calc["classification"] == "incomparable" assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，上游节点「整理」的配置有变化，无法判断变化来源"] # 并行支路上的旁路整理改了表达式 calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12, aside="nodes.fetch.rows[:2]")))["calc"] assert calc["classification"] == "incomparable" assert calc["reasons"] == ["输入「vars.adj」不知道由哪个节点写入，节点「旁路整理」的配置有变化，无法判断变化来源"] def test_a_run_input_upstream_of_a_code_node_is_a_parameter_change(): """代码节点上游查到的数据两次一模一样，只有运行输入变了：它交的数变了算参数变化。""" def make(run_id, goal, value): run = Fake(run_id, graph=weekly_graph(after_fetch=(calc_node(),)), inputs={"goal": goal}) run.add("node.started", "start", {"node_type": "input"}) run.add("node.finished", "start", {"artifact": run.put({"goal": goal})}) run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, BASE_ROWS, BASE_ROWS)) return run calc = by_id(run_compare(make("r1", 1000, 10), make("r2", 1500, 15)))["calc"] assert calc["classification"] == "params_only" assert calc["reasons"] == ["输入「vars.adj」来自「调整」，它的上游只有运行输入有变化"] def test_a_value_from_a_querying_node_that_moved_without_its_data_is_incomparable(): """调用工具节点整份交出的结果（不是 cell() 取的格）：它查到的行两次一样，交出的值（这里是查询耗时）却变了， 变化不来自数据；查到的行也变了的，照常是数据变化。""" def make(run_id, elapsed, rows=BASE_ROWS, graph=None): run = Fake(run_id, graph=graph) run.tool("fetch", region_sql(), rows=rows) run.agent("bot", region_sql(), rows=rows) run.card([metric("cost", "vars.res.elapsed_ms", elapsed, [scalar_in("vars.res.elapsed_ms", elapsed, "fetch", "tool")], unit="毫秒", decimals=None)]) return run cost = by_id(run_compare(make("r1", 3), make("r2", 5)))["cost"] assert cost["classification"] == "incomparable" and any("区域取数" in r for r in cost["reasons"]) cost = by_id(run_compare(make("r1", 3), make("r2", 5, NEXT_ROWS)))["cost"] assert cost["classification"] == "data_only" # 取数节点自己改了配置（SQL 没动）：它整份交出的结果可能跟着变，算模板变化 graph = weekly_graph() graph["nodes"][1]["data"]["config"]["fail_fast"] = False cost = by_id(run_compare(make("r1", 3), make("r2", 5, NEXT_ROWS, graph)))["cost"] assert cost["classification"] == "sql_changed", cost["reasons"] assert cost["reasons"] == ["上游节点「区域取数」的配置有变化"] def test_data_from_upstream_is_only_claimed_when_the_graph_is_the_same(): """改了它下游的节点（成果字段）：它的上游没变，变化照样来自上游数据，但不能说「模板没有变化」。""" def make(run_id, rows, value, graph): run = Fake(run_id, graph=graph) run.tool("fetch", region_sql(), rows=rows) run.agent("bot", region_sql(), rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run base_graph = weekly_graph(after_fetch=(calc_node(),)) target_graph = copy.deepcopy(base_graph) target_graph["nodes"][-1]["data"]["config"] = {"fields": [{"name": "周报", "value": "{{ nodes.write.text }}"}]} calc = by_id(run_compare(make("r1", BASE_ROWS, 10, base_graph), make("r2", NEXT_ROWS, 12, target_graph)))["calc"] assert calc["classification"] == "data_only" assert calc["reasons"] == [compare.DATA_FROM_UPSTREAM_SCOPED] def test_the_code_node_itself_changed_is_a_template_change(): def make(run_id, code, value): run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code=code, assign_to="adj"),))) run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, BASE_ROWS, BASE_ROWS)) return run calc = by_id(run_compare(make("r1", "x = 1", 10), make("r2", "x = 2", 11)))["calc"] assert calc["classification"] == "sql_changed" and any("调整" in r for r in calc["reasons"]) def test_rewired_edges_are_a_template_change(): """调整节点改由取数节点接进来（各节点的配置都没动）：它交的数变了算模板变化，不能说变化来自上游数据。""" calc_node = node("calc", "code", "调整", code="result = 1", assign_to="adj") base_graph = weekly_graph(between=(calc_node,)) target_graph = copy.deepcopy(base_graph) target_graph["edges"] = [e for e in target_graph["edges"] if (e["source"], e["target"]) != ("bot", "calc")] target_graph["edges"].append({"source": "fetch", "target": "calc"}) def make(run_id, graph, value): run = Fake(run_id, graph=graph) run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, BASE_ROWS, BASE_ROWS)) return run result = run_compare(make("r1", base_graph, 10), make("r2", target_graph, 12)) calc = by_id(result)["calc"] assert calc["classification"] == "sql_changed" and any("连线" in r for r in calc["reasons"]), calc["reasons"] assert compare.EDGES_CHANGED in result["notes"] def uncited(run_id, share, *, graph=None, sqls=None): """Agent 没开按出处核对，口径卡用了它交的占比 vars.kpi.share；东区金额照旧取取数节点的那一格。""" run = Fake(run_id, graph=graph) fetch = run.tool("fetch", region_sql()) run.plain_agent("bot", sqls or [region_sql()], {"kpi": {"share": share}}) run.card([metric("east", EAST, BASE_ROWS[0][1], [cell_in(0, "gmv", BASE_ROWS[0][1], fetch)], name="东区金额"), metric("share", "vars.kpi.share * 100", share * 100, [scalar_in("vars.kpi.share", share, "bot", "agent")], unit="%", fmt="plain")]) return run def test_a_number_the_agent_gave_without_citing_is_incomparable(): """Agent 没按出处交的数：两次查到的数据一样，数却变了（模型本身的差异），不能说是数据变化。""" result = run_compare(uncited("r1", 0.46), uncited("r2", 0.52)) share = by_id(result)["share"] assert share["classification"] == "incomparable" and share["attribution"] is None assert any("区域分析" in r and "查询快照" in r for r in share["reasons"]), share["reasons"] assert by_id(result)["east"]["classification"] == "unchanged" # 交的数一样：照常没变 assert by_id(run_compare(uncited("r1", 0.46), uncited("r2", 0.46)))["share"]["classification"] == "unchanged" def test_an_extra_query_by_the_writer_is_a_sql_change(): extra = "SELECT COUNT(*) AS n FROM refunds" share = by_id(run_compare(uncited("r1", 0.46), uncited("r2", 0.52, sqls=[region_sql(), extra])))["share"] assert share["classification"] == "sql_changed" and any("多了一次查询" in r for r in share["reasons"]) # 反过来这次少查了一次：同样是取数变了 share = by_id(run_compare(uncited("r1", 0.46, sqls=[region_sql(), extra]), uncited("r2", 0.52)))["share"] assert share["classification"] == "sql_changed" and any("少了一次查询" in r for r in share["reasons"]) def test_an_input_on_a_snapshot_on_one_side_only_is_incomparable(): """基准运行没找到东区金额那一格落在哪份快照上（卡上只记了值），这次找到了：只有一边对得上查询快照，比不了取数。""" def make(run_id, rows, *, located): run = Fake(run_id) fetch = run.tool("fetch", region_sql(), rows=rows) run.agent("bot", region_sql(), rows=rows) item = cell_in(0, "gmv", rows[0][1], fetch) if not located: kept = {k: v for k, v in item.items() if k not in ("artifact", "locator", "eid")} item = {**kept, "status": "unresolved"} run.card([metric("east", EAST, rows[0][1], [item], name="东区金额")]) return run east = by_id(run_compare(make("r1", BASE_ROWS, located=False), make("r2", NEXT_ROWS, located=True)))["east"] assert east["classification"] == "incomparable" and east["attribution"] is None assert east["reasons"] == [f"输入「{EAST}」只有一边对得上查询快照，无法比较取数"] def test_a_changed_default_model_is_a_template_change_for_agents_that_follow_it(): base_graph = weekly_graph() base_graph["defaults"] = {"provider": "anthropic", "model": "model-a"} target_graph = copy.deepcopy(base_graph) target_graph["defaults"]["model"] = "model-b" result = run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=target_graph)) share = by_id(result)["share"] assert share["classification"] == "sql_changed" assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型）有变化"] assert result["template"]["same_graph"] is False and compare.DEFAULTS_CHANGED in result["notes"] # 取数节点不用模型：东区金额不受影响 assert by_id(result)["east"]["classification"] == "unchanged" # 按出处交的字段：取法可能跟着模型变了 total = by_id(run_compare(week("r1", graph=base_graph), week("r2", NEXT_ROWS, graph=target_graph)))["total"] assert total["classification"] == "sql_changed" assert total["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型）有变化，它按出处交来的字段可能换了取法"] # 一次改了好几项：写界面上的叫法（模型接入不写 provider），多的只写项数 many = copy.deepcopy(base_graph) many["defaults"] = {"provider": "openai", "model": "model-b", "temperature": 0.5, "max_tokens": 2000} share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=many)))["share"] assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型接入、模型、温度等 4 项）有变化"] # 同时改了 Agent 不读的结论句裁判：只写它跟着变的模型 judged = copy.deepcopy(target_graph) judged["defaults"]["judge"] = {"max_cost_usd": 1.0} share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=judged)))["share"] assert share["reasons"] == ["上游节点「区域分析」跟随的工作流默认设置（模型）有变化"] # Agent 自己指定了模型：默认设置变了它不跟着变 for graph in (base_graph, target_graph): graph["nodes"][2]["data"]["config"]["model"] = "model-x" share = by_id(run_compare(uncited("r1", 0.46, graph=base_graph), uncited("r2", 0.52, graph=target_graph)))["share"] assert share["classification"] == "incomparable" def test_defaults_that_agents_do_not_read_leave_them_alone(): """只改了 Agent 不读的默认设置：结论句裁判、结论句的处理只有报告节点读，审批策略只管调用工具前问不问人。按出处 交的字段、取数节点的格子照常是数据变化。""" for before, after in (({"judge": {"max_cost_usd": 0.5}}, {"judge": {"max_cost_usd": 1.0}}), ({"claims": "off"}, {"claims": "judge"}), ({"approval": "dangerous"}, {"approval": "never"})): base_graph, target_graph = weekly_graph(), weekly_graph() base_graph["defaults"], target_graph["defaults"] = before, after result = run_compare(week("r1", graph=base_graph), week("r2", NEXT_ROWS, graph=target_graph)) found = {m["metric_id"]: (m["classification"], m["reasons"]) for m in result["metrics"]} assert found == {"east": ("data_only", [compare.DATA_FROM_QUERIES]), "total": ("data_only", [compare.DATA_FROM_QUERIES])}, (before, found) assert compare.DEFAULTS_CHANGED in result["notes"] def test_a_model_default_reaches_every_node_that_calls_the_model(): """默认模型换了，跟着换的不只 Agent：让模型分类的条件分支、让模型修复的结构校验（默认开着）、用模型重排的知识检索 也是。它们在代码节点上游时，代码节点交的数变了不能说来自上游数据；不调模型的照常。""" def calc(extra, before=None, after=None): def make(run_id, defaults, rows, value): graph = weekly_graph(after_fetch=(extra, calc_node())) graph["defaults"] = defaults run = Fake(run_id, graph=graph) run.tool("fetch", region_sql(), rows=rows) run.agent("bot", region_sql(), rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run before = before or {"provider": "anthropic", "model": "model-a"} after = after or {"provider": "anthropic", "model": "model-b"} return by_id(run_compare(make("r1", before, BASE_ROWS, 10), make("r2", after, NEXT_ROWS, 12)))["calc"] cases = [{"key": "peak", "condition": "true"}] for extra in (node("route", "branch", "分流", mode="llm", instruction="判断是否旺季", cases=cases), node("route", "validate", "校验", schema={"type": "object"}), node("route", "retrieve", "检索", query="旺季", rerank="model")): found = calc(extra) assert found["classification"] == "incomparable", (extra["type"], found["reasons"]) assert found["reasons"] == [f"输入「vars.adj」来自「调整」，上游节点「{extra['data']['label']}」跟随的工作流默认" "设置（模型）有变化，无法判断变化来源"] for extra in (node("route", "branch", "分流", mode="expression", cases=cases), node("route", "validate", "校验", schema={"type": "object"}, repair_with_llm=False), node("route", "retrieve", "检索", query="旺季")): found = calc(extra) assert found["classification"] == "data_only", (extra["type"], found["reasons"]) # 调不调模型也可能由默认设置定：条件分支没写判断方式、默认设置写了让模型判断的，跟着默认模型变；结构校验的默认 # 设置关了让模型修复的，不跟着变 asks = node("route", "branch", "分流", instruction="判断是否旺季", cases=cases) found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "llm", "model": "model-b"}) assert found["classification"] == "incomparable", found["reasons"] assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（模型）有变化，" "无法判断变化来源"] found = calc(node("route", "validate", "校验", schema={"type": "object"}), {"repair_with_llm": False, "model": "model-a"}, {"repair_with_llm": False, "model": "model-b"}) assert found["classification"] == "data_only", found["reasons"] # 默认的判断方式这次改成了按表达式：基准运行那次让模型判断过，跟着变的是判断方式和模型两项 found = calc(asks, {"mode": "llm", "model": "model-a"}, {"mode": "expression", "model": "model-b"}) assert found["reasons"] == ["输入「vars.adj」来自「调整」，上游节点「分流」跟随的工作流默认设置（判断方式、模型）" "有变化，无法判断变化来源"] #: 只顺着节点模块里的函数往下找：别处的函数不经 NodeContext.cfg 读节点配置 _NODE_MODULES = "app.engine.nodes." def _defined_here(obj): """obj 是定义在节点模块里的函数或类（装饰过的取原函数），否则 None。""" obj = inspect.unwrap(obj) if inspect.isfunction(obj) else getattr(obj, "func", obj) if (inspect.isfunction(obj) or inspect.isclass(obj)) and obj.__module__.startswith(_NODE_MODULES): return obj return None def _settings_read(runner): """一类节点的执行器运行时按 NodeContext.cfg 读的配置项（节点上没写就回退到默认设置）：执行器自己的函数体，加上它 用到的、定义在节点模块里的函数和类，一层层往下找（函数里就地 import 的也算）。返回 (配置项, 认不出键名的读法, 走到的函数)。键名是变量的，只认在字面量元组上循环出来的（Agent 的预算三项）；别的变量键名认不出是哪几项， 原样交回，测试据此失败。""" read, unknown, seen = set(), [], set() todo = [runner] while todo: fn = _defined_here(todo.pop()) if fn is None or (fn.__module__, fn.__qualname__) in seen: continue seen.add((fn.__module__, fn.__qualname__)) names = vars(importlib.import_module(fn.__module__)) # 它所在模块里的名字（类没有 __globals__） parts = list(ast.walk(ast.parse(textwrap.dedent(inspect.getsource(fn))))) imported = {alias.asname or alias.name: (part.module, alias.name) for part in parts if isinstance(part, ast.ImportFrom) and part.module for alias in part.names} looped = {} # 循环变量 → 它在字面量元组上取到的键名：{k: ctx.cfg(k) for k in ("max_steps", …)} for part in parts: for gen in getattr(part, "generators", None) or ([part] if isinstance(part, ast.For) else []): if isinstance(gen.target, ast.Name) and isinstance(gen.iter, (ast.Tuple, ast.List)) and all( isinstance(e, ast.Constant) and isinstance(e.value, str) for e in gen.iter.elts): looped[gen.target.id] = {e.value for e in gen.iter.elts} for part in parts: called = part.func if isinstance(part, ast.Call) else None if isinstance(called, ast.Attribute) and called.attr == "cfg": key = part.args[0] if part.args else None if isinstance(key, ast.Constant) and isinstance(key.value, str): read.add(key.value) elif isinstance(key, ast.Name) and key.id in looped: read |= looped[key.id] else: unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}") elif isinstance(called, ast.Attribute) and called.attr == "approval_mode": read.add("approval") # NodeContext.approval_mode 同样先读节点、再回退到默认设置 elif isinstance(called, ast.Name) and called.id == "_setting": # 报告模块的 _setting(spec, 节点, 键) 和 NodeContext.cfg 同一个规则 key = part.args[2] if len(part.args) > 2 else None if isinstance(key, ast.Constant) and isinstance(key.value, str): read.add(key.value) else: unknown.append(f"{fn.__qualname__}: {ast.unparse(part)}") elif isinstance(part, ast.Name) and isinstance(part.ctx, ast.Load): # 调用、当回调传出去的函数都算用到了 target = names.get(part.id) if target is None and part.id in imported: module, name = imported[part.id] target = getattr(importlib.import_module(module), name, None) todo.append(target) elif isinstance(part, ast.Attribute) and isinstance(part.value, ast.Name): module = names.get(part.value.id) if inspect.ismodule(module): todo.append(getattr(module, part.attr, None)) return read, unknown, seen def test_every_setting_a_node_reads_is_known_to_the_comparison(): """节点运行时按 NodeContext.cfg 读配置，节点上没写就回退到图级默认设置。对比按「哪类节点读哪几项」把默认设置的 变化落到节点上（compare._CFG_KEYS）：执行器新读了一项、这里没登记，默认设置改了这一项就会被漏掉；登记了不读的， 默认设置改了它就会被误标。按节点类型比，不按模块比：一个模块管几类节点（Agent 和模型调用、调用工具和代码……）， 其中一类读了、另一类没读的，不能拿另一类的登记充数。""" reached = set() for kind, runner in RUNNERS.items(): read, unknown, seen = _settings_read(runner) assert not unknown, (str(kind), unknown) known = compare._CFG_KEYS.get(str(kind), frozenset()) | compare._CFG_KEYS_IF.get(str(kind), frozenset()) assert read == known, (str(kind), "没登记", sorted(read - known), "登记了不读", sorted(known - read)) reached |= seen # 键名不是字面量的读法也认得出来 assert {"max_steps", "budget_tokens", "budget_usd"} <= _settings_read(RUNNERS[NodeType.AGENT])[0] # 节点模块里每个按 cfg 读配置的函数都走到了：走不到的说明上面漏了某种用法，查出来的配置项就不全 holders = set() for info in pkgutil.iter_modules(nodes_package.__path__): module = importlib.import_module(_NODE_MODULES + info.name) for top in ast.parse(inspect.getsource(module)).body: if isinstance(top, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and any( isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute) and n.func.attr in ("cfg", "approval_mode") for n in ast.walk(top)): holders.add((module.__name__, top.name)) assert holders and holders <= reached, sorted(holders - reached) def cited(run_id, rows=BASE_ROWS, *, column="gmv", span=None, sql=None): """Agent 按出处交回数组字段：引用快照的哪一列、哪一段行由调用方定。""" run = Fake(run_id) sql = sql or region_sql() run.tool("fetch", sql, rows=rows) bot = run.agent("bot", sql, rows=rows, column=column, span=span) first, last = span or (0, len(rows) - 1) values = [r[COLUMNS.index(column)] for r in rows[first:last + 1]] run.card([metric("total", "sum(vars.kpi.amounts)", sum(values), [amounts_in(bot, rows, column=column, span=span)])]) return run def test_an_agent_citing_another_column_is_not_a_data_change(): """SQL 一字不差、数据一模一样，Agent 这次引用的是退款列：取的不是同一个数。""" total = by_id(run_compare(cited("r1"), cited("r2", column="refund")))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert any("gmv" in r and "refund" in r for r in total["reasons"]), total["reasons"] def test_an_agent_citing_other_rows_is_not_a_data_change(): """Agent 这次只引用了后两行：东区的数还在快照里，不能把它算成「东区的数据没了」。""" total = by_id(run_compare(cited("r1"), cited("r2", span=(1, 2))))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert any("行" in r for r in total["reasons"]), total["reasons"] # 两边引用的都是整份结果（这次多了一个区域）：照常是数据变化 total = by_id(run_compare(cited("r1"), cited("r2", [*NEXT_ROWS, ["北区", 90.0, 0.0]])))["total"] assert total["classification"] == "data_only" and total["attribution"]["rows"] #: 按区域排序的结果这次多了北区，排在南区前面，西区挪到第 4 行 WITH_NORTH = [["东区", 1200.5, 100.0], ["北区", 90.0, 0.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0]] def test_the_same_rows_are_not_the_same_citation_when_only_one_side_is_the_whole_result(): """Agent 两次都引用第 1 至 3 行：基准运行里那是整份结果，这次西区还在第 4 行、没被引用。行段一样也不是同一处， 不能把西区算成「数据没了」（占比也会超过 100%）；反过来（基准只引用了一部分、这次是整份）也一样。""" total = by_id(run_compare(cited("r1"), cited("r2", WITH_NORTH, span=(0, 2))))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert total["reasons"] == ["Agent「区域分析」在基准运行中引用了整份结果，这次只引用了第 1 至 3 行（结果共 4 行），" "取的不是同一组数"] total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), cited("r2")))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert total["reasons"] == ["Agent「区域分析」在基准运行中只引用了第 1 至 3 行（结果共 4 行），这次引用了整份结果，" "取的不是同一组数"] # 两边都只引用了前 3 行、都不是整份结果：同一处，照常是数据变化 later = [["东区", 1500.5, 120.0], ["北区", 95.0, 0.0], ["南区", 700.0, 40.0], ["西区", 600.0, 10.0]] total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), cited("r2", later, span=(0, 2))))["total"] assert total["classification"] == "data_only" and total["attribution"]["rows"] def test_the_same_rows_are_not_the_same_citation_when_a_cited_region_moved_out_of_them(): """两边都只引用第 1 至 3 行、都不是整份结果：基准运行引用了西区，这次北区排进了前 3 行，西区挪到第 4 行，还在这次 的快照里，只是没被引用。取的不是同一组数，不能把西区算成「数据没了」（占比会远超 100%）；反过来（这次引用的区域 在基准运行的快照里、没被引用）也一样。西区这次真的没有了的，照常是数据变化。""" with_middle = [["东区", 1200.5, 100.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]] total = by_id(run_compare(cited("r1", with_middle, span=(0, 2)), cited("r2", WITH_NORTH, span=(0, 2))))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」这次仍在查询结果第 4 行，不在引用的" "行里，取的不是同一组数"] total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), cited("r2", with_middle, span=(0, 2))))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert total["reasons"] == ["Agent「区域分析」两次都引用第 1 至 3 行，「西区」在基准运行中位于查询结果第 4 行，不在" "引用的行里，取的不是同一组数"] # 按金额从大到小取前 3 名：北区这次挤进了前 3，西区落到第 4 行 top = region_sql().replace("ORDER BY region", "ORDER BY gmv DESC") before = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["西区", 600.0, 0.0], ["中区", 30.0, 0.0]] after = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["西区", 600.0, 0.0]] total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top), cited("r2", after, span=(0, 2), sql=top)))["total"] assert total["classification"] == "incomparable" and total["attribution"] is None assert "「西区」这次仍在查询结果第 4 行" in total["reasons"][0] # 对照：西区这次真的没有了（不在这次的快照里），照常是数据变化，按区域归因，占比之和为 1 gone = [["东区", 1200.0, 0.0], ["南区", 800.0, 0.0], ["北区", 700.0, 0.0], ["中区", 30.0, 0.0]] total = by_id(run_compare(cited("r1", before, span=(0, 2), sql=top), cited("r2", gone, span=(0, 2), sql=top)))["total"] assert (total["classification"], total["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES]) rows = total["attribution"]["rows"] assert [(r["key"], r["base"], r["target"], r["share"]) for r in rows] == [ (["北区"], None, 700.0, 7.0), (["西区"], 600.0, None, -6.0), (["东区"], 1200.0, 1200.0, 0.0), (["南区"], 800.0, 800.0, 0.0)] assert sum(exact(r["share"]) for r in rows) == 1 # 区域列设了遮罩：不拿它认是哪一区，只能按行号认，遮住的区域名不出现在结果里 def masked(run): run.masked = {a: frozenset({"region"}) for a in run.queries} return run result = run_compare(masked(cited("r1", with_middle, span=(0, 2))), masked(cited("r2", WITH_NORTH, span=(0, 2)))) assert by_id(result)["total"]["classification"] == "data_only" text = json.dumps(result, ensure_ascii=False) assert all(region not in text for region in ("东区", "南区", "西区", "北区", "中区")) def test_an_input_cites_the_whole_result_only_when_it_covers_every_row(): """怎么认整份结果（compare._whole）：容器本身就是一份查询结果的（没有行号、行段）、数组字段的行段从第一行到最后 一行的是；单格、只有一段行的不是；快照读不出来的不知道。Agent 的字段在真实运行里总带着行号或行段，容器这一支 只是兜底，造不出真实的运行来走它，在这里直接核对。""" query = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1, snap={"rows": [["东区"], ["南区"], ["西区"]]}) lost = compare._Query(artifact="q1", node_id="bot", tool=TOOL, ordinal=1) def cites(row=None, rows=None): return SimpleNamespace(row=row, rows=rows) assert compare._whole(cites(), query) is True assert compare._whole(cites(rows=(0, 2)), query) is True assert compare._whole(cites(rows=(0, 1)), query) is False assert compare._whole(cites(row=0), query) is False assert compare._whole(cites(rows=(0, 2)), lost) is None def test_cited_rows_on_an_unreadable_snapshot_are_not_taken_as_the_same(): """基准运行 Agent 那次查询的快照文件不在了：两边都引用第 1 至 3 行，可基准那次是不是整份结果核对不了，不报成 数据变化。""" base = cited("r1") del base.store[base.queries[1]] total = by_id(run_compare(base, cited("r2", WITH_NORTH, span=(0, 2))))["total"] assert total["classification"] == "incomparable" assert total["reasons"] == ["基准运行的查询快照无法读取，无法核对 Agent「区域分析」引用的是不是整份结果"] # 这次引用的是整份结果、基准那次的行段不同：同样核对不了是不是两边都是整份 total = by_id(run_compare(base, cited("r2", WITH_NORTH)))["total"] assert total["classification"] == "incomparable" and "无法核对" in total["reasons"][0] # 反过来只有这次运行的快照读不出来：说的是这次运行 def unreadable(run): del run.store[run.queries[1]] return run total = by_id(run_compare(cited("r1"), unreadable(cited("r2", WITH_NORTH, span=(0, 2)))))["total"] assert total["classification"] == "incomparable" assert total["reasons"] == ["这次运行的查询快照无法读取，无法核对 Agent「区域分析」引用的是不是整份结果"] # 基准运行只引用了一部分、这次的行段又不同：这次是不是整份结果都不是同一处，照实说行段换了 total = by_id(run_compare(cited("r1", WITH_NORTH, span=(0, 2)), unreadable(cited("r2", span=(1, 2)))))["total"] assert total["classification"] == "incomparable" assert total["reasons"] == ["Agent「区域分析」引用的行由第 1 至 3 行改为第 2 至 3 行，取的不是同一组数"] def test_an_element_of_a_cited_array_follows_its_row(): """口径卡取 Agent 数组字段里的第一个元素：Agent 交的行段换了起点，第一个元素落到了别的行，不是同一个数；起点没变 的照常是数据变化。""" def make(run_id, rows, span): run = Fake(run_id) run.tool("fetch", region_sql(), rows=rows) bot = run.agent("bot", region_sql(), rows=rows, span=span) row = span[0] item = {"path": "vars.kpi.amounts[0]", "value": rows[row][1], "node_id": "bot", "via": "agent_field", "field": "amounts[0]", "ref": f"Q1.r{row}.gmv", "artifact": bot, "locator": {"row": row, "column": "gmv"}, "eid": f"ev:cell:{bot[:8]}:{row}:gmv", "status": "verified"} run.card([metric("first", "vars.kpi.amounts[0]", rows[row][1], [item])]) return run first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), make("r2", NEXT_ROWS, (1, 2))))["first"] assert first["classification"] == "incomparable" assert first["reasons"] == ["Agent「区域分析」引用的行由第 1 行改为第 2 行，取的不是同一个数"] first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), make("r2", NEXT_ROWS, (0, 2))))["first"] assert (first["classification"], first["reasons"]) == ("data_only", [compare.DATA_FROM_QUERIES]) # 起点没变，可这次北区排在了第一行：基准运行取的是东区，东区这次还在第 2 行，取的不是同一个数 first = by_id(run_compare(make("r1", BASE_ROWS, (0, 2)), make("r2", [["北区", 90.0, 0.0], *NEXT_ROWS], (0, 3))))["first"] assert first["classification"] == "incomparable" assert first["reasons"] == ["Agent「区域分析」两次都引用第 1 行，「东区」这次仍在查询结果第 2 行，不在引用的行里，" "取的不是同一个数"] def test_no_split_when_an_old_leaf_now_cites_another_column(): """口径变了（v1 → v2），旧口径用到的 Agent 字段这次引用的是退款列：说不清它只是数据变了，不拆，说明为什么。""" def make(run_id, column, expression, factor, version): run = Fake(run_id) run.tool("fetch", region_sql()) bot = run.agent("bot", region_sql(), column=column) value = sum(r[COLUMNS.index(column)] for r in BASE_ROWS) * factor run.card([metric("total", expression, value, [amounts_in(bot, BASE_ROWS, column=column)])], version=version) return run total = by_id(run_compare(make("r1", "gmv", "sum(vars.kpi.amounts)", 1, "v1"), make("r2", "refund", "sum(vars.kpi.amounts) * 2", 2, "v2")))["total"] assert total["classification"] == "caliber_and_data" and total["decomposition"] is None assert total["decomposition_note"].startswith("无法确认旧口径用到的输入只是数据有变化") assert "refund" in total["decomposition_note"] def test_an_unknown_writer_on_one_side_is_named_plainly(): """一边记不出 vars.adj 是哪个节点写的：说清是哪一边没记录，不写「None」。""" def make(run_id, writer, value): run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code="x", assign_to="adj"),))) run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.code("calc", value, "adj") item = scalar_in("vars.adj", value, writer, "code" if writer else "state") run.card([metric("calc", "vars.adj * 2", value * 2, [item])]) return run calc = by_id(run_compare(make("r1", None, 10), make("r2", "calc", 12)))["calc"] assert calc["classification"] == "incomparable" assert not any("None" in r for r in calc["reasons"]) and any("基准运行" in r for r in calc["reasons"]) def test_a_list_run_input_is_compared_by_its_content(): """运行输入是列表：卡上只记了「[2 项]」，按入口节点的产出还原出完整值再比。""" def make(run_id, targets): run = Fake(run_id, inputs={"targets": targets}) run.add("node.started", "start", {"node_type": "input"}) run.add("node.finished", "start", {"artifact": run.put({"targets": targets})}) item = {"path": "input.targets", "value": f"[{len(targets)} 项]", "node_id": "start", "via": "input", "field": "targets", "status": "ok"} run.card([metric("goal", "sum(input.targets)", float(sum(targets)), [item])]) return run goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 500.0])))["goal"] assert goal["classification"] == "params_only" assert goal["reasons"] == ["运行输入「input.targets」的内容有变化"] goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 200.0])))["goal"] assert goal["classification"] == "unchanged" def test_a_list_run_input_without_its_full_value_is_not_taken_as_unchanged(): """运行输入是列表，入口节点当时的产出不在（还原不出完整值），卡上只记了「[2 项]」：大概一样、指标的值却变了， 核对不了它变没变，不报成数据变化；项数变了的照常是参数变化；指标的值没变的照常没变。""" def make(run_id, targets): run = Fake(run_id, inputs={"targets": targets}) item = {"path": "input.targets", "value": f"[{len(targets)} 项]", "node_id": "start", "via": "input", "field": "targets", "status": "ok"} run.card([metric("goal", "sum(input.targets)", float(sum(targets)), [item])]) return run goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 500.0])))["goal"] assert goal["classification"] == "incomparable" assert goal["reasons"] == ["运行输入「input.targets」是列表或对象，当时没有保存完整值，无法核对是否有变化"] goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 200.0, 50.0])))["goal"] assert goal["classification"] == "params_only" assert goal["reasons"] == ["运行输入「input.targets」的内容有变化"] goal = by_id(run_compare(make("r1", [100.0, 200.0]), make("r2", [100.0, 200.0])))["goal"] assert (goal["classification"], goal["reasons"]) == ("unchanged", []) def test_a_run_input_change_is_a_parameter_change(): def make(run_id, goal): run = Fake(run_id, inputs={"goal": goal}) run.add("node.started", "start", {"node_type": "input"}) run.add("node.finished", "start", {"artifact": run.put({"goal": goal})}) fetch = run.tool("fetch", region_sql()) run.agent("bot", region_sql()) run.card([metric("reach", f"round({EAST} / input.goal * 100, 1)", round(1200.5 / goal * 100, 1), [cell_in(0, "gmv", 1200.5, fetch), scalar_in("input.goal", goal, "start", "input")], unit="%", fmt="plain")]) return run result = run_compare(make("r1", 1000), make("r2", 1500)) reach = by_id(result)["reach"] assert reach["classification"] == "params_only" and any("input.goal" in r for r in reach["reasons"]) assert reach["delta_rendered"] == "-40.0 个百分点" assert any("goal" in n for n in result["notes"]) def test_metrics_and_cards_on_one_side_only(): def extra(fetch, bot, rows, bot_rows): return [*default_metrics(fetch, bot, rows, bot_rows), metric("west", "cell(nodes.fetch, 2, 'gmv')", rows[2][1], [cell_in(2, "gmv", rows[2][1], fetch)])] base = week("r1") target = week("r2", metrics=extra) target.card([metric("cnt", "len(vars.kpi.amounts)", 3, [amounts_in(target.queries[1], BASE_ROWS)], unit="个", decimals=None)], node_id="card2", caliber="补充口径") result = run_compare(base, target) metrics = by_id(result) assert metrics["west"]["classification"] == "added" and metrics["west"]["base"] is None assert metrics["cnt"]["classification"] == "added" and metrics["cnt"]["card_id"] == "card2" assert result["summary"]["added"] == 2 swapped = by_id(run_compare(target, base)) assert swapped["west"]["classification"] == "removed" and swapped["west"]["target"] is None def test_an_unreadable_card_makes_its_metrics_incomparable(): base, target = week("r1"), week("r2", NEXT_ROWS) base.tampered.add(base.cards[0][1]) base.cards = [] # 接口只交回复验过哈希的口径卡 result = run_compare(base, target) assert {m["classification"] for m in result["metrics"]} == {"incomparable"} assert any("哈希" in r for r in by_id(result)["east"]["reasons"]) def test_a_card_executed_several_times_uses_the_last_one(): base = week("r1") target = Fake("r2") fetch = target.tool("fetch", region_sql(), rows=NEXT_ROWS) bot = target.agent("bot", region_sql(), rows=NEXT_ROWS) target.card(default_metrics(fetch, bot, BASE_ROWS, BASE_ROWS)) target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS), runs=2) result = run_compare(base, target) assert by_id(result)["east"]["target"] == 1500.5 assert any("执行了 3 次" in n for n in result["notes"]) def test_verify_queries_are_left_out(): base = week("r1") target = Fake("r2") fetch = target.tool("fetch", region_sql(), rows=NEXT_ROWS) bot = target.agent("bot", region_sql(), rows=NEXT_ROWS, verify=["SELECT COUNT(*) AS n FROM orders"]) target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS)) result = run_compare(base, target) assert [(q["node_id"], q["structure"]) for q in result["queries"]] == [("fetch", "same"), ("bot", "same")] # -------------------------------------------------------------------------- # 查询配对、模板对比 # -------------------------------------------------------------------------- def test_queries_fall_back_to_normalized_sql_pairing(): """节点换了 ID：按「节点 + 第几次 + 工具」配不上，再按规范化 SQL 配（大小写、空白不同也认）。""" base = week("r1") target = Fake("r2", graph=weekly_graph()) target.tool("pull", region_sql(WEEK37).lower(), rows=NEXT_ROWS) target.agent("bot", region_sql(WEEK37), rows=NEXT_ROWS) result = run_compare(base, target) pull = result_queries(result, "pull") assert pull["structure"] == "params_only" and pull["base_sql"] == region_sql() assert len(result["queries"]) == 2 def test_an_agent_asking_in_another_order_is_paired_by_sql(): """Agent 这次换了查询的先后：同一节点里先按规范化 SQL 配，不按第几次硬配成两条「结构变了」。""" by_region = region_sql() by_day = "SELECT day, SUM(amount) AS gmv FROM orders WHERE day >= '2026-08-31' GROUP BY day" def make(run_id, sqls): run = Fake(run_id) run.add("node.started", "bot", {"node_type": "agent"}) for i, sql in enumerate(sqls): artifact = run.snapshot(sql, ["k", "gmv"], [["甲", 1.0]]) run.add("tool.start", "bot", {"tool": "db_query__shop", "args": {"sql": sql}, "call_id": f"c{i}"}) run.add("tool.end", "bot", {"tool": "db_query__shop", "call_id": f"c{i}", "query_artifact": artifact}) run.add("node.finished", "bot", {"duration_ms": 1, "attempt": 1}) return run result = run_compare(make("r1", [by_region, by_day]), make("r2", [by_day.replace("2026-08-31", "2026-09-07"), region_sql(WEEK37)])) pairs = [(q["structure"], q["base_sql"][:20]) for q in result["queries"]] assert pairs == [("params_only", by_day[:20]), ("params_only", by_region[:20])] def test_swapped_queries_with_the_same_literals_pair_up_exactly(): """Agent 两次都查了东区和南区，只是先后互换：先配字面量也相同的，两条都没变；下游的数照常是数据变化。""" def region(name): return f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{name}'" def make(run_id, regions, value, gmv): run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code="x", assign_to="adj"),))) run.plain_agent("bot", [region(r) for r in regions], {"text": "ok"}, ["gmv"], [[gmv]], types={"gmv": "number"}) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, None, None)) return run result = run_compare(make("r1", ["东区", "南区"], 10, 1.0), make("r2", ["南区", "东区"], 12, 2.0)) assert [(q["structure"], q["literals"]) for q in result["queries"]] == [("same", []), ("same", [])] # Agent 两次交出的一模一样（只是查询的先后换了）：它没带进模型本身的差异 assert by_id(result)["calc"]["classification"] == "data_only" def test_renamed_nodes_pair_their_queries_by_literals_first(): """节点换了 ID，同一个模板查了东区和南区、这次先后互换：跨节点按 SQL 配时同样先配字面量也相同的，两条都没变， 不会配成两条「参数不同」。""" def make(run_id, node_id, regions): run = Fake(run_id) for region in regions: run.tool(node_id, f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}'", ["gmv"], [[1.0]], types={"gmv": "number"}) return run result = run_compare(make("r1", "fetch", ["东区", "南区"]), make("r2", "pull", ["南区", "东区"])) assert [(q["node_id"], q["structure"], q["literals"]) for q in result["queries"]] == [ ("pull", "same", []), ("pull", "same", [])] def test_a_switched_data_source_is_a_structure_change(): """SQL 一字不差、换了数据源：查的不是同一份数据。""" result = run_compare(week("r1"), week("r2", NEXT_ROWS, source="shop_backup")) fetch = result_queries(result, "fetch") assert fetch["structure"] == "changed" and any("shop_backup" in d for d in fetch["differences"]) assert by_id(result)["east"]["classification"] == "sql_changed" def test_a_literal_edited_in_the_workflow_is_a_template_change(): """图里写死的筛选值改了（'paid' 改成 'refunded'）：执行的 SQL 只差字面量，但这是改模板，不是运行时代入的参数。""" paid, refunded = region_sql(extra=" AND status = 'paid'"), region_sql(extra=" AND status = 'refunded'") result = run_compare(week("r1", sql=paid, graph=weekly_graph(paid)), week("r2", NEXT_ROWS, sql=refunded, graph=weekly_graph(refunded))) east = by_id(result)["east"] assert east["classification"] == "sql_changed", east["reasons"] assert any("区域取数" in r and "'paid'" in r for r in east["reasons"]) assert east["decomposition"] is None and east["attribution"] is None # 查询表照实写：执行的 SQL 确实只差这一处字面量 assert result_queries(result, "fetch")["structure"] == "params_only" def test_reformatting_the_workflow_sql_is_not_a_change(): loose = region_sql().replace(" FROM ", "\n FROM ").lower() result = run_compare(week("r1"), week("r2", NEXT_ROWS, sql=loose, graph=weekly_graph(loose))) assert {m["classification"] for m in result["metrics"]} == {"data_only"} # 代码节点在它下游：取数节点不算改了配置，变化照常来自上游数据 def make(run_id, sql, rows, value): run = Fake(run_id, graph=weekly_graph(sql, after_fetch=(calc_node(),))) run.tool("fetch", sql, rows=rows) run.agent("bot", sql, rows=rows) run.code("calc", value, "adj") run.card(code_metrics(value)(None, None, rows, rows)) return run calc = by_id(run_compare(make("r1", region_sql(), BASE_ROWS, 10), make("r2", loose, NEXT_ROWS, 12)))["calc"] assert (calc["classification"], calc["reasons"]) == ("data_only", [compare.DATA_FROM_UPSTREAM_SCOPED]) def test_renaming_a_node_is_not_a_change(): graph = weekly_graph() graph["nodes"][2]["data"]["label"] = "区域分析（新）" result = run_compare(week("r1"), week("r2", NEXT_ROWS, graph=graph)) assert {m["classification"] for m in result["metrics"]} == {"data_only"} [renamed] = result["template"]["nodes"] assert (renamed["node_id"], renamed["keys"]) == ("bot", ["label"]) MERGE_SQL = "SELECT region, gmv FROM a ORDER BY region" OUTER_SQL = "SELECT region, gmv FROM m ORDER BY region" MERGED_TYPES = {"region": "text", "gmv": "number"} def merged(run_id, sql, rows, *, template=None, nested=False, twice=False, lost=False, bare=False, unsealed=False): """取数节点查一次，合并节点把它当输入 a 再执行一条合并 SQL（nested：再套一层合并；twice：这次多接一个输入 b）， 口径卡取合并结果的一格。追不下去的几种（只用于一层合并）：lost 合并的快照文件不在了；bare 合并没记录输入（快照、 merge.end 里都没有）；unsealed 上游那次查询不在封存范围内（事件里没有它）。""" nodes = [node("start", "input", "开始"), node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": template or sql}), node("join", "merge", "合并", sql=MERGE_SQL, inputs={"a": "fetch", **({"b": "fetch"} if twice else {})})] if nested: nodes.append(node("outer", "merge", "再合并", sql=OUTER_SQL, inputs={"m": "join"})) nodes.append(node("card", "metrics", "口径卡", caliber="周报口径", caliber_version="v1")) run = Fake(run_id, graph=chain(*nodes)) fetch = run.tool("fetch", sql, rows=rows) table = [[r[0], r[1]] for r in rows] inputs = [("a", "fetch", fetch)] + ([("b", "fetch", fetch)] if twice else []) top, owner = run.merge("join", MERGE_SQL, inputs, ["region", "gmv"], table, types=MERGED_TYPES), "join" if nested: top, owner = run.merge("outer", OUTER_SQL, [("m", "join", top)], ["region", "gmv"], table, types=MERGED_TYPES), "outer" run.card([metric("east", f"cell(nodes.{owner}, 0, 'gmv')", table[0][1], [cell_in(0, "gmv", table[0][1], top, node_id=owner)], name="东区金额")]) if lost: del run.store[top] if bare: run.store[top] = {k: v for k, v in run.store[top].items() if k != "inputs"} run.events = [(seq, etype, nid, {k: v for k, v in data.items() if k != "inputs"} if etype == "merge.end" else data) for seq, etype, nid, data in run.events] if unsealed: run.events = [e for e in run.events if e[2] != "fetch"] return run def test_a_merged_cell_follows_the_queries_feeding_the_merge(): """合并 SQL 一字没改，喂它的取数 SQL 加了筛选：读合并结果的指标是 SQL 变了，不拆、不归因（嵌套的合并也追到底）。""" paid = region_sql(extra=" AND status = 'paid'") for nested in (False, True): east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, nested=nested), merged("r2", paid, NEXT_ROWS, nested=nested)))["east"] assert east["classification"] == "sql_changed", (nested, east["reasons"]) assert east["decomposition"] is None and east["attribution"] is None assert any("区域取数" in r and "SQL" in r for r in east["reasons"]), east["reasons"] # 上游只是代进去的日期不同：参数变化 template = region_sql(("{{ input.start }}", "{{ input.end }}")) east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, template=template), merged("r2", region_sql(WEEK37), NEXT_ROWS, template=template)))["east"] assert east["classification"] == "params_only", east["reasons"] # 合并 SQL 和上游都没变：数据变化 east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS), merged("r2", region_sql(), NEXT_ROWS)))["east"] assert east["classification"] == "data_only", east["reasons"] # 合并节点这次多接了一个输入 east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS), merged("r2", region_sql(), NEXT_ROWS, twice=True))) assert east["east"]["classification"] == "sql_changed" and any("「b」" in r for r in east["east"]["reasons"]) def test_a_merged_cell_whose_inputs_cannot_be_followed(): """合并的快照文件不在了：按 merge.end 记的输入接着追，上游加了筛选照样认得出。合并没记录输入、上游那次查询不在 封存范围内：说不清上游的取数变没变，不报成数据变化。""" paid = region_sql(extra=" AND status = 'paid'") east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, lost=True), merged("r2", paid, NEXT_ROWS, lost=True)))["east"] assert east["classification"] == "sql_changed", east["reasons"] assert any("「a」" in r and "区域取数" in r for r in east["reasons"]), east["reasons"] for broken, said in (({"bare": True}, "没有记录它的输入"), ({"unsealed": True}, "不在封存范围内")): east = by_id(run_compare(merged("r1", region_sql(), BASE_ROWS, **broken), merged("r2", region_sql(), NEXT_ROWS, **broken)))["east"] assert east["classification"] == "incomparable", (broken, east["reasons"]) assert any(said in r for r in east["reasons"]), (broken, east["reasons"]) def test_an_agent_reconfigured_with_the_same_sql_is_a_template_change(): """Agent 改了提示词、SQL 一字不差：它按出处交来的字段可能换了取法，不能报成数据变化。""" graph = weekly_graph() graph["nodes"][2]["data"]["config"]["prompt"] = "按区域列出金额，从大到小" result = run_compare(week("r1"), week("r2", NEXT_ROWS, graph=graph)) metrics = by_id(result) assert metrics["total"]["classification"] == "sql_changed" assert any("区域分析" in r for r in metrics["total"]["reasons"]) # 取数节点的格子不经过这个 Agent：照旧是数据变化 assert metrics["east"]["classification"] == "data_only" def test_a_merge_query_is_compared_by_its_merge_sql(): """合并查询只记在节点的台账里（没有 tool.end），SQL 在快照和 merge.end 里；快照文件不在了也比得了。""" def make(run_id, sql, *, drop=False): run = Fake(run_id) run.tool("fetch", region_sql()) snap = {"columns": ["region", "gmv"], "rows": [["东区", 1.0]], "truncated": False, "sql": sql, "source": MERGE_SOURCE, "sources": ["shop"]} artifact = run.put(snap) run.add("node.started", "join", {"node_type": "merge"}) run.add("merge.end", "join", {"sql": sql, "query_artifact": artifact, "rows": 1}) run.add("node.finished", "join", {"duration_ms": 1, "attempt": 1, "evidence": [ {"kind": "query", "node_id": "join", "exec": 1, "artifact": artifact, "source": MERGE_SOURCE, "columns": ["region", "gmv"], "rows": 1, "truncated": False}]}) if drop: del run.store[artifact] return run result = run_compare(make("r1", "SELECT region, gmv FROM a WHERE gmv > 10"), make("r2", "SELECT region, gmv FROM a WHERE gmv > 20")) join = result_queries(result, "join") assert (join["ordinal"], join["structure"], join["literals"]) == (1, "params_only", [{"base": "10", "target": "20"}]) result = run_compare(make("r1", "SELECT region, gmv FROM a WHERE gmv > 10", drop=True), make("r2", "SELECT region, gmv FROM a WHERE gmv > 10 AND region <> '西区'")) join = result_queries(result, "join") assert join["structure"] == "changed" and join["base_sql"] == "SELECT region, gmv FROM a WHERE gmv > 10" assert "无法读取" in join["problem"] def test_a_node_run_in_a_loop_is_paired_execution_by_execution(): def make(run_id, regions): run = Fake(run_id) for region in regions: run.tool("fetch", f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}'", ["gmv"], [[1.0]], types={"gmv": "number"}) return run result = run_compare(make("r1", ["东区", "南区", "西区"]), make("r2", ["东区", "北区"])) assert [(q["ordinal"], q["structure"]) for q in result["queries"]] == [(1, "same"), (2, "params_only"), (3, "removed")] assert result["queries"][1]["literals"] == [{"base": "'南区'", "target": "'北区'"}] def test_unpaired_queries_are_added_or_removed(): base = week("r1") target = week("r2") target.tool("extra", "SELECT COUNT(*) AS n FROM refunds", ["n"], [[7]], types={"n": "number"}) result = run_compare(base, target) extra = result_queries(result, "extra") assert extra["structure"] == "added" and extra["base_sql"] is None and extra["target_rows"] == 1 removed = result_queries(run_compare(target, base), "extra") assert removed["structure"] == "removed" and removed["target_sql"] is None def test_the_template_lists_added_removed_and_reconfigured_nodes(): base_graph = weekly_graph(between=(node("calc", "code", "调整", code="x = 1"),)) target_graph = weekly_graph(region_sql(WEEK37), between=(node("check", "dual_check", "双盲复算"),)) target_graph["nodes"][-2]["data"]["label"] = "写周报（新）" target_graph["nodes"][-1]["data"]["config"] = {"fields": [{"name": "周报", "value": "{{ nodes.write.text }}"}]} base = week("r1", graph=base_graph) target = week("r2", graph=target_graph, version="v1") base.version, target.version = 3, 4 template = run_compare(base, target)["template"] assert (template["base_version"], template["target_version"], template["same_graph"]) == (3, 4, False) changes = {n["node_id"]: (n["change"], n["keys"]) for n in template["nodes"]} assert changes == {"fetch": ("config", ["args.sql"]), "check": ("added", []), "write": ("config", ["label"]), "out": ("config", ["fields"]), "calc": ("removed", [])} labels = {n["node_id"]: n["labels"] for n in template["nodes"]} assert labels["write"] == ["节点名称"] and labels["out"] == ["成果字段"] # 配置的值（这里是 SQL 和模板）一个字都不展示 assert "2026-09" not in json.dumps(template, ensure_ascii=False) def test_a_missing_snapshot_file_is_reported_and_withheld(): """基准运行的快照文件不在了：SQL 从 tool.start 的参数里取，结构照比；输入的值核对不了遮没遮，不给；不归因。""" base, target = week("r1"), week("r2", NEXT_ROWS) fetch = base.queries[0] del base.store[fetch] result = run_compare(base, target) query = result_queries(result, "fetch") assert query["structure"] == "same" and query["base_sql"] == region_sql() and query["base_rows"] is None assert "无法读取" in query["problem"] east = by_id(result)["east"] assert east["classification"] == "data_only" and east["inputs"][0]["base"] is None assert east["attribution"]["rows"] == [] and "无法读取" in east["attribution"]["note"] def test_runs_without_metric_cards_still_compare_queries(): base, target = Fake("r1"), Fake("r2") base.tool("fetch", region_sql()) target.tool("fetch", region_sql(WEEK37), rows=NEXT_ROWS) result = run_compare(base, target) assert result["metrics"] == [] and sum(result["summary"].values()) == 0 assert any("没有口径卡" in n for n in result["notes"]) assert [q["structure"] for q in result["queries"]] == ["params_only"] def test_the_comparison_does_not_touch_the_runs(): base, target = week("r1"), week("r2", NEXT_ROWS) before = (list(base.events), list(target.events), dict(base.store), dict(target.store)) run_compare(base, target) assert (base.events, target.events, base.store, target.store) == before