Skip to content

Dependency reads

dependency_reads

Read authority derived from the target test interpreter's declared dependencies.

DependencyEnvironment dataclass

Trusted runner inputs, transported with the phase contract.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass(frozen=True)
class DependencyEnvironment:
    """Trusted runner inputs, transported with the phase contract."""

    interpreter: str
    package_root: str
    pythonpath: tuple[str, ...] = ()

    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> DependencyEnvironment:
        """Validate absolute runner locations; never infer a service interpreter."""
        interpreter, root = raw.get("interpreter"), raw.get("package_root")
        paths = raw.get("pythonpath", ())
        if not isinstance(interpreter, str) or not Path(interpreter).is_absolute():
            raise ValueError("dependency interpreter must be absolute")
        if not isinstance(root, str) or not Path(root).is_absolute():
            raise ValueError("dependency package_root must be absolute")
        if not isinstance(paths, (list, tuple)) or any(
            not isinstance(p, str) or not Path(p).is_absolute() for p in paths
        ):
            raise ValueError("dependency pythonpath entries must be absolute")
        return cls(interpreter, os.path.realpath(root), tuple(str(p) for p in paths))

    def to_mapping(self) -> dict[str, object]:
        """Serialize the runner inputs without granting any write prefix."""
        return {
            "interpreter": self.interpreter,
            "package_root": self.package_root,
            "pythonpath": list(self.pythonpath),
        }
from_mapping(raw) classmethod

Validate absolute runner locations; never infer a service interpreter.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@classmethod
def from_mapping(cls, raw: Mapping[str, object]) -> DependencyEnvironment:
    """Validate absolute runner locations; never infer a service interpreter."""
    interpreter, root = raw.get("interpreter"), raw.get("package_root")
    paths = raw.get("pythonpath", ())
    if not isinstance(interpreter, str) or not Path(interpreter).is_absolute():
        raise ValueError("dependency interpreter must be absolute")
    if not isinstance(root, str) or not Path(root).is_absolute():
        raise ValueError("dependency package_root must be absolute")
    if not isinstance(paths, (list, tuple)) or any(
        not isinstance(p, str) or not Path(p).is_absolute() for p in paths
    ):
        raise ValueError("dependency pythonpath entries must be absolute")
    return cls(interpreter, os.path.realpath(root), tuple(str(p) for p in paths))
to_mapping()

Serialize the runner inputs without granting any write prefix.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
55
56
57
58
59
60
61
def to_mapping(self) -> dict[str, object]:
    """Serialize the runner inputs without granting any write prefix."""
    return {
        "interpreter": self.interpreter,
        "package_root": self.package_root,
        "pythonpath": list(self.pythonpath),
    }

DependencySources dataclass

Import prefixes and their resolved package/file roots, never site-packages.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass(frozen=True)
class DependencySources:
    """Import prefixes and their resolved package/file roots, never site-packages."""

    roots: tuple[tuple[str, str], ...] = ()

    def permits(self, path: str | Path) -> bool:
        """Authorize canonical dependency descendants, refusing escaping symlinks."""
        candidate = Path(path).resolve()
        return any(
            candidate == Path(root)
            or (Path(root).is_dir() and candidate.is_relative_to(Path(root)))
            for _, root in self.roots
        )

    def resolve_symbol(
        self,
        symbol: str,
        _seen: frozenset[str] = frozenset(),
    ) -> tuple[Path, str, str] | None:
        """Resolve modules and static re-exports without importing dependency code."""
        if symbol in _seen:
            return None
        resolved = self._locate_symbol(symbol)
        if resolved is None or not resolved[2]:
            return resolved
        imported = _reexport_reference(resolved)
        if imported is not None:
            return self.resolve_symbol(imported, _seen | {symbol})
        return resolved

    def _locate_symbol(self, symbol: str) -> tuple[Path, str, str] | None:
        """Resolve the longest module prefix without importing dependency code."""
        parts = symbol.split(".")
        for length in range(len(parts), 0, -1):
            module = ".".join(parts[:length])
            for prefix, raw in self.roots:
                if module != prefix and not module.startswith(prefix + "."):
                    continue
                root = Path(raw)
                if root.is_file() and module != prefix:
                    continue
                suffix = module[len(prefix) :].lstrip(".").split(".")
                base = root.joinpath(*suffix) if suffix != [""] else root
                candidates = (
                    (base / "__init__.py", base.with_suffix(".py"))
                    if root.is_dir()
                    else (root,)
                )
                for candidate in candidates:
                    if candidate.is_file() and self.permits(candidate):
                        return candidate.resolve(), module, ".".join(parts[length:])
        return None
