"""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()