fix: make the plugin generation store actually reclaimable (#4689)

src/autoskillit/cli/_plugin_artifact.py

@@ -9,7 +9,9 @@
 from typing import TYPE_CHECKING
 
 from autoskillit.core import (
+    _AUTOSKILLIT_PLUGIN_KEY,
     ArtifactLease,
+    LegacyRetiringEvidence,
     PluginArtifactIdentity,
     PluginArtifactKind,
     PluginArtifactLifecycleLease,
@@ -25,6 +27,7 @@
     RetiringCacheReadResult,
     RetiringCacheState,
     due_retiring_records,
+    generation_store_root,
     get_logger,
     installed_plugin_artifact_lease_path,
     installed_plugin_artifact_manifest_path,
@@ -422,6 +425,32 @@ def try_reclaim(
     ) -> RetirementOutcome:
         return self._retirement.try_reclaim(record, now)
 
+    def try_promote_legacy_evidence(
+        self,
+        evidence: LegacyRetiringEvidence,
+        now: datetime,
+    ) -> RetirementOutcome:
+        return self._retirement.try_promote_legacy_evidence(
+            evidence,
+            now,
+            identity_for_path=self.identity_for_path,
+        )
+
+    def identity_for_path(self, managed_path: Path) -> PluginArtifactIdentity:
+        """Validate and return the exact current identity at a managed path."""
+        managed_path = Path(managed_path)
+        if not self._contains(managed_path):
+            raise PluginArtifactValidationError(
+                f"installed plugin is outside managed root: {managed_path}"
+            )
+        return _read_and_validate_identity(
+            managed_path,
+            expected_semantic_key=installed_plugin_semantic_key(
+                _AUTOSKILLIT_PLUGIN_KEY,
+                Path(managed_path).name,
+            ),
+        )
+
 
 class DefaultPluginRetirementCoordinator:
     """Cross-kind retirement dispatcher used by startup and explicit sweeps."""
@@ -431,15 +460,14 @@ def __init__(
         *,
         projection_owner: PluginArtifactRetirementOwner,
         installed_owner: PluginArtifactRetirementOwner,
+        generation_owner: PluginArtifactRetirementOwner,
     ) -> None:
-        self._owners = {
+        self._owners: dict[PluginArtifactKind, PluginArtifactRetirementOwner] = {
             PluginArtifactKind.PROJECTION: projection_owner,
             PluginArtifactKind.INSTALLED_PLUGIN: installed_owner,
+            PluginArtifactKind.PLUGIN_GENERATION: generation_owner,
         }
-        self._managed_roots = {
-            PluginArtifactKind.PROJECTION: projection_owner.managed_root,
-            PluginArtifactKind.INSTALLED_PLUGIN: installed_owner.managed_root,
-        }
+        self._managed_roots = {kind: owner.managed_root for kind, owner in self._owners.items()}
 
     def migrate_legacy_cache(self) -> RetiringCacheReadResult:
         """Upgrade path-only retirement evidence before exact-v2 mutations."""
@@ -466,24 +494,52 @@ def sweep_due(self, now: datetime) -> tuple[RetirementOutcome, ...]:
                     stacklevel=2,
                 )
             return ()
-        outcomes = [RetirementOutcome.LEGACY_EVIDENCE for _item in state.legacy_evidence]
+        outcomes: list[RetirementOutcome] = []
+        for evidence in state.legacy_evidence:
+            owner = (
+                self._owners.get(evidence.recognized_kind) if evidence.recognized_kind else None
+            )
+            if owner is None:
+                outcomes.append(RetirementOutcome.LEGACY_EVIDENCE)
+                continue
+            outcomes.append(owner.try_promote_legacy_evidence(evidence, now))
+        # Re-read after promotion so newly minted records reclaim in this pass.
         for record in due_retiring_records(now):
-            owner = self._owners[record.artifact_kind]
-            outcomes.append(owner.try_reclaim(record, now))
+            record_owner = self._owners.get(record.artifact_kind)
+            if record_owner is None:
+                outcomes.append(RetirementOutcome.REJECTED_IDENTITY)
+                continue
+            outcomes.append(record_owner.try_reclaim(record, now))
         return tuple(outcomes)
 
 
 def default_plugin_retirement_coordinator() -> DefaultPluginRetirementCoordinator:
-    """Compose installed and projected owners without leaking CLI into server lifespan."""
-    from autoskillit.workspace import ProjectedPluginRetirementOwner
+    """Compose every artifact-kind owner over its own disjoint managed root.
+
+    The generation store and the legacy Claude plugin cache are separate trees.
+    A single owner cannot serve both: ``try_reclaim`` rejects any record outside
+    its ``managed_root`` and never removes it, so a misrouted record is
+    re-rejected on every sweep for the life of the install.
+    """
+    from autoskillit.workspace import (
+        GenerationArtifactRetirementOwner,
+        ProjectedPluginRetirementOwner,
+    )
 
-    projection_root = Path.home() / ".autoskillit" / "plugin-projections"
+    home = Path.home()
+    projection_root = home / ".autoskillit" / "plugin-projections"
     installed_owner = InstalledPluginArtifactRetirementOwner(
         current_installed_plugin_root().parent
     )
+    generation_owner = GenerationArtifactRetirementOwner(
+        generation_store_root(home, _AUTOSKILLIT_PLUGIN_KEY),
+        home=home,
+        plugin_ref=_AUTOSKILLIT_PLUGIN_KEY,
+    )
     return DefaultPluginRetirementCoordinator(
         projection_owner=ProjectedPluginRetirementOwner(projection_root),
         installed_owner=installed_owner,
+        generation_owner=generation_owner,
     )
 
 

src/autoskillit/core/__init__.pyi

@@ -33,6 +33,9 @@ from ._plugin_artifact_identity import (
 from ._plugin_artifact_identity import (
     generation_artifact_root as generation_artifact_root,
 )
+from ._plugin_artifact_identity import (
+    generation_plugin_selector_path as generation_plugin_selector_path,
+)
 from ._plugin_artifact_identity import (
     generation_selector_path as generation_selector_path,
 )
@@ -64,12 +67,16 @@ from ._plugin_artifact_identity import (
 from ._plugin_artifact_identity import (
     resolve_current_generation as resolve_current_generation,
 )
+from ._plugin_artifact_identity import (
+    resolve_current_generation_for_plugin as resolve_current_generation_for_plugin,
+)
 from ._plugin_cache import KitchenProcessIdentity as KitchenProcessIdentity
 from ._plugin_cache import PluginArtifactRetirementEngine as PluginArtifactRetirementEngine
 from ._plugin_cache import _InstallLock as _InstallLock
 from ._plugin_cache import any_kitchen_open as any_kitchen_open
 from ._plugin_cache import append_retiring_record as append_retiring_record
 from ._plugin_cache import due_retiring_records as due_retiring_records
+from ._plugin_cache import is_reclaimable_artifact_path as is_reclaimable_artifact_path
 from ._plugin_cache import kitchen_entry_alive as kitchen_entry_alive
 from ._plugin_cache import migrate_retiring_cache_v1 as migrate_retiring_cache_v1
 from ._plugin_cache import read_active_kitchens_registry as read_active_kitchens_registry

src/autoskillit/core/_plugin_artifact_identity.py

@@ -78,6 +78,36 @@ def resolve_current_generation(home: Path, plugin_ref: str, version: str) -> Pat
     return target if target.is_dir() and target.parent == version_root else None
 
 
+def generation_plugin_selector_path(home: Path, plugin_ref: str) -> Path:
+    """Return the version-independent ``current`` symlink for a plugin.
+
+    Lives one level above the per-version selector, so a consumer that must
+    bake an absolute path into persisted config (the Codex hooks config) can
+    name a path that survives version bumps instead of pinning one exact
+    incarnation that retirement will later reclaim.
+    """
+    return generation_store_root(home, plugin_ref) / "current"
+
+
+def resolve_current_generation_for_plugin(home: Path, plugin_ref: str) -> Path | None:
+    """Resolve the version-independent selector to the active generation.
+
+    Returns ``None`` when no selector exists or the target is not a generation
+    directory directly beneath a version directory of this plugin's store.
+    """
+    selector = generation_plugin_selector_path(home, plugin_ref)
+    if not selector.is_symlink():
+        return None
+    try:
+        store_root = selector.parent.resolve(strict=True)
+        target = selector.resolve(strict=True)
+    except OSError:
+        return None
+    if not target.is_dir():
+        return None
+    return target if target.parent.parent == store_root else None
+
+
 def installed_plugin_artifact_manifest_path(managed_root: Path) -> Path:
     """Return the stable external manifest for one installed plugin root."""
     root = Path(managed_root)

src/autoskillit/core/_plugin_cache.py

@@ -294,6 +294,22 @@ def _legacy_record_id(version: str, path: str, retired_at: str) -> str:
     return hashlib.sha256(payload).hexdigest()
 
 
+def is_reclaimable_artifact_path(path: Path, managed_root: Path) -> bool:
+    """Return whether *path* may ever be treated as a retirable incarnation.
+
+    Only a direct, non-hidden child of *managed_root* qualifies. Dot-prefixed
+    entries are managed infrastructure, never artifacts — most importantly
+    ``plugin-projections/.artifact-leases``, the directory holding the lock
+    files every live session's inherited reader lease is held on. Reclaiming
+    it would delete the lease infrastructure out from under every running
+    session.
+
+    Nested descendants are excluded too: an artifact is exactly one level deep,
+    so anything deeper is a component of an artifact rather than an artifact.
+    """
+    return not path.name.startswith(".") and path.parent == managed_root
+
+
 def _classify_legacy_path(
     path: str,
     managed_roots: Mapping[PluginArtifactKind, Path],
@@ -313,6 +329,8 @@ def _classify_legacy_path(
         except OSError:
             continue
         if location != managed_root and location.is_relative_to(managed_root):
+            if not is_reclaimable_artifact_path(location, managed_root):
+                return None, "legacy path is managed infrastructure, not an artifact incarnation"
             return kind, None
     return None, "legacy path is outside known managed roots"
 
@@ -702,6 +720,70 @@ def try_reclaim(self, record: RetiringArtifactRecord, now: datetime) -> Retireme
         finally:
             writer.close_preserving()
 
+    def try_promote_legacy_evidence(
+        self,
+        evidence: LegacyRetiringEvidence,
+        now: datetime,
+        *,
+        identity_for_path: Callable[[Path], PluginArtifactIdentity],
+    ) -> RetirementOutcome:
+        """Promote path-only legacy evidence into an exact v2 record.
+
+        Migrated v1 evidence carries no incarnation or digest, so it can never
+        authorize deletion by itself. This re-derives an exact identity from
+        disk and hands it to the normal queue; the artifact is then removed by
+        ``try_reclaim`` under the same lease and identity checks as any other
+        record. Nothing is deleted here.
+
+        Eligibility is re-derived from scratch rather than trusting the stored
+        ``recognized_kind`` — evidence already persisted with a wrong kind
+        must not become authority now.
+        """
+        if evidence.recognized_kind is not self.artifact_kind:
+            return RetirementOutcome.LEGACY_EVIDENCE
+        try:
+            path = destination_location(Path(evidence.path))
+        except (OSError, TypeError, ValueError):
+            return RetirementOutcome.LEGACY_EVIDENCE
+        if not is_reclaimable_artifact_path(path, self.managed_root):
+            return RetirementOutcome.LEGACY_EVIDENCE
+        if not self.contains(path):
+            return RetirementOutcome.LEGACY_EVIDENCE
+        if not path.exists():
+            # Nothing left to protect (also covers a broken symlink: exists()
+            # follows the link and is False when the target is gone).
+            remove_retiring_records((evidence.record_id,))
+            return RetirementOutcome.RECORD_REMOVED
+        try:
+            writer = ArtifactLease.acquire_exclusive(self._lease_path(path), blocking=False)
+        except ArtifactLeaseContention:
+            return RetirementOutcome.DEFERRED_CONTENDED
+        except (OSError, RuntimeError) as exc:
+            self._logger.warning(
+                "plugin_artifact_legacy_promotion_lease_failed",
+                artifact_kind=self.artifact_kind.value,
+                path=str(path),
+                error=str(exc),
+            )
+            return RetirementOutcome.DEFERRED_IO_ERROR
+        try:
+            if self._is_current is not None and self._is_current(path):
+                return RetirementOutcome.DEFERRED_CONTENDED
+            try:
+                identity = identity_for_path(path)
+            except (
+                PluginArtifactValidationError,
+                PluginArtifactUnavailableError,
+                OSError,
+            ):
+                # Cannot positively identify it; never delete on ambiguity.
+                return RetirementOutcome.LEGACY_EVIDENCE
+            created = self.enqueue_retirement(identity, now).created
+            remove_retiring_records((evidence.record_id,))
+        finally:
+            writer.close_preserving()
+        return RetirementOutcome.RECLAIMED if created else RetirementOutcome.RECORD_REMOVED
+
     def _log_reclaim(
         self,
         record: RetiringArtifactRecord,

src/autoskillit/core/types/_type_plugin_source.py

@@ -59,10 +59,23 @@ def consumes_artifact(self) -> bool:
 
 
 class PluginArtifactKind(StrEnum):
-    """Managed plugin artifact families sharing the retirement queue."""
+    """Managed plugin artifact families sharing the retirement queue.
+
+    ``PLUGIN_GENERATION`` routes the generation store
+    (``~/.autoskillit/plugin-generations/``) to its own retirement owner.
+    ``INSTALLED_PLUGIN`` remains bound to the legacy Claude-cache tree, which
+    ``reconcile_install_artifacts()`` still enqueues. The two roots are
+    disjoint, so one owner cannot serve both: a record routed to an owner whose
+    ``managed_root`` does not contain it is rejected on every sweep forever.
+
+    This is a routing key for the retirement queue only. The on-disk manifest's
+    own ``artifact_kind`` stays ``INSTALLED_PLUGIN`` for both trees, since they
+    share one manifest format.
+    """
 
     PROJECTION = "projection"
     INSTALLED_PLUGIN = "installed_plugin"
+    PLUGIN_GENERATION = "plugin_generation"
 
 
 class RetiringCacheState(StrEnum):

src/autoskillit/core/types/_type_protocols_workspace.py

@@ -12,6 +12,7 @@
 from ._type_enums import SkillExecutionRole, SkillInvalidityKind, SkillSource
 from ._type_exploration import RepositoryProfileId
 from ._type_plugin_source import (
+    LegacyRetiringEvidence,
     PluginArtifactIdentity,
     PluginLaunchBinding,
     PluginLoadMode,
@@ -87,6 +88,12 @@ def try_reclaim(
         now: datetime,
     ) -> RetirementOutcome: ...
 
+    def try_promote_legacy_evidence(
+        self,
+        evidence: LegacyRetiringEvidence,
+        now: datetime,
+    ) -> RetirementOutcome: ...
+
 
 @runtime_checkable
 class PluginRetirementCoordinator(Protocol):

src/autoskillit/execution/backends/_codex_hooks.py

@@ -87,6 +87,17 @@ def _resolve_codex_hooks_dir(plugin_dir: Path | None = None) -> Path:
     short-lived resolve→validate of the current generation selector is
     performed through the same generation-store authority as launch binding.
 
+    The bindingless path prefers the *version-independent* selector. Codex
+    bakes this absolute path into ``~/.codex/config.toml`` and re-reads it at
+    every launch without re-resolving, and a per-session snapshot of that
+    config may outlive the version it was written against. A path pinned to one
+    version's incarnation therefore dangles as soon as that generation is
+    retired, taking every safety guard with it. The plugin-level selector is
+    re-pointed on each publish, so the same absolute path stays valid across
+    version bumps. ``trusted_hash`` is unaffected: it hashes the dispatcher's
+    bytes through the symlink, and ``_dispatch.py`` is byte-identical across
+    versions by design.
+
     If neither the generation store nor the legacy installed cache supplies a
     dispatcher, the dev-checkout ``HOOKS_DIR`` is used as the terminal fallback.
     """
@@ -98,16 +109,25 @@ def _resolve_codex_hooks_dir(plugin_dir: Path | None = None) -> Path:
 
     # Bindingless path: resolve from generation store with short-lived lease
     from autoskillit import __version__
-    from autoskillit.core import resolve_current_generation
+    from autoskillit.core import (
+        generation_plugin_selector_path,
+        resolve_current_generation,
+        resolve_current_generation_for_plugin,
+    )
+
+    if resolve_current_generation_for_plugin(Path.home(), _AUTOSKILLIT_PLUGIN_KEY) is not None:
+        candidate = generation_plugin_selector_path(Path.home(), _AUTOSKILLIT_PLUGIN_KEY) / "hooks"
+        if (candidate / "_dispatch.py").is_file():
+            return candidate
 
-    generation_dir = resolve_current_generation(Path.home(), "autoskillit", __version__)
+    generation_dir = resolve_current_generation(Path.home(), _AUTOSKILLIT_PLUGIN_KEY, __version__)
     if generation_dir is not None:
         candidate = generation_dir / "hooks"
         if (candidate / "_dispatch.py").is_file():
             return candidate
 
     # Fall back to legacy installed cache
-    cache_root = installed_plugin_artifact_root(Path.home(), "autoskillit", __version__)
+    cache_root = installed_plugin_artifact_root(Path.home(), _AUTOSKILLIT_PLUGIN_KEY, __version__)
     try:
         identity = read_installed_plugin_artifact_identity(
             cache_root,