fix(install): preserve refused skill files when deployment receipts are missing

src/apm_cli/integration/base_integrator.py

@@ -358,7 +358,13 @@ def copy_directory_files(
                         "Preserve your changes before reinstalling with --force."
                     )
                     if diagnostics is not None:
-                        diagnostics.warn(message)
+                        if expected is None:
+                            # A refused write must not subsequently become a
+                            # stale deletion. Abort before cleanup/lock commit
+                            # and report an incomplete install to the caller.
+                            diagnostics.error(message)
+                        else:
+                            diagnostics.warn(message)
                     else:
                         _rich_warning(message)
                     if on_member_skipped is not None:

src/apm_cli/integration/skill_integrator.py

@@ -1108,7 +1108,9 @@ def _promote_sub_skills_standalone(
                 parent_name,
                 warn=is_primary,
                 owned_by=owned_by if is_primary else None,
-                diagnostics=diagnostics if is_primary else None,
+                # Every physical target must report safety errors; warn/logger
+                # still restrict duplicate promotion summaries to the primary.
+                diagnostics=diagnostics,
                 managed_files=managed_files,
                 force=force,
                 project_root=project_root,
@@ -1333,7 +1335,7 @@ def _integrate_native_skill(
                 current_key or skill_name,
                 warn=is_primary,
                 owned_by=owned_by if is_primary else None,
-                diagnostics=diagnostics if is_primary else None,
+                diagnostics=diagnostics,
                 managed_files=managed_files,
                 force=force,
                 project_root=project_root,
@@ -1443,7 +1445,7 @@ def _integrate_skill_bundle(
                 parent_name,
                 warn=is_primary,
                 owned_by=owned_by if is_primary else None,
-                diagnostics=diagnostics if is_primary else None,
+                diagnostics=diagnostics,
                 managed_files=managed_files,
                 force=force,
                 project_root=project_root,

tests/integration/test_missing_receipt_cleanup.py

@@ -0,0 +1,91 @@
+"""Run the source CLI in a disposable home; never touch live APM state."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+from pathlib import Path
+
+import pytest
+
+from tests.utils.apm_lifecycle_runner import ApmLifecycleRunner
+from tests.utils.isolated_apm_environment import IsolatedApmEnvironment
+
+pytestmark = [pytest.mark.integration]
+
+
+@pytest.mark.parametrize("user_scope", [False, True])
+@pytest.mark.parametrize("edited", [False, True])
+@pytest.mark.parametrize("package_kind", ["apm_package", "claude_skill", "skill_bundle"])
+@pytest.mark.parametrize("secondary_only", [False, True])
+def test_missing_receipt_cannot_turn_refused_writes_into_stale_deletions(
+    tmp_path: Path,
+    apm_engine_command,
+    user_scope: bool,
+    edited: bool,
+    package_kind: str,
+    secondary_only: bool,
+):
+    isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ))
+    env = isolated.subprocess_env()
+    project = isolated.work_root / "consumer"
+    project.mkdir(parents=True)
+    root = isolated.config_root if user_scope else project
+    root.mkdir(parents=True, exist_ok=True)
+    package = isolated.work_root / "package"
+    source = (
+        package
+        / {"apm_package": ".apm/skills/demo", "claude_skill": "", "skill_bundle": "skills/demo"}[
+            package_kind
+        ]
+    )
+    source.mkdir(parents=True)
+    if package_kind != "claude_skill":
+        (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n")
+    (source / "SKILL.md").write_text("---\nname: demo\ndescription: Receipt fixture\n---\nv1\n")
+    (source / "a.md").write_text("v1\n")
+    (source / "b.md").write_text("v1\n")
+    targets = "agent-skills,claude" if secondary_only else "agent-skills"
+    target_manifest = "\n".join(f"- {target}" for target in targets.split(","))
+    (root / "apm.yml").write_text(
+        f"name: consumer\nversion: 1.0.0\ndependencies:\n  apm:\n  - path: {package}\ntargets:\n{target_manifest}\n"
+    )
+    runner = ApmLifecycleRunner(apm_engine_command)
+    args = ("install", *(("-g",) if user_scope else ()), "--target", targets, "--no-policy")
+    first = runner.run(args, cwd=project, env=env)
+    assert first.returncode == 0, first.stdout + first.stderr
+    deploy_root = isolated.home if user_scope else project
+    secondary_root = ".claude/skills" if secondary_only else ".agents/skills"
+    deployed_skills = list((deploy_root / secondary_root).glob("*/SKILL.md"))
+    assert len(deployed_skills) == 1
+    deployed = deployed_skills[0].parent
+    if secondary_only:
+        # Only the secondary target retains old-lock files. The primary can
+        # install successfully, so its diagnostics cannot mask a lost error.
+        primary = deploy_root / ".agents/skills"
+        assert len(list(primary.glob("*/SKILL.md"))) == 1
+        shutil.rmtree(primary)
+    if edited:
+        (deployed / "SKILL.md").write_text("my local edits\n")
+    before = {p.name: p.read_bytes() for p in deployed.iterdir()}
+    lock_before = (root / "apm.lock.yaml").read_bytes()
+    receipts = list((isolated.home / ".apm").glob("deployment-receipts*/*.json"))
+    assert len(receipts) == 1
+    assert json.loads(receipts[0].read_text())["version"] in (1, 2)
+    receipts[0].unlink()
+    if package_kind != "claude_skill":
+        (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.1.0\n")
+    (source / "a.md").write_text("v2\n")
+    (source / "b.md").unlink()
+    (source / "c.md").write_text("v2\n")
+    for _ in range(2):
+        result = runner.run(args, cwd=project, env=env)
+        assert result.returncode != 0, result.stdout + result.stderr
+        assert "no trustworthy local deployment baseline" in " ".join(
+            (result.stdout + result.stderr).split()
+        )
+        assert "stale files cleaned" not in result.stdout
+        for name, content in before.items():
+            assert (deployed / name).read_bytes() == content
+        assert (root / "apm.lock.yaml").read_bytes() == lock_before

tests/integration/test_obsolete_dependency_claim_lifecycle.py

@@ -102,14 +102,28 @@ def audit(*, ci: bool, expected: int = 0) -> dict:
     extra.unlink()
     if not already_absent and historical != "concrete":
         # Source removal alone cannot authorize deletion of an unattributed file.
-        run(*install)
+        prior_lock = lock_path.read_bytes()
+        prior_outputs = {path: (deploy_root / path).read_bytes() for path in obsolete | surviving}
+        for _ in range(2):
+            refused = run(*install, expected=1)
+            assert "no trustworthy local deployment baseline" in " ".join(refused.stdout.split())
+            assert lock_path.read_bytes() == prior_lock
+            assert {
+                path: (deploy_root / path).read_bytes() for path in prior_outputs
+            } == prior_outputs
         retained = load_yaml(lock_path)
         shared = ".agents/skills/demo/extra.md"
         assert (deploy_root / shared).read_bytes() == b"Old dependency asset.\n"
         assert shared in retained["dependencies"][0]["deployed_files"]
         assert shared in retained["dependencies"][0]["deployed_file_hashes"]
-        assert any(row["value"] == shared for row in retained["deployments"])
-        (deploy_root / shared).unlink()
+        if historical == "legacy":
+            assert any(row["value"] == shared for row in retained["deployments"])
+        else:
+            assert "deployments" not in retained
+        # Model explicit removal of the retained assets, then exercise normal
+        # claim retirement below without granting an overwrite through --force.
+        for path in obsolete:
+            (deploy_root / path).unlink()
     if already_absent:
         for path in obsolete:
             (deploy_root / path).unlink()
@@ -199,8 +213,19 @@ def test_historical_absence_respects_live_preservation_boundaries(
     elif protection in {"untracked-content", "trusted-untracked-content"}:
         untracked.write_bytes(b"Untracked personal content.\n")
     arguments = ("--target", "claude") if protection == "unprocessed-target" else ()
-    expected_exit = 1 if protection == "required-selection" else 0
+    refuses_install = protection in {"edited-marker", "untracked-content", "required-selection"}
+    expected_exit = 1 if refuses_install else 0
     prior_bytes = lock_path.read_bytes()
+
+    def deployed_bytes():
+        return {
+            str(path.relative_to(scenario.deploy_root)): path.read_bytes()
+            for root in (".agents", ".claude")
+            for path in (scenario.deploy_root / root / "skills/s").rglob("*")
+            if path.is_file()
+        }
+
+    prior_files = deployed_bytes()
     for _ in range(2):
         result = scenario.run("install", *arguments, expected=expected_exit)
         repaired = load_yaml(lock_path)
@@ -224,13 +249,14 @@ def test_historical_absence_respects_live_preservation_boundaries(
             assert personal not in dependency["deployed_files"]
             assert personal not in dependency["deployed_file_hashes"]
             assert all(row["value"] != personal for row in repaired["deployments"])
-        if protection == "untracked-content":
+        if protection in {"edited-marker", "untracked-content"}:
             assert "no trustworthy local deployment baseline" in " ".join(result.stdout.split())
         assert not (scenario.deploy_root / shared).exists()
         if protection == "edited-marker":
             assert marker.read_bytes() == b"User-edited skill marker.\n"
-        elif protection == "required-selection":
+        if refuses_install:
             assert lock_path.read_bytes() == prior_bytes
+            assert deployed_bytes() == prior_files
 
 
 @pytest.mark.parametrize("user_scope", [False, True], ids=["project", "user"])