qlib/backend/tests/test_experiment_archive.py

"""Experiment 归档补齐（2026-09）测试。

覆盖五件事：
1. **同步端点归档**：`POST /api/backtests` / `POST /api/factor-tests` 落库（此前只写
   进程内存），并用响应头 `X-Experiment-Id` 返回归档 id；归档失败时仍返回计算结果
   并用 `X-Archive-Error` 如实暴露（AGENT §7 不静默）。
2. **data_version**：真实数据指纹（交易日 + 各表行数），格式与列宽（varchar(40)）合规。
3. **列表过滤/分页**：`kind` / `q` / `limit` / `offset` + `X-Total-Count`（不静默截断）。
4. **DELETE**：200 → 再 GET 404 → 再 DELETE 404，且不触碰 job 记录。
5. **归档完整性**：默认不再按 60 只截断个股曲线；超体积预算时裁剪并留下
   `archive_meta`（机器可读）+ `unimplemented`（人可读）证据；
   完整结果只存 experiment 一份（`GET /api/jobs/{id}` 契约不变，老记录可回退）。
"""

from __future__ import annotations

import json
import re
from datetime import date, datetime
from types import SimpleNamespace

import pytest
from app.api import deps
from app.application.services import experiment_archive as ea
from app.domain.entities.market import Stock
from app.domain.entities.research import (
    BacktestResult,
    CostSpec,
    ExperimentRecord,
    ExperimentSummary,
    FactorSpec,
    JobRecord,
    JobStatus,
    ResearchSpec,
    SelectionSpec,
    UniverseSpec,
)
from app.infrastructure.persistence.sqlalchemy.base import Base
from app.infrastructure.persistence.sqlalchemy.models.jobs import ExperimentModel, JobModel
from app.infrastructure.persistence.sqlalchemy.models.market import StockDailyModel
from app.infrastructure.persistence.sqlalchemy.repositories.jobs_impl import (
    SqlAlchemyExperimentRepository,
    SqlAlchemyJobRepository,
)
from app.main import app
from app.quant.engine import LocalEngine
from app.quant.service import ResearchService
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, select
from sqlalchemy.orm import sessionmaker

from conftest_quant import bars_dataframe_to_daily_bars, synthetic_daily

_SYMS = ["60000" + str(i) + ".SH" for i in range(5)]

_BACKTEST_BODY = {
    "type": "backtest",
    "universe": {"exclude_st": False, "min_listing_days": 0},
    "factors": [{"name": "momentum_20", "weight": 1.0}],
    "selection": {"top_n": 2},
    "rebalance": "monthly",
    "period": ["2024-03-01", "2024-10-31"],
}

# data_version 格式：d<YYYYMMDD|->，后跟可选的行数段（MySQL 近似值带 ≈ 与 k）
_DATA_VERSION_RE = re.compile(
    r"^d(\d{8}|-)(;n(≈\d+k|\d+))?(;a(≈\d+k|\d+))?(;b(≈\d+k|\d+))?$"
)


class _MemStockRepo:
    def __init__(self, stocks: list[Stock]) -> None:
        self._stocks = stocks

    def get_by_symbol(self, symbol: str) -> Stock | None:
        return next((s for s in self._stocks if s.symbol == symbol), None)

    def list(self) -> list[Stock]:
        return self._stocks


class _MemDailyRepo:
    def __init__(self, bars) -> None:
        self._bars = bars

    def get_range_many(self, symbols, start, end, adjust="none"):
        return [b for b in self._bars if b.symbol in symbols and start <= b.trade_date <= end]

    def get_range(self, symbol, start, end):
        return [b for b in self._bars if b.symbol == symbol and start <= b.trade_date <= end]


@pytest.fixture()
def api(tmp_path):
    """TestClient + 临时 SQLite（job/experiment/行情表）+ 内存行情仓储。"""
    drifts = {sym: 0.003 - 0.0015 * i for i, sym in enumerate(_SYMS)}
    bars = bars_dataframe_to_daily_bars(synthetic_daily(drifts, n=300))
    stocks = [
        Stock(symbol=sym, name=f"测试股份{i}", list_date=date(1999, 11, 10))
        for i, sym in enumerate(_SYMS)
    ]
    service = ResearchService(_MemStockRepo(stocks), _MemDailyRepo(bars), LocalEngine())

    engine = create_engine(f"sqlite:///{tmp_path / 'archive.db'}", future=True)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine, expire_on_commit=False)

    def _session_override():
        with Session() as s:
            yield s

    app.dependency_overrides[deps.get_session] = _session_override
    app.dependency_overrides[deps._service_factory] = lambda: service
    app.dependency_overrides[deps._stock_repo_factory] = lambda: _MemStockRepo(stocks)
    with TestClient(app) as client:
        yield client, Session
    app.dependency_overrides.clear()


def _exp_by_id(Session, exp_id: str) -> ExperimentRecord | None:
    with Session() as session:
        return SqlAlchemyExperimentRepository(session).get(exp_id)


class TestSyncEndpointsArchive:
    """任务 1：同步端点必须落库，并通过 X-Experiment-Id 暴露归档 id。"""

    def test_backtest_archives_to_db_and_returns_header(self, api) -> None:
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        assert resp.status_code == 200
        body = resp.json()
        assert body["equity_curve"], "body 形状不变：仍是完整 BacktestResult"

        exp_id = resp.headers.get("X-Experiment-Id")
        assert exp_id and exp_id.startswith("EXP-"), "必须用响应头返回归档 id"
        exp = _exp_by_id(Session, exp_id)
        assert exp is not None, "同步回测必须在 experiment 表落库（不再只写进程内存）"
        assert exp.kind == "backtest"
        assert json.loads(exp.spec_json)["factors"][0]["name"] == "momentum_20"
        assert exp.summary_text and "总收益" in exp.summary_text
        assert exp.code_version, "归档必须带代码版本"
        assert exp.job_id is None, "同步端点没有 Job，job_id 应为 None"
        # 归档结果可解码为完整 BacktestResult
        archived = BacktestResult.model_validate_json(exp.result_json)
        assert archived.summary.total_return_pct == body["summary"]["total_return_pct"]

        # 详情接口能读回
        detail = client.get(f"/api/experiments/{exp_id}")
        assert detail.status_code == 200
        assert detail.json()["result"]["summary"]["total_return_pct"] == (
            body["summary"]["total_return_pct"]
        )

    def test_factor_test_archives_with_kind(self, api) -> None:
        client, Session = api
        payload = dict(_BACKTEST_BODY, type="factor_test")
        resp = client.post("/api/factor-tests", json=payload)
        assert resp.status_code == 200
        assert resp.json()["factor_name"] == "momentum_20"

        exp_id = resp.headers.get("X-Experiment-Id")
        assert exp_id
        exp = _exp_by_id(Session, exp_id)
        assert exp is not None
        assert exp.kind == "factor_test"
        assert exp.result_json and json.loads(exp.result_json)["factor_name"] == "momentum_20"

    def test_archive_failure_keeps_result_and_exposes_reason(self, api, monkeypatch) -> None:
        """归档失败不得吞掉计算结果，且失败必须如实暴露（AGENT §7 / §24）。"""
        client, _Session = api

        def _boom(self, experiment):  # noqa: ANN001
            raise RuntimeError("db down")

        monkeypatch.setattr(SqlAlchemyExperimentRepository, "save", _boom)
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        assert resp.status_code == 200, "归档失败也必须返回已算出的结果"
        assert resp.json()["equity_curve"], "结果本身必须完整返回"
        assert "X-Experiment-Id" not in resp.headers
        err = resp.headers.get("X-Archive-Error")
        assert err and "RuntimeError" in err and "db down" in err, (
            f"归档失败原因必须如实暴露，实际：{err!r}"
        )


