fix: fail closed on malformed observation/control JSON and purge stale plugin imports

Four fail-closed and correctness bugs identified during codebase audit:

1. _observation_pair_failures silently passed non-dict JSON by returning
   an empty failure list. Now reports a malformed-JSON failure.

2. _usage in heldout_runner accepted non-dict stats (e.g. null) without
   raising, potentially proceeding with incomplete result data. Now
   validates stats is a dict before access.

3. load_registry cache had no invalidation path, making test fixtures that
   mutate the registry unreliable. Added invalidate_registry_cache().

4. install_source_only_importer inserted a SourceOnlyFinder but did not
   purge already-imported modules from sys.modules, so stale bytecode could
   shadow the finder. Now purges the target package from sys.modules first.

Includes formal regression tests for all four fixes.

Author: morluto

benchmarks/tooling/benchmark_contracts.py

@@ -223,7 +223,7 @@ def _observation_pair_failures() -> list[str]:
     treatment = _read_json(treatment_path)
     control = _read_json(control_path)
     if not isinstance(treatment, dict) or not isinstance(control, dict):
-        return []
+        return ["observation/control pair contains malformed (non-object) JSON"]
 
     def normalized(value: dict[str, Any]) -> dict[str, Any]:
         copy: dict[str, Any] = json.loads(json.dumps(value))

benchmarks/tooling/harbor_suite.py

@@ -553,6 +553,11 @@ def load_registry(path: Path = REGISTRY_PATH) -> tuple[Suite, ...]:
     return result
 
 
+def invalidate_registry_cache() -> None:
+    """Drop cached registry entries so subsequent load_registry calls re-parse."""
+    _load_registry_cache.clear()
+
+
 def get_suite(dataset: str, *, path: Path = REGISTRY_PATH) -> Suite:
     short = dataset.removeprefix(DATASET_PREFIX)
     for suite in load_registry(path):
@@ -1085,6 +1090,7 @@ def report_ok(message: str) -> None:
     "iter_task_dirs",
     "load_environment_profiles",
     "load_registry",
+    "invalidate_registry_cache",
     "report_failures",
     "report_ok",
     "select_task_refs",

benchmarks/tooling/heldout_runner.py

@@ -97,7 +97,9 @@ def _usage(result_path: Path) -> tuple[int, float]:
     if not isinstance(result, dict):
         raise HarborSuiteError("Harbor result must be an object")
     stats = result.get("stats")
