分析相关用例的数据源夹具前后都删干净：tests/fixtures/sources 加 sqlite_source（建之前先把同名的旧源连同数据目录、剖析和现查缓存删掉，交回新 id，可顺手探好表结构）和 forget_source（删源、数据目录随外键级联删除，再清掉 coverage 现查缓存 _LIVE 里按这个源 id 记的表范围），tests/conftest.py 加共用夹具 scratch_source（按它们建、测试结束时删）。以前各文件的 shop 夹具复用名叫 shop 的源却不清它的数据目录（复核时一条写了目录的端到端串扰了同进程后跑的用例），现查缓存按源 id 记、5 分钟内同一个进程里可能拿到别的用例的旧范围。test_analysis_e2e、test_analysis_e2e_claims、test_analysis_tools、test_metric_comparability、test_contract_analysis_claims 的 shop 夹具改用它（直接建遮罩源的两条用例一并改，去掉 try/finally）；test_analysis_tools 开头「算子还没落地」的说明改成照实写替身和真算子的端到端各在哪。测试：旧源留下的数据目录和现查范围在建新源时清掉，删源后现查缓存里没有它

backend/tests/conftest.py

@@ -111,6 +111,24 @@ def pytest_collection_modifyitems(config, items) -> None:
             item.add_marker(skip)
 
 
+@pytest.fixture
+async def scratch_source():
+    """测试里建的 SQLite 数据源：source_id = await scratch_source("shop", 库文件路径, options={...})。同名的旧源先删掉（连同
+    它的数据目录、剖析和现查缓存，tests/fixtures/sources.sqlite_source），测试结束时再删一遍、清掉现查缓存（forget_source）。
+    分析相关的用例都建名叫 shop 的源：别的用例留下的数据目录、5 分钟内现查过的表范围会改掉这次的判定。"""
+    from tests.fixtures.sources import forget_source, sqlite_source
+
+    made: list[str] = []
+
+    async def make(name: str, path, **kw) -> str:
+        made.append(await sqlite_source(name, path, **kw))
+        return made[-1]
+
+    yield make
+    for source_id in made:
+        await forget_source(source_id)
+
+
 @pytest.fixture(scope="session")
 def scenic_db(tmp_path_factory) -> str:
     """合成的景区业务库（tests/fixtures/catalog/scenic.py）：50 张表、真外键与只能靠命名推断的关系并存。

backend/tests/fixtures/sources.py

@@ -3,9 +3,15 @@
 数据源表在每个测试进程的测试库里是共用的：留下来的源会出现在助手的数据源清单里（50 张表的合成库还会
 触发挑表、多一次模型调用），别的用例的结果就跟着执行顺序变。阶段 1B 就是这样被绊过一次。
 删数据源时它的数据目录随外键级联删除（测试库开着 PRAGMA foreign_keys=ON）。
+
+分析相关的用例都建名叫 shop 的源（sqlite_source / forget_source，tests/conftest.py 的 scratch_source 夹具按它们建、按它们删）：
+别的用例留下的数据目录（金额标成了流量，窗口长度不同就从有条件可比变成不可比）、coverage 的现查缓存里 5 分钟内按同一个
+源 id 记下的表范围（另一份库的日期范围），都会改掉这次的判定。所以建之前先把同名的旧源连同这两样删干净，用完再删一遍。
 """
 from __future__ import annotations
 
+from typing import Any
+
 from sqlalchemy import delete, select
 
 from app.data.engine import engines
@@ -26,6 +32,42 @@ async def drop_source(source_id: str | None = None, *, name: str | None = None)
     await engines.invalidate(source_id)
 
 
