Skip to content
Open
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
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from __future__ import annotations

import logging
import os
from dataclasses import dataclass
from datetime import date
from pathlib import Path
Expand Down Expand Up @@ -130,11 +131,55 @@ def parse_deprecation_configs(raw: dict[str, Any]) -> list[DeprecationConfig]:
return tombstones


# Users author configuration that can reference a bundle in two places under
# a ``.amplifier/`` directory: the settings files at its top level
# (settings.yaml, settings.local.yaml, ...) and hand-written bundle files
# under ``bundles/``. Other top-level ``*.yaml`` files (some tool-written) are
# scanned too; they are few and small. The subdirectories are tool-managed state --
# the resolved module cache (#344), per-project session transcripts, and
# data stores of installed bundles (knowledge bases, evaluation runs, recipe
# state) -- which can be very large (tens of thousands of directories on a
# well-used machine) and can mention bundle names as *data*, producing false
# "evidence". Scanning only the authored locations keeps session:start cheap
# and the evidence meaningful.
_BUNDLES_SUBDIR = "bundles"
_BUNDLES_MAX_DEPTH = 4
_BUNDLES_SKIP_DIRS = frozenset({".git", "cache"})


def _authored_yaml_files(amp_dir: Path):
"""Yield top-level ``*.yaml`` files and ``*.yaml`` under ``bundles/``
(``.git`` and ``cache`` directories skipped, depth-capped)."""
try:
entries = sorted(amp_dir.iterdir())
except OSError:
return
for entry in entries:
if entry.suffix == ".yaml" and entry.is_file():
yield entry
bundles = amp_dir / _BUNDLES_SUBDIR
if not bundles.is_dir():
return
for dirpath, dirnames, filenames in os.walk(bundles):
depth = len(Path(dirpath).relative_to(bundles).parts)
dirnames[:] = (
[]
if depth >= _BUNDLES_MAX_DEPTH
else sorted(d for d in dirnames if d not in _BUNDLES_SKIP_DIRS)
)
for filename in sorted(filenames):
if filename.endswith(".yaml"):
yield Path(dirpath) / filename


def find_source_files(bundle_name: str, search_dirs: list[Path]) -> list[str]:
"""Scan .amplifier/ directories for files referencing the deprecated bundle.
"""Scan authored ``.amplifier/`` configuration for references to the deprecated bundle.

Best-effort: silently skips unreadable files and missing directories.
Searches for any YAML file under .amplifier/ that contains the bundle_name string.
Looks only at top-level ``*.yaml`` in ``.amplifier/`` (settings files and a
few tool-written files) and ``*.yaml`` under ``.amplifier/bundles/`` --
never tool-managed state (module cache, sessions, bundle data stores).
Results are de-duplicated by real path and returned in scan order.

Args:
bundle_name: Name of the deprecated bundle to search for.
Expand All @@ -144,21 +189,20 @@ def find_source_files(bundle_name: str, search_dirs: list[Path]) -> list[str]:
List of absolute file paths containing references to the bundle.
"""
found: list[str] = []
seen: set[str] = set()

for base_dir in search_dirs:
amp_dir = base_dir / ".amplifier"
if not amp_dir.is_dir():
continue

for yaml_file in amp_dir.rglob("*.yaml"):
# Skip resolved/cached artifacts (e.g. .amplifier/cache/...) — the
# tombstone's own carrier config is cached on every install and would
# otherwise always self-match, defeating require_evidence gating. (#344)
# Scope the check to the path *below* .amplifier so a "cache" segment
# in an ancestor of base_dir (e.g. a user project under some cache/
# dir) doesn't suppress genuine authored config.
if "cache" in yaml_file.relative_to(amp_dir).parts:
for yaml_file in _authored_yaml_files(amp_dir):
# os.path.realpath never raises on symlink loops (Path.resolve
# does before Python 3.13).
key = os.path.realpath(yaml_file)
if key in seen:
continue
seen.add(key)
try:
content = yaml_file.read_text(encoding="utf-8")
if bundle_name in content:
Expand Down
76 changes: 75 additions & 1 deletion modules/hooks-deprecation/tests/test_deprecation_hook.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for the deprecation hook module."""