-    if isinstance(stats, dict) and any(
+    if not isinstance(stats, dict):
+        raise HarborSuiteError("Harbor result stats must be an object")
+    if any(
         stats.get(key, 0)
         for key in (
             "n_errored_trials",

src/jacobian/implementation.py

@@ -76,7 +76,18 @@ def install_source_only_importer(entrypoint: str) -> None:
     """Force the entrypoint package's future imports to compile measured source."""
 
     module_name, _ = split_entrypoint(entrypoint)
-    sys.meta_path.insert(0, _SourceOnlyFinder(module_name.split(".", 1)[0]))
+    top_level = module_name.split(".", 1)[0]
+    # Purge any pre-imported modules from the target package so the
+    # SourceOnlyFinder recompiles them from measured source.  Without
+    # this, already-imported modules in sys.modules would shadow the
+    # finder and serve potentially stale bytecode.
+    stale = [
+        name for name in list(sys.modules)
+        if name == top_level or name.startswith(top_level + ".")
+    ]
+    for name in stale:
+        del sys.modules[name]
+    sys.meta_path.insert(0, _SourceOnlyFinder(top_level))
 
 
 def split_entrypoint(entrypoint: str) -> tuple[str, str]:

tests/unit/tooling/test_audit_fixes.py

@@ -0,0 +1,111 @@
+"""Formal tests for the audit-confirmed bugs and their fixes.
+
+These tests serve as formal validation that:
+1. _observation_pair_failures fails closed on malformed JSON (Bug 1a)
+2. _usage rejects non-dict stats (Bug 1b)
+3. _load_registry_cache provides invalidation (Bug 3a)
+4. install_source_only_importer purges sys.modules (Bug TOCTOU)
+"""
+import json
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+# Bug 1a: _observation_pair_failures should fail closed on non-dict JSON
+def test_observation_pair_failures_fails_closed_on_non_dict(tmp_path):
+    """Formally prove that malformed observation JSON is not silently accepted."""
+    from benchmarks.tooling import benchmark_contracts
+
+    # When treatment/control JSON is a valid JSON but not a dict (e.g., a list),
+    # _observation_pair_failures should return failures, not empty list.
+    treatment_path = Path(benchmark_contracts.BENCHMARKS) / "datasets" / "mathematical-benchmarks-v1" / "jobs" / "jacobian-observation.json"
+    control_path = Path(benchmark_contracts.BENCHMARKS) / "config" / "mathematical-benchmarks-v1-control.json"
+
+    # Mock _read_json to return non-dict values
+    original_read_json = benchmark_contracts._read_json
+
+    try:
+        # Test: JSON array instead of object
+        def mock_read_json_array(path):
+            return []
+        benchmark_contracts._read_json = mock_read_json_array
+        failures = benchmark_contracts._observation_pair_failures()
+        assert len(failures) > 0, "Bug 1a: malformed JSON should not silently pass"
+        assert "malformed" in failures[0].lower()
+
+        # Test: JSON null
+        def mock_read_json_null(path):
+            return None
+        benchmark_contracts._read_json = mock_read_json_null
+        failures = benchmark_contracts._observation_pair_failures()
+        assert len(failures) > 0, "Bug 1a: null JSON should not silently pass"
+    finally:
+        benchmark_contracts._read_json = original_read_json
+
+
+# Bug 1b: _usage should reject non-dict stats
+def test_usage_rejects_non_dict_stats():
+    """Formally prove that non-dict stats is rejected."""
+    from benchmarks.tooling import heldout_runner
+
+    # Create a temporary result file with stats as null
+    import tempfile
+    with tempfile.NamedTemporaryFile(suffix=".json", mode="w", delete=False) as f:
+        json.dump({"stats": None}, f)
+        path = Path(f.name)
+
+    try:
+        with pytest.raises(Exception) as exc_info:
+            heldout_runner._usage(path)
+        assert "stats" in str(exc_info.value).lower()
+    finally:
+        path.unlink(missing_ok=True)
+
+
+# Bug 3a: Registry cache should be invalidatable
+def test_registry_cache_invalidation():
+    """Formally prove that cache invalidation works."""
+    from benchmarks.tooling import harbor_suite
+
+    assert hasattr(harbor_suite, "invalidate_registry_cache"), "invalidate_registry_cache should exist"
+    harbor_suite.invalidate_registry_cache()
+    assert harbor_suite._load_registry_cache == {}, "Cache should be empty after invalidation"
+
+
+# Bug TOCTOU: install_source_only_importer should purge sys.modules
+def test_source_only_importer_purges_sys_modules():
+    """Formally prove that pre-imported modules are purged."""
+    import importlib
+    from jacobian.implementation import install_source_only_importer
+
+    # Create a fake module in sys.modules that looks like the target package
+    fake_module = type(sys)("fake_test_package")
+    fake_module.some_value = "stale"
+    sys.modules["fake_test_package"] = fake_module
+    sys.modules["fake_test_package.helper"] = type(sys)("fake_test_package.helper")
+
+    try:
+        install_source_only_importer("fake_test_package:main")
+        assert "fake_test_package" not in sys.modules, "Pre-imported module should be purged"
+        assert "fake_test_package.helper" not in sys.modules, "Pre-imported submodule should be purged"
+    finally:
+        sys.modules.pop("fake_test_package", None)
+        sys.modules.pop("fake_test_package.helper", None)
+        # Clean up meta_path
+        from jacobian.implementation import _SourceOnlyFinder
+        sys.meta_path = [f for f in sys.meta_path if not isinstance(f, _SourceOnlyFinder)]
+
+
+if __name__ == "__main__":
+    test_observation_pair_failures_fails_closed_on_non_dict(None)
+    print("Bug 1a test passed")
+    test_usage_rejects_non_dict_stats()
+    print("Bug 1b test passed")
+    test_registry_cache_invalidation()
+    print("Bug 3a test passed")
+    test_source_only_importer_purges_sys_modules()
+    print("TOCTOU test passed")
+    print("\nAll audit fix tests passed!")