Require per-extension ownership for generic symlinks

Reject cross-extension retargeted symlinks in generic command and skill ownership checks, and preserve them during skill cleanup. Exercise refresh, forced reinstall, removal, and update rollback with identical target bytes.

Assisted-by: GitHub Copilot (model: GPT-6 Sol, autonomous)

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: mnriem

src/specify_cli/agents.py

@@ -634,7 +634,7 @@ def _generic_owned_output(path: Path, source_id: str, project_root: Path) -> boo
         if path.stat().st_nlink > 1:
             return False
         if path.is_symlink() and not path.resolve().is_relative_to(
-            (project_root / ".specify/extensions").resolve()
+            (project_root / ".specify/extensions" / source_id).resolve()
         ):
             return False
         relative = path.relative_to(project_root.resolve()).as_posix()

src/specify_cli/extensions/__init__.py

@@ -1680,7 +1680,7 @@ def _replacement(match: re.Match[str]) -> str:
             if selected_ai == "generic" and skill_dir_preexists:
                 metadata = self.registry.get(manifest.id) or {}
                 if skill_name not in self._generic_owned_names(
-                    metadata, [skill_name], skills=True
+                    metadata, [skill_name], skills=True, extension_id=manifest.id
                 ):
                     continue
             if skill_file.exists() or skill_file.is_symlink():
