Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions python/akf/check.py
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,14 @@ def check_file(filepath: str, threshold: float = 0.6) -> CheckResult:
if any(is_expired(claim) for claim in unit.claims):
return CheckResult(status="STALE", exit_code=1, reason="claims_expired", **base)

# Transitive staleness: this file is unchanged, but a local dependency it
# imports moved — what the stamp verified is no longer true (#124).
recorded_deps = (unit.meta or {}).get("deps")
if recorded_deps:
from .deps import changed_deps
if changed_deps(filepath, recorded_deps):
return CheckResult(status="STALE", exit_code=1, reason="dependency_changed", **base)

if overall < threshold:
return CheckResult(status="LOW", exit_code=1, reason="below_threshold", **base)

Expand Down
106 changes: 106 additions & 0 deletions python/akf/deps.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,106 @@
"""First-degree local dependency resolution for dependency-aware staleness.

A stamp on ``auth.py`` is a claim about ``auth.py`` *as it behaved with the
modules it imports*. If a local helper it imports changes, the stamp is no
longer trustworthy even though ``auth.py``'s own bytes never moved (issue
#124). At stamp time we record the content hashes of the file's first-degree
local imports; ``akf check`` flips STALE when any of them no longer match.

Scope is deliberately conservative: Python files only, first-degree imports
only, and only modules that resolve to files near the stamped file (same
directory or package-relative). Stdlib and site-packages imports are ignored
— they're versioned by the environment, not the repo.
"""

from __future__ import annotations

import ast
import hashlib
import os
from typing import Dict, Optional


def _hash_file(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
return "sha256:" + h.hexdigest()[:16]


def _module_to_path(base_dir: str, module: str, level: int = 0) -> Optional[str]:
"""Resolve a module name to a local file path, or None if not local.

``level`` is the relative-import depth (``from .. import x`` -> 2).
"""
root = base_dir
for _ in range(max(0, level - 1)):
root = os.path.dirname(root)

parts = module.split(".") if module else []
candidates = []
if parts:
candidates.append(os.path.join(root, *parts) + ".py")
candidates.append(os.path.join(root, *parts, "__init__.py"))
elif level:
candidates.append(os.path.join(root, "__init__.py"))

for cand in candidates:
if os.path.isfile(cand):
return cand
return None


def resolve_local_deps(filepath: str) -> Dict[str, str]:
"""Map each first-degree local import of a Python file to its content hash.

Keys are paths relative to the stamped file's directory (stable across
checkouts); values are ``sha256:<16 hex>`` content hashes. Returns an
empty dict for non-Python files or files that don't parse.
"""
if not filepath.endswith(".py"):
return {}

try:
with open(filepath, "r", encoding="utf-8", errors="replace") as f:
tree = ast.parse(f.read())
except (OSError, SyntaxError):
return {}

base_dir = os.path.dirname(os.path.abspath(filepath))
deps: Dict[str, str] = {}

for node in ast.walk(tree):
if isinstance(node, ast.Import):
targets = [(alias.name, 0) for alias in node.names]
elif isinstance(node, ast.ImportFrom):
targets = [(node.module or "", node.level)]
else:
continue

for module, level in targets:
path = _module_to_path(base_dir, module, level)
if path is None:
continue
rel = os.path.relpath(path, base_dir)
if rel not in deps:
try:
deps[rel] = _hash_file(path)
except OSError:
continue

return deps


def changed_deps(filepath: str, recorded: Dict[str, str]) -> list:
"""Return the recorded dependencies whose content no longer matches."""
base_dir = os.path.dirname(os.path.abspath(filepath))
changed = []
for rel, expected in recorded.items():
dep_path = os.path.join(base_dir, rel)
try:
if _hash_file(dep_path) != expected:
changed.append(rel)
except OSError:
changed.append(rel) # dependency deleted or unreadable
return changed
7 changes: 7 additions & 0 deletions python/akf/stamp.py
Original file line number Diff line number Diff line change
Expand Up @@ -228,6 +228,13 @@ def stamp_file(
file_hash = hashlib.sha256(fh.read()).hexdigest()[:16]
unit = unit.model_copy(update={"integrity_hash": f"sha256:{file_hash}"})

# Record first-degree local import hashes (Python files) so `akf check`
# can flag staleness when a dependency changes, not just this file (#124).
from .deps import resolve_local_deps
dep_hashes = resolve_local_deps(filepath)
if dep_hashes:
unit = unit.model_copy(update={"meta": {**(unit.meta or {}), "deps": dep_hashes}})

# Embed into the file using universal format layer
from .universal import embed as _embed
_embed(filepath, metadata=unit.to_dict(compact=True))
Expand Down
80 changes: 80 additions & 0 deletions python/tests/test_deps.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
"""Tests for dependency-aware staleness (#124)."""

import os
import time

import pytest

from akf.check import check_file
from akf.deps import resolve_local_deps, changed_deps
from akf.stamp import stamp_file


@pytest.fixture
def pkg(tmp_path):
"""A small package: main.py imports helper.py and pkg/util.py."""
(tmp_path / "helper.py").write_text("def help(): return 1\n")
(tmp_path / "pkg").mkdir()
(tmp_path / "pkg" / "__init__.py").write_text("")
(tmp_path / "pkg" / "util.py").write_text("def util(): return 2\n")
main = tmp_path / "main.py"
main.write_text(
"import helper\n"
"from pkg import util\n"
"import os\n" # stdlib — must be ignored
"import requests\n" # third-party — must be ignored
"def run(): return helper.help() + util.util()\n"
)
return main


class TestResolveLocalDeps:
def test_finds_local_imports_only(self, pkg):
deps = resolve_local_deps(str(pkg))
assert "helper.py" in deps
assert os.path.join("pkg", "__init__.py") in deps
assert all("os" not in k and "requests" not in k for k in deps)
assert all(v.startswith("sha256:") for v in deps.values())

def test_non_python_file_empty(self, tmp_path):
f = tmp_path / "doc.md"
f.write_text("# hi\n")
assert resolve_local_deps(str(f)) == {}

def test_changed_deps_detects_edit(self, pkg):
deps = resolve_local_deps(str(pkg))
assert changed_deps(str(pkg), deps) == []
(pkg.parent / "helper.py").write_text("def help(): return 999\n")
assert "helper.py" in changed_deps(str(pkg), deps)

def test_deleted_dep_counts_as_changed(self, pkg):
deps = resolve_local_deps(str(pkg))
(pkg.parent / "helper.py").unlink()
assert "helper.py" in changed_deps(str(pkg), deps)


class TestDependencyStaleness:
def _freeze_mtime(self, path):
"""Keep the stamped file's own mtime inside the stamp tolerance."""
past = time.time() - 1
os.utime(path, (past, past))

def test_dep_change_flips_stale(self, pkg):
stamp_file(str(pkg), agent="claude-code", evidence=["tests pass"])
assert check_file(str(pkg)).status == "OK"

# Edit only the dependency — main.py's bytes never move.
(pkg.parent / "helper.py").write_text("def help(): return 999\n")
self._freeze_mtime(pkg)
result = check_file(str(pkg))
assert result.status == "STALE"
assert result.reason == "dependency_changed"

def test_untouched_deps_stay_ok(self, pkg):
stamp_file(str(pkg), agent="claude-code", evidence=["tests pass"])
self._freeze_mtime(pkg)
assert check_file(str(pkg)).status == "OK"

def test_deps_recorded_in_stamp_meta(self, pkg):
unit = stamp_file(str(pkg), agent="claude-code")
assert "helper.py" in unit.meta["deps"]
Loading