fix: close final backtest contract boundaries

src/quant_engine/artifact.py

@@ -32,6 +32,7 @@ from quant_engine.risk import CovarianceSnapshot, labeled_component_risk
 
 RESEARCH_ARTIFACT_SCHEMA_VERSION = "1.1.0"
 BACKTEST_EVIDENCE_SCHEMA_VERSION = "1.0.0"
+_MAX_SAFE_INTEGER = (1 << 53) - 1
 
 RISK_COLUMNS = [
     "run_id",
@@ -136,7 +137,7 @@ class ResearchRunArtifact:
         payload = {
             "schema_version": self.schema_version,
             "tables": {
-                name: _frame_records(frame)
+                name: _frame_records(frame, name)
                 for name, frame in self._internal_table_frames().items()
             },
         }
@@ -275,8 +276,8 @@ class BacktestEvidenceManifest:
         cls,
         value: Any,
         *,
-        backtest_run_ref: BacktestRunRef,
         artifact: ResearchRunArtifact,
+        backtest_run_ref: BacktestRunRef | None = None,
     ) -> Self:
         if type(value) is not dict:
             _manifest_fail(BacktestContractErrorCode.TYPE_ERROR, "$", "must be an object")
@@ -350,12 +351,59 @@ class BacktestEvidenceManifest:
                 "$.qualification",
                 "unsupported evidence qualification",
             ) from error
-        rebuilt = build_backtest_evidence_manifest(
-            backtest_run_ref,
-            artifact,
-            artifact_available_at=value["artifact_available_at"],
-            qualification=qualification,
-        )
+        raw_run_reference = value["run_reference"]
+        if type(raw_run_reference) is not dict:
+            _manifest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                "$.run_reference",
+                "must be an object",
+            )
+        if any(type(key) is not str for key in raw_run_reference):
+            _manifest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                "$.run_reference",
+                "object keys must be strings",
+            )
+        reference_fields = {"kind", "value"}
+        if set(raw_run_reference) != reference_fields:
+            _manifest_fail(
+                BacktestContractErrorCode.INVALID_VALUE,
+                "$.run_reference",
+                "must contain exactly kind and value",
+            )
+        reference_kind = raw_run_reference["kind"]
+        if reference_kind == "legacy_backtest_run":
+            if qualification is not EvidenceQualification.LEGACY_EXPLORATORY:
+                _manifest_fail(
+                    BacktestContractErrorCode.READINESS_ESCALATION,
+                    "$.qualification",
+                    "legacy evidence cannot be promoted",
+                )
+            legacy_run = _legacy_run_from_dict(raw_run_reference["value"])
+            rebuilt = build_legacy_backtest_evidence_manifest(
+                legacy_run,
+                artifact,
+                artifact_available_at=value["artifact_available_at"],
+            )
+        elif reference_kind == "backtest_run_ref":
+            if backtest_run_ref is None:
+                _manifest_fail(
+                    BacktestContractErrorCode.TYPE_ERROR,
+                    "$.backtest_run_ref",
+                    "complete BacktestRunRef is required",
+                )
+            rebuilt = build_backtest_evidence_manifest(
+                backtest_run_ref,
+                artifact,
+                artifact_available_at=value["artifact_available_at"],
+                qualification=qualification,
+            )
+        else:
+            _manifest_fail(
+                BacktestContractErrorCode.INVALID_VALUE,
+                "$.run_reference.kind",
+                "unsupported run-reference kind",
+            )
         if value != rebuilt.to_dict():
             _manifest_fail(
                 BacktestContractErrorCode.IDENTITY_MISMATCH,
@@ -384,13 +432,39 @@ def _aware_timestamp(value: str | pd.Timestamp, name: str) -> pd.Timestamp:
     return timestamp
 
 
-def _json_value(value: object) -> object:
-    if value is None or isinstance(value, str | bool | int):
+def _json_value(value: object, path: str = "$") -> object:
+    if value is None or type(value) is bool:
+        return value
+    if type(value) is int:
+        if abs(value) > _MAX_SAFE_INTEGER:
+            _manifest_fail(
+                BacktestContractErrorCode.INVALID_VALUE,
+                path,
+                "integer exceeds the canonical safe range",
+            )
+        return value
+    if type(value) is str:
+        try:
+            value.encode("utf-8")
+        except UnicodeEncodeError as error:
+            raise BacktestContractError(
+                BacktestContractErrorCode.INVALID_FORMAT,
+                path,
+                "text must be valid UTF-8",
+            ) from error
+        return value
+    if type(value) is float:
+        if math.isnan(value):
+            return {"non_finite_float": "nan"}
+        if math.isinf(value):
+            return {
+                "non_finite_float": (
+                    "positive_infinity" if value > 0 else "negative_infinity"
+                )
+            }
         return value
-    if isinstance(value, float):
-        return value if math.isfinite(value) else None
     if isinstance(value, np.generic):
-        return _json_value(value.item())
+        return _json_value(value.item(), path)
     if isinstance(value, pd.Timestamp):
         return value.isoformat()
     if isinstance(value, datetime):
@@ -398,13 +472,34 @@ def _json_value(value: object) -> object:
     if isinstance(value, date):
         return value.isoformat()
     if isinstance(value, Mapping):
-        return {
-            str(key): _json_value(item)
-            for key, item in sorted(value.items(), key=lambda pair: str(pair[0]))
-        }
+        normalized: dict[str, object] = {}
+        for key, item in value.items():
+            if type(key) is not str:
+                _manifest_fail(
+                    BacktestContractErrorCode.TYPE_ERROR,
+                    f"{path}.keys",
+                    "object keys must be strings",
+                )
+            try:
+                key.encode("utf-8")
+            except UnicodeEncodeError as error:
+                raise BacktestContractError(
+                    BacktestContractErrorCode.INVALID_FORMAT,
+                    f"{path}.keys",
+                    "object keys must be valid UTF-8",
+                ) from error
+            normalized[key] = _json_value(item, f"{path}.{key}")
+        return dict(sorted(normalized.items()))
     if isinstance(value, list | tuple):
-        return [_json_value(item) for item in value]
-    raise TypeError(f"value of type {type(value).__name__} is not JSON serializable")
+        return [
+            _json_value(item, f"{path}[{index}]")
+            for index, item in enumerate(value)
+        ]
+    _manifest_fail(
+        BacktestContractErrorCode.TYPE_ERROR,
+        path,
+        f"unsupported canonical value type: {type(value).__name__}",
+    )
 
 
 def _canonical_mapping_json(values: Mapping[str, object]) -> str:
@@ -418,24 +513,47 @@ def _canonical_mapping_json(values: Mapping[str, object]) -> str:
     )
 
 
