feat(s51): MAF-free transitive import-graph probe

Gate: pytest tests/test_hitl_loadbearing.py tests/test_okf.py → 24 passed.
This commit is contained in:
Kjell Tore Guttormsen 2026-07-15 19:41:42 +02:00
commit eaca95a325

View file

@ -12,6 +12,7 @@ Seams pinned here:
from __future__ import annotations
import ast
import json
from pathlib import Path
@ -186,3 +187,87 @@ def test_inbox_idset_skips_malformed_features(tmp_path: Path) -> None:
)
assert [p.run_id for p in hitl.pending(str(outbox), str(inbox))] == ["run-1"]
# --- transitive import-graph MAF-freedom probe (Step 5) -------------------------------------------
_SRC_DIR = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser"
_MAF_ROOTS = {"agent_framework", "mcp"}
def _first_party_submodule_edges(tree: ast.Module) -> set[str]:
"""Collect ``portfolio_optimiser.<name>`` submodule import edges from a parsed module —
``from portfolio_optimiser.<name> import ...`` and ``import portfolio_optimiser.<name>``.
DISCOUNTED by design (Assumption 3): the package top ``from portfolio_optimiser import X`` (X is
a re-exported symbol from the eager ``__init__`` ``__all__`` set, e.g. ``run_project``), and
``__init__`` itself never traversed, else every walk would reach ``run`` ``agent_framework``
and the probe would be unsatisfiable for a clean module. Known limitation: a hypothetical
``from portfolio_optimiser import verdicts`` (submodule via the package) is indistinguishable from
a symbol import here and would be discounted optional future hardening; the specced detach
(``from portfolio_optimiser.verdicts import ...``) is a ``len>=2`` edge and IS caught."""
edges: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
parts = (node.module or "").split(".")
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
edges.add(parts[1])
elif isinstance(node, ast.Import):
for alias in node.names:
parts = alias.name.split(".")
if len(parts) >= 2 and parts[0] == "portfolio_optimiser" and parts[1] != "__init__":
edges.add(parts[1])
return edges
def _imports_maf(tree: ast.Module) -> bool:
for node in ast.walk(tree):
if isinstance(node, ast.Import):
if any(a.name.split(".")[0] in _MAF_ROOTS for a in node.names):
return True
elif isinstance(node, ast.ImportFrom):
if (node.module or "").split(".")[0] in _MAF_ROOTS:
return True
return False
def test_hitl_transitive_import_graph_is_maf_free() -> None:
"""LOAD-BEARING: no module reachable from ``hitl.py``'s OWN first-party import edges imports
``agent_framework``/``mcp``. A static AST BFS (never traversing ``__init__.py`` see the discount
note) NOT a ``sys.modules`` probe, which is unsatisfiable because ``__init__.py:11`` eagerly
loads ``run`` MAF. hitl re-implements the outbox shape + admits-logic inline, so its steady
state is an EMPTY first-party edge set. Detach point (documented): add
``from portfolio_optimiser.verdicts import load_verdicts_from_dir`` to ``hitl.py`` the walk
reaches ``verdicts.py`` (``agent_framework`` at line 29) RED. A clean hitl GREEN."""
seen: set[str] = set()
queue = list(_first_party_submodule_edges(ast.parse((_SRC_DIR / "hitl.py").read_text("utf-8"))))
maf_bearing: list[str] = []
while queue:
mod = queue.pop()
if mod in seen:
continue
seen.add(mod)
mod_file = _SRC_DIR / f"{mod}.py"
if not mod_file.is_file():
continue
tree = ast.parse(mod_file.read_text("utf-8"))
if _imports_maf(tree):
maf_bearing.append(mod)
queue.extend(_first_party_submodule_edges(tree) - seen)
assert maf_bearing == [], (
f"hitl's transitive first-party import graph reaches MAF-bearing module(s): {sorted(maf_bearing)}"
)
def test_hitl_source_has_no_network_import() -> None:
"""hitl.py imports no network library (``socket``/``urllib``/``http``/``requests``/``httpx``) —
the inspection tool reads local folders only, never egresses (målbilde §1 no-silent-egress)."""
forbidden = {"socket", "urllib", "http", "requests", "httpx"}
tree = ast.parse((_SRC_DIR / "hitl.py").read_text("utf-8"))
imported: set[str] = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
imported |= {a.name.split(".")[0] for a in node.names}
elif isinstance(node, ast.ImportFrom):
imported.add((node.module or "").split(".")[0])
assert imported & forbidden == set(), f"hitl must not import network libs: {imported & forbidden}"