Fix offline test isolation and restore mature CPU coverage

ci/conditional-test-policy.json

@@ -34,40 +34,40 @@
       "gate": "cuda-runtime",
       "reason": "Titan has a verified RTX 4090 and Gitea GPU runner, but no GitHub self-hosted CUDA runner is registered or selectable for this workflow.",
       "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/110",
-      "opened": "2026-08-16",
-      "expires": "2026-09-15",
+      "opened": "2026-09-20",
+      "expires": "2026-10-20",
       "blocked_claim": "CUDA runtime support, compatibility, correctness, and performance are not claimed while this waiver is active."
     },
     {
       "gate": "bitsandbytes-runtime",
       "reason": "Titan has a verified RTX 4090 and Gitea GPU runner, but no GitHub self-hosted CUDA and bitsandbytes runner is registered or selectable for this workflow.",
       "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/110",
-      "opened": "2026-08-16",
-      "expires": "2026-09-15",
+      "opened": "2026-09-20",
+      "expires": "2026-10-20",
       "blocked_claim": "bitsandbytes runtime support, compatibility, correctness, and performance are not claimed while this waiver is active."
     },
     {
       "gate": "mps-runtime",
       "reason": "Mutsu is a verified 16 GB Apple M4 builder, but no GitHub self-hosted MPS runner is registered or selectable for this workflow.",
       "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/110",
-      "opened": "2026-08-16",
-      "expires": "2026-09-15",
+      "opened": "2026-09-20",
+      "expires": "2026-10-20",
       "blocked_claim": "Apple MPS runtime support, compatibility, correctness, and performance are not claimed while this waiver is active."
     },
     {
       "gate": "mlx-runtime",
       "reason": "Mutsu is a verified 16 GB Apple M4 builder, but no GitHub self-hosted MLX runner is registered or selectable for this workflow.",
       "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/110",
-      "opened": "2026-08-16",
-      "expires": "2026-09-15",
+      "opened": "2026-09-20",
+      "expires": "2026-10-20",
       "blocked_claim": "Apple MLX runtime support, compatibility, correctness, and performance are not claimed while this waiver is active."
     },
     {
       "gate": "remote-execution",
       "reason": "No least-privileged remote test target or required repository credentials are configured.",
       "issue": "https://github.com/elder-plinius/OBLITERATUS/issues/110",
-      "opened": "2026-08-16",
-      "expires": "2026-09-15",
+      "opened": "2026-09-20",
+      "expires": "2026-10-20",
       "blocked_claim": "Remote execution support, compatibility, correctness, and performance are not claimed while this waiver is active."
     }
   ],

ci/test-risk-map.json

@@ -214,7 +214,8 @@
         "tests/test_distributed_gloo.py",
         "tests/test_distributed_launcher.py",
         "tests/test_distributed_preflight.py",
