feat(extensions): Quality of life improvements for RFC-aligned catalog integration (#1776)

* feat(extensions): implement automatic updates with atomic backup/restore

- Implement automatic extension updates with download from catalog
- Add comprehensive backup/restore mechanism for failed updates:
  - Backup registry entry before update
  - Backup extension directory
  - Backup command files for all AI agents
  - Backup hooks from extensions.yml
- Add extension ID verification after install
- Add KeyboardInterrupt handling to allow clean cancellation
- Fix enable/disable to preserve installed_at timestamp by using
  direct registry manipulation instead of registry.add()
- Add rollback on any update failure with command file,
  hook, and registry restoration

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): comprehensive name resolution and error handling improvements

- Add shared _resolve_installed_extension helper for ID/display name resolution
  with proper ambiguous name handling (shows table of matches)
- Add _resolve_catalog_extension helper for catalog lookups by ID or display name
- Update enable/disable/update/remove commands to use name resolution helpers
- Fix extension_info to handle catalog errors gracefully:
  - Fallback to local installed info when catalog unavailable
  - Distinguish "catalog unavailable" from "not found in catalog"
  - Support display name lookup for both installed and catalog extensions
- Use resolved display names in all status messages for consistency
- Extract _print_extension_info helper for DRY catalog info printing

Addresses reviewer feedback:
- Ambiguous name handling in enable/disable/update
- Catalog error fallback for installed extensions
- UX message clarity (catalog unavailable vs not found)
- Resolved ID in status messages

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): properly detect ambiguous names in extension_info

The extension_info command was breaking on the first name match without
checking for ambiguity. This fix separates ID matching from name matching
and checks for ambiguity before selecting a match, consistent with the
_resolve_installed_extension() helper used by other commands.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* refactor(extensions): add public update() method to ExtensionRegistry

Add a proper public API for updating registry metadata while preserving
installed_at timestamp, instead of directly mutating internal registry
data and calling private _save() method.

Changes:
- Add ExtensionRegistry.update() method that preserves installed_at
- Update enable/disable commands to use registry.update()
- Update rollback logic to use registry.update()

This decouples the CLI from registry internals and maintains proper
encapsulation.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): safely access optional author field in extension_info

ExtensionManifest doesn't expose an author property - the author field
is optional in extension.yml and stored in data["extension"]["author"].
Use safe dict access to avoid AttributeError.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): address multiple reviewer comments

- ExtensionRegistry.update() now preserves original installed_at timestamp
- Add ExtensionRegistry.restore() for rollback (entry was removed)
- Clean up wrongly installed extension on ID mismatch before rollback
- Remove unused catalog_error parameter from _print_extension_info()

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): check _install_allowed for updates, preserve backup on failed rollback

- Skip automatic updates for extensions from catalogs with install_allowed=false
- Only delete backup directory on successful rollback, preserve it on failure
  for manual recovery

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): address reviewer feedback on update/rollback logic

- Hook rollback: handle empty backup_hooks by checking `is not None`
  instead of truthiness (falsy empty dict would skip hook cleanup)
- extension_info: use resolved_installed_id for catalog lookup when
  extension was found by display name (prevents wrong catalog match)
- Rollback: always remove extension dir first, then restore if backup
  exists (handles case when no original dir existed before update)
- Validate extension ID from ZIP before installing, not after
  (avoids side effects of installing wrong extension before rollback)
- Preserve enabled state during updates: re-apply disabled state and
  hook enabled flags after successful update
- Optimize _resolve_catalog_extension: pass query to catalog.search()
  instead of fetching all extensions
- update() now merges metadata with existing entry instead of replacing
  (preserves fields like registered_commands when only updating enabled)
- Add tests for ExtensionRegistry.update() and restore() methods:
  - test_update_preserves_installed_at
  - test_update_merges_with_existing
  - test_update_raises_for_missing_extension
  - test_restore_overwrites_completely
  - test_restore_can_recreate_removed_entry

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* docs(extensions): update RFC to reflect implemented status

- Change status from "Draft" to "Implemented"
- Update all Implementation Phases to show completed items
- Add new features implemented beyond original RFC:
  - Display name resolution for all commands
  - Ambiguous name handling with tables
  - Atomic update with rollback
  - Pre-install ID validation
  - Enabled state preservation
  - Registry update/restore methods
  - Catalog error fallback
  - _install_allowed flag
  - Cache invalidation
- Convert Open Questions to Resolved Questions with decisions
- Add remaining Open Questions (sandboxing, signatures) as future work
- Fix table of contents links

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): address third round of PR review comments

- Refactor extension_info to use _resolve_installed_extension() helper
  with new allow_not_found parameter instead of duplicating resolution logic
- Fix rollback hook restoration to not create empty hooks: {} in config
  when original config had no hooks section
- Fix ZIP pre-validation to handle nested extension.yml files (GitHub
  auto-generated ZIPs have structure like repo-name-branch/extension.yml)
- Replace unused installed_manifest variable with _ placeholder
- Add display name to update status messages for better UX

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(extensions): address fourth round of PR review comments

Rollback fixes:
- Preserve installed_at timestamp after successful update (was reset by
  install_from_zip calling registry.add)
- Fix rollback to only delete extension_dir if backup exists (avoids
  destroying valid installation when failure happens before modification)
