fix: close final backtest contract boundaries

fix: harden backtest contract boundaries
feat: close backtest evidence contract
2026-09-01 12:34:12 +08:00 · 2026-09-01 12:17:16 +08:00 · 2026-09-01 12:04:48 +08:00 · 2026-09-01 11:10:18 +08:00 · 2026-09-01 08:36:41 +08:00 · 2026-08-30 16:14:34 +08:00
12 changed files with 6475 additions and 44 deletions
@@ -1,7 +1,7 @@
 {
  "schema_version": 1,
  "module_id": "quant_engine",
-  "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 1, "effective_from": "2026-08-20T00:00:00+08:00"},
+  "authority": {"scope": "module_metadata", "subject": "quant_engine", "owner": "quant-engine-owner", "source": "MODULE_SPEC.yaml", "revision": 3, "effective_from": "2026-09-01T00:00:00+08:00"},
  "repository": {"name": "quant_engine", "workspace_id": "researchhub", "type": "research_engine", "maturity": "operational"},
  "bounded_context": {
    "domain": "quantitative-research-engine",
@@ -18,13 +18,25 @@
    {"id": "factor-and-indicator-calculation", "summary": "Calculate reusable alpha factors and technical indicators from caller-supplied data.", "status": "operational"},
    {"id": "execution-simulation", "summary": "Simulate costs, slippage, market constraints, fills, NAV, and PnL without live order routing.", "status": "operational"},
    {"id": "portfolio-backtesting", "summary": "Run weight-based backtests and benchmark comparisons.", "status": "operational"},
+    {"id": "backtest-evidence-contracts", "summary": "Identify governed offline backtest inputs and close existing research artifact evidence without persistence or decision authority.", "status": "operational"},
    {"id": "risk-and-performance-analysis", "summary": "Calculate portfolio decomposition, risk contribution, and performance statistics.", "status": "operational"}
  ],
  "data": {"owns": [
    {"asset_id": "quantitative-model-implementations", "kind": "model", "classification": "internal"},
    {"asset_id": "simulation-and-metric-results", "kind": "artifact", "classification": "confidential"}
  ]},
-  "contracts": {"provides": [], "consumes": []},
+  "contracts": {
+    "provides": [
+      {"contract_id": "researchhub.factor-definition", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"},
+      {"contract_id": "researchhub.factor-set-ref", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/factor_contracts.py"},
+      {"contract_id": "researchhub.backtest-run-ref", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/governed_pipeline.py"},
+      {"contract_id": "researchhub.backtest-evidence-manifest", "version": "1.0.0", "authority": "quant_engine", "path": "src/quant_engine/artifact.py"}
+    ],
+    "consumes": [
+      {"contract_id": "researchhub.dataset-snapshot", "version": "1.0.0", "authority": "researchhub.data", "admission": "qualified_immutable_envelope"},
+      {"contract_id": "researchhub.data-foundation", "version": "1.0.0", "authority": "researchhub.data", "admission": "content_addressed_selected_views"}
+    ]
+  },
  "dependencies": [],
  "agent_context": {
    "default_entrypoints": [
@@ -19,14 +19,15 @@
 ## 模块

 - `alpha_factors` — 158 alpha 公式 + 24 基础算子（移植自 qlib alpha158）
+- `factor_contracts` — `FactorDefinition` / `FactorSetRef` v1 纯计算合同、严格 PIT/availability 输入准入与显式 legacy 投影
 - `execution` — A 股长仓执行仿真（成本/滑点/现金约束）+ 稀疏调仓/完整交易日 Ledger + 可投影成交与 NAV 审计；T+1、涨跌停、成交量与价差提供独立约束函数
 - `indicators` — 50+ 技术指标（MACD / KDJ / 布林 / ATR / ADX / 等）
 - `data_adapter` — 桥接 qtdb_pro 长表与新模块（rename / long-wide / 复权 / vwap 代理）
 - `backtest` — weight-based 多日仿真（rebalance_table / compute_nav / compare_to_benchmark）
 - `portfolio_construction` — 多期因子分数 → Top-K → 等权目标权重表
 - `research_pipeline` — 因子日 → 下一真实交易日 → 显式执行价 → 日末估值 → 成本后绩效（防前视编排）
- `governed_pipeline` — 数据快照 → 因子版本 → 策略版本 → 回测运行 → 目标组合 → 风险决策 → Paper 订单意图；全链路带确定性 ID，风险拒绝时禁止生成订单意图
- `artifact` — 版本化、确定性、存储中立的完整 research run 事实表与 manifest
+- `governed_pipeline` — 数据快照 → 因子版本 → 策略版本 → 回测运行 → 目标组合 → 风险决策 → Paper 订单意图；同时拥有输入/配置/重放血缘决定的 `BacktestRunRef`
+- `artifact` — 版本化、确定性、存储中立的完整 research run 事实表，以及只映射现有表的 `BacktestEvidenceManifest`
 - `attribution` — 基于实际成交后持仓的隔夜 / 日内 / 交易成本逐日收益归因与闭合审计
 - `metrics` — 绝对绩效 + 严格日期对齐的 TE / IR / alpha / beta 基准相对绩效
 - `factor_library` — 通用方法（turnover / winsorize / IC / OLS / jb_test）
@@ -195,6 +196,69 @@ print(backtest.stats())
 print(backtest.benchmark_report())
 ```

+## 因子／特征合同 v1
+
+`quant_engine.factor_contracts` 提供 `researchhub.factor-definition` 与
+`researchhub.factor-set-ref` `1.0.0`。合同使用受限 canonical JSON：只接受 ASCII
+lower-snake-case object key、UTF-8 string、bool/null 和 safe integer；小数参数必须用显式
+canonical decimal string。定义、输入映射、上游证据、输出 schema/content 和 lineage 的任一
+语义变化都会产生新 identity。
+
+创建 `FactorSetRef` 必须提供完整且可重算 identity 的 `DatasetSnapshotEnvelope` 与
+`DataFoundationEnvelope`，不能用 ID 字符串或布尔值代替资格证明。每个因子输入都要映射到一个
+实际选中的 `StandardizedViewRef`，schema 必须同时匹配定义和 view；未消费、缺失、重复或跨
+snapshot/Foundation/PIT 的 view 都会失败关闭。snapshot PIT 可以早于 Foundation/view PIT，
+但始终满足 knowledge ≤ snapshot PIT ≤ Foundation/view/FactorSet PIT ≤ evaluation。
+
+```python
+from quant_engine.factor_contracts import (
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorDefinition,
+    FactorSetRef,
+)
+
+snapshot = DatasetSnapshotEnvelope.from_dict(dataset_snapshot_v1)
+foundation = DataFoundationEnvelope.from_dict(data_foundation_v1)
+
+# definition 必须是完整的 FactorDefinition；FactorSetRef.create 还要求显式 input bindings、
+# view availability、output quality/coverage、canonical output bytes 和 immutable artifact ref。
+factor_set = FactorSetRef.create(
+    definitions=(definition,),
+    dataset_snapshot=snapshot,
+    foundation=foundation,
+    **explicit_factor_set_evidence,
+)
+```
+
+`availability_mode="as_available"` 声明 source/view 和计算产物在历史 evaluation 前实际可用；
+`"retrospective_replay"` 保留历史 evaluation，但要求真实 publication/view creation、compute 和
+artifact 时间位于之后，并固定 `historical_availability="not_established"`。两种模式都不会授予
+decision、real-data、production、paper 或 live readiness。
+
+旧 `governed_pipeline.FactorVersion` 的四字段构造器、`factor_id@version`、run/target/risk/order
+identity 均保持不变。迁移只能通过 content-addressed `LegacyFactorBinding`，再显式调用
+`bind_legacy_factor()` 或 `project_legacy_factor()`；后者是有损投影，不表示旧 digest 与新定义
+digest 等价，也不会把旧 run 静默升级为新合同。
+
+## 回测引用与证据合同 v1
+
+`quant_engine.governed_pipeline.BacktestRunRef` 是合格回测运行身份的唯一权威。`run_id` 只由
+已验收的 Dataset Snapshot / Data Foundation / `FactorSetRef` 身份、universe、日历与公司行动
+祖先、策略、执行/成本模型、严格整数 seed、完整代码提交、环境锁、配置、时间和重放血缘决定；
+它不包含任何输出摘要。重放必须绑定直接父运行、连续 attempt 和不变的
+`replay_spec_digest`，输入漂移或血缘环会失败关闭。
+
+`quant_engine.artifact.BacktestEvidenceManifest` 只摘要 `ResearchRunArtifact` 已有的九张事实表。
+固定 `offline_research_v1` 映射为 `run`、`signal`、`fill`、`position_nav`、`performance`、
+`attribution`、`risk_snapshot` 与 `replay`；每张表都保留列模式摘要、行数和内容摘要，空 risk
+表也必须有稳定 schema。`manifest_id` 由完整 RunRef 与输出证据决定，因此结果变化不会反向改变
+`run_id`。当前 artifact 不拥有订单或拒绝事实，所以此画像明确不声明 `order` / `rejection`。
+
+旧 `BacktestRun` 只能通过 `build_legacy_backtest_evidence_manifest()` 显式映射为
+`LEGACY_EXPLORATORY`；不能隐式提升为 `CONTRACT_QUALIFIED`。所有资格均只描述离线证据闭合，
+不表示投资有效、组合获批、Paper、生产或实盘就绪。
+
 ## 治理垂直切片

 `governed_pipeline` 不复制因子、回测、组合或执行算法，只编排现有能力并补充版本与风险契约。
@@ -11,26 +11,84 @@ import hashlib
 import json
 import math
 import re
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from dataclasses import asdict, dataclass
 from datetime import UTC, datetime
 from enum import StrEnum
 from types import MappingProxyType
+from typing import Any, Never, Self

 import pandas as pd

 from quant_engine.execution import ExecutionConfig
+from quant_engine.factor_contracts import (
+    ContractErrorCode,
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorContractError,
+    FactorDefinition,
+    FactorSetRef,
+    LegacyFactorBinding,
+    canonical_json,
+    canonical_json_bytes,
+)
 from quant_engine.research_pipeline import FactorBacktestResult, run_factor_backtest_research

 _SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_PREFIXED_SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$")
 _GIT_SHA = re.compile(r"^[0-9a-f]{40}$")
+_LOGICAL_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$")
+_SEMVER = re.compile(
+    r"^(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\."
+    r"(?:0|[1-9][0-9]*)"
+    r"(?:-(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*)"
+    r"(?:\.(?:0|[1-9][0-9]*|[0-9A-Za-z-]*[A-Za-z-][0-9A-Za-z-]*))*)?"
+    r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$"
+)
+_MAX_SAFE_INTEGER = (1 << 53) - 1
+_FORBIDDEN_ID_TOKENS = frozenset(
+    {
+        "latest",
+        "provider",
+        "source",
+        "broker",
+        "credential",
+        "locator",
+        "uri",
+        "s3",
+        "http",
+        "file",
+        "postgres",
+        "mysql",
+        "clickhouse",
+        "table",
+        "sql",
+        "tushare",
+        "wind",
+        "bloomberg",
+        "qtdb",
+        "edb",
+        "db",
+        "database",
+        "schema",
+        "bucket",
+        "path",
+        "endpoint",
+    }
+)

 __all__ = [
+    "BACKTEST_RUN_REF_SCHEMA_VERSION",
    "DatasetSnapshot",
    "FactorVersion",
+    "bind_legacy_factor",
+    "project_legacy_factor",
    "StrategyStage",
    "StrategyVersion",
    "BacktestRun",
+    "BacktestContractErrorCode",
+    "BacktestContractError",
+    "BacktestRunRef",
    "PortfolioTarget",
    "RiskPolicy",
    "RiskDecisionStatus",
@@ -43,6 +101,684 @@ __all__ = [
 ]


+BACKTEST_RUN_REF_SCHEMA_VERSION = "1.0.0"
+
+
+class BacktestContractErrorCode(StrEnum):
+    """Stable rejection categories shared by the S3 public contracts."""
+
+    TYPE_ERROR = "type_error"
+    MISSING_FIELD = "missing_field"
+    UNKNOWN_FIELD = "unknown_field"
+    INVALID_FORMAT = "invalid_format"
+    INVALID_VALUE = "invalid_value"
+    IDENTITY_MISMATCH = "identity_mismatch"
+    QUALIFICATION_REJECTED = "qualification_rejected"
+    TIME_ORDER_VIOLATION = "time_order_violation"
+    INPUT_CLOSURE_VIOLATION = "input_closure_violation"
+    LINEAGE_VIOLATION = "lineage_violation"
+    EVIDENCE_MISMATCH = "evidence_mismatch"
+    READINESS_ESCALATION = "readiness_escalation"
+
+
+class BacktestContractError(ValueError):
+    """Typed deterministic contract rejection with an exact JSON path."""
+
+    def __init__(self, code: BacktestContractErrorCode, path: str, detail: str) -> None:
+        self.code = code
+        self.path = path
+        self.detail = detail
+        super().__init__(f"{code.value} at {path}: {detail}")
+
+
+def _backtest_fail(
+    code: BacktestContractErrorCode,
+    path: str,
+    detail: str,
+) -> Never:
+    raise BacktestContractError(code, path, detail)
+
+
+def _backtest_object(value: Any, path: str, fields: Sequence[str]) -> dict[str, Any]:
+    if type(value) is not dict:
+        _backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an object")
+    if any(type(key) is not str for key in value):
+        _backtest_fail(
+            BacktestContractErrorCode.TYPE_ERROR,
+            path,
+            "object keys must be strings",
+        )
+    required = set(fields)
+    missing = sorted(required - set(value))
+    if missing:
+        _backtest_fail(
+            BacktestContractErrorCode.MISSING_FIELD,
+            f"{path}.{missing[0]}",
+            "field is required",
+        )
+    unknown = sorted(set(value) - required)
+    if unknown:
+        _backtest_fail(
+            BacktestContractErrorCode.UNKNOWN_FIELD,
+            f"{path}.{unknown[0]}",
+            "field is not permitted",
+        )
+    return value
+
+
+def _backtest_text(value: Any, path: str) -> str:
+    if type(value) is not str:
+        _backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be a string")
+    try:
+        value.encode("utf-8")
+    except UnicodeEncodeError as error:
+        raise BacktestContractError(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be valid UTF-8 text",
+        ) from error
+    if not value or value != value.strip():
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be non-empty canonical text",
+        )
+    return value
+
+
+def _backtest_logical_id(value: Any, path: str) -> str:
+    logical_id = _backtest_text(value, path)
+    tokens = set(re.split(r"[._-]+", logical_id))
+    if _LOGICAL_ID.fullmatch(logical_id) is None or tokens & _FORBIDDEN_ID_TOKENS:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a stable storage-neutral logical identifier",
+        )
+    return logical_id
+
+
+def _backtest_semver(value: Any, path: str) -> str:
+    version = _backtest_text(value, path)
+    if _SEMVER.fullmatch(version) is None:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a canonical semantic version",
+        )
+    return version
+
+
+def _backtest_digest(value: Any, path: str) -> str:
+    digest = _backtest_text(value, path)
+    if _PREFIXED_SHA256.fullmatch(digest) is None:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a lowercase sha256 digest",
+        )
+    return digest
+
+
+def _backtest_git_revision(value: Any, path: str) -> str:
+    revision = _backtest_text(value, path)
+    if _GIT_SHA.fullmatch(revision) is None:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a lowercase 40-character Git commit",
+        )
+    return revision
+
+
+def _backtest_instant(value: Any, path: str) -> tuple[str, datetime]:
+    text = _backtest_text(value, path)
+    if not text.endswith("Z"):
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a canonical UTC instant",
+        )
+    try:
+        parsed = datetime.fromisoformat(text.removesuffix("Z") + "+00:00")
+    except ValueError as error:
+        raise BacktestContractError(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a canonical UTC instant",
+        ) from error
+    normalized = parsed.astimezone(UTC).isoformat().replace("+00:00", "Z")
+    if normalized != text:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_FORMAT,
+            path,
+            "must be a canonical UTC instant",
+        )
+    return text, parsed
+
+
+def _backtest_integer(value: Any, path: str, *, minimum: int = 0) -> int:
+    if type(value) is not int:
+        _backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an integer")
+    if value < minimum:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_VALUE,
+            path,
+            f"must be at least {minimum}",
+        )
+    if value > _MAX_SAFE_INTEGER:
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_VALUE,
+            path,
+            "integer exceeds the canonical safe range",
+        )
+    return value
+
+
+def _backtest_string_tuple(value: Any, path: str) -> tuple[str, ...]:
+    if type(value) not in {tuple, list}:
+        _backtest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an array")
+    normalized = tuple(
+        _backtest_text(item, f"{path}[{index}]") for index, item in enumerate(value)
+    )
+    if len(normalized) != len(set(normalized)):
+        _backtest_fail(
+            BacktestContractErrorCode.INVALID_VALUE,
+            path,
+            "items must be unique",
+        )
+    return tuple(sorted(normalized))
+
+
+def _content_address_digest(value: str, prefix: str, path: str) -> str:
+    if not value.startswith(prefix) or _SHA256.fullmatch(value.removeprefix(prefix)) is None:
+        _backtest_fail(
+            BacktestContractErrorCode.IDENTITY_MISMATCH,
+            path,
+            "authority identity is not content addressed",
+        )
+    return f"sha256:{value.removeprefix(prefix)}"
+
+
+def _digest_document(value: object) -> str:
+    return f"sha256:{hashlib.sha256(canonical_json_bytes(value)).hexdigest()}"
+
+
+def _selected_revision_closure(
+    foundation: DataFoundationEnvelope,
+    factor_set: FactorSetRef,
+    field: str,
+) -> tuple[str, ...]:
+    payload = foundation.to_dict()
+    raw_views = payload.get("standardized_views")
+    if type(raw_views) is not list:
+        _backtest_fail(
+            BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+            "$.foundation.standardized_views",
+            "validated Foundation views are required",
+        )
+    selected = set(factor_set.selected_view_ref_ids)
+    found: set[str] = set()
+    seen_views: set[str] = set()
+    for index, raw_view in enumerate(raw_views):
+        if type(raw_view) is not dict:
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                f"$.foundation.standardized_views[{index}]",
+                "validated view object is required",
+            )
+        view_id = raw_view.get("view_ref_id")
+        if view_id not in selected:
+            continue
+        seen_views.add(str(view_id))
+        revisions = raw_view.get(field)
+        path = f"$.foundation.standardized_views[{index}].{field}"
+        if type(revisions) is not list:
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                path,
+                "revision closure is required",
+            )
+        for revision_index, revision_id in enumerate(revisions):
+            found.add(_backtest_text(revision_id, f"{path}[{revision_index}]"))
+    if seen_views != selected:
+        _backtest_fail(
+            BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+            "$.factor_set.selected_view_ref_ids",
+            "selected FactorSet views are not Foundation members",
+        )
+    return tuple(sorted(found))
+
+
+@dataclass(frozen=True, slots=True, init=False)
+class BacktestRunRef:
+    """Immutable input/configuration/replay identity determined before output exists."""
+
+    contract_name: str
+    schema_version: str
+    run_id: str
+    dataset_snapshot_id: str
+    dataset_content_digest: str
+    dataset_manifest_digest: str
+    foundation_id: str
+    foundation_digest: str
+    factor_set_id: str
+    factor_set_digest: str
+    factor_output_content_digest: str
+    universe_digest: str
+    trading_calendar_revision_ids: tuple[str, ...]
+    trading_calendar_digest: str
+    corporate_action_revision_ids: tuple[str, ...]
+    corporate_action_digest: str
+    strategy_id: str
+    strategy_version: str
+    strategy_digest: str
+    execution_model_version: str
+    execution_model_digest: str
+    cost_model_version: str
+    cost_model_digest: str
+    random_seed: int
+    code_revision: str
+    environment_lock_digest: str
+    configuration_digest: str
+    evaluation_at: str
+    computed_at: str
+    replay_spec_digest: str
+    replay_parent_run_id: str | None
+    replay_reason: str | None
+    replay_attempt: int
+    replay_ancestor_run_ids: tuple[str, ...]
+
+    @classmethod
+    def create(
+        cls,
+        *,
+        dataset_snapshot: DatasetSnapshotEnvelope,
+        foundation: DataFoundationEnvelope,
+        factor_set: FactorSetRef,
+        universe_digest: str,
+        trading_calendar_revision_ids: Sequence[str],
+        corporate_action_revision_ids: Sequence[str],
+        strategy_id: str,
+        strategy_version: str,
+        strategy_digest: str,
+        execution_model_version: str,
+        execution_model_digest: str,
+        cost_model_version: str,
+        cost_model_digest: str,
+        random_seed: int,
+        code_revision: str,
+        environment_lock_digest: str,
+        configuration_digest: str,
+        evaluation_at: str,
+        computed_at: str,
+        parent: BacktestRunRef | None = None,
+        replay_reason: str | None = None,
+        replay_attempt: int = 0,
+    ) -> Self:
+        return cls._build(
+            dataset_snapshot=dataset_snapshot,
+            foundation=foundation,
+            factor_set=factor_set,
+            universe_digest=universe_digest,
+            trading_calendar_revision_ids=trading_calendar_revision_ids,
+            corporate_action_revision_ids=corporate_action_revision_ids,
+            strategy_id=strategy_id,
+            strategy_version=strategy_version,
+            strategy_digest=strategy_digest,
+            execution_model_version=execution_model_version,
+            execution_model_digest=execution_model_digest,
+            cost_model_version=cost_model_version,
+            cost_model_digest=cost_model_digest,
+            random_seed=random_seed,
+            code_revision=code_revision,
+            environment_lock_digest=environment_lock_digest,
+            configuration_digest=configuration_digest,
+            evaluation_at=evaluation_at,
+            computed_at=computed_at,
+            parent=parent,
+            replay_reason=replay_reason,
+            replay_attempt=replay_attempt,
+        )
+
+    @classmethod
+    def _build(
+        cls,
+        *,
+        dataset_snapshot: Any,
+        foundation: Any,
+        factor_set: Any,
+        universe_digest: Any,
+        trading_calendar_revision_ids: Any,
+        corporate_action_revision_ids: Any,
+        strategy_id: Any,
+        strategy_version: Any,
+        strategy_digest: Any,
+        execution_model_version: Any,
+        execution_model_digest: Any,
+        cost_model_version: Any,
+        cost_model_digest: Any,
+        random_seed: Any,
+        code_revision: Any,
+        environment_lock_digest: Any,
+        configuration_digest: Any,
+        evaluation_at: Any,
+        computed_at: Any,
+        parent: BacktestRunRef | None,
+        replay_reason: Any,
+        replay_attempt: Any,
+    ) -> Self:
+        if not isinstance(dataset_snapshot, DatasetSnapshotEnvelope):
+            _backtest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                "$.dataset_snapshot",
+                "complete validated DatasetSnapshotEnvelope required",
+            )
+        if not isinstance(foundation, DataFoundationEnvelope):
+            _backtest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                "$.foundation",
+                "complete validated DataFoundationEnvelope required",
+            )
+        if not isinstance(factor_set, FactorSetRef):
+            _backtest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                "$.factor_set",
+                "complete validated FactorSetRef required",
+            )
+        if foundation.dataset_snapshot_id != dataset_snapshot.snapshot_id:
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                "$.foundation.dataset_snapshot_id",
+                "Foundation snapshot mismatch",
+            )
+        if (
+            factor_set.dataset_snapshot_id != dataset_snapshot.snapshot_id
+            or factor_set.foundation_id != foundation.foundation_id
+            or factor_set.upstream_evidence.snapshot_content_digest
+            != dataset_snapshot.content_digest
+            or factor_set.upstream_evidence.snapshot_manifest_digest
+            != dataset_snapshot.manifest_digest
+        ):
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                "$.factor_set",
+                "FactorSet upstream authority mismatch",
+            )
+        expected_calendars = _selected_revision_closure(
+            foundation,
+            factor_set,
+            "trading_calendar_revision_ids",
+        )
+        supplied_calendars = _backtest_string_tuple(
+            trading_calendar_revision_ids,
+            "$.trading_calendar_revision_ids",
+        )
+        if not expected_calendars or supplied_calendars != expected_calendars:
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                "$.trading_calendar_revision_ids",
+                "must exactly match selected FactorSet calendar ancestry",
+            )
+        expected_actions = _selected_revision_closure(
+            foundation,
+            factor_set,
+            "corporate_action_revision_ids",
+        )
+        supplied_actions = _backtest_string_tuple(
+            corporate_action_revision_ids,
+            "$.corporate_action_revision_ids",
+        )
+        if supplied_actions != expected_actions:
+            _backtest_fail(
+                BacktestContractErrorCode.INPUT_CLOSURE_VIOLATION,
+                "$.corporate_action_revision_ids",
+                "must exactly match selected FactorSet corporate-action ancestry",
+            )
+
+        normalized_evaluation, evaluation_time = _backtest_instant(
+            evaluation_at,
+            "$.evaluation_at",
+        )
+        normalized_computed, computed_time = _backtest_instant(computed_at, "$.computed_at")
+        _, factor_available = _backtest_instant(
+            factor_set.artifact_available_at,
+            "$.factor_set.artifact_available_at",
+        )
+        _, factor_evaluation = _backtest_instant(
+            factor_set.evaluation_at,
+            "$.factor_set.evaluation_at",
+        )
+        if evaluation_time < factor_evaluation or evaluation_time < factor_available:
+            _backtest_fail(
+                BacktestContractErrorCode.TIME_ORDER_VIOLATION,
+                "$.evaluation_at",
+                "run evaluation must not precede FactorSet evaluation or availability",
+            )
+        if computed_time < evaluation_time or computed_time < factor_available:
+            _backtest_fail(
+                BacktestContractErrorCode.TIME_ORDER_VIOLATION,
+                "$.computed_at",
+                "FactorSet availability and evaluation must not follow computation",
+            )
+
+        normalized_seed = _backtest_integer(random_seed, "$.random_seed")
+        replay_count = _backtest_integer(replay_attempt, "$.replay_attempt")
+        spec_payload: dict[str, object] = {
+            "dataset_snapshot_id": dataset_snapshot.snapshot_id,
+            "dataset_content_digest": dataset_snapshot.content_digest,
+            "dataset_manifest_digest": dataset_snapshot.manifest_digest,
+            "foundation_id": foundation.foundation_id,
+            "foundation_digest": _content_address_digest(
+                foundation.foundation_id,
+                "rhdfv1:sha256:",
+                "$.foundation.foundation_id",
+            ),
+            "factor_set_id": factor_set.factor_set_id,
+            "factor_set_digest": _content_address_digest(
+                factor_set.factor_set_id,
+                "rhfactorsetv1:sha256:",
+                "$.factor_set.factor_set_id",
+            ),
+            "factor_output_content_digest": factor_set.output_content_digest,
+            "universe_digest": _backtest_digest(universe_digest, "$.universe_digest"),
+            "trading_calendar_revision_ids": list(supplied_calendars),
+            "trading_calendar_digest": _digest_document(list(supplied_calendars)),
+            "corporate_action_revision_ids": list(supplied_actions),
+            "corporate_action_digest": _digest_document(list(supplied_actions)),
+            "strategy_id": _backtest_logical_id(strategy_id, "$.strategy_id"),
+            "strategy_version": _backtest_semver(strategy_version, "$.strategy_version"),
+            "strategy_digest": _backtest_digest(strategy_digest, "$.strategy_digest"),
+            "execution_model_version": _backtest_semver(
+                execution_model_version,
+                "$.execution_model_version",
+            ),
+            "execution_model_digest": _backtest_digest(
+                execution_model_digest,
+                "$.execution_model_digest",
+            ),
+            "cost_model_version": _backtest_semver(cost_model_version, "$.cost_model_version"),
+            "cost_model_digest": _backtest_digest(
+                cost_model_digest,
+                "$.cost_model_digest",
+            ),
+            "random_seed": normalized_seed,
+            "code_revision": _backtest_git_revision(code_revision, "$.code_revision"),
+            "environment_lock_digest": _backtest_digest(
+                environment_lock_digest,
+                "$.environment_lock_digest",
+            ),
+            "configuration_digest": _backtest_digest(
+                configuration_digest,
+                "$.configuration_digest",
+            ),
+            "evaluation_at": normalized_evaluation,
+        }
+        replay_spec_digest = _digest_document(spec_payload)
+        if parent is None:
+            if replay_reason is not None:
+                _backtest_fail(
+                    BacktestContractErrorCode.LINEAGE_VIOLATION,
+                    "$.replay_reason",
+                    "root run cannot declare a replay reason",
+                )
+            if replay_count != 0:
+                _backtest_fail(
+                    BacktestContractErrorCode.LINEAGE_VIOLATION,
+                    "$.replay_attempt",
+                    "root run must use attempt zero",
+                )
+            normalized_reason = None
+            parent_run_id = None
+            ancestors: tuple[str, ...] = ()
+        else:
+            if not isinstance(parent, BacktestRunRef):
+                _backtest_fail(
+                    BacktestContractErrorCode.TYPE_ERROR,
+                    "$.parent",
+                    "BacktestRunRef parent is required",
+                )
+            normalized_reason = _backtest_logical_id(replay_reason, "$.replay_reason")
+            if replay_count != parent.replay_attempt + 1:
+                _backtest_fail(
+                    BacktestContractErrorCode.LINEAGE_VIOLATION,
+                    "$.replay_attempt",
+                    "replay attempt must follow its parent",
+                )
+            if replay_spec_digest != parent.replay_spec_digest:
+                _backtest_fail(
+                    BacktestContractErrorCode.LINEAGE_VIOLATION,
+                    "$.replay_spec_digest",
+                    "replay cannot claim changed deterministic inputs",
+                )
+            _, parent_computed = _backtest_instant(parent.computed_at, "$.parent.computed_at")
+            if computed_time <= parent_computed:
+                _backtest_fail(
+                    BacktestContractErrorCode.TIME_ORDER_VIOLATION,
+                    "$.computed_at",
+                    "replay computation must follow its parent",
+                )
+            parent_run_id = parent.run_id
+            ancestors = (*parent.replay_ancestor_run_ids, parent.run_id)
+            if len(ancestors) != len(set(ancestors)):
+                _backtest_fail(
+                    BacktestContractErrorCode.LINEAGE_VIOLATION,
+                    "$.replay_ancestor_run_ids",
+                    "replay lineage contains a cycle",
+                )
+
+        payload: dict[str, object] = {
+            "contract_name": "researchhub.backtest-run-ref",
+            "schema_version": BACKTEST_RUN_REF_SCHEMA_VERSION,
+            **spec_payload,
+            "computed_at": normalized_computed,
+            "replay_spec_digest": replay_spec_digest,
+            "replay_parent_run_id": parent_run_id,
+            "replay_reason": normalized_reason,
+            "replay_attempt": replay_count,
+            "replay_ancestor_run_ids": list(ancestors),
+        }
+        run_id = f"rhbacktestrunv1:sha256:{hashlib.sha256(canonical_json_bytes(payload)).hexdigest()}"
+        values: dict[str, object] = {
+            **payload,
+            "run_id": run_id,
+            "trading_calendar_revision_ids": supplied_calendars,
+            "corporate_action_revision_ids": supplied_actions,
+            "replay_ancestor_run_ids": ancestors,
+        }
+        instance = object.__new__(cls)
+        for name, value in values.items():
+            object.__setattr__(instance, name, value)
+        return instance
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "contract_name": self.contract_name,
+            "schema_version": self.schema_version,
+            "run_id": self.run_id,
+            "dataset_snapshot_id": self.dataset_snapshot_id,
+            "dataset_content_digest": self.dataset_content_digest,
+            "dataset_manifest_digest": self.dataset_manifest_digest,
+            "foundation_id": self.foundation_id,
+            "foundation_digest": self.foundation_digest,
+            "factor_set_id": self.factor_set_id,
+            "factor_set_digest": self.factor_set_digest,
+            "factor_output_content_digest": self.factor_output_content_digest,
+            "universe_digest": self.universe_digest,
+            "trading_calendar_revision_ids": list(self.trading_calendar_revision_ids),
+            "trading_calendar_digest": self.trading_calendar_digest,
+            "corporate_action_revision_ids": list(self.corporate_action_revision_ids),
+            "corporate_action_digest": self.corporate_action_digest,
+            "strategy_id": self.strategy_id,
+            "strategy_version": self.strategy_version,
+            "strategy_digest": self.strategy_digest,
+            "execution_model_version": self.execution_model_version,
+            "execution_model_digest": self.execution_model_digest,
+            "cost_model_version": self.cost_model_version,
+            "cost_model_digest": self.cost_model_digest,
+            "random_seed": self.random_seed,
+            "code_revision": self.code_revision,
+            "environment_lock_digest": self.environment_lock_digest,
+            "configuration_digest": self.configuration_digest,
+            "evaluation_at": self.evaluation_at,
+            "computed_at": self.computed_at,
+            "replay_spec_digest": self.replay_spec_digest,
+            "replay_parent_run_id": self.replay_parent_run_id,
+            "replay_reason": self.replay_reason,
+            "replay_attempt": self.replay_attempt,
+            "replay_ancestor_run_ids": list(self.replay_ancestor_run_ids),
+        }
+
+    def to_json(self) -> str:
+        return canonical_json(self.to_dict())
+
+    @classmethod
+    def from_dict(
+        cls,
+        value: Any,
+        *,
+        dataset_snapshot: DatasetSnapshotEnvelope,
+        foundation: DataFoundationEnvelope,
+        factor_set: FactorSetRef,
+        parent: BacktestRunRef | None = None,
+    ) -> Self:
+        field_names = tuple(cls.__dataclass_fields__)
+        item = _backtest_object(value, "$", field_names)
+        rebuilt = cls._build(
+            dataset_snapshot=dataset_snapshot,
+            foundation=foundation,
+            factor_set=factor_set,
+            universe_digest=item["universe_digest"],
+            trading_calendar_revision_ids=item["trading_calendar_revision_ids"],
+            corporate_action_revision_ids=item["corporate_action_revision_ids"],
+            strategy_id=item["strategy_id"],
+            strategy_version=item["strategy_version"],
+            strategy_digest=item["strategy_digest"],
+            execution_model_version=item["execution_model_version"],
+            execution_model_digest=item["execution_model_digest"],
+            cost_model_version=item["cost_model_version"],
+            cost_model_digest=item["cost_model_digest"],
+            random_seed=item["random_seed"],
+            code_revision=item["code_revision"],
+            environment_lock_digest=item["environment_lock_digest"],
+            configuration_digest=item["configuration_digest"],
+            evaluation_at=item["evaluation_at"],
+            computed_at=item["computed_at"],
+            parent=parent,
+            replay_reason=item["replay_reason"],
+            replay_attempt=item["replay_attempt"],
+        )
+        expected = rebuilt.to_dict()
+        for name in field_names:
+            if item[name] != expected[name]:
+                _backtest_fail(
+                    BacktestContractErrorCode.IDENTITY_MISMATCH,
+                    f"$.{name}",
+                    "serialized run reference does not match admitted authorities",
+                )
+        return rebuilt
+
+
 def _required_text(value: str, name: str) -> str:
    normalized = value.strip()
    if not normalized:
@@ -160,6 +896,80 @@ class FactorVersion:
        return f"{self.factor_id}@{self.version}"


+def _validate_legacy_factor_binding(
+    legacy: FactorVersion,
+    definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> None:
+    if not isinstance(legacy, FactorVersion):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.legacy",
+            "FactorVersion is required",
+        )
+    if not isinstance(definition, FactorDefinition):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.definition",
+            "complete FactorDefinition is required",
+        )
+    if not isinstance(binding, LegacyFactorBinding):
+        raise FactorContractError(
+            ContractErrorCode.TYPE_ERROR,
+            "$.binding",
+            "LegacyFactorBinding is required",
+        )
+    expected = (
+        definition.definition_id,
+        legacy.factor_id,
+        legacy.version,
+        legacy.definition_sha256,
+        legacy.dataset_schema_version,
+        definition.input_schema_digest,
+    )
+    actual = (
+        binding.definition_id,
+        binding.legacy_factor_id,
+        binding.legacy_version,
+        binding.legacy_definition_sha256,
+        binding.legacy_dataset_schema_version,
+        binding.canonical_input_schema_digest,
+    )
+    if actual != expected:
+        raise FactorContractError(
+            ContractErrorCode.LEGACY_BINDING_MISMATCH,
+            "$.binding",
+            "binding does not exactly associate the supplied legacy and canonical identities",
+        )
+
+
+def bind_legacy_factor(
+    legacy: FactorVersion,
+    complete_definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> FactorDefinition:
+    """Validate an explicit migration binding without manufacturing missing semantics."""
+
+    _validate_legacy_factor_binding(legacy, complete_definition, binding)
+    return complete_definition
+
+
+def project_legacy_factor(
+    complete_definition: FactorDefinition,
+    binding: LegacyFactorBinding,
+) -> FactorVersion:
+    """Project a canonical definition into its recorded, explicitly lossy legacy identity."""
+
+    legacy = FactorVersion(
+        factor_id=binding.legacy_factor_id,
+        version=binding.legacy_version,
+        definition_sha256=binding.legacy_definition_sha256,
+        dataset_schema_version=binding.legacy_dataset_schema_version,
+    )
+    _validate_legacy_factor_binding(legacy, complete_definition, binding)
+    return legacy
+
+
 class StrategyStage(StrEnum):
    DRAFT = "Draft"
    RESEARCH = "Research"
@@ -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"
+  }
+}
@@ -0,0 +1,206 @@
+{
+  "dataset_snapshot": {
+    "contract_name": "researchhub.dataset-snapshot",
+    "schema_version": "1.0.0",
+    "snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+    "descriptor": {
+      "dataset": {
+        "dataset_id": "rhdataset:market:0123456789abcdef0123456789abcdef",
+        "dataset_kind": "market",
+        "record_schema_version": "1.0.0",
+        "dimensions": ["instrument_id", "effective_time"]
+      },
+      "published_at": "2026-01-02T07:05:00Z",
+      "time_semantics": {
+        "effective_time": {
+          "start_inclusive": "2026-01-02T07:00:00Z",
+          "end_inclusive": "2026-01-02T07:00:00Z"
+        },
+        "knowledge_time": {
+          "start_inclusive": "2026-01-02T07:01:00Z",
+          "end_inclusive": "2026-01-02T07:01:00Z"
+        },
+        "pit_cutoff": "2026-01-02T07:01:00Z"
+      },
+      "content": {
+        "digest_algorithm": "sha256",
+        "canonicalization": "RFC8785",
+        "record_order": "canonical-record-byte-order",
+        "content_digest": "sha256:44ea11ba64dc2e6fd55c6d8e038c5edc84ee38d6fd46e38b15a1d5409662a020",
+        "logical_manifest": {
+          "record_count": 2,
+          "chunks": [
+            {
+              "chunk_index": 0,
+              "content_digest": "sha256:44ea11ba64dc2e6fd55c6d8e038c5edc84ee38d6fd46e38b15a1d5409662a020",
+              "record_count": 2
+            }
+          ]
+        },
+        "manifest_digest": "sha256:d991bb2f8f6b80525f93c51e0b371213a3ed4649dffb073ed4605bfbd32349bd",
+        "record_count": 2
+      },
+      "lineage": {
+        "publisher": {"id": "researchhub.data", "version": "1.0.0"},
+        "transformation": {
+          "id": "rhtransform:00112233445566778899aabbccddeeff",
+          "version": "1.0.0"
+        },
+        "upstream_snapshot_ids": [],
+        "upstream_content_digests": []
+      },
+      "quality": {
+        "status": "passed",
+        "checks": [
+          {
+            "check_id": "completeness",
+            "status": "passed",
+            "severity": "blocking",
+            "evidence_digest": "sha256:876fc2fcc6414ddc3f824a47f475d34c82a53d2bda5dc72a234a3f3164e8e2ec"
+          },
+          {
+            "check_id": "pit_time_integrity",
+            "status": "passed",
+            "severity": "blocking",
+            "evidence_digest": "sha256:90a6cc46b9f2ab317a1c6d14dc173784e19b5621cd7fe956e8f41338cbdc5944"
+          }
+        ]
+      },
+      "qualification": {
+        "status": "qualified",
+        "policy_id": "researchhub.dataset-snapshot.pit",
+        "policy_version": "1.0.0",
+        "evaluated_at": "2026-01-02T07:04:00Z",
+        "evidence_digest": "sha256:e192462f9022f2b477f73cdbe9e6c9f891ebcfdc2b4ed4f8ddd7b1ff107ee6a6"
+      }
+    }
+  },
+  "data_foundation": {
+    "contract_name": "researchhub.data-foundation",
+    "schema_version": "1.0.0",
+    "foundation_id": "rhdfv1:sha256:d848237ab753ee9432ae78ec1f93b6ac45c8072d6694023b7f288203daf9d838",
+    "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+    "pit_cutoff": "2026-01-03T00:00:00Z",
+    "instrument_routes": [
+      {
+        "route_revision_id": "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "revision_number": 1,
+        "symbol": "600000",
+        "mic": "XSHG",
+        "currency": "CNY",
+        "asset_class": "equity",
+        "instrument_type": "stock",
+        "calendar_id": "rhcalendar:11112222333344445555666677778888",
+        "effective_from": "2020-01-01T00:00:00Z",
+        "knowledge_time": "2026-01-01T07:00:00Z",
+        "evidence_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+      }
+    ],
+    "trading_calendar_revisions": [
+      {
+        "calendar_revision_id": "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078",
+        "calendar_id": "rhcalendar:11112222333344445555666677778888",
+        "session_date": "2026-01-02",
+        "revision_number": 1,
+        "status": "open",
+        "sessions": [
+          {"opens_at": "2026-01-02T01:30:00Z", "closes_at": "2026-01-02T07:00:00Z"}
+        ],
+        "knowledge_time": "2026-01-01T08:00:00Z",
+        "evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+      }
+    ],
+    "corporate_action_revisions": [
+      {
+        "action_revision_id": "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275",
+        "action_id": "rhaction:99998888777766665555444433332222",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "revision_number": 1,
+        "action_type": "cash_dividend",
+        "status": "confirmed",
+        "effective_time": "2026-01-02T00:00:00Z",
+        "knowledge_time": "2026-01-01T09:00:00Z",
+        "terms_digest": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
+        "evidence_digest": "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+      }
+    ],
+    "standardized_views": [
+      {
+        "view_ref_id": "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c",
+        "view_id": "rhview:abcdef0123456789abcdef0123456789",
+        "view_version": "1.0.0",
+        "dataset_snapshot_id": "rhdsv1:sha256:f63a29b4795c63fb7d6b2d3b5544cee9274b633db77c75c63d50a340c0827d57",
+        "pit_cutoff": "2026-01-03T00:00:00Z",
+        "schema_digest": "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+        "content_digest": "sha256:123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0",
+        "transformation_digest": "sha256:23456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef01",
+        "instrument_route_revision_ids": [
+          "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86"
+        ],
+        "trading_calendar_revision_ids": [
+          "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078"
+        ],
+        "corporate_action_revision_ids": [
+          "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275"
+        ]
+      }
+    ],
+    "revision_lineage": [
+      {
+        "revision_kind": "instrument_route",
+        "revision_id": "rhroutev1:sha256:ca67013250e28ab4cce16570607379a8792400e62415ee6cb71c75508e2f3d86",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T07:00:00Z",
+        "evidence_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+      },
+      {
+        "revision_kind": "trading_calendar",
+        "revision_id": "rhcalv1:sha256:1f4ca22557063389badd669cf774bb35234066e847646682dccfe411e252a078",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T08:00:00Z",
+        "evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+      },
+      {
+        "revision_kind": "corporate_action",
+        "revision_id": "rhcav1:sha256:0f947df29f152bfa2c6ab0da7a0d464670c7ad2526cd10f2a7939b42fee5c275",
+        "revision_number": 1,
+        "knowledge_time": "2026-01-01T09:00:00Z",
+        "evidence_digest": "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+      }
+    ],
+    "readiness": {
+      "evidence_scope": "synthetic_fixture",
+      "contract_validation": {
+        "status": "validated",
+        "evidence_digests": [
+          "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
+        ]
+      },
+      "real_data_validation": {"status": "not_validated", "evidence_digests": []},
+      "production_validation": {"status": "not_validated", "evidence_digests": []},
+      "live_validation": {"status": "not_validated", "evidence_digests": []}
+    }
+  },
+  "output_schema": {
+    "columns": ["evaluation_at", "factor_id", "instrument_id", "value"],
+    "schema_version": "1.0.0"
+  },
+  "output_content": {
+    "rows": [
+      {
+        "evaluation_at": "2026-01-03T11:00:00Z",
+        "factor_id": "alpha_005",
+        "instrument_id": "rhinstrument:0123456789abcdef0123456789abcdef",
+        "value": "0.125"
+      }
+    ]
+  },
+  "expected": {
+    "definition_id": "rhfactorv1:sha256:978fb8000d318373844a5e044ca14bf377e01ebe8d85964b826ecd2af9085ce9",
+    "input_schema_digest": "sha256:4501aeab99b4bcc25a1b8813ebe197fb498053fd710d73746bf20fc8eeb4bfa7",
+    "factor_set_id": "rhfactorsetv1:sha256:e9339581cf569e92459f672e8081337712e7bf98e58ad42d60d7ed13f9b5a021",
+    "output_artifact_id": "rhfactoroutputv1:sha256:a4803b5ff66d12d3a0e7e8e5b8cca953cbc137e2bf41514b7c4e5f05da5ee68b",
+    "legacy_binding_id": "rhlegacyfactorv1:sha256:541bc5a9469f9c8e4c2d696a9972fc5f2e6e2bef218b86f728823994b915dede"
+  }
+}
@@ -1,32 +1,58 @@
 from __future__ import annotations

 import json
-import unittest
 from pathlib import Path


 ROOT = Path(__file__).resolve().parents[2]


-class ModuleSpecTests(unittest.TestCase):
-    def test_module_spec_declares_pure_research_engine_boundary(self) -> None:
+def test_module_spec_declares_pure_research_engine_boundary() -> None:
    spec = json.loads((ROOT / "MODULE_SPEC.yaml").read_text(encoding="utf-8"))
-        self.assertEqual(spec["module_id"], "quant_engine")
-        self.assertEqual(spec["authority"]["subject"], spec["module_id"])
-        self.assertEqual(spec["repository"]["type"], "research_engine")
-        self.assertEqual(spec["bounded_context"]["domain"], "quantitative-research-engine")
+    assert spec["module_id"] == "quant_engine"
+    assert spec["authority"]["subject"] == spec["module_id"]
+    assert spec["repository"]["type"] == "research_engine"
+    assert spec["bounded_context"]["domain"] == "quantitative-research-engine"
    prohibited = " ".join(spec["bounded_context"]["prohibited_responsibilities"]).lower()
    for term in ("investment advice", "live order", "credentials", "source facts"):
-            self.assertIn(term, prohibited)
-        self.assertEqual(spec["contracts"], {"provides": [], "consumes": []})
-        self.assertEqual(spec["dependencies"], [])
-        self.assertTrue(
-            all(
+        assert term in prohibited
+    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"),
+    }
+    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"]
+    } == {
+        ("researchhub.dataset-snapshot", "1.0.0"),
+        ("researchhub.data-foundation", "1.0.0"),
+    }
+    assert all(
+        item["authority"] == "researchhub.data"
+        for item in spec["contracts"]["consumes"]
+    )
+    assert spec["dependencies"] == []
+    assert all(
        command["required"] and not command["network"]
        for command in spec["verification"]["commands"]
    )
-        )


 if __name__ == "__main__":
-    unittest.main()
+    test_module_spec_declares_pure_research_engine_boundary()
@@ -7,6 +7,13 @@ import pandas as pd
 import pytest

 import quant_engine.alpha_factors as alpha_factors_module
+from quant_engine.factor_contracts import (
+    FactorContractError,
+    FactorInput,
+    ProducerIdentity,
+    factor_definition_from_alpha158,
+    factor_input_schema_digest,
+)
 from quant_engine.alpha_factors import (
    ALPHA158_REGISTRY,
    ALPHA158_PHASE1_OPERATOR_SPECS,
@@ -434,6 +441,50 @@ def test_alpha_registry_required_fields():
        assert required <= set(meta.keys()), f"{alpha_id} missing fields"


+def test_alpha_registry_adapts_to_definition_without_copying_formula_or_inputs():
+    factor_input = FactorInput(
+        "market",
+        "sha256:" + "1" * 64,
+        tuple(ALPHA158_REGISTRY["alpha_005"]["inputs"]),
+    )
+    definition = factor_definition_from_alpha158(
+        "alpha_005",
+        version="1.0.0",
+        parameters={},
+        inputs=(factor_input,),
+        implementation_digest="sha256:" + "2" * 64,
+        input_schema_digest=factor_input_schema_digest((factor_input,)),
+        valid_from="2026-01-01T00:00:00Z",
+        valid_until="2027-01-01T00:00:00Z",
+        warmup_sessions=10,
+        lag_sessions=1,
+        producer=ProducerIdentity("quant_engine", "1.0.0"),
+        code_revision="c" * 40,
+    )
+
+    assert definition.formula == ALPHA158_REGISTRY["alpha_005"]["formula"]
+    assert definition.inputs[0].required_columns == tuple(
+        ALPHA158_REGISTRY["alpha_005"]["inputs"]
+    )
+
+    incomplete = FactorInput("market", "sha256:" + "1" * 64, ("close",))
+    with pytest.raises(FactorContractError, match="exactly correspond"):
+        factor_definition_from_alpha158(
+            "alpha_005",
+            version="1.0.0",
+            parameters={},
+            inputs=(incomplete,),
+            implementation_digest="sha256:" + "2" * 64,
+            input_schema_digest=factor_input_schema_digest((incomplete,)),
+            valid_from="2026-01-01T00:00:00Z",
+            valid_until="2027-01-01T00:00:00Z",
+            warmup_sessions=10,
+            lag_sessions=1,
+            producer=ProducerIdentity("quant_engine", "1.0.0"),
+            code_revision="c" * 40,
+        )
+
+
 def test_get_alpha_meta_success():
    """已知 alpha_id 返回完整 meta。"""
    meta = get_alpha_meta("alpha_001")
@@ -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()
@@ -0,0 +1,938 @@
+"""Versioned factor-definition and factor-set contract conformance."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from dataclasses import FrozenInstanceError
+from pathlib import Path
+from typing import Any, Callable
+
+import pytest
+
+from quant_engine.factor_contracts import (
+    ActorIdentity,
+    AvailabilityMode,
+    ContractErrorCode,
+    Causation,
+    DataFoundationEnvelope,
+    DatasetSnapshotEnvelope,
+    FactorContractError,
+    FactorDefinition,
+    FactorInput,
+    FactorSetRef,
+    HistoricalAvailability,
+    InputBinding,
+    LegacyFactorBinding,
+    OutputArtifactRef,
+    OutputCoverage,
+    OutputQuality,
+    OutputQualityCheck,
+    PayloadValidation,
+    ProducerIdentity,
+    TypedParameter,
+    ViewAvailability,
+    canonical_json_bytes,
+    factor_definition_from_alpha158,
+    factor_input_schema_digest,
+    validate_factor_catalog,
+)
+from quant_engine.governed_pipeline import (
+    FactorVersion,
+    bind_legacy_factor,
+    project_legacy_factor,
+)
+
+
+FIXTURE_PATH = Path(__file__).parent / "fixtures" / "factor-contracts-v1.golden.json"
+VIEW_REF_ID = "rhviewrefv1:sha256:bf776bcd26d940fafde1d650776a5505fb3fe8b5b068c351622bf2c42385629c"
+VIEW_SCHEMA_DIGEST = "sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+
+
+def _golden() -> dict[str, Any]:
+    loaded = json.loads(FIXTURE_PATH.read_text(encoding="utf-8"))
+    assert isinstance(loaded, dict)
+    return loaded
+
+
+def _sha256(value: bytes) -> str:
+    return f"sha256:{hashlib.sha256(value).hexdigest()}"
+
+
+def _reidentify(item: dict[str, Any], field: str, prefix: str) -> None:
+    payload = {key: value for key, value in item.items() if key != field}
+    item[field] = f"{prefix}{hashlib.sha256(canonical_json_bytes(payload)).hexdigest()}"
+
+
+def _snapshot_and_foundation(
+    fixture: dict[str, Any] | None = None,
+) -> tuple[DatasetSnapshotEnvelope, DataFoundationEnvelope]:
+    source = _golden() if fixture is None else fixture
+    return (
+        DatasetSnapshotEnvelope.from_dict(source["dataset_snapshot"]),
+        DataFoundationEnvelope.from_dict(source["data_foundation"]),
+    )
+
+
+def _definition(
+    *,
+    inputs: tuple[FactorInput, ...] | None = None,
+    **overrides: Any,
+) -> FactorDefinition:
+    factor_inputs = inputs or (
+        FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume")),
+    )
+    arguments: dict[str, Any] = {
+        "factor_id": "alpha_005",
+        "version": "1.0.0",
+        "formula": "correlation(close, volume, 10)",
+        "parameters": {},
+        "implementation_digest": "sha256:" + "1" * 64,
+        "input_schema_digest": factor_input_schema_digest(factor_inputs),
+        "inputs": factor_inputs,
+        "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,
+    }
+    arguments.update(overrides)
+    return FactorDefinition.create(**arguments)
+
+
+def _golden_definition() -> FactorDefinition:
+    factor_input = FactorInput("market", VIEW_SCHEMA_DIGEST, ("close", "volume"))
+    return 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,
+    )
+
+
+def _factor_set_arguments(
+    *,
+    fixture: dict[str, Any] | None = None,
+    snapshot: DatasetSnapshotEnvelope | None = None,
+    foundation: DataFoundationEnvelope | None = None,
+    definition: FactorDefinition | None = None,
+) -> dict[str, Any]:
+    source = _golden() if fixture is None else fixture
+    if snapshot is None or foundation is None:
+        parsed_snapshot, parsed_foundation = _snapshot_and_foundation(source)
+        snapshot = snapshot or parsed_snapshot
+        foundation = foundation or parsed_foundation
+    selected_definition = definition or _golden_definition()
+    output_schema_bytes = canonical_json_bytes(source["output_schema"])
+    output_content_bytes = canonical_json_bytes(source["output_content"])
+    artifact = OutputArtifactRef.create(
+        schema_digest=_sha256(output_schema_bytes),
+        content_digest=_sha256(output_content_bytes),
+    )
+    return {
+        "definitions": (selected_definition,),
+        "dataset_snapshot": snapshot,
+        "foundation": foundation,
+        "selected_view_ref_ids": (VIEW_REF_ID,),
+        "input_bindings": (
+            InputBinding(
+                selected_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,
+        "availability_mode": AvailabilityMode.AS_AVAILABLE,
+        "evaluation_at": "2026-01-03T11:00:00Z",
+        "computed_at": "2026-01-03T10:15:00Z",
+        "artifact_available_at": "2026-01-03T10:20:00Z",
+        "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,
+    }
+
+
+def _factor_set(**overrides: Any) -> FactorSetRef:
+    arguments = _factor_set_arguments()
+    arguments.update(overrides)
+    return FactorSetRef.create(**arguments)
+
+
+def _assert_error(
+    error: pytest.ExceptionInfo[FactorContractError],
+    code: ContractErrorCode,
+    path: str,
+) -> None:
+    assert error.value.code is code
+    assert error.value.path == path
+
+
+def _mutate_artifact_schema_binding(value: dict[str, Any]) -> None:
+    artifact = value["output_artifact_ref"]
+    artifact["schema_digest"] = "sha256:" + "0" * 64
+    _reidentify(artifact, "artifact_id", "rhfactoroutputv1:sha256:")
+
+
+def test_golden_contracts_are_content_addressed_round_trippable_and_deeply_immutable() -> None:
+    fixture = _golden()
+    original_snapshot = copy.deepcopy(fixture["dataset_snapshot"])
+    original_foundation = copy.deepcopy(fixture["data_foundation"])
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    definition = _golden_definition()
+    factor_set = FactorSetRef.create(**_factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition))
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id="factor:demo-momentum",
+        legacy_version="1.0.0",
+        legacy_definition_sha256="b" * 64,
+        legacy_dataset_schema_version="1.0.0",
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+
+    assert snapshot.pit_cutoff == "2026-01-02T07:01:00Z"
+    assert foundation.pit_cutoff == factor_set.pit_cutoff == "2026-01-03T00:00:00Z"
+    assert snapshot.pit_cutoff != foundation.pit_cutoff
+    assert definition.definition_id == fixture["expected"]["definition_id"]
+    assert definition.input_schema_digest == fixture["expected"]["input_schema_digest"]
+    assert factor_set.factor_set_id == fixture["expected"]["factor_set_id"]
+    assert factor_set.output_artifact_ref.artifact_id == fixture["expected"]["output_artifact_id"]
+    assert binding.binding_id == fixture["expected"]["legacy_binding_id"]
+    assert not definition.to_json().endswith("\n")
+    assert not factor_set.to_json().endswith("\n")
+    assert FactorDefinition.from_json(definition.to_json()) == definition
+
+    reparsed = FactorSetRef.from_json(
+        factor_set.to_json(),
+        definitions=(definition,),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+        output_schema_bytes=canonical_json_bytes(fixture["output_schema"]),
+        output_content_bytes=canonical_json_bytes(fixture["output_content"]),
+    )
+    reference_only = FactorSetRef.from_json(
+        factor_set.to_json(),
+        definitions=(definition,),
+        dataset_snapshot=snapshot,
+        foundation=foundation,
+    )
+    assert reparsed.factor_set_id == factor_set.factor_set_id
+    assert reparsed.payload_validation is PayloadValidation.PAYLOAD_REVALIDATED
+    assert reference_only.payload_validation is PayloadValidation.REFERENCE_ONLY
+
+    fixture["dataset_snapshot"]["descriptor"]["dataset"]["dimensions"].append("forbidden")
+    fixture["data_foundation"]["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
+    assert snapshot.to_dict() == original_snapshot
+    assert foundation.to_dict() == original_foundation
+    returned = snapshot.to_dict()
+    returned["descriptor"]["dataset"]["dimensions"].append("also_forbidden")
+    assert snapshot.to_dict() == original_snapshot
+    with pytest.raises(FrozenInstanceError):
+        snapshot.snapshot_id = "rhdsv1:sha256:" + "0" * 64  # type: ignore[misc]
+
+
+@pytest.mark.parametrize("variant", ["whitespace", "key_order"])
+def test_contract_decoders_reject_non_canonical_json(variant: str) -> None:
+    fixture = _golden()
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    definition = _golden_definition()
+    factor_set = FactorSetRef.create(
+        **_factor_set_arguments(
+            fixture=fixture,
+            snapshot=snapshot,
+            foundation=foundation,
+            definition=definition,
+        )
+    )
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id="factor:demo-momentum",
+        legacy_version="1.0.0",
+        legacy_definition_sha256="b" * 64,
+        legacy_dataset_schema_version="1.0.0",
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+
+    def non_canonical(value: str) -> str:
+        if variant == "whitespace":
+            return value + "\n"
+        loaded = json.loads(value)
+        reversed_items = dict(reversed(tuple(loaded.items())))
+        return json.dumps(reversed_items, ensure_ascii=False, separators=(",", ":"))
+
+    decoders = (
+        lambda value: FactorDefinition.from_json(value),
+        lambda value: FactorSetRef.from_json(
+            value,
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        ),
+        lambda value: LegacyFactorBinding.from_json(value, definition=definition),
+    )
+    for decoder, encoded in zip(
+        decoders,
+        (definition.to_json(), factor_set.to_json(), binding.to_json()),
+        strict=True,
+    ):
+        with pytest.raises(FactorContractError) as exc_info:
+            decoder(non_canonical(encoded))
+        assert exc_info.value.code is ContractErrorCode.INVALID_FORMAT
+        assert exc_info.value.path == "$"
+
+
+def test_factor_definition_identity_is_order_independent_where_semantics_are_unordered() -> None:
+    first_input = FactorInput("prices", "sha256:" + "6" * 64, ("close",))
+    second_input = FactorInput("volumes", "sha256:" + "7" * 64, ("volume",))
+    inputs = (first_input, second_input)
+    parameters_a = {
+        "window": TypedParameter("integer", 10),
+        "weights": TypedParameter("json", {"fast": [1, 2], "slow": [3, 4]}),
+    }
+    parameters_b = {
+        "weights": TypedParameter("json", {"slow": [3, 4], "fast": [1, 2]}),
+        "window": TypedParameter("integer", 10),
+    }
+    first = _definition(
+        inputs=inputs,
+        parameters=parameters_a,
+        input_schema_digest=factor_input_schema_digest(inputs),
+    )
+    second = _definition(
+        inputs=tuple(reversed(inputs)),
+        parameters=parameters_b,
+        input_schema_digest=factor_input_schema_digest(tuple(reversed(inputs))),
+    )
+    assert first.definition_id == second.definition_id
+    assert first.to_json() == second.to_json()
+
+    semantic_changes = (
+        _definition(factor_id="alpha_006"),
+        _definition(version="1.0.1"),
+        _definition(formula="correlation(close, volume, 11)"),
+        _definition(parameters={"window": TypedParameter("integer", 10)}),
+        _definition(implementation_digest="sha256:" + "9" * 64),
+        _definition(valid_until="2027-01-02T00:00:00Z"),
+        _definition(warmup_sessions=11),
+        _definition(lag_sessions=2),
+        _definition(producer=ProducerIdentity("quant_engine", "1.0.1")),
+        _definition(code_revision="d" * 40),
+    )
+    assert all(changed.definition_id != _golden_definition().definition_id for changed in semantic_changes)
+    assert len({changed.definition_id for changed in semantic_changes}) == len(semantic_changes)
+
+
+def test_parameter_types_decimal_profile_and_detached_nested_values_are_strict() -> None:
+    nested = {"ordered": [1, {"flag": True}]}
+    parameter = TypedParameter("json", nested)
+    nested["ordered"].append(2)
+    definition = _definition(parameters={"payload": parameter})
+    assert definition.to_dict()["parameters"]["payload"]["value"] == {
+        "ordered": [1, {"flag": True}]
+    }
+    integer_definition = _definition(parameters={"value": TypedParameter("integer", 1)})
+    string_definition = _definition(parameters={"value": TypedParameter("string", "1")})
+    assert integer_definition.definition_id != string_definition.definition_id
+
+    for parameter_type, value, code in (
+        ("decimal", "1.0", ContractErrorCode.INVALID_FORMAT),
+        ("decimal", "1e3", ContractErrorCode.INVALID_FORMAT),
+        ("decimal", "-0", ContractErrorCode.INVALID_FORMAT),
+        ("integer", True, ContractErrorCode.TYPE_ERROR),
+        ("json", 1.5, ContractErrorCode.TYPE_ERROR),
+        ("json", {"é": "bad-key"}, ContractErrorCode.INVALID_FORMAT),
+        ("json", 9_007_199_254_740_992, ContractErrorCode.INVALID_VALUE),
+    ):
+        with pytest.raises(FactorContractError) as error:
+            TypedParameter(parameter_type, value)
+        assert error.value.code is code
+    assert TypedParameter("decimal", "10.25").to_dict()["value"] == "10.25"
+
+
+def test_catalog_rejects_duplicate_and_overlapping_logical_validity_but_allows_adjacency() -> None:
+    base = _golden_definition()
+    adjacent = _definition(valid_from="2027-01-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
+    assert len(validate_factor_catalog((adjacent, base))) == 2
+    with pytest.raises(FactorContractError) as duplicate:
+        validate_factor_catalog((base, base))
+    _assert_error(duplicate, ContractErrorCode.INVALID_VALUE, "$.definitions")
+    overlapping = _definition(valid_from="2026-06-01T00:00:00Z", valid_until="2028-01-01T00:00:00Z")
+    with pytest.raises(FactorContractError) as overlap:
+        validate_factor_catalog((base, overlapping))
+    _assert_error(overlap, ContractErrorCode.TIME_ORDER_VIOLATION, "$.definitions")
+
+
+def test_upstream_contracts_reject_unknown_fields_identity_forgery_and_unqualified_input() -> None:
+    unknown = _golden()["dataset_snapshot"]
+    unknown["provider"] = "forbidden"
+    with pytest.raises(FactorContractError) as unknown_error:
+        DatasetSnapshotEnvelope.from_dict(unknown)
+    _assert_error(unknown_error, ContractErrorCode.UNKNOWN_FIELD, "$.provider")
+
+    forged = _golden()["data_foundation"]
+    forged["standardized_views"][0]["schema_digest"] = "sha256:" + "0" * 64
+    with pytest.raises(FactorContractError) as forged_error:
+        DataFoundationEnvelope.from_dict(forged)
+    assert forged_error.value.code is ContractErrorCode.IDENTITY_MISMATCH
+    assert forged_error.value.path.endswith("view_ref_id")
+
+    rejected_source = _golden()
+    rejected_source["dataset_snapshot"]["descriptor"]["qualification"]["status"] = "rejected"
+    _reidentify(rejected_source["dataset_snapshot"], "snapshot_id", "rhdsv1:sha256:")
+    rejected_snapshot = DatasetSnapshotEnvelope.from_dict(rejected_source["dataset_snapshot"])
+    _, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as rejected_error:
+        FactorSetRef.create(
+            **_factor_set_arguments(snapshot=rejected_snapshot, foundation=foundation)
+        )
+    _assert_error(
+        rejected_error,
+        ContractErrorCode.QUALIFICATION_REJECTED,
+        "$.dataset_snapshot.descriptor.qualification",
+    )
+
+
+def test_foundation_rejects_future_knowledge_and_per_view_calendar_borrowing() -> None:
+    future = _golden()["data_foundation"]
+    action = future["corporate_action_revisions"][0]
+    old_action_id = action["action_revision_id"]
+    action["knowledge_time"] = "2026-01-03T00:00:01Z"
+    _reidentify(action, "action_revision_id", "rhcav1:sha256:")
+    future["standardized_views"][0]["corporate_action_revision_ids"] = [action["action_revision_id"]]
+    lineage = next(item for item in future["revision_lineage"] if item["revision_id"] == old_action_id)
+    lineage["revision_id"] = action["action_revision_id"]
+    lineage["knowledge_time"] = action["knowledge_time"]
+    _reidentify(future["standardized_views"][0], "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(future, "foundation_id", "rhdfv1:sha256:")
+    with pytest.raises(FactorContractError) as future_error:
+        DataFoundationEnvelope.from_dict(future)
+    _assert_error(
+        future_error,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+        "$.revision_lineage.knowledge_time",
+    )
+
+    uncovered = _golden()["data_foundation"]
+    original_route_id = uncovered["instrument_routes"][0]["route_revision_id"]
+    second_calendar = copy.deepcopy(uncovered["trading_calendar_revisions"][0])
+    second_calendar["calendar_id"] = "rhcalendar:99990000111122223333444455556666"
+    _reidentify(second_calendar, "calendar_revision_id", "rhcalv1:sha256:")
+    uncovered["trading_calendar_revisions"].append(second_calendar)
+    route = uncovered["instrument_routes"][0]
+    route["calendar_id"] = second_calendar["calendar_id"]
+    _reidentify(route, "route_revision_id", "rhroutev1:sha256:")
+    route_lineage = next(item for item in uncovered["revision_lineage"] if item["revision_id"] == original_route_id)
+    route_lineage["revision_id"] = route["route_revision_id"]
+    uncovered["revision_lineage"].append(
+        {
+            "revision_kind": "trading_calendar",
+            "revision_id": second_calendar["calendar_revision_id"],
+            "revision_number": 1,
+            "knowledge_time": second_calendar["knowledge_time"],
+            "evidence_digest": second_calendar["evidence_digest"],
+        }
+    )
+    view = uncovered["standardized_views"][0]
+    view["instrument_route_revision_ids"] = [route["route_revision_id"]]
+    _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(uncovered, "foundation_id", "rhdfv1:sha256:")
+    with pytest.raises(FactorContractError) as calendar_error:
+        DataFoundationEnvelope.from_dict(uncovered)
+    assert calendar_error.value.code is ContractErrorCode.INPUT_CLOSURE_VIOLATION
+    assert "selected route calendar" in calendar_error.value.detail
+
+
+def _replay_fixture() -> dict[str, Any]:
+    fixture = _golden()
+    snapshot = fixture["dataset_snapshot"]
+    snapshot["descriptor"]["published_at"] = "2026-01-04T00:00:00Z"
+    _reidentify(snapshot, "snapshot_id", "rhdsv1:sha256:")
+    foundation = fixture["data_foundation"]
+    foundation["dataset_snapshot_id"] = snapshot["snapshot_id"]
+    for view in foundation["standardized_views"]:
+        view["dataset_snapshot_id"] = snapshot["snapshot_id"]
+        _reidentify(view, "view_ref_id", "rhviewrefv1:sha256:")
+    _reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
+    return fixture
+
+
+def test_as_available_and_retrospective_replay_keep_distinct_time_claims() -> None:
+    as_available = _factor_set()
+    assert as_available.historical_availability is HistoricalAvailability.DECLARED_AS_AVAILABLE
+
+    replay_source = _replay_fixture()
+    snapshot, foundation = _snapshot_and_foundation(replay_source)
+    replay_view_id = next(iter(foundation.views))
+    arguments = _factor_set_arguments(
+        fixture=replay_source,
+        snapshot=snapshot,
+        foundation=foundation,
+    )
+    arguments.update(
+        selected_view_ref_ids=(replay_view_id,),
+        input_bindings=(
+            InputBinding(
+                arguments["definitions"][0].definition_id,
+                "market",
+                replay_view_id,
+                VIEW_SCHEMA_DIGEST,
+            ),
+        ),
+        view_availability=(
+            ViewAvailability(replay_view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),
+        ),
+        availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
+        computed_at="2026-01-04T00:20:00Z",
+        artifact_available_at="2026-01-04T00:25:00Z",
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    replay = FactorSetRef.create(**arguments)
+    assert replay.evaluation_at == "2026-01-03T11:00:00Z"
+    assert replay.computed_at == "2026-01-04T00:20:00Z"
+    assert replay.historical_availability is HistoricalAvailability.NOT_ESTABLISHED
+
+    replay_source_args = _factor_set_arguments(
+        fixture=replay_source,
+        snapshot=snapshot,
+        foundation=foundation,
+    )
+    replay_source_args.update(
+        selected_view_ref_ids=(replay_view_id,),
+        input_bindings=(
+            InputBinding(
+                replay_source_args["definitions"][0].definition_id,
+                "market",
+                replay_view_id,
+                VIEW_SCHEMA_DIGEST,
+            ),
+        ),
+        view_availability=(
+            ViewAvailability(replay_view_id, "2026-01-02T23:50:00Z", "sha256:" + "2" * 64),
+        ),
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    with pytest.raises(FactorContractError) as late_publication:
+        FactorSetRef.create(**replay_source_args)
+    _assert_error(
+        late_publication,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+        "$.dataset_snapshot.descriptor.published_at",
+    )
+
+
+@pytest.mark.parametrize(
+    ("overrides", "path"),
+    [
+        ({"view_availability": (ViewAvailability(VIEW_REF_ID, "2026-01-03T00:00:01Z", "sha256:" + "2" * 64),)}, "$.view_availability"),
+        ({"computed_at": "2026-01-02T23:40:00Z"}, "$.computed_at"),
+        ({"artifact_available_at": "2026-01-03T10:14:00Z"}, "$.artifact_available_at"),
+        ({"artifact_available_at": "2026-01-03T11:00:01Z"}, "$.artifact_available_at"),
+        ({"evaluation_at": "2026-01-03T11:00:00"}, "$.evaluation_at"),
+    ],
+)
+def test_as_available_time_failures_are_typed(overrides: dict[str, Any], path: str) -> None:
+    with pytest.raises(FactorContractError) as error:
+        _factor_set(**overrides)
+    assert error.value.code in {
+        ContractErrorCode.INVALID_FORMAT,
+        ContractErrorCode.TIME_ORDER_VIOLATION,
+    }
+    assert error.value.path == path
+
+
+def test_replay_rejects_backdating_and_historical_availability_promotion() -> None:
+    source = _replay_fixture()
+    snapshot, foundation = _snapshot_and_foundation(source)
+    view_id = next(iter(foundation.views))
+    arguments = _factor_set_arguments(fixture=source, snapshot=snapshot, foundation=foundation)
+    definition = arguments["definitions"][0]
+    arguments.update(
+        selected_view_ref_ids=(view_id,),
+        input_bindings=(InputBinding(definition.definition_id, "market", view_id, VIEW_SCHEMA_DIGEST),),
+        view_availability=(ViewAvailability(view_id, "2026-01-04T00:10:00Z", "sha256:" + "2" * 64),),
+        availability_mode=AvailabilityMode.RETROSPECTIVE_REPLAY,
+        computed_at="2026-01-04T00:20:00Z",
+        artifact_available_at="2026-01-04T00:25:00Z",
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    replay = FactorSetRef.create(**arguments)
+    promoted = replay.to_dict()
+    promoted["historical_availability"] = "declared_as_available"
+    _reidentify(promoted, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as promotion_error:
+        FactorSetRef.from_dict(
+            promoted,
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        promotion_error,
+        ContractErrorCode.READINESS_ESCALATION,
+        "$.historical_availability",
+    )
+    arguments["computed_at"] = "2026-01-03T11:30:00Z"
+    with pytest.raises(FactorContractError) as backdated_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(backdated_error, ContractErrorCode.TIME_ORDER_VIOLATION, "$.computed_at")
+
+
+def _multi_view_fixture() -> tuple[dict[str, Any], str]:
+    fixture = _golden()
+    foundation = fixture["data_foundation"]
+    second = copy.deepcopy(foundation["standardized_views"][0])
+    second["view_id"] = "rhview:11111111222222223333333344444444"
+    second["schema_digest"] = "sha256:" + "6" * 64
+    second["content_digest"] = "sha256:" + "7" * 64
+    second["transformation_digest"] = "sha256:" + "8" * 64
+    _reidentify(second, "view_ref_id", "rhviewrefv1:sha256:")
+    foundation["standardized_views"].append(second)
+    _reidentify(foundation, "foundation_id", "rhdfv1:sha256:")
+    return fixture, second["view_ref_id"]
+
+
+def test_multi_input_mapping_requires_exact_consumption_closure_and_is_order_independent() -> None:
+    fixture, second_view_id = _multi_view_fixture()
+    snapshot, foundation = _snapshot_and_foundation(fixture)
+    inputs = (
+        FactorInput("prices", VIEW_SCHEMA_DIGEST, ("close",)),
+        FactorInput("volumes", "sha256:" + "6" * 64, ("volume",)),
+    )
+    definition = _definition(
+        inputs=inputs,
+        formula="correlation(close, volume, 10)",
+        input_schema_digest=factor_input_schema_digest(inputs),
+    )
+    first_binding = InputBinding(definition.definition_id, "prices", VIEW_REF_ID, VIEW_SCHEMA_DIGEST)
+    second_binding = InputBinding(definition.definition_id, "volumes", second_view_id, "sha256:" + "6" * 64)
+    first_availability = ViewAvailability(VIEW_REF_ID, "2026-01-02T23:40:00Z", "sha256:" + "2" * 64)
+    second_availability = ViewAvailability(second_view_id, "2026-01-02T23:50:00Z", "sha256:" + "6" * 64)
+    base = _factor_set_arguments(fixture=fixture, snapshot=snapshot, foundation=foundation, definition=definition)
+    base.update(
+        selected_view_ref_ids=(VIEW_REF_ID, second_view_id),
+        input_bindings=(first_binding, second_binding),
+        view_availability=(first_availability, second_availability),
+        causation=Causation("foundation", foundation.foundation_id),
+    )
+    first = FactorSetRef.create(**base)
+    reordered = dict(base)
+    reordered.update(
+        selected_view_ref_ids=(second_view_id, VIEW_REF_ID),
+        input_bindings=(second_binding, first_binding),
+        view_availability=(second_availability, first_availability),
+    )
+    assert FactorSetRef.create(**reordered).factor_set_id == first.factor_set_id
+
+    for invalid_bindings, invalid_views in (
+        ((first_binding,), (VIEW_REF_ID, second_view_id)),
+        ((first_binding, second_binding), (VIEW_REF_ID,)),
+        ((first_binding, second_binding), (VIEW_REF_ID, second_view_id, VIEW_REF_ID)),
+    ):
+        invalid = dict(base)
+        invalid.update(input_bindings=invalid_bindings, selected_view_ref_ids=invalid_views)
+        with pytest.raises(FactorContractError) as error:
+            FactorSetRef.create(**invalid)
+        assert error.value.code in {
+            ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+            ContractErrorCode.INVALID_VALUE,
+        }
+
+
+@pytest.mark.parametrize(
+    ("mutate", "code", "path"),
+    [
+        (lambda value: value["producer"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.producer.id"),
+        (lambda value: value["producer"].pop("version"), ContractErrorCode.MISSING_FIELD, "$.producer.version"),
+        (lambda value: value["producer"].update(id="other_engine"), ContractErrorCode.LINEAGE_VIOLATION, "$.producer.id"),
+        (lambda value: value["producer"].update(version="latest"), ContractErrorCode.INVALID_FORMAT, "$.producer.version"),
+        (lambda value: value.update(code_revision="bad"), ContractErrorCode.INVALID_FORMAT, "$.code_revision"),
+        (lambda value: value["actor"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.actor.kind"),
+        (lambda value: value["actor"].pop("id"), ContractErrorCode.MISSING_FIELD, "$.actor.id"),
+        (lambda value: value["actor"].update(kind="robot"), ContractErrorCode.INVALID_VALUE, "$.actor.kind"),
+        (lambda value: value["actor"].update(id="latest"), ContractErrorCode.INVALID_VALUE, "$.actor.id"),
+        (lambda value: value.pop("correlation_id"), ContractErrorCode.MISSING_FIELD, "$.correlation_id"),
+        (lambda value: value.update(correlation_id="latest"), ContractErrorCode.INVALID_VALUE, "$.correlation_id"),
+        (lambda value: value["causation"].pop("kind"), ContractErrorCode.MISSING_FIELD, "$.causation.kind"),
+        (lambda value: value["causation"].update(kind="run"), ContractErrorCode.INVALID_VALUE, "$.causation.kind"),
+        (lambda value: value["causation"].update(id="rhdfv1:sha256:" + "0" * 64), ContractErrorCode.LINEAGE_VIOLATION, "$.causation.id"),
+        (lambda value: value.pop("output_artifact_ref"), ContractErrorCode.MISSING_FIELD, "$.output_artifact_ref"),
+        (lambda value: value["output_artifact_ref"].update(artifact_id="rhfactoroutputv1:sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.output_artifact_ref.artifact_id"),
+        (_mutate_artifact_schema_binding, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref"),
+        (lambda value: value.update(availability_mode="implicit_fallback"), ContractErrorCode.INVALID_VALUE, "$.availability_mode"),
+        (lambda value: value.pop("computed_at"), ContractErrorCode.MISSING_FIELD, "$.computed_at"),
+        (lambda value: value.update(decision_eligible=True), ContractErrorCode.READINESS_ESCALATION, "$.decision_eligible"),
+        (lambda value: value.update(evidence_scope="real_data"), ContractErrorCode.READINESS_ESCALATION, "$.evidence_scope"),
+        (lambda value: value["upstream_evidence"].update(qualification_evidence_digest="sha256:" + "0" * 64), ContractErrorCode.IDENTITY_MISMATCH, "$.upstream_evidence"),
+    ],
+)
+def test_lineage_artifact_and_readiness_fields_have_independent_typed_negatives(
+    mutate: Callable[[dict[str, Any]], Any],
+    code: ContractErrorCode,
+    path: str,
+) -> None:
+    factor_set = _factor_set()
+    value = factor_set.to_dict()
+    mutate(value)
+    if "factor_set_id" in value:
+        _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(error, code, path)
+
+
+def test_output_schema_content_bytes_cannot_be_swapped_or_forged() -> None:
+    factor_set = _factor_set()
+    fixture = _golden()
+    snapshot, foundation = _snapshot_and_foundation()
+    schema_bytes = canonical_json_bytes(fixture["output_schema"])
+    content_bytes = canonical_json_bytes(fixture["output_content"])
+    with pytest.raises(FactorContractError) as swapped:
+        FactorSetRef.from_dict(
+            factor_set.to_dict(),
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+            output_schema_bytes=content_bytes,
+            output_content_bytes=schema_bytes,
+        )
+    _assert_error(swapped, ContractErrorCode.ARTIFACT_MISMATCH, "$.output_artifact_ref")
+    with pytest.raises(FactorContractError) as noncanonical:
+        FactorSetRef.create(
+            **{
+                **_factor_set_arguments(),
+                "output_schema_bytes": json.dumps(fixture["output_schema"], indent=2).encode(),
+            }
+        )
+    _assert_error(noncanonical, ContractErrorCode.INVALID_FORMAT, "$.output_schema_bytes")
+
+
+def test_unsuccessful_output_quality_or_coverage_cannot_form_a_factor_set() -> None:
+    with pytest.raises(FactorContractError) as failed_quality:
+        _factor_set(
+            output_quality=OutputQuality(
+                "failed",
+                (OutputQualityCheck("finite_values", "failed", "sha256:" + "3" * 64),),
+            )
+        )
+    _assert_error(failed_quality, ContractErrorCode.INVALID_VALUE, "$.output_quality")
+
+    for coverage in (
+        OutputCoverage("incomplete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
+        OutputCoverage("complete", 2, 1, "row", "alpha_005.cn_a", "sha256:" + "4" * 64),
+    ):
+        with pytest.raises(FactorContractError) as incomplete:
+            _factor_set(output_coverage=coverage)
+        _assert_error(incomplete, ContractErrorCode.INVALID_VALUE, "$.output_coverage")
+
+
+def test_external_snapshot_definition_and_view_references_cannot_be_substituted() -> None:
+    factor_set = _factor_set()
+    snapshot, foundation = _snapshot_and_foundation()
+    value = factor_set.to_dict()
+    value["dataset_snapshot_id"] = "rhdsv1:sha256:" + "0" * 64
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as snapshot_error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        snapshot_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.dataset_snapshot_id",
+    )
+
+    value = factor_set.to_dict()
+    value["definition_ids"] = ["rhfactorv1:sha256:" + "0" * 64]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    with pytest.raises(FactorContractError) as definition_error:
+        FactorSetRef.from_dict(
+            value,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(
+        definition_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.definition_ids",
+    )
+
+    arguments = _factor_set_arguments()
+    arguments["selected_view_ref_ids"] = ("rhviewrefv1:sha256:" + "0" * 64,)
+    with pytest.raises(FactorContractError) as view_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(
+        view_error,
+        ContractErrorCode.INPUT_CLOSURE_VIOLATION,
+        "$.selected_view_ref_ids",
+    )
+
+
+@pytest.mark.parametrize(
+    "invalid_definition_id",
+    [
+        {"unexpected": "object"},
+        ["array"],
+        42,
+        True,
+        None,
+    ],
+)
+def test_factor_set_ref_definition_ids_reject_non_string_types(
+    invalid_definition_id: Any,
+) -> None:
+    factor_set = _factor_set()
+    definition = _golden_definition()
+    value = factor_set.to_dict()
+    value["definition_ids"] = [definition.definition_id, invalid_definition_id]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_json(
+            canonical_json_bytes(value),
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+
+    _assert_error(error, ContractErrorCode.TYPE_ERROR, "$.definition_ids[1]")
+
+
+def test_factor_set_ref_definition_ids_still_reject_duplicate_strings() -> None:
+    factor_set = _factor_set()
+    definition = _golden_definition()
+    value = factor_set.to_dict()
+    value["definition_ids"] = [definition.definition_id, definition.definition_id]
+    _reidentify(value, "factor_set_id", "rhfactorsetv1:sha256:")
+    snapshot, foundation = _snapshot_and_foundation()
+
+    with pytest.raises(FactorContractError) as error:
+        FactorSetRef.from_json(
+            canonical_json_bytes(value),
+            definitions=(definition,),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+
+    _assert_error(error, ContractErrorCode.INVALID_VALUE, "$.definition_ids")
+
+
+def test_factor_set_parent_requires_exact_identity_and_correlation() -> None:
+    parent = _factor_set()
+    child_arguments = _factor_set_arguments()
+    child_arguments.update(
+        output_content_bytes=canonical_json_bytes({"rows": [{"value": "0.250"}]}),
+        causation=Causation("factor_set", parent.factor_set_id),
+        parent=parent,
+    )
+    child_arguments["output_artifact_ref"] = OutputArtifactRef.create(
+        schema_digest=_sha256(child_arguments["output_schema_bytes"]),
+        content_digest=_sha256(child_arguments["output_content_bytes"]),
+    )
+    child = FactorSetRef.create(**child_arguments)
+    assert child.causation.id == parent.factor_set_id
+    missing_parent = child.to_dict()
+    snapshot, foundation = _snapshot_and_foundation()
+    with pytest.raises(FactorContractError) as missing_error:
+        FactorSetRef.from_dict(
+            missing_parent,
+            definitions=(_golden_definition(),),
+            dataset_snapshot=snapshot,
+            foundation=foundation,
+        )
+    _assert_error(missing_error, ContractErrorCode.LINEAGE_VIOLATION, "$.causation")
+    wrong_correlation = dict(child_arguments)
+    wrong_correlation["correlation_id"] = "different_run"
+    with pytest.raises(FactorContractError) as correlation_error:
+        FactorSetRef.create(**wrong_correlation)
+    _assert_error(correlation_error, ContractErrorCode.LINEAGE_VIOLATION, "$.correlation_id")
+
+
+def test_legacy_bridge_is_explicit_lossy_and_preserves_all_four_historical_fields() -> None:
+    definition = _golden_definition()
+    legacy = FactorVersion(
+        factor_id="factor:demo-momentum",
+        version="1.0.0",
+        definition_sha256="b" * 64,
+        dataset_schema_version="1.0.0",
+    )
+    binding = LegacyFactorBinding.create(
+        definition=definition,
+        legacy_factor_id=legacy.factor_id,
+        legacy_version=legacy.version,
+        legacy_definition_sha256=legacy.definition_sha256,
+        legacy_dataset_schema_version=legacy.dataset_schema_version,
+        canonical_input_schema_digest=definition.input_schema_digest,
+        correspondence_evidence_digest="sha256:" + "5" * 64,
+    )
+    assert bind_legacy_factor(legacy, definition, binding) is definition
+    assert project_legacy_factor(definition, binding) == legacy
+    assert legacy.version_id == "factor:demo-momentum@1.0.0"
+    assert legacy.definition_sha256 != definition.definition_id.rsplit(":", maxsplit=1)[-1]
+    assert LegacyFactorBinding.from_json(binding.to_json(), definition=definition) == binding
+
+    mismatched = FactorVersion(
+        factor_id="factor:different",
+        version=legacy.version,
+        definition_sha256=legacy.definition_sha256,
+        dataset_schema_version=legacy.dataset_schema_version,
+    )
+    with pytest.raises(FactorContractError) as mismatch_error:
+        bind_legacy_factor(mismatched, definition, binding)
+    _assert_error(mismatch_error, ContractErrorCode.LEGACY_BINDING_MISMATCH, "$.binding")
+
+
+def test_bare_legacy_factor_or_id_cannot_enter_factor_set_contract() -> None:
+    legacy = FactorVersion("factor:demo-momentum", "1.0.0", "b" * 64, "1.0.0")
+    arguments = _factor_set_arguments()
+    arguments["definitions"] = (legacy,)
+    with pytest.raises(FactorContractError) as legacy_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(legacy_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")
+    arguments["definitions"] = (legacy.version_id,)
+    with pytest.raises(FactorContractError) as id_error:
+        FactorSetRef.create(**arguments)
+    _assert_error(id_error, ContractErrorCode.TYPE_ERROR, "$.definitions[0]")
@@ -119,6 +119,24 @@ def test_governed_slice_is_reproducible_and_creates_only_paper_intent() -> None:
    assert result.order_intent.environment == "paper"
    assert result.order_intent.risk_decision_id == result.risk_decision.decision_id
    assert result.order_intent.portfolio_target_id == result.portfolio_target.target_id
+    assert result.factor_version.version_id == "factor:demo-momentum@1.0.0"
+    assert result.factor_version.definition_sha256 == "b" * 64
+    assert result.strategy_version.version_id == "strategy:demo-top2@1.0.0"
+    assert result.backtest_run.run_id == (
+        "backtest-run:e74403571f6a73c98b380220748957a422518c0bed883fabc4b93ebe13f05a37"
+    )
+    assert result.backtest_run.config_hash == (
+        "40a3c804a2dc940161a626d1a5d25817c13463e685005e37fc48c41d1e20b87b"
+    )
+    assert result.portfolio_target.target_id == (
+        "portfolio-target:ab2d398489aa9a292ee155a1098e9340beac0a924874cdaeb5b7b4d379ac9ce8"
+    )
+    assert result.risk_decision.decision_id == (
+        "risk-decision:95924926bd327e44beeb15a63f47d14c5e77b97533fc3227fba2eda68e9b423d"
+    )
+    assert result.order_intent.intent_id == (
+        "order-intent:73c349086c2c05b68424ace9286896eb21507b9080da5ff9b96b58c88d8f6ac1"
+    )

    repeated = run_governed_factor_slice(
        factor_scores=_scores(),
Author	SHA1	Message	Date
ao gong	6a7de4b1f3	fix: close final backtest contract boundaries CI / lite (pull_request) Successful in 10s Details	2026-09-01 12:34:12 +08:00
ao gong	36d444924a	fix: harden backtest contract boundaries	2026-09-01 12:17:16 +08:00
ao gong	e1e10ce293	feat: close backtest evidence contract	2026-09-01 12:04:48 +08:00
ageorge156	598c2b92a2	fix: type-check factor set definition ids (#16 ) CI / lite (push) Successful in 8s Details	2026-09-01 11:10:18 +08:00
ageorge156	62ed09842d	fix: enforce canonical factor contract json (#15 ) CI / lite (push) Successful in 7s Details	2026-09-01 08:36:41 +08:00
ageorge156	e782e223f7	feat(governance): add risk-gated paper research slice (#14 ) CI / lite (push) Successful in 8s Details	2026-08-30 16:14:34 +08:00