-        "tests/test_cli.py"
+        "tests/test_cli.py",
+        "tests/test_distributed_validation_edges.py"
       ]
     },
     {
@@ -358,7 +359,8 @@
         "tests/test_novel_analysis.py",
         "tests/test_projection_math_contracts.py",
         "tests/test_whitened_svd_oracles.py",
-        "tests/test_visualization.py"
+        "tests/test_visualization.py",
+        "tests/test_analysis_display_contracts.py"
       ]
     },
     {

docs/conditional-testing.md

@@ -23,7 +23,7 @@ waiver for at most 30 days. Each waiver names one gate, a canonical tracking iss
 the reason, its opening and expiry dates, and the support claim it blocks. Invalid,
 duplicate, future-dated, overlong, or expired waivers fail the policy job. The current
 waivers are tracked by [issue #110](https://github.com/elder-plinius/OBLITERATUS/issues/110)
-and expire on 2026-09-15; they are not evidence that any waived backend works.
+and expire on 2026-10-20; they are not evidence that any waived backend works.
 They block support, compatibility, correctness, and performance claims for the
 waived environment.
 
@@ -33,6 +33,29 @@ Apple M4 builder. Exact-head operator probes may be attached to issue #110, but
 they do not make the scheduled GitHub lanes runnable or establish a broad backend
 support claim.
 
+### September 2026 waiver renewal
+
+The validation-repair PR proposes renewing the five no-support waivers from
+2026-09-20 through 2026-10-20, within the existing 30-day maximum. The previous
+waivers expired on 2026-09-15; this renewal does not cover that lapse or turn old
+operator results into fresh evidence. Review and merge of the PR is the renewal
+decision; issue #110 remains open.
+
+The latest inventory recorded in
+[issue #110](https://github.com/elder-plinius/OBLITERATUS/issues/110#issuecomment-5547991209)
+reports unavailable runner inventory, no gate-enabling repository variables, and
+no remote-lane secrets. This repair has no fresh evidence that scheduled CUDA,
+bitsandbytes, MPS, MLX, or remote lanes are configured. It preserves the reasons
+and blocked claims pending that evidence. Ad hoc Titan access alone does not
+establish scheduled coverage or validate Apple and remote environments.
+
+Before 2026-10-20, maintainers must configure and verify the affected lanes or
+review another bounded renewal. Real-clock policy checks still reject expired
+waivers. Structural unit tests use explicit dates so unrelated assertions do not
+change with the calendar; separate tests assert that expired waivers fail closed.
+No coverage thresholds, gate-selection requirements, or evidence-freshness rules
+are relaxed by this renewal.
+
 ## Hosted gates
 
 The model gate downloads only

tests/test_analysis_display_contracts.py

@@ -0,0 +1,201 @@
+"""Numerical and presentation contracts for analysis results on the CPU."""
+
+from types import SimpleNamespace as NS
+
+import matplotlib.pyplot as plt
+import numpy as np
+import pytest
+import torch
+
+from obliteratus.analysis import visualization as plots
+from obliteratus.analysis.spectral_certification import SpectralCertifier
+from obliteratus.analysis.tuned_lens import RefusalTunedLens, TunedLensProbe
+
+pytestmark = pytest.mark.cpu
+
+
+@pytest.fixture(autouse=True)
+def close_figures():
+    yield
+    plt.close("all")
+
+
+@pytest.mark.parametrize("save", [False, True])
+def test_token_spectrum_preserves_order_signs_and_layer_selection(tmp_path, monkeypatch, save):
+    shown = []
+    monkeypatch.setattr(plt, "show", lambda: shown.append(True))
+    layer = NS(
+        layer_idx=7,
+        top_promoted=[("yes", 3.0), ("perhaps", 1.0)],
+        top_suppressed=[("no", -4.0), ("never", -2.0)],
+        refusal_compliance_gap=2.5,
+        refusal_specificity=0.75,
+    )
+    result = NS(per_layer={7: layer}, strongest_refusal_layer=7)
+    path = tmp_path / "tokens.png" if save else None
+    fig = plots.plot_logit_lens_spectrum(
+        result, layer_idx=7 if save else None, output_path=path, title="Spectrum" if save else None
+    )
+    ax = fig.axes[0]
+    assert [patch.get_width() for patch in ax.patches] == [-2, -4, 3, 1]
+    assert [label.get_text() for label in ax.get_yticklabels()] == [
+        "'never'",
+        "'no'",
+        "'yes'",
+        "'perhaps'",
+    ]
+    assert "2.5000" in ax.texts[0].get_text()
+    assert ax.get_title() == ("Spectrum" if save else "Logit Lens — Layer 7")
+    if save:
+        assert path.read_bytes().startswith(b"\x89PNG")
+        assert not shown
+    else:
+        assert shown == [True]
+    assert plots.plot_logit_lens_spectrum(result, layer_idx=99) is None
+
+
+@pytest.mark.parametrize("save", [False, True])
+def test_pareto_averages_scores_and_preserves_reference_coordinates(tmp_path, monkeypatch, save):
+    monkeypatch.setattr(plt, "show", lambda: None)
+    path = tmp_path / "pareto.png" if save else None
+    fig = plots.plot_capability_safety_pareto(
+        {"a": NS(score=0.2), "b": NS(score=0.8)},
+        0.3,
+        other_points=[(0.9, 0.6, "Baseline")],
+        output_path=path,
+    )
+    assert np.asarray(fig.axes[0].collections[0].get_offsets()).tolist() == [[0.3, 0.5]]
+    assert np.asarray(fig.axes[0].collections[1].get_offsets()).tolist() == [[0.9, 0.6]]
+    assert "Baseline" in [text.get_text() for text in fig.axes[0].texts]
+    if save:
+        assert path.read_bytes().startswith(b"\x89PNG")
+    empty = plots.plot_capability_safety_pareto({}, 0.0)
+    assert np.asarray(empty.axes[0].collections[0].get_offsets()).tolist() == [[0.0, 0.0]]
+
+
+def test_topology_missing_means_have_zero_strength_and_squeezes_direction(monkeypatch):
+    monkeypatch.setattr(plt, "show", lambda: None)
+    fig = plots.plot_refusal_topology(
+        {4: torch.tensor([[3.0, 0.0]]), 8: torch.tensor([0.0, 1.0])},
+        {4: torch.tensor([2.0, 7.0])},
+        {4: torch.tensor([0.0, 7.0])},
+        [4],
+    )
+    assert [p.get_height() for p in fig.axes[0].patches] == [2.0, 0.0]
+    assert [t.get_text() for t in fig.axes[0].get_xticklabels()] == ["4", "8"]
+
+
+@pytest.mark.parametrize(
+    "path", ["embed_out", "output", "transformer.wte", "model.embed_tokens", "gpt_neox.embed_in"]
+)
+def test_tuned_lens_finds_supported_output_or_tied_embeddings(path):
+    weight = torch.tensor([[1.0, 2.0], [3.0, 4.0]])
+    node = NS(weight=weight)
+    for part in reversed(path.split(".")):
+        node = NS(**{part: node})
+    torch.testing.assert_close(RefusalTunedLens()._get_unembedding_matrix(node), weight)
+
+
+def test_tuned_lens_missing_output_is_actionable():
+    with pytest.raises(RuntimeError, match="Cannot locate unembedding"):
+        RefusalTunedLens()._get_unembedding_matrix(NS())
+
+
+def test_tuned_lens_group_filters_invalid_and_unencodable_tokens():
+    def encode(text, **kwargs):
+        if text == "error":
+            raise ValueError("unknown token")
+        return {"valid": [1, 0], "empty": [], "negative": [-1], "outside": [3]}[text]
+
+    boosts = RefusalTunedLens()._get_token_group_boosts(
+        torch.tensor([2.0, 5.0, 9.0]),
+        NS(encode=encode),
+        ["error", "empty", "negative", "outside", "valid"],
+    )
+    assert boosts == [5.0]
+
+
+def test_tuned_lens_affine_direction_and_report_have_known_oracle():
+    lens = RefusalTunedLens(top_k=1)
+    tokenizer = NS(
+        decode=lambda ids: str(ids[0]), encode=lambda text, **kw: [0] if text == "sorry" else []
+    )
+    model = NS(
+        lm_head=NS(weight=torch.eye(2)),
+        model=NS(norm=NS(weight=torch.tensor([2.0, 3.0]), bias=torch.tensor([1.0, -1.0]))),
+    )
+    probe = TunedLensProbe(5, torch.eye(2), torch.tensor([100.0, 100.0]), 0.0)
+    result = lens.analyze_all_layers(
+        {5: torch.tensor([[1.0, 0.0]]), 6: torch.ones(2)}, {5: probe}, model, tokenizer
+    )
+    layer = result.per_layer[5]
+    assert layer.top_promoted == [("0", 3.0)]
+    assert layer.top_suppressed == [("1", -1.0)]
+    assert layer.refusal_compliance_gap == 3.0
+    assert layer.correction_magnitude == 0.0
+    report = lens.format_report(result)
+    assert "Strongest refusal layer: 5" in report
+    assert "Mean refusal-compliance gap: 3.0000" in report
+    assert "Top promoted:" in report
+    assert "Top suppressed:" in report
+    empty = lens.analyze_all_layers({6: torch.ones(2)}, {}, model, tokenizer)
+    assert empty.per_layer == {}
+    assert empty.mean_refusal_compliance_gap == 0.0
+    assert "No layers analyzed." in lens.format_report(empty)
+
+
+def test_tuned_lens_comparison_intersects_layers_and_detects_reversed_ranking():
+    result = NS(
+        per_layer={
+            1: NS(refusal_compliance_gap=1.0),
+            2: NS(refusal_compliance_gap=2.0),
+            3: NS(refusal_compliance_gap=3.0),
+        }
+    )
+    assert RefusalTunedLens.compare_with_logit_lens(result, {1: 1.0}) == 1.0
+    assert (
+        RefusalTunedLens.compare_with_logit_lens(result, {1: 3.0, 2: 2.0, 3: 1.0, 4: 100.0}) == -1.0
+    )
+
+
+def test_spectral_diagonal_oracle_and_degenerate_condition_estimates():
+    certifier = SpectralCertifier()
+    covariance = torch.diag(torch.tensor([0.0, 2.0, 8.0]))
+    result = certifier._eigenvalue_analysis(covariance, bbp_threshold=4.0, mp_upper=2.0)
+    torch.testing.assert_close(result.eigenvalues, torch.tensor([8.0, 2.0, 0.0]))
+    torch.testing.assert_close(
+        covariance @ result.eigenvectors, result.eigenvectors @ torch.diag(result.eigenvalues)
+    )
+    assert result.above_threshold == [0]
+    assert result.in_bulk == [1]
+    assert result.signal_subspace_dim == 1
+    assert certifier._estimate_condition_number(covariance) == 4.0
+    assert certifier._estimate_condition_number(torch.zeros(2, 2)) == 1.0
+    assert certifier._estimate_noise_variance(covariance, n=2, d=3) == 2.0
+    ratio = (1 - (3 / 12) ** 0.5) ** 2 + (3 / 12) ** (1 / 3)
+    assert certifier._estimate_noise_variance(covariance, n=12, d=3) == pytest.approx(2 / ratio)
+    assert certifier._estimate_noise_variance(torch.zeros(2, 2), n=2, d=2) == 1e-10
+    for spectrum in [
+        torch.tensor([]),
+        torch.tensor([1.0]),
+        torch.tensor([1.0, float("nan")]),
+        torch.tensor([-1.0, 2.0]),
+    ]:
+        assert certifier._estimate_condition_number_from_spectrum(spectrum) == 1.0
+    assert certifier._estimate_condition_number_from_spectrum(torch.tensor([1.0, 1e8])) == 1e6
+
+
+def test_spectral_linear_algebra_failure_returns_conservative_fallbacks(monkeypatch):
+    def fail(*args, **kwargs):
+        raise RuntimeError("eigensolver failed")
+
+    monkeypatch.setattr(torch.linalg, "eigvalsh", fail)
+    monkeypatch.setattr(torch.linalg, "eigh", fail)
+    certifier = SpectralCertifier()
+    assert certifier._estimate_noise_variance(torch.eye(2), 2, 2) == 1.0
+    assert certifier._estimate_condition_number(torch.eye(2)) == 1.0
+    result = certifier._eigenvalue_analysis(torch.eye(2), 1.0, 1.0)
+    assert result.signal_subspace_dim == 0
+    assert result.above_threshold == []
+    assert result.in_bulk == []
+    torch.testing.assert_close(result.eigenvalues, torch.zeros(1))

tests/test_conditional_gate_scripts.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from datetime import date
+from datetime import date, timedelta
 from pathlib import Path
 
 from scripts import check_conditional_policy
@@ -29,14 +29,33 @@ def _waiver(gate: str, **overrides) -> dict:
 
 
 def test_committed_conditional_policy_is_complete():
+    policy = json.loads((ROOT / "ci" / "conditional-test-policy.json").read_text())
+    # Structural tests use the declared review date. The CI policy command
+    # separately enforces expiry against the real clock.
+    reviewed = max(date.fromisoformat(waiver["opened"]) for waiver in policy["environment_waivers"])
     assert check_conditional_policy.validate(
         ROOT / "ci" / "conditional-test-policy.json",
         ROOT / "ci" / "test-quality-policy.json",
         ROOT / ".github" / "workflows" / "conditional-tests.yml",
-        today=TODAY,
+        today=reviewed,
     ) == []
 
 
+def test_committed_waivers_fail_closed_after_expiry():
+    policy = json.loads((ROOT / "ci" / "conditional-test-policy.json").read_text())
+    expired = max(date.fromisoformat(waiver["expires"]) for waiver in policy["environment_waivers"])
+    errors = check_conditional_policy.validate(
+        ROOT / "ci" / "conditional-test-policy.json",
+        ROOT / "ci" / "test-quality-policy.json",
+        ROOT / ".github" / "workflows" / "conditional-tests.yml",
+        today=expired + timedelta(days=1),
+    )
+    assert errors == [
+        f"environment waiver {waiver['gate']} expired on {waiver['expires']}"
+        for waiver in policy["environment_waivers"]
+    ]
+
+
 def test_cuda_job_replaces_locked_cpu_torch_with_same_version_cuda_build():
     workflow = (ROOT / ".github" / "workflows" / "conditional-tests.yml").read_text()
     cuda_job = workflow.split("  cuda:\n", maxsplit=1)[1].split("  mps:\n", maxsplit=1)[0]
@@ -74,6 +93,7 @@ def test_jetson_job_is_manual_physical_trusted_and_retains_sanitized_evidence():
 
 def test_policy_rejects_unknown_cpu_exclusion_gate(tmp_path):
     policy = json.loads((ROOT / "ci" / "conditional-test-policy.json").read_text())
+    policy["environment_waivers"] = []
     quality = {
         "mature_cpu_scope": {
             "exclusions": [{"path": "obliteratus/device.py", "conditional_gate": "missing"}]
@@ -87,6 +107,7 @@ def test_policy_rejects_unknown_cpu_exclusion_gate(tmp_path):
         policy_path,
         quality_path,
         ROOT / ".github" / "workflows" / "conditional-tests.yml",
+        today=TODAY,
     )
     assert errors == ["CPU exclusion obliteratus/device.py references unknown gate missing"]
 

tests/test_distributed_validation_edges.py

@@ -0,0 +1,292 @@
+"""Deterministic failure and system-probe contracts for distributed admission."""
+
+from dataclasses import replace
+import ctypes
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from obliteratus.distributed import preflight
+from obliteratus.distributed.config import DistributedPreflightConfig
+from obliteratus.distributed.contracts import ContractError, RuntimeContractError
+from tests.test_distributed_launcher import _profile
+
+
+pytestmark = pytest.mark.cpu
+
+
+@pytest.mark.parametrize(
+    ("section", "key", "value", "message"),
+    [
+        ("run", "world_size", True, "integer"),
+        ("run", "world_size", 1, "between"),
+        ("run", "run_id", "", "bounded non-empty"),
+        ("run", "run_id", "bad", "invalid format"),
+        ("software", "python", "\ud800", "valid UTF-8"),
+        ("source", "path", "relative", "absolute local"),
+        ("network", "allowed_master_cidrs", [], "non-empty bounded"),
+        ("network", "allowed_master_cidrs", ["bad"], "invalid network"),
+        ("network", "allowed_master_cidrs", ["8.8.8.0/24"], "private networks"),
+        ("network", "allowed_master_cidrs", ["10.0.0.0/8"] * 2, "duplicates"),
+        ("topology", "dimension_divisors", [], "non-empty bounded"),
+        ("execution", "allowed_environment_keys", ["PATH", "PATH"], "unique bounded"),
+        ("execution", "allowed_environment_keys", ["HF_TOKEN"], "secret or proxy"),
+    ],
+)
+def test_profile_rejects_invalid_field_values(tmp_path, section, key, value, message):
+    _profile(tmp_path)
+    path = tmp_path / "profile.json"
+    payload = json.loads(path.read_text())
+    payload[section][key] = value
+    path.write_text(json.dumps(payload))
+    with pytest.raises(ContractError, match=message):
+        DistributedPreflightConfig.from_file(path)
+
+
+@pytest.mark.parametrize(
+    ("raw", "message"),
+    [
+        (b"", "regular file"),
+        (b"\xff", "strict UTF-8 JSON"),
+        (b"{", "strict UTF-8 JSON"),
+        (b"[]", "must be an object"),
+        (b'{"schema_version":1,"schema_version":1}', "duplicate"),
+        (b"{}", "required profile section"),
+    ],
+)
+def test_profile_rejects_invalid_serialization(tmp_path, raw, message):
+    path = tmp_path / "profile.json"
+    path.write_bytes(raw)
+    with pytest.raises(ContractError, match=message):
+        DistributedPreflightConfig.from_file(path)
+
+
+@pytest.mark.parametrize(
+    ("change", "message"),
+    [
+        ({"rendezvous_id": "1" * 32}, "distinct"),
+        ({"local_world_size": 3}, "divisible"),
+        ({"tensor_parallel_size": 4}, "equal world_size"),
+        ({"max_source_bytes": 1}, "cannot exceed"),
+        ({"coordinator_rank": 2}, "smaller than"),
+        ({"device_kind": "mps"}, "cuda or cpu"),
+        ({"evidence_tier": "protocol_cpu"}, "does not match"),
+        ({"local_files_only": False}, "remain true"),
+    ],
+)
+def test_direct_profile_replacement_rechecks_admission(tmp_path, change, message):
+    with pytest.raises(ContractError, match=message):
+        replace(_profile(tmp_path), **change).validate()
+
+
+@pytest.mark.parametrize("kind", ["missing", "symlink", "directory"])
+def test_profile_requires_regular_file(tmp_path, kind):
+    target = tmp_path / "target"
+    if kind == "symlink":
+        target.symlink_to(tmp_path / "absent")
+    elif kind == "directory":
+        target.mkdir()
+    with pytest.raises(ContractError, match="regular"):
+        DistributedPreflightConfig.from_file(target)
+
+
+@pytest.mark.parametrize("kind", ["empty", "oversized", "invalid_utf8", "missing"])
+def test_checkout_identity_file_is_bounded_and_utf8(tmp_path, kind):
+    path = tmp_path / "HEAD"
+    if kind != "missing":
+        path.write_bytes({"empty": b"", "oversized": b"a" * 4097, "invalid_utf8": b"\xff"}[kind])
+    with pytest.raises(ContractError, match="identity"):
+        preflight._bounded_text_file(path)
+
+
+@pytest.mark.parametrize(
+    "head", ["garbage", "ref: outside", "ref: refs/../secret", "ref: refs/heads/bad\\name"]
+)
+def test_checkout_rejects_unsafe_head_references(tmp_path, head):
+    git = tmp_path / ".git"
+    git.mkdir()
+    (git / "HEAD").write_text(head)
+    with pytest.raises(ContractError, match="invalid format"):
+        preflight.checkout_commit(tmp_path)
+
+
+def test_checkout_resolves_worktree_common_packed_reference(tmp_path):
+    worktree = tmp_path / "worktree"
+    worktree.mkdir()
+    metadata = tmp_path / "metadata"
+    metadata.mkdir()
+    common = tmp_path / "common"
+    common.mkdir()
+    (worktree / ".git").write_text("gitdir: ../metadata\n")
+    (metadata / "HEAD").write_text("ref: refs/heads/main\n")
+    (metadata / "commondir").write_text("../common\n")
+    commit = "a" * 40
+    (common / "packed-refs").write_text(f"# packed refs\n{commit} refs/heads/main\n^{commit}\n")
+    assert preflight.checkout_commit(worktree) == commit
+    (common / "packed-refs").write_text(f"{commit} refs/heads/other\n")
+    with pytest.raises(ContractError, match="reference is unavailable"):
+        preflight.checkout_commit(worktree)
+
+
+@pytest.mark.parametrize("mode", ["init_failure", "version_failure", "unavailable", "success"])
+def test_driver_version_probe_handles_library_results(monkeypatch, mode):
+    def version(pointer):
+        ctypes.cast(pointer, ctypes.POINTER(ctypes.c_int))[0] = 12040
+        return int(mode == "version_failure")
+
+    def library(name):
+        assert name == "libcuda.so.1"
+        if mode == "unavailable":
+            raise OSError("no driver installed")
+        return SimpleNamespace(
+            cuInit=lambda flags: int(mode == "init_failure"), cuDriverGetVersion=version
+        )
+
+    monkeypatch.setattr(preflight.ctypes, "CDLL", library)
+    assert preflight._cuda_driver_version() == ("12040" if mode == "success" else "unavailable")
+
+
+@pytest.mark.parametrize(
+    "version, expected", [((2, 28, 3), "2.28.3"), (22803, "22803"), (None, "unavailable")]
+)
+def test_nccl_probe_normalizes_version_and_missing_backend(monkeypatch, version, expected):
+    def probe():
+        if version is None:
+            raise RuntimeError("backend unavailable")
+        return version
+
+    monkeypatch.setattr(preflight.torch.cuda.nccl, "version", probe)
+    assert preflight._nccl_version() == expected
+
+
+@pytest.fixture
+def system_probe_inputs(tmp_path, monkeypatch):
+    config = _profile(tmp_path)
+    source = preflight.SourceIdentity("a" * 64, "b" * 64, "c" * 64, 3, 256)
+    monkeypatch.setattr(preflight, "inspect_source", lambda *args, **kwargs: source)
+    monkeypatch.setattr(preflight, "validate_network_interface", lambda *args, **kwargs: None)
+    monkeypatch.setattr(preflight, "_host_memory", lambda: (16000, 8000))
+    monkeypatch.setattr(preflight.shutil, "disk_usage", lambda path: SimpleNamespace(free=32000))
+    monkeypatch.setattr(preflight, "_nccl_version", lambda: "2.28.3")
+    monkeypatch.setattr(preflight, "_cuda_driver_version", lambda: "12040")
+    monkeypatch.setattr(preflight, "checkout_commit", lambda: "d" * 40)
+    monkeypatch.setattr(preflight, "checkout_code_digest", lambda: "e" * 64)
+    monkeypatch.setattr(preflight, "storage_mount_digest", lambda path: "f" * 64)
+    monkeypatch.setattr(preflight.platform, "node", lambda: "test-host")
+    monkeypatch.setattr(preflight.platform, "machine", lambda: "x86_64")
+    return config, SimpleNamespace(rank=0, local_rank=1), source
+
+
+@pytest.mark.parametrize("kind", ["cpu", "cuda"])
+def test_system_probe_records_measured_resources(system_probe_inputs, monkeypatch, kind):
+    config, launch, source = system_probe_inputs
+    selected = []
+    monkeypatch.setattr(preflight.torch.cuda, "is_available", lambda: True)
+    monkeypatch.setattr(preflight.torch.cuda, "set_device", selected.append)
+    monkeypatch.setattr(
+        preflight.torch.cuda,
+        "get_device_properties",
+        lambda rank: SimpleNamespace(uuid="GPU-test", name="Test GPU", major=8, minor=6),
+    )
+    monkeypatch.setattr(preflight.torch.cuda, "mem_get_info", lambda rank: (4000, 12000))
+    result = preflight.SystemProbes().collect(replace(config, device_kind=kind), launch)
+    assert result.source == source
+    assert (
+        result.total_host_memory_bytes,
+        result.free_host_memory_bytes,
+        result.free_staging_bytes,
+    ) == (16000, 8000, 32000)
+    assert result.commit_sha == "d" * 40
+    assert result.code_digest == "e" * 64
+    assert result.storage_identity == "f" * 64
+    assert dict(result.software_versions)["driver"] == "12040"
+    if kind == "cuda":
+        assert selected == [1]
+        assert (result.device_identity, result.device_name, result.compute_capability) == (
+            "GPU-test",
+            "Test GPU",
+            "8.6",
+        )
+        assert (result.total_device_memory_bytes, result.free_device_memory_bytes) == (12000, 4000)
+    else:
+        assert selected == []
+        assert result.device_identity == "cpu:test-host:x86_64:1"
+        assert (result.total_device_memory_bytes, result.free_device_memory_bytes) == (16000, 8000)
+
+
+@pytest.mark.parametrize("failure", ["unavailable", "no_uuid", "checkout", "staging"])
+def test_system_probe_fails_closed_on_missing_identity(system_probe_inputs, monkeypatch, failure):
+    config, launch, _ = system_probe_inputs
+    monkeypatch.setattr(preflight.torch.cuda, "is_available", lambda: failure != "unavailable")
+    monkeypatch.setattr(preflight.torch.cuda, "set_device", lambda rank: None)
+    monkeypatch.setattr(
+        preflight.torch.cuda, "get_device_properties", lambda rank: SimpleNamespace()
+    )
+    if failure == "checkout":
+        config = replace(config, device_kind="cpu")
+
+        def missing_commit():
+            raise ContractError("unavailable")
+
+        monkeypatch.setattr(preflight, "checkout_commit", missing_commit)
+    if failure == "staging":
+        config = replace(config, staging_path=config.staging_path / "missing")
+    with pytest.raises(RuntimeContractError) as error:
+        preflight.SystemProbes().collect(config, launch)
+    assert error.value.code == (
+        "LMS_STORAGE_PROFILE_MISMATCH" if failure == "staging" else "LMS_RUNTIME_PROFILE_MISMATCH"
+    )
+
+
+@pytest.mark.parametrize(
+    "kind, code",
+    [
+        ("writable", "LMS_SOURCE_BOUNDARY_VIOLATION"),
+        ("hardlink", "LMS_SOURCE_BOUNDARY_VIOLATION"),
+        ("oversized", "LMS_RESOURCE_ADMISSION_DENIED"),
+        ("expired", "LMS_STAGE_TIMEOUT"),
+        ("missing", "LMS_SOURCE_BOUNDARY_VIOLATION"),
+        ("replaced", "LMS_SOURCE_CHANGED"),
+    ],
+)
+def test_source_hash_rejects_unsafe_or_changed_file(tmp_path, monkeypatch, kind, code):
+    path = tmp_path / "source.json"
+    path.write_bytes(b"original")
+    path.chmod(0o444)
+    if kind == "writable":
+        path.chmod(0o644)
+    elif kind == "hardlink":
+        (tmp_path / "alias.json").hardlink_to(path)
+    elif kind == "missing":
+        path.unlink()
+    elif kind == "replaced":
+        original_read = preflight.os.read
+        replaced = False
+
+        def replace_after_read(descriptor, size):
+            nonlocal replaced
+            result = original_read(descriptor, size)
+            if not replaced:
+                replacement = tmp_path / "replacement.json"
+                replacement.write_bytes(b"modified")
+                replacement.chmod(0o444)
+                replacement.replace(path)
+                replaced = True
+            return result
+
+        monkeypatch.setattr(preflight.os, "read", replace_after_read)
+    monkeypatch.setattr(preflight.time, "monotonic", lambda: 10.0)
+    with pytest.raises(RuntimeContractError) as error:
+        preflight._hash_file(
+            path,
+            deadline=9.0 if kind == "expired" else 11.0,
+            max_bytes=1 if kind == "oversized" else 100,
+        )
+    assert error.value.code == code
+
+
+def test_host_memory_converts_system_page_counts_to_bytes(monkeypatch):
+    values = {"SC_PAGE_SIZE": 4096, "SC_PHYS_PAGES": 1000, "SC_AVPHYS_PAGES": 250}
+    monkeypatch.setattr(preflight.os, "sysconf", values.__getitem__)
+    assert preflight._host_memory() == (4_096_000, 1_024_000)

tests/test_notebook_contract.py

@@ -4,9 +4,12 @@ from __future__ import annotations
 
 import ast
 import json
+import os
 from pathlib import Path
 from types import SimpleNamespace
 
+import pytest
+
 
 def test_abliterate_notebook_stage_callback_uses_stage_result_contract(capsys):
     """The Colab callback should consume StageResult.stage and .message."""
@@ -53,7 +56,10 @@ def _run_harness(monkeypatch, *, token=None, failure=None):
     import sys
     from unittest.mock import Mock
 
-    monkeypatch.delenv("HF_TOKEN", raising=False)
+    # Record even an initially absent key before notebook code writes directly
+    # to os.environ, so teardown restores the caller's original environment.
+    monkeypatch.setenv("HF_TOKEN", "")
+    monkeypatch.delenv("HF_TOKEN")
     events = []
 
     def download(**kwargs):
@@ -81,6 +87,31 @@ def _run_harness(monkeypatch, *, token=None, failure=None):
     return namespace, events, hub, secret
 
 
+@pytest.mark.parametrize("original_token", [None, "fake-caller-token"])
+@pytest.mark.parametrize("access_denied", [False, True])
+def test_notebook_harness_restores_token_after_execution(monkeypatch, original_token, access_denied):
+    if original_token is None:
+        monkeypatch.delenv("HF_TOKEN", raising=False)
+    else:
+        monkeypatch.setenv("HF_TOKEN", original_token)
+
+    with monkeypatch.context() as notebook_patch:
+        namespace, _, _, _ = _run_harness(
+            notebook_patch,
+            token="fake-notebook-token",
+            failure=PermissionError("denied") if access_denied else None,
+        )
+        source = _notebook_code("def check_model_access")
+        if access_denied:
+            with pytest.raises(RuntimeError, match="Model access"):
+                exec(source, namespace)
+        else:
+            exec(source, namespace)
+            assert os.environ["HF_TOKEN"] == "fake-notebook-token"
+
+    assert os.environ.get("HF_TOKEN") == original_token
+
+
 def test_default_gated_model_stops_before_pipeline_without_access(monkeypatch, capsys):
     import traceback
     import pytest