Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added EOF
Empty file.
26 changes: 26 additions & 0 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2365,6 +2365,19 @@ def integration_switch(
)
raise typer.Exit(1)

# Unregister extension commands for the old agent so they don't
# remain as orphans in the old agent's directory.
try:
from .extensions import ExtensionManager

ext_mgr = ExtensionManager(project_root)
ext_mgr.unregister_agent_artifacts(installed_key)
except Exception as ext_err:
console.print(
f"[yellow]Warning:[/yellow] Could not clean up extension artifacts "
f"(commands, skills, registry entries) for '{installed_key}': {ext_err}"
)
Comment thread
mnriem marked this conversation as resolved.

# Clear metadata so a failed Phase 2 doesn't leave stale references
_remove_integration_json(project_root)
opts = load_init_options(project_root)
Expand Down Expand Up @@ -2404,6 +2417,19 @@ def integration_switch(
_write_integration_json(project_root, target_integration.key)
_update_init_options_for_integration(project_root, target_integration, script_type=selected_script)

# Re-register extension commands for the new agent so that
# previously-installed extensions are available in the new integration.
try:
from .extensions import ExtensionManager

ext_mgr = ExtensionManager(project_root)
ext_mgr.register_enabled_extensions_for_agent(target)
except Exception as ext_err:
console.print(
f"[yellow]Warning:[/yellow] Could not register extension commands, skills, "
f"or related artifacts for '{target}': {ext_err}"
)

except Exception as e:
# Attempt rollback of any files written by setup
try:
Expand Down
179 changes: 170 additions & 9 deletions src/specify_cli/extensions.py
Original file line number Diff line number Diff line change
Expand Up @@ -962,29 +962,40 @@ def _register_extension_skills(

return written

def _unregister_extension_skills(self, skill_names: List[str], extension_id: str) -> None:
def _unregister_extension_skills(
self,
skill_names: List[str],
extension_id: str,
skills_dir: Optional[Path] = None,
) -> None:
"""Remove SKILL.md directories for extension skills.

Called during extension removal to clean up skill files that
were created by ``_register_extension_skills()``.

If ``_get_skills_dir()`` returns ``None`` (e.g. the user removed
init-options.json or toggled ai_skills after installation), we
fall back to scanning all known agent skills directories so that
orphaned skill directories are still cleaned up. In that case
each candidate directory is verified against the SKILL.md
``metadata.source`` field before removal to avoid accidentally
deleting user-created skills with the same name.
If *skills_dir* is not provided and ``_get_skills_dir()`` returns
``None`` (e.g. the user removed init-options.json or toggled
ai_skills after installation), we fall back to scanning all known
agent skills directories so that orphaned skill directories are
still cleaned up. In that case each candidate directory is
verified against the SKILL.md ``metadata.source`` field before
removal to avoid accidentally deleting user-created skills with
the same name.

Args:
skill_names: List of skill names to remove.
extension_id: Extension ID used to verify ownership during
fallback candidate scanning.
skills_dir: Optional explicit skills directory to use instead
of resolving via ``_get_skills_dir()``. Useful when the
caller needs to target a specific agent's skills directory
regardless of the currently-active agent in init-options.
"""
if not skill_names:
return

skills_dir = self._get_skills_dir()
if skills_dir is None:
skills_dir = self._get_skills_dir()

if skills_dir:
# Fast path: we know the exact skills directory
Expand Down Expand Up @@ -1332,6 +1343,156 @@ def remove(self, extension_id: str, keep_config: bool = False) -> bool:

return True

@staticmethod
def _valid_name_list(value: Any) -> List[str]:
"""Return string entries from a registry list, ignoring corrupt values."""
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, str)]

def unregister_agent_artifacts(self, agent_name: str) -> None:
"""Remove extension files registered for a specific agent.

Extension command files are tracked per agent in ``registered_commands``.
Extension skills are scoped to the provided *agent_name*; they are removed
from that agent's skills directory (resolved via its integration config)
and the registry field is cleared.

Skips cleanup when *agent_name* is not a supported agent to avoid
losing registry entries while leaving orphaned files on disk.
"""
if not agent_name:
return

registrar = CommandRegistrar()
if agent_name not in registrar.AGENT_CONFIGS:
return

# Resolve the skills directory for the specific agent so cleanup is
# agent-scoped and does not depend on the currently-active agent in
# init-options. Use the same helper that extension install uses.
from . import _get_skills_dir as resolve_skills_dir

agent_skills_dir = resolve_skills_dir(self.project_root, agent_name)

