# ADR-057: CommandRegistry - Unified Command Discovery and Routing
Status: APPROVED (Phase 3 Implementation In Progress) Issue: #551 ARCH-COMMANDS Date: 2026-01-22 Decision Makers: Lead Developer, Chief Architect, PM
Commands in Piper Morgan are scattered across 6 registration points with no single source of truth:
| Interface | Commands | Registration Location |
|---|---|---|
| CLI (argparse) | 6 | main.py |
| CLI (Click) | 23+ | cli/commands/*.py |
| Web Chat Patterns | 541+ → ~20 intents | services/intent_service/pre_classifier.py |
| Slack Commands | 2 | services/integrations/slack/webhook_router.py |
| URL Routes | 202 | web/api/routes/*.py |
| Action Registry | 1+ | services/actions/action_registry.py |
/piper help hardcoded, doesn’t reflect actual capabilitiesFrom docs/internal/architecture/current/command-inventory.md:
Key Gap Categories:
┌─────────────────────┐
│ CommandRegistry │
│ (Single Source) │
└─────────┬───────────┘
│
┌───────────────┬───────────┼───────────┬───────────────┐
│ │ │ │ │
▼ ▼ ▼ ▼ ▼
┌───────────┐ ┌───────────┐ ┌─────────┐ ┌─────────┐ ┌───────────┐
│ CLI │ │ Web Chat │ │ Slack │ │ URL │ │ Discovery │
│ Adapter │ │ Adapter │ │ Adapter │ │ Adapter │ │ Handler │
└───────────┘ └───────────┘ └─────────┘ └─────────┘ └───────────┘
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any, Callable
from enum import Enum
class CommandInterface(Enum):
"""Interfaces where a command can be exposed"""
CLI = "cli"
WEB_CHAT = "web_chat"
SLACK = "slack"
URL = "url"
ALL = "all" # Exposed on all interfaces
class CommandCategory(Enum):
"""Functional categories for command organization"""
CALENDAR = "calendar"
TODOS = "todos"
PROJECTS = "projects"
GITHUB = "github"
STANDUP = "standup"
SETTINGS = "settings"
HELP = "help"
ADMIN = "admin"
@dataclass
class InterfaceConfig:
"""Configuration for a specific interface"""
enabled: bool = True
aliases: List[str] = field(default_factory=list)
description_override: Optional[str] = None
requires_auth: bool = True
# Interface-specific options
slack_response_type: str = "ephemeral" # or "in_channel"
cli_group: Optional[str] = None # Click group name
url_method: str = "GET" # HTTP method
url_path: Optional[str] = None # Route path override
@dataclass
class CommandDefinition:
"""Central definition of a command across all interfaces"""
# Identity
name: str # Canonical name (e.g., "calendar_today")
display_name: str # Human-readable (e.g., "Today's Calendar")
description: str # What it does
category: CommandCategory
# Interface Exposure
interfaces: Dict[CommandInterface, InterfaceConfig] = field(default_factory=dict)
# Handler Reference (existing handlers continue to work)
handler_module: str # e.g., "services.intent_service.canonical_handlers"
handler_name: str # e.g., "_handle_temporal_query"
# Discovery Metadata
examples: List[str] = field(default_factory=list) # Example invocations
keywords: List[str] = field(default_factory=list) # Search terms
help_text: Optional[str] = None # Detailed help
# Execution Metadata
requires_integration: Optional[str] = None # e.g., "calendar", "github"
execution_type: str = "query" # "query", "mutation", "action"
def is_available_on(self, interface: CommandInterface) -> bool:
"""Check if command is available on given interface"""
if CommandInterface.ALL in self.interfaces:
return self.interfaces[CommandInterface.ALL].enabled
return interface in self.interfaces and self.interfaces[interface].enabled
def get_interface_config(self, interface: CommandInterface) -> Optional[InterfaceConfig]:
"""Get configuration for specific interface"""
if CommandInterface.ALL in self.interfaces:
return self.interfaces[CommandInterface.ALL]
return self.interfaces.get(interface)
from typing import Dict, List, Optional, Callable
import logging
class CommandRegistry:
"""Central registry for all Piper commands"""
_commands: Dict[str, CommandDefinition] = {}
_by_category: Dict[CommandCategory, List[str]] = {}
_by_interface: Dict[CommandInterface, List[str]] = {}
_initialized: bool = False
@classmethod
def register(cls, command: CommandDefinition) -> None:
"""Register a command definition"""
cls._commands[command.name] = command
# Index by category
if command.category not in cls._by_category:
cls._by_category[command.category] = []
cls._by_category[command.category].append(command.name)
# Index by interface
for interface in command.interfaces:
if interface not in cls._by_interface:
cls._by_interface[interface] = []
cls._by_interface[interface].append(command.name)
@classmethod
def get_command(cls, name: str) -> Optional[CommandDefinition]:
"""Get a command by canonical name"""
return cls._commands.get(name)
@classmethod
def list_commands(cls,
interface: Optional[CommandInterface] = None,
category: Optional[CommandCategory] = None) -> List[CommandDefinition]:
"""List commands, optionally filtered"""
commands = list(cls._commands.values())
if interface:
commands = [c for c in commands if c.is_available_on(interface)]
if category:
commands = [c for c in commands if c.category == category]
return commands
@classmethod
def get_help(cls, interface: CommandInterface) -> str:
"""Generate help text for an interface"""
commands = cls.list_commands(interface=interface)
# Group by category
by_category: Dict[CommandCategory, List[CommandDefinition]] = {}
for cmd in commands:
if cmd.category not in by_category:
by_category[cmd.category] = []
by_category[cmd.category].append(cmd)
# Format help
lines = ["**Available Commands**\n"]
for category, cmds in sorted(by_category.items(), key=lambda x: x[0].value):
lines.append(f"\n**{category.value.title()}**")
for cmd in cmds:
config = cmd.get_interface_config(interface)
desc = config.description_override if config and config.description_override else cmd.description
lines.append(f" • {cmd.display_name}: {desc}")
return "\n".join(lines)
@classmethod
def find_by_keyword(cls, keyword: str, interface: Optional[CommandInterface] = None) -> List[CommandDefinition]:
"""Find commands matching a keyword"""
keyword_lower = keyword.lower()
matches = []
for cmd in cls.list_commands(interface=interface):
if (keyword_lower in cmd.name.lower() or
keyword_lower in cmd.display_name.lower() or
any(keyword_lower in kw.lower() for kw in cmd.keywords)):
matches.append(cmd)
return matches
class SlackCommandAdapter:
"""Adapts CommandRegistry for Slack slash commands"""
@staticmethod
def get_slash_commands() -> Dict[str, Callable]:
"""Get all Slack slash commands from registry"""
commands = CommandRegistry.list_commands(interface=CommandInterface.SLACK)
slash_commands = {}
for cmd in commands:
config = cmd.get_interface_config(CommandInterface.SLACK)
if config and config.enabled:
# Primary command
slash_commands[f"/{cmd.name}"] = SlackCommandAdapter._create_handler(cmd)
# Aliases
for alias in config.aliases:
slash_commands[f"/{alias}"] = SlackCommandAdapter._create_handler(cmd)
return slash_commands
@staticmethod
def _create_handler(cmd: CommandDefinition):
"""Create Slack handler wrapper for command"""
async def handler(payload: dict) -> dict:
# Load actual handler
module = importlib.import_module(cmd.handler_module)
handler_fn = getattr(module, cmd.handler_name)
# Execute
result = await handler_fn(payload)
# Format for Slack
config = cmd.get_interface_config(CommandInterface.SLACK)
return {
"response_type": config.slack_response_type if config else "ephemeral",
"text": result.get("message", str(result))
}
return handler
@staticmethod
def build_help_response() -> dict:
"""Build Slack-formatted help response"""
help_text = CommandRegistry.get_help(CommandInterface.SLACK)
return {
"response_type": "ephemeral",
"blocks": [
{
"type": "section",
"text": {"type": "mrkdwn", "text": help_text}
}
]
}
CommandRegistry classCommandDefinition dataclass/standup in CommandRegistry/piper help from registry| Registry | Purpose | Relationship |
|---|---|---|
| PluginRegistry | Integration lifecycle | CommandRegistry may query for available integrations |
| ActionRegistry | Mutation execution | Commands may reference actions for mutations |
| ServiceContainer | Service lifecycle | CommandRegistry handlers access services |
Integration: CommandRegistry complements rather than replaces these. Commands may:
PluginRegistry.get_enabled_plugins() for availabilityActionRegistry.execute() for mutationsServiceContainer.get_service()# services/commands/command_definitions.py
from services.commands.registry import CommandRegistry, CommandDefinition, CommandInterface, InterfaceConfig, CommandCategory
# Standup command - available everywhere
STANDUP_COMMAND = CommandDefinition(
name="standup",
display_name="Daily Standup",
description="Generate your daily standup report",
category=CommandCategory.STANDUP,
interfaces={
CommandInterface.ALL: InterfaceConfig(
enabled=True,
aliases=["standup", "daily"],
slack_response_type="in_channel",
cli_group="standup",
url_path="/api/v1/standup/generate"
)
},
handler_module="services.standup.standup_service",
handler_name="generate_standup",
examples=["show standup", "/standup", "what's my standup?"],
keywords=["standup", "daily", "yesterday", "today", "blockers"],
requires_integration=None, # No external integration required
execution_type="query"
)
# Calendar today - not yet on Slack (gap to close)
CALENDAR_TODAY_COMMAND = CommandDefinition(
name="calendar_today",
display_name="Today's Calendar",
description="Show your meetings for today",
category=CommandCategory.CALENDAR,
interfaces={
CommandInterface.WEB_CHAT: InterfaceConfig(enabled=True),
CommandInterface.CLI: InterfaceConfig(enabled=True, cli_group="cal"),
CommandInterface.SLACK: InterfaceConfig(enabled=False), # Gap: #551 Phase 3 will enable
CommandInterface.URL: InterfaceConfig(enabled=True, url_path="/api/v1/calendar/today")
},
handler_module="services.intent_service.canonical_handlers",
handler_name="_handle_temporal_query",
examples=["what meetings do I have today?", "cal today", "show calendar"],
keywords=["calendar", "meetings", "today", "schedule"],
requires_integration="calendar",
execution_type="query"
)
def register_all():
"""Register all command definitions"""
CommandRegistry.register(STANDUP_COMMAND)
CommandRegistry.register(CALENDAR_TODAY_COMMAND)
# ... more commands
PROPOSED - Awaiting PM review before implementation.
services/commands/registry.py with core classesservices/commands/definitions.py with initial commands/standup as proof of concept/piper help from registryservices/commands/
├── __init__.py
├── registry.py # CommandRegistry class
├── definitions.py # CommandDefinition dataclass
├── adapters/
│ ├── __init__.py
│ ├── base.py # BaseAdapter interface
│ ├── slack_adapter.py
│ ├── cli_adapter.py
│ └── webchat_adapter.py
└── commands/
├── __init__.py
├── standup.py # STANDUP_COMMAND
├── calendar.py # CALENDAR_* commands
└── todos.py # TODO_* commands
docs/internal/architecture/current/command-inventory.md: Phase 1 inventoryservices/plugins/plugin_registry.py: Existing registry patternservices/actions/action_registry.py: Existing action patternservices/container/service_container.py: Service lifecycle patternADR-057 created: 2026-01-22 Phase 2 of Issue #551 ARCH-COMMANDS