- Fix rollback to remove NEW command files created by failed install
  (files that weren't in original backup are now cleaned up)
- Fix rollback to delete hooks key entirely when backup_hooks is None
  (original config had no hooks key, so restore should remove it)

Cross-command consistency fix:
- Add display name resolution to `extension add` command using
  _resolve_catalog_extension() helper (was only in `extension info`)
- Use resolved extension ID for download_extension() call, not original
  argument which may be a display name

Security fix (fail-closed):
- Malformed catalog config (empty/missing URLs) now raises ValidationError
  instead of silently falling back to built-in catalogs

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(lint): address ruff linting errors and registry.update() semantics

- Remove unused import ExtensionError in extension_info
- Remove extraneous f-prefix from strings without placeholders
- Use registry.restore() instead of registry.update() for installed_at
  preservation (update() always preserves existing installed_at, ignoring
  our override)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: iamaeroplane <michal.bachorik@gmail.com>
Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Michal Bachorik
2026-03-13 13:23:37 +01:00
committed by GitHub
parent 82f8a13f83
commit 58ce653908
4 changed files with 1238 additions and 288 deletions

View File

@@ -1813,6 +1813,126 @@ def get_speckit_version() -> str:
return "unknown"
def _resolve_installed_extension(
argument: str,
installed_extensions: list,
command_name: str = "command",
allow_not_found: bool = False,
) -> tuple[Optional[str], Optional[str]]:
"""Resolve an extension argument (ID or display name) to an installed extension.
Args:
argument: Extension ID or display name provided by user
installed_extensions: List of installed extension dicts from manager.list_installed()
command_name: Name of the command for error messages (e.g., "enable", "disable")
allow_not_found: If True, return (None, None) when not found instead of raising
Returns:
Tuple of (extension_id, display_name), or (None, None) if allow_not_found=True and not found
Raises:
typer.Exit: If extension not found (and allow_not_found=False) or name is ambiguous
"""
from rich.table import Table
# First, try exact ID match
for ext in installed_extensions:
if ext["id"] == argument:
return (ext["id"], ext["name"])
# If not found by ID, try display name match
name_matches = [ext for ext in installed_extensions if ext["name"].lower() == argument.lower()]
if len(name_matches) == 1:
# Unique display-name match
return (name_matches[0]["id"], name_matches[0]["name"])
elif len(name_matches) > 1:
# Ambiguous display-name match
console.print(
f"[red]Error:[/red] Extension name '{argument}' is ambiguous. "
"Multiple installed extensions share this name:"
)
table = Table(title="Matching extensions")
table.add_column("ID", style="cyan", no_wrap=True)
table.add_column("Name", style="white")
table.add_column("Version", style="green")
for ext in name_matches:
table.add_row(ext.get("id", ""), ext.get("name", ""), str(ext.get("version", "")))
console.print(table)
console.print("\nPlease rerun using the extension ID:")
console.print(f" [bold]specify extension {command_name} <extension-id>[/bold]")
raise typer.Exit(1)
else:
# No match by ID or display name
if allow_not_found:
return (None, None)
console.print(f"[red]Error:[/red] Extension '{argument}' is not installed")
raise typer.Exit(1)
def _resolve_catalog_extension(
argument: str,
catalog,
command_name: str = "info",
) -> tuple[Optional[dict], Optional[Exception]]:
"""Resolve an extension argument (ID or display name) from the catalog.
Args:
argument: Extension ID or display name provided by user
catalog: ExtensionCatalog instance
command_name: Name of the command for error messages
Returns:
Tuple of (extension_info, catalog_error)
- If found: (ext_info_dict, None)
- If catalog error: (None, error)
- If not found: (None, None)
"""
from rich.table import Table
from .extensions import ExtensionError
try:
# First try by ID
ext_info = catalog.get_extension_info(argument)
if ext_info:
return (ext_info, None)
# Try by display name - search using argument as query, then filter for exact match
search_results = catalog.search(query=argument)
name_matches = [ext for ext in search_results if ext["name"].lower() == argument.lower()]
if len(name_matches) == 1:
return (name_matches[0], None)
elif len(name_matches) > 1:
# Ambiguous display-name match in catalog
console.print(
f"[red]Error:[/red] Extension name '{argument}' is ambiguous. "
"Multiple catalog extensions share this name:"
)
table = Table(title="Matching extensions")
table.add_column("ID", style="cyan", no_wrap=True)
table.add_column("Name", style="white")
table.add_column("Version", style="green")
table.add_column("Catalog", style="dim")
for ext in name_matches:
table.add_row(
ext.get("id", ""),
ext.get("name", ""),
str(ext.get("version", "")),
ext.get("_catalog_name", ""),
)
console.print(table)
console.print("\nPlease rerun using the extension ID:")
console.print(f" [bold]specify extension {command_name} <extension-id>[/bold]")
raise typer.Exit(1)
# Not found
return (None, None)
except ExtensionError as e:
return (None, e)
@extension_app.command("list")
def extension_list(
available: bool = typer.Option(False, "--available", help="Show available extensions from catalog"),
@@ -2111,8 +2231,11 @@ def extension_add(
# Install from catalog
catalog = ExtensionCatalog(project_root)
# Check if extension exists in catalog
ext_info = catalog.get_extension_info(extension)
# Check if extension exists in catalog (supports both ID and display name)
ext_info, catalog_error = _resolve_catalog_extension(extension, catalog, "add")
if catalog_error:
console.print(f"[red]Error:[/red] Could not query extension catalog: {catalog_error}")
raise typer.Exit(1)
if not ext_info:
console.print(f"[red]Error:[/red] Extension '{extension}' not found in catalog")
console.print("\nSearch available extensions:")
@@ -2132,9 +2255,10 @@ def extension_add(
)
raise typer.Exit(1)
# Download extension ZIP
# Download extension ZIP (use resolved ID, not original argument which may be display name)
extension_id = ext_info['id']
console.print(f"Downloading {ext_info['name']} v{ext_info.get('version', 'unknown')}...")
zip_path = catalog.download_extension(extension)
zip_path = catalog.download_extension(extension_id)
try:
# Install from downloaded ZIP
@@ -2167,7 +2291,7 @@ def extension_add(
@extension_app.command("remove")
def extension_remove(
extension: str = typer.Argument(help="Extension ID to remove"),
extension: str = typer.Argument(help="Extension ID or name to remove"),
keep_config: bool = typer.Option(False, "--keep-config", help="Don't remove config files"),
force: bool = typer.Option(False, "--force", help="Skip confirmation"),
):
@@ -2185,25 +2309,19 @@ def extension_remove(
manager = ExtensionManager(project_root)
# Check if extension is installed
if not manager.registry.is_installed(extension):
console.print(f"[red]Error:[/red] Extension '{extension}' is not installed")
raise typer.Exit(1)
# Resolve extension ID from argument (handles ambiguous names)
installed = manager.list_installed()
extension_id, display_name = _resolve_installed_extension(extension, installed, "remove")
# Get extension info
ext_manifest = manager.get_extension(extension)
if ext_manifest:
ext_name = ext_manifest.name
cmd_count = len(ext_manifest.commands)
else:
ext_name = extension
cmd_count = 0
# Get extension info for command count
ext_manifest = manager.get_extension(extension_id)
cmd_count = len(ext_manifest.commands) if ext_manifest else 0
# Confirm removal
if not force:
console.print("\n[yellow]⚠ This will remove:[/yellow]")
console.print(f"{cmd_count} commands from AI agent")
console.print(f" • Extension directory: .specify/extensions/{extension}/")
console.print(f" • Extension directory: .specify/extensions/{extension_id}/")
if not keep_config:
console.print(" • Config files (will be backed up)")
console.print()
@@ -2214,15 +2332,15 @@ def extension_remove(
raise typer.Exit(0)
# Remove extension
success = manager.remove(extension, keep_config=keep_config)
success = manager.remove(extension_id, keep_config=keep_config)
if success:
console.print(f"\n[green]✓[/green] Extension '{ext_name}' removed successfully")
console.print(f"\n[green]✓[/green] Extension '{display_name}' removed successfully")
if keep_config:
console.print(f"\nConfig files preserved in .specify/extensions/{extension}/")
console.print(f"\nConfig files preserved in .specify/extensions/{extension_id}/")
else:
console.print(f"\nConfig files backed up to .specify/extensions/.backup/{extension}/")
console.print(f"\nTo reinstall: specify extension add {extension}")
console.print(f"\nConfig files backed up to .specify/extensions/.backup/{extension_id}/")
console.print(f"\nTo reinstall: specify extension add {extension_id}")
else:
console.print("[red]Error:[/red] Failed to remove extension")
raise typer.Exit(1)
@@ -2320,7 +2438,7 @@ def extension_info(
extension: str = typer.Argument(help="Extension ID or name"),
):
"""Show detailed information about an extension."""
from .extensions import ExtensionCatalog, ExtensionManager, ExtensionError
from .extensions import ExtensionCatalog, ExtensionManager
project_root = Path.cwd()
@@ -2333,118 +2451,181 @@ def extension_info(
catalog = ExtensionCatalog(project_root)
manager = ExtensionManager(project_root)
installed = manager.list_installed()
try:
ext_info = catalog.get_extension_info(extension)
# Try to resolve from installed extensions first (by ID or name)
# Use allow_not_found=True since the extension may be catalog-only
resolved_installed_id, resolved_installed_name = _resolve_installed_extension(
extension, installed, "info", allow_not_found=True
)
if not ext_info:
console.print(f"[red]Error:[/red] Extension '{extension}' not found in catalog")
console.print("\nTry: specify extension search")
raise typer.Exit(1)
# Try catalog lookup (with error handling)
# If we resolved an installed extension by display name, use its ID for catalog lookup
# to ensure we get the correct catalog entry (not a different extension with same name)
lookup_key = resolved_installed_id if resolved_installed_id else extension
ext_info, catalog_error = _resolve_catalog_extension(lookup_key, catalog, "info")
# Header
verified_badge = " [green]✓ Verified[/green]" if ext_info.get("verified") else ""
console.print(f"\n[bold]{ext_info['name']}[/bold] (v{ext_info['version']}){verified_badge}")
console.print(f"ID: {ext_info['id']}")
# Case 1: Found in catalog - show full catalog info
if ext_info:
_print_extension_info(ext_info, manager)
return
# Case 2: Installed locally but catalog lookup failed or not in catalog
if resolved_installed_id:
# Get local manifest info
ext_manifest = manager.get_extension(resolved_installed_id)
metadata = manager.registry.get(resolved_installed_id)
console.print(f"\n[bold]{resolved_installed_name}[/bold] (v{metadata.get('version', 'unknown')})")
console.print(f"ID: {resolved_installed_id}")
console.print()
# Description
console.print(f"{ext_info['description']}")
console.print()
# Author and License
console.print(f"[dim]Author:[/dim] {ext_info.get('author', 'Unknown')}")
console.print(f"[dim]License:[/dim] {ext_info.get('license', 'Unknown')}")
# Source catalog
if ext_info.get("_catalog_name"):
install_allowed = ext_info.get("_install_allowed", True)
install_note = "" if install_allowed else " [yellow](discovery only)[/yellow]"
console.print(f"[dim]Source catalog:[/dim] {ext_info['_catalog_name']}{install_note}")
console.print()
# Requirements
if ext_info.get('requires'):
console.print("[bold]Requirements:[/bold]")
reqs = ext_info['requires']
if reqs.get('speckit_version'):
console.print(f" • Spec Kit: {reqs['speckit_version']}")
if reqs.get('tools'):
for tool in reqs['tools']:
tool_name = tool['name']
tool_version = tool.get('version', 'any')
required = " (required)" if tool.get('required') else " (optional)"
console.print(f"{tool_name}: {tool_version}{required}")
if ext_manifest:
console.print(f"{ext_manifest.description}")
console.print()
# Author is optional in extension.yml, safely retrieve it
author = ext_manifest.data.get("extension", {}).get("author")
if author:
console.print(f"[dim]Author:[/dim] {author}")
console.print()
# Provides
if ext_info.get('provides'):
console.print("[bold]Provides:[/bold]")
provides = ext_info['provides']
if provides.get('commands'):
console.print(f" • Commands: {provides['commands']}")
if provides.get('hooks'):
console.print(f" • Hooks: {provides['hooks']}")
console.print()
if ext_manifest.commands:
console.print("[bold]Commands:[/bold]")
for cmd in ext_manifest.commands:
console.print(f"{cmd['name']}: {cmd.get('description', '')}")
console.print()
# Tags
if ext_info.get('tags'):
tags_str = ", ".join(ext_info['tags'])
console.print(f"[bold]Tags:[/bold] {tags_str}")
console.print()
# Statistics
stats = []
if ext_info.get('downloads') is not None:
stats.append(f"Downloads: {ext_info['downloads']:,}")
if ext_info.get('stars') is not None:
stats.append(f"Stars: {ext_info['stars']}")
if stats:
console.print(f"[bold]Statistics:[/bold] {' | '.join(stats)}")
console.print()
# Links
console.print("[bold]Links:[/bold]")
if ext_info.get('repository'):
console.print(f" • Repository: {ext_info['repository']}")
if ext_info.get('homepage'):
console.print(f" • Homepage: {ext_info['homepage']}")
if ext_info.get('documentation'):
console.print(f" • Documentation: {ext_info['documentation']}")
if ext_info.get('changelog'):
console.print(f" • Changelog: {ext_info['changelog']}")
console.print()
# Installation status and command
is_installed = manager.registry.is_installed(ext_info['id'])
install_allowed = ext_info.get("_install_allowed", True)
if is_installed:
console.print("[green]✓ Installed[/green]")
console.print(f"\nTo remove: specify extension remove {ext_info['id']}")
elif install_allowed:
console.print("[yellow]Not installed[/yellow]")
console.print(f"\n[cyan]Install:[/cyan] specify extension add {ext_info['id']}")
# Show catalog status
if catalog_error:
console.print(f"[yellow]Catalog unavailable:[/yellow] {catalog_error}")
console.print("[dim]Note: Using locally installed extension; catalog info could not be verified.[/dim]")
else:
catalog_name = ext_info.get("_catalog_name", "community")
console.print("[yellow]Not installed[/yellow]")
console.print(
f"\n[yellow]⚠[/yellow] '{ext_info['id']}' is available in the '{catalog_name}' catalog "
f"but not in your approved catalog. Add it to .specify/extension-catalogs.yml "
f"with install_allowed: true to enable installation."
)
console.print("[yellow]Note:[/yellow] Not found in catalog (custom/local extension)")
except ExtensionError as e:
console.print(f"\n[red]Error:[/red] {e}")
raise typer.Exit(1)
console.print()
console.print("[green]✓ Installed[/green]")
console.print(f"\nTo remove: specify extension remove {resolved_installed_id}")
return
# Case 3: Not found anywhere
if catalog_error:
console.print(f"[red]Error:[/red] Could not query extension catalog: {catalog_error}")
console.print("\nTry again when online, or use the extension ID directly.")
else:
console.print(f"[red]Error:[/red] Extension '{extension}' not found")
console.print("\nTry: specify extension search")
raise typer.Exit(1)
def _print_extension_info(ext_info: dict, manager):
"""Print formatted extension info from catalog data."""
# Header
verified_badge = " [green]✓ Verified[/green]" if ext_info.get("verified") else ""
console.print(f"\n[bold]{ext_info['name']}[/bold] (v{ext_info['version']}){verified_badge}")
console.print(f"ID: {ext_info['id']}")
console.print()
# Description
console.print(f"{ext_info['description']}")
console.print()
# Author and License
console.print(f"[dim]Author:[/dim] {ext_info.get('author', 'Unknown')}")
console.print(f"[dim]License:[/dim] {ext_info.get('license', 'Unknown')}")
# Source catalog
if ext_info.get("_catalog_name"):
install_allowed = ext_info.get("_install_allowed", True)
install_note = "" if install_allowed else " [yellow](discovery only)[/yellow]"
console.print(f"[dim]Source catalog:[/dim] {ext_info['_catalog_name']}{install_note}")
console.print()
# Requirements
if ext_info.get('requires'):
console.print("[bold]Requirements:[/bold]")
reqs = ext_info['requires']
if reqs.get('speckit_version'):
console.print(f" • Spec Kit: {reqs['speckit_version']}")
if reqs.get('tools'):
for tool in reqs['tools']:
tool_name = tool['name']
tool_version = tool.get('version', 'any')
required = " (required)" if tool.get('required') else " (optional)"
console.print(f"{tool_name}: {tool_version}{required}")
console.print()
# Provides
if ext_info.get('provides'):
console.print("[bold]Provides:[/bold]")
provides = ext_info['provides']
if provides.get('commands'):
console.print(f" • Commands: {provides['commands']}")
if provides.get('hooks'):
console.print(f" • Hooks: {provides['hooks']}")
console.print()
# Tags
if ext_info.get('tags'):
tags_str = ", ".join(ext_info['tags'])
console.print(f"[bold]Tags:[/bold] {tags_str}")
console.print()
# Statistics
stats = []
if ext_info.get('downloads') is not None:
stats.append(f"Downloads: {ext_info['downloads']:,}")
if ext_info.get('stars') is not None:
stats.append(f"Stars: {ext_info['stars']}")
if stats:
console.print(f"[bold]Statistics:[/bold] {' | '.join(stats)}")
console.print()
# Links
console.print("[bold]Links:[/bold]")
if ext_info.get('repository'):
console.print(f" • Repository: {ext_info['repository']}")
if ext_info.get('homepage'):
console.print(f" • Homepage: {ext_info['homepage']}")
if ext_info.get('documentation'):
console.print(f" • Documentation: {ext_info['documentation']}")
if ext_info.get('changelog'):
console.print(f" • Changelog: {ext_info['changelog']}")
console.print()
# Installation status and command
is_installed = manager.registry.is_installed(ext_info['id'])
install_allowed = ext_info.get("_install_allowed", True)
if is_installed:
console.print("[green]✓ Installed[/green]")
console.print(f"\nTo remove: specify extension remove {ext_info['id']}")
elif install_allowed:
console.print("[yellow]Not installed[/yellow]")
console.print(f"\n[cyan]Install:[/cyan] specify extension add {ext_info['id']}")
else:
catalog_name = ext_info.get("_catalog_name", "community")
console.print("[yellow]Not installed[/yellow]")
console.print(
f"\n[yellow]⚠[/yellow] '{ext_info['id']}' is available in the '{catalog_name}' catalog "
f"but not in your approved catalog. Add it to .specify/extension-catalogs.yml "
f"with install_allowed: true to enable installation."
)
@extension_app.command("update")
def extension_update(
extension: str = typer.Argument(None, help="Extension ID to update (or all)"),
extension: str = typer.Argument(None, help="Extension ID or name to update (or all)"),
):
"""Update extension(s) to latest version."""
from .extensions import ExtensionManager, ExtensionCatalog, ExtensionError
from .extensions import (
ExtensionManager,
ExtensionCatalog,
ExtensionError,
ValidationError,
CommandRegistrar,
HookExecutor,
)
from packaging import version as pkg_version
import shutil
project_root = Path.cwd()
@@ -2457,18 +2638,17 @@ def extension_update(
manager = ExtensionManager(project_root)
catalog = ExtensionCatalog(project_root)
speckit_version = get_speckit_version()
try:
# Get list of extensions to update
installed = manager.list_installed()
if extension:
# Update specific extension
if not manager.registry.is_installed(extension):
console.print(f"[red]Error:[/red] Extension '{extension}' is not installed")
raise typer.Exit(1)
extensions_to_update = [extension]
# Update specific extension - resolve ID from argument (handles ambiguous names)
extension_id, _ = _resolve_installed_extension(extension, installed, "update")
extensions_to_update = [extension_id]
else:
# Update all extensions
installed = manager.list_installed()
extensions_to_update = [ext["id"] for ext in installed]
if not extensions_to_update:
@@ -2482,7 +2662,16 @@ def extension_update(
for ext_id in extensions_to_update:
# Get installed version
metadata = manager.registry.get(ext_id)
installed_version = pkg_version.Version(metadata["version"])
if metadata is None or "version" not in metadata:
console.print(f"{ext_id}: Registry entry corrupted or missing (skipping)")
continue
try:
installed_version = pkg_version.Version(metadata["version"])
except pkg_version.InvalidVersion:
console.print(
f"{ext_id}: Invalid installed version '{metadata.get('version')}' in registry (skipping)"
)
continue
# Get catalog info
ext_info = catalog.get_extension_info(ext_id)
@@ -2490,12 +2679,24 @@ def extension_update(
console.print(f"{ext_id}: Not found in catalog (skipping)")
continue
catalog_version = pkg_version.Version(ext_info["version"])
# Check if installation is allowed from this catalog
if not ext_info.get("_install_allowed", True):
console.print(f"{ext_id}: Updates not allowed from '{ext_info.get('_catalog_name', 'catalog')}' (skipping)")
continue
try:
catalog_version = pkg_version.Version(ext_info["version"])
except pkg_version.InvalidVersion:
console.print(
f"{ext_id}: Invalid catalog version '{ext_info.get('version')}' (skipping)"
)
continue
if catalog_version > installed_version:
updates_available.append(
{
"id": ext_id,
"name": ext_info.get("name", ext_id), # Display name for status messages
"installed": str(installed_version),
"available": str(catalog_version),
"download_url": ext_info.get("download_url"),
@@ -2521,25 +2722,288 @@ def extension_update(
console.print("Cancelled")
raise typer.Exit(0)
# Perform updates
# Perform updates with atomic backup/restore
console.print()
updated_extensions = []
failed_updates = []
registrar = CommandRegistrar()
hook_executor = HookExecutor(project_root)
for update in updates_available:
ext_id = update["id"]
console.print(f"📦 Updating {ext_id}...")
extension_id = update["id"]
ext_name = update["name"] # Use display name for user-facing messages
console.print(f"📦 Updating {ext_name}...")
# TODO: Implement download and reinstall from URL
# For now, just show message
console.print(
"[yellow]Note:[/yellow] Automatic update not yet implemented. "
"Please update manually:"
)
console.print(f" specify extension remove {ext_id} --keep-config")
console.print(f" specify extension add {ext_id}")
# Backup paths
backup_base = manager.extensions_dir / ".backup" / f"{extension_id}-update"
backup_ext_dir = backup_base / "extension"
backup_commands_dir = backup_base / "commands"
backup_config_dir = backup_base / "config"
console.print(
"\n[cyan]Tip:[/cyan] Automatic updates will be available in a future version"
)
# Store backup state
backup_registry_entry = None
backup_hooks = None # None means no hooks key in config; {} means hooks key existed
backed_up_command_files = {}
try:
# 1. Backup registry entry (always, even if extension dir doesn't exist)
backup_registry_entry = manager.registry.get(extension_id)
# 2. Backup extension directory
extension_dir = manager.extensions_dir / extension_id
if extension_dir.exists():
backup_base.mkdir(parents=True, exist_ok=True)
if backup_ext_dir.exists():
shutil.rmtree(backup_ext_dir)
shutil.copytree(extension_dir, backup_ext_dir)
# Backup config files separately so they can be restored
# after a successful install (install_from_directory clears dest dir).
config_files = list(extension_dir.glob("*-config.yml")) + list(
extension_dir.glob("*-config.local.yml")
)
for cfg_file in config_files:
backup_config_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(cfg_file, backup_config_dir / cfg_file.name)
# 3. Backup command files for all agents
registered_commands = backup_registry_entry.get("registered_commands", {})
for agent_name, cmd_names in registered_commands.items():
if agent_name not in registrar.AGENT_CONFIGS:
continue
agent_config = registrar.AGENT_CONFIGS[agent_name]
commands_dir = project_root / agent_config["dir"]
for cmd_name in cmd_names:
cmd_file = commands_dir / f"{cmd_name}{agent_config['extension']}"
if cmd_file.exists():
backup_cmd_path = backup_commands_dir / agent_name / cmd_file.name
backup_cmd_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(cmd_file, backup_cmd_path)
backed_up_command_files[str(cmd_file)] = str(backup_cmd_path)
# Also backup copilot prompt files
if agent_name == "copilot":
prompt_file = project_root / ".github" / "prompts" / f"{cmd_name}.prompt.md"
if prompt_file.exists():
backup_prompt_path = backup_commands_dir / "copilot-prompts" / prompt_file.name
backup_prompt_path.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(prompt_file, backup_prompt_path)
backed_up_command_files[str(prompt_file)] = str(backup_prompt_path)
# 4. Backup hooks from extensions.yml
# Use backup_hooks=None to indicate config had no "hooks" key (don't create on restore)
# Use backup_hooks={} to indicate config had "hooks" key with no hooks for this extension
config = hook_executor.get_project_config()
if "hooks" in config:
backup_hooks = {} # Config has hooks key - preserve this fact
for hook_name, hook_list in config["hooks"].items():
ext_hooks = [h for h in hook_list if h.get("extension") == extension_id]
if ext_hooks:
backup_hooks[hook_name] = ext_hooks
# 5. Download new version
zip_path = catalog.download_extension(extension_id)
try:
# 6. Validate extension ID from ZIP BEFORE modifying installation
# Handle both root-level and nested extension.yml (GitHub auto-generated ZIPs)
with zipfile.ZipFile(zip_path, "r") as zf:
import yaml
manifest_data = None
namelist = zf.namelist()
# First try root-level extension.yml
if "extension.yml" in namelist:
with zf.open("extension.yml") as f:
manifest_data = yaml.safe_load(f) or {}
else:
# Look for extension.yml in a single top-level subdirectory
# (e.g., "repo-name-branch/extension.yml")
manifest_paths = [n for n in namelist if n.endswith("/extension.yml") and n.count("/") == 1]
if len(manifest_paths) == 1:
with zf.open(manifest_paths[0]) as f:
manifest_data = yaml.safe_load(f) or {}
if manifest_data is None:
raise ValueError("Downloaded extension archive is missing 'extension.yml'")
zip_extension_id = manifest_data.get("extension", {}).get("id")
if zip_extension_id != extension_id:
raise ValueError(
f"Extension ID mismatch: expected '{extension_id}', got '{zip_extension_id}'"
)
# 7. Remove old extension (handles command file cleanup and registry removal)
manager.remove(extension_id, keep_config=True)
# 8. Install new version
_ = manager.install_from_zip(zip_path, speckit_version)
# Restore user config files from backup after successful install.
new_extension_dir = manager.extensions_dir / extension_id
if backup_config_dir.exists() and new_extension_dir.exists():
for cfg_file in backup_config_dir.iterdir():
if cfg_file.is_file():
shutil.copy2(cfg_file, new_extension_dir / cfg_file.name)
# 9. Restore metadata from backup (installed_at, enabled state)
if backup_registry_entry:
# Copy current registry entry to avoid mutating internal
# registry state before explicit restore().
current_metadata = manager.registry.get(extension_id)
if current_metadata is None:
raise RuntimeError(
f"Registry entry for '{extension_id}' missing after install — update incomplete"
)
new_metadata = dict(current_metadata)
# Preserve the original installation timestamp
if "installed_at" in backup_registry_entry:
new_metadata["installed_at"] = backup_registry_entry["installed_at"]
# If extension was disabled before update, disable it again
if not backup_registry_entry.get("enabled", True):
new_metadata["enabled"] = False
# Use restore() instead of update() because update() always
# preserves the existing installed_at, ignoring our override
manager.registry.restore(extension_id, new_metadata)
# Also disable hooks in extensions.yml if extension was disabled
if not backup_registry_entry.get("enabled", True):
config = hook_executor.get_project_config()
if "hooks" in config:
for hook_name in config["hooks"]:
for hook in config["hooks"][hook_name]:
if hook.get("extension") == extension_id:
hook["enabled"] = False
hook_executor.save_project_config(config)
finally:
# Clean up downloaded ZIP
if zip_path.exists():
zip_path.unlink()
# 10. Clean up backup on success
if backup_base.exists():
shutil.rmtree(backup_base)
console.print(f" [green]✓[/green] Updated to v{update['available']}")
updated_extensions.append(ext_name)
except KeyboardInterrupt:
raise
except Exception as e:
console.print(f" [red]✗[/red] Failed: {e}")
failed_updates.append((ext_name, str(e)))
# Rollback on failure
console.print(f" [yellow]↩[/yellow] Rolling back {ext_name}...")
try:
# Restore extension directory
# Only perform destructive rollback if backup exists (meaning we
# actually modified the extension). This avoids deleting a valid
# installation when failure happened before changes were made.
extension_dir = manager.extensions_dir / extension_id
if backup_ext_dir.exists():
if extension_dir.exists():
shutil.rmtree(extension_dir)
shutil.copytree(backup_ext_dir, extension_dir)
# Remove any NEW command files created by failed install
# (files that weren't in the original backup)
try:
new_registry_entry = manager.registry.get(extension_id)
if new_registry_entry is None:
new_registered_commands = {}
else:
new_registered_commands = new_registry_entry.get("registered_commands", {})
for agent_name, cmd_names in new_registered_commands.items():
if agent_name not in registrar.AGENT_CONFIGS:
continue
agent_config = registrar.AGENT_CONFIGS[agent_name]
commands_dir = project_root / agent_config["dir"]
for cmd_name in cmd_names:
cmd_file = commands_dir / f"{cmd_name}{agent_config['extension']}"
# Delete if it exists and wasn't in our backup
if cmd_file.exists() and str(cmd_file) not in backed_up_command_files:
cmd_file.unlink()
# Also handle copilot prompt files
if agent_name == "copilot":
prompt_file = project_root / ".github" / "prompts" / f"{cmd_name}.prompt.md"
if prompt_file.exists() and str(prompt_file) not in backed_up_command_files:
prompt_file.unlink()
except KeyError:
pass # No new registry entry exists, nothing to clean up
# Restore backed up command files
for original_path, backup_path in backed_up_command_files.items():
backup_file = Path(backup_path)
if backup_file.exists():
original_file = Path(original_path)
original_file.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(backup_file, original_file)
# Restore hooks in extensions.yml
# - backup_hooks=None means original config had no "hooks" key
# - backup_hooks={} or {...} means config had hooks key
config = hook_executor.get_project_config()
if "hooks" in config:
modified = False
if backup_hooks is None:
# Original config had no "hooks" key; remove it entirely
del config["hooks"]
modified = True
else:
# Remove any hooks for this extension added by failed install
for hook_name, hooks_list in config["hooks"].items():
original_len = len(hooks_list)
config["hooks"][hook_name] = [
h for h in hooks_list
if h.get("extension") != extension_id
]
if len(config["hooks"][hook_name]) != original_len:
modified = True
# Add back the backed up hooks if any
if backup_hooks:
for hook_name, hooks in backup_hooks.items():
if hook_name not in config["hooks"]:
config["hooks"][hook_name] = []
config["hooks"][hook_name].extend(hooks)
modified = True
if modified:
hook_executor.save_project_config(config)
# Restore registry entry (use restore() since entry was removed)
if backup_registry_entry:
manager.registry.restore(extension_id, backup_registry_entry)
console.print(" [green]✓[/green] Rollback successful")
# Clean up backup directory only on successful rollback
if backup_base.exists():
shutil.rmtree(backup_base)
except Exception as rollback_error:
console.print(f" [red]✗[/red] Rollback failed: {rollback_error}")
console.print(f" [dim]Backup preserved at: {backup_base}[/dim]")
# Summary
console.print()
if updated_extensions:
console.print(f"[green]✓[/green] Successfully updated {len(updated_extensions)} extension(s)")
if failed_updates:
console.print(f"[red]✗[/red] Failed to update {len(failed_updates)} extension(s):")
for ext_name, error in failed_updates:
console.print(f"{ext_name}: {error}")
raise typer.Exit(1)
except ValidationError as e:
console.print(f"\n[red]Validation Error:[/red] {e}")
raise typer.Exit(1)
except ExtensionError as e:
console.print(f"\n[red]Error:[/red] {e}")
raise typer.Exit(1)
@@ -2547,7 +3011,7 @@ def extension_update(
@extension_app.command("enable")
def extension_enable(
extension: str = typer.Argument(help="Extension ID to enable"),
extension: str = typer.Argument(help="Extension ID or name to enable"),
):
"""Enable a disabled extension."""
from .extensions import ExtensionManager, HookExecutor
@@ -2564,34 +3028,38 @@ def extension_enable(
manager = ExtensionManager(project_root)
hook_executor = HookExecutor(project_root)
if not manager.registry.is_installed(extension):
console.print(f"[red]Error:[/red] Extension '{extension}' is not installed")
raise typer.Exit(1)
# Resolve extension ID from argument (handles ambiguous names)
installed = manager.list_installed()
extension_id, display_name = _resolve_installed_extension(extension, installed, "enable")
# Update registry
metadata = manager.registry.get(extension)
metadata = manager.registry.get(extension_id)
if metadata is None:
console.print(f"[red]Error:[/red] Extension '{extension_id}' not found in registry (corrupted state)")
raise typer.Exit(1)
if metadata.get("enabled", True):
console.print(f"[yellow]Extension '{extension}' is already enabled[/yellow]")
console.print(f"[yellow]Extension '{display_name}' is already enabled[/yellow]")
raise typer.Exit(0)
metadata["enabled"] = True
manager.registry.add(extension, metadata)
manager.registry.update(extension_id, metadata)
# Enable hooks in extensions.yml
config = hook_executor.get_project_config()
if "hooks" in config:
for hook_name in config["hooks"]:
for hook in config["hooks"][hook_name]:
if hook.get("extension") == extension:
if hook.get("extension") == extension_id:
hook["enabled"] = True
hook_executor.save_project_config(config)
console.print(f"[green]✓[/green] Extension '{extension}' enabled")
console.print(f"[green]✓[/green] Extension '{display_name}' enabled")
@extension_app.command("disable")
def extension_disable(
extension: str = typer.Argument(help="Extension ID to disable"),
extension: str = typer.Argument(help="Extension ID or name to disable"),
):
"""Disable an extension without removing it."""
from .extensions import ExtensionManager, HookExecutor
@@ -2608,31 +3076,35 @@ def extension_disable(
manager = ExtensionManager(project_root)
hook_executor = HookExecutor(project_root)
if not manager.registry.is_installed(extension):
console.print(f"[red]Error:[/red] Extension '{extension}' is not installed")
raise typer.Exit(1)
# Resolve extension ID from argument (handles ambiguous names)
installed = manager.list_installed()
extension_id, display_name = _resolve_installed_extension(extension, installed, "disable")
# Update registry
metadata = manager.registry.get(extension)
metadata = manager.registry.get(extension_id)
if metadata is None:
console.print(f"[red]Error:[/red] Extension '{extension_id}' not found in registry (corrupted state)")
raise typer.Exit(1)
if not metadata.get("enabled", True):
console.print(f"[yellow]Extension '{extension}' is already disabled[/yellow]")
console.print(f"[yellow]Extension '{display_name}' is already disabled[/yellow]")
raise typer.Exit(0)
metadata["enabled"] = False
manager.registry.add(extension, metadata)
manager.registry.update(extension_id, metadata)
# Disable hooks in extensions.yml
config = hook_executor.get_project_config()
if "hooks" in config:
for hook_name in config["hooks"]:
for hook in config["hooks"][hook_name]:
if hook.get("extension") == extension:
if hook.get("extension") == extension_id:
hook["enabled"] = False
hook_executor.save_project_config(config)
console.print(f"[green]✓[/green] Extension '{extension}' disabled")
console.print(f"[green]✓[/green] Extension '{display_name}' disabled")
console.print("\nCommands will no longer be available. Hooks will not execute.")
console.print(f"To re-enable: specify extension enable {extension}")
console.print(f"To re-enable: specify extension enable {extension_id}")
def main():