Comment thread
mnriem marked this conversation as resolved.
for ext_id, metadata in self.registry.list().items():
updates: Dict[str, Any] = {}

registered_commands = metadata.get("registered_commands", {})
if isinstance(registered_commands, dict) and agent_name in registered_commands:
command_names = self._valid_name_list(registered_commands.get(agent_name))
if command_names:
registrar.unregister_commands({agent_name: command_names}, self.project_root)

Comment thread
mnriem marked this conversation as resolved.
new_registered = copy.deepcopy(registered_commands)
new_registered.pop(agent_name, None)
updates["registered_commands"] = new_registered

registered_skills = self._valid_name_list(metadata.get("registered_skills", []))
if registered_skills:
# Only pass the resolved skills_dir when it actually exists.
# Otherwise let _unregister_extension_skills fall back to
# scanning all known agent skills directories, which is useful
# for cleaning up stale entries created by earlier installs.
skills_dir = agent_skills_dir if agent_skills_dir.is_dir() else None
self._unregister_extension_skills(
registered_skills, ext_id, skills_dir=skills_dir
)

# Only reconcile registry state when cleanup was scoped to a
# specific existing directory. When skills_dir is None,
# _unregister_extension_skills falls back to scanning multiple
# candidate directories, so agent_skills_dir cannot be used to
# infer what was removed. When skills_dir is set,
# _unregister_extension_skills may intentionally skip deletion
# when ownership cannot be verified (e.g., corrupted/missing
# SKILL.md or mismatching metadata.source). Only drop registry
# entries for skill directories that were actually removed so
# future cleanup attempts can still find skipped ones.
if skills_dir is not None:
remaining_skills = [
skill_name
for skill_name in registered_skills
if (skills_dir / skill_name).is_dir()
]
if remaining_skills != registered_skills:
updates["registered_skills"] = remaining_skills
Comment thread
mnriem marked this conversation as resolved.

if updates:
self.registry.update(ext_id, updates)

def register_enabled_extensions_for_agent(self, agent_name: str) -> None:
"""Register installed, enabled extensions for ``agent_name``.

This is intended to be called after switching integrations. Command
registration is scoped to the explicit ``agent_name`` argument, but some
behavior still depends on the current init-options state (for example,
skills-mode handling uses the active ``ai`` / ``ai_skills`` settings).

Callers should therefore pass the agent that has just been made active
in init-options; in normal use, ``agent_name`` is expected to match the
current ``ai`` value. This mirrors extension install behavior while
avoiding stale default-mode command directories when that active agent
is running in skills mode (notably Copilot ``--skills``).
"""
if not agent_name:
return

from . import load_init_options

registrar = CommandRegistrar()
agent_config = registrar.AGENT_CONFIGS.get(agent_name)
init_options = load_init_options(self.project_root)
if not isinstance(init_options, dict):
init_options = {}

active_agent = init_options.get("ai")
skills_mode_active = (
active_agent == agent_name
and bool(init_options.get("ai_skills"))
and bool(agent_config)
and agent_config.get("extension") != "/SKILL.md"
)

for ext_id, metadata in self.registry.list().items():
if not metadata.get("enabled", True):
continue

manifest = self.get_extension(ext_id)
if manifest is None:
continue

ext_dir = self.extensions_dir / ext_id
updates: Dict[str, Any] = {}

if agent_config and not skills_mode_active:
registered = registrar.register_commands_for_agent(
agent_name, manifest, ext_dir, self.project_root
)
registered_commands = metadata.get("registered_commands", {})
if not isinstance(registered_commands, dict):
registered_commands = {}
new_registered = copy.deepcopy(registered_commands)
if registered:
new_registered[agent_name] = registered
Comment thread
mnriem marked this conversation as resolved.
else:
# Registration returned empty list (e.g., corrupted
# manifest pointing at missing command files). Clear
# stale entry so later cleanup doesn't try to remove
# files that were never written.
new_registered.pop(agent_name, None)
if new_registered != registered_commands:
updates["registered_commands"] = new_registered

registered_skills = self._register_extension_skills(manifest, ext_dir)
if registered_skills:
existing_skills = self._valid_name_list(metadata.get("registered_skills", []))
merged_skills = list(dict.fromkeys(existing_skills + registered_skills))
updates["registered_skills"] = merged_skills
Comment thread
cyliu0 marked this conversation as resolved.

if updates:
self.registry.update(ext_id, updates)

def list_installed(self) -> List[Dict[str, Any]]:
"""List all installed extensions with metadata.

Expand Down
Loading
Loading