permits(path)

Authorize canonical dependency descendants, refusing escaping symlinks.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
70
71
72
73
74
75
76
77
def permits(self, path: str | Path) -> bool:
    """Authorize canonical dependency descendants, refusing escaping symlinks."""
    candidate = Path(path).resolve()
    return any(
        candidate == Path(root)
        or (Path(root).is_dir() and candidate.is_relative_to(Path(root)))
        for _, root in self.roots
    )
resolve_symbol(symbol, _seen=frozenset())

Resolve modules and static re-exports without importing dependency code.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def resolve_symbol(
    self,
    symbol: str,
    _seen: frozenset[str] = frozenset(),
) -> tuple[Path, str, str] | None:
    """Resolve modules and static re-exports without importing dependency code."""
    if symbol in _seen:
        return None
    resolved = self._locate_symbol(symbol)
    if resolved is None or not resolved[2]:
        return resolved
    imported = _reexport_reference(resolved)
    if imported is not None:
        return self.resolve_symbol(imported, _seen | {symbol})
    return resolved

current_read_contract()

Return the transport-bound contract, falling back only for local CLI calls.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def current_read_contract() -> WriteContract | None:
    """Return the transport-bound contract, falling back only for local CLI calls."""
    from axm.tools.write_scope import (
        WriteContract,
        restrict_write_contract,
        write_contract_from_env,
    )

    value = _READ_CONTRACT.get()
    if value is _UNBOUND:
        contract = write_contract_from_env()
    else:
        contract = value if isinstance(value, WriteContract) else None
    return restrict_write_contract(contract, _READ_RESTRICTIONS.get())

dependency_read_context(contract)

Bind trusted transport authority for one tool invocation.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
224
225
226
227
228
229
230
231
@contextmanager
def dependency_read_context(contract: WriteContract | None) -> Iterator[None]:
    """Bind trusted transport authority for one tool invocation."""
    token = _READ_CONTRACT.set(contract)
    try:
        yield
    finally:
        _READ_CONTRACT.reset(token)

dependency_sources(environment)

Probe distribution metadata in the test environment; errors fail closed.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def dependency_sources(environment: DependencyEnvironment) -> DependencySources:
    """Probe distribution metadata in the test environment; errors fail closed."""
    probe = Path(__file__).with_name("_dependency_probe.py")
    result = subprocess.run(  # noqa: S603 - trusted runner, no shell
        [environment.interpreter, "-I", "-B", str(probe)],
        input=json.dumps(environment.to_mapping()),
        text=True,
        capture_output=True,
        timeout=30,
        check=False,
    )
    if result.returncode:
        raise ValueError(f"dependency resolution failed: {result.stderr.strip()}")
    roots = json.loads(result.stdout)
    if not isinstance(roots, list):
        raise ValueError("invalid dependency resolution response")
    return DependencySources(
        tuple((str(item[0]), os.path.realpath(item[1])) for item in roots)
    )

read_context_bound()

Whether a transport explicitly bound authority, including a denial None.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
209
210
211
def read_context_bound() -> bool:
    """Whether a transport explicitly bound authority, including a denial None."""
    return _READ_CONTRACT.get() is not _UNBOUND or bool(_READ_RESTRICTIONS.get())

resolve_read_path(base, candidate)