class TestDataVersion:
    """任务 2：data_version 必须是真实、可比较、且塞得进 varchar(40) 的指纹。"""

    def test_format_and_column_width(self, api) -> None:
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"])
        assert exp is not None
        assert exp.data_version, "data_version 不允许为空（此前永远是 NULL）"
        assert _DATA_VERSION_RE.match(exp.data_version), exp.data_version
        assert len(exp.data_version) <= 40, "experiment.data_version 是 varchar(40)"

    def test_real_values_from_seeded_bars(self, api) -> None:
        """种子 3 行日线（SQLite 走精确 COUNT(*)）→ 指纹必须是精确的真实值。"""
        client, Session = api
        days = [date(2024, 9, 26), date(2024, 9, 27), date(2024, 9, 30)]
        with Session() as session:
            session.execute(
                StockDailyModel.__table__.insert(),
                [
                    {
                        "symbol": f"60010{i}.SH",
                        "trade_date": d,
                        "source": "tushare",
                        "adjust": "none",
                        "open": 1.0,
                        "high": 1.0,
                        "low": 1.0,
                        "close": 1.0,
                        "volume": 1.0,
                        "amount": 1.0,
                    }
                    for i, d in enumerate(days)
                ],
            )
            session.commit()
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"])
        assert exp is not None
        assert exp.data_version == "d20240930;n3;a0;b0", exp.data_version

    def test_unavailable_when_tables_missing(self, tmp_path) -> None:
        """取不到数据口径时如实降级为 unavailable，绝不编造数字。"""
        from app.infrastructure.persistence.sqlalchemy.data_version import compute_data_version

        engine = create_engine(f"sqlite:///{tmp_path / 'empty.db'}", future=True)
        with sessionmaker(bind=engine)() as session:
            assert compute_data_version(session) == "unavailable"


