Make workspace contract verification unambiguous

tests/fixtures/ignored-workspace-v1/benchmark.json

@@ -38,6 +38,9 @@
     "clock": "CLOCK_MONOTONIC_RAW elapsed nanoseconds",
     "statistic": "median",
     "relative_statistic": "median of paired routed/direct elapsed-time ratios",
+    "median_definition": "sort numeric values ascending and select value at zero-based index floor(count/2); every measured count is odd",
+    "within_run_relative_mad_formula": "median(abs(sample - median(samples))) / median(samples), evaluated separately for direct elapsed times, routed elapsed times, and paired routed/direct ratios in each workload, preparation state, and independent run",
+    "between_run_median_spread_formula": "(max(run_medians) - min(run_medians)) / median(run_medians), evaluated across the three independent-run medians separately for direct elapsed time, routed elapsed time, and paired routed/direct ratio in each workload and preparation state",
     "outlier_policy": "retain every completed measured sample",
     "sample_timeout": "four times the workload absolute budget",
     "within_run_relative_mad_limit": 0.10,
@@ -49,12 +52,40 @@
     "primed": "fresh materialization followed by one untimed complete workload invocation before measured samples"
   },
   "normalization": {
-    "fixture_identity": "SHA-256 of canonical byte-path, kind, mode, size, and regular-file content records",
-    "result_identity": "workload-specific canonical output plus post-workload fixture identity",
     "environment": "reject a run that differs from any reference environment field",
     "background_load": "reject a sample when another process consumes more than five percent of one reference core or benchmark memory pressure causes reclaim",
     "pass_rule": "every independent run and preparation state meets both the absolute median and paired median-ratio budget and both variance limits"
   },