Resolve a local read or an authorized external dependency source.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
240
241
242
243
244
245
246
def resolve_read_path(base: Path, candidate: str) -> Path | None:
    """Resolve a local read or an authorized external dependency source."""
    path = (base / candidate).resolve()
    contract = current_read_contract()
    if contract is not None:
        return path if contract.readable(str(path)) else None
    return path if path.is_relative_to(base.resolve()) else None

restricted_read_context(contract)

Retain an origin's authority; live caller authority can only narrow it.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
214
215
216
217
218
219
220
221
@contextmanager
def restricted_read_context(contract: WriteContract | None) -> Iterator[None]:
    """Retain an origin's authority; live caller authority can only narrow it."""
    token = _READ_RESTRICTIONS.set((*_READ_RESTRICTIONS.get(), contract))
    try:
        yield
    finally:
        _READ_RESTRICTIONS.reset(token)

session_dependencies()

Get read authority from the current session, never from ambient sys.path.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
165
166
167
168
169
170
171
172
def session_dependencies() -> DependencySources:
    """Get read authority from the current session, never from ambient sys.path."""
    # The transport context wins over process environment.

    contract = current_read_contract()
    if contract is None or contract.dependency_environment is None:
        return DependencySources()
    return dependency_sources(contract.dependency_environment)

session_dependency_environment(project_path)

Resolve a declared runtime, refusing implicit execution via read authority.

Return None only without a transport/local contract. Under a contract a matching declared environment is mandatory, including for readable roots. Keep the interpreter's lexical venv path; resolving its symlink loses venv activation semantics.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
def session_dependency_environment(
    project_path: str | Path,
) -> DependencyEnvironment | None:
    """Resolve a declared runtime, refusing implicit execution via read authority.

    Return None only without a transport/local contract. Under a contract a
    matching declared environment is mandatory, including for readable roots.
    Keep the interpreter's lexical venv path; resolving its symlink loses venv
    activation semantics.
    """
    contract = current_read_contract()
    if contract is None:
        return None
    environments = contract.dependency_environments
    if contract.dependency_environment is not None:
        environments = (contract.dependency_environment, *environments)
    root = Path(project_path).resolve()
    matches = [
        env
        for env in environments
        if root.is_relative_to(Path(env.package_root).resolve())
    ]
    if not matches:
        raise ValueError(f"no declared runtime for: {root}")
    selected = _unambiguous_runtime(matches, root)
    if not Path(selected.interpreter).is_file():
        raise ValueError(
            f"declared runtime interpreter missing: {selected.interpreter}"
        )
    return selected

session_readable(path)

Use precisely the same canonical read decision as the transport.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
234
235
236
237
def session_readable(path: Path) -> bool:
    """Use precisely the same canonical read decision as the transport."""
    contract = current_read_contract()
    return contract is not None and contract.readable(str(path.resolve()))

unresolved_dependency_refusals(calls, environment)

Attribute unresolved dependency access failures using backend evidence.

Source code in packages/axm/src/axm/tools/dependency_reads.py
Python
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
def unresolved_dependency_refusals(
    calls: tuple[dict[str, object], ...],
    environment: DependencyEnvironment,
) -> tuple[str, ...]:
    """Attribute unresolved dependency access failures using backend evidence."""
    sources = dependency_sources(environment)
    failed: dict[str, str] = {}
    for call in calls:
        name = str(call.get("name", "")).rsplit("__", 1)[-1]
        arguments = call.get("arguments")
        if name not in {"read_file", "ast_inspect", "ast_search"} or not isinstance(
            arguments, Mapping
        ):
            continue
        paths = _call_sources(name, arguments, environment, sources)
        text = str(call.get("result"))
        if not _read_failed(call):
            for location in paths:
                failed.pop(location, None)
        elif any(
            marker in text
            for marker in (
                "ESCAPE",
                "not found",
                "Permission denied",
                "path escapes",
                "dependency resolution failed",
            )
        ):
            for location in paths:
                failed[location] = (
                    f"{name} {location} arguments={dict(arguments)!r}: {text}"
                )
    return tuple(failed.values())