class TestListFilters:
    """任务 3：列表过滤/分页在 SQL 层完成，并用 X-Total-Count 暴露总数。"""

    @staticmethod
    def _seed(Session) -> list[str]:
        """直接造 4 条归档：2 个 backtest（其一含 DIVIDEND_YIELD 因子）、1 个
        factor_test、1 个 selection。"""
        rows = [
            ("EXP-A1", "backtest", "DIVIDEND_YIELD", "总收益 12.00% · 年化 8.00% · 回撤 5.00%"),
            ("EXP-A2", "backtest", "momentum_20", "总收益 3.00% · 年化 2.00% · 回撤 1.00%"),
            ("EXP-A3", "factor_test", "momentum_20", "IC 0.0100 · RankIC 0.0200 · 样本 100 日"),
            ("EXP-A4", "selection", "momentum_60", "as_of 2024-10-31 · 选出 3 / 评估 100"),
        ]
        with Session() as session:
            repo = SqlAlchemyExperimentRepository(session)
            for i, (exp_id, kind, factor, summary) in enumerate(rows):
                repo.save(
                    ExperimentRecord(
                        id=exp_id,
                        kind=kind,
                        spec_json=json.dumps(
                            {
                                "factors": [{"name": factor, "weight": 1.0}],
                                "period": ["2024-03-01", "2024-10-31"],
                                "rebalance": "monthly",
                                "selection": {"top_n": i + 1},
                            }
                        ),
                        result_json=json.dumps({"kind": kind, "n": i}),
                        summary_text=summary,
                        code_version="abc1234",
                        data_version="d20260904;n1;a2;b3",
                        created_at=datetime(2026, 9, 4, 10, i, 0),
                    )
                )
            session.commit()
        return [r[0] for r in rows]

    def test_kind_filter_and_total_count(self, api) -> None:
        client, Session = api
        self._seed(Session)
        resp = client.get("/api/experiments", params={"kind": "backtest"})
        assert resp.status_code == 200
        body = resp.json()
        assert {e["id"] for e in body} == {"EXP-A1", "EXP-A2"}
        assert resp.headers["X-Total-Count"] == "2"

    def test_q_is_case_insensitive_over_factor_name_and_summary(self, api) -> None:
        client, Session = api
        self._seed(Session)
        # 因子名（在 spec_json 里，大小写不敏感）
        r1 = client.get("/api/experiments", params={"q": "dividend_yield"})
        assert [e["id"] for e in r1.json()] == ["EXP-A1"]
        assert r1.headers["X-Total-Count"] == "1"
        # id 片段
        r2 = client.get("/api/experiments", params={"q": "exp-a"})
        assert r2.headers["X-Total-Count"] == "4"
        # summary_text（中文）
        r3 = client.get("/api/experiments", params={"q": "选出 3"})
        assert [e["id"] for e in r3.json()] == ["EXP-A4"]
        # kind + q 组合
        r4 = client.get("/api/experiments", params={"kind": "backtest", "q": "momentum"})
        assert [e["id"] for e in r4.json()] == ["EXP-A2"]

    def test_like_wildcards_and_injection_are_bound(self, api) -> None:
        client, Session = api
        self._seed(Session)
        # `%` 是 LIKE 通配符，必须被转义成字面量：种子里只有 EXP-A1/A2 的摘要带
        # 百分号，若未转义会返回全部 4 条
        r = client.get("/api/experiments", params={"q": "%"})
        assert {e["id"] for e in r.json()} == {"EXP-A1", "EXP-A2"}
        assert r.headers["X-Total-Count"] == "2"
        # 注入尝试必须是普通字符串（参数绑定），不得改变语义
        inj = client.get("/api/experiments", params={"q": "' OR 1=1 --"})
        assert inj.json() == [] and inj.headers["X-Total-Count"] == "0"

    def test_limit_offset_and_total_count_not_silently_truncated(self, api) -> None:
        client, Session = api
        self._seed(Session)
        page1 = client.get("/api/experiments", params={"limit": 2, "offset": 0})
        assert len(page1.json()) == 2
        assert page1.headers["X-Total-Count"] == "4", "总数必须暴露，客户端才知道被截断"
        page2 = client.get("/api/experiments", params={"limit": 2, "offset": 2})
        assert len(page2.json()) == 2
        ids = {e["id"] for e in page1.json()} | {e["id"] for e in page2.json()}
        assert len(ids) == 4, "分页不得重复/漏项（created_at 同秒时靠 id 次级排序稳定）"
        # 默认 limit（200）远大于旧硬编码的 50
        assert client.get("/api/experiments").status_code == 200

    def test_meta_includes_data_version_job_id_and_result_bytes(self, api) -> None:
        client, Session = api
        seeded = self._seed(Session)
        row = next(e for e in client.get("/api/experiments").json() if e["id"] == "EXP-A1")
        assert row["data_version"] == "d20260904;n1;a2;b3", "列表必须暴露数据指纹"
        assert row["job_id"] is None
        assert row["result_bytes"] == len(json.dumps({"kind": "backtest", "n": 0}))

        # 详情路径（result_json 在内存，len() 零成本）与列表口径必须一致
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp_id = resp.headers["X-Experiment-Id"]
        exp = _exp_by_id(Session, exp_id)
        assert exp is not None
        detail = client.get(f"/api/experiments/{exp_id}").json()
        assert detail["result_bytes"] == len(exp.result_json)
        assert detail["data_version"] == exp.data_version
        assert detail["job_id"] is None
        listed = next(e for e in client.get("/api/experiments").json() if e["id"] == exp_id)
        assert listed["result_bytes"] == detail["result_bytes"], "列表与详情体积口径一致"
        assert set(seeded) == {"EXP-A1", "EXP-A2", "EXP-A3", "EXP-A4"}

    def test_repository_list_filtered_returns_summaries_without_result_json(self, api) -> None:
        _client, Session = api
        self._seed(Session)
        with Session() as session:
            repo = SqlAlchemyExperimentRepository(session)
            rows = repo.list_filtered(kind="backtest", limit=10)
            assert all(isinstance(r, ExperimentSummary) for r in rows)
            assert not hasattr(rows[0], "result_json"), "列表不得拉取 MEDIUMTEXT 大字段"
            assert repo.count_filtered() == 4
            assert repo.count_filtered(kind="selection") == 1