+  "identity_encoding": {
+    "integer_encoding": "unsigned integers are fixed-width big-endian; the command exit status is signed two's-complement i32 big-endian",
+    "fixture": {
+      "context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d666978747572652d763100",
+      "root_entry": "omitted",
+      "path": "raw relative pathname bytes with slash separators, no leading or trailing slash, and no NUL",
+      "ordering": "ascending lexicographic order of complete raw path byte strings",
+      "kind_tags": {
+        "directory": 1,
+        "regular": 2,
+        "symlink": 3
+      },
+      "mode": "u32 of st_mode masked with octal 7777",
+      "size": "u64 payload length for regular files and symlinks; zero for directories",
+      "payload": "regular file bytes, raw symlink target bytes, or empty bytes for a directory",
+      "record": "u32(path length) || path || u8(kind tag) || u32(mode) || u64(size) || SHA-256(payload)",
+      "stream": "context bytes || u32(record count) || records in declared ordering",
+      "identity": "SHA-256(stream)",
+      "unsupported_entry": "reject identity construction for any other file kind"
+    },
+    "result": {
+      "context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d726573756c742d763100",
+      "workload": "u32(UTF-8 workload ID length) || UTF-8 workload ID",
+      "command_record": "i32(exit status) || u64(stdout length) || exact stdout bytes || u64(stderr length) || exact stderr bytes",
+      "ordering": "command records remain in command_sequence order",
+      "stream": "context bytes || framed workload || initial fixture identity || u32(command record count) || ordered command records || post-workload fixture identity",
+      "identity": "SHA-256(stream)",
+      "successful_sample": "every command exit status is zero and both fixture identities and every output byte are retained"
+    }
+  },
   "workloads": [
     {
       "id": "metadata-10000",

tests/tooling/test_ignored_workspace_contract.py

@@ -1,12 +1,41 @@
 from __future__ import annotations
 
+import copy
+from fractions import Fraction
+import hashlib
 import json
 from pathlib import Path
+import struct
 import unittest
 
 
 ROOT = Path(__file__).resolve().parents[2]
 FIXTURE = ROOT / "tests/fixtures/ignored-workspace-v1"
+EXPECTED_GIT_COMPATIBILITY = {
+    "status-diff": "supported",
+    "ordinary-add-reset-commit": "supported-host-backed-paths",
+    "checkout-restore": "supported",
+    "branch-reference-update": "supported",
+    "merge-rebase-stash": "supported",
+    "linked-worktree": "supported",
+    "nested-repository": "supported",
+    "force-add-isolated-only": "EBUSY-before-index-publication",
+    "host-config-write": "EROFS",
+    "default-hook-host-access": "sandbox-transient-only",
+}
+EXPECTED_FORCE_ADD_CASE = {
+    "id": "guest-force-add-isolated-only",
+    "tags": ["ignored-to-tracked-transition", "git-add", "no-silent-loss"],
+    "path_hex": "7461726765742f617070",
+    "setup": ["isolated-object", "path-ignored", "host-path-absent"],
+    "actions": ["sandbox-git-add-force"],
+    "expect": [
+        "EBUSY-before-index-publication",
+        "prior-index-retained",
+        "isolated-object-visible",
+        "host-path-absent",
+    ],
+}
 
 
 def load(name: str) -> dict:
@@ -14,6 +43,109 @@ def load(name: str) -> dict:
         return json.load(source)
 
 
+def odd_median(values: list[int | Fraction]) -> int | Fraction:
+    if not values or len(values) % 2 == 0:
+        raise ValueError("median input must have odd length")
+    ordered = sorted(values)
+    return ordered[len(ordered) // 2]
+
+
+def relative_mad(values: list[int | Fraction]) -> Fraction:
+    center = odd_median(values)
+    if center == 0:
+        raise ValueError("relative MAD center must be positive")
+    return Fraction(odd_median([abs(value - center) for value in values]), center)
+
+
+def median_spread(run_medians: list[int]) -> Fraction:
+    center = odd_median(run_medians)
+    if center == 0:
+        raise ValueError("median spread center must be positive")
+    return Fraction(max(run_medians) - min(run_medians), center)
+
+
+def fixture_identity(records: list[tuple[bytes, str, int, bytes]]) -> bytes:
+    encoding = load("benchmark.json")["identity_encoding"]["fixture"]
+    kinds = encoding["kind_tags"]
+    ordered = sorted(records, key=lambda record: record[0])
+    if len({record[0] for record in ordered}) != len(ordered):
+        raise ValueError("duplicate fixture path")
+    stream = bytearray.fromhex(encoding["context_hex"])
+    stream.extend(struct.pack(">I", len(ordered)))
+    for path, kind, mode, payload in ordered:
+        components = path.split(b"/")
+        if (
+            not path
+            or path.startswith(b"/")
+            or path.endswith(b"/")
+            or b"\x00" in path
+            or any(component in (b"", b".", b"..") for component in components)
+        ):
+            raise ValueError("invalid fixture path")
+        if kind not in kinds:
+            raise ValueError("unsupported fixture kind")
+        if kind == "directory" and payload:
+            raise ValueError("directory fixture payload must be empty")
+        size = 0 if kind == "directory" else len(payload)
+        stream.extend(struct.pack(">I", len(path)))
+        stream.extend(path)
+        stream.extend(struct.pack(">B", kinds[kind]))
+        stream.extend(struct.pack(">I", mode & 0o7777))
+        stream.extend(struct.pack(">Q", size))
+        stream.extend(hashlib.sha256(payload).digest())
+    return hashlib.sha256(stream).digest()
+
+
+def result_identity(
+    workload: str,
+    initial_fixture: bytes,
+    commands: list[tuple[int, bytes, bytes]],
+    post_fixture: bytes,
+) -> bytes:
+    encoding = load("benchmark.json")["identity_encoding"]["result"]
+    workload_bytes = workload.encode("utf-8")
+    if not workload_bytes or len(initial_fixture) != 32 or len(post_fixture) != 32:
+        raise ValueError("invalid result identity input")
+    stream = bytearray.fromhex(encoding["context_hex"])
+    stream.extend(struct.pack(">I", len(workload_bytes)))
+    stream.extend(workload_bytes)
+    stream.extend(initial_fixture)
+    stream.extend(struct.pack(">I", len(commands)))
+    for exit_status, stdout, stderr in commands:
+        stream.extend(struct.pack(">iQ", exit_status, len(stdout)))
+        stream.extend(stdout)
+        stream.extend(struct.pack(">Q", len(stderr)))
+        stream.extend(stderr)
+    stream.extend(post_fixture)
+    return hashlib.sha256(stream).digest()
+
+
+def validate_semantic_consistency(contract: dict, corpus: dict) -> None:
+    compatibility = {
+        operation["operation"]: operation["support"]
+        for operation in contract["git_compatibility"]
+    }
+    if compatibility != EXPECTED_GIT_COMPATIBILITY:
+        raise ValueError("Git compatibility differs from the selected contract")
+    transitions = {
+        transition["id"]: transition
+        for transition in contract["transition_contracts"]
+    }
+    guest_publication = transitions["isolated-to-tracked-guest-publication"]
+    if guest_publication != {
+        "id": "isolated-to-tracked-guest-publication",
+        "trigger": "guest index publication tracks an isolated-only path",
+        "result": "EBUSY",
+        "index_mutation": "none",
+        "isolated_state": "retained-visible",
+        "copy": "none",
+    }:
+        raise ValueError("guest force-add transition differs from the selected contract")
+    cases = {case["id"]: case for case in corpus["cases"]}
+    if cases["guest-force-add-isolated-only"] != EXPECTED_FORCE_ADD_CASE:
+        raise ValueError("guest force-add corpus differs from the selected contract")
+
+
 class IgnoredWorkspaceContractTests(unittest.TestCase):
     def test_contract_dispatch_and_selected_mechanism_are_exact(self) -> None:
         contract = load("contract.json")
@@ -77,6 +209,82 @@ class IgnoredWorkspaceContractTests(unittest.TestCase):
             "EBUSY-before-index-publication",
         )
 
+    def test_transition_matrix_is_exactly_the_selected_contract(self) -> None:
+        transitions = load("contract.json")["transition_contracts"]
+        self.assertEqual(
+            transitions,
+            [
+                {
+                    "id": "cross-route-rename",
+                    "operations": ["rename", "rename-noreplace", "rename-exchange"],
+                    "result": "EXDEV",
+                    "mutation": "none",
+                },
+                {
+                    "id": "rename-whiteout",
+                    "operations": ["rename-whiteout"],
+                    "result": "EOPNOTSUPP",
+                    "mutation": "none",
+                },
+                {
+                    "id": "cross-route-hard-link",
+                    "operations": ["link"],
+                    "result": "EXDEV",
+                    "mutation": "none",
+                },
+                {
+                    "id": "ignored-to-host-policy",
+                    "trigger": "path becomes non-ignored without becoming tracked",
+                    "new_lookups": "host-or-absent",
+                    "open_handles": "remain-isolated",
+                    "isolated_state": "retained-hidden",
+                    "copy": "none",
+                },
+                {
+                    "id": "host-to-ignored-policy",
+                    "trigger": "untracked path becomes ignored",
+                    "new_lookups": "retained-isolated-or-absent",
+                    "open_handles": "remain-host",
+                    "host_state": "retained-hidden",
+                    "copy": "none",
+                },
+                {
+                    "id": "isolated-to-tracked-guest-publication",
+                    "trigger": "guest index publication tracks an isolated-only path",
+                    "result": "EBUSY",
+                    "index_mutation": "none",
+                    "isolated_state": "retained-visible",
+                    "copy": "none",
+                },
+                {
+                    "id": "isolated-to-tracked-host-established",
+                    "trigger": "published index and type-compatible host object both exist",
+                    "new_lookups": "host",
+                    "open_handles": "remain-isolated",
+                    "isolated_state": "retained-hidden",
+                    "copy": "none",
+                },
+                {
+                    "id": "isolated-to-tracked-unmediated-inconsistent",
+                    "trigger": "unmediated index publication tracks a path without a host object",
+                    "result": "EIO",
+                    "path_state": "fail-closed-until-consistent",
+                    "isolated_state": "retained-hidden-recoverable",
+                    "copy": "none",
+                },
+            ],
+        )
+
+    def test_git_compatibility_is_exactly_the_selected_contract(self) -> None:
+        compatibility = {
+            operation["operation"]: operation["support"]
+            for operation in load("contract.json")["git_compatibility"]
+        }
+        self.assertEqual(
+            compatibility,
+            EXPECTED_GIT_COMPATIBILITY,
+        )
+
     def test_operation_matrix_covers_required_surface(self) -> None:
         operations = {
             operation["operation"]: operation
@@ -160,6 +368,43 @@ class IgnoredWorkspaceContractTests(unittest.TestCase):
         self.assertIn("access-error-not-host-fallback", cases["router-loss-fails-closed"]["expect"])
         self.assertIn("host-sentinel-unchanged", cases["router-loss-fails-closed"]["expect"])
 
+    def test_force_add_corpus_is_exactly_fail_closed_without_copy_back(self) -> None:
+        cases = {case["id"]: case for case in load("corpus.json")["cases"]}
+        self.assertEqual(
+            cases["guest-force-add-isolated-only"],
+            EXPECTED_FORCE_ADD_CASE,
+        )
+
+    def test_reproduced_semantic_mutations_are_refused(self) -> None:
+        contract = load("contract.json")
+        corpus = load("corpus.json")
+        validate_semantic_consistency(contract, corpus)
+
+        unsupported_merge = copy.deepcopy(contract)
+        merge = next(
+            operation
+            for operation in unsupported_merge["git_compatibility"]
+            if operation["operation"] == "merge-rebase-stash"
+        )
+        merge["support"] = "unsupported"
+        with self.assertRaisesRegex(ValueError, "Git compatibility"):
+            validate_semantic_consistency(unsupported_merge, corpus)
+
+        copied_force_add = copy.deepcopy(corpus)
+        force_add = next(
+            case
+            for case in copied_force_add["cases"]
+            if case["id"] == "guest-force-add-isolated-only"
+        )
+        force_add["actions"] = [
+            "sandbox-git-add-force",
+            "copy-isolated-bytes-to-host",
+            "publish-index",
+        ]
+        force_add["expect"] = ["success", "host-copy", "index-published"]
+        with self.assertRaisesRegex(ValueError, "force-add corpus"):
+            validate_semantic_consistency(contract, copied_force_add)
+
     def test_benchmark_reference_conditions_are_complete(self) -> None:
         reference = load("benchmark.json")["reference_environment"]
         required = {
@@ -196,11 +441,84 @@ class IgnoredWorkspaceContractTests(unittest.TestCase):
             sampling["relative_statistic"],
             "median of paired routed/direct elapsed-time ratios",
         )
+        self.assertEqual(
+            sampling["within_run_relative_mad_formula"],
+            "median(abs(sample - median(samples))) / median(samples), evaluated separately for direct elapsed times, routed elapsed times, and paired routed/direct ratios in each workload, preparation state, and independent run",
+        )
+        self.assertEqual(
+            sampling["between_run_median_spread_formula"],
+            "(max(run_medians) - min(run_medians)) / median(run_medians), evaluated across the three independent-run medians separately for direct elapsed time, routed elapsed time, and paired routed/direct ratio in each workload and preparation state",
+        )
         self.assertEqual(sampling["outlier_policy"], "retain every completed measured sample")
         self.assertLessEqual(sampling["within_run_relative_mad_limit"], 0.10)
         self.assertLessEqual(sampling["between_run_median_spread_limit"], 0.10)
         self.assertEqual(sampling["budget_tolerance"], 0.0)
 
+    def test_variance_formulas_distinguish_the_reproduced_values(self) -> None:
+        values = [100, 110, 111]
+        self.assertEqual(relative_mad(values), Fraction(1, 110))
+        self.assertEqual(median_spread(values), Fraction(1, 10))
+        self.assertNotEqual(relative_mad(values), median_spread(values))
+
+    def test_identity_encoding_is_exactly_framed_and_ordered(self) -> None:
+        encoding = load("benchmark.json")["identity_encoding"]
+        self.assertEqual(
+            encoding,
+            {
+                "integer_encoding": "unsigned integers are fixed-width big-endian; the command exit status is signed two's-complement i32 big-endian",
+                "fixture": {
+                    "context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d666978747572652d763100",
+                    "root_entry": "omitted",
+                    "path": "raw relative pathname bytes with slash separators, no leading or trailing slash, and no NUL",
+                    "ordering": "ascending lexicographic order of complete raw path byte strings",
+                    "kind_tags": {"directory": 1, "regular": 2, "symlink": 3},
+                    "mode": "u32 of st_mode masked with octal 7777",
+                    "size": "u64 payload length for regular files and symlinks; zero for directories",
+                    "payload": "regular file bytes, raw symlink target bytes, or empty bytes for a directory",
+                    "record": "u32(path length) || path || u8(kind tag) || u32(mode) || u64(size) || SHA-256(payload)",
+                    "stream": "context bytes || u32(record count) || records in declared ordering",
+                    "identity": "SHA-256(stream)",
+                    "unsupported_entry": "reject identity construction for any other file kind",
+                },
+                "result": {
+                    "context_hex": "6c617a617265742d69676e6f7265642d776f726b73706163652d726573756c742d763100",
+                    "workload": "u32(UTF-8 workload ID length) || UTF-8 workload ID",
+                    "command_record": "i32(exit status) || u64(stdout length) || exact stdout bytes || u64(stderr length) || exact stderr bytes",
+                    "ordering": "command records remain in command_sequence order",
+                    "stream": "context bytes || framed workload || initial fixture identity || u32(command record count) || ordered command records || post-workload fixture identity",
+                    "identity": "SHA-256(stream)",
+                    "successful_sample": "every command exit status is zero and both fixture identities and every output byte are retained",
+                },
+            },
+        )
+
+    def test_fixture_identity_has_one_exact_canonical_encoding(self) -> None:
+        records = [
+            (b"link", "symlink", 0o777, b"a/\xff"),
+            (b"a/\xff", "regular", 0o640, b"x"),
+            (b"a", "directory", 0o755, b""),
+        ]
+        self.assertEqual(
+            fixture_identity(records).hex(),
+            "72f539c740d0048772a2380969c2b22fa970f725958d17b7a534069e92d1994d",
+        )
+        self.assertEqual(fixture_identity(records), fixture_identity(list(reversed(records))))
+        with self.assertRaisesRegex(ValueError, "unsupported fixture kind"):
+            fixture_identity([(b"socket", "socket", 0o600, b"")])
+
+    def test_result_identity_frames_commands_outputs_and_fixture_states(self) -> None:
+        initial = bytes.fromhex("11" * 32)
+        post = bytes.fromhex("22" * 32)
+        commands = [(0, b"a", b""), (1, b"", b"b")]
+        self.assertEqual(
+            result_identity("git", initial, commands, post).hex(),
+            "0fe91ab5411628dc22cb2fb3161486535794c8044d74453c7ab7eafd1c93b4e7",
+        )
+        self.assertNotEqual(
+            result_identity("git", initial, list(reversed(commands)), post),
+            result_identity("git", initial, commands, post),
+        )
+
     def test_each_workload_has_absolute_and_relative_budgets(self) -> None:
         benchmark = load("benchmark.json")
         workloads = benchmark["workloads"]