fix(install): harden stale-file cleanup with per-file content-hash provenance (#666 follow-up)

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Harden `apm install` stale-file cleanup to prevent unsafe lockfile deletions, preserve user-edited files via per-file SHA-256 provenance, and improve cleanup reporting during install and `--dry-run` (#666, #762)
 - Fix `apm marketplace add` silently failing for private repos by using credentials when probing `marketplace.json` (#701)
 - Harden marketplace plugin normalization to enforce that manifest-declared `agents`/`skills`/`commands`/`hooks` paths resolve inside the plugin root (#760)
 - Stop `test_auto_detect_through_proxy` from making real `api.github.com` calls by passing a mock `auth_resolver`, fixing flaky macOS CI rate-limit failures (#759)

docs/src/content/docs/reference/cli-commands.md

@@ -115,9 +115,20 @@ Exceptions:
 - MCP servers already configured but with changed manifest config are re-applied automatically (`updated`)
 - APM packages removed from `apm.yml` have their deployed files cleaned up on the next full `apm install`
 - APM packages whose ref/version changed in `apm.yml` are re-downloaded automatically (no `--update` needed)
-- Files previously deployed by a still-present package that are no longer produced (e.g. the package renamed or removed a primitive) are removed from disk and from `apm.lock.yaml` `deployed_files` on the next `apm install`
 - `--force` remains available for full overwrite/reset scenarios
 
+**Stale-file cleanup:**
+
+`apm install` removes files that a still-present package previously deployed but no longer produces -- for example after a package renames or drops a primitive. This keeps the workspace consistent with the manifest without any manual `apm prune`/`uninstall` step. Behaviour:
+
+- Scope: only files recorded under that package's `deployed_files` in `apm.lock.yaml` are eligible
+- Safety gate: paths that escape the project root or fall outside known integration prefixes are refused
+- Directory entries are refused outright -- APM only deletes individual files
+- Per-file provenance: APM records a content hash for each deployed file; if the on-disk content has changed since deploy time the file is treated as user-edited and kept (with a warning explaining how to remove it manually)
+- Skipped when integration reports an error for the package (avoids deleting a file that just failed to redeploy)
+- Files that fail to delete are kept in `deployed_files` and retried on the next `apm install`
+- Use `apm install --dry-run` to preview package-level orphan cleanup; intra-package stale cleanup is not previewed because it requires running integration
+
 **Examples:**
 ```bash
 # Install all dependencies from apm.yml

src/apm_cli/core/command_logger.py

@@ -170,6 +170,7 @@ def __init__(
     ):
         super().__init__("install", verbose=verbose, dry_run=dry_run)
         self.partial = partial  # True when specific packages are passed to `apm install`
+        self._stale_cleaned_total = 0  # Accumulated by stale_cleanup / orphan_cleanup
 
     # --- Validation phase ---
 
@@ -305,10 +306,76 @@ def package_type_info(self, type_label: str):
             return
         _rich_echo(f"    Package type: {type_label}", color="dim")
 
+    # --- Cleanup phase (stale and orphan file removal) ---
+
+    def stale_cleanup(self, dep_key: str, count: int):
+        """Log per-package stale-file cleanup outcome at default verbosity.
+
+        Stale-file deletion is a destructive operation in the user's
+        tracked workspace (unlike npm's ``node_modules``); it must be
+        visible without ``--verbose``. Rendered as an info line so it
+        groups visually with other phase messages, not as a tree item
+        (the originating package line was emitted earlier in the install
+        sequence and is no longer adjacent).
+        """
+        if count <= 0:
+            return
+        self._stale_cleaned_total += count
+        noun = "file" if count == 1 else "files"
+        _rich_info(f"Cleaned {count} stale {noun} from {dep_key}", symbol="info")
+
+    def orphan_cleanup(self, count: int):
+        """Log post-install orphan-file cleanup outcome at default verbosity.
+
+        Same visibility rationale as :meth:`stale_cleanup`: file deletion
+        in the user's workspace must be visible by default.
+        """
+        if count <= 0:
+            return
+        self._stale_cleaned_total += count
+        noun = "file" if count == 1 else "files"
+        _rich_info(
+            f"Cleaned {count} {noun} from packages no longer in apm.yml",
+            symbol="info",
+        )
+
+    @property
+    def stale_cleaned_total(self) -> int:
+        """Total files removed by stale + orphan cleanup during this install."""
+        return self._stale_cleaned_total
+
+    def cleanup_skipped_user_edit(self, rel_path: str, dep_key: str):
+        """Log a stale-file deletion that was skipped because the user
+        edited the file after APM deployed it.
+
+        Yellow inline at default verbosity -- the user needs to know APM
+        kept the file and a manual decision is pending.
+        """
+        _rich_warning(
+            f"  Kept user-edited file {rel_path} (from {dep_key}); "
+            "delete manually if no longer needed",
+            symbol="warning",
+        )
+
     # --- Install summary ---
 
-    def install_summary(self, apm_count: int, mcp_count: int, errors: int = 0):
-        """Log final install summary."""
+    def install_summary(
+        self,
+        apm_count: int,
+        mcp_count: int,
+        errors: int = 0,
+        stale_cleaned: int = 0,
+    ):
+        """Log final install summary.
+
+        Args:
+            apm_count: Number of APM dependencies installed.
+            mcp_count: Number of MCP servers installed.
+            errors: Number of errors collected during install.
+            stale_cleaned: Total stale + orphan files removed during
+                this install. Reported as a parenthetical so existing
+                callers and assertion patterns continue to work.
+        """
         parts = []
         if apm_count > 0:
             noun = "dependency" if apm_count == 1 else "dependencies"
@@ -317,14 +384,22 @@ def install_summary(self, apm_count: int, mcp_count: int, errors: int = 0):
             noun = "server" if mcp_count == 1 else "servers"
             parts.append(f"{mcp_count} MCP {noun}")
 
+        cleanup_suffix = ""
+        if stale_cleaned > 0:
+            file_noun = "file" if stale_cleaned == 1 else "files"
+            cleanup_suffix = f" ({stale_cleaned} stale {file_noun} cleaned)"
+
         if parts:
             summary = " and ".join(parts)
             if errors > 0:
                 _rich_warning(
-                    f"Installed {summary} with {errors} error(s).", symbol="warning"
+                    f"Installed {summary}{cleanup_suffix} with {errors} error(s).",
+                    symbol="warning",
                 )
             else:
-                _rich_success(f"Installed {summary}.", symbol="sparkles")
+                _rich_success(
+                    f"Installed {summary}{cleanup_suffix}.", symbol="sparkles"
+                )
         elif errors > 0:
             _rich_error(
                 f"Installation failed with {errors} error(s).", symbol="error"

src/apm_cli/deps/lockfile.py

@@ -32,6 +32,7 @@ class LockedDependency:
     resolved_by: Optional[str] = None
     package_type: Optional[str] = None
     deployed_files: List[str] = field(default_factory=list)
+    deployed_file_hashes: Dict[str, str] = field(default_factory=dict)
     source: Optional[str] = None  # "local" for local deps, None/absent for remote
     local_path: Optional[str] = None  # Original local path (relative to project root)
     content_hash: Optional[str] = None  # SHA-256 of package file tree
@@ -72,6 +73,10 @@ def to_dict(self) -> Dict[str, Any]:
             result["package_type"] = self.package_type
         if self.deployed_files:
             result["deployed_files"] = sorted(self.deployed_files)
+        if self.deployed_file_hashes:
+            result["deployed_file_hashes"] = dict(
+                sorted(self.deployed_file_hashes.items())
+            )
         if self.source:
             result["source"] = self.source
         if self.local_path:
@@ -116,6 +121,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "LockedDependency":
             resolved_by=data.get("resolved_by"),
             package_type=data.get("package_type"),
             deployed_files=deployed_files,
+            deployed_file_hashes=dict(data.get("deployed_file_hashes") or {}),
             source=data.get("source"),
             local_path=data.get("local_path"),
             content_hash=data.get("content_hash"),
@@ -183,6 +189,7 @@ class LockFile:
     mcp_servers: List[str] = field(default_factory=list)
     mcp_configs: Dict[str, dict] = field(default_factory=dict)
     local_deployed_files: List[str] = field(default_factory=list)
+    local_deployed_file_hashes: Dict[str, str] = field(default_factory=dict)
 
     def add_dependency(self, dep: LockedDependency) -> None:
         """Add a dependency to the lock file."""
@@ -217,6 +224,10 @@ def to_yaml(self) -> str:
             data["mcp_configs"] = dict(sorted(self.mcp_configs.items()))
         if self.local_deployed_files:
             data["local_deployed_files"] = sorted(self.local_deployed_files)
+        if self.local_deployed_file_hashes:
+            data["local_deployed_file_hashes"] = dict(
+                sorted(self.local_deployed_file_hashes.items())
+            )
         from ..utils.yaml_io import yaml_to_str
         return yaml_to_str(data)
 
@@ -238,6 +249,9 @@ def from_yaml(cls, yaml_str: str) -> "LockFile":
         lock.mcp_servers = list(data.get("mcp_servers", []))
         lock.mcp_configs = dict(data.get("mcp_configs") or {})
         lock.local_deployed_files = list(data.get("local_deployed_files", []))
+        lock.local_deployed_file_hashes = dict(
+            data.get("local_deployed_file_hashes") or {}
+        )
         return lock
 
     def write(self, path: Path) -> None: