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
41 changes: 34 additions & 7 deletions amplifier_app_cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -718,7 +718,16 @@ def _populate_mode_shortcuts(self) -> None:
CommandProcessor.MODE_SHORTCUTS.update(shortcuts)

def _populate_skill_shortcuts(self) -> None:
"""Populate SKILL_SHORTCUTS from skills discovery."""
"""Populate SKILL_SHORTCUTS from skills discovery.

NOTE (additive-only cache): this only ever ADDS entries via
``.update()`` on the CLASS-level dict -- it never removes a shortcut
that has since disappeared from discovery. That's an accepted
limitation for this fix: skills don't disappear mid-session, so a
stale-but-present entry is not a practical problem. Safe to call
repeatedly (e.g. to refresh after lazy skill resolution) precisely
because it's additive rather than a rebuild.
"""
discovery = self.session.coordinator.get_capability("skills_discovery")
if discovery and hasattr(discovery, "get_shortcuts"):
shortcuts = discovery.get_shortcuts()
Expand Down Expand Up @@ -790,12 +799,10 @@ def process_input(self, user_input: str) -> tuple[str, dict[str, Any]]:
# canonical skill name when the lookup key is an alias (the
# skill's `shortcut:` frontmatter field). Older skills bundles
# don't populate "name" — fall back to the lookup key.
if shortcut_name in self.SKILL_SHORTCUTS:
entry = self.SKILL_SHORTCUTS[shortcut_name]
def _dispatch_skill_shortcut(name: str) -> tuple[str, dict[str, Any]]:
entry = self.SKILL_SHORTCUTS[name]
canonical = (
entry.get("name", shortcut_name)
if isinstance(entry, dict)
else shortcut_name
entry.get("name", name) if isinstance(entry, dict) else name
)
return (
"load_skill",
Expand All @@ -806,6 +813,22 @@ def process_input(self, user_input: str) -> tuple[str, dict[str, Any]]:
},
)

if shortcut_name in self.SKILL_SHORTCUTS:
return _dispatch_skill_shortcut(shortcut_name)

# Defense-in-depth: SKILL_SHORTCUTS is populated once, additively,
# at __init__ time (see _populate_skill_shortcuts). Skills sourced
# from @namespace:skills packs resolve LAZILY on first
# provider:request, which happens AFTER that startup snapshot --
# so a miss here does not necessarily mean the command is
# unknown. Refresh against the live skills_discovery capability
# and recheck once before giving up. Cheap (a dict read off an
# already-resolved capability) and fails soft if the capability
# is absent, exactly like the initial population.
self._populate_skill_shortcuts()
if shortcut_name in self.SKILL_SHORTCUTS:
return _dispatch_skill_shortcut(shortcut_name)

return "unknown_command", {"command": command}

# Regular prompt
Expand Down Expand Up @@ -2024,7 +2047,11 @@ def _format_help(self) -> str:
lines.append(f" /{name}")

# Add dynamic skills section if skills are available
# Use cached SKILL_SHORTCUTS (same source as process_input) for consistency
# Refresh the cache first so displayed shortcuts reflect skills that
# resolved lazily (e.g. @namespace:skills packs) after __init__'s
# startup snapshot -- then use SKILL_SHORTCUTS (same source as
# process_input) for consistency.
self._populate_skill_shortcuts()
shortcuts = self.SKILL_SHORTCUTS
if shortcuts:
lines.append("")
Expand Down
181 changes: 181 additions & 0 deletions tests/test_skill_shortcuts_live_fallback.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,181 @@
"""Tests for defense-in-depth stale-SKILL_SHORTCUTS handling.

SKILL_SHORTCUTS is populated once, additively, at CommandProcessor.__init__
time (see _populate_skill_shortcuts). Skills sourced from @namespace:skills
packs resolve LAZILY -- on first provider:request -- which happens AFTER
that startup snapshot. These tests simulate that timing: the discovery
capability's get_shortcuts() return value changes AFTER the CommandProcessor
is constructed (as it would once the lazy source resolves), and verify that:

1. Dispatch (process_input) falls back to a live re-check of the discovery
capability on a cache miss, so a skill that only became visible after
__init__ is still dispatched correctly.
2. A genuinely unknown command (never present in discovery, live or not)
still returns 'unknown_command'.
3. /help ("_format_help") reflects a skill that appeared after __init__,
because it refreshes the cache before rendering.
"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from amplifier_app_cli.main import CommandProcessor
from helpers import _make_command_processor


@pytest.fixture(autouse=True)
def reset_skill_shortcuts():
"""Clear SKILL_SHORTCUTS before and after every test in this module.