+async def sqlite_source(name: str, path: Any, *, options: dict[str, Any] | None = None, schema: bool = False) -> str:
+    """建一个指向这份 SQLite 库文件的只读数据源（启用），交回 id。同名的旧源先删掉（forget_source：连同它的数据目录、剖析
+    和现查缓存），这次的源是新的 id。schema 为真时顺手探好表结构（要写数据目录的用例用）。用完调 forget_source。"""
+    await forget_source(name=name)
+    async with SessionLocal() as session:
+        row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True,
+                         options=dict(options or {}))
+        session.add(row)
+        await session.flush()
+        if schema:
+            from app.data.introspect import introspect
+
+            row.schema_cache = await introspect(row)
+        await session.commit()
+        source_id = row.id
+    await engines.invalidate(source_id)
+    return source_id
+
+
+async def forget_source(source_id: str | None = None, *, name: str | None = None) -> None:
+    """按 id 或名字删一个数据源（drop_source：数据目录、剖析随外键级联删除，连接池作废），再清掉 coverage 现查缓存里它的
+    表范围（data/coverage 的 _LIVE，按数据源 id 记、5 分钟有效：同一个进程里后面的用例会拿到这份旧范围）。不存在就什么都
+    不做。"""
+    from app.data import coverage
+
+    if source_id is None:
+        async with SessionLocal() as session:
+            source_id = (await session.execute(
+                select(DataSource.id).where(DataSource.name == name))).scalar_one_or_none()
+        if source_id is None:
+            return
+    await drop_source(source_id)
+    for key in [k for k in coverage._LIVE if k and k[0] == source_id]:
+        coverage._LIVE.pop(key, None)
+
+
 async def introspect_now(client, source_id: str, *, expect: int = 202) -> dict:
     """探查结构并等它做完（探查是后台任务：POST /introspect 回 202 和任务，next-round 03 A1），返回数据源此刻的样子
     （DataSourceOut）。expect 不是 202 时只发请求、断言状态码，返回响应体。"""

backend/tests/test_analysis_e2e.py

@@ -19,9 +19,8 @@ import app.analysis as analysis_ops
 from app.analysis import contract
 from app.core import artifact_store
 from app.core.artifact_store import canonical_json, content_hash
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine.runner import run_manager
 
 TOOL = "db_query__shop"
@@ -64,24 +63,15 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，用完再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders())
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 def node(nid, ntype, **config):
@@ -349,3 +339,49 @@ async def test_a_ratio_between_two_regions_is_not_a_period_comparison(south_week
     assert issuance["tier"] == "degraded"
     assert issuance["gaps"] == ["指标「华东对华南的倍数」两期不可比：两边除了日期，筛选的还不一样"]
     recomputes(artifact_store.load(found["artifact"]))
+
+
+# --------------------------------------------------------------------------
+# 测试夹具：复用名叫 shop 的源，前后都删干净
+# --------------------------------------------------------------------------
+
+
+async def test_a_scratch_source_starts_clean_and_forgets_its_probes(tmp_path):
+    """分析相关的用例都建名叫 shop 的源（tests/conftest.py 的 scratch_source）：别的用例留下的数据目录、coverage 的现查缓存里
+    按同一个源 id 记下的表范围（5 分钟有效）会改掉这次的判定。sqlite_source 建的源是新的 id，同名的旧源连同数据目录、现查
+    缓存先删掉；forget_source 删源（数据目录随外键级联删除）、清掉现查缓存里它的表范围。"""
+    from app.analysis.probe import make_prober
+    from app.data import catalog as data_catalog
+    from app.data import coverage
+    from app.data.catalog import make_item
+    from app.db.models import CatalogNote, DataSource
+    from tests.fixtures.sources import forget_source, sqlite_source
+
+    path = tmp_path / "other.db"
+    db = sqlite3.connect(path)
+    db.executescript("CREATE TABLE orders (day TEXT, amount INTEGER); INSERT INTO orders VALUES ('2026-01-05', 1);")
+    db.close()
+
+    async def leave_traces(source_id):
+        async with SessionLocal() as session:
+            await data_catalog.write_entry(session, source_id, "orders", {"columns": {"amount": {
+                "measure": make_item("flow", "human")}}}, if_version=0, actor=None)
+            row = await session.get(DataSource, source_id)
+        probe = await make_prober({"scratch": row})("scratch", "orders", "day")
+        assert probe["max"] == "2026-01-05" and probe["basis"] == "live"
+        assert any(k[0] == source_id for k in coverage._LIVE)
+
+    async def traces(source_id):
+        async with SessionLocal() as session:
+            notes = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all()
+            row = await session.get(DataSource, source_id)
+        return {"row": row is not None, "catalog": len(notes), "probes": sum(k[0] == source_id for k in coverage._LIVE)}
+
+    old = await sqlite_source("scratch", path)
+    await leave_traces(old)
+    new = await sqlite_source("scratch", path)
+    assert new != old and await traces(old) == {"row": False, "catalog": 0, "probes": 0}
+    assert await traces(new) == {"row": True, "catalog": 0, "probes": 0}
+    await leave_traces(new)
+    await forget_source(new)
+    assert await traces(new) == {"row": False, "catalog": 0, "probes": 0}

backend/tests/test_analysis_e2e_claims.py

@@ -26,15 +26,12 @@ from app.analysis import contract
 from app.core import artifact_store
 from app.data import catalog as data_catalog
 from app.data.catalog import make_item
-from app.data.engine import engines
-from app.data.introspect import introspect
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine import analysis_claims as ac
 from app.engine.evidence import iter_segments
 from app.engine.runner import run_manager
 from app.main import app
-from tests.fixtures.sources import drop_source
 
 TOOL = "db_query__shop"
 DECOMPOSE = contract.TOOL_NAMES["decompose"]
@@ -74,7 +71,7 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, region TEXT, channel TEXT, amount INTEGER);")
@@ -84,25 +81,14 @@ async def shop(tmp_path):
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [(i, w, r, c, a) for i, (r, w, c, a) in enumerate(rows, 1)])
     db.commit()
     db.close()
