From 7e8624d07df00d9385293a50cf5dfca36f310782 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 02:10:57 +0800 Subject: [PATCH 1/4] feat: add strict vnext result store contract Co-Authored-By: Codex --- .../vnext_result_store_contract.py | 165 ++++++++++++++++++ tests/test_vnext_result_store_contract.py | 95 ++++++++++ 2 files changed, 260 insertions(+) create mode 100644 src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py create mode 100644 tests/test_vnext_result_store_contract.py diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py new file mode 100644 index 0000000..307a683 --- /dev/null +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py @@ -0,0 +1,165 @@ +"""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. +""" + +from __future__ import annotations + +import hashlib +import json +import math +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 +_PROFILES = frozenset({"SOXL", "TQQQ"}) +_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, int)): + return value + if isinstance(value, float) and math.isfinite(value): + return value + if isinstance(value, tuple): + return [_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 _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_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() + + +@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) + if self.canonical_profile not in _PROFILES: + raise VNextContractError() + 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() + _text(self.computed_at) + object.__setattr__(self, "params", MappingProxyType(_params(self.params))) + + def _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, + "persist_mode": self.persist_mode.value, + "strategy_id": self.strategy_id, + "run_id": self.run_id, + "param_set_id": self.param_set_id, + "param_version": self.param_version, + "computed_at": self.computed_at, + "source_revision": self.source_revision, + "params": dict(self.params), + } + + def to_wire(self) -> dict[str, Any]: + payload = self._payload() + return {**payload, "identity_digest": _digest(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._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() + payload = {key: data[key] for key in _WIRE_FIELDS if key != "identity_digest"} + if not isinstance(data["identity_digest"], str) or _digest(payload) != data["identity_digest"]: + raise VNextContractError() + 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=data["params"], + ) + if contract.to_wire() != dict(data): + raise VNextContractError() + return contract + except VNextContractError: + raise + except Exception as exc: + raise VNextContractError() from exc diff --git a/tests/test_vnext_result_store_contract.py b/tests/test_vnext_result_store_contract.py new file mode 100644 index 0000000..322ef96 --- /dev/null +++ b/tests/test_vnext_result_store_contract.py @@ -0,0 +1,95 @@ +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 ( + 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/")) + + 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_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") + 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_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() From e63ead352f16bc34b2bee2c98d2ef9f2a39dffbe Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 02:20:09 +0800 Subject: [PATCH 2/4] fix: separate vnext identity from wire metadata Co-Authored-By: Codex --- .../vnext_result_store_contract.py | 55 +++++++++++++++---- tests/test_vnext_result_store_contract.py | 34 ++++++++++++ 2 files changed, 78 insertions(+), 11 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py index 307a683..f10e803 100644 --- a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py @@ -2,6 +2,12 @@ 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 @@ -54,7 +60,7 @@ def _param_value(value: Any) -> Any: if isinstance(value, float) and math.isfinite(value): return value if isinstance(value, tuple): - return [_param_value(item) for item in value] + return tuple(_param_value(item) for item in value) raise VNextContractError() @@ -64,6 +70,20 @@ def _params(value: Any) -> dict[str, Any]: 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, int)) or (isinstance(item, float) and math.isfinite(item)): + 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() @@ -80,6 +100,14 @@ 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 @@ -113,7 +141,7 @@ def __post_init__(self) -> None: _text(self.computed_at) object.__setattr__(self, "params", MappingProxyType(_params(self.params))) - def _payload(self) -> dict[str, Any]: + def _identity_payload(self) -> dict[str, Any]: return { "namespace": VNEXT_NAMESPACE, "schema_version": _SCHEMA_VERSION, @@ -121,23 +149,28 @@ def _payload(self) -> dict[str, Any]: "canonical_profile": self.canonical_profile, "execution_timing": self.execution_timing.value, "result_identity_version": self.result_identity_version, - "persist_mode": self.persist_mode.value, "strategy_id": self.strategy_id, "run_id": self.run_id, "param_set_id": self.param_set_id, "param_version": self.param_version, - "computed_at": self.computed_at, "source_revision": self.source_revision, - "params": dict(self.params), + "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(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._payload())}.json" + 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: @@ -146,16 +179,16 @@ def decode_vnext_wire(data: Mapping[str, Any]) -> VNextResultContract: raise VNextContractError() if data["namespace"] != VNEXT_NAMESPACE or data["schema_version"] != _SCHEMA_VERSION: raise VNextContractError() - payload = {key: data[key] for key in _WIRE_FIELDS if key != "identity_digest"} - if not isinstance(data["identity_digest"], str) or _digest(payload) != data["identity_digest"]: - 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=data["params"], + 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 diff --git a/tests/test_vnext_result_store_contract.py b/tests/test_vnext_result_store_contract.py index 322ef96..1b289d4 100644 --- a/tests/test_vnext_result_store_contract.py +++ b/tests/test_vnext_result_store_contract.py @@ -47,6 +47,27 @@ def test_order_independence_and_identity_changes(self) -> None: 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): @@ -62,6 +83,19 @@ 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_adversarial_wire_values_reject(self) -> None: wire = _contract().to_wire() for mutation in ( From c1a76ae2791188d7aeee8da5698f08ab9838843a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 02:36:58 +0800 Subject: [PATCH 3/4] fix: bound vnext json integers Co-Authored-By: Codex --- .../vnext_result_store_contract.py | 11 +++++++++-- tests/test_vnext_result_store_contract.py | 13 +++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py index f10e803..d21278c 100644 --- a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py @@ -25,6 +25,7 @@ VNEXT_NAMESPACE = "qpk-vnext/result/v1" _SCHEMA_VERSION = "qpk.vnext.result.v1" _MAX_SEGMENT = 100 +MAX_SAFE_JSON_INTEGER = 2**53 - 1 _PROFILES = frozenset({"SOXL", "TQQQ"}) _WIRE_FIELDS = frozenset({ "namespace", "schema_version", "domain", "canonical_profile", "execution_timing", @@ -55,7 +56,11 @@ def _segment(value: Any) -> str: def _param_value(value: Any) -> Any: if value is None: raise VNextContractError() - if isinstance(value, (str, bool, int)): + 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 @@ -78,7 +83,9 @@ def thaw(item: Any) -> Any: 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, int)) or (isinstance(item, float) and math.isfinite(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)} diff --git a/tests/test_vnext_result_store_contract.py b/tests/test_vnext_result_store_contract.py index 1b289d4..5375c9f 100644 --- a/tests/test_vnext_result_store_contract.py +++ b/tests/test_vnext_result_store_contract.py @@ -6,6 +6,7 @@ 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, @@ -96,6 +97,18 @@ def test_direct_mutable_values_reject_but_wire_lists_are_frozen(self) -> None: 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_adversarial_wire_values_reject(self) -> None: wire = _contract().to_wire() for mutation in ( From 92e7cbf3e421e0b8305da1b8614325edce4ff1e4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Wed, 15 Jul 2026 03:00:51 +0800 Subject: [PATCH 4/4] fix: generalize vnext profiles and audit timestamps Co-Authored-By: Codex --- .../vnext_result_store_contract.py | 29 ++++++++++++++++--- tests/test_vnext_result_store_contract.py | 16 ++++++++++ 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py index d21278c..7b8242b 100644 --- a/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py +++ b/src/quant_platform_kit/strategy_lifecycle/vnext_result_store_contract.py @@ -15,6 +15,8 @@ import hashlib import json import math +import re +from datetime import datetime from dataclasses import dataclass from enum import Enum from types import MappingProxyType @@ -26,7 +28,7 @@ _SCHEMA_VERSION = "qpk.vnext.result.v1" _MAX_SEGMENT = 100 MAX_SAFE_JSON_INTEGER = 2**53 - 1 -_PROFILES = frozenset({"SOXL", "TQQQ"}) +_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", @@ -99,6 +101,26 @@ def _text(value: Any) -> str: 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() @@ -132,8 +154,7 @@ class VNextResultContract: def __post_init__(self) -> None: _segment(self.domain) - if self.canonical_profile not in _PROFILES: - raise VNextContractError() + _canonical_profile(self.canonical_profile) try: object.__setattr__(self, "execution_timing", ExecutionTiming(self.execution_timing)) object.__setattr__(self, "persist_mode", PersistMode(self.persist_mode)) @@ -145,7 +166,7 @@ def __post_init__(self) -> None: raise VNextContractError() if not isinstance(self.param_version, int) or isinstance(self.param_version, bool) or self.param_version < 1: raise VNextContractError() - _text(self.computed_at) + _computed_at(self.computed_at) object.__setattr__(self, "params", MappingProxyType(_params(self.params))) def _identity_payload(self) -> dict[str, Any]: diff --git a/tests/test_vnext_result_store_contract.py b/tests/test_vnext_result_store_contract.py index 5375c9f..41e3ad2 100644 --- a/tests/test_vnext_result_store_contract.py +++ b/tests/test_vnext_result_store_contract.py @@ -40,6 +40,8 @@ def test_round_trip_and_namespace_isolation(self) -> None: 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}) @@ -77,6 +79,9 @@ def test_required_none_legacy_any_alias_and_unknown_values_reject(self) -> None: _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") @@ -109,6 +114,17 @@ def test_json_safe_integer_boundary_and_nested_rejection(self) -> None: 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 (