口径账本：合并查询的格照实写为什么没比对；账本测试按用例各用一个数据源隔开

- ledger/match.py：合并查询的快照不和口径比（它的数来自各个输入的查询），引用了它的格以前标成「没有比对记录」，现在写明
  「合并查询的格不和口径比对」；它不进工件的 queries，统计不数它。
- 测试：账本的用例以前在开头把别的用例发布的版本一律退役，好让候选口径只剩自己的——并行时会碰到同一个进程里别的测试模块的
  数据。改成每个用例一个名字不同的数据源（候选口径本来就按数据源预筛），不动别人的数据。补启动补比随调度开关起后台任务的用例。

backend/app/ledger/match.py

@@ -67,6 +67,7 @@ SNAPSHOT_TAMPERED = "查询快照与哈希不一致，疑似被修改，不比
 SNAPSHOT_MISSING = "查询快照读不出来，不比对"
 NO_SQL = "查询快照里没有 SQL，不比对"
 PARSE_FAILED = "这次的 SQL 无法解析，没有比对"
+MERGED = "合并查询的格不和口径比对（它的数来自各个输入的查询）"
 MORE_CANDIDATES = "另有 {n} 份相关口径没有比对（每条查询最多比 {max} 份）"
 
 #: 不比对的原因（GET 账本时照实说）
@@ -364,6 +365,10 @@ def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None],
     by_key = {c.key: {"level": c.level, "fields": c.summary()["fields"], "expressions": list(c.expressions)}
               for c in considered}
     by_artifact = {q["artifact"]: q for q in results}
+    # 合并查询的快照不比（read_queries 跳过了它），引用了它的格照实写为什么没比对；不进工件的 queries，统计不数它
+    for artifact in sealed.queries:
+        by_artifact.setdefault(artifact, {"artifact": artifact, "status": "skipped", "reason": MERGED,
+                                          "candidates": []})
     sealed_cards = {artifact for _, artifact in sealed.ledger}
 
     def load_card(artifact: str) -> dict[str, Any] | None:

backend/tests/ledger_samples.py

