fix: enforce canonical factor contract json

src/quant_engine/factor_contracts.py

@@ -157,6 +157,14 @@ def _duplicate_key_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
 
 
 def _parse_json_object(value: str | bytes, path: str) -> dict[str, Any]:
+    try:
+        source_bytes = value if isinstance(value, bytes) else value.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        raise FactorContractError(
+            ContractErrorCode.INVALID_FORMAT,
+            path,
+            "invalid UTF-8 JSON",
+        ) from exc
     try:
         loaded = json.loads(value, object_pairs_hook=_duplicate_key_pairs)
     except (UnicodeDecodeError, json.JSONDecodeError) as exc:
@@ -168,6 +176,8 @@ def _parse_json_object(value: str | bytes, path: str) -> dict[str, Any]:
     if type(loaded) is not dict:
         _fail(ContractErrorCode.TYPE_ERROR, path, "top-level JSON value must be an object")
     _assert_canonical_profile(loaded, path)
+    if source_bytes != canonical_json_bytes(loaded):
+        _fail(ContractErrorCode.INVALID_FORMAT, path, "JSON is not canonical")
     return cast(dict[str, Any], loaded)
 
 

tests/test_factor_contracts.py

@@ -263,6 +263,57 @@ def test_golden_contracts_are_content_addressed_round_trippable_and_deeply_immut
         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",))