import logging
import os
from datetime import date, timedelta
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -162,7 +163,7 @@ def test_handles_missing_amplifier_dir(self, tmp_path):
assert results == []

def test_scans_nested_yaml_files(self, tmp_path):
"""Finds bundle references in nested .amplifier/ subdirectories."""
"""Finds bundle references in nested directories under .amplifier/bundles/."""
amp_dir = tmp_path / ".amplifier" / "bundles"
amp_dir.mkdir(parents=True)
config = amp_dir / "my-config.yaml"
Expand Down Expand Up @@ -1301,3 +1302,76 @@ async def test_invalid_config_does_not_register_contributor(self):
await mount(coordinator, {})

assert coordinator.register_contributor_calls == []


class TestFindSourceFilesAuthoredOnly:
"""Only authored config is evidence: top-level settings files and bundles/."""

def _amp(self, tmp_path):
amp = tmp_path / ".amplifier"
amp.mkdir()
return amp

def test_ignores_tool_managed_state(self, tmp_path):
amp = self._amp(tmp_path)
(amp / "settings.yaml").write_text("bundle:\n app: [lsp-python]\n")
noise = {
"cache/dep/x.yaml", # resolved modules (#344)
"projects/-p/sessions/s1/metadata.yaml", # session storage
"team-knowledge/kb/capabilities/lsp-python.yaml", # a bundle's data store
"evaluation/run1/config.yaml",
"resolve/instance.yaml", # a subdir's top level
".hidden/x.yaml",
}
for rel in noise:
f = amp / rel
f.parent.mkdir(parents=True, exist_ok=True)
f.write_text("lsp-python\n")
assert find_source_files("lsp-python", [tmp_path]) == [
str(amp / "settings.yaml")
]

def test_finds_nested_authored_bundles_within_depth(self, tmp_path):
amp = self._amp(tmp_path)
shallow = amp / "bundles" / "team" / "profile.yaml"
deep = amp / "bundles" / "a" / "b" / "c" / "d" / "e" / "far.yaml"
for f in (shallow, deep):
f.parent.mkdir(parents=True)
f.write_text("includes: [lsp-python]\n")
assert find_source_files("lsp-python", [tmp_path]) == [str(shallow)]

def test_skips_cache_and_git_under_bundles(self, tmp_path):
amp = self._amp(tmp_path)
for rel in ("bundles/cache/x.yaml", "bundles/.git/x.yaml"):
f = amp / rel
f.parent.mkdir(parents=True)
f.write_text("lsp-python\n")
assert find_source_files("lsp-python", [tmp_path]) == []

def test_other_hidden_dirs_under_bundles_still_scanned(self, tmp_path):
amp = self._amp(tmp_path)
f = amp / "bundles" / ".team" / "x.yaml"
f.parent.mkdir(parents=True)
f.write_text("lsp-python\n")
assert find_source_files("lsp-python", [tmp_path]) == [str(f)]

def test_symlink_loop_under_bundles_is_skipped(self, tmp_path):
amp = self._amp(tmp_path)
bundles = amp / "bundles"
bundles.mkdir()
(amp / "settings.yaml").write_text("lsp-python\n")
loop = bundles / "loop.yaml"
try:
os.symlink(loop, loop)
except (OSError, NotImplementedError):
pytest.skip("symlinks unavailable")
assert find_source_files("lsp-python", [tmp_path]) == [
str(amp / "settings.yaml")
]

def test_cwd_equal_to_home_reports_each_file_once(self, tmp_path):
amp = self._amp(tmp_path)
(amp / "settings.local.yaml").write_text("lsp-python\n")
assert find_source_files("lsp-python", [tmp_path, tmp_path]) == [
str(amp / "settings.local.yaml")
]
Loading