srr84/agent-data-layer

1"""Self-checks for the two JOIN-1 CI guards (AEP §4/§6).2 3Each guard gets a POSITIVE CONTROL (a deliberately-planted violation must be4DETECTED) plus a NEGATIVE CONTROL (the real production tree is clean / exit 0).5The positive control is what proves the guard can actually FAIL — a guard that6only ever returns "clean" is not a guard. The planted violations live in tmp7fixtures (or a tmp src tree the guard is pointed at); the real tree is never8mutated.9 10New reader -> WALKTHROUGH.md (authored at a later milestone).11 12```text13spec traceability (audit map — safe to skip)14AEP §4 (no-stub-reachable: @stub/Stub reachable on the production path -> FAIL),15        §6 (INV-C1 store-access/no-new-ingress + open-param + typed-param-as-selector16        -> FAIL; both HARD CI gates at M0/JOIN-1), §8 (JOIN-1 machine bar)17Validation §1 (deterministic-core unit layer; no model/server)18code-discipline §3 (a positive control: a planted violation is DETECTED; then the19        real tree is confirmed clean)20```21"""22 23from __future__ import annotations24 25import importlib.util26import subprocess27import sys28import textwrap29from pathlib import Path30from types import ModuleType31from typing import Callable, cast32 33_REPO_ROOT = Path(__file__).resolve().parents[1]34_SCRIPTS = _REPO_ROOT / "scripts"35_PYTHON = sys.executable36 37 38def _load_script(name: str) -> ModuleType:39    """Import a scripts/*.py guard as a module (they are not on a package path)."""40    path = _SCRIPTS / name41    spec = importlib.util.spec_from_file_location(path.stem, path)42    assert spec is not None and spec.loader is not None43    module = importlib.util.module_from_spec(spec)44    spec.loader.exec_module(module)45    return module46 47 48def _run_guard(script: str) -> subprocess.CompletedProcess[str]:49    """Run a guard script as a subprocess and return its completed process."""50    return subprocess.run(51        [_PYTHON, str(_SCRIPTS / script)],52        capture_output=True,53        text=True,54    )55 56 57def _bindings_fn(guard: ModuleType) -> Callable[[ModuleType], list[str]]:58    """The guard's ``_module_stub_bindings`` as a typed callable (it is dynamically loaded)."""59    return cast(Callable[[ModuleType], list[str]], guard._module_stub_bindings)60 61 62# ---------------------------------------------------------------------------63# Guard 1: no-stub-reachable64# ---------------------------------------------------------------------------65 66 67def test_no_stub_reachable_clean_on_real_tree() -> None:68    """NEGATIVE control: the real production tree has no reachable stub -> exit 0."""69    result = _run_guard("check_no_stub_reachable.py")70    assert result.returncode == 0, result.stderr71 72 73def test_no_stub_reachable_detects_planted_stub() -> None:74    """POSITIVE control: a @stub-tagged symbol bound in a module IS detected.75 76    Builds a throwaway module that binds a @stub callable, then runs the guard's77    own symbol scanner over it. The scanner must report the planted stub — proving78    the marker walk actually flags a stub, not that it always returns clean. This is79    also the regression for the ``__module__``-ownership false negative: the wrapper80    carries the convention module's ``__module__`` yet must still be flagged where81    it is BOUND.82    """83    guard = _load_script("check_no_stub_reachable.py")84    bindings = _bindings_fn(guard)85    from agent_data_layer.contracts.stub import stub86 87    def _impl() -> None:88        return None89 90    not_yet_built = stub(_impl)91 92    planted = ModuleType("planted_production_module")93    setattr(planted, "not_yet_built", not_yet_built)94 95    found = bindings(planted)96    assert "not_yet_built" in found97 98 99def test_no_stub_reachable_detects_planted_stub_subclass() -> None:100    """POSITIVE control: a Stub SUBCLASS bound in a module IS detected."""101    guard = _load_script("check_no_stub_reachable.py")102    bindings = _bindings_fn(guard)103    from agent_data_layer.contracts.stub import Stub104 105    class _NotBuilt(Stub):106        pass107 108    planted = ModuleType("planted_module_with_subclass")109    setattr(planted, "NotBuilt", _NotBuilt)110 111    assert "NotBuilt" in bindings(planted)112 113 114def test_no_stub_reachable_ignores_non_stub_symbol() -> None:115    """A plain (non-stub) symbol is NOT flagged (no false positive)."""116    guard = _load_script("check_no_stub_reachable.py")117    bindings = _bindings_fn(guard)118 119    def ordinary() -> None:120        return None121 122    plain = ModuleType("plain_module")123    setattr(plain, "ordinary", ordinary)124    assert bindings(plain) == []125 126 127def test_no_stub_reachable_ignores_reimported_marker_base() -> None:128    """The Stub marker BASE re-imported into a module is NOT a live stub (no false +)."""129    guard = _load_script("check_no_stub_reachable.py")130    bindings = _bindings_fn(guard)131    from agent_data_layer.contracts.stub import Stub132 133    reexporter = ModuleType("module_that_imports_the_base")134    setattr(reexporter, "Stub", Stub)135    assert bindings(reexporter) == []136 137 138# ---------------------------------------------------------------------------139# Guard 2: INV-C1 data-flow140# ---------------------------------------------------------------------------141 142 143def test_inv_c1_clean_on_real_tree() -> None:144    """NEGATIVE control: the real production tree satisfies INV-C1 -> exit 0."""145    result = _run_guard("check_inv_c1_dataflow.py")146    assert result.returncode == 0, result.stderr147 148 149def _check_inv_c1_source(tmp_path: Path, body: str) -> list[str]:150    """Write ``body`` to a tmp .py under a fake non-allowlisted module and check it.151 152    Points the guard's SRC root at the tmp tree so ``_rel_module`` resolves the file153    to ``executor/planted`` — a module NOT on the store-access allowlist, so any154    planted violation is in scope.155    """156    guard = _load_script("check_inv_c1_dataflow.py")157    src = tmp_path / "src" / "agent_data_layer" / "executor"158    src.mkdir(parents=True)159    planted = src / "planted.py"160    planted.write_text(textwrap.dedent(body), encoding="utf-8")161    setattr(guard, "_SRC", tmp_path / "src" / "agent_data_layer")162    check_file = cast(Callable[[Path], list[str]], guard.check_file)163    return check_file(planted)164 165 166def test_inv_c1_detects_index_read_outside_allowlist(tmp_path: Path) -> None:167    """POSITIVE control: an index-field read in a non-allowlisted module IS detected."""168    problems = _check_inv_c1_source(169        tmp_path,170        """171        def leak(idx: object) -> object:172            return idx.all_events  # new ingress outside the allowlist173        """,174    )175    assert any("new ingress" in p for p in problems), problems176 177 178def test_inv_c1_detects_typed_param_as_selector(tmp_path: Path) -> None:179    """POSITIVE control: a free-form selector on a store-derived value IS detected."""180    problems = _check_inv_c1_source(181        tmp_path,182        """183        def leak(event: object, needle: str) -> bool:184            return event.shelf_state.startswith(needle)  # selector on a store field185        """,186    )187    assert any("exact-match key" in p for p in problems), problems188 189 190def test_inv_c1_detects_in_membership_against_store_value(tmp_path: Path) -> None:191    """POSITIVE control: an `in` probe against a store-derived value IS detected."""192    problems = _check_inv_c1_source(193        tmp_path,194        """195        def leak(needle: str, idx: object) -> bool:196            return needle in idx.all_events  # membership selector against the index197        """,198    )199    assert any("'in' membership" in p for p in problems), problems200 201 202def test_inv_c1_detects_open_paramspec(tmp_path: Path) -> None:203    """POSITIVE control: a ParamSpec widened to an open type IS detected."""204    problems = _check_inv_c1_source(205        tmp_path,206        """207        def make() -> object:208            return ParamSpec(name="q", type="FreeText", enum_domain=None, required=True)209        """,210    )211    assert any("closed ParamType set" in p for p in problems), problems212 213 214def test_inv_c1_ignores_id_format_check(tmp_path: Path) -> None:215    """NEGATIVE: an id-format .startswith on a plain string is NOT a store selector."""216    problems = _check_inv_c1_source(217        tmp_path,218        """219        def validate_id(store_id: str) -> bool:220            return store_id.startswith("st-")  # format check on a plain string, allowed221        """,222    )223    assert problems == [], problems224