@@ -2082,7 +2082,8 @@ def _restore_generic_refresh_artifacts(
             raise ExtensionError("Could not restore generic artifacts: " + "; ".join(errors))
 
     def _generic_owned_names(
-        self, metadata: Dict[str, Any], names: List[str], *, skills: bool
+        self, metadata: Dict[str, Any], names: List[str], *,
+        skills: bool, extension_id: str,
     ) -> List[str]:
         """Keep customized or untracked generic artifacts out of cleanup."""
         from ..integrations.generic import registration_directory
@@ -2105,7 +2106,7 @@ def _generic_owned_names(
             if path.stat().st_nlink > 1:
                 continue
             if path.is_symlink() and not path.resolve().is_relative_to(
-                self.extensions_dir.resolve()
+                (self.extensions_dir / extension_id).resolve()
             ):
                 continue
             relative = path.relative_to(root).as_posix()
@@ -2125,7 +2126,9 @@ def _complete_generic_refresh(
         """Require each invocation to be newly written or still hash-owned."""
         missing = sorted(set(expected) - set(registered))
         retained = (
-            self._generic_owned_names(metadata, missing, skills=skills)
+            self._generic_owned_names(
+                metadata, missing, skills=skills, extension_id=extension_id
+            )
             if missing else []
         )
         absent = sorted(set(missing) - set(retained))
@@ -2236,6 +2239,14 @@ def _unregister_extension_skills(
             skill_names, extension_id, skills_dir=skills_dir
         ):
             skill_file = skill_subdir / "SKILL.md"
+            if (
+                generic_hashes is not None
+                and skill_file.is_symlink()
+                and not skill_file.resolve().is_relative_to(
+                    (self.extensions_dir / extension_id).resolve()
+                )
+            ):
+                continue
             if generic_hashes is not None and skill_file.is_relative_to(
                 self.project_root.resolve()
             ):
@@ -2510,7 +2521,8 @@ def install_from_directory(
             )
             owned = (
                 set(self._generic_owned_names(
-                    self.registry.get(manifest.id) or {}, list(names), skills=skills,
+                    self.registry.get(manifest.id) or {}, list(names),
+                    skills=skills, extension_id=manifest.id,
                 ))
                 if force and self.registry.is_installed(manifest.id) else set()
             )
@@ -3696,7 +3708,7 @@ def unregister_agent_artifacts(
                 )
                 if agent_name == "generic":
                     command_names = self._generic_owned_names(
-                        metadata, command_names, skills=False
+                        metadata, command_names, skills=False, extension_id=ext_id
                     )
                 if command_names:
                     registrar.unregister_commands(
@@ -3713,7 +3725,7 @@ def unregister_agent_artifacts(
             if registered_skills and not commands_only:
                 if agent_name == "generic":
                     registered_skills = self._generic_owned_names(
-                        metadata, registered_skills, skills=True
+                        metadata, registered_skills, skills=True, extension_id=ext_id
                     )
                 # Always pass the explicit, agent-scoped skills_dir — even
                 # when it doesn't currently exist on disk. This method must
@@ -4051,7 +4063,7 @@ def register_enabled_extensions_for_agent(self, agent_name: str, *, force: bool
                             ]
                             if agent_name == "generic":
                                 to_remove = self._generic_owned_names(
-                                    metadata, to_remove, skills=True
+                                    metadata, to_remove, skills=True, extension_id=ext_id
                                 )
                             if to_remove:
                                 self._unregister_extension_skills(
@@ -4139,7 +4151,8 @@ def register_enabled_extensions_for_agent(self, agent_name: str, *, force: bool
                             if fully_replaced:
                                 if agent_name == "generic":
                                     fully_replaced = self._generic_owned_names(
-                                        metadata, fully_replaced, skills=False
+                                        metadata, fully_replaced, skills=False,
+                                        extension_id=ext_id,
                                     )
                                 registrar.unregister_commands(
                                     {agent_name: fully_replaced}, self.project_root

src/specify_cli/extensions/command_enable.py

@@ -66,7 +66,7 @@ def extension_enable(
                     if skills else set(manager._collect_manifest_command_names(manifest))
                 )
                 owned = set(manager._generic_owned_names(
-                    refreshed, list(expected), skills=skills,
+                    refreshed, list(expected), skills=skills, extension_id=extension_id,
                 ))
                 missing = expected - owned
                 if missing:

tests/integrations/test_integration_generic.py

@@ -169,7 +169,8 @@ def test_generic_skill_symlinked_directory_is_not_owned(
         source.write_text(source.read_text(encoding="utf-8") + "\nnew source\n", encoding="utf-8")
         manager.register_enabled_extensions_for_agent("generic", force=True)
         assert manager._generic_owned_names(
-            manager.registry.get("sample"), ["speckit-sample-run"], skills=True,
+            manager.registry.get("sample"), ["speckit-sample-run"],
+            skills=True, extension_id="sample",
         ) == []
     elif operation == "force":
         with pytest.raises(ExtensionError, match="cannot be replaced safely"):
@@ -1453,6 +1454,71 @@ def install_update(self, _zip_path, speckit_version, *, catalog_name=None):
     assert not artifact.exists()
 
 
+def test_generic_update_rollback_preserves_cross_extension_skill_link(
+    tmp_path, generic_extension, monkeypatch,
+):
+    from typer.testing import CliRunner
+    from specify_cli import app
+
+    project = generic_project(tmp_path, skills=True)
+    manager = ExtensionManager(project)
+    manager.install_from_directory(generic_extension, "1.0.0", link_commands=True)
+    artifact = project / ".custom/commands/speckit-sample-run/SKILL.md"
+    if not artifact.is_symlink():
+        pytest.skip("dev-mode symlinks are unavailable")
+    other_file = manager.extensions_dir / "other/file.md"
+    other_file.parent.mkdir()
+    original = artifact.read_bytes()
+    other_file.write_bytes(original)
+    artifact.unlink()
+    artifact.symlink_to(os.path.relpath(other_file, artifact.parent))
+    metadata = manager.registry.get("sample")
+
+    updated_source = tmp_path / "updated-source"
+    shutil.copytree(generic_extension, updated_source)
+    manifest_path = updated_source / "extension.yml"
+    manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
+    manifest["extension"]["version"] = "2.0.0"
+    manifest_path.write_text(yaml.safe_dump(manifest), encoding="utf-8")
+    archive = tmp_path / "sample-update.zip"
+    with zipfile.ZipFile(archive, "w") as zip_file:
+        for file in updated_source.rglob("*"):
+            if file.is_file():
+                zip_file.write(file, file.relative_to(updated_source))
+
+    original_unlink = Path.unlink
+    removed_user_link = []
+
+    def track_unlink(path, *args, **kwargs):
+        if path == artifact:
+            removed_user_link.append(path)
+        return original_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", track_unlink)
+    with (
+        patch.object(Path, "cwd", return_value=project),
+        patch.object(ExtensionCatalog, "get_extension_info", return_value={
+            "id": "sample",
+            "name": "Sample",
+            "version": "2.0.0",
+            "_install_allowed": True,
+        }),
+        patch.object(ExtensionCatalog, "download_extension", return_value=archive),
+        patch.object(ExtensionManager, "install_from_zip", side_effect=RuntimeError("update failed")),
+    ):
+        result = CliRunner().invoke(
+            app, ["extension", "update", "sample"], input="y\n",
+        )
+
+    assert result.exit_code == 1
+    assert "update failed" in result.output
+    assert not removed_user_link
+    assert artifact.is_symlink()
+    assert artifact.resolve() == other_file.resolve()
+    assert other_file.read_bytes() == original
+    assert ExtensionManager(project).registry.get("sample") == metadata
+
+
 @pytest.mark.parametrize("skills", [False, True])
 @pytest.mark.parametrize("collision", [False, True])
 @pytest.mark.parametrize("multiple_prior_dirs", [False, True])
@@ -1723,6 +1789,46 @@ def test_generic_dev_extension_removes_links(tmp_path, generic_extension, skills
     assert not artifact.is_symlink()
 
 
+@pytest.mark.parametrize("skills", [False, True])
+@pytest.mark.parametrize("operation", ["refresh", "force", "remove"])
+def test_generic_retargeted_cross_extension_link_is_not_owned(
+    tmp_path, generic_extension, skills, operation, capsys,
+):
+    project = generic_project(tmp_path, skills=skills)
+    manager = ExtensionManager(project)
+    manager.install_from_directory(generic_extension, "1.0.0", link_commands=True)
+    artifact = project / ".custom/commands" / (
+        "speckit-sample-run/SKILL.md" if skills else "speckit.sample.run.md"
+    )
+    if not artifact.is_symlink():
+        pytest.skip("dev-mode symlinks are unavailable")
+    other_file = manager.extensions_dir / "other" / "file.md"
+    other_file.parent.mkdir()
+    original = artifact.read_bytes()
+    other_file.write_bytes(original)
+    artifact.unlink()
+    artifact.symlink_to(os.path.relpath(other_file, artifact.parent))
+    metadata = manager.registry.get("sample")
+
+    if operation == "refresh":
+        source = manager.extensions_dir / "sample/commands/run.md"
+        source.write_text(source.read_text(encoding="utf-8") + "\nnew source\n", encoding="utf-8")
+        manager.register_enabled_extensions_for_agent("generic", force=skills)
+        warning = capsys.readouterr().out
+        assert "Missing" in warning and "invocation artifacts" in warning
+        assert manager.registry.get("sample") == metadata
+    elif operation == "force":
+        with pytest.raises(ExtensionError, match="cannot be replaced safely"):
+            manager.install_from_directory(generic_extension, "1.0.0", force=True)
+        assert manager.registry.get("sample") == metadata
+    else:
+        assert manager.remove("sample")
+
+    assert artifact.is_symlink()
+    assert artifact.resolve() == other_file.resolve()
+    assert other_file.read_bytes() == original
+
+
 @pytest.mark.parametrize("skills", [False, True])
 def test_generic_dev_extension_preserves_edited_link(tmp_path, generic_extension, skills):
     project = generic_project(tmp_path, skills=skills)