fix: apm install no longer permanently blocked by policy after lockfile wipe (required-packages-deployed catch-22)

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Non-skill integrators (agent, instruction, prompt, command, hook script-copy) silently adopt byte-identical pre-existing files so a degraded `deployed_files=[]` lockfile no longer permanently blocks installs gated by `required-packages-deployed`. (#1313)
 - `apm audit` drift check now returns skip-with-info (`passed=True`) when the install cache is cold, instead of failing the audit; bare `apm audit` surfaces the skip reason on stderr so CI pipelines that have not yet run `apm install` are not incorrectly red-marked. (#1289)
 - `extends: org` now correctly layers `dependencies.require` and `dependencies.deny` from the parent policy when the child omits the `dependencies:` block entirely; `None` signals "no opinion" (transparent) while `[]` signals explicit override. (#1290)
 - CI self-check job now uses `setup-only: true` + `apm audit --ci --no-drift` so managed files are not overwritten by `apm install` before `content-integrity` runs; documented the audit-only CI pattern and the install-before-audit blind spot in the enterprise and CI/CD guides. (#1291)

src/apm_cli/install/services.py

@@ -228,14 +228,15 @@ def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[
 
     # Aggregate per-primitive across targets so we emit ONE line per kind
     # (per the 1/2/3+ collapse rule), not one per target.
-    # Structure: { prim_name: {"files": int, "label": str, "paths": [str]} }
+    # Structure: { prim_name: {"files": int, "adopted": int, "label": str, "paths": [str]} }
     _per_kind: dict[str, dict[str, Any]] = {}
 
     for _prim_name, _entry in _dispatch.items():
         if _entry.multi_target:
             continue  # skills handled separately
         _integrator = _INTEGRATOR_KWARGS[_prim_name]
         _agg_files = 0
+        _agg_adopted = 0
         _agg_paths: list[str] = []
         _label = _prim_name
         for _target in targets:
@@ -253,9 +254,28 @@ def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[
             result["links_resolved"] += _int_result.links_resolved
             for tp in _int_result.target_paths:
                 deployed.append(_deployed_path_entry(tp, project_root, targets))
-            if _int_result.files_integrated <= 0:
+            _adopted_attr = getattr(_int_result, "files_adopted", 0)
+            # Coerce defensively: subclasses (e.g. HookIntegrationResult)
+            # always set this, but tests use MagicMock results which
+            # auto-attribute to MagicMock objects whose ``__int__`` is 1.
+            # Treat anything that is not a real int as 0 so we never
+            # invent fake adopt counts.
+            _adopted = _adopted_attr if isinstance(_adopted_attr, int) else 0
+            # Show the per-kind line whenever ANY work happened -- either
+            # a fresh integrate or a silent adopt of pre-existing
+            # byte-identical files. Adopt-only runs (e.g. re-install
+            # after lockfile wipe) used to print nothing here, which made
+            # the install summary look like a no-op even though the
+            # lockfile WAS being repopulated. Surfacing adopt counts
+            # restores operator trust in CI.
+            if _int_result.files_integrated <= 0 and _adopted <= 0:
                 continue
             _agg_files += _int_result.files_integrated
+            _agg_adopted += _adopted
+            # Only count fresh integrations against the package counter
+            # so totals like "3 prompts integrated" stay truthful;
+            # adopted files are surfaced separately in the per-kind
+            # line.
             result[_entry.counter_key] += _int_result.files_integrated
             _effective_root = _mapping.deploy_root or _target.root_dir
             _deploy_dir = (
@@ -278,9 +298,10 @@ def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[
                 _label = _prim_name
             _agg_paths.append(_deploy_dir)
 
-        if _agg_files > 0:
+        if _agg_files > 0 or _agg_adopted > 0:
             _per_kind[_prim_name] = {
                 "files": _agg_files,
+                "adopted": _agg_adopted,
                 "label": _label,
                 "paths": _agg_paths,
             }
@@ -291,12 +312,24 @@ def _format_target_collapse(paths: list[str], verbose: bool) -> tuple[str, list[
             continue
         _info = _per_kind[_prim_name]
         _suffix, _expansion = _format_target_collapse(_info["paths"], _verbose)
+        # Build the verb + count phrase. When at least one file was
+        # freshly integrated we lead with "N X integrated"; pure-adopt
+        # runs (no fresh writes) lead with "N X adopted" so the line
+        # still appears and the count is truthful.
+        _files = _info["files"]
+        _adopted = _info["adopted"]
+        if _files > 0:
+            _verb_phrase = f"{_files} {_info['label']} integrated"
+            if _adopted > 0:
+                _verb_phrase = f"{_verb_phrase} ({_adopted} adopted)"
+        else:
+            _verb_phrase = f"{_adopted} {_info['label']} adopted"
         if _expansion:
-            _log_integration(f"  |-- {_info['files']} {_info['label']} integrated:")
+            _log_integration(f"  |-- {_verb_phrase}:")
             for line in _expansion:
                 _log_integration(line)
         else:
-            _log_integration(f"  |-- {_info['files']} {_info['label']} integrated -> {_suffix}")
+            _log_integration(f"  |-- {_verb_phrase} -> {_suffix}")
 
     skill_result = skill_integrator.integrate_package_skill(
         package_info,

src/apm_cli/integration/agent_integrator.py

@@ -14,6 +14,7 @@
 import yaml
 
 from apm_cli.integration.base_integrator import BaseIntegrator, IntegrationResult
+from apm_cli.utils.path_security import PathTraversalError, ensure_path_within
 from apm_cli.utils.paths import portable_relpath
 
 if TYPE_CHECKING:
@@ -119,6 +120,7 @@ def integrate_agents_for_target(
 
         files_integrated = 0
         files_skipped = 0
+        files_adopted = 0
         target_paths: list[Path] = []
         total_links_resolved = 0
 
@@ -129,8 +131,36 @@ def integrate_agents_for_target(
                 target,
             )
             target_path = agents_dir / target_filename
+            # Defense-in-depth: target_filename comes from
+            # get_target_filename_for_target which strips path separators,
+            # but assert containment under agents_dir so a future
+            # regression cannot smuggle a traversal sequence past the
+            # adopt branch (which fires *before* check_collision and
+            # would otherwise blindly trust the computed path). Mirrors
+            # the guard already in command_integrator and
+            # instruction_integrator.
+            try:
+                ensure_path_within(target_path, agents_dir)
+            except PathTraversalError as exc:
+                if diagnostics is not None:
+                    diagnostics.warn(
+                        message=f"Rejected agent target path: {exc}",
+                        package=package_info.package.name,
+                    )
+                files_skipped += 1
+                continue
+
             rel_path = portable_relpath(target_path, project_root)
 
+            if self.is_content_identical_to_source(target_path, source_file):
+                # Pre-existing file is byte-identical to source -- silently
+                # adopt so deployed_files reflects reality. See
+                # BaseIntegrator.is_content_identical_to_source for the
+                # full rationale (catch-22 fix).
+                target_paths.append(target_path)
+                files_adopted += 1
+                continue
+
             if self.check_collision(
                 target_path,
                 rel_path,
@@ -160,6 +190,7 @@ def integrate_agents_for_target(
             files_skipped=files_skipped,
             target_paths=target_paths,
             links_resolved=total_links_resolved,
+            files_adopted=files_adopted,
         )
 
     def sync_for_target(
@@ -383,6 +414,7 @@ def integrate_package_agents(
 
         files_integrated = 0
         files_skipped = 0
+        files_adopted = 0
         target_paths: list[Path] = []
         total_links_resolved = 0
 
@@ -393,18 +425,31 @@ def integrate_package_agents(
                 copilot,
             )
             target_path = agents_dir / target_filename
-            rel_path = portable_relpath(target_path, project_root)
-
-            if self.check_collision(
-                target_path, rel_path, managed_files, force, diagnostics=diagnostics
-            ):
+            try:
+                ensure_path_within(target_path, agents_dir)
+            except PathTraversalError as exc:
+                if diagnostics is not None:
+                    diagnostics.warn(
+                        message=f"Rejected agent target path: {exc}",
+                        package=package_info.package.name,
+                    )
                 files_skipped += 1
                 continue
+            rel_path = portable_relpath(target_path, project_root)
 
-            links_resolved = self.copy_agent(source_file, target_path)
-            total_links_resolved += links_resolved
-            files_integrated += 1
-            target_paths.append(target_path)
+            if self.is_content_identical_to_source(target_path, source_file):
+                target_paths.append(target_path)
+                files_adopted += 1
+            else:
+                if self.check_collision(
+                    target_path, rel_path, managed_files, force, diagnostics=diagnostics
+                ):
+                    files_skipped += 1
+                    continue
+                links_resolved = self.copy_agent(source_file, target_path)
+                total_links_resolved += links_resolved
+                files_integrated += 1
+                target_paths.append(target_path)
 
             if claude_agents_dir:
                 claude_target = KNOWN_TARGETS["claude"]
@@ -414,8 +459,20 @@ def integrate_package_agents(
                     claude_target,
                 )
                 claude_path = claude_agents_dir / claude_filename
+                try:
+                    ensure_path_within(claude_path, claude_agents_dir)
+                except PathTraversalError as exc:
+                    if diagnostics is not None:
+                        diagnostics.warn(
+                            message=f"Rejected claude agent target path: {exc}",
+                            package=package_info.package.name,
+                        )
+                    continue
                 claude_rel = portable_relpath(claude_path, project_root)
-                if not self.check_collision(
+                if self.is_content_identical_to_source(claude_path, source_file):
+                    target_paths.append(claude_path)
+                    files_adopted += 1
+                elif not self.check_collision(
                     claude_path, claude_rel, managed_files, force, diagnostics=diagnostics
                 ):
                     self.copy_agent(source_file, claude_path)
@@ -429,8 +486,20 @@ def integrate_package_agents(
                     cursor_target,
                 )
                 cursor_path = cursor_agents_dir / cursor_filename
+                try:
+                    ensure_path_within(cursor_path, cursor_agents_dir)
+                except PathTraversalError as exc:
+                    if diagnostics is not None:
+                        diagnostics.warn(
+                            message=f"Rejected cursor agent target path: {exc}",
+                            package=package_info.package.name,
+                        )
+                    continue
                 cursor_rel = portable_relpath(cursor_path, project_root)
-                if not self.check_collision(
+                if self.is_content_identical_to_source(cursor_path, source_file):
+                    target_paths.append(cursor_path)
+                    files_adopted += 1
+                elif not self.check_collision(
                     cursor_path, cursor_rel, managed_files, force, diagnostics=diagnostics
                 ):
                     self.copy_agent(source_file, cursor_path)
@@ -442,6 +511,7 @@ def integrate_package_agents(
             files_skipped=files_skipped,
             target_paths=target_paths,
             links_resolved=total_links_resolved,
+            files_adopted=files_adopted,
         )
 
     # DEPRECATED: use get_target_filename_for_target(KNOWN_TARGETS["claude"], ...) instead.