+    # scratch_source 建的是新源（同名的旧源连同数据目录、现查缓存先删掉），测试结束时删掉、数据目录随外键级联删除：同一个
+    # 测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定（比如金额标成了流量，窗口长度不同就从有条件可比
+    # 变成不可比）
+    source_id = await scratch_source("shop", path, schema=True)
     async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        row.schema_cache = await introspect(row)
-        await session.commit()
         for table, notes in NOTES.items():
-            current = await data_catalog.read_entry(session, row.id, table)
-            await data_catalog.write_entry(session, row.id, table, notes, if_version=current.version if current else 0,
-                                           actor=None)
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    # 删掉这个源（数据目录随外键级联删除）：同一个测试进程里别的用例也用名叫 shop 的源，留着这份目录会改掉它们的判定
-    # （比如金额标成了流量，窗口长度不同就从有条件可比变成不可比）
-    await drop_source(source_id)
+            await data_catalog.write_entry(session, source_id, table, notes, if_version=0, actor=None)
+    return source_id
 
 
 def _node(nid, ntype, **config):

backend/tests/test_analysis_tools.py

@@ -1,8 +1,9 @@
 """agent 分析工具（docs/design/analysis-round/00-总方案.md §3.2）：参数、方法说明、查询容器、台账、编号、下钻、次数上限、
 遮罩列、续跑。
 
-算子（app.analysis.run）还没落地：换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的
-东西、算子交回后入口做的事。模型换成脚本（MockChatModel._decide），不调用任何模型。库是虚构的零售库 shop。
+算子（app.analysis.run）换成按请求交回假结论的替身（tests/fixtures/analysis_ops.FakeOps），断言入口交给算子的东西、算子
+交回后入口做的事；五个工具接真算子的端到端见 tests/test_analysis_e2e.py。模型换成脚本（MockChatModel._decide），不调用任何
+模型。库是虚构的零售库 shop。
 """
 from __future__ import annotations
 
@@ -18,9 +19,8 @@ from app.analysis import contract
 from app.analysis.contract import CallError
 from app.core import artifact_store
 from app.core.artifact_store import canonical_json, content_hash
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import Artifact, DataSource, Run, RunEvent
+from app.db.models import Artifact, Run, RunEvent
 from app.engine.runner import run_manager
 from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE
 from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE
@@ -45,30 +45,20 @@ async def engine_up():
     await run_manager.shutdown()
 
 
-async def _shop(tmp_path, options=None):
+async def _shop(tmp_path, scratch_source, options=None):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, phone TEXT);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS)
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    return source_id
+    return await scratch_source("shop", path, options=options)
 
 
 @pytest.fixture