SKILL_SHORTCUTS is a CLASS-level dict shared across CommandProcessor
instances -- reset it so state never leaks between tests.
"""
CommandProcessor.SKILL_SHORTCUTS.clear()
yield
CommandProcessor.SKILL_SHORTCUTS.clear()


def _make_lazy_discovery(initial_shortcuts: dict) -> MagicMock:
"""A skills_discovery mock whose get_shortcuts() result can be mutated
later, simulating a skill resolving lazily after CommandProcessor
construction.
"""
mock_discovery = MagicMock()
mock_discovery.get_shortcuts.return_value = dict(initial_shortcuts)
return mock_discovery


class TestDispatchFallsBackToLiveDiscovery:
"""A skill that appears in discovery AFTER __init__ must still dispatch."""

def test_skill_added_after_init_still_dispatches(self):
"""A skill unresolved at __init__ time resolves via live fallback."""
# At construction time, discovery has no shortcuts yet (lazy source
# not yet resolved).
mock_discovery = _make_lazy_discovery({})
cp = _make_command_processor(skills_discovery=mock_discovery)

# Sanity: the shortcut is genuinely absent from the startup snapshot.
assert "wayfinder-pack" not in CommandProcessor.SKILL_SHORTCUTS

# Simulate lazy resolution: the skill now exists in the live
# capability (as it would after a first provider:request).
mock_discovery.get_shortcuts.return_value = {
"wayfinder-pack": {
"name": "wayfinder-pack",
"description": "Wayfinder pack skill",
}
}

action, data = cp.process_input("/wayfinder-pack")

assert action == "load_skill"
assert data["skill_name"] == "wayfinder-pack"
assert data["command"] == "/wayfinder-pack"

def test_skill_added_after_init_with_arguments(self):
"""Arguments after a late-resolving skill shortcut are preserved."""
mock_discovery = _make_lazy_discovery({})
cp = _make_command_processor(skills_discovery=mock_discovery)

mock_discovery.get_shortcuts.return_value = {
"seam-test": {"name": "seam-test", "description": "Seam test skill"}
}

action, data = cp.process_input("/seam-test check the seams")

assert action == "load_skill"
assert data["skill_name"] == "seam-test"
assert data["arguments"] == "check the seams"

def test_alias_shortcut_resolves_to_canonical_name_after_refresh(self):
"""A late-resolving alias entry still maps to its canonical name."""
mock_discovery = _make_lazy_discovery({})
cp = _make_command_processor(skills_discovery=mock_discovery)

mock_discovery.get_shortcuts.return_value = {
"dc": {"name": "design-council", "description": "Design council"}
}

action, data = cp.process_input("/dc")

assert action == "load_skill"
assert data["skill_name"] == "design-council"

def test_fallback_refreshes_class_level_cache(self):
"""A successful live fallback should also update SKILL_SHORTCUTS
(additive), so subsequent lookups don't need to refresh again."""
mock_discovery = _make_lazy_discovery({})
cp = _make_command_processor(skills_discovery=mock_discovery)

mock_discovery.get_shortcuts.return_value = {
"simplify": {"name": "simplify"}
}

cp.process_input("/simplify")

assert "simplify" in CommandProcessor.SKILL_SHORTCUTS


class TestUnknownCommandStillUnknownAfterFallback:
"""A command absent from discovery, live or cached, is still unknown."""

def test_genuinely_unknown_command_returns_unknown_command(self):
"""A command never present in discovery must still be unknown_command."""
mock_discovery = _make_lazy_discovery({"simplify": {"name": "simplify"}})
cp = _make_command_processor(skills_discovery=mock_discovery)

action, data = cp.process_input("/totally-not-a-real-skill")

assert action == "unknown_command"
assert data["command"] == "/totally-not-a-real-skill"

def test_no_discovery_capability_fails_soft(self):
"""With no skills_discovery capability at all, fallback must not
raise and must still return unknown_command (fail-soft parity with
the pre-existing miss path)."""
cp = _make_command_processor() # no skills_discovery

action, data = cp.process_input("/notaskill")

assert action == "unknown_command"
assert data["command"] == "/notaskill"

def test_discovery_without_get_shortcuts_fails_soft(self):
"""A discovery object lacking get_shortcuts() must not raise during
the live-fallback refresh."""

class SimpleDiscovery:
pass

cp = _make_command_processor(skills_discovery=SimpleDiscovery())

action, _data = cp.process_input("/notaskill")

assert action == "unknown_command"


class TestHelpDisplayFreshness:
"""/help ("_format_help") must reflect skills that resolved after __init__."""

def test_help_shows_skill_added_after_init(self):
"""A skill shortcut that only appears after construction should show
up in the /help output, because _format_help refreshes the cache
before rendering."""
mock_discovery = _make_lazy_discovery({})
cp = _make_command_processor(skills_discovery=mock_discovery)

mock_discovery.get_shortcuts.return_value = {
"wayfinder-pack": {
"name": "wayfinder-pack",
"description": "Wayfinder pack skill",
}
}

help_text = cp._format_help()

assert "/wayfinder-pack" in help_text
assert "Wayfinder pack skill" in help_text
Loading