""" Tests for scanner lifecycle signals (tasks 2.1-2.3 of lifecycle-aware-doc-hygiene). Covers: - 2.1: directory-rule match prunes the walk + emits one aggregate entry; IGNORE-surface match prunes with zero emission; file-rule match attaches a lifecycle signal alongside existing objective signals; unmatched files flow through unchanged. - 2.2: temporary-tier age (git commit time, mtime fallback for untracked, one-stat directory-inode mtime for untracked dirs); retain-recent-N grouping/ranking by rule match entry. - 2.3: delete-once-served — served_when_path substitution + filesystem check (deterministic, scanner-proven); served_when free text is classifier-judged only, scanner asserts nothing about satisfaction. sys.path manipulation is intentional per file-ownership constraints (no shared conftest). """ from __future__ import annotations import sys from pathlib import Path # Prepend scripts/ to sys.path so we can import scanner/rulebook without a package _SCRIPTS_DIR = Path(__file__).parent.parent / "scripts" if str(_SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPTS_DIR)) from rulebook import RuleMatch # noqa: E402 from scanner import Scanner # noqa: E402 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_tree(tmp: Path, files: dict) -> None: for rel, content in files.items(): p = tmp / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content, encoding="utf-8") def _rule(glob, lifetime="temporary", origin="global", **extra) -> dict: r = {"glob": glob, "confirmed_by": "human", "confirmed_on": "2026-07-14", "source": "test"} if lifetime is not None: r["lifetime"] = lifetime r.update(extra) return r class _FakeRulebook: """A minimal rulebook stand-in: one canned RuleMatch per glob->config. Each registered entry is `(is_dir_only, matcher_fn, level, is_ignore, rule)`. `matcher_fn(path, is_dir)` decides whether the entry applies. This keeps scanner tests fully isolated from rulebook.py's glob-compilation internals (already covered by test_rulebook.py) while exercising the scanner's *consumption* of RuleMatch objects. """ def __init__(self): self._entries = [] # list of (matcher_fn, level, is_ignore, rule, origin) def register(self, matcher_fn, level, rule, is_ignore=False, origin="global"): self._entries.append((matcher_fn, level, is_ignore, rule, origin)) def query(self, path, is_dir=False): norm = path.strip("/") for matcher_fn, level, is_ignore, rule, origin in self._entries: if matcher_fn(norm, is_dir): return RuleMatch(rule=rule, level=level, is_ignore=is_ignore, origin=origin) return None def _run(root: Path, rulebook=None, git_commit_time_fn=None, now_fn=None, **kwargs) -> dict: return Scanner( root=root, git_log_fn=lambda p: [], now_fn=now_fn or (lambda: 0.0), rulebook=rulebook, git_commit_time_fn=git_commit_time_fn, **kwargs, ).run() # --------------------------------------------------------------------------- # 2.1: directory-rule prune + aggregate entry # --------------------------------------------------------------------------- class TestDirectoryRulePrune: def test_directory_rule_prunes_walk_and_emits_one_aggregate_entry(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-2026-07-01/a.md": "content a", "autoresearch/run-2026-07-01/b.md": "content b", "autoresearch/run-2026-07-01/nested/c.md": "content c", "docs/live.md": "live content", }) rule = _rule("autoresearch/*/**", lifetime="temporary") rb = _FakeRulebook() rb.register( lambda p, is_dir: is_dir and p == "autoresearch/run-2026-07-01", "directory", rule, ) result = _run(tmp_path, rulebook=rb) # No file beneath the pruned directory is opened/shortlisted. for path in result["shortlist"]: assert not path.startswith("autoresearch/run-2026-07-01/"), path # Exactly one aggregate entry for the directory path itself. assert "autoresearch/run-2026-07-01" in result["shortlist"] assert result["shortlist"].count("autoresearch/run-2026-07-01") == 1 sigs = result["signals"]["autoresearch/run-2026-07-01"] lifecycle_sigs = [s for s in sigs if s["name"] == "lifecycle"] assert len(lifecycle_sigs) == 1 assert lifecycle_sigs[0]["lifetime"] == "temporary" assert lifecycle_sigs[0]["rule_ref"] == "autoresearch/*/**" # Unrelated file flows through unchanged. assert "docs/live.md" in result["shortlist"] assert "docs/live.md" not in result["signals"] or all( s["name"] != "lifecycle" for s in result["signals"]["docs/live.md"] ) def test_ignore_surface_prunes_with_zero_emission(self, tmp_path): _make_tree(tmp_path, { "graphify-out/graph.json": "{}", "graphify-out/nested/index.md": "index", "docs/live.md": "live content", }) ignore_rule = _rule("graphify-out/**", lifetime=None) rb = _FakeRulebook() rb.register( lambda p, is_dir: is_dir and p == "graphify-out", "directory", ignore_rule, is_ignore=True, ) result = _run(tmp_path, rulebook=rb) for path in result["shortlist"]: assert not path.startswith("graphify-out"), path assert "graphify-out" not in result["shortlist"] assert "graphify-out" not in result["signals"] def test_file_rule_attaches_lifecycle_signal_alongside_existing_signals(self, tmp_path): _make_tree(tmp_path, { "HANDOFF-2026-07-01.md": "[broken](./missing.md)", }) rule = _rule( "HANDOFF-2026-07-01.md", lifetime="delete-once-served", served_when="the handoff has been read and actioned", ) rb = _FakeRulebook() rb.register( lambda p, is_dir: (not is_dir) and p == "HANDOFF-2026-07-01.md", "file", rule, ) result = _run(tmp_path, rulebook=rb) sigs = result["signals"]["HANDOFF-2026-07-01.md"] names = {s["name"] for s in sigs} assert "lifecycle" in names assert "broken_reference" in names def test_unmatched_files_flow_through_unchanged(self, tmp_path): _make_tree(tmp_path, {"docs/plain.md": "plain content, nothing special"}) rb = _FakeRulebook() # no registered rules -> always None result = _run(tmp_path, rulebook=rb) assert "docs/plain.md" in result["shortlist"] assert "docs/plain.md" not in result["signals"] # --------------------------------------------------------------------------- # 2.2: temporary-tier age + retain-recent-N # --------------------------------------------------------------------------- class TestTemporaryTierAge: def test_tracked_file_age_from_git_commit_time(self, tmp_path): _make_tree(tmp_path, {"tmp/a.md": "content"}) rule = _rule("tmp/a.md", lifetime="temporary", max_age_days=3, retain_recent=3) rb = _FakeRulebook() rb.register(lambda p, is_dir: (not is_dir) and p == "tmp/a.md", "file", rule) # now = 10 days after commit -> age_days == 10 commit_time = "2026-06-24T00:00:00+00:00" now_ts = __import__("datetime").datetime.fromisoformat(commit_time).timestamp() + 10 * 86400 result = _run( tmp_path, rulebook=rb, git_commit_time_fn=lambda p: commit_time, now_fn=lambda: now_ts, ) sig = [s for s in result["signals"]["tmp/a.md"] if s["name"] == "lifecycle"][0] assert abs(sig["age_days"] - 10.0) < 0.01 def test_untracked_file_age_falls_back_to_mtime(self, tmp_path): _make_tree(tmp_path, {"tmp/b.md": "content"}) rule = _rule("tmp/b.md", lifetime="temporary") rb = _FakeRulebook() rb.register(lambda p, is_dir: (not is_dir) and p == "tmp/b.md", "file", rule) result = _run( tmp_path, rulebook=rb, git_commit_time_fn=lambda p: None, # untracked now_fn=lambda: (tmp_path / "tmp/b.md").stat().st_mtime + 5 * 86400, ) sig = [s for s in result["signals"]["tmp/b.md"] if s["name"] == "lifecycle"][0] assert abs(sig["age_days"] - 5.0) < 0.01 def test_untracked_directory_age_is_one_stat_not_recursive(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-x/old_nested_file.md": "old", }) # Make the nested file's mtime much older than the directory's own # mtime, to prove the directory's own inode mtime is what's used. import os nested = tmp_path / "autoresearch/run-x/old_nested_file.md" old_time = 1_000_000.0 os.utime(nested, (old_time, old_time)) dir_path = tmp_path / "autoresearch/run-x" dir_mtime = dir_path.stat().st_mtime rule = _rule("autoresearch/*/**", lifetime="temporary") rb = _FakeRulebook() rb.register( lambda p, is_dir: is_dir and p == "autoresearch/run-x", "directory", rule ) result = _run( tmp_path, rulebook=rb, git_commit_time_fn=lambda p: None, now_fn=lambda: dir_mtime + 2 * 86400, ) sig = [s for s in result["signals"]["autoresearch/run-x"] if s["name"] == "lifecycle"][0] # Age reflects the directory's own mtime (~2 days), not the ancient # nested file's mtime (which would be a huge age if it were used). assert abs(sig["age_days"] - 2.0) < 0.01 def test_retain_recent_n_keeps_newest_3_regardless_of_age(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-1/x.md": "1", "autoresearch/run-2/x.md": "2", "autoresearch/run-3/x.md": "3", "autoresearch/run-4/x.md": "4", "autoresearch/run-5/x.md": "5", }) rule = _rule("autoresearch/*/**", lifetime="temporary", retain_recent=3, max_age_days=3) rb = _FakeRulebook() def matcher(p, is_dir): return is_dir and p.startswith("autoresearch/run-") rb.register(matcher, "directory", rule) # Ages: run-1 oldest (10d) ... run-5 newest (1d). ages = { "autoresearch/run-1": 10.0, "autoresearch/run-2": 8.0, "autoresearch/run-3": 6.0, "autoresearch/run-4": 4.0, "autoresearch/run-5": 1.0, } def commit_time_fn(abs_path): return None # force mtime fallback, patched via now_fn/mtime below # Use mtime fallback: set directory mtimes so that now - mtime == age. import os now_ts = 2_000_000.0 for rel, age in ages.items(): os.utime(tmp_path / rel, (now_ts - age * 86400, now_ts - age * 86400)) result = _run( tmp_path, rulebook=rb, git_commit_time_fn=commit_time_fn, now_fn=lambda: now_ts, ) retention_by_dir = { path: [s for s in sigs if s["name"] == "lifecycle"][0]["retention"] for path, sigs in result["signals"].items() if path.startswith("autoresearch/run-") } # Newest 3 (run-3, run-4, run-5) always kept regardless of age. assert retention_by_dir["autoresearch/run-5"]["kept"] is True assert retention_by_dir["autoresearch/run-4"]["kept"] is True assert retention_by_dir["autoresearch/run-3"]["kept"] is True # Ranked 4th and 5th (run-2, run-1) are older than max_age_days=3 -> # deletable. assert retention_by_dir["autoresearch/run-2"]["kept"] is False assert retention_by_dir["autoresearch/run-2"]["deletable"] is True assert retention_by_dir["autoresearch/run-1"]["kept"] is False assert retention_by_dir["autoresearch/run-1"]["deletable"] is True def test_4th_ranked_entry_younger_than_max_age_is_not_deletable(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-1/x.md": "1", "autoresearch/run-2/x.md": "2", "autoresearch/run-3/x.md": "3", "autoresearch/run-4/x.md": "4", }) rule = _rule("autoresearch/*/**", lifetime="temporary", retain_recent=3, max_age_days=3) rb = _FakeRulebook() rb.register(lambda p, is_dir: is_dir and p.startswith("autoresearch/run-"), "directory", rule) import os now_ts = 2_000_000.0 ages = { "autoresearch/run-1": 1.0, # 4th-ranked (oldest), but younger than max_age_days "autoresearch/run-2": 0.8, "autoresearch/run-3": 0.5, "autoresearch/run-4": 0.1, } for rel, age in ages.items(): os.utime(tmp_path / rel, (now_ts - age * 86400, now_ts - age * 86400)) result = _run( tmp_path, rulebook=rb, git_commit_time_fn=lambda p: None, now_fn=lambda: now_ts ) retention_by_dir = { path: [s for s in sigs if s["name"] == "lifecycle"][0]["retention"] for path, sigs in result["signals"].items() if path.startswith("autoresearch/run-") } assert retention_by_dir["autoresearch/run-1"]["kept"] is False assert retention_by_dir["autoresearch/run-1"]["deletable"] is False # --------------------------------------------------------------------------- # 2.3: delete-once-served # --------------------------------------------------------------------------- class TestDeleteOnceServed: def test_served_when_path_substitution_and_satisfied(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-2026-07-01/report.md": "report", "autoresearch/run-2026-07-01/archive/report.md": "archived copy", }) rule = _rule( "autoresearch/*/report.md", lifetime="delete-once-served", served_when_path="autoresearch/run-2026-07-01/archive/{name}", ) rb = _FakeRulebook() rb.register( lambda p, is_dir: (not is_dir) and p == "autoresearch/run-2026-07-01/report.md", "file", rule, ) result = _run(tmp_path, rulebook=rb) sig = [ s for s in result["signals"]["autoresearch/run-2026-07-01/report.md"] if s["name"] == "lifecycle" ][0] assert sig["served"]["kind"] == "scanner-proven" assert sig["served"]["resolved_path"] == "autoresearch/run-2026-07-01/archive/report.md" assert sig["served"]["satisfied"] is True def test_served_when_path_not_satisfied(self, tmp_path): _make_tree(tmp_path, { "autoresearch/run-2026-07-01/report.md": "report", }) rule = _rule( "autoresearch/*/report.md", lifetime="delete-once-served", served_when_path="autoresearch/run-2026-07-01/archive/{name}", ) rb = _FakeRulebook() rb.register( lambda p, is_dir: (not is_dir) and p == "autoresearch/run-2026-07-01/report.md", "file", rule, ) result = _run(tmp_path, rulebook=rb) sig = [ s for s in result["signals"]["autoresearch/run-2026-07-01/report.md"] if s["name"] == "lifecycle" ][0] assert sig["served"]["kind"] == "scanner-proven" assert sig["served"]["satisfied"] is False def test_served_when_free_text_is_classifier_judged_only(self, tmp_path): _make_tree(tmp_path, {"HANDOFF-2026-07-01.md": "handoff content"}) rule = _rule( "HANDOFF-2026-07-01.md", lifetime="delete-once-served", served_when="the handoff has been read and actioned by the team", ) rb = _FakeRulebook() rb.register( lambda p, is_dir: (not is_dir) and p == "HANDOFF-2026-07-01.md", "file", rule ) result = _run(tmp_path, rulebook=rb) sig = [ s for s in result["signals"]["HANDOFF-2026-07-01.md"] if s["name"] == "lifecycle" ][0] assert sig["served"]["kind"] == "classifier-judged" assert sig["served"]["served_when"] == "the handoff has been read and actioned by the team" # Scanner asserts nothing about satisfaction for classifier-judged signals. assert "satisfied" not in sig["served"]