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: with (FIXTURE / name).open("rb") as source: 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") self.assertEqual(contract["schema"], "lazaret-ignored-workspace-contract-v1") self.assertEqual(contract["mechanism"], "rootless-host-fuse-router") self.assertEqual(contract["route_classes"], ["host", "isolated"]) def test_cross_route_namespace_operations_are_directionally_bounded(self) -> None: transitions = { transition["id"]: transition for transition in load("contract.json")["transition_contracts"] } for transition_id in ( "cross-route-rename", "rename-whiteout", "cross-route-hard-link", ): with self.subTest(transition_id=transition_id): self.assertEqual(transitions[transition_id]["mutation"], "none") self.assertEqual(transitions["cross-route-rename"]["result"], "EXDEV") self.assertEqual( transitions["qualified-host-to-isolated-rename"], { "id": "qualified-host-to-isolated-rename", "operations": ["rename", "rename-noreplace"], "result": "success", "mutation": "durable-host-to-isolated-migration", "copy": "host-to-isolated-only", }, ) self.assertEqual(transitions["cross-route-hard-link"]["result"], "EXDEV") self.assertEqual(transitions["rename-whiteout"]["result"], "EOPNOTSUPP") def test_transitions_copy_only_toward_isolated_state_between_generations(self) -> None: transitions = load("contract.json")["transition_contracts"] copying = [ transition["id"] for transition in transitions if "copy" in transition and transition["copy"] != "none" ] self.assertEqual( copying, ["qualified-host-to-isolated-rename", "host-to-ignored-policy"], ) handle_transitions = { transition["id"]: transition["open_handles"] for transition in transitions if "open_handles" in transition } self.assertEqual( handle_transitions, {}, ) def test_isolated_only_force_add_is_explicitly_refused(self) -> None: contract = load("contract.json") transitions = { transition["id"]: transition for transition in contract["transition_contracts"] } transition = transitions["isolated-to-tracked-guest-publication"] self.assertEqual(transition["result"], "EBUSY") self.assertEqual(transition["index_mutation"], "none") compatibility = { operation["operation"]: operation["support"] for operation in contract["git_compatibility"] } self.assertEqual( compatibility["force-add-isolated-only"], "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": "qualified-host-to-isolated-rename", "operations": ["rename", "rename-noreplace"], "result": "success", "mutation": "durable-host-to-isolated-migration", "copy": "host-to-isolated-only", }, { "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": "next Domain initialization classifies an isolated path as host-backed", "new_lookups": "host-or-absent", "isolated_state": "deleted-before-readiness", "copy": "none", }, { "id": "host-to-ignored-policy", "trigger": "next Domain initialization classifies a host-backed path as ignored", "new_lookups": "host-clone-or-absent", "host_state": "retained-unchanged", "isolated_state": "host-clone-replaces-stale-state", "copy": "host-to-isolated-only", }, { "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", }, ], ) 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 for operation in load("contract.json")["filesystem_operations"] } required = { "lookup-getattr-readdir", "create-mkdir-open-read-write-truncate-fallocate-fsync", "rename", "rename-noreplace", "rename-exchange", "rename-whiteout", "hard-link", "symlink-readlink", "user-xattrs", "posix-acls", "sparse-seek-hole-data-punch-hole", "copy-file-range", "utimens-nanoseconds", "mmap-shared-private", "flock-posix-record-lock", "fifo-unix-socket", "character-block-device-create", } self.assertTrue(required.issubset(operations)) self.assertEqual( operations["rename"]["cross_route"], "qualified-host-to-isolated-or-EXDEV", ) self.assertEqual(operations["hard-link"]["cross_route"], "EXDEV") self.assertEqual(operations["copy-file-range"]["cross_route"], "EXDEV") self.assertEqual(operations["character-block-device-create"]["host"], "EPERM") def test_corpus_ids_and_byte_paths_are_canonical(self) -> None: cases = load("corpus.json")["cases"] ids = [case["id"] for case in cases] self.assertEqual(len(ids), len(set(ids))) for case in cases: with self.subTest(case=case["id"]): path_hex = case["path_hex"] self.assertEqual(path_hex, path_hex.lower()) self.assertEqual(bytes.fromhex(path_hex).hex(), path_hex) self.assertNotIn(b"\x00", bytes.fromhex(path_hex)) non_utf8 = next(case for case in cases if case["id"] == "non-utf8-ignore-match") with self.assertRaises(UnicodeDecodeError): bytes.fromhex(non_utf8["path_hex"]).decode("utf-8") def test_corpus_covers_every_required_behavior(self) -> None: cases = load("corpus.json")["cases"] tags = {tag for case in cases for tag in case["tags"]} required = { "non-utf8-names", "ignore-precedence", "core-ignore-case", "worktrees", "nested-repositories", "mmap", "locks", "xattrs", "acls", "sparse", "copy-range", "rename", "link", "symlink", "timestamps", "special-files", "concurrent-mutation", "cross-policy-rename", "ignored-to-tracked-transition", "startup-sealed-policy", "startup-policy-transition", "host-to-isolated-copy", "no-silent-loss", "no-copy-back", } self.assertTrue(required.issubset(tags), sorted(required - tags)) def test_corpus_asserts_backing_state_for_refusal_and_loss_cases(self) -> None: cases = {case["id"]: case for case in load("corpus.json")["cases"]} self.assertEqual( cases["cross-route-file-rename"]["expect"], ["EXDEV", "source-unchanged", "destination-absent"], ) self.assertIn("isolated-object-deleted", cases["unmediated-inconsistent-index"]["expect"]) 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: qualification = load("benchmark.json")["qualification_environment"] required = { "architecture", "kernel", "cpu", "memory", "storage", "filesystem", "qualification_result_fields", "qualification_result_policy", } self.assertEqual(set(qualification), required) self.assertEqual(qualification["architecture"], "x86_64") self.assertEqual( set(qualification["cpu"]), {"physical_cores_minimum", "allocation", "governor"}, ) self.assertGreaterEqual(qualification["cpu"]["physical_cores_minimum"], 8) self.assertGreaterEqual(qualification["memory"]["bytes_minimum"], 16 * 1024**3) self.assertEqual( qualification["storage"], { "kind": "local-block-device", "network_backed": False, "copy_on_write": False, }, ) self.assertEqual(qualification["filesystem"]["type"], "XFS") self.assertEqual(qualification["filesystem"]["block_bytes"], 4096) self.assertIn("cpu_model", qualification["qualification_result_fields"]) self.assertIn("storage_model", qualification["qualification_result_fields"]) def test_sampling_protocol_fixes_statistics_and_variance(self) -> None: sampling = load("benchmark.json")["sampling"] self.assertEqual(sampling["independent_runs"], 3) self.assertEqual(sampling["warmup_samples_per_cell"], 2) self.assertEqual(sampling["measured_samples_per_cell"], 15) self.assertEqual(sampling["statistic"], "median") self.assertEqual( sampling["relative_statistic"], "median of paired routed/direct elapsed-time ratios", ) self.assertEqual( sampling["direct_calibration_formula"], "direct placement median <= absolute_median_ms / relative_median_ratio in every independent run and preparation state", ) 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"] self.assertEqual( {workload["category"] for workload in workloads}, {"metadata", "git", "package-manager", "build"}, ) for workload in workloads: with self.subTest(workload=workload["id"]): self.assertGreater(workload["absolute_median_ms"], 0) self.assertGreater(workload["relative_median_ratio"], 1.0) self.assertEqual(workload["preparation_states"], ["unprimed", "primed"]) self.assertEqual(len(workload["fixture_seed"]), 64) self.assertTrue(workload["fixture_algorithm"]) self.assertTrue(workload["command_sequence"]) self.assertTrue(all(command for command in workload["command_sequence"])) self.assertTrue(workload["result_oracle"]) def test_specification_names_the_committed_contract(self) -> None: specification = (ROOT / "docs/spec/lazaret.md").read_text() self.assertIn("#### Ignored Workspace State routing contract", specification) self.assertIn("`lazaret-ignored-workspace-contract-v1`", specification) self.assertIn("rootless host FUSE filesystem", specification) if __name__ == "__main__": unittest.main()