diff --git a/CHANGELOG.md b/CHANGELOG.md index 4bbe6496d6..2ddea8259b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -40,6 +40,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 every OS. Existing Windows lockfiles may retain the old CRLF-domain hash; keep the lockfile so its `resolved_commit` pins remain intact while the automatic one-time repair tracked in #2628 lands. (closes #2619) +- `apm compile` now reduces matching work for literal scoped `applyTo` patterns + in large repositories while preserving historical placement. It shares a + source inventory across discovery and placement while preserving cleanup + behavior, and preserves commas in character classes. (#2595) - Hook commands such as `"${CLAUDE_PLUGIN_ROOT}"/hooks/probe.py` now rewrite to `"${CLAUDE_PLUGIN_ROOT}/hooks/probe.py"` and warn when a supported plugin-root placeholder remains unresolved instead of silently deploying a dead hook. diff --git a/scripts/check_compile_inventory_authority.py b/scripts/check_compile_inventory_authority.py new file mode 100644 index 0000000000..89e72392fe --- /dev/null +++ b/scripts/check_compile_inventory_authority.py @@ -0,0 +1,75 @@ +"""Verify compile traversal is owned by the shared inventory.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).parents[1] +INVENTORY = ROOT / "src/apm_cli/compilation/inventory.py" +OPTIMIZER = ROOT / "src/apm_cli/compilation/context_optimizer.py" +DISCOVERY = ROOT / "src/apm_cli/primitives/discovery.py" +DISTRIBUTED = ROOT / "src/apm_cli/compilation/distributed_compiler.py" +AGENTS = ROOT / "src/apm_cli/compilation/agents_compiler.py" + + +def _has_all(source: str, required: tuple[str, ...]) -> bool: + """Return whether every required contract fragment appears in source.""" + return all(fragment in source for fragment in required) + + +def main() -> int: + """Return nonzero when compile traversal has a duplicate authority.""" + inventory = INVENTORY.read_text(encoding="utf-8") + optimizer = OPTIMIZER.read_text(encoding="utf-8") + discovery = DISCOVERY.read_text(encoding="utf-8") + distributed = DISTRIBUTED.read_text(encoding="utf-8") + agents = AGENTS.read_text(encoding="utf-8") + + valid = ( + inventory.count("class CompileInventory") == 1 + and inventory.count("os.walk(") == 1 + and "os.walk(" not in optimizer + and "os.walk(" not in distributed + and _has_all( + optimizer, + ( + "from .inventory import CompileInventory", + "inventory = self._inventory or CompileInventory.collect(self.base_dir)", + "inventory.files_under(self._scan_top_level_roots)", + ), + ) + and _has_all( + discovery, + ( + "inventory: CompileInventory | None = None", + "inventory.files_within(base_path)", + ), + ) + and _has_all( + distributed, + ( + "source_inventory: CompileInventory | None = None", + "deploy_inventory: CompileInventory | None = None", + "for directory_path, (relative_path, files) in sorted(cleanup_directories.items()):", + ), + ) + and _has_all( + agents, + ( + "self._source_inventory = CompileInventory.collect(", + "self.source_dir, exclude_patterns=config.exclude", + "source_inventory=self._source_inventory", + "deploy_inventory=self._deploy_inventory", + ), + ) + ) + if valid: + return 0 + + print("[x] Compile traversal must route through compilation/inventory.py") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/lint-architecture-boundaries.sh b/scripts/lint-architecture-boundaries.sh index 73b0f15c39..0ca73663b9 100755 --- a/scripts/lint-architecture-boundaries.sh +++ b/scripts/lint-architecture-boundaries.sh @@ -195,20 +195,8 @@ if [ -n "$compiled_write_hits" ]; then violations=$((violations + 1)) fi distributed_compiler="src/apm_cli/compilation/distributed_compiler.py" -nested_worktree_walk_count=$(grep -Fc \ - 'for directory, child_dirs, files in os.walk(self.base_dir, followlinks=False):' \ - "$distributed_compiler" || true) -nested_worktree_boundary_count=$(grep -Fc \ - '(directory_path / ".git").is_file()' \ - "$distributed_compiler" || true) -nested_worktree_prune_count=$(grep -Fc 'child_dirs.clear()' "$distributed_compiler" || true) -nested_worktree_rglob_hits=$(grep -En 'rglob\("AGENTS\.md"\)' "$distributed_compiler" || true) -if [ "$nested_worktree_walk_count" -ne 1 ] \ - || [ "$nested_worktree_boundary_count" -ne 1 ] \ - || [ "$nested_worktree_prune_count" -ne 1 ] \ - || [ -n "$nested_worktree_rglob_hits" ]; then - echo "[x] Nested worktree cleanup must prune .git-file roots" - [ -n "$nested_worktree_rglob_hits" ] && echo "$nested_worktree_rglob_hits" +if ! python3 scripts/check_compile_inventory_authority.py; then + echo "[x] Compile traversal must use the shared inventory" violations=$((violations + 1)) fi agents_source_attribution_output=$(python3 scripts/check_agents_source_attribution_owner.py \ @@ -1678,6 +1666,9 @@ apply_to_owner="src/apm_cli/utils/patterns.py" apply_to_normalizer_defs=$(grep -rEc --include='*.py' \ '^def _?normalize_apply_to\(' src/apm_cli \ | awk -F: '{sum += $2} END {print sum + 0}') +apply_to_prefix_defs=$(grep -rEc --include='*.py' \ + '^def literal_apply_to_top_level_roots\(' src/apm_cli \ + | awk -F: '{sum += $2} END {print sum + 0}') apply_to_parser="src/apm_cli/primitives/parser.py" hidden_tool_placement_owner="src/apm_cli/compilation/context_optimizer.py" hidden_tool_tree_defs=$(grep -rEc --include='*.py' \ @@ -1685,13 +1676,17 @@ hidden_tool_tree_defs=$(grep -rEc --include='*.py' \ | awk -F: '{sum += $2} END {print sum + 0}') if [ "$apply_to_normalizer_defs" -ne 1 ] \ || ! grep -q '^def normalize_apply_to(' "$apply_to_owner" \ + || [ "$apply_to_prefix_defs" -ne 1 ] \ + || ! grep -q '^def literal_apply_to_top_level_roots(' "$apply_to_owner" \ || ! grep -q 'from apm_cli.utils.patterns import normalize_apply_to' "$apply_to_parser" \ || grep -Eq '^def _?normalize_apply_to\(' "$apply_to_parser" \ || ! grep -q 'normalize_apply_to(metadata.get("applyTo"), default="")' "$apply_to_parser" \ || [ "$hidden_tool_tree_defs" -ne 1 ] \ || ! grep -q '^PLACEMENT_HIDDEN_TOOL_TREES = frozenset(' "$hidden_tool_placement_owner" \ + || ! grep -q 'literal_apply_to_top_level_roots(' "$hidden_tool_placement_owner" \ + || grep -q '^ def _targeted_top_level_roots(' "$hidden_tool_placement_owner" \ || ! grep -q 'not self._is_supported_hidden_tool_root(path)' "$hidden_tool_placement_owner"; then - echo "[x] applyTo normalization must use utils/patterns.py and hidden placement ContextOptimizer" + echo "[x] applyTo parsing must use utils/patterns.py and hidden placement ContextOptimizer" violations=$((violations + 1)) fi diff --git a/src/apm_cli/compilation/agents_compiler.py b/src/apm_cli/compilation/agents_compiler.py index d4a348aee2..b1a384d266 100644 --- a/src/apm_cli/compilation/agents_compiler.py +++ b/src/apm_cli/compilation/agents_compiler.py @@ -23,10 +23,11 @@ from ..primitives.discovery import discover_primitives from ..primitives.models import Instruction, PrimitiveCollection from ..utils.path_security import PathTraversalError, ensure_path_within -from ..utils.paths import portable_relpath +from ..utils.paths import portable_relpath, resolve_base_and_source_dirs from ..version import get_version from .claude_formatter import CLAUDE_HEADER, ClaudeFormatter from .constants import BUILD_ID_PLACEHOLDER +from .inventory import CompileInventory from .link_resolver import resolve_markdown_links, validate_link_targets from .template_builder import ( TemplateData, @@ -336,13 +337,14 @@ def __init__(self, base_dir: str = ".", source_dir: str | None = None): compile --root`` redirects writes but sources remain in ``$PWD``. """ - self.base_dir = Path(base_dir) - self.source_dir = Path(source_dir) if source_dir else self.base_dir + self.base_dir, self.source_dir = resolve_base_and_source_dirs(base_dir, source_dir) self.warnings: list[str] = [] self.errors: list[str] = [] self._logger = None self._distributed_placement: dict[Path, tuple[Instruction, ...]] | None = None self._distributed_context_optimizer: ContextOptimizer | None = None + self._source_inventory: CompileInventory | None = None + self._deploy_inventory: CompileInventory | None = None def _log(self, method: str, message: str, **kwargs): """Delegate to logger if available, else no-op.""" @@ -403,6 +405,14 @@ def compile( # Placement is valid only for this invocation's primitive snapshot. self._distributed_placement = None self._distributed_context_optimizer = None + self._source_inventory = CompileInventory.collect( + self.source_dir, exclude_patterns=config.exclude + ) + self._deploy_inventory = ( + self._source_inventory + if self.source_dir == self.base_dir and not config.exclude + else CompileInventory.collect(self.base_dir) + ) try: # Use provided primitives or discover them (with dependency support) @@ -412,6 +422,7 @@ def compile( primitives = discover_primitives( str(self.source_dir), exclude_patterns=config.exclude, + inventory=self._source_inventory, ) else: # Use enhanced discovery with dependencies (Task 4 integration) @@ -420,6 +431,7 @@ def compile( primitives = discover_primitives_with_dependencies( str(self.source_dir), exclude_patterns=config.exclude, + inventory=self._source_inventory, ) # Route to targets based on config.target. @@ -555,6 +567,8 @@ def _compile_distributed( str(self.base_dir), exclude_patterns=config.exclude, source_dir=str(self.source_dir), + source_inventory=self._source_inventory, + deploy_inventory=self._deploy_inventory, ) # Skip instructions in AGENTS.md when they are already deployed to the @@ -865,6 +879,8 @@ def _compile_claude_md( str(self.base_dir), exclude_patterns=config.exclude, source_dir=str(self.source_dir), + source_inventory=self._source_inventory, + deploy_inventory=self._deploy_inventory, ) placement_map = self._get_distributed_placement( config, diff --git a/src/apm_cli/compilation/context_optimizer.py b/src/apm_cli/compilation/context_optimizer.py index f8cf2d00a4..fe5462a2df 100644 --- a/src/apm_cli/compilation/context_optimizer.py +++ b/src/apm_cli/compilation/context_optimizer.py @@ -7,7 +7,6 @@ import builtins import fnmatch -import os import time from collections import defaultdict from dataclasses import dataclass, field @@ -24,7 +23,12 @@ from ..primitives.models import Instruction from ..utils.exclude import matches_glob, should_exclude, validate_exclude_patterns from ..utils.paths import portable_relpath -from ..utils.patterns import has_top_level_comma, parse_apply_to +from ..utils.patterns import ( + has_top_level_comma, + literal_apply_to_top_level_roots, + parse_apply_to, +) +from .inventory import CompileInventory # CRITICAL: Shadow Click commands to prevent namespace collision # When this module is imported during 'apm compile', Click's active context @@ -133,7 +137,12 @@ class ContextOptimizer: LOW_DISTRIBUTION_THRESHOLD = 0.3 HIGH_DISTRIBUTION_THRESHOLD = 0.7 - def __init__(self, base_dir: str = ".", exclude_patterns: builtins.list[str] | None = None): + def __init__( + self, + base_dir: str = ".", + exclude_patterns: builtins.list[str] | None = None, + inventory: CompileInventory | None = None, + ): """Initialize the context optimizer. Args: @@ -153,6 +162,7 @@ def __init__(self, base_dir: str = ".", exclude_patterns: builtins.list[str] | N self._glob_set_cache: builtins.dict[str, builtins.set[Path]] = {} self._file_list_cache: builtins.list[Path] | None = None self._placement_hidden_tool_trees: frozenset[str] = frozenset() + self._scan_top_level_roots: frozenset[str] | None = None self._inheritance_cache: builtins.dict[Path, builtins.list[Path]] = {} # (#171) self._timing_enabled = False self._phase_timings: builtins.dict[str, float] = {} @@ -172,6 +182,7 @@ def __init__(self, base_dir: str = ".", exclude_patterns: builtins.list[str] | N # Configurable exclusion patterns (validated at init time) self._exclude_patterns = validate_exclude_patterns(exclude_patterns) + self._inventory = inventory def enable_timing(self, verbose: bool = False): """Enable performance timing instrumentation.""" @@ -270,6 +281,9 @@ def optimize_instruction_placement( # Shared file discovery runs once per compile batch, so traverse the # union of roots explicitly targeted by its instructions. self._placement_hidden_tool_trees = self._targeted_hidden_tool_roots(instructions) + self._scan_top_level_roots = literal_apply_to_top_level_roots( + instruction.apply_to for instruction in instructions + ) self._file_list_cache = None self._glob_cache.clear() self._glob_set_cache.clear() @@ -299,7 +313,7 @@ def process_instructions(): instruction=instruction, pattern="(global)", matching_directories=1, - total_directories=len(self._directory_cache), + total_directories=self._placement_directory_count(), distribution_score=1.0, strategy=PlacementStrategy.DISTRIBUTED, placement_directories=[self.base_dir], @@ -485,90 +499,62 @@ def get_compilation_results( ) def _analyze_project_structure(self) -> None: - """Analyze the project structure; populate both ``_directory_cache`` and ``_file_list_cache``. - - This is the single canonical ``os.walk`` traversal for the entire - optimization pipeline. Both caches are built in one pass so that - :meth:`_get_all_files` (used by :meth:`_cached_glob` / the symlink-safe - glob replacement) never needs a separate walk. - - Ordering guarantee: subdirectories are sorted before descent and files - are sorted within each directory, so ``_file_list_cache`` is - deterministic regardless of OS-level readdir order. - """ - # Rebuild from scratch for both direct calls and optimize orchestration. + """Project full accounting and scoped candidate files from one inventory.""" self._directory_cache.clear() self._pattern_cache.clear() self._file_list_cache = [] self._files_by_directory.clear() self._children_by_directory.clear() - visited_dirs: builtins.set[Path] = set() - - for root, dirs, files in os.walk(self.base_dir): - current_path = Path(root) - - # Guard against symlink-induced infinite loops. - if current_path in visited_dirs: - dirs[:] = [] - continue - visited_dirs.add(current_path) - - relative_path = self._relative_path(current_path) - depth = len(relative_path.parts) if relative_path is not None else 0 - - # Only supported agent-tool roots participate in placement. Other - # hidden paths (including nested caches) stay pruned entirely. - if self._contains_unsupported_hidden_directory(current_path, relative_path): - dirs[:] = [] - continue - - # Safety net: skip if a default-excluded component snuck through. - if any(part in DEFAULT_EXCLUDED_DIRNAMES for part in relative_path.parts): - dirs[:] = [] - continue - - # Skip paths matching configurable exclusion patterns. - if self._should_exclude_path(current_path): - dirs[:] = [] + inventory = self._inventory or CompileInventory.collect(self.base_dir) + self._inventory = inventory + selected_files = ( + None + if self._scan_top_level_roots is None + else set(inventory.files_under(self._scan_top_level_roots)) + ) + for entry in inventory.directories: + current_path = entry.path + if not self._is_placement_path(current_path, entry.relative_path): continue - # Prune and sort subdirectories. Sorting ensures that - # ``_file_list_cache`` has a stable, deterministic order across - # platforms (the OS-level readdir order is not guaranteed). - dirs[:] = sorted(d for d in dirs if not self._should_exclude_subdir(current_path / d)) - - # Populate the children-by-directory index with admitted child dirs. - self._children_by_directory[current_path] = [current_path / d for d in dirs] - - # project_files preserves os.walk entries for project accounting; - # matching_files keeps the old is_file() filter, excluding broken - # symlinks and other non-regular entries from pattern matching. - project_files: builtins.list[Path] = [] - matching_files: builtins.list[Path] = [] - for file in sorted(files): - if not file.startswith("."): - file_path = current_path / file - self._file_list_cache.append(file_path) - project_files.append(file_path) - if file_path.is_file(): - matching_files.append(file_path) - - # Build the directory cache (only for directories that contain - # at least one non-hidden file; empty directories are skipped). - total_files = len(project_files) - if total_files == 0: + project_files = [ + current_path / name for name in entry.file_names if not name.startswith(".") + ] + self._children_by_directory[current_path] = [ + current_path / name + for name in entry.child_names + if self._is_placement_path(current_path / name) + ] + if not project_files: continue - self._files_by_directory[current_path] = matching_files - - analysis = DirectoryAnalysis( - directory=current_path, depth=depth, total_files=total_files + self._directory_cache[current_path] = DirectoryAnalysis( + directory=current_path, + depth=entry.depth, + total_files=len(project_files), + file_types={path.suffix for path in project_files}, + ) + candidate_files = ( + project_files + if selected_files is None + else [path for path in project_files if path in selected_files] ) - for fp in project_files: - analysis.file_types.add(fp.suffix) + matching_files = [path for path in candidate_files if path.is_file()] + self._file_list_cache.extend(candidate_files) + if matching_files: + self._files_by_directory[current_path] = matching_files - self._directory_cache[current_path] = analysis + def _is_placement_path(self, path: Path, relative_path: Path | None = None) -> bool: + """Return whether an inventory path participates in placement accounting.""" + relative_path = relative_path or self._relative_path(path) + if relative_path is None: + return False + return not ( + self._contains_unsupported_hidden_directory(path, relative_path) + or any(part in DEFAULT_EXCLUDED_DIRNAMES for part in relative_path.parts) + or self._should_exclude_path(path) + ) def _should_exclude_subdir(self, path: Path) -> bool: """Check if a subdirectory should be pruned from os.walk traversal. @@ -612,12 +598,15 @@ def _contains_unsupported_hidden_directory( relative_path = relative_path or self._relative_path(path) if relative_path is None: return True - # os.walk reaches a nested directory only after _should_exclude_subdir - # admitted its parent, so the top-level component is sufficient here. return bool( relative_path.parts - and relative_path.parts[0].startswith(".") - and relative_path.parts[0] not in self._placement_hidden_tool_trees + and ( + ( + relative_path.parts[0].startswith(".") + and relative_path.parts[0] not in self._placement_hidden_tool_trees + ) + or any(part.startswith(".") for part in relative_path.parts[1:]) + ) ) def _targeted_hidden_tool_roots( @@ -655,6 +644,10 @@ def _should_exclude_path(self, path: Path) -> bool: """ return should_exclude(path, self.base_dir, self._exclude_patterns) + def _placement_directory_count(self) -> int: + """Count every directory in historic placement accounting.""" + return len(self._directory_cache) + def _find_optimal_placements( self, instruction: Instruction, verbose: bool = False ) -> builtins.list[Path]: @@ -722,7 +715,7 @@ def _solve_placement_optimization( instruction=instruction, pattern=pattern, matching_directories=0, - total_directories=len(self._directory_cache), + total_directories=self._placement_directory_count(), distribution_score=0.0, strategy=PlacementStrategy.DISTRIBUTED, placement_directories=[placement], @@ -771,7 +764,7 @@ def _solve_placement_optimization( instruction=instruction, pattern=pattern, matching_directories=len(matching_directories), - total_directories=len(self._directory_cache), + total_directories=self._placement_directory_count(), distribution_score=distribution_score, strategy=strategy, placement_directories=placements, @@ -982,8 +975,8 @@ def _calculate_distribution_score(self, matching_directories: builtins.set[Path] Returns: float: Distribution score accounting for spread and depth diversity. """ - total_dirs_with_files = len( - [d for d in self._directory_cache.values() if d.total_files > 0] + total_dirs_with_files = sum( + analysis.total_files > 0 for analysis in self._directory_cache.values() ) if total_dirs_with_files == 0: return 0.0 diff --git a/src/apm_cli/compilation/distributed_compiler.py b/src/apm_cli/compilation/distributed_compiler.py index 9623d9a2d7..96f39992c7 100644 --- a/src/apm_cli/compilation/distributed_compiler.py +++ b/src/apm_cli/compilation/distributed_compiler.py @@ -7,7 +7,6 @@ import builtins import logging -import os from collections import defaultdict from collections.abc import MutableMapping from dataclasses import dataclass, field @@ -21,6 +20,7 @@ from .constants import BUILD_ID_PLACEHOLDER from .constitution import find_constitution from .context_optimizer import ContextOptimizer +from .inventory import CompileInventory from .link_resolver import UnifiedLinkResolver from .template_builder import ( build_attributed_instructions, @@ -113,6 +113,8 @@ def __init__( base_dir: str = ".", exclude_patterns: builtins.list[str] | None = None, source_dir: str | None = None, + source_inventory: CompileInventory | None = None, + deploy_inventory: CompileInventory | None = None, ): """Initialize the distributed AGENTS.md compiler. @@ -127,19 +129,54 @@ def __init__( ``base_dir`` for back-compat; set explicitly when ``apm compile --root`` redirects writes but sources remain in ``$PWD``. + source_inventory: Shared source-tree filesystem snapshot. + deploy_inventory: Shared deploy-tree snapshot for orphan cleanup. """ self.base_dir, self.source_dir = resolve_base_and_source_dirs(base_dir, source_dir) + self._exclude_patterns = exclude_patterns + self._owns_source_inventory = source_inventory is None + self._owns_deploy_inventory = deploy_inventory is None + self._source_inventory = source_inventory + self._deploy_inventory = deploy_inventory self.warnings: builtins.list[str] = [] self.errors: builtins.list[str] = [] self.total_files_written = 0 self.context_optimizer = ContextOptimizer( - str(self.source_dir), exclude_patterns=exclude_patterns + str(self.source_dir), + exclude_patterns=exclude_patterns, + inventory=self._source_inventory, ) self.link_resolver = UnifiedLinkResolver(self.source_dir) self.output_formatter = CompilationFormatter() self._placement_map = None + def _refresh_owned_inventories(self) -> None: + """Refresh direct-construction snapshots at the start of a compile.""" + if self._owns_source_inventory: + self._source_inventory = CompileInventory.collect( + self.source_dir, exclude_patterns=self._exclude_patterns + ) + if self._owns_deploy_inventory: + self._deploy_inventory = ( + self._source_inventory + if self.base_dir == self.source_dir and not self._exclude_patterns + else CompileInventory.collect(self.base_dir) + ) + self.context_optimizer._inventory = self._source_inventory + + def _deploy_inventory_for_cleanup(self) -> CompileInventory: + """Return the deploy snapshot, collecting it lazily for direct callers.""" + if self._deploy_inventory is None: + if self.base_dir == self.source_dir and not self._exclude_patterns: + self._source_inventory = self._source_inventory or CompileInventory.collect( + self.source_dir, exclude_patterns=self._exclude_patterns + ) + self._deploy_inventory = self._source_inventory + else: + self._deploy_inventory = CompileInventory.collect(self.base_dir) + return self._deploy_inventory + def _source_to_base(self, path: Path) -> Path: """Map a path rooted at source_dir to the equivalent base_dir path. @@ -170,6 +207,7 @@ def compile_distributed( """ self.warnings.clear() self.errors.clear() + self._refresh_owned_inventories() try: # Configuration with defaults aligned to Minimal Context Principle @@ -842,22 +880,40 @@ def _find_orphaned_agents_files( orphaned_files = [] generated_set = set(generated_paths) suppressed_set = set(suppressed_empty_paths or []) - - for directory, child_dirs, files in os.walk(self.base_dir, followlinks=False): - directory_path = Path(directory) + deploy_inventory = self._deploy_inventory_for_cleanup() + nested_worktree_roots = tuple( + entry.relative_path + for entry in deploy_inventory.directories if ( - directory_path != self.base_dir - and ".git" in files - and (directory_path / ".git").is_file() - ): - _logger.debug( - "Skipping nested Git worktree during orphan cleanup: %s", - portable_relpath(directory_path, self.base_dir), - ) - child_dirs.clear() + entry.path != self.base_dir + and ".git" in entry.file_names + and (entry.path / ".git").is_file() + ) + ) + cleanup_directories = { + entry.path: (entry.relative_path, entry.file_names) + for entry in deploy_inventory.directories + } + for suppressed_path in suppressed_set: + try: + relative_path = suppressed_path.parent.relative_to(self.base_dir) + except ValueError: continue + cleanup_directories.setdefault(suppressed_path.parent, (relative_path, ())) - child_dirs[:] = [child for child in child_dirs if child not in _CLEANUP_SKIP_DIRS] + for directory_path, (relative_path, files) in sorted(cleanup_directories.items()): + if any(part in _CLEANUP_SKIP_DIRS for part in relative_path.parts): + continue + if any( + relative_path.is_relative_to(worktree_root) + for worktree_root in nested_worktree_roots + ): + if relative_path in nested_worktree_roots: + _logger.debug( + "Skipping nested Git worktree during orphan cleanup: %s", + portable_relpath(directory_path, self.base_dir), + ) + continue if "AGENTS.md" not in files: continue diff --git a/src/apm_cli/compilation/inventory.py b/src/apm_cli/compilation/inventory.py new file mode 100644 index 0000000000..1ee2372b9b --- /dev/null +++ b/src/apm_cli/compilation/inventory.py @@ -0,0 +1,122 @@ +"""Read-only project inventory shared by compilation phases.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from ..utils.exclude import should_exclude, validate_exclude_patterns + +_UNIVERSALLY_SKIPPED_DIRS = frozenset( + { + ".git", + "node_modules", + "__pycache__", + ".pytest_cache", + } +) + + +@dataclass(frozen=True) +class InventoryDirectory: + """One directory observed during a compilation inventory walk.""" + + path: Path + relative_path: Path + depth: int + child_names: tuple[str, ...] + file_names: tuple[str, ...] + + +@dataclass(frozen=True) +class CompileInventory: + """Deterministic, read-only filesystem snapshot for one compile invocation. + + The inventory records filesystem facts only. Consumers retain ownership of + primitive classification, placement, and orphan-cleanup decisions. + """ + + root: Path + directories: tuple[InventoryDirectory, ...] + _directory_positions: dict[Path, int] + + @classmethod + def collect( + cls, + root: Path, + *, + exclude_patterns: list[str] | None = None, + ) -> CompileInventory: + """Collect one exclusion-aware, symlink-safe project snapshot.""" + try: + root = root.resolve() + except OSError: + root = root.absolute() + safe_patterns = validate_exclude_patterns(exclude_patterns) + directories: list[InventoryDirectory] = [] + + for directory, child_dirs, file_names in os.walk(root, followlinks=False): + path = Path(directory) + if should_exclude(path, root, safe_patterns): + child_dirs[:] = [] + continue + + admitted_children = sorted( + name + for name in child_dirs + if name not in _UNIVERSALLY_SKIPPED_DIRS + and not should_exclude(path / name, root, safe_patterns) + ) + child_dirs[:] = admitted_children + relative_path = path.relative_to(root) + directories.append( + InventoryDirectory( + path=path, + relative_path=relative_path, + depth=len(relative_path.parts), + child_names=tuple(admitted_children), + file_names=tuple(sorted(file_names)), + ) + ) + + return cls( + root=root, + directories=tuple(directories), + _directory_positions={entry.path: index for index, entry in enumerate(directories)}, + ) + + def contains_directory(self, path: Path) -> bool: + """Return whether *path* was observed as a directory.""" + return path in self._directory_positions + + def files_under(self, roots: frozenset[str] | None = None) -> tuple[Path, ...]: + """Return non-hidden candidate files, optionally under literal roots.""" + files: list[Path] = [] + for entry in self.directories: + if roots is not None and ( + not entry.relative_path.parts or entry.relative_path.parts[0] not in roots + ): + continue + files.extend(entry.path / name for name in entry.file_names if not name.startswith(".")) + return tuple(files) + + def files_within(self, directory: Path) -> tuple[Path, ...]: + """Return every recorded file beneath an inventory directory.""" + try: + resolved_directory = directory.resolve() + relative_directory = resolved_directory.relative_to(self.root) + except (OSError, ValueError): + return () + start = self._directory_positions.get(resolved_directory) + if start is None: + return () + files: list[Path] = [] + # os.walk emits each directory's subtree contiguously in this snapshot. + for entry in self.directories[start:]: + try: + entry.relative_path.relative_to(relative_directory) + except ValueError: + break + files.extend(entry.path / name for name in entry.file_names) + return tuple(files) diff --git a/src/apm_cli/primitives/discovery.py b/src/apm_cli/primitives/discovery.py index 798105c09a..e4bccd7196 100644 --- a/src/apm_cli/primitives/discovery.py +++ b/src/apm_cli/primitives/discovery.py @@ -1,10 +1,13 @@ """Discovery functionality for primitive files.""" +from __future__ import annotations + import fnmatch import logging import os import time from pathlib import Path +from typing import TYPE_CHECKING from ..constants import DEFAULT_SKIP_DIRS from ..utils import perf_stats @@ -12,6 +15,9 @@ from .models import PrimitiveCollection from .parser import parse_primitive_file, parse_skill_file +if TYPE_CHECKING: + from ..compilation.inventory import CompileInventory + logger = logging.getLogger(__name__) from ..deps.lockfile import LockFile # noqa: E402 from ..models.apm_package import APMPackage # noqa: E402 @@ -111,6 +117,7 @@ def _discovery_cache_key( def discover_primitives( base_dir: str = ".", exclude_patterns: list[str] | None = None, + inventory: CompileInventory | None = None, ) -> PrimitiveCollection: """Find all APM primitive files in the project. @@ -131,7 +138,7 @@ def discover_primitives( """ started = time.perf_counter() cache_key = _discovery_cache_key(base_dir, exclude_patterns) - cached = _DISCOVERY_CACHE.get(cache_key) + cached = None if inventory is not None else _DISCOVERY_CACHE.get(cache_key) if cached is not None: perf_stats.record_discovery( base_dir=str(base_dir), @@ -145,7 +152,12 @@ def discover_primitives( # Find and parse files for each primitive type for primitive_type, patterns in LOCAL_PRIMITIVE_PATTERNS.items(): # noqa: B007 - files = find_primitive_files(base_dir, patterns, exclude_patterns=safe_patterns) + files = find_primitive_files( + base_dir, + patterns, + exclude_patterns=safe_patterns, + inventory=inventory, + ) for file_path in files: try: @@ -157,7 +169,8 @@ def discover_primitives( # Discover SKILL.md at project root _discover_local_skill(base_dir, collection, exclude_patterns=safe_patterns) - _DISCOVERY_CACHE[cache_key] = collection + if inventory is None: + _DISCOVERY_CACHE[cache_key] = collection perf_stats.record_discovery( base_dir=str(base_dir), duration_s=time.perf_counter() - started, @@ -169,6 +182,7 @@ def discover_primitives( def discover_primitives_with_dependencies( base_dir: str = ".", exclude_patterns: list[str] | None = None, + inventory: CompileInventory | None = None, ) -> PrimitiveCollection: """Enhanced primitive discovery including dependency sources. @@ -188,7 +202,12 @@ def discover_primitives_with_dependencies( safe_patterns = validate_exclude_patterns(exclude_patterns) # Phase 1: Local primitives (highest priority) - scan_local_primitives(base_dir, collection, exclude_patterns=safe_patterns) + scan_local_primitives( + base_dir, + collection, + exclude_patterns=safe_patterns, + inventory=inventory, + ) # Phase 1b: Local SKILL.md _discover_local_skill(base_dir, collection, exclude_patterns=safe_patterns) @@ -196,7 +215,7 @@ def discover_primitives_with_dependencies( # Phase 2: Dependency primitives (lower priority, with conflict detection) # Plugins are normalized into standard APM packages during install # (apm.yml + .apm/ are synthesized), so scan_dependency_primitives handles them. - scan_dependency_primitives(base_dir, collection) + scan_dependency_primitives(base_dir, collection, inventory=inventory) return collection @@ -205,6 +224,7 @@ def scan_local_primitives( base_dir: str, collection: PrimitiveCollection, exclude_patterns: list[str] | None = None, + inventory: CompileInventory | None = None, ) -> None: """Scan local .apm/ directory for primitives. @@ -215,7 +235,12 @@ def scan_local_primitives( """ # Find and parse files for each primitive type for primitive_type, patterns in LOCAL_PRIMITIVE_PATTERNS.items(): # noqa: B007 - files = find_primitive_files(base_dir, patterns, exclude_patterns=exclude_patterns) + files = find_primitive_files( + base_dir, + patterns, + exclude_patterns=exclude_patterns, + inventory=inventory, + ) # Filter out files from apm_modules to avoid conflicts with dependency scanning local_files = [] @@ -253,7 +278,11 @@ def _is_under_directory(file_path: Path, directory: Path) -> bool: return False -def scan_dependency_primitives(base_dir: str, collection: PrimitiveCollection) -> None: +def scan_dependency_primitives( + base_dir: str, + collection: PrimitiveCollection, + inventory: CompileInventory | None = None, +) -> None: """Scan all dependencies in apm_modules/ with priority handling. Args: @@ -277,7 +306,12 @@ def scan_dependency_primitives(base_dir: str, collection: PrimitiveCollection) - dep_path = apm_modules_path.joinpath(*parts) if dep_path.exists() and dep_path.is_dir(): - scan_directory_with_source(dep_path, collection, source=f"dependency:{dep_name}") + scan_directory_with_source( + dep_path, + collection, + source=f"dependency:{dep_name}", + inventory=inventory, + ) def get_dependency_declaration_order(base_dir: str) -> list[str]: @@ -409,7 +443,11 @@ def _matches_any_pattern(rel_path: str, patterns: list[str]) -> bool: def _scan_patterns( - base_dir: Path, patterns: dict[str, list[str]], collection: PrimitiveCollection, source: str + base_dir: Path, + patterns: dict[str, list[str]], + collection: PrimitiveCollection, + source: str, + inventory: CompileInventory | None = None, ) -> None: """Walk *base_dir* once, match files against all patterns, parse and collect. @@ -424,30 +462,42 @@ def _scan_patterns( """ if not base_dir.exists(): return + try: + base_dir = base_dir.resolve() + except OSError: + base_dir = base_dir.absolute() # Flatten all patterns into a single list for matching all_patterns: list[str] = [] for _primitive_type, type_patterns in patterns.items(): all_patterns.extend(type_patterns) - base_str = str(base_dir) - for dirpath, _dirnames, filenames in os.walk(base_str, followlinks=False): - for filename in filenames: - full_path = os.path.join(dirpath, filename) - rel_path = os.path.relpath(full_path, base_str).replace(os.sep, "/") - if not _matches_any_pattern(rel_path, all_patterns): - continue - file_path = Path(full_path) - if file_path.is_file() and _is_readable(file_path): - try: - primitive = parse_primitive_file(file_path, source=source) - collection.add_primitive(primitive) - except Exception as e: - print(f"Warning: Failed to parse dependency primitive {file_path}: {e}") + files = inventory.files_within(base_dir) if inventory is not None else None + if files is None: + base_str = str(base_dir) + files = tuple( + Path(dirpath) / filename + for dirpath, _dirnames, filenames in os.walk(base_str, followlinks=False) + for filename in filenames + ) + + for file_path in files: + rel_path = file_path.relative_to(base_dir).as_posix() + if not _matches_any_pattern(rel_path, all_patterns): + continue + if not file_path.is_symlink() and file_path.is_file() and _is_readable(file_path): + try: + primitive = parse_primitive_file(file_path, source=source) + collection.add_primitive(primitive) + except Exception as e: + print(f"Warning: Failed to parse dependency primitive {file_path}: {e}") def scan_directory_with_source( - directory: Path, collection: PrimitiveCollection, source: str + directory: Path, + collection: PrimitiveCollection, + source: str, + inventory: CompileInventory | None = None, ) -> None: """Scan a directory for primitives with a specific source tag. @@ -459,14 +509,26 @@ def scan_directory_with_source( # Scan .apm directory within the dependency apm_dir = directory / ".apm" if apm_dir.exists(): - _scan_patterns(apm_dir, DEPENDENCY_PRIMITIVE_PATTERNS, collection, source) + _scan_patterns( + apm_dir, + DEPENDENCY_PRIMITIVE_PATTERNS, + collection, + source, + inventory=inventory, + ) # Also scan .github directory — some packages store primitives there instead of (or # in addition to) .apm/. Without this, dependency instructions in .github/instructions/ # are silently skipped in the normal compile path (issue #631). github_dir = directory / ".github" if github_dir.exists(): - _scan_patterns(github_dir, DEPENDENCY_GITHUB_PRIMITIVE_PATTERNS, collection, source) + _scan_patterns( + github_dir, + DEPENDENCY_GITHUB_PRIMITIVE_PATTERNS, + collection, + source, + inventory=inventory, + ) # Check for SKILL.md in the dependency root _discover_skill_in_directory(directory, collection, source) @@ -507,7 +569,7 @@ def _discover_skill_in_directory( source (str): Source identifier for the skill. """ skill_path = directory / "SKILL.md" - if skill_path.exists() and _is_readable(skill_path): + if not skill_path.is_symlink() and skill_path.exists() and _is_readable(skill_path): try: skill = parse_skill_file(skill_path, source=source) collection.add_primitive(skill) @@ -583,6 +645,7 @@ def find_primitive_files( base_dir: str, patterns: list[str], exclude_patterns: list[str] | None = None, + inventory: CompileInventory | None = None, ) -> list[Path]: """Find primitive files matching the given patterns. @@ -619,6 +682,15 @@ def find_primitive_files( all_files: list[Path] = [] files_visited = 0 + if inventory is not None: + return _find_primitive_inventory_files( + inventory, + base_path, + pattern_tuples, + exclude_patterns, + started, + ) + for root, dirs, files in os.walk(base_str): # Prune excluded directories BEFORE descending. ``DEFAULT_SKIP_DIRS`` # check is a frozenset lookup; the ``_exclude_matches_dir`` call @@ -696,6 +768,38 @@ def find_primitive_files( return valid_files +def _find_primitive_inventory_files( + inventory: CompileInventory, + base_path: Path, + pattern_tuples: list[tuple[str, ...]], + exclude_patterns: list[str] | None, + started: float, +) -> list[Path]: + """Classify a compile inventory with the existing primitive glob grammar.""" + candidates = inventory.files_within(base_path) + valid_files: list[Path] = [] + for file_path in candidates: + relative_path = file_path.relative_to(base_path) + if any(part in DEFAULT_SKIP_DIRS for part in relative_path.parts[:-1]): + continue + path_parts = relative_path.parts + if not any(_glob_match_parts(path_parts, pattern) for pattern in pattern_tuples): + continue + if exclude_patterns and should_exclude(file_path, base_path, exclude_patterns): + continue + if file_path.is_file() and not file_path.is_symlink(): + valid_files.append(file_path) + + perf_stats.record_walk( + base_dir=str(base_path), + pattern_count=len(pattern_tuples), + duration_s=time.perf_counter() - started, + files_visited=len(candidates), + files_matched=len(valid_files), + ) + return valid_files + + def _exclude_matches_dir( dir_path: Path, base_path: Path, diff --git a/src/apm_cli/utils/patterns.py b/src/apm_cli/utils/patterns.py index 73c4232216..c557de5b3b 100644 --- a/src/apm_cli/utils/patterns.py +++ b/src/apm_cli/utils/patterns.py @@ -7,9 +7,12 @@ from __future__ import annotations +from collections.abc import Iterable + _APPLY_TO_ESCAPE = "\\" _APPLY_TO_SEPARATOR = "," _ESCAPABLE_APPLY_TO_CHARS = frozenset({_APPLY_TO_SEPARATOR, _APPLY_TO_ESCAPE}) +_GLOB_META_CHARACTERS = frozenset({"*", "?", "[", "{"}) class _ApplyToPattern(str): @@ -40,6 +43,7 @@ def has_top_level_comma(pattern: str) -> bool: return False depth = 0 + in_character_class = False index = 0 while index < len(pattern): ch = pattern[index] @@ -50,7 +54,12 @@ def has_top_level_comma(pattern: str) -> bool: ): index += 2 continue - if ch == "{": + if in_character_class: + if ch == "]": + in_character_class = False + elif ch == "[": + in_character_class = True + elif ch == "{": depth += 1 elif ch == "}": if depth > 0: @@ -112,9 +121,17 @@ def _escape_apply_to_segment(pattern: str) -> str: """Encode one YAML-list pattern so top-level commas retain their boundary.""" escaped: list[str] = [] depth = 0 + in_character_class = False for char in pattern: - if char == _APPLY_TO_ESCAPE: + if in_character_class: + escaped.append(char) + if char == "]": + in_character_class = False + elif char == _APPLY_TO_ESCAPE: escaped.append(_APPLY_TO_ESCAPE * 2) + elif char == "[": + in_character_class = True + escaped.append(char) elif char == "{": depth += 1 escaped.append(char) @@ -147,6 +164,7 @@ def parse_apply_to(value: str | None) -> list[str]: return [] segments: list[_ApplyToPattern] = [] depth = 0 + in_character_class = False current: list[str] = [] escaped_top_level_comma = False @@ -167,7 +185,14 @@ def append_current() -> None: ) index += 2 continue - if char == "{": + if in_character_class: + current.append(char) + if char == "]": + in_character_class = False + elif char == "[": + in_character_class = True + current.append(char) + elif char == "{": depth += 1 current.append(char) elif char == "}": @@ -183,3 +208,78 @@ def append_current() -> None: index += 1 append_current() return [segment for segment in (s.strip() for s in segments) if segment] + + +def literal_apply_to_top_level_roots( + apply_to_values: Iterable[str | None], +) -> frozenset[str] | None: + """Return provable literal roots for a batch of ``applyTo`` expressions. + + ``None`` means a root-restricted scan could omit a matching file and callers + must retain their full traversal. Each expression must contain one or more + scoped patterns, whose first path component is literal. The returned roots + are the union across comma-separated expressions. + """ + roots: set[str] = set() + + for apply_to in apply_to_values: + if ( + not apply_to + or not apply_to.strip() + or _APPLY_TO_ESCAPE in apply_to + or not _has_balanced_glob_groups(apply_to) + ): + return None + + patterns = parse_apply_to(apply_to) + if not patterns: + return None + + for pattern in patterns: + root = _literal_top_level_root(pattern) + if root is None: + return None + roots.add(root) + + return frozenset(roots) if roots else None + + +def _has_balanced_glob_groups(pattern: str) -> bool: + """Return whether brace and character-class delimiters are balanced.""" + brace_depth = 0 + in_character_class = False + + for character in pattern: + if in_character_class: + if character == "]": + in_character_class = False + continue + if character == "[": + in_character_class = True + elif character == "{": + brace_depth += 1 + elif character == "}": + if brace_depth == 0: + return False + brace_depth -= 1 + elif character == "]": + return False + + return brace_depth == 0 and not in_character_class + + +def _literal_top_level_root(pattern: str) -> str | None: + """Return a pattern's literal first directory component, if provable.""" + normalized = pattern + while normalized.startswith("./"): + normalized = normalized[2:] + + first, separator, _ = normalized.partition("/") + if ( + not separator + or not first + or first in {".", ".."} + or any(character in _GLOB_META_CHARACTERS for character in first) + ): + return None + return first diff --git a/tests/integration/test_architecture_apply_to_patterns.py b/tests/integration/test_architecture_apply_to_patterns.py index 480936f432..5d9518af1f 100644 --- a/tests/integration/test_architecture_apply_to_patterns.py +++ b/tests/integration/test_architecture_apply_to_patterns.py @@ -23,16 +23,26 @@ def test_apply_to_normalization_and_hidden_placement_have_canonical_owners() -> guard = (root / "scripts/lint-architecture-boundaries.sh").read_text(encoding="utf-8") assert patterns.count("def normalize_apply_to(") == 1 + assert patterns.count("def literal_apply_to_top_level_roots(") == 1 assert "from apm_cli.utils.patterns import normalize_apply_to" in parser assert "def _normalize_apply_to(" not in parser assert "PLACEMENT_HIDDEN_TOOL_TREES = frozenset(" in optimizer assert "def _targeted_hidden_tool_roots(" in optimizer + assert "literal_apply_to_top_level_roots(" in optimizer + assert "def _targeted_top_level_roots(" not in optimizer assert "self._placement_hidden_tool_trees" in optimizer assert "not self._is_supported_hidden_tool_root(path)" in optimizer + inventory = (root / "src/apm_cli/compilation/inventory.py").read_text(encoding="utf-8") + inventory_guard = (root / "scripts/check_compile_inventory_authority.py").read_text( + encoding="utf-8" + ) + assert inventory.count("class CompileInventory") == 1 + assert inventory.count("os.walk(") == 1 + assert "os.walk(" not in optimizer + assert "Compile traversal must route through compilation/inventory.py" in inventory_guard assert "| applyTo normalization and hidden-tool placement |" in owner_table assert ( - "applyTo normalization must use utils/patterns.py and hidden placement ContextOptimizer" - in guard + "applyTo parsing must use utils/patterns.py and hidden placement ContextOptimizer" in guard ) @@ -70,8 +80,84 @@ def test_apply_to_owner_guard_rejects_a_parser_normalizer(tmp_path: Path) -> Non timeout=300, ) + assert result.returncode == 1 + assert "applyTo parsing must use utils/patterns.py and hidden placement ContextOptimizer" in ( + result.stdout + ) + + +def test_apply_to_owner_guard_rejects_optimizer_local_prefix_parser(tmp_path: Path) -> None: + """AC31 rejects restoring local traversal-prefix parsing.""" + root = Path(__file__).parents[2] + sandbox = tmp_path / "repo" + shutil.copytree( + root, + sandbox, + ignore=shutil.ignore_patterns( + ".git", + ".venv", + ".pytest_cache", + "__pycache__", + "build", + "dist", + "node_modules", + ), + ) + optimizer = sandbox / "src/apm_cli/compilation/context_optimizer.py" + optimizer.write_text( + optimizer.read_text(encoding="utf-8") + + "\n def _targeted_top_level_roots(self) -> frozenset[str]:\n" + + " return frozenset()\n", + encoding="utf-8", + ) + + result = subprocess.run( + ("bash", "scripts/lint-architecture-boundaries.sh"), + cwd=sandbox, + capture_output=True, + text=True, + check=False, + timeout=300, + ) + assert result.returncode == 1 assert ( - "applyTo normalization must use utils/patterns.py and hidden placement ContextOptimizer" - in (result.stdout) + "applyTo parsing must use utils/patterns.py and hidden placement ContextOptimizer" + in result.stdout ) + + +def test_compile_inventory_guard_rejects_optimizer_walk(tmp_path: Path) -> None: + """The optimizer must not restore a private project traversal.""" + root = Path(__file__).parents[2] + sandbox = tmp_path / "repo" + shutil.copytree( + root, + sandbox, + ignore=shutil.ignore_patterns( + ".git", + ".venv", + ".pytest_cache", + "__pycache__", + "build", + "dist", + "node_modules", + ), + ) + optimizer = sandbox / "src/apm_cli/compilation/context_optimizer.py" + optimizer.write_text( + optimizer.read_text(encoding="utf-8") + "\n# os.walk(self.base_dir)\n", + encoding="utf-8", + ) + + result = subprocess.run( + ("python3", "scripts/check_compile_inventory_authority.py"), + cwd=sandbox, + capture_output=True, + text=True, + check=False, + timeout=300, + ) + + assert result.returncode == 1 + assert "Compile traversal must route through compilation/inventory.py" in result.stdout diff --git a/tests/integration/test_compile_clean_nested_worktree.py b/tests/integration/test_compile_clean_nested_worktree.py index d1b4d18230..79449f5375 100644 --- a/tests/integration/test_compile_clean_nested_worktree.py +++ b/tests/integration/test_compile_clean_nested_worktree.py @@ -61,7 +61,10 @@ def test_compile_clean_preserves_nested_git_worktree_agents_file( parent.mkdir() _run_git(parent, environment, "init", "--initial-branch=main") (parent / ".gitignore").write_text(".worktrees/\n", encoding="utf-8") - (parent / "apm.yml").write_text("name: nested-worktree\nversion: 1.0.0\n", encoding="utf-8") + (parent / "apm.yml").write_text( + "name: nested-worktree\nversion: 1.0.0\ncompilation:\n exclude:\n - excluded\n", + encoding="utf-8", + ) instructions = parent / ".apm" / "instructions" instructions.mkdir(parents=True) (instructions / "root.instructions.md").write_text( @@ -86,6 +89,13 @@ def test_compile_clean_preserves_nested_git_worktree_agents_file( parent_orphan = parent / "stale" / "AGENTS.md" parent_orphan.parent.mkdir() parent_orphan.write_bytes(_GENERATED_AGENTS) + excluded_orphan = parent / "excluded" / "AGENTS.md" + excluded_orphan.parent.mkdir() + excluded_orphan.write_bytes(_GENERATED_AGENTS) + hand_authored = parent / "hand-authored" / "AGENTS.md" + hand_authored.parent.mkdir() + hand_authored.write_bytes(b"# Team-owned instructions\n") + hand_authored_before = hand_authored.read_bytes() result = ApmLifecycleRunner((str(apm_binary_path),)).run( ("compile", "--clean"), @@ -96,6 +106,8 @@ def test_compile_clean_preserves_nested_git_worktree_agents_file( assert result.returncode == 0, f"stdout={result.stdout!r}\nstderr={result.stderr!r}" assert not parent_orphan.exists() + assert not excluded_orphan.exists() + assert hand_authored.read_bytes() == hand_authored_before assert nested_agents.read_bytes() == nested_agents_before assert nested_descendant_agents.read_bytes() == nested_descendant_agents_before assert _run_git(nested, environment, "status", "--porcelain").stdout == "" diff --git a/tests/integration/test_compile_universal_apply_to_fastpath_e2e.py b/tests/integration/test_compile_universal_apply_to_fastpath_e2e.py index a9409630ea..f075831b87 100644 --- a/tests/integration/test_compile_universal_apply_to_fastpath_e2e.py +++ b/tests/integration/test_compile_universal_apply_to_fastpath_e2e.py @@ -9,6 +9,7 @@ from apm_cli.cli import cli from apm_cli.compilation.context_optimizer import ContextOptimizer +from apm_cli.compilation.inventory import CompileInventory UNIVERSAL_SENTINEL = "Universal fast path sentinel." EXPLICIT_SENTINEL = "Explicit all files sentinel." @@ -63,6 +64,39 @@ def _build_project(project_root: Path) -> set[str]: return expected_directories +def _agents_snapshot(project_root: Path) -> dict[str, bytes]: + """Return generated AGENTS.md files keyed by portable project path.""" + return { + path.relative_to(project_root).as_posix(): path.read_bytes() + for path in sorted(project_root.rglob("AGENTS.md")) + } + + +def _write_literal_scope_fixture(project_root: Path, apply_to: str) -> None: + """Create a placement-threshold fixture with unrelated project accounting.""" + (project_root / "apm.yml").write_text( + "name: literal-scope-e2e\nversion: 0.1.0\ntarget: agents\n", + encoding="utf-8", + ) + instructions_dir = project_root / ".apm" / "instructions" + instructions_dir.mkdir(parents=True) + (instructions_dir / "literal.instructions.md").write_text( + "---\n" + "description: Literal-root placement rule\n" + f'applyTo: "{apply_to}"\n' + "---\n\n" + "Literal scope sentinel.\n", + encoding="utf-8", + ) + for directory in ("api", "cli", "worker", "web"): + _write_project_file(project_root, f"src/{directory}/app.py") + # Keep the historical ``src`` placement candidate populated without + # widening the instruction's four matching directories. + _write_project_file(project_root, "src/README.md") + for index in range(20): + _write_project_file(project_root, f"vendor-{index:02d}/large.txt") + + def test_compile_agents_universal_apply_to_fast_path_preserves_match_set( tmp_path: Path, monkeypatch, @@ -138,3 +172,47 @@ def spy_file_matches(self: ContextOptimizer, file_path: Path, pattern: str) -> b agents_content = agents_md.read_text(encoding="utf-8") assert UNIVERSAL_SENTINEL in agents_content assert EXPLICIT_SENTINEL in agents_content + + +def test_compile_literal_roots_preserve_generated_artifacts_against_full_fallback( + tmp_path: Path, + monkeypatch, +) -> None: + """Literal-root pruning must be invisible in generated AGENTS.md artifacts.""" + project_root = tmp_path + literal_apply_to = "src/**/*.py" + _write_literal_scope_fixture(project_root, literal_apply_to) + + monkeypatch.chdir(project_root) + with patch( + "apm_cli.compilation.inventory.CompileInventory.collect", + wraps=CompileInventory.collect, + ) as collect: + literal_result = CliRunner().invoke( + cli, + ["compile", "--target", "agents"], + catch_exceptions=False, + ) + assert literal_result.exit_code == 0, literal_result.output + literal_artifacts = _agents_snapshot(project_root) + + for path in project_root.rglob("AGENTS.md"): + path.unlink() + + # A separate ``**/*.never`` primitive would become a root-level rule and + # legitimately alter generated bytes. Force the classifier's documented + # conservative result instead, while running the real CLI and optimizer. + with patch( + "apm_cli.compilation.context_optimizer.literal_apply_to_top_level_roots", + return_value=None, + ): + full_fallback_result = CliRunner().invoke( + cli, + ["compile", "--target", "agents"], + catch_exceptions=False, + ) + assert full_fallback_result.exit_code == 0, full_fallback_result.output + assert collect.call_count == 2 + + assert _agents_snapshot(project_root) == literal_artifacts + assert set(literal_artifacts) == {"src/AGENTS.md"} diff --git a/tests/unit/compilation/test_agents_compiler_coverage.py b/tests/unit/compilation/test_agents_compiler_coverage.py index 2f956da3a0..33deef0284 100644 --- a/tests/unit/compilation/test_agents_compiler_coverage.py +++ b/tests/unit/compilation/test_agents_compiler_coverage.py @@ -195,7 +195,26 @@ def test_compile_local_only_calls_basic_discover(self): ) as mock_disc: result = compiler.compile(config) # no primitives passed → discovers # noqa: F841 - mock_disc.assert_called_once_with(str(compiler.base_dir), exclude_patterns=config.exclude) + mock_disc.assert_called_once() + args, kwargs = mock_disc.call_args + self.assertEqual(args, (str(compiler.base_dir),)) + self.assertEqual(kwargs["exclude_patterns"], config.exclude) + self.assertIn("inventory", kwargs) + + def test_compile_uses_excluded_source_and_full_deploy_inventories(self): + """Source exclusions do not hide stale deploy outputs from cleanup.""" + (Path(self.tmp) / "vendor" / "generated.py").parent.mkdir(parents=True) + (Path(self.tmp) / "vendor" / "generated.py").touch() + compiler = AgentsCompiler(self.tmp) + config = CompilationConfig(strategy="single-file", dry_run=True, exclude=["vendor"]) + + compiler.compile(config, _make_primitives()) + + assert compiler._source_inventory is not None + assert compiler._deploy_inventory is not None + vendor = (Path(self.tmp) / "vendor").resolve() + self.assertFalse(compiler._source_inventory.contains_directory(vendor)) + self.assertTrue(compiler._deploy_inventory.contains_directory(vendor)) # --------------------------------------------------------------------------- diff --git a/tests/unit/compilation/test_compile_inventory.py b/tests/unit/compilation/test_compile_inventory.py new file mode 100644 index 0000000000..02c4c79134 --- /dev/null +++ b/tests/unit/compilation/test_compile_inventory.py @@ -0,0 +1,50 @@ +"""Component coverage for the shared compile inventory.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from apm_cli.compilation.inventory import CompileInventory + +pytestmark = pytest.mark.component + + +def _touch(base: Path, relative_path: str) -> None: + """Create one fixture file.""" + path = base / relative_path + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("x\n", encoding="utf-8") + + +@pytest.mark.windows_compat +def test_inventory_is_sorted_exclusion_aware_and_does_not_follow_symlinks(tmp_path: Path) -> None: + """The snapshot keeps portable relative paths without following directory links.""" + _touch(tmp_path, "src/z.py") + _touch(tmp_path, "src/a.py") + _touch(tmp_path, "vendor/ignored.py") + _touch(tmp_path, ".git/config") + _touch(tmp_path, "node_modules/package/index.js") + _touch(tmp_path, "__pycache__/inventory.pyc") + _touch(tmp_path, ".pytest_cache/metadata") + (tmp_path / "linked").symlink_to(tmp_path / "src", target_is_directory=True) + + inventory = CompileInventory.collect(tmp_path, exclude_patterns=["vendor"]) + + assert [entry.relative_path.as_posix() for entry in inventory.directories] == [".", "src"] + assert inventory.directories[1].file_names == ("a.py", "z.py") + assert inventory.files_under(frozenset({"src"})) == ( + tmp_path / "src/a.py", + tmp_path / "src/z.py", + ) + assert inventory.files_within(tmp_path / "src") == ( + tmp_path / "src/a.py", + tmp_path / "src/z.py", + ) + assert not inventory.contains_directory(tmp_path / "vendor") + assert not inventory.contains_directory(tmp_path / ".git") + assert not inventory.contains_directory(tmp_path / "node_modules") + assert not inventory.contains_directory(tmp_path / "__pycache__") + assert not inventory.contains_directory(tmp_path / ".pytest_cache") + assert not inventory.contains_directory(tmp_path / "linked") diff --git a/tests/unit/compilation/test_context_optimizer_iterdir_indexes.py b/tests/unit/compilation/test_context_optimizer_iterdir_indexes.py index 1dc70fa2d3..f9783c4a40 100644 --- a/tests/unit/compilation/test_context_optimizer_iterdir_indexes.py +++ b/tests/unit/compilation/test_context_optimizer_iterdir_indexes.py @@ -190,8 +190,8 @@ def test_instruction_relevance_no_listdir(self, tmp_path: Path) -> None: optimizer._directory_cache[tmp_path / "src"].pattern_matches.clear() with patch( - "apm_cli.compilation.context_optimizer.os.listdir", - side_effect=AssertionError("relevance must not rescan the directory"), + "apm_cli.compilation.context_optimizer.CompileInventory.collect", + side_effect=AssertionError("relevance must not recollect the inventory"), ): assert optimizer._is_instruction_relevant(instruction, tmp_path / "src") diff --git a/tests/unit/compilation/test_context_optimizer_single_walk.py b/tests/unit/compilation/test_context_optimizer_single_walk.py index 99290801bc..8987b89237 100644 --- a/tests/unit/compilation/test_context_optimizer_single_walk.py +++ b/tests/unit/compilation/test_context_optimizer_single_walk.py @@ -8,8 +8,6 @@ from __future__ import annotations -import os -from collections.abc import Iterator from pathlib import Path from unittest.mock import patch @@ -23,7 +21,7 @@ def _make_instruction( name: str = "inst", - apply_to: str = "**/*.py", + apply_to: str | None = "**/*.py", ) -> Instruction: return Instruction( name=name, @@ -40,39 +38,13 @@ def _touch(base: Path, rel: str) -> None: p.touch() -def _walk_with_nested_subtree( - base: Path, - subtree_name: str, - descended: list[Path], -) -> Iterator[tuple[str, list[str], list[str]]]: - root_dirs = [subtree_name] - yield str(base), root_dirs, ["app.py"] - if subtree_name not in root_dirs: - return +class TestCompileInventoryProjection: + """Regression coverage for full accounting and scoped candidate projection.""" - subtree = base / subtree_name - child_dirs = ["deep"] - yield str(subtree), child_dirs, [] - if child_dirs: - deep = subtree / "deep" - descended.append(deep) - yield str(deep), [], ["ignored.py"] - - -class TestSingleWalkPopulatesBothCaches: - """Regression: optimize_instruction_placement must call os.walk exactly once - and populate both _directory_cache and _file_list_cache in that single pass. - """ - - def test_single_walk_populates_directory_cache_and_file_list_cache( + def test_single_inventory_populates_directory_cache_and_file_list_cache( self, tmp_path: Path ) -> None: - """One os.walk traversal must populate both caches. - - Regression for the dual-walk footprint: before the fix, - _analyze_project_structure and _get_all_files each ran their own - os.walk, doubling filesystem I/O on every compile. - """ + """One inventory walk must populate both optimizer caches.""" _touch(tmp_path, "src/main.py") _touch(tmp_path, "src/utils.py") _touch(tmp_path, "tests/test_main.py") @@ -80,22 +52,16 @@ def test_single_walk_populates_directory_cache_and_file_list_cache( optimizer = ContextOptimizer(base_dir=str(tmp_path)) instruction = _make_instruction(apply_to="**/*.py") - real_walk = os.walk - walk_call_count = 0 - - def counting_walk(top, **kwargs): - nonlocal walk_call_count - walk_call_count += 1 - yield from real_walk(top, **kwargs) + from apm_cli.compilation.inventory import CompileInventory - with patch("apm_cli.compilation.context_optimizer.os.walk", side_effect=counting_walk): + with patch( + "apm_cli.compilation.context_optimizer.CompileInventory.collect", + wraps=CompileInventory.collect, + ) as collect: + optimizer = ContextOptimizer(base_dir=str(tmp_path)) optimizer.optimize_instruction_placement([instruction]) - assert walk_call_count == 1, ( - f"expected exactly 1 os.walk call, got {walk_call_count} " - "(dual-walk regression: _analyze_project_structure and _get_all_files " - "must share the same traversal result)" - ) + assert collect.call_count == 1 assert optimizer._directory_cache, "_directory_cache must be non-empty after optimization" assert optimizer._file_list_cache is not None, "_file_list_cache must not be None" @@ -110,30 +76,187 @@ def counting_walk(top, **kwargs): assert "utils.py" in file_names assert "test_main.py" in file_names + def test_literal_apply_to_prefix_prunes_unrelated_top_level_subtrees( + self, tmp_path: Path + ) -> None: + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "vendor/huge.txt") + + optimizer = ContextOptimizer(base_dir=str(tmp_path)) + optimizer.optimize_instruction_placement([_make_instruction(apply_to="src/**/*.py")]) + + assert tmp_path / "src/main.py" in optimizer._file_list_cache + assert tmp_path / "vendor/huge.txt" not in optimizer._file_list_cache + assert tmp_path / "vendor" in optimizer._directory_cache + + def test_literal_prefix_matches_full_walk_placement_at_project_scale( + self, tmp_path: Path + ) -> None: + """A 6,602-file tree scopes matching without changing placement output.""" + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "src/worker.py") + for directory_index in range(419): + file_count = 16 if directory_index < 315 else 15 + for file_index in range(file_count): + _touch(tmp_path, f"pkg-{directory_index:03d}/file-{file_index:02d}.txt") + + instruction = _make_instruction(apply_to="src/**/*.py") + scoped = ContextOptimizer(base_dir=str(tmp_path)) + scoped_match_calls: list[Path] = [] + original_file_matches = ContextOptimizer._file_matches_pattern + + def counting_file_matches(self: ContextOptimizer, file_path: Path, pattern: str) -> bool: + scoped_match_calls.append(file_path) + return original_file_matches(self, file_path, pattern) + + with patch.object(ContextOptimizer, "_file_matches_pattern", counting_file_matches): + scoped_placement = scoped.optimize_instruction_placement([instruction]) + + full = ContextOptimizer(base_dir=str(tmp_path)) + with patch( + "apm_cli.compilation.context_optimizer.literal_apply_to_top_level_roots", + return_value=None, + ): + full_placement = full.optimize_instruction_placement([instruction]) + + def placement_snapshot( + placement: dict[Path, list[Instruction]], + ) -> list[tuple[str, tuple[str, ...]]]: + return sorted( + ( + str(directory.relative_to(tmp_path)), + tuple(instruction.name for instruction in instructions), + ) + for directory, instructions in placement.items() + ) + + assert len(full._directory_cache) == 420 + assert len(scoped._directory_cache) == 420 + assert len(full._file_list_cache) == 6602 + assert len(scoped._file_list_cache) == 2 + assert placement_snapshot(scoped_placement) == placement_snapshot(full_placement) + assert placement_snapshot(scoped_placement) == [("src", ("inst",))] + assert set(scoped_match_calls) <= set(scoped._file_list_cache) + assert len(scoped_match_calls) <= len(scoped._file_list_cache) + assert scoped._optimization_decisions[0].matching_directories == 1 + assert full._optimization_decisions[0].matching_directories == 1 + + @pytest.mark.parametrize( + "apply_to", + [ + None, + "**/*.py", + "*.py", + "{src,docs}/**", + "*/src/**/*.py", + "src/**/*.py,**/*.md", + "src/{api,cli/**", + ], + ) + def test_unprovable_prefix_keeps_full_walk(self, tmp_path: Path, apply_to: str | None) -> None: + """A global or ambiguous segment kills the root-pruning mutation.""" + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "vendor/huge.txt") + + optimizer = ContextOptimizer(base_dir=str(tmp_path)) + optimizer.optimize_instruction_placement([_make_instruction(apply_to=apply_to)]) + + assert tmp_path / "vendor/huge.txt" in optimizer._file_list_cache + assert tmp_path / "vendor" in optimizer._directory_cache + + def test_unscoped_candidates_do_not_build_redundant_full_file_projection( + self, tmp_path: Path + ) -> None: + """Global fallback reuses each inventory directory's complete file list.""" + from apm_cli.compilation.inventory import CompileInventory + + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "vendor/huge.txt") + inventory = CompileInventory.collect(tmp_path) + optimizer = ContextOptimizer(base_dir=str(tmp_path), inventory=inventory) + + with patch.object( + CompileInventory, + "files_under", + side_effect=AssertionError("unscoped projection must reuse complete directory files"), + ): + optimizer.optimize_instruction_placement([_make_instruction(apply_to="**/*.py")]) + + assert set(optimizer._file_list_cache) == { + tmp_path / "src/main.py", + tmp_path / "vendor/huge.txt", + } + + def test_comma_list_unions_ten_literal_roots(self, tmp_path: Path) -> None: + """Comma lists retain every literal root and prune unrelated siblings.""" + roots = [f"package-{index:02d}" for index in range(10)] + for root in roots: + _touch(tmp_path, f"{root}/src/main.py") + _touch(tmp_path, "vendor/huge.txt") + apply_to = ",".join(f"{root}/src/**/*.py" for root in roots) + + optimizer = ContextOptimizer(base_dir=str(tmp_path)) + optimizer.optimize_instruction_placement([_make_instruction(apply_to=apply_to)]) + + assert set(optimizer._scan_top_level_roots or ()) == set(roots) + assert tmp_path / "vendor/huge.txt" not in optimizer._file_list_cache + assert {path.parent.parent.name for path in optimizer._file_list_cache} == set(roots) + assert optimizer._optimization_decisions[0].matching_directories == len(roots) + + def test_hidden_root_and_literal_root_are_scanned_together(self, tmp_path: Path) -> None: + """A targeted hidden root remains eligible alongside ordinary roots.""" + _touch(tmp_path, ".github/instructions/guide.md") + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "vendor/huge.txt") + + optimizer = ContextOptimizer(base_dir=str(tmp_path)) + optimizer.optimize_instruction_placement( + [_make_instruction(apply_to=".github/**/*.md,src/**/*.py")] + ) + + assert tmp_path / ".github/instructions/guide.md" in optimizer._file_list_cache + assert tmp_path / "src/main.py" in optimizer._file_list_cache + assert tmp_path / "vendor/huge.txt" not in optimizer._file_list_cache + assert optimizer._optimization_decisions[0].matching_directories == 2 + + def test_excluded_literal_root_stays_excluded(self, tmp_path: Path) -> None: + """Configured exclusions still win when applyTo names their root.""" + _touch(tmp_path, "src/main.py") + _touch(tmp_path, "vendor/generated.py") + + optimizer = ContextOptimizer(base_dir=str(tmp_path), exclude_patterns=["vendor"]) + optimizer.optimize_instruction_placement([_make_instruction(apply_to="vendor/**/*.py")]) + + assert tmp_path / "vendor/generated.py" not in optimizer._file_list_cache + assert tmp_path / "vendor" not in optimizer._directory_cache + assert optimizer._optimization_decisions[0].matching_directories == 0 + + def test_universal_scan_does_not_follow_directory_symlinks(self, tmp_path: Path) -> None: + """The root filter leaves the no-follow traversal contract unchanged.""" + _touch(tmp_path, "src/main.py") + (tmp_path / "linked").symlink_to(tmp_path / "src", target_is_directory=True) + + optimizer = ContextOptimizer(base_dir=str(tmp_path)) + optimizer.optimize_instruction_placement([_make_instruction(apply_to="**/*.py")]) + + assert tmp_path / "src/main.py" in optimizer._file_list_cache + assert tmp_path / "linked" not in optimizer._directory_cache + assert optimizer._optimization_decisions[0].matching_directories == 1 + class TestGetAllFilesRoutesThroughAnalyze: - """_get_all_files must route through _analyze_project_structure when cache is None.""" + """_get_all_files must project the inventory only once.""" def test_get_all_files_before_optimize_triggers_single_walk(self, tmp_path: Path) -> None: - """_get_all_files called before optimize must use _analyze_project_structure.""" + """_get_all_files called before optimize must project the inventory.""" _touch(tmp_path, "app.py") _touch(tmp_path, "lib/helper.py") optimizer = ContextOptimizer(base_dir=str(tmp_path)) assert optimizer._file_list_cache is None - real_walk = os.walk - walk_call_count = 0 + files = optimizer._get_all_files() - def counting_walk(top, **kwargs): - nonlocal walk_call_count - walk_call_count += 1 - yield from real_walk(top, **kwargs) - - with patch("apm_cli.compilation.context_optimizer.os.walk", side_effect=counting_walk): - files = optimizer._get_all_files() - - assert walk_call_count == 1 assert optimizer._file_list_cache is not None assert files is optimizer._file_list_cache assert set(optimizer._directory_cache) == {tmp_path, tmp_path / "lib"} @@ -157,24 +280,14 @@ def test_optimize_rebuilds_direct_cache_for_selected_hidden_root(self, tmp_path: assert tmp_path / ".github/instructions" in optimizer._directory_cache def test_get_all_files_reuses_cache_without_second_walk(self, tmp_path: Path) -> None: - """After the first walk, a second call to _get_all_files reuses the cache.""" + """After the first projection, a second call reuses the cache.""" _touch(tmp_path, "app.py") optimizer = ContextOptimizer(base_dir=str(tmp_path)) first = optimizer._get_all_files() - real_walk = os.walk - walk_call_count = 0 - - def counting_walk(top, **kwargs): - nonlocal walk_call_count - walk_call_count += 1 - yield from real_walk(top, **kwargs) + second = optimizer._get_all_files() - with patch("apm_cli.compilation.context_optimizer.os.walk", side_effect=counting_walk): - second = optimizer._get_all_files() - - assert walk_call_count == 0, "second _get_all_files call must not walk again" assert first is second @@ -212,51 +325,6 @@ def test_configurable_exclude_patterns_prune_both_caches(self, tmp_path: Path) - assert "vendor" not in dir_names assert "lib.py" not in file_names - def test_default_exclusion_safety_net_prunes_nested_subtree(self, tmp_path: Path) -> None: - """The safety net stops descent when parent pruning is bypassed.""" - _touch(tmp_path, "app.py") - descended: list[Path] = [] - optimizer = ContextOptimizer(base_dir=str(tmp_path)) - - with ( - patch( - "apm_cli.compilation.context_optimizer.os.walk", - side_effect=lambda _top: _walk_with_nested_subtree( - tmp_path, - "node_modules", - descended, - ), - ), - patch.object(optimizer, "_should_exclude_subdir", return_value=False), - ): - optimizer._analyze_project_structure() - - assert descended == [] - - def test_configurable_exclusion_safety_net_prunes_nested_subtree( - self, - tmp_path: Path, - ) -> None: - """A current-path exclusion stops descent when parent pruning is bypassed.""" - _touch(tmp_path, "app.py") - descended: list[Path] = [] - optimizer = ContextOptimizer(base_dir=str(tmp_path), exclude_patterns=["vendor"]) - - with ( - patch( - "apm_cli.compilation.context_optimizer.os.walk", - side_effect=lambda _top: _walk_with_nested_subtree( - tmp_path, - "vendor", - descended, - ), - ), - patch.object(optimizer, "_should_exclude_subdir", return_value=False), - ): - optimizer._analyze_project_structure() - - assert descended == [] - class TestHiddenToolRootsInUnifiedWalk: """Supported hidden-tool roots in applyTo are admitted; others are pruned.""" diff --git a/tests/unit/primitives/test_discovery_parser.py b/tests/unit/primitives/test_discovery_parser.py index ac7595963d..ea569d4349 100644 --- a/tests/unit/primitives/test_discovery_parser.py +++ b/tests/unit/primitives/test_discovery_parser.py @@ -8,6 +8,7 @@ from pathlib import Path from unittest.mock import MagicMock, patch +from apm_cli.compilation.inventory import CompileInventory from apm_cli.primitives.discovery import ( _discover_local_skill, _discover_skill_in_directory, @@ -309,6 +310,45 @@ def test_with_apm_dir_discovers_primitives(self): self.assertEqual(len(collection.instructions), 1) self.assertEqual(collection.instructions[0].source, "dependency:owner/repo") + def test_inventory_scan_skips_symlinked_dependency_primitive(self): + """Inventory-backed dependency discovery does not dereference file links.""" + dep_dir = Path(self.tmp) / "owner" / "repo" + target = Path(self.tmp) / "outside.instructions.md" + _write(target, INSTRUCTION_CONTENT) + link = dep_dir / ".apm" / "instructions" / "linked.instructions.md" + link.parent.mkdir(parents=True) + try: + link.symlink_to(target) + except OSError: + self.skipTest("file symlinks are unavailable on this platform") + + collection = PrimitiveCollection() + scan_directory_with_source( + dep_dir, + collection, + source="dependency:owner/repo", + inventory=CompileInventory.collect(Path(self.tmp)), + ) + + self.assertEqual(len(collection.instructions), 0) + + def test_dependency_skill_symlink_is_not_discovered(self): + """Dependency skills do not load content from outside their package.""" + dep_dir = Path(self.tmp) / "owner" / "repo" + target = Path(self.tmp) / "outside-skill.md" + _write(target, SKILL_CONTENT) + link = dep_dir / "SKILL.md" + link.parent.mkdir(parents=True) + try: + link.symlink_to(target) + except OSError: + self.skipTest("file symlinks are unavailable on this platform") + + collection = PrimitiveCollection() + scan_directory_with_source(dep_dir, collection, source="dependency:owner/repo") + + self.assertEqual(len(collection.skills), 0) + def test_parse_error_in_dep_primitive_warns_and_continues(self): dep_dir = Path(self.tmp) / "owner" / "repo" _write( diff --git a/tests/unit/utils/test_patterns.py b/tests/unit/utils/test_patterns.py index cc5a1fb3b7..b6e74a27c4 100644 --- a/tests/unit/utils/test_patterns.py +++ b/tests/unit/utils/test_patterns.py @@ -2,6 +2,7 @@ from apm_cli.utils.patterns import ( has_top_level_comma, + literal_apply_to_top_level_roots, normalize_apply_to, parse_apply_to, yaml_double_quote, @@ -55,6 +56,12 @@ def test_brace_alternation_mixed_with_top_level_comma(self): "**/*.py", ] + def test_character_class_comma_is_not_a_list_separator(self): + assert parse_apply_to("src/[a,b]/**/*.py,docs/**/*.md") == [ + "src/[a,b]/**/*.py", + "docs/**/*.md", + ] + def test_nested_braces(self): assert parse_apply_to("**/{a,{b,c}},**/*.py") == [ "**/{a,{b,c}}", @@ -108,6 +115,40 @@ def test_list_literal_comma_preserves_pattern_boundary(self): assert parse_apply_to(normalized) == ["src/foo,bar/*.py", "**/*.pyi"] +class TestLiteralApplyToTopLevelRoots: + """Tests for conservative traversal-root analysis.""" + + def test_unions_literal_roots_across_expressions_and_comma_lists(self): + roots = literal_apply_to_top_level_roots( + ["src/**/*.py,docs/**/*.md", "packages/*/src/**/*.py"] + ) + + assert roots == frozenset({"src", "docs", "packages"}) + + def test_retains_literal_prefix_before_later_brace_glob(self): + assert literal_apply_to_top_level_roots(["src/{api,cli}/**"]) == frozenset({"src"}) + + def test_returns_none_for_global_or_unprovable_patterns(self): + unprovable = [ + [None], + [""], + ["**/*.py"], + ["*.py"], + ["*/src/**/*.py"], + ["[sd]rc/**/*.py"], + ["{src,docs}/**"], + ["src/**/*.py,**/*.md"], + ["src/{api,cli/**"], + ["/src/**/*.py"], + ["../src/**/*.py"], + [r"src\**\*.py"], + [r"src/foo\,bar/**/*.py"], + ] + + for patterns in unprovable: + assert literal_apply_to_top_level_roots(patterns) is None + + class TestYamlDoubleQuote: """Unit tests for yaml_double_quote() defence-in-depth escaping."""