fix: close final backtest contract boundaries (#17)

2026-09-01 12:40:04 +08:00
parent 598c2b92a2
commit 78d65b4db0
7 changed files with 2569 additions and 30 deletions
@@ -0,0 +1,17 @@
+{
+  "run_id": "rhbacktestrunv1:sha256:5036c771c44a2adade9ea590eee0d8cf824ff8f519fadd8ba424e5d914856386",
+  "replay_spec_digest": "sha256:20f07fcb526bc38b4ba3d71d6c3b00a8c63ad96cab0fecd869326503ab42d98b",
+  "manifest_id": "rhbacktestevidencev1:sha256:f94733e849433f62f3da1e1ec8999891b93d49c7d657832d64e64bef9e211117",
+  "evidence_digest": "sha256:f3913894d032c389c64eef59058b3cbc694cb9cc14ce9cee2699f0068200b650",
+  "table_content_digests": {
+    "run": "sha256:7d947ec93f714641669cbb14bd69dd8cf30387aabb7f3a086918fc2e878ab05c",
+    "signals": "sha256:72dc15064cc45d7c51d2dd4b8c3a6d8c7d4155232d5d70d3c9e7697fab70ce50",
+    "trades": "sha256:2b8b9321e7993941ac486cf50cab5b4c6425b2571a4701f0eb02273ec9a96c53",
+    "positions": "sha256:b456a48fab51742b05084ca6dcaf01c03dfe5b39ad71215ada070e9b1f59f7ea",
+    "nav": "sha256:25649efce860b76410f87dbd36c81085887620086dc0b48b07099918f65d9c78",
+    "performance": "sha256:0856439ea7ec84e38887ccfa0067324f2e9cd293543e4ef683b9c2beb9eb34cf",
+    "attribution": "sha256:ad0f12f668d0a2d9ae5b3636d29989ab4bafcfb95f5de77abeb52a6d9e95d366",
+    "attribution_daily": "sha256:d4459ad650f88871d7b1e40392033037b5818ef92fdf1ee021868929fc1455a5",
+    "risk": "sha256:4816dd4812b5ff2e97e74bf34ca2221bfde675b387a96e277683557f0ad7d975"
+  }
+}
@@ -16,19 +16,26 @@ def test_module_spec_declares_pure_research_engine_boundary() -> None:
    prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower()
    for term in ("investment advice", "live order", "credentials", "source facts"):
        assert term in prohibited
-    assert spec["authority"]["revision"] == 2
+    assert spec["authority"]["revision"] == 3
    assert {
        (item["contract_id"], item["version"])
        for item in spec["contracts"]["provides"]
    } == {
        ("researchhub.factor-definition", "1.0.0"),
        ("researchhub.factor-set-ref", "1.0.0"),
+        ("researchhub.backtest-run-ref", "1.0.0"),
+        ("researchhub.backtest-evidence-manifest", "1.0.0"),
    }
-    assert all(
-        item["authority"] == "quant_engine"
-        and item["path"] == "src/quant_engine/factor_contracts.py"
-        for item in spec["contracts"]["provides"]
-    )
+    expected_paths = {
+        "researchhub.factor-definition": "src/quant_engine/factor_contracts.py",
+        "researchhub.factor-set-ref": "src/quant_engine/factor_contracts.py",
+        "researchhub.backtest-run-ref": "src/quant_engine/governed_pipeline.py",
+        "researchhub.backtest-evidence-manifest": "src/quant_engine/artifact.py",
+    }
+    assert all(item["authority"] == "quant_engine" for item in spec["contracts"]["provides"])
+    assert {
+        item["contract_id"]: item["path"] for item in spec["contracts"]["provides"]
+    } == expected_paths
    assert {
        (item["contract_id"], item["version"])
        for item in spec["contracts"]["consumes"]
@@ -0,0 +1,773 @@
+"""Backtest run-reference and closed-evidence contract conformance."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from dataclasses import replace
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+import pytest
+
+from quant_engine.artifact import (
+    BacktestEvidenceManifest,
+    EvidenceQualification,
+    RESEARCH_ARTIFACT_SCHEMA_VERSION,
+    ResearchRunArtifact,
+    build_backtest_evidence_manifest,
+    build_legacy_backtest_evidence_manifest,
+    build_research_run_artifact,
+)
+from quant_engine.execution import ExecutionConfig
+from quant_engine.factor_contracts import (
+    ActorIdentity,
+    AvailabilityMode,
+    Causation,
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorInput,
+    FactorSetRef,
+    InputBinding,
+    OutputArtifactRef,
+    OutputCoverage,
+    OutputQuality,
+    OutputQualityCheck,
+    ProducerIdentity,
+    ViewAvailability,
+    canonical_json_bytes,
+    factor_definition_from_alpha158,
+    factor_input_schema_digest,
+)
+from quant_engine.governed_pipeline import (
+    BacktestContractError,
+    BacktestContractErrorCode,
+    BacktestRun,
+    BacktestRunRef,
+)
+from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research
+
+
+ROOT = Path(__file__).resolve().parents[1]
+FACTOR_FIXTURE = ROOT / "tests" / "fixtures" / "factor-contracts-v1.golden.json"
+BACKTEST_FIXTURE = ROOT / "tests" / "fixtures" / "backtest-evidence-v1.golden.json"
+VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
+VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+CALENDAR_REVISION_ID = "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078"
+ACTION_REVISION_ID = "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275"
+PARAMETERS = {"lag_sessions": 1, "top_k": 1}
+
+
+def _sha256(value: bytes) -> str:
+    return f"sha256:{hashlib.sha256(value).hexdigest()}"
+
+
+def _accepted_authorities(
+    *,
+    factor_evaluation_at: str = "2026-01-03T11:00:00Z",
+    factor_computed_at: str = "2026-01-03T10:15:00Z",
+    factor_artifact_available_at: str = "2026-01-03T10:20:00Z",
+    factor_availability_mode: AvailabilityMode = AvailabilityMode.AS_AVAILABLE,
+) -> tuple[
+    DatasetSnapshotEnvelope,
+    DataFoundationEnvelope,
+    FactorSetRef,
+]:
+    fixture = json.loads(FACTOR_FIXTURE.read_text(encoding="utf-8"))
+    snapshot = DatasetSnapshotEnvelope.from_dict(fixture["dataset_snapshot"])
+    foundation = DataFoundationEnvelope.from_dict(fixture["data_foundation"])
+    factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
+    definition = factor_definition_from_alpha158(
+        "alpha_005",
+        version="1.0.0",
+        parameters={},
+        inputs=(factor_input,),
+        implementation_digest="sha256:" + "1" * 64,
+        input_schema_digest=factor_input_schema_digest((factor_input,)),
+        valid_from="2026-01-01T00:00:00.000000Z",
+        valid_until="2027-01-01T00:00:00Z",
+        warmup_sessions=10,
+        lag_sessions=1,
+        producer=ProducerIdentity("quant_engine", "1.0.0"),
+        code_revision="c" * 40,
+    )
+    output_schema_bytes = canonical_json_bytes(fixture["output_schema"])
+    output_content_bytes = canonical_json_bytes(fixture["output_content"])
+    artifact_ref = OutputArtifactRef.create(
+        schema_digest=_sha256(output_schema_bytes),
+        content_digest=_sha256(output_content_bytes),
+    )
+    factor_set = FactorSetRef.create(
+        definitions=(definition,),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+        selected_view_ref_ids=(VIEW_REF_ID,),
+        input_bindings=(
+            InputBinding(definition.definition_id, "market", VIEW_REF_ID, VIEW_SCHEMA_DIGEST),
+        ),
+        view_availability=(
+            ViewAvailability(VIEW_REF_ID, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
+        ),
+        output_quality=OutputQuality(
+            "passed",
+            (OutputQualityCheck("finite_values", "passed", "sha256:" + "3" * 64),),
+        ),
+        output_coverage=OutputCoverage(
+            "complete", 1, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64
+        ),
+        output_schema_bytes=output_schema_bytes,
+        output_content_bytes=output_content_bytes,
+        output_artifact_ref=artifact_ref,
+        availability_mode=factor_availability_mode,
+        evaluation_at=factor_evaluation_at,
+        computed_at=factor_computed_at,
+        artifact_available_at=factor_artifact_available_at,
+        producer=ProducerIdentity("quant_engine", "1.0.0"),
+        code_revision="c" * 40,
+        actor=ActorIdentity("service", "factor_worker_v1"),
+        correlation_id="research_run_001",
+        causation=Causation("foundation", foundation.foundation_id),
+        evidence_scope="synthetic_fixture",
+        decision_eligible=False,
+    )
+    return snapshot, foundation, factor_set
+
+
+def _config_digest(parameters: dict[str, object] | None = None) -> str:
+    encoded = json.dumps(
+        PARAMETERS if parameters is None else parameters,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        allow_nan=False,
+    ).encode("utf-8")
+    return _sha256(encoded)
+
+
+def _run_ref(**overrides: Any) -> BacktestRunRef:
+    snapshot, foundation, factor_set = _accepted_authorities()
+    arguments: dict[str, Any] = {
+        "dataset_snapshot": snapshot,
+        "foundation": foundation,
+        "factor_set": factor_set,
+        "universe_digest": "sha256:" + "5" * 64,
+        "trading_calendar_revision_ids": (CALENDAR_REVISION_ID,),
+        "corporate_action_revision_ids": (ACTION_REVISION_ID,),
+        "strategy_id": "alpha-top1",
+        "strategy_version": "1.0.0",
+        "strategy_digest": "sha256:" + "6" * 64,
+        "execution_model_version": "1.0.0",
+        "execution_model_digest": "sha256:" + "7" * 64,
+        "cost_model_version": "1.0.0",
+        "cost_model_digest": "sha256:" + "8" * 64,
+        "random_seed": 7,
+        "code_revision": "d" * 40,
+        "environment_lock_digest": "sha256:" + "9" * 64,
+        "configuration_digest": _config_digest(),
+        "evaluation_at": "2026-01-08T01:00:00Z",
+        "computed_at": "2026-01-08T02:00:00Z",
+    }
+    arguments.update(overrides)
+    return BacktestRunRef.create(**arguments)
+
+
+def _backtest_result() -> FactorBacktestResult:
+    dates = pd.date_range("2026-01-05", periods=4, freq="B")
+    scores = pd.DataFrame({"A": [2.0, 0.0], "B": [1.0, 3.0]}, index=dates[:2])
+    opens = pd.DataFrame(
+        {"A": [10.0, 10.0, 15.0, 15.0], "B": [20.0, 20.0, 20.0, 21.0]},
+        index=dates,
+    )
+    closes = pd.DataFrame(
+        {"A": [10.0, 12.0, 15.0, 15.0], "B": [20.0, 20.0, 18.0, 21.0]},
+        index=dates,
+    )
+    return run_factor_backtest_research(
+        scores,
+        opens,
+        closes,
+        top_k=1,
+        execution_price_field="open",
+        valuation_price_field="close",
+        initial_cash=1_000.0,
+        config=ExecutionConfig(
+            commission_bps=0,
+            stamp_tax_bps=0,
+            slippage_bps=0,
+            min_trade_amount=0,
+        ),
+    )
+
+
+def _artifact(run_ref: BacktestRunRef, *, run_id: str | None = None) -> ResearchRunArtifact:
+    result = _backtest_result()
+    benchmark = pd.Series(
+        [0.0, 0.01, -0.01, 0.02],
+        index=result.returns.index,
+        name="benchmark_return",
+    )
+    return build_research_run_artifact(
+        result,
+        run_id=run_ref.run_id if run_id is None else run_id,
+        strategy_id=run_ref.strategy_id,
+        strategy_name="Alpha Top 1",
+        strategy_version=run_ref.strategy_version,
+        engine_version="1.2.0",
+        code_revision=run_ref.code_revision,
+        data_snapshot_id=run_ref.dataset_snapshot_id,
+        calendar="CN-A",
+        timezone="Asia/Shanghai",
+        started_at="2026-01-08T10:00:00+08:00",
+        finished_at="2026-01-08T10:01:00+08:00",
+        parameters=PARAMETERS,
+        benchmark_id="000300.SH",
+        benchmark_returns=benchmark,
+    )
+
+
+def _assert_error(
+    error: pytest.ExceptionInfo[BacktestContractError],
+    code: BacktestContractErrorCode,
+    path: str,
+) -> None:
+    assert error.value.code is code
+    assert error.value.path == path
+
+
+def test_backtest_run_ref_is_deterministic_and_binds_only_opaque_authorities() -> None:
+    first = _run_ref()
+    second = _run_ref()
+
+    assert first == second
+    assert first.run_id.startswith("rhbacktestrunv1:sha256:")
+    assert first.replay_spec_digest.startswith("sha256:")
+    assert first.dataset_snapshot_id.startswith("rhdsv1:sha256:")
+    assert first.foundation_id.startswith("rhdfv1:sha256:")
+    assert first.factor_set_id.startswith("rhfactorsetv1:sha256:")
+    assert first.trading_calendar_revision_ids == (CALENDAR_REVISION_ID,)
+    assert first.corporate_action_revision_ids == (ACTION_REVISION_ID,)
+    assert first.replay_parent_run_id is None
+    assert first.replay_attempt == 0
+    assert first.replay_ancestor_run_ids == ()
+    snapshot, foundation, factor_set = _accepted_authorities()
+    assert BacktestRunRef.from_dict(
+        first.to_dict(),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+        factor_set=factor_set,
+    ) == first
+    forbidden = ("latest", "locator", "uri", "credential", "provider", "broker")
+    assert not any(token in first.to_json().lower() for token in forbidden)
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("universe_digest", "sha256:" + "a" * 64),
+        ("strategy_digest", "sha256:" + "b" * 64),
+        ("execution_model_digest", "sha256:" + "c" * 64),
+        ("cost_model_digest", "sha256:" + "e" * 64),
+        ("random_seed", 8),
+        ("code_revision", "e" * 40),
+        ("environment_lock_digest", "sha256:" + "f" * 64),
+        ("configuration_digest", "sha256:" + "0" * 64),
+        ("evaluation_at", "2026-01-08T01:00:01Z"),
+        ("computed_at", "2026-01-08T02:00:01Z"),
+    ],
+)
+def test_every_governed_run_input_mutation_changes_run_identity(
+    field: str,
+    value: object,
+) -> None:
+    assert _run_ref(**{field: value}).run_id != _run_ref().run_id
+
+
+def test_backtest_run_ref_rejects_unclosed_upstream_and_unsafe_scalars() -> None:
+    with pytest.raises(BacktestContractError) as wrong_calendar:
+        _run_ref(trading_calendar_revision_ids=())
+    _assert_error(
+        wrong_calendar,
+        BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.trading_calendar_revision_ids",
+    )
+    with pytest.raises(BacktestContractError) as wrong_action:
+        _run_ref(corporate_action_revision_ids=())
+    _assert_error(
+        wrong_action,
+        BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.corporate_action_revision_ids",
+    )
+    with pytest.raises(BacktestContractError) as bool_seed:
+        _run_ref(random_seed=True)
+    _assert_error(bool_seed, BacktestContractErrorCode.TYPE_ERROR, "$.random_seed")
+    with pytest.raises(BacktestContractError) as bad_revision:
+        _run_ref(code_revision="abc")
+    _assert_error(bad_revision, BacktestContractErrorCode.INVALID_FORMAT, "$.code_revision")
+    with pytest.raises(BacktestContractError) as bad_digest:
+        _run_ref(universe_digest="5" * 64)
+    _assert_error(bad_digest, BacktestContractErrorCode.INVALID_FORMAT, "$.universe_digest")
+    with pytest.raises(BacktestContractError) as lookahead:
+        _run_ref(computed_at="2026-01-08T00:59:59Z")
+    _assert_error(lookahead, BacktestContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at")
+    with pytest.raises(BacktestContractError) as factor_type:
+        _run_ref(factor_set="rhfactorsetv1:sha256:" + "0" * 64)
+    _assert_error(factor_type, BacktestContractErrorCode.TYPE_ERROR, "$.factor_set")
+
+
+@pytest.mark.parametrize(
+    ("factor_times", "expected_path"),
+    [
+        (
+            {
+                "factor_evaluation_at": "2026-01-08T01:00:01Z",
+                "factor_computed_at": "2026-01-08T00:59:59Z",
+                "factor_artifact_available_at": "2026-01-08T01:00:00Z",
+            },
+            "$.evaluation_at",
+        ),
+        (
+            {
+                "factor_evaluation_at": "2026-01-03T11:00:00Z",
+                "factor_computed_at": "2026-01-08T01:00:00Z",
+                "factor_artifact_available_at": "2026-01-08T01:00:01Z",
+                "factor_availability_mode": AvailabilityMode.RETROSPECTIVE_REPLAY,
+            },
+            "$.evaluation_at",
+        ),
+    ],
+)
+def test_run_ref_evaluation_closes_factor_pit(
+    factor_times: dict[str, Any],
+    expected_path: str,
+) -> None:
+    snapshot, foundation, factor_set = _accepted_authorities(**factor_times)
+
+    with pytest.raises(BacktestContractError) as lookahead:
+        _run_ref(
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+            factor_set=factor_set,
+        )
+
+    _assert_error(
+        lookahead,
+        BacktestContractErrorCode.TIME_ORDER_VIOLATION,
+        expected_path,
+    )
+
+
+def test_run_ref_rejects_aliases_locators_unsafe_integers_and_invalid_text() -> None:
+    with pytest.raises(BacktestContractError) as mutable_alias:
+        _run_ref(strategy_id="latest")
+    _assert_error(
+        mutable_alias,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.strategy_id",
+    )
+    with pytest.raises(BacktestContractError) as physical_uri:
+        _run_ref(execution_model_version="s3://model-bucket/current")
+    _assert_error(
+        physical_uri,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.execution_model_version",
+    )
+    with pytest.raises(BacktestContractError) as unsafe_seed:
+        _run_ref(random_seed=2**53)
+    _assert_error(
+        unsafe_seed,
+        BacktestContractErrorCode.INVALID_VALUE,
+        "$.random_seed",
+    )
+    with pytest.raises(BacktestContractError) as invalid_unicode:
+        _run_ref(strategy_id="\ud800")
+    _assert_error(
+        invalid_unicode,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.strategy_id",
+    )
+
+    run_ref = _run_ref()
+    mixed_keys = run_ref.to_dict()
+    mixed_keys[1] = "not-a-contract-key"  # type: ignore[index]
+    snapshot, foundation, factor_set = _accepted_authorities()
+    with pytest.raises(BacktestContractError) as invalid_key:
+        BacktestRunRef.from_dict(
+            mixed_keys,
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+            factor_set=factor_set,
+        )
+    _assert_error(invalid_key, BacktestContractErrorCode.TYPE_ERROR, "$")
+
+
+@pytest.mark.parametrize(
+    "physical_id",
+    ["db.table", "source_alpha", "wind.model", "qtdb_view", "bloomberg-signal"],
+)
+def test_run_ref_rejects_physical_terms_in_logical_ids(physical_id: str) -> None:
+    with pytest.raises(BacktestContractError) as physical:
+        _run_ref(strategy_id=physical_id)
+    _assert_error(
+        physical,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.strategy_id",
+    )
+
+
+@pytest.mark.parametrize("version", ["1.0.0-.", "1.0.0-foo..bar", "1.0.0-01"])
+def test_run_ref_requires_strict_semver_prerelease_identifiers(version: str) -> None:
+    with pytest.raises(BacktestContractError) as invalid:
+        _run_ref(strategy_version=version)
+    _assert_error(
+        invalid,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.strategy_version",
+    )
+
+    assert _run_ref(strategy_version="1.0.0-alpha.1").strategy_version == "1.0.0-alpha.1"
+
+
+def test_replay_lineage_is_acyclic_and_cannot_claim_changed_inputs() -> None:
+    parent = _run_ref()
+    replay = _run_ref(
+        computed_at="2026-01-08T03:00:00Z",
+        parent=parent,
+        replay_reason="deterministic_reproduction",
+        replay_attempt=1,
+    )
+
+    assert replay.run_id != parent.run_id
+    assert replay.replay_spec_digest == parent.replay_spec_digest
+    assert replay.replay_parent_run_id == parent.run_id
+    assert replay.replay_ancestor_run_ids == (parent.run_id,)
+
+    with pytest.raises(BacktestContractError) as changed_input:
+        _run_ref(
+            universe_digest="sha256:" + "a" * 64,
+            computed_at="2026-01-08T03:00:00Z",
+            parent=parent,
+            replay_reason="changed_universe",
+            replay_attempt=1,
+        )
+    _assert_error(
+        changed_input,
+        BacktestContractErrorCode.LINEAGE_VIOLATION,
+        "$.replay_spec_digest",
+    )
+    with pytest.raises(BacktestContractError) as skipped_attempt:
+        _run_ref(
+            computed_at="2026-01-08T03:00:00Z",
+            parent=parent,
+            replay_reason="skipped_attempt",
+            replay_attempt=2,
+        )
+    _assert_error(
+        skipped_attempt,
+        BacktestContractErrorCode.LINEAGE_VIOLATION,
+        "$.replay_attempt",
+    )
+
+
+def test_offline_research_manifest_closes_exact_existing_evidence_mapping() -> None:
+    run_ref = _run_ref()
+    artifact = _artifact(run_ref)
+    first = build_backtest_evidence_manifest(
+        run_ref,
+        artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+        qualification=EvidenceQualification.CONTRACT_QUALIFIED,
+    )
+    second = build_backtest_evidence_manifest(
+        run_ref,
+        artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+        qualification=EvidenceQualification.CONTRACT_QUALIFIED,
+    )
+
+    assert first == second
+    assert first.manifest_id.startswith("rhbacktestevidencev1:sha256:")
+    assert first.run_id == run_ref.run_id
+    assert first.profile == "offline_research_v1"
+    assert first.qualification is EvidenceQualification.CONTRACT_QUALIFIED
+    mapping = {
+        item.category: tuple(table.logical_name for table in item.tables)
+        for item in first.evidence
+    }
+    assert mapping == {
+        "run": ("run",),
+        "signal": ("signals",),
+        "fill": ("trades",),
+        "position_nav": ("positions", "nav"),
+        "performance": ("performance",),
+        "attribution": ("attribution", "attribution_daily"),
+        "risk_snapshot": ("risk",),
+        "replay": (),
+    }
+    assert "order" not in mapping
+    assert "rejection" not in mapping
+    risk = next(item for item in first.evidence if item.category == "risk_snapshot")
+    assert risk.tables[0].row_count == 0
+    assert risk.tables[0].schema_digest.startswith("sha256:")
+
+    changed_performance = artifact.performance
+    changed_performance.loc[0, "n_days"] += 1
+    changed_artifact = replace(artifact, _performance=changed_performance)
+    changed = build_backtest_evidence_manifest(
+        run_ref,
+        changed_artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+    )
+    assert changed.manifest_id != first.manifest_id
+    assert run_ref.run_id == first.run_id == changed.run_id
+
+
+def test_manifest_rejects_missing_mismatched_or_duplicate_evidence() -> None:
+    run_ref = _run_ref()
+    artifact = _artifact(run_ref)
+    with pytest.raises(BacktestContractError) as wrong_run:
+        build_backtest_evidence_manifest(
+            run_ref,
+            _artifact(run_ref, run_id="different-run"),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        wrong_run,
+        BacktestContractErrorCode.IDENTITY_MISMATCH,
+        "$.artifact.tables.run.run_id",
+    )
+    missing_signals = replace(artifact, _signals=None)  # type: ignore[arg-type]
+    with pytest.raises(BacktestContractError) as missing_table:
+        build_backtest_evidence_manifest(
+            run_ref,
+            missing_signals,
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        missing_table,
+        BacktestContractErrorCode.TYPE_ERROR,
+        "$.artifact.tables.signals",
+    )
+    with pytest.raises(BacktestContractError) as digest_mismatch:
+        build_backtest_evidence_manifest(
+            run_ref,
+            artifact,
+            artifact_available_at="2026-01-08T02:05:00Z",
+            expected_table_digests={"performance": "sha256:" + "0" * 64},
+        )
+    _assert_error(
+        digest_mismatch,
+        BacktestContractErrorCode.EVIDENCE_MISMATCH,
+        "$.artifact.tables.performance.content_digest",
+    )
+    manifest = build_backtest_evidence_manifest(
+        run_ref,
+        artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+    )
+    duplicate = manifest.to_dict()
+    duplicate["evidence"].append(copy.deepcopy(duplicate["evidence"][0]))
+    with pytest.raises(BacktestContractError) as duplicate_category:
+        BacktestEvidenceManifest.from_dict(
+            duplicate,
+            backtest_run_ref=run_ref,
+            artifact=artifact,
+        )
+    _assert_error(
+        duplicate_category,
+        BacktestContractErrorCode.INVALID_VALUE,
+        "$.evidence[8].category",
+    )
+
+
+def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding() -> None:
+    run_ref = _run_ref()
+    artifact = _artifact(run_ref)
+
+    with pytest.raises(BacktestContractError) as unsupported_schema:
+        build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, schema_version="999.0.0"),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        unsupported_schema,
+        BacktestContractErrorCode.INVALID_VALUE,
+        "$.artifact.schema_version",
+    )
+
+    identities: set[str] = set()
+    for value in (float("nan"), float("inf"), float("-inf")):
+        performance = artifact.performance
+        performance.loc[0, "alpha"] = value
+        manifest = build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, _performance=performance),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+        assert manifest.artifact_schema_version == RESEARCH_ARTIFACT_SCHEMA_VERSION
+        identities.add(manifest.manifest_id)
+    assert len(identities) == 3
+
+    content_digests: set[str] = set()
+    for value in (float("nan"), {"non_finite_float": "nan"}):
+        performance = artifact.performance.astype(object)
+        performance.at[0, "alpha"] = value
+        manifest = build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, _performance=performance),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+        performance_entry = next(
+            entry for entry in manifest.evidence if entry.category == "performance"
+        )
+        content_digests.add(performance_entry.tables[0].content_digest)
+    assert len(content_digests) == 2
+
+    unsupported = artifact.performance.astype(object)
+    unsupported.loc[0, "alpha"] = object()
+    with pytest.raises(BacktestContractError) as unsupported_cell:
+        build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, _performance=unsupported),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        unsupported_cell,
+        BacktestContractErrorCode.TYPE_ERROR,
+        "$.artifact.tables.performance.rows[0].alpha",
+    )
+
+    invalid_nested_key = artifact.performance.astype(object)
+    invalid_nested_key.at[0, "alpha"] = {"\ud800": "value"}
+    with pytest.raises(BacktestContractError) as invalid_utf8:
+        build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, _performance=invalid_nested_key),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        invalid_utf8,
+        BacktestContractErrorCode.INVALID_FORMAT,
+        "$.artifact.tables.performance.rows[0].alpha.keys",
+    )
+
+    unsafe_integer = artifact.performance.astype(object)
+    unsafe_integer.loc[0, "alpha"] = 10**5000
+    with pytest.raises(BacktestContractError) as unsafe_cell:
+        build_backtest_evidence_manifest(
+            run_ref,
+            replace(artifact, _performance=unsafe_integer),
+            artifact_available_at="2026-01-08T02:05:00Z",
+        )
+    _assert_error(
+        unsafe_cell,
+        BacktestContractErrorCode.INVALID_VALUE,
+        "$.artifact.tables.performance.rows[0].alpha",
+    )
+
+
+def test_artifact_canonical_content_has_typed_collision_free_cell_encoding() -> None:
+    run_ref = _run_ref()
+    artifact = _artifact(run_ref)
+
+    content_hashes: set[str] = set()
+    for value in (
+        float("nan"),
+        float("inf"),
+        float("-inf"),
+        {"non_finite_float": "nan"},
+    ):
+        performance = artifact.performance.astype(object)
+        performance.at[0, "alpha"] = value
+        mutated = replace(artifact, _performance=performance)
+        content_hashes.add(mutated.content_sha256)
+        assert "non_finite_float" in mutated.canonical_json()
+    assert len(content_hashes) == 4
+
+    unsupported = artifact.performance.astype(object)
+    unsupported.loc[0, "alpha"] = object()
+    with pytest.raises(BacktestContractError) as unsupported_cell:
+        replace(artifact, _performance=unsupported).canonical_json()
+    _assert_error(
+        unsupported_cell,
+        BacktestContractErrorCode.TYPE_ERROR,
+        "$.tables.performance.rows[0].alpha",
+    )
+
+
+def test_legacy_bridge_is_explicit_and_cannot_be_contract_qualified() -> None:
+    run_ref = _run_ref()
+    legacy_run = BacktestRun(
+        run_id="legacy-run-001",
+        dataset_snapshot_id=run_ref.dataset_snapshot_id,
+        factor_version_id="alpha_005@1.0.0",
+        strategy_version_id="alpha-top1@1.0.0",
+        code_revision=run_ref.code_revision,
+        config_hash=_config_digest().removeprefix("sha256:"),
+        created_at=datetime(2026, 1, 8, 2, 0, tzinfo=UTC),
+    )
+    artifact = _artifact(run_ref, run_id=legacy_run.run_id)
+    manifest = build_legacy_backtest_evidence_manifest(
+        legacy_run,
+        artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+    )
+
+    assert manifest.qualification is EvidenceQualification.LEGACY_EXPLORATORY
+    assert manifest.run_id == legacy_run.run_id
+    assert manifest.backtest_run_ref is None
+    assert manifest.to_dict()["run_reference"]["kind"] == "legacy_backtest_run"
+    assert BacktestEvidenceManifest.from_dict(
+        manifest.to_dict(),
+        artifact=artifact,
+    ) == manifest
+    with pytest.raises(BacktestContractError) as implicit_promotion:
+        build_backtest_evidence_manifest(  # type: ignore[arg-type]
+            legacy_run,
+            artifact,
+            artifact_available_at="2026-01-08T02:05:00Z",
+            qualification=EvidenceQualification.CONTRACT_QUALIFIED,
+        )
+    _assert_error(
+        implicit_promotion,
+        BacktestContractErrorCode.TYPE_ERROR,
+        "$.backtest_run_ref",
+    )
+
+
+def test_golden_contract_and_architecture_boundary() -> None:
+    run_ref = _run_ref()
+    artifact = _artifact(run_ref)
+    manifest = build_backtest_evidence_manifest(
+        run_ref,
+        artifact,
+        artifact_available_at="2026-01-08T02:05:00Z",
+    )
+    golden = json.loads(BACKTEST_FIXTURE.read_text(encoding="utf-8"))
+
+    table_digests = {
+        table.logical_name: table.content_digest
+        for item in manifest.evidence
+        for table in item.tables
+    }
+    assert golden == {
+        "run_id": run_ref.run_id,
+        "replay_spec_digest": run_ref.replay_spec_digest,
+        "manifest_id": manifest.manifest_id,
+        "evidence_digest": manifest.evidence_digest,
+        "table_content_digests": table_digests,
+    }
+    governed_source = (ROOT / "src" / "quant_engine" / "governed_pipeline.py").read_text(
+        encoding="utf-8"
+    )
+    artifact_source = (ROOT / "src" / "quant_engine" / "artifact.py").read_text(
+        encoding="utf-8"
+    )
+    assert "from quant_engine.artifact" not in governed_source
+    assert "BacktestRunRef" in governed_source
+    assert "BacktestEvidenceManifest" not in governed_source
+    assert "BacktestEvidenceManifest" in artifact_source
+    assert not (ROOT / "src" / "quant_engine" / "backtest_contracts.py").exists()