class TestDelete:
    """任务 4：DELETE 只删归档，不删 Job 记录。"""

    def test_delete_then_404(self, api) -> None:
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp_id = resp.headers["X-Experiment-Id"]

        deleted = client.delete(f"/api/experiments/{exp_id}")
        assert deleted.status_code == 200
        assert deleted.json() == {"deleted": exp_id}
        assert _exp_by_id(Session, exp_id) is None
        assert client.get(f"/api/experiments/{exp_id}").status_code == 404
        again = client.delete(f"/api/experiments/{exp_id}")
        assert again.status_code == 404
        assert "不存在" in again.json()["detail"]

    def test_delete_keeps_job_record_and_reports_missing_result(self, api) -> None:
        """删除归档不触碰 job 表；job 的结果只剩归档一份，故如实说明原因而非静默 200 空值。"""
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp_id = resp.headers["X-Experiment-Id"]
        with Session() as session:
            job = JobRecord(
                id="JOB-DEL-1",
                kind="backtest",
                spec_json=json.dumps(_BACKTEST_BODY),
                status=JobStatus.SUCCESS,
                experiment_id=exp_id,
                created_at=datetime.now(),
            )
            SqlAlchemyJobRepository(session).create(job)
            session.commit()

        assert client.delete(f"/api/experiments/{exp_id}").status_code == 200
        with Session() as session:
            kept = SqlAlchemyJobRepository(session).get("JOB-DEL-1")
            assert kept is not None, "job 记录是执行历史，DELETE 不得连带删除"
            assert kept.experiment_id == exp_id
        view = client.get("/api/jobs/JOB-DEL-1")
        assert view.status_code == 200
        assert view.json()["result"] is None
        assert "已不存在" in view.json()["result_unavailable_reason"]