-def _frame_records(frame: pd.DataFrame) -> list[dict[str, object]]:
+def _frame_records(frame: pd.DataFrame, table_name: str) -> list[dict[str, object]]:
+    try:
+        raw_records = frame.to_dict(orient="records")
+    except (TypeError, ValueError) as error:
+        raise BacktestContractError(
+            BacktestContractErrorCode.TYPE_ERROR,
+            f"$.tables.{table_name}",
+            "table records cannot be serialized",
+        ) from error
     return [
-        {str(key): _json_value(value) for key, value in row.items()}
-        for row in frame.to_dict(orient="records")
+        {
+            str(key): _evidence_json_value(
+                value,
+                f"$.tables.{table_name}.rows[{row_index}].{key}",
+            )
+            for key, value in row.items()
+        }
+        for row_index, row in enumerate(raw_records)
     ]
 
 
 def _evidence_json_value(value: object, path: str) -> object:
-    if value is None or type(value) in {str, bool, int}:
-        if type(value) is str:
-            try:
-                value.encode("utf-8")
-            except UnicodeEncodeError as error:
-                raise BacktestContractError(
-                    BacktestContractErrorCode.INVALID_FORMAT,
-                    path,
-                    "text must be valid UTF-8",
-                ) from error
+    if value is None or type(value) is bool:
+        return value
+    if type(value) is int:
+        if abs(value) > _MAX_SAFE_INTEGER:
+            _manifest_fail(
+                BacktestContractErrorCode.INVALID_VALUE,
+                path,
+                "integer exceeds the canonical safe range",
+            )
+        return value
+    if type(value) is str:
+        try:
+            value.encode("utf-8")
+        except UnicodeEncodeError as error:
+            raise BacktestContractError(
+                BacktestContractErrorCode.INVALID_FORMAT,
+                path,
+                "text must be valid UTF-8",
+            ) from error
         return value
     if type(value) is float:
         if math.isnan(value):
@@ -453,11 +571,22 @@ def _evidence_json_value(value: object, path: str) -> object:
             if type(key) is not str:
                 _manifest_fail(
                     BacktestContractErrorCode.TYPE_ERROR,
-                    path,
+                    f"{path}.keys",
                     "object cell keys must be strings",
                 )
+            try:
+                key.encode("utf-8")
+            except UnicodeEncodeError as error:
+                raise BacktestContractError(
+                    BacktestContractErrorCode.INVALID_FORMAT,
+                    f"{path}.keys",
+                    "object cell keys must be valid UTF-8",
+                ) from error
             normalized[key] = _evidence_json_value(item, f"{path}.{key}")
-        return dict(sorted(normalized.items()))
+        return {
+            "canonical_value_type": "object",
+            "value": dict(sorted(normalized.items())),
+        }
     if isinstance(value, list | tuple):
         return [
             _evidence_json_value(item, f"{path}[{index}]")
@@ -476,6 +605,7 @@ def _evidence_cell_kind(value: object, path: str) -> str:
     if type(value) is bool:
         return "boolean"
     if type(value) is int:
+        _evidence_json_value(value, path)
         return "integer"
     if type(value) is float:
         return "number"
@@ -639,6 +769,69 @@ def _legacy_run_dict(run: BacktestRun) -> dict[str, object]:
     }
 
 
+def _legacy_run_from_dict(value: object) -> BacktestRun:
+    path = "$.run_reference.value"
+    if type(value) is not dict:
+        _manifest_fail(BacktestContractErrorCode.TYPE_ERROR, path, "must be an object")
+    if any(type(key) is not str for key in value):
+        _manifest_fail(
+            BacktestContractErrorCode.TYPE_ERROR,
+            path,
+            "object keys must be strings",
+        )
+    fields = {
+        "run_id",
+        "dataset_snapshot_id",
+        "factor_version_id",
+        "strategy_version_id",
+        "code_revision",
+        "config_hash",
+        "created_at",
+    }
+    if set(value) != fields:
+        _manifest_fail(
+            BacktestContractErrorCode.INVALID_VALUE,
+            path,
+            "legacy run reference has incomplete or unknown fields",
+        )
+
+    def required_text(field: str) -> str:
+        item = value[field]
+        item_path = f"{path}.{field}"
+        if type(item) is not str:
+            _manifest_fail(
+                BacktestContractErrorCode.TYPE_ERROR,
+                item_path,
+                "must be a string",
+            )
+        try:
+            item.encode("utf-8")
+        except UnicodeEncodeError as error:
+            raise BacktestContractError(
+                BacktestContractErrorCode.INVALID_FORMAT,
+                item_path,
+                "must be valid UTF-8 text",
+            ) from error
+        if not item or item != item.strip():
+            _manifest_fail(
+                BacktestContractErrorCode.INVALID_FORMAT,
+                item_path,
+                "must be non-empty canonical text",
+            )
+        return item
+
+    _, created_at = _manifest_instant(value["created_at"], f"{path}.created_at")
+    return BacktestRun(
+        run_id=required_text("run_id"),
+        dataset_snapshot_id=required_text("dataset_snapshot_id"),
+        factor_version_id=required_text("factor_version_id"),
+        strategy_version_id=required_text("strategy_version_id"),
+        code_revision=required_text("code_revision"),
+        config_hash=required_text("config_hash"),
+        created_at=created_at,
+    )
+
+
 def _artifact_frames(artifact: ResearchRunArtifact) -> dict[str, pd.DataFrame]:
     if not isinstance(artifact, ResearchRunArtifact):
         _manifest_fail(

src/quant_engine/governed_pipeline.py

@@ -40,13 +40,17 @@ _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]*)(?:-[0-9a-z.-]+)?$"
+    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",
@@ -57,6 +61,19 @@ _FORBIDDEN_ID_TOKENS = frozenset(
         "postgres",
         "mysql",
         "clickhouse",
+        "table",
+        "sql",
+        "tushare",
+        "wind",
+        "bloomberg",
+        "qtdb",
+        "edb",
+        "db",
+        "database",
+        "schema",
+        "bucket",
+        "path",
+        "endpoint",
     }
 )
 
@@ -528,6 +545,16 @@ class BacktestRunRef:
             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,

tests/test_backtest_contracts.py

@@ -65,7 +65,13 @@ def _sha256(value: bytes) -> str:
     return f"sha256:{hashlib.sha256(value).hexdigest()}"
 
 
-def _accepted_authorities() -> tuple[
+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,
@@ -115,10 +121,10 @@ def _accepted_authorities() -> tuple[
         output_schema_bytes=output_schema_bytes,
         output_content_bytes=output_content_bytes,
         output_artifact_ref=artifact_ref,
-        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",
+        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"),
@@ -311,6 +317,48 @@ def test_backtest_run_ref_rejects_unclosed_upstream_and_unsafe_scalars() -> None
     _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")
@@ -355,6 +403,33 @@ def test_run_ref_rejects_aliases_locators_unsafe_integers_and_invalid_text() ->
     _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(
@@ -536,6 +611,21 @@ def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding()
         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:
@@ -550,6 +640,63 @@ def test_manifest_binds_supported_schema_and_has_collision_free_cell_encoding()
         "$.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()
@@ -573,6 +720,10 @@ def test_legacy_bridge_is_explicit_and_cannot_be_contract_qualified() -> None:
     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,