@@ -1,6 +1,7 @@
 """口径账本测试共用的虚构样本：一个店铺库、一份受管口径「周销售」、跑一次探索运行。数据全是编的。
 
-- 店铺库 shop（SQLite）：orders(id, region, status, amount, paid_at, phone)，九单，状态有 paid / refunded / open。
+- 店铺库（SQLite）：orders(id, region, status, amount, paid_at, phone)，九单，状态有 paid / refunded / open。每个用例一个
+  名字不同的数据源（fresh_source）：候选口径按数据源预筛，别的用例发布的口径不会混进来，也不必去动别的用例的数据。
 - 口径「周销售」v1（受管）：入口两个字段 start / end，取数节点 fetch 的 SQL 是
   SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}'，
   口径卡 card 上一个指标 gmv = cell(nodes.fetch, 0, 'gmv')。
@@ -55,11 +56,30 @@ def build_shop(path) -> None:
     db.close()
 
 
+#: 这个用例的数据源名（fresh_source 换）：图里的查询工具是 db_query__<它>
+CURRENT = {"source": "shop"}
+
+
+def tool() -> str:
+    return f"db_query__{CURRENT['source']}"
+
+
+async def fresh_source(path) -> str:
+    """建库、起一个这个用例专用的数据源（名字不重复），返回数据源名。"""
+    import uuid
+
+    build_shop(path)
+    CURRENT["source"] = f"shop_{uuid.uuid4().hex[:8]}"
+    await set_source(path)
+    return CURRENT["source"]
+
+
 async def set_source(path, *, options: dict[str, Any] | None = None) -> str:
+    name = CURRENT["source"]
     async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
+        row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
         if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
+            row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True)
             session.add(row)
         row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
         row.options = dict(options or {})
@@ -73,7 +93,7 @@ def caliber_graph(sql: str = TEMPLATE, *, expression: str = "cell(nodes.fetch, 0
     return chain(
         node("start", "input", fields=[{"name": "start", "required": True}, {"name": "end", "required": True},
                                        {"name": "store", "default": "all"}]),
-        node("fetch", "tool", tool="db_query__shop", args={"sql": sql}),
+        node("fetch", "tool", tool=tool(), args={"sql": sql}),
         node("card", "metrics", caliber="周销售", caliber_version="v1", metrics=[
             {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands", "expression": expression}]),
         node("write", "report", instructions="写周销售"),
@@ -99,7 +119,7 @@ async def publish(name: str = "周销售", *, graph: dict[str, Any] | None = Non
 
 
 def explore_graph(sql: str, *, card_expression: str | None = None) -> dict[str, Any]:
-    parts = [node("start", "input"), node("q", "tool", tool="db_query__shop", args={"sql": sql})]
+    parts = [node("start", "input"), node("q", "tool", tool=tool(), args={"sql": sql})]
     if card_expression:
         parts.append(node("mine", "metrics", caliber="我的口径", caliber_version="v1", metrics=[
             {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands",

backend/tests/test_ledger_api.py

@@ -18,7 +18,7 @@ from httpx import ASGITransport, AsyncClient
 from sqlalchemy import select
 
 from app.db.base import SessionLocal
-from app.db.models import AuditEvent, Workflow, WorkflowVersion
+from app.db.models import AuditEvent, Workflow
 from app.engine.runner import run_manager
 from app.ledger import match as ledger_match
 from app.main import app
@@ -27,16 +27,9 @@ from tests import ledger_samples as S
 
 @pytest.fixture(autouse=True)
 async def engine_up(tmp_path):
-    from app.db.base import utcnow
-
     await run_manager.setup()
     path = tmp_path / "shop.db"
-    S.build_shop(path)
-    await S.set_source(path)
-    async with SessionLocal() as session:
-        for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars():
-            row.retired_at = utcnow()
-        await session.commit()
+    await S.fresh_source(path)
     yield path
     await ledger_match.drain()
     await run_manager.shutdown()
@@ -115,8 +108,8 @@ async def test_masked_binding_values_and_sql_fragments_are_withheld(client, monk
     await S.publish(graph=S.caliber_graph(template))
     graph = S.chain(
         S.node("start", "input"),
-        S.node("p", "tool", tool="db_query__shop", args={"sql": "SELECT phone FROM orders ORDER BY id"}),
-        S.node("q", "tool", tool="db_query__shop",
+        S.node("p", "tool", tool=S.tool(), args={"sql": "SELECT phone FROM orders ORDER BY id"}),
+        S.node("q", "tool", tool=S.tool(),
                args={"sql": "SELECT SUM(amount) AS gmv FROM orders WHERE phone = '13800000001'"}),
         S.node("write", "report", instructions="写答案"),
         S.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}]))

backend/tests/test_ledger_match.py

@@ -30,16 +30,7 @@ from tests import ledger_samples as S
 async def engine_up(tmp_path):
     await run_manager.setup()
     path = tmp_path / "shop.db"
-    S.build_shop(path)
-    await S.set_source(path)
-    # 每个用例自己发布口径：先把别的用例发布的退役掉，候选口径只看这一个用例的
-    from app.db.models import WorkflowVersion
-    from app.db.base import utcnow
-
-    async with SessionLocal() as session:
-        for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars():
-            row.retired_at = utcnow()
-        await session.commit()
+    await S.fresh_source(path)
     yield path
     await ledger_match.drain()
     await run_manager.shutdown()
@@ -68,7 +59,7 @@ async def test_an_equivalent_query_is_proven_and_its_number_comes_from_the_calib
     assert content["format"] == store.FORMAT and content["algo"] == store.ALGO and content["run_id"] == row.id
     assert content["manifest_hash"] == row.manifest_hash
     [query] = content["queries"]
-    assert query["status"] == "proven" and query["source"] == "shop" and query["tables"] == ["orders"]
+    assert query["status"] == "proven" and query["source"] == S.CURRENT["source"] and query["tables"] == ["orders"]
     [cand] = query["candidates"]
     assert cand["caliber"] == f"{wf}@1" and cand["level"] == "governed" and cand["node_id"] == "fetch"
     assert cand["binding"] == {"start": "2026-09-01", "end": "2026-09-08"}
@@ -283,7 +274,7 @@ async def test_only_sealed_query_snapshots_are_compared(monkeypatch):
 
     await S.publish()
     row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。")
-    await put_json({"sql": S.WITH_REFUNDS, "source": "shop", "columns": ["gmv"], "rows": [[1]]},
+    await put_json({"sql": S.WITH_REFUNDS, "source": S.CURRENT["source"], "columns": ["gmv"], "rows": [[1]]},
                    kind="query_snapshot", run_id=row.id, node_id="q")
     await ledger_match.match_run(row.id, trigger="on_demand")
     recorded, _ = await matched(row.id)
@@ -298,3 +289,17 @@ async def test_cells_inside_a_table_are_labelled_the_same_way(monkeypatch):
     [labels] = [r["labels"] for r in recorded.content["reports"].values()]
     assert len(labels) >= 2 and all(label["kind"] == "caliber" and label["caliber"] == f"{wf}@1"
                                     for label in labels.values())
+
+
+async def test_backfill_starts_in_the_background_when_the_scheduler_is_on(monkeypatch):
+    started = []
+
+    async def fake_backfill(**kw):
+        started.append(kw)
+        return 0
+
+    monkeypatch.setattr(settings, "scheduler", True)
+    monkeypatch.setattr(ledger_match, "backfill", fake_backfill)
+    assert ledger_match.start_backfill() is True
+    await ledger_match.drain()
+    assert started == [{}]