运行对比：补合并查询、循环里多次执行的配对测试，口径卡读不出来时的说法补上「的」

合并查询只记在节点台账里、SQL 另在 merge.end 里，快照文件不在了照样比得了 SQL；循环里同一个节点执行几次的按先后
一一配上，多出来的一次标「删除」。口径卡指标集读不出来、被改过时的说明改成「口径卡「X」的指标集与哈希不一致」。

backend/app/engine/compare.py

@@ -317,11 +317,12 @@ def _cards(side: Side) -> dict[str, _Card]:
 
 
 def _card_problem(load: Callable[[str], Any], artifact: str) -> str:
+    """指标集为什么用不了，接在「口径卡「X」」后面。"""
     try:
         content = load(artifact)
     except ValueError:
-        return "指标集与哈希不一致，疑似被修改"
-    return "无法读取指标集" if content is None else "指标集的内容不完整"
+        return "的指标集与哈希不一致，疑似被修改"
+    return "的指标集无法读取" if content is None else "的指标集内容不完整"
 
 
 def _plan_card(side: Side, card: _Card, queries: list[_Query]) -> None:

backend/tests/test_compare_api.py

@@ -28,7 +28,6 @@ from app.db.models import Artifact, DataSource, Run, RunEvent, Workflow
 from app.engine.runner import run_manager, verify_manifest
 from app.main import app
 from tests.fixtures.compare_runs import (
-    BASE_ROWS,
     EAST,
     NEXT_ROWS,
     Fake,

backend/tests/test_compare_engine.py

@@ -16,6 +16,7 @@ import json
 from decimal import Decimal
 
 from app.engine import compare
+from app.engine.merge_query import MERGE_SOURCE
 from tests.fixtures.compare_runs import (
     BASE_ROWS,
     EAST,
@@ -530,6 +531,47 @@ def test_an_agent_reconfigured_with_the_same_sql_is_a_template_change():
     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")