Repair what the ratchet counts with NRA plans; type imports by when they run

pyproject.toml

@@ -42,6 +42,7 @@ dev = [
     "ruff==0.16.0",
     "mypy==2.3.0",
     "uv==0.13.1",
+    "nominal-refactor-advisor @ git+https://github.com/OpenHCSDev/NominalRefactorAdvisor@main",
 ]
 
 [project.urls]

tests/guards/test_debt_ratchet.py

@@ -17,7 +17,7 @@
 from fixture_git import IDENTITY
 from process_tree_fixture import TREE, identities
 
-from tools.debt_ratchet import (
+from debt_ratchet import (
     AUDIT_SCRIPTS,
     BannedVocabulary,
     BroadExceptDefault,

tests/test_carry.py

@@ -9,7 +9,7 @@
 from pathlib import Path
 
 import pytest
-from tools.debt_ratchet import UncarriedRecordFormat, git
+from debt_ratchet import UncarriedRecordFormat, git
 
 from agent_comms.cli_commands import StopCliCommand, TargetBatchResult
 from agent_comms.goal_attempts import GoalAttemptStore

tests/test_module_layers_loads.py

@@ -0,0 +1,46 @@
+"""An import loads its target when its module loads, unless a function call or a type
+checker is the only thing that runs it."""
+
+from pathlib import Path
+
+from module_layers import Package
+
+SOURCES = {
+    "__init__.py": "",
+    "top.py": "",
+    "body.py": "",
+    "decorator.py": "",
+    "default.py": "",
+    "checked.py": "",
+    "user.py": (
+        "from typing import TYPE_CHECKING\n"
+        "import importlib\n"
+        "from . import top\n"
+        "if TYPE_CHECKING:\n"
+        "    from . import checked\n"
+        "@importlib.import_module('agent_comms.decorator').wrap\n"
+        "def f(\n"
+        "    x=importlib.import_module('agent_comms.default'),\n"
+        "    y=lambda: importlib.import_module('agent_comms.body'),\n"
+        "):\n"
+        "    from . import body\n"
+    ),
+}
+
+
+def test_only_imports_that_run_while_the_module_loads_load_their_target(tmp_path: Path) -> None:
+    for name, source in SOURCES.items():
+        path = tmp_path / "src" / "agent_comms" / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(source)
+    package = Package(tmp_path)
+    loads = {
+        edge.target: edge.loads for edge in package.imports if edge.source == "agent_comms.user"
+    }
+    assert loads == {
+        "agent_comms.top": True,
+        "agent_comms.decorator": True,
+        "agent_comms.default": True,
+        "agent_comms.checked": False,
+        "agent_comms.body": False,
+    }

tests/test_repairs.py

@@ -0,0 +1,55 @@
+import subprocess
+import sys
+from pathlib import Path
+
+from repairs import ImportCycle, ImportHoist
+
+PACKAGE = {
+    "__init__.py": "",
+    "leaf.py": "VALUE = 1\n",
+    "loop.py": "from .user import used\n\nLOOPED = used\n",
+    "user.py": (
+        "def used():\n"
+        "    from .leaf import VALUE\n"
+        "\n"
+        "    return VALUE\n"
+        "\n"
+        "\n"
+        "def looped():\n"
+        "    from .loop import LOOPED\n"
+        "\n"
+        "    return LOOPED\n"
+    ),
+}
+
+
+def tree(root: Path) -> Path:
+    for name, source in PACKAGE.items():
+        path = root / "src" / "agent_comms" / name
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(source)
+    return root
+
+
+def test_a_local_import_moves_up_unless_a_load_time_import_leads_back(tmp_path: Path) -> None:
+    repo = tree(tmp_path)
+    hoist = ImportHoist(repo, ["src/agent_comms/user.py"])
+    diff, changed = hoist.simulation()
+    assert changed == 1
+    assert "+from .leaf import VALUE" in diff
+    assert "-    from .leaf import VALUE" in diff
+    assert "-    from .loop import LOOPED" not in diff
+    assert hoist.left == {ImportCycle: 1}
+
+    assert ImportHoist(repo, ["src/agent_comms/user.py"]).apply() == ("src/agent_comms/user.py",)
+    source = (repo / "src/agent_comms/user.py").read_text()
+    assert source.startswith("from .leaf import VALUE\n")
+    assert "    from .loop import LOOPED\n" in source
+    loaded = subprocess.run(
+        [sys.executable, "-c", "import agent_comms.loop, agent_comms.user"],
+        cwd=repo,
+        env={"PYTHONPATH": str(repo / "src")},
+        capture_output=True,
+        text=True,
+    )
+    assert loaded.returncode == 0, loaded.stderr

tools/debt_ratchet.py

@@ -28,7 +28,6 @@
 import tomllib
 import types
 import typing
-import typing_extensions
 from typing import Annotated
 from abc import abstractmethod
 from collections import Counter
@@ -52,6 +51,7 @@
 from agent_comms.declared_family import DeclaredFamily
 from agent_comms.installer.carry import STEPS
 from stored_formats import StoredForm
+from typing_names import TypingImports, TypingReference
 from agent_comms.field_codec import FieldCodec, ViewExclusion, boundary_decoder, projected
 from agent_comms.mro_dispatch import MroDispatch, handles
 from agent_comms.phase_evidence import MainTree, PlanReading
@@ -2206,125 +2206,6 @@ def occurrences(node: ast.AST) -> int:
         return 0
 
 
-class TypingImports(MroDispatch):
-    """Resolve unshadowed module imports from the actual stdlib declarations.
-
-    This bounded syntactic measure never executes inspected source. Nested
-    bindings conservatively disqualify a name rather than inferring its flow.
-    """
-
-    typing_modules: typing.ClassVar[dict[str, types.ModuleType]] = {
-        module.__name__: module for module in (typing, typing_extensions)
-    }
-
-    def __init__(self, declarations: set[ast.AST]) -> None:
-        self.declarations = declarations
-        self.values: dict[str, object] = {}
-        self.shadowed: set[str] = set()
-
-    def bind(self, node: ast.AST, name: str, value: object) -> None:
-        if node not in self.declarations or name in self.values:
-            self.shadowed.add(name)
-        self.values[name] = value
-
-    @handles(ast.Import)
-    def modules(self, node: ast.Import) -> None:
-        for item in node.names:
-            self.bind(node, item.asname or item.name.split(".")[0],
-                      self.typing_modules.get(item.name))
-
-    @handles(ast.ImportFrom)
-    def members(self, node: ast.ImportFrom) -> None:
-        for item in node.names:
-            declaration = None
-            module = self.typing_modules.get(node.module) if node.level == 0 else None
-            if module is not None:
-                declaration = vars(module).get(item.name)
-            self.bind(node, item.asname or item.name, declaration)
-
-    @handles(ast.Name)
-    def assignment(self, node: ast.Name) -> None:
-        if isinstance(node.ctx, (ast.Store, ast.Del)):
-            self.shadowed.add(node.id)
-
-    @handles(ast.Attribute)
-    def mutation(self, node: ast.Attribute) -> None:
-        if isinstance(node.ctx, (ast.Store, ast.Del)):
-            for descendant in ast.walk(node.value):
-                if isinstance(descendant, ast.Name):
-                    self.shadowed.add(descendant.id)
-
-    @handles(ast.arg)
-    def parameter(self, node: ast.arg) -> None:
-        self.shadowed.add(node.arg)
-
-    @handles(ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)
-    def named_binding(self, node) -> None:
-        self.shadowed.add(node.name)
-
-    @handles(ast.ExceptHandler)
-    def exception_binding(self, node: ast.ExceptHandler) -> None:
-        name = node.name
-        if name is not None:
-            self.shadowed.add(name)
-
-    # Honor the package's Python 3.11 contract without inventing parser nodes.
-    if hasattr(ast, "type_param"):
-        @handles(ast.type_param)
-        def type_parameter(self, node) -> None:
-            self.shadowed.add(node.name)
-
-    @classmethod
-    def of(cls, tree: ast.Module) -> dict[str, object]:
-        imports = cls(set(tree.body))
-        for node in ast.walk(tree):
-            imports.dispatch_sync(node)
-        if "*" in imports.values:
-            return {}  # A star import can replace any otherwise known binding.
-        return {name: value for name, value in imports.values.items()
-                if name not in imports.shadowed}
-
-    @classmethod
-    def at_any_depth(cls, tree: ast.Module) -> dict[str, object]:
-        """Every name an import binds to ``typing``, wherever the import is written.
-
-        For a measure that counts uses of a typing name: skipping an import inside
-        ``if TYPE_CHECKING:`` or a function body would hide the uses it enables.
-        """
-        imports = cls(set(ast.walk(tree)))
-        for node in ast.walk(tree):
-            imports.dispatch_sync(node)
-        return {name: value for name, value in imports.values.items() if value is not None}
-
-
-class TypingReference(MroDispatch):
-    """One AST expression's resolved imported declaration, never evaluated code."""
-
-    def __init__(self, imports: dict[str, object]) -> None:
-        self.imports = imports
-        self.value: object = None
-
-    @property
-    def is_any(self) -> bool:
-        return self.value is typing.Any
-
-    @handles(ast.Name)
-    def name(self, node: ast.Name) -> None:
-        self.value = self.imports.get(node.id)
-
-    @handles(ast.Attribute)
-    def member(self, node: ast.Attribute) -> None:
-        parent = TypingReference(self.imports)
-        parent.dispatch_sync(node.value)
-        for module in TypingImports.typing_modules.values():
-            if parent.value is module:
-                self.value = vars(module).get(node.attr)
-
-    def raw_access_count(self) -> int:
-        """Typing declarations contribute no raw-record operation; unknowns do."""
-        return int(getattr(self.value, "__module__", None) != typing.__name__)
-
-
 class StringSubscript(OccurrenceMeasure):
     @classmethod
     def lines(cls, source: bytes, filename: str) -> list[int]:
@@ -2347,6 +2228,14 @@ def occurrences(node: ast.AST) -> int:
         )
 
 
+class AnyReference(TypingReference):
+    """A reference AnyName asks one question of: does it resolve to ``typing.Any``."""
+
+    @property
+    def is_any(self) -> bool:
+        return self.value is typing.Any
+
+
 class AnyInTypeExpression(MroDispatch):
     """Count ``typing.Any`` in one type expression without evaluating it.
 
@@ -2371,7 +2260,7 @@ def resolved(self, node: ast.AST) -> object:
 
     @handles(ast.Name, ast.Attribute)
     def reference(self, node: ast.expr) -> None:
-        reference = TypingReference(self.imports)
+        reference = AnyReference(self.imports)
         reference.dispatch_sync(node)
         if reference.is_any:
             self.references.append(node)
@@ -2448,7 +2337,7 @@ def declare(self, site: str, expression: ast.AST | None, scope: tuple[ast.AST, .
     def reference(self, node: ast.Name | ast.Attribute, scope: tuple[ast.AST, ...]) -> None:
         if not isinstance(node.ctx, ast.Load) or node in self.counted:
             return
-        reference = TypingReference(self.imports)
+        reference = AnyReference(self.imports)
         reference.dispatch_sync(node)
         if reference.is_any:
             self.counted.add(node)