fix: invalidate external step caches and restore backup preimages

Invalidate source-derived Python caches outside custom packages as well as in-package caches. Include extension configuration backup paths in existing component snapshots, preserving prior files, directories and absence. Cover equal-size/equal-mtime reloads, cache deletion failures, nine real rollback histories and backup I/O failures.

Assisted-by: GitHub Copilot (model: GPT-6 Astra, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: marcelsafin

docs/reference/bundles.md

@@ -59,7 +59,7 @@ specify bundle update [<bundle_id>]
 
 Re-resolves a bundle and **refreshes** its components through each primitive's update path, bringing already-installed components up to the bundle's newly pinned versions while preserving primitive-level overrides (such as preset priority). Provide a bundle id, or use `--all` to update everything installed.
 
-Before refreshing or removing an owned component, the bundler snapshots its installed files and registry metadata. If a component operation or provenance write fails, it attempts to restore those local snapshots, including disabled state, user configuration, extension hook settings, and generated command files and skill resources for current and previously active integrations, without downloading an older version. Previously absent outputs are also restored to absence. Custom steps are restored before dependent workflows. Recovery is best-effort and reports incomplete restoration; these temporary snapshots cover failures during the command, not process crashes or unrelated project files.
+Before refreshing or removing an owned component, the bundler snapshots its installed files and registry metadata. If a component operation or provenance write fails, it attempts to restore those local snapshots, including disabled state, user configuration, extension hook settings and pre-existing configuration backups, and generated command files and skill resources for current and previously active integrations, without downloading an older version. Previously absent outputs and configuration backups are also restored to absence. Custom steps are restored before dependent workflows. Recovery is best-effort and reports incomplete restoration; these temporary snapshots cover failures during the command, not process crashes or unrelated project files.
 
 > **Pin enforcement is install-time only.** Idempotency checks are id-based, not version-aware: a component that is already present is skipped during `install` without comparing its on-disk version to the manifest pin. Version pins are therefore guaranteed to be applied only when the bundler actually installs a component for the first time or refreshes it. Run `specify bundle update` to re-apply every owned component at its pinned version.
 

src/specify_cli/bundler/services/artifacts.py

@@ -1,4 +1,4 @@
-"""Component-scoped preimages of generated integration outputs."""
+"""Component-scoped preimages of manager-generated outputs and backups."""
 from __future__ import annotations
 
 import os
@@ -84,6 +84,7 @@ def snapshot_generated_artifacts(
                     )
             active_skills = manager._resolve_agent_skills_dir(active) if active else None
         else:
+            paths.add(manager.extensions_dir / ".backup" / snapshot.component.id)
             paths.update(manager._find_extension_skill_dirs(
                 skills, snapshot.component.id, create_skills_dir=False
             ))

src/specify_cli/workflows/__init__.py

@@ -189,6 +189,10 @@ def _load_custom_steps(project_root: Path, registry: dict[str, StepBase]) -> Non
                     raise OSError(f"Refusing symlinked bytecode cache: {cache_dir}")
                 if cache_dir.is_dir():
                     _shutil.rmtree(cache_dir)
+            # PYTHONPYCACHEPREFIX can place caches outside the step package.
+            for source_file in step_dir.rglob("*.py"):
+                cache_file = Path(_importlib_util.cache_from_source(str(source_file)))
+                cache_file.unlink(missing_ok=True)
             _importlib.invalidate_caches()
 
             # Treat the step directory as a proper package so that relative

tests/integration/test_bundler_state_rollback.py

@@ -137,6 +137,102 @@ def fail_save(*_args):
         assert HookExecutor(project).get_project_config()["hooks"] == original_hooks
 
 
+@pytest.mark.parametrize("installed_components", ["extensions"], indirect=True)
+@pytest.mark.parametrize("backup_state", ["absent", "empty", "contents"])
+@pytest.mark.parametrize("operation", ["refresh", "drop", "remove"])
+def test_save_failure_restores_extension_backup_preimage(
+    installed_components, monkeypatch, backup_state, operation,
+):
+    from specify_cli import _assets
+
+    project, _, manager_type, installer, plan, metadata = installed_components
+    backup_root = project / ".specify/extensions/.backup"
+    if backup_state != "absent":
+        owned_backup = backup_root / "owned"
+        owned_backup.mkdir(parents=True)
+        unrelated = backup_root / "unrelated"
+        unrelated.mkdir()
+        (unrelated / "saved-config.yml").write_text("unrelated\n", encoding="utf-8")
+        if backup_state == "contents":
+            (owned_backup / "owned-config.yml").write_text(
+                "setting: previous backup\n", encoding="utf-8"
+            )
+            (owned_backup / "notes").mkdir()
+            (owned_backup / "notes/context.txt").write_text(
+                "retained backup context\n", encoding="utf-8"
+            )
+
+    def backup_tree():
+        if not backup_root.exists():
+            return None
+        return {
+            str(path.relative_to(backup_root)): path.read_bytes() if path.is_file() else None
+            for path in backup_root.rglob("*")
+        }
+
+    before = backup_tree()
+    original_record = records_path(project).read_bytes()
+    source = _assets._locate_bundled_extension("owned")
+    (source / "replacement-config.yml").write_text("new defaults\n", encoding="utf-8")
+
+    def fail_save(*_args):
+        raise OSError("provenance write refused")
+
+    monkeypatch.setattr("specify_cli.bundler.services.installer.save_records", fail_save)
+    with pytest.raises(BundlerError, match="provenance write refused"):
+        if operation == "remove":
+            remove_bundle(project, "demo-bundle", installer)
+        else:
+            install_bundle(
+                project,
+                plan(["owned", "keeper"] if operation == "refresh" else ["keeper"]),
+                installer, refresh=True,
+            )
+    assert backup_tree() == before
+    assert manager_type(project).registry.get("owned") == metadata
+    assert records_path(project).read_bytes() == original_record
+    assert not (project / ".specify/extensions/owned/replacement-config.yml").exists()
+
+
+@pytest.mark.parametrize("installed_components", ["extensions"], indirect=True)
+@pytest.mark.parametrize("failure", ["snapshot", "restore"])
+def test_extension_backup_io_failure_is_reported(
+    installed_components, monkeypatch, failure,
+):
+    import shutil
+    from pathlib import Path
+
+    project, _, manager_type, installer, _, metadata = installed_components
+    backup = project / ".specify/extensions/.backup/owned"
+    backup.mkdir(parents=True)
+    original = backup / "owned-config.yml"
+    original.write_text("prior backup\n", encoding="utf-8")
+    original_record = records_path(project).read_bytes()
+    copy_tree = shutil.copytree
+
+    def fail_copy(source, destination, *args, **kwargs):
+        attempted = source if failure == "snapshot" else destination
+        if Path(attempted) == backup:
+            raise PermissionError("backup I/O denied")
+        return copy_tree(source, destination, *args, **kwargs)
+
+    def fail_save(*_args):
+        raise OSError("provenance write refused")
+
+    monkeypatch.setattr("specify_cli.bundler.services.artifacts.shutil.copytree", fail_copy)
+    if failure == "restore":
+        monkeypatch.setattr("specify_cli.bundler.services.installer.save_records", fail_save)
+    message = "backup I/O denied" if failure == "snapshot" else "Rollback was incomplete"
+    with pytest.raises(BundlerError, match=message):
+        remove_bundle(project, "demo-bundle", installer)
+    assert manager_type(project).registry.get("owned") == metadata
+    assert records_path(project).read_bytes() == original_record
+    if failure == "snapshot":
+        assert original.read_text(encoding="utf-8") == "prior backup\n"
+    else:
+        assert not backup.exists()
+
+
 @pytest.mark.parametrize("kind", ["steps", "workflows"])
 def test_dropped_component_restores_local_payload_and_exact_registry(
     tmp_path, monkeypatch, kind

tests/workflows/test_registry_isolation.py

@@ -198,3 +198,60 @@ def test_symlinked_cache_is_not_used_when_it_cannot_be_invalidated(projects):
     assert load_custom_steps(project_a) == []
     assert "shared" not in STEP_REGISTRY
     assert external.is_dir()
+
+
+@pytest.fixture
+def external_bytecode_cache(projects, monkeypatch):
+    import sys
+
+    project_a, _ = projects
+    prefix = project_a.parent / "bytecode"
+    monkeypatch.setattr(sys, "pycache_prefix", str(prefix))
+    package = project_a / ".specify/workflows/steps/shared"
+    assert load_custom_steps(project_a) == ["shared"]
+    assert STEP_REGISTRY["shared"].execute({}, None).output["project"] == "project-a"
+    for name in ("__init__.py", "helper.py"):
+        cache = Path(importlib.util.cache_from_source(str(package / name)))
+        assert cache.is_relative_to(prefix)
+        assert cache.is_file()
+    assert not (package / "__pycache__").exists()
+    return project_a, package, prefix
+
+
+def test_external_caches_are_invalidated_for_package_and_delayed_imports(
+    external_bytecode_cache,
+):
+    project, package, prefix = external_bytecode_cache
+    unrelated = prefix / "unrelated.pyc"
+    unrelated.write_bytes(b"unrelated cache")
+    for name in ("__init__.py", "helper.py"):
+        source = package / name
+        stat = source.stat()
+        original = source.read_text(encoding="utf-8")
+        updated = original.replace("project-a", "project-b")
+        assert updated != original and len(updated) == len(original)
+        source.write_text(updated, encoding="utf-8")
+        os.utime(source, ns=(stat.st_atime_ns, stat.st_mtime_ns))
+
+    assert load_custom_steps(project) == ["shared"]
+    assert STEP_REGISTRY["shared"].project_marker == "project-b"
+    assert STEP_REGISTRY["shared"].execute({}, None).output["project"] == "project-b"
+    assert unrelated.read_bytes() == b"unrelated cache"
+
+
+@pytest.mark.parametrize("source_name", ["__init__.py", "helper.py"])
+def test_failed_external_cache_deletion_skips_package(
+    external_bytecode_cache, monkeypatch, source_name,
+):
+    project, package, _ = external_bytecode_cache
+    cache = Path(importlib.util.cache_from_source(str(package / source_name)))
+    unlink = Path.unlink
+
+    def deny_cache_removal(path, *args, **kwargs):
+        if path == cache:
+            raise PermissionError("external cache deletion denied")
+        return unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", deny_cache_removal)
+    assert load_custom_steps(project) == []
+    assert "shared" not in STEP_REGISTRY