class TestCurveCompleteness:
    """P0：默认完整存档（不再按 60 只截断）；超预算时裁剪并留证据。"""

    @staticmethod
    def _spec(top_n: int, symbols: list[str]) -> ResearchSpec:
        return ResearchSpec(
            type="backtest",
            universe=UniverseSpec(
                exclude_st=False, min_listing_days=0, symbols=list(symbols)
            ),
            factors=[FactorSpec(name="momentum_20")],
            selection=SelectionSpec(top_n=top_n, hold_top_x=top_n, allow_substitute=False, defer_buy=True),
            rebalance="monthly",
            selection_interval_months=6,
            rebalance_interval_months=6,
            period=(date(2024, 3, 1), date(2024, 6, 30)),
            costs=CostSpec(
                commission_rate=0.0, stamp_tax_rate=0.0, slippage_rate=0.0
            ),
        )

    def test_default_stores_every_held_symbol_beyond_60(self) -> None:
        syms = [f"60{i:04d}.SH" for i in range(70)]
        drifts = {s: 0.001 * (i % 4) + 0.0005 for i, s in enumerate(syms)}
        daily = synthetic_daily(drifts, n=120)
        res = LocalEngine().run_backtest(daily, self._spec(len(syms), syms))

        held = {a.symbol for a in res.fills if a.symbol}
        assert len(held) > 60, f"构造的样本应持有 60 只以上，实际 {len(held)}"
        curve_syms = {c.symbol for c in res.symbol_curves}
        assert held <= curve_syms, "期内持有的每只都必须有曲线（完整存档，默认不截断）"
        assert not [n for n in res.unimplemented if "个股收益曲线" in n], "默认不得出现截断说明"

    def test_configured_limit_truncates_and_notes_actual_limit(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        """配置了上限就截断 —— 且 note 里写清「实际保留几只 / 共几只」。"""
        from app.core import config as config_mod
        from app.quant import local_engine as le

        monkeypatch.setattr(
            config_mod,
            "get_settings",
            lambda: SimpleNamespace(research_archive_curve_limit=3),
        )
        assert le._MAX_SYMBOL_CURVES is None, "默认代码路径不设上限（配置优先）"
        syms = [f"60{i:04d}.SH" for i in range(10)]
        drifts = {s: 0.001 * (i % 3) + 0.0005 for i, s in enumerate(syms)}
        daily = synthetic_daily(drifts, n=90)
        res = LocalEngine().run_backtest(daily, self._spec(len(syms), syms))

        assert len(res.symbol_curves) == 3
        note = [n for n in res.unimplemented if "个股收益曲线" in n]
        assert note and "3 只" in note[0] and "共持有" in note[0]

    def test_archive_budget_truncates_with_machine_readable_evidence(
        self, api, monkeypatch
    ) -> None:
        """归档字节预算：超预算时裁剪曲线，并把证据同时写进 archive_meta 与 unimplemented。

        预算不靠猜：先按默认预算归档一次拿到「完整归档的字节数」L，再把预算设为
        L - 2,000 重新归档 —— 必然需要裁掉若干条曲线，且裁剪后必须真的落到预算内。
        """
        client, Session = api
        body = {
            **_BACKTEST_BODY,
            "selection": {"top_n": 5},
            "universe": {**_BACKTEST_BODY["universe"], "symbols": _SYMS},
        }
        first = client.post("/api/backtests", json=body)
        assert first.status_code == 200
        assert first.json()["archive_meta"]["truncated"] is False, "默认预算下应完整存档"
        full = _exp_by_id(Session, first.headers["X-Experiment-Id"])
        assert full is not None
        full_bytes = len(full.result_json.encode("utf-8"))
        assert first.json()["archive_meta"]["curves_total"] >= 3, "样本需有足够曲线才能验证裁剪"

        budget = full_bytes - 2_000
        monkeypatch.setattr(ea, "_archive_budget_bytes", lambda: budget)
        resp = client.post("/api/backtests", json=body)
        assert resp.status_code == 200
        meta = resp.json()["archive_meta"]
        assert meta["truncated"] is True, f"预算 {budget} < 完整体积 {full_bytes}，必须触发裁剪"
        assert 0 < meta["curves_stored"] < meta["curves_total"], meta
        assert meta["budget_bytes"] == budget
        assert meta["over_budget"] is False
        assert any("个股收益曲线" in n for n in resp.json()["unimplemented"])

        exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"])
        assert exp is not None
        assert len(exp.result_json.encode("utf-8")) <= budget, "裁剪后必须真的放进预算"
        archived = json.loads(exp.result_json)
        assert archived["archive_meta"]["truncated"] is True
        assert archived["archive_meta"]["curves_stored"] == len(archived["symbol_curves"])
        assert any("个股收益曲线" in n for n in archived["unimplemented"])
        assert archived["archive_meta"]["result_bytes"] == len(exp.result_json.encode("utf-8"))

    def test_archive_over_budget_is_labelled_not_hidden(self, api, monkeypatch) -> None:
        """预算小到连归档主体都放不下：必须标 over_budget + 说明，绝不假装达标。"""
        client, Session = api
        budget = 5_000
        monkeypatch.setattr(ea, "_archive_budget_bytes", lambda: budget)

        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        assert resp.status_code == 200
        body = resp.json()
        meta = body["archive_meta"]
        assert meta["curves_stored"] == 0 and meta["truncated"] is True
        assert meta["over_budget"] is True, "主体超预算必须显式标注，而不是静默超标"
        assert any("超过配置的体积预算" in n for n in body["unimplemented"])
        exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"])
        assert exp is not None
        archived = json.loads(exp.result_json)
        assert archived["archive_meta"]["over_budget"] is True
        assert any("超过配置的体积预算" in n for n in archived["unimplemented"])
        # 如实记录「确实超标」这一事实（不掩盖，也不假装裁剪到位）
        assert archived["archive_meta"]["result_bytes"] > budget

    def test_archive_meta_complete_when_under_budget(self, api) -> None:
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"])
        assert exp is not None
        archived = json.loads(exp.result_json)
        meta = archived["archive_meta"]
        assert meta["truncated"] is False
        assert meta["curves_stored"] == meta["curves_total"] == len(archived["symbol_curves"])
        assert meta["result_bytes"] == len(exp.result_json.encode("utf-8"))
        assert meta["result_chars"] == len(exp.result_json)


class TestSingleResultCopy:
    """P1：完整结果只存 experiment 一份；GET /api/jobs/{id} 契约与老记录兼容。"""

    def test_submit_and_run_reads_through_in_memory_only(self, tmp_path) -> None:
        """`submit_and_run` 的返回对象仍带完整结果（既有调用方如
        scripts/run_dividend_case.py 依赖它），但**数据库里不存第二份**。"""
        from app.application.services.job_executor import submit_and_run

        engine = create_engine(f"sqlite:///{tmp_path / 'submit.db'}", future=True)
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine, expire_on_commit=False)
        drifts = {sym: 0.003 - 0.0015 * i for i, sym in enumerate(_SYMS)}
        bars = bars_dataframe_to_daily_bars(synthetic_daily(drifts, n=300))
        stocks = [
            Stock(symbol=sym, name=f"测试股份{i}", list_date=date(1999, 11, 10))
            for i, sym in enumerate(_SYMS)
        ]
        spec = ResearchSpec.model_validate(_BACKTEST_BODY)
        done = submit_and_run(
            spec,
            factories={
                "session_factory": Session,
                "job_repo_factory": lambda s: SqlAlchemyJobRepository(s),
                "experiment_repo_factory": lambda s: SqlAlchemyExperimentRepository(s),
                "stock_repo_factory": lambda s: _MemStockRepo(stocks),
                "daily_repo_factory": lambda s: _MemDailyRepo(bars),
                "engine": LocalEngine(),
            },
        )
        assert done.status == JobStatus.SUCCESS
        assert done.experiment_id
        assert done.result_json, "返回对象必须带完整结果（内存读透，供既有调用方使用）"
        assert BacktestResult.model_validate_json(done.result_json).symbol_curves

        with Session() as session:
            persisted = SqlAlchemyJobRepository(session).get(done.id)
            exp = SqlAlchemyExperimentRepository(session).get(done.experiment_id)
        assert persisted is not None and persisted.result_json is None, "数据库不得存第二份"
        assert exp is not None and exp.result_json, "完整结果在 experiment 侧"

    def test_job_view_reads_experiment_and_keeps_full_result(self, api) -> None:
        from app.api.jobs import _job_view

        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp_id = resp.headers["X-Experiment-Id"]
        with Session() as session:
            job = JobRecord(
                id="JOB-VIEW-1",
                kind="backtest",
                spec_json=json.dumps(_BACKTEST_BODY),
                status=JobStatus.SUCCESS,
                result_json=None,  # 新形态：job 侧不再保存结果副本
                experiment_id=exp_id,
                created_at=datetime.now(),
            )
            SqlAlchemyJobRepository(session).create(job)
            session.commit()
            view = _job_view(job, SqlAlchemyExperimentRepository(session))
        assert view["result_source"] == "experiment"
        assert view["result"].symbol_curves, "契约不变：GET /api/jobs/{id} 仍返回完整结果"
        assert (
            view["result"].summary.total_return_pct
            == resp.json()["summary"]["total_return_pct"]
        )
        assert client.get("/api/jobs/JOB-VIEW-1").json()["result"]["symbol_curves"]

    def test_legacy_job_record_still_decodes_from_result_json(self, api) -> None:
        """老形态（result_json 有值、experiment_id 为空）必须继续可用。"""
        from app.api.jobs import _job_view

        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        full = json.dumps(resp.json(), ensure_ascii=False)
        with Session() as session:
            job = JobRecord(
                id="JOB-LEGACY-1",
                kind="backtest",
                spec_json=json.dumps(_BACKTEST_BODY),
                status=JobStatus.SUCCESS,
                result_json=full,
                experiment_id=None,
                created_at=datetime.now(),
            )
            SqlAlchemyJobRepository(session).create(job)
            session.commit()
            view = _job_view(job, SqlAlchemyExperimentRepository(session))
        assert view["result_source"] == "job"
        assert view["result"].symbol_curves
        assert view["result"].summary.total_return_pct == resp.json()["summary"]["total_return_pct"]

    def test_job_table_no_longer_duplicates_result(self, api) -> None:
        """json 列确实没有第二份结果（体积审计的机器可读证据）。"""
        client, Session = api
        resp = client.post("/api/backtests", json=_BACKTEST_BODY)
        exp_id = resp.headers["X-Experiment-Id"]
        with Session() as session:
            job = JobRecord(
                id="JOB-SIZE-1",
                kind="backtest",
                spec_json=json.dumps(_BACKTEST_BODY),
                status=JobStatus.SUCCESS,
                experiment_id=exp_id,
                created_at=datetime.now(),
            )
            SqlAlchemyJobRepository(session).create(job)
            session.commit()
        with Session() as session:
            row = session.execute(
                select(JobModel.result_json).where(JobModel.id == "JOB-SIZE-1")
            ).scalar()
            exp_row = session.execute(
                select(ExperimentModel.result_json).where(ExperimentModel.id == exp_id)
            ).scalar()
        assert row is None, "job 侧不得再存结果副本"
        assert exp_row, "完整结果必须在 experiment 侧"
        archived = _exp_by_id(Session, exp_id)
        assert archived is not None
        assert len(exp_row) == len(archived.result_json)