-async def shop(tmp_path):
-    source_id = await _shop(tmp_path)
-    yield
-    await engines.invalidate(source_id)
+async def shop(tmp_path, scratch_source):
+    return await _shop(tmp_path, scratch_source)
 
 
 class Script:
@@ -407,31 +397,25 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop):
     assert ops.calls[0]["settings"]["max_calls"] == 1
 
 
-async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path):
-    source_id = await _shop(tmp_path, {"mask_columns": ["phone"]})
-    try:
-        ops = FakeOps(monkeypatch)
-        Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})])
-        await run_graph(agent_graph())
-        masked = ops.calls[0]["snapshots"]["Q1"]["masked"]
-        # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在
-        assert masked[0] == "tel" and "phone" in masked
-    finally:
-        await engines.invalidate(source_id)
+async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source):
+    await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]})
+    ops = FakeOps(monkeypatch)
+    Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})])
+    await run_graph(agent_graph())
+    masked = ops.calls[0]["snapshots"]["Q1"]["masked"]
+    # 按 SQL 认出改了名字的遮罩列（phone AS tel）排在前面，数据源设的列名也在
+    assert masked[0] == "tel" and "phone" in masked
 
 
-async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path):
-    source_id = await _shop(tmp_path, {"mask_columns": ["phone"]})
-    try:
-        ops = FakeOps(monkeypatch)
-        Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})])
-        await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True})
-        snap = ops.calls[0]["snapshots"]["Q1"]
-        assert "tel" in snap["masked"]
-        tel = snap["columns"].index("tel")
-        assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照"
-    finally:
-        await engines.invalidate(source_id)
+async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path, scratch_source):
+    await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]})
+    ops = FakeOps(monkeypatch)
+    Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})])
+    await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True})
+    snap = ops.calls[0]["snapshots"]["Q1"]
+    assert "tel" in snap["masked"]
+    tel = snap["columns"].index("tel")
+    assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照"
 
 
 # --------------------------------------------------------------------------

backend/tests/test_contract_analysis_claims.py

@@ -17,9 +17,8 @@ from httpx import ASGITransport, AsyncClient
 from langchain_core.messages import AIMessage
 from sqlalchemy import select
 
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion
+from app.db.models import Run, RunEvent, Workflow, WorkflowVersion
 from app.engine import analysis_claims as ac
 from app.engine import compiler
 from app.engine.nodes.io import _analysis_claims, _comparability
@@ -271,25 +270,15 @@ async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, mo
 
 
 @pytest.fixture
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
+    """虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，测试结束时再删）。"""
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);")
     db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)])
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        else:
-            row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop):

backend/tests/test_metric_comparability.py

@@ -19,9 +19,8 @@ from sqlalchemy import select
 
 from app.analysis.contract import CallError
 from app.core import artifact_store
-from app.data.engine import engines
 from app.db.base import SessionLocal
-from app.db.models import DataSource, Run, RunEvent
+from app.db.models import Run, RunEvent
 from app.engine.nodes.metrics import COMPARABILITY_SKIPPED
 from app.engine.runner import run_manager
 from tests.fixtures.analysis_ops import FakeOps
@@ -41,7 +40,7 @@ async def engine_up():
 
 
 @pytest.fixture(autouse=True)
-async def shop(tmp_path):
+async def shop(tmp_path, scratch_source):
     path = tmp_path / "shop.db"
     db = sqlite3.connect(path)
     db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, store INTEGER, amount REAL, "
@@ -51,17 +50,7 @@ async def shop(tmp_path):
         (3, "2026-10-13", "华东", 1203, 450.0, 0.0), (4, "2026-10-15", "华北", 1204, 300.0, 15.0)])
     db.commit()
     db.close()
-    async with SessionLocal() as session:
-        row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none()
-        if row is None:
-            row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True)
-            session.add(row)
-        row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {}
-        await session.commit()
-        source_id = row.id
-    await engines.invalidate(source_id)
-    yield
-    await engines.invalidate(source_id)
+    return await scratch_source("shop", path)
 
 
 def node(nid, ntype, **config):