test(#831): add failing tests for verbatim-hashed fragments (B_payload/H_payload, no-self-ref, source-invalidation)

packages/mosaic/package.json

@@ -25,7 +25,7 @@
     "lint": "eslint src",
     "typecheck": "tsc --noEmit",
     "test": "vitest run --passWithNoTests && pnpm run test:framework-shell",
-    "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
+    "test:framework-shell": "python3 src/lease-broker/daemon_deadline_unittest.py && python3 src/lease-broker/normative_fragments_unittest.py && python3 src/mutator-gate/runtime_tools_unittest.py && python3 src/mutator-gate/runtime_launch_guard_unittest.py && python3 framework/tools/lease-broker/check-runtime-launches.py --root ../.. && bash framework/tools/codex/test-pr-diff-context.sh"
   },
   "dependencies": {
     "@mosaicstack/brain": "workspace:*",

packages/mosaic/src/lease-broker/normative_fragments_unittest.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""RED-first contract tests for verbatim-hashed normative fragments."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+
+MODULE_PATH = Path(__file__).parents[2] / "framework/tools/lease-broker/normative_fragments.py"
+
+
+def shipped_module():
+    # Each test reaches the shipped implementation; no test doubles or local
+    # reimplementation of construction are allowed on this admission surface.
+    assert MODULE_PATH.is_file(), f"shipped construction module is missing: {MODULE_PATH}"
+    spec = importlib.util.spec_from_file_location("normative_fragments", MODULE_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError("unable to load normative fragment construction")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def fragment(module, source_id: str, content: bytes):
+    return module.NormativeFragment(
+        source_id=source_id,
+        content=content,
+        expected_sha256=hashlib.sha256(content).hexdigest(),
+    )
+
+
+def valid_fragments(module):
+    return [
+        fragment(module, "authority/constitution", b"Constitution\n"),
+        fragment(module, "authority/runtime", b"Runtime\n"),
+    ]
+
+
+class NormativeFragmentsTest(unittest.TestCase):
+    def test_t24_claude_and_pi_builders_are_byte_identical_and_one_way(self) -> None:
+        module = shipped_module()
+        fragments = valid_fragments(module)
+
+        claude = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+        pi = module.build_for_pi(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+
+        self.assertEqual(claude.injectionDecision, "ACCEPTED")
+        self.assertTrue(claude.promotion)
+        self.assertEqual(claude.b_payload, pi.b_payload)
+        self.assertEqual(claude.h_payload, pi.h_payload)
+        self.assertEqual(
+            claude.h_payload,
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + module.length_frame([claude.b_payload])).hexdigest(),
+        )
+        self.assertNotIn(b"h_payload", claude.b_payload)
+
+        mutated_fragment = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[
+                fragment(module, "authority/constitution", b"Constitution changed\n"),
+                fragment(module, "authority/runtime", b"Runtime\n"),
+            ],
+        )
+        mutated_metadata = module.build_for_claude(
+            manifest_version=2,
+            generator_version="wi4-test",
+            fragments=fragments,
+        )
+        self.assertNotEqual(claude.h_payload, mutated_fragment.h_payload)
+        self.assertNotEqual(claude.h_payload, mutated_metadata.h_payload)
+
+    def test_length_framing_and_domain_separation_prevent_ambiguous_construction(self) -> None:
+        module = shipped_module()
+        left = module.length_frame([b"ab", b"c"])
+        right = module.length_frame([b"a", b"bc"])
+
+        self.assertNotEqual(left, right)
+        self.assertNotEqual(
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + right).hexdigest(),
+        )
+        self.assertNotEqual(
+            hashlib.sha256(module.HASH_DOMAIN_SEPARATOR + left).hexdigest(),
+            hashlib.sha256(b"other-context\x00" + left).hexdigest(),
+        )
+
+    def test_source_invalidation_missing_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        missing = module.NormativeFragment(
+            source_id="authority/missing",
+            content=None,
+            expected_sha256=hashlib.sha256(b"missing").hexdigest(),
+        )
+
+        result = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[missing],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "missing")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+    def test_source_invalidation_oversize_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        content = b"x" * (module.MAX_FRAGMENT_BYTES + 1)
+        oversize = fragment(module, "authority/oversize", content)
+
+        result = module.build_for_pi(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[oversize],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "oversize")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+    def test_source_invalidation_hash_mismatch_refuses_real_construction_and_promotion(self) -> None:
+        module = shipped_module()
+        mismatch = module.NormativeFragment(
+            source_id="authority/hash-mismatch",
+            content=b"trusted bytes",
+            expected_sha256=hashlib.sha256(b"different bytes").hexdigest(),
+        )
+
+        result = module.build_for_claude(
+            manifest_version=1,
+            generator_version="wi4-test",
+            fragments=[mismatch],
+        )
+
+        self.assertEqual(result.injectionDecision, "REFUSED")
+        self.assertFalse(result.promotion)
+        self.assertEqual(result.source_reason, "hash-mismatch")
+        self.assertIsNone(result.b_payload)
+        self.assertIsNone(result.h_payload)
+
+
+if __name__ == "__main__":
+    unittest.main()