feat: add strict vNext result/store contract

src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py

@@ -0,0 +1,226 @@
+"""Strict side-by-side vNext result/store contract foundation.
+
+Pure validation and deterministic wire/key generation only; this module performs
+no store, cache, publish, runner, or filesystem I/O and never reads legacy data.
+
+Identity subset (and only this subset) feeds ``identity_digest`` and ``key``:
+namespace, schema_version, domain, canonical_profile, execution_timing,
+result_identity_version, strategy_id, run_id, param_set_id, param_version,
+source_revision, and canonical params. ``persist_mode`` and ``computed_at`` are
+wire metadata and are intentionally excluded from identity.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import re
+from datetime import datetime
+from dataclasses import dataclass
+from enum import Enum
+from types import MappingProxyType
+from typing import Any, Mapping
+
+from quant_platform_kit.strategy_lifecycle.capabilities import ExecutionTiming, PersistMode
+
+VNEXT_NAMESPACE = "qpk-vnext/result/v1"
+_SCHEMA_VERSION = "qpk.vnext.result.v1"
+_MAX_SEGMENT = 100
+MAX_SAFE_JSON_INTEGER = 2**53 - 1
+_RFC3339_UTC = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z$")
+_WIRE_FIELDS = frozenset({
+    "namespace", "schema_version", "domain", "canonical_profile", "execution_timing",
+    "result_identity_version", "persist_mode", "strategy_id", "run_id", "param_set_id",
+    "param_version", "computed_at", "source_revision", "params", "identity_digest",
+})
+
+
+class VNextContractError(ValueError):
+    """Sanitized validation error for all vNext contract failures."""
+
+    def __init__(self) -> None:
+        super().__init__("invalid vNext result/store contract")
+
+
+def _segment(value: Any) -> str:
+    if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT:
+        raise VNextContractError()
+    if value in {".", ".."} or "/" in value or "\\" in value or value.startswith("/"):
+        raise VNextContractError()
+    if value != value.strip() or value != value.strip("-._"):
+        raise VNextContractError()
+    if any(char not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-" for char in value):
+        raise VNextContractError()
+    return value
+
+
+def _param_value(value: Any) -> Any:
+    if value is None:
+        raise VNextContractError()
+    if isinstance(value, (str, bool)):
+        return value
+    if isinstance(value, int):
+        if abs(value) > MAX_SAFE_JSON_INTEGER:
+            raise VNextContractError()
+        return value
+    if isinstance(value, float) and math.isfinite(value):
+        return value
+    if isinstance(value, tuple):
+        return tuple(_param_value(item) for item in value)
+    raise VNextContractError()
+
+
+def _params(value: Any) -> dict[str, Any]:
+    if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value):
+        raise VNextContractError()
+    return {key: _param_value(value[key]) for key in sorted(value)}
+
+
+def _wire_params(value: Any) -> Mapping[str, Any]:
+    if not isinstance(value, Mapping) or any(not isinstance(key, str) or not key for key in value):
+        raise VNextContractError()
+    def thaw(item: Any) -> Any:
+        if isinstance(item, list):
+            return tuple(thaw(child) for child in item)
+        if isinstance(item, tuple):
+            return tuple(thaw(child) for child in item)
+        if isinstance(item, (str, bool)) or (isinstance(item, float) and math.isfinite(item)):
+            return item
+        if isinstance(item, int) and abs(item) <= MAX_SAFE_JSON_INTEGER:
+            return item
+        raise VNextContractError()
+    return {key: thaw(value[key]) for key in sorted(value)}
+
+
+def _text(value: Any) -> str:
+    if not isinstance(value, str) or not value or len(value) > _MAX_SEGMENT:
+        raise VNextContractError()
+    if any(ord(char) < 32 or char in "/\\" for char in value):
+        raise VNextContractError()
+    return value
+
+
+def _canonical_profile(value: Any) -> str:
+    """Validate an uppercase canonical profile; policy allowlists belong upstream."""
+    value = _segment(value)
+    if value != value.upper():
+        raise VNextContractError()
+    return value
+
+
+def _computed_at(value: Any) -> str:
+    """Validate canonical UTC RFC3339 audit time without timezone conversion."""
+    value = _text(value)
+    if not _RFC3339_UTC.fullmatch(value):
+        raise VNextContractError()
+    try:
+        datetime.fromisoformat(value.removesuffix("Z"))
+    except ValueError as exc:
+        raise VNextContractError() from exc
+    return value
+
+
+def _canonical_json(value: Mapping[str, Any]) -> bytes:
+    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+
+
+def _digest(value: Mapping[str, Any]) -> str:
+    return hashlib.sha256(_canonical_json(value)).hexdigest()
+
+
+def _thaw_params(value: Mapping[str, Any]) -> dict[str, Any]:
+    def thaw(item: Any) -> Any:
+        if isinstance(item, tuple):
+            return [thaw(child) for child in item]
+        return item
+    return {key: thaw(value[key]) for key in sorted(value)}
+
+
+@dataclass(frozen=True)
+class VNextResultContract:
+    domain: str
+    canonical_profile: str
+    execution_timing: ExecutionTiming
+    result_identity_version: int
+    persist_mode: PersistMode
+    strategy_id: str
+    run_id: str
+    param_set_id: str
+    param_version: int
+    computed_at: str
+    source_revision: str
+    params: Mapping[str, Any]
+
+    def __post_init__(self) -> None:
+        _segment(self.domain)
+        _canonical_profile(self.canonical_profile)
+        try:
+            object.__setattr__(self, "execution_timing", ExecutionTiming(self.execution_timing))
+            object.__setattr__(self, "persist_mode", PersistMode(self.persist_mode))
+        except Exception as exc:
+            raise VNextContractError() from exc
+        for value in (self.strategy_id, self.run_id, self.param_set_id, self.source_revision):
+            _segment(value)
+        if not isinstance(self.result_identity_version, int) or isinstance(self.result_identity_version, bool) or self.result_identity_version < 1:
+            raise VNextContractError()
+        if not isinstance(self.param_version, int) or isinstance(self.param_version, bool) or self.param_version < 1:
+            raise VNextContractError()
+        _computed_at(self.computed_at)
+        object.__setattr__(self, "params", MappingProxyType(_params(self.params)))
+
+    def _identity_payload(self) -> dict[str, Any]:
+        return {
+            "namespace": VNEXT_NAMESPACE,
+            "schema_version": _SCHEMA_VERSION,
+            "domain": self.domain,
+            "canonical_profile": self.canonical_profile,
+            "execution_timing": self.execution_timing.value,
+            "result_identity_version": self.result_identity_version,
+            "strategy_id": self.strategy_id,
+            "run_id": self.run_id,
+            "param_set_id": self.param_set_id,
+            "param_version": self.param_version,
+            "source_revision": self.source_revision,
+            "params": _thaw_params(self.params),
+        }
+
+    def _payload(self) -> dict[str, Any]:
+        return {
+            **self._identity_payload(),
+            "persist_mode": self.persist_mode.value,
+            "computed_at": self.computed_at,
+        }
+
+    def to_wire(self) -> dict[str, Any]:
+        payload = self._payload()
+        return {**payload, "identity_digest": _digest(self._identity_payload())}
+
+    @property
+    def key(self) -> str:
+        return f"{VNEXT_NAMESPACE}/{self.domain}/{self.canonical_profile}/{self.strategy_id}/{self.run_id}/{self.execution_timing.value}/i{self.result_identity_version}/p{self.param_version}/{_digest(self._identity_payload())}.json"
+
+
+def decode_vnext_wire(data: Mapping[str, Any]) -> VNextResultContract:
+    try:
+        if not isinstance(data, Mapping) or set(data) != _WIRE_FIELDS:
+            raise VNextContractError()
+        if data["namespace"] != VNEXT_NAMESPACE or data["schema_version"] != _SCHEMA_VERSION:
+            raise VNextContractError()
+        wire_params = _wire_params(data["params"])
+        contract = VNextResultContract(
+            domain=data["domain"], canonical_profile=data["canonical_profile"],
+            execution_timing=data["execution_timing"], result_identity_version=data["result_identity_version"],
+            persist_mode=data["persist_mode"], strategy_id=data["strategy_id"], run_id=data["run_id"],
+            param_set_id=data["param_set_id"], param_version=data["param_version"],
+            computed_at=data["computed_at"], source_revision=data["source_revision"], params=wire_params,
+        )
+        if not isinstance(data["identity_digest"], str) or _digest(contract._identity_payload()) != data["identity_digest"]:
+            raise VNextContractError()
+        if contract.to_wire() != dict(data):
+            raise VNextContractError()
+        return contract
+    except VNextContractError:
+        raise
+    except Exception as exc:
+        raise VNextContractError() from exc

tests/test_vnext_result_store_contract.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from dataclasses import replace
+
+from quant_platform_kit.strategy_lifecycle.capabilities import ExecutionTiming, PersistMode
+from quant_platform_kit.strategy_lifecycle.vnext_result_store_contract import (
+    MAX_SAFE_JSON_INTEGER,
+    VNEXT_NAMESPACE,
+    VNextContractError,
+    VNextResultContract,
+    decode_vnext_wire,
+)
+
+
+def _contract(**overrides):
+    values = {
+        "domain": "us_equity",
+        "canonical_profile": "SOXL",
+        "execution_timing": ExecutionTiming.NEXT_OPEN,
+        "result_identity_version": 1,
+        "persist_mode": PersistMode.DURABLE,
+        "strategy_id": "soxl_trend",
+        "run_id": "run-001",
+        "param_set_id": "baseline",
+        "param_version": 1,
+        "computed_at": "2026-07-15T00:00:00Z",
+        "source_revision": "abc123",
+        "params": {"lookback": 20, "threshold": 0.5},
+    }
+    values.update(overrides)
+    return VNextResultContract(**values)
+
+
+class VNextResultStoreContractTests(unittest.TestCase):
+    def test_round_trip_and_namespace_isolation(self) -> None:
+        contract = _contract()
+        wire = contract.to_wire()
+        self.assertEqual(wire["namespace"], VNEXT_NAMESPACE)
+        self.assertEqual(decode_vnext_wire(wire), contract)
+        self.assertTrue(contract.key.startswith("qpk-vnext/result/v1/"))
+        upro = _contract(canonical_profile="UPRO")
+        self.assertEqual(decode_vnext_wire(upro.to_wire()), upro)
+
+    def test_order_independence_and_identity_changes(self) -> None:
+        first = _contract(params={"b": 2, "a": 1})
+        second = _contract(params={"a": 1, "b": 2})
+        self.assertEqual(first.to_wire(), second.to_wire())
+        self.assertEqual(first.key, second.key)
+        self.assertNotEqual(first.key, replace(first, run_id="run-002").key)
+
+    def test_wire_metadata_does_not_change_identity(self) -> None:
+        first = _contract(params={"nested": (1, (2, 3))})
+        second = replace(first, persist_mode=PersistMode.EPHEMERAL, computed_at="2026-07-16T00:00:00Z")
+        self.assertEqual(first.key, second.key)
+        self.assertEqual(first.to_wire()["identity_digest"], second.to_wire()["identity_digest"])
+        self.assertNotEqual(first.to_wire()["persist_mode"], second.to_wire()["persist_mode"])
+        self.assertNotEqual(first.to_wire()["computed_at"], second.to_wire()["computed_at"])
+
+    def test_identity_fields_change_digest_and_key(self) -> None:
+        base = _contract()
+        variants = (
+            replace(base, execution_timing=ExecutionTiming.NEXT_CLOSE),
+            replace(base, canonical_profile="TQQQ"),
+            replace(base, source_revision="def456"),
+            replace(base, param_set_id="candidate"),
+            replace(base, result_identity_version=2),
+        )
+        for variant in variants:
+            self.assertNotEqual(base.key, variant.key)
+            self.assertNotEqual(base.to_wire()["identity_digest"], variant.to_wire()["identity_digest"])
+
+    def test_required_none_legacy_any_alias_and_unknown_values_reject(self) -> None:
+        for field in ("execution_timing", "canonical_profile", "strategy_id", "run_id", "source_revision"):
+            with self.assertRaises(VNextContractError):
+                _contract(**{field: None})
+        with self.assertRaises(VNextContractError):
+            _contract(execution_timing=None)
+        with self.assertRaises(VNextContractError):
+            _contract(canonical_profile="soxl_soxx_trend_income")
+        for profile in ("", "soxl", "A/B", "..", "X" * 101):
+            with self.assertRaises(VNextContractError):
+                _contract(canonical_profile=profile)
+        with self.assertRaises(VNextContractError):
+            _contract(execution_timing="ANY")
+
+    def test_persist_modes_are_explicit_without_io(self) -> None:
+        self.assertEqual(_contract(persist_mode=PersistMode.EPHEMERAL).to_wire()["persist_mode"], "ephemeral")
+        self.assertEqual(_contract(persist_mode=PersistMode.DURABLE).to_wire()["persist_mode"], "durable")
+
+    def test_tuple_and_nested_tuple_round_trip_freezes_params(self) -> None:
+        contract = _contract(params={"levels": (1, ("a", True))})
+        self.assertEqual(decode_vnext_wire(contract.to_wire()), contract)
+        self.assertIsInstance(contract.params["levels"], tuple)
+        with self.assertRaises(AttributeError):
+            contract.params["levels"].append(2)
+
+    def test_direct_mutable_values_reject_but_wire_lists_are_frozen(self) -> None:
+        with self.assertRaises(VNextContractError):
+            _contract(params={"values": [1, 2]})
+        decoded = decode_vnext_wire(_contract(params={"values": (1, 2)}).to_wire())
+        self.assertEqual(decoded.params["values"], (1, 2))
+
+    def test_json_safe_integer_boundary_and_nested_rejection(self) -> None:
+        boundary = _contract(params={"low": -MAX_SAFE_JSON_INTEGER, "high": MAX_SAFE_JSON_INTEGER})
+        self.assertEqual(decode_vnext_wire(boundary.to_wire()), boundary)
+        self.assertEqual(_contract(params={"flag": True}).params["flag"], True)
+        for value in (MAX_SAFE_JSON_INTEGER + 1, -(MAX_SAFE_JSON_INTEGER + 1)):
+            with self.assertRaises(VNextContractError):
+                _contract(params={"value": value})
+            wire = _contract(params={"value": (1,)}).to_wire()
+            wire["params"] = {"nested": [value]}
+            with self.assertRaises(VNextContractError):
+                decode_vnext_wire(wire)
+
+    def test_computed_at_requires_canonical_utc_rfc3339(self) -> None:
+        for timestamp in (
+            "", " 2026-07-15T00:00:00Z", "2026-07-15T00:00:00+00:00",
+            "2026-02-30T00:00:00Z", "2026-07-15", "not-a-time",
+        ):
+            with self.assertRaises(VNextContractError):
+                _contract(computed_at=timestamp)
+        for timestamp in ("2026-07-15T00:00:00Z", "2026-07-15T00:00:00.123456Z"):
+            contract = _contract(computed_at=timestamp)
+            self.assertEqual(decode_vnext_wire(contract.to_wire()), contract)
+
+    def test_adversarial_wire_values_reject(self) -> None:
+        wire = _contract().to_wire()
+        for mutation in (
+            {"namespace": "strategy_lifecycle"},
+            {"execution_timing": None},
+            {"canonical_profile": "SOXL_SOXX_TREND_INCOME"},
+            {"result_identity_version": True},
+            {"params": {"bad": []}},
+            {"unknown": "field"},
+        ):
+            candidate = copy.deepcopy(wire)
+            candidate.update(mutation)
+            with self.assertRaises(VNextContractError):
+                decode_vnext_wire(candidate)
+        missing = copy.deepcopy(wire)
+        del missing["run_id"]
+        with self.assertRaises(VNextContractError):
+            decode_vnext_wire(missing)
+
+    def test_unsafe_and_overlong_identity_segments_reject(self) -> None:
+        for field in ("domain", "strategy_id", "run_id", "param_set_id", "source_revision"):
+            with self.assertRaises(VNextContractError):
+                _contract(**{field: "../unsafe"})
+            with self.assertRaises(VNextContractError):
+                _contract(**{field: "x" * 101})
+        with self.assertRaises(VNextContractError):
+            _contract(param_version=True)
+
+
+if __name__ == "__main__":
+    unittest.main()