diff --git a/AGENTS.md b/AGENTS.md index fd624f4cb..187c435a0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -242,9 +242,13 @@ In M3, the old `ResourceProvider` hierarchy was replaced with native pydantic-ai - `AggregatedResourceSource` — Composes multiple `ResourceSource` instances, routes by URI scheme. - `AgentContext` (`capabilities/agent_context.py`) — Frozen dataclass carrying `agent_registry`, `delegation`, `session`, `scope`, `resources`, `host`. Constructed by RunLoop per-turn. - `DelegationService` (`capabilities/delegation.py`) — Protocol exposing `spawn_subagent(name, prompt)` and `get_available_agents()`. Limits tools to operations they need without exposing `AgentPool`. -- `ChangeEvent` (`capabilities/change_event.py`) — Frozen dataclass for capability change notifications (`on_change()` stream). +- `ChangeEvent` (`capabilities/change_event.py`) — Frozen dataclass for capability change notifications (`on_change()` stream). `kind` accepts `"commands_changed"` in addition to `"tools_changed"`, `"prompts_changed"`, `"resources_changed"`, `"skills_changed"`. +- `CommandResource` (`capabilities/resource_protocols.py`) — `@runtime_checkable Protocol` for capabilities that produce slash commands. Implements `list_commands() -> Sequence[CommandEntry]` and `async get_command(name) -> CommandEntry | None`. `CommandEntry` carries an optional `handler: Callable[[str, AgentContext], Awaitable[str]]` for direct execution. +- `CommandBridge` (`capabilities/command_bridge.py`) — Connects `ExtensionRegistry` to protocol servers (ACP, OpenCode). Discovers commands from all `CommandResource` capabilities, de-duplicates by name (TURN → AGENT → SESSION → POOL), executes via `CommandEntry.handler`, and watches for `"commands_changed"`/`"skills_changed"`/`"prompts_changed"` events. Per-session lifecycle with `Scope(SESSION)`. - Entry-point registry (`capabilities/registry.py`) — Discovers custom capabilities via `agentpool.capabilities` entry-point group. +**Command Registration via Capabilities:** Capabilities implementing `CommandResource` can publish slash commands visible to ACP and OpenCode clients. `CommandBridge` discovers these via `ExtensionRegistry.get_command_resources(scope)`, converts them to protocol-specific command formats, and routes execution back through `CommandEntry.handler`. See `examples/custom_command_capability.py` for a complete example. + **Deleted alongside ResourceProviders:** - `src/agentpool/tools/factory.py` (194 LOC, 6 `ToolsetFactory` classes) — became dead code after all providers migrated. - `src/agentpool/tools/manager.py` (364 LOC, `ToolManager`) — all `agent.tools.X` access migrated to direct capability references. diff --git a/docs/rfcs/draft/RFC-0058-capability-command-bridge.md b/docs/rfcs/draft/RFC-0058-capability-command-bridge.md new file mode 100644 index 000000000..66cdb8c2c --- /dev/null +++ b/docs/rfcs/draft/RFC-0058-capability-command-bridge.md @@ -0,0 +1,712 @@ +--- +rfc_id: RFC-0058 +title: "Capability-Command Bridge: Connecting CommandResource Capabilities to Protocol Servers" +status: REVIEW +author: Sisyphus +reviewers: + - name: Oracle + status: completed + - name: Metis + status: completed + - name: Momus + status: completed +created: 2026-07-22 +last_updated: 2026-07-22 +decision_date: +related_rfcs: + - RFC-0016 (Unified Skill-to-Slash Command Architecture) + - RFC-0032 (ACP Slash Commands Protocol Compliance) + - RFC-0051 (Extension Source Architecture) +related_openspec: + - openspec/changes/capability-command-bridge/ +--- + +# RFC-0058: Capability-Command Bridge + +## Overview + +AgentPool's v1 capability architecture introduced `CommandResource` as a mixin protocol for capabilities to expose slash commands, and `ExtensionRegistry.get_command_resources()` to query them. However, protocol servers (ACP, OpenCode) bypass this infrastructure entirely — they read commands directly from `SkillsRegistry` and YAML manifest config. Additionally, `CommandResource` is discovery-only: `CommandEntry` carries metadata but no execution handler, so even if discovered, a custom capability's commands cannot be invoked. + +This RFC proposes a `CommandBridge` component that connects `ExtensionRegistry` command resources to protocol server command stores, providing both discovery and execution paths. The change is additive — existing skill commands and MCP prompt commands continue to work unchanged, while any capability implementing `CommandResource` gains automatic command registration across all command-capable protocol servers. + +## Table of Contents + +- [Background & Context](#background--context) +- [Problem Statement](#problem-statement) +- [Goals & Non-Goals](#goals--non-goals) +- [Evaluation Criteria](#evaluation-criteria) +- [Options Analysis](#options-analysis) +- [Recommendation](#recommendation) +- [Technical Design](#technical-design) +- [Security Considerations](#security-considerations) +- [Implementation Plan](#implementation-plan) +- [Open Questions](#open-questions) +- [Decision Record](#decision-record) +- [References](#references) + +--- + +## Background & Context + +### Current State + +AgentPool's command system has three layers that are partially connected: + +**Layer 1 — Capability Protocol (`src/agentpool/capabilities/resource_protocols.py`):** +- `CommandResource` is a `@runtime_checkable` Protocol with `list_commands() -> Sequence[CommandEntry]` and `get_command(name) -> CommandEntry | None` +- `CommandEntry` is a dataclass: `name`, `description`, `skill_uri`, `source` +- Two capabilities implement `CommandResource`: `SkillManagerCap` (maps skills to commands) and `McpServerCap` (maps MCP prompts to commands) + +**Layer 2 — Registry (`src/agentpool/capabilities/extension_registry.py`):** +- `ExtensionRegistry` manages capabilities at 4 scope levels: POOL → SESSION → AGENT → TURN +- `get_command_resources(scope)` returns all visible capabilities implementing `CommandResource` +- This method has **no callers outside of tests** + +**Layer 3 — Protocol Servers:** +- **ACP server** (`session_agent_mgmt.py`): `_register_skill_commands()` reads from `host_context.skills_registry.list_skills()` directly, creates `SkillCommand` objects, feeds them to `ACPSkillBridge`. `_register_manifest_commands()` reads YAML config. Neither calls `get_command_resources()`. +- **OpenCode server** (`skill_bridge.py`): `OpenCodeSkillBridge` wraps `SkillCommand` → `SlashedCommand` via `create_skill_command()`. `GET /command` endpoint lists from MCP prompts + `skill_bridge`. Does not call `get_command_resources()`. +- **AG-UI server**: Treats skills as tools (not slash commands) via `AGUISkillBridge`. Orthogonal to this RFC. +- **MCP, OpenAI API, A2A servers**: No slash command support. Out of scope. + +### Historical Context + +- **RFC-0016** established the skill-to-slash-command architecture, creating `SkillCommand`, `create_skill_command()`, and per-protocol bridges (`ACPSkillBridge`, `OpenCodeSkillBridge`). +- **RFC-0032** aligned ACP slash command advertisement with the ACP spec (moved from `initialize` to `session/update`). +- **M3 capability migration** replaced `ResourceProvider` hierarchy with `AbstractCapability` and mixin protocols (`CommandResource`, `SkillResource`, `McpResource`). `ExtensionRegistry` was introduced with `get_command_resources()`, but protocol servers were never rewired to use it. +- The gap was identified during investigation of why a custom capability implementing `CommandResource` has its commands invisible to all protocol servers. + +### Glossary + +| Term | Definition | +|------|------------| +| `CommandResource` | `@runtime_checkable` Protocol for capabilities that produce slash commands | +| `CommandEntry` | Dataclass carrying command metadata (`name`, `description`, `skill_uri`, `source`) | +| `ExtensionRegistry` | 4-level scope registry for capabilities (POOL → SESSION → AGENT → TURN) | +| `CommandBridge` | Proposed new class connecting `ExtensionRegistry` to protocol server `CommandStore`s | +| `CommandStore` | Per-session store of registered slash commands in protocol servers | +| `SlashedCommand` | Protocol-agnostic command abstraction in `src/agentpool/commands/base.py` | +| `SkillCommand` | `SlashedCommand` subclass wrapping a `Skill` (defined in `skills/command.py`) | +| `ACPSkillBridge` | ACP-specific bridge converting `SkillCommand` → ACP `AvailableCommand` | +| `OpenCodeSkillBridge` | OpenCode-specific bridge converting `SkillCommand` → `SlashedCommand` | +| `create_skill_command()` | Shared function in `skill_bridge.py` creating `SlashedCommand` from `SkillCommand` | +| `ChangeEvent` | Dataclass emitted by capabilities to signal resource changes (`kind: str`) | + +--- + +## Problem Statement + +### The Problem + +A custom capability implementing `CommandResource` has its commands invisible to all protocol servers. Two specific gaps cause this: + +1. **Discovery gap**: Protocol servers never call `ExtensionRegistry.get_command_resources()`. They read from `SkillsRegistry` and YAML manifest config only. +2. **Execution gap**: `CommandResource` is discovery-only. `CommandEntry` carries metadata but no execution handler. Even if discovered, there is no way to invoke a command back to the producing capability without protocol-specific wiring per capability type. + +### Evidence + +- `ExtensionRegistry.get_command_resources()` exists (line ~340 in `extension_registry.py`) but has zero callers in production code paths (only in tests). +- `_register_skill_commands()` in `session_agent_mgmt.py` calls `host_context.skills_registry.list_skills()` directly, bypassing the capability system. +- `OpenCodeSkillBridge` in `skill_bridge.py` wraps `SkillCommand` objects only — no mechanism for non-skill `CommandEntry` objects. +- `CommandEntry` in `resource_protocols.py` has fields `name`, `description`, `skill_uri`, `source` — no `handler` or callable field. + +### Impact of Inaction + +- **Cost**: Every new capability type that wants to expose slash commands requires manual wiring in each protocol server (ACP, OpenCode). This is O(servers × capability types) wiring effort. +- **Risk**: The `CommandResource` protocol and `ExtensionRegistry.get_command_resources()` are dead code — they exist but are never exercised in production, creating a false impression that the system supports capability-driven commands. +- **Opportunity**: Custom capabilities (e.g., a hypothetical `JiraCapability` exposing `/create-ticket`) cannot participate in the slash command system without bypassing the capability architecture entirely. + +--- + +## Goals & Non-Goals + +### Goals (In Scope) + +1. Enable any capability implementing `CommandResource` to have its commands automatically discovered and registered by all command-capable protocol servers (ACP, OpenCode) +2. Provide an execution path from command invocation back to the producing capability, without protocol-specific wiring per capability type +3. Maintain backward compatibility — existing skill commands and MCP prompt commands continue to work unchanged +4. Keep the change minimal — connect existing pipes, don't create new abstractions + +### Non-Goals (Out of Scope) + +1. Adding command support to protocols that don't have commands (MCP, OpenAI API, A2A) +2. AG-UI protocol integration — AG-UI treats skills as tools (not slash commands) via `AGUISkillBridge`; the `CommandBridge` pattern is for slash command protocols only +3. Modifying `AbstractCapability` or `AbstractToolset` upstream in pydantic-ai +4. Unifying all command types into a single class hierarchy — static YAML commands, skill commands, and capability commands coexist +5. Modifying `create_skill_command()` in `skill_bridge.py` — this function is shared between ACP and OpenCode servers and MUST NOT be modified + +### Success Criteria + +- [ ] A custom capability implementing `CommandResource` (not `SkillManagerCap`, not `McpServerCap`) has its commands visible in ACP `AvailableCommandsUpdate` +- [ ] A custom capability's commands can be invoked by an ACP client via `CommandBridge.execute()` +- [ ] A custom capability's commands appear in the OpenCode `/command` endpoint response +- [ ] Existing skill commands and MCP prompt commands continue to work without behavior change +- [ ] `ExtensionRegistry.get_command_resources()` is called in production code paths (not just tests) +- [ ] No breaking changes — `CommandEntry` consumers that ignore the new `handler` field are unaffected + +--- + +## Evaluation Criteria + +The following criteria are used to objectively evaluate each option: + +| Criterion | Weight | Description | Minimum Threshold | +|-----------|--------|-------------|-------------------| +| Discovery completeness | High | All `CommandResource` capabilities are discovered, not just skills | Must cover custom capabilities | +| Execution path | High | Commands can be invoked back to the producing capability | Must work without per-capability wiring | +| Backward compatibility | High | Existing skill/MCP prompt commands work unchanged | Zero behavior change for existing users | +| Implementation effort | Medium | LOC changed, files modified, new modules | <500 LOC total change | +| Protocol server coupling | Medium | Protocol servers should not duplicate discovery logic | Single integration point | +| Type safety | Medium | No `Any` types, no `@ts-ignore`-equivalent patterns | Full type hints, mypy --strict clean | +| Testability | Medium | New behavior is unit-testable in isolation | Unit tests for all new public methods | + +--- + +## Options Analysis + +### Option 1: `CommandBridge` with `handler` on `CommandEntry` + +**Description** + +Add an optional `handler: Callable[[str, AgentContext], Awaitable[str]] | None` field to `CommandEntry`. Create a new `CommandBridge` class that sits between `ExtensionRegistry` and protocol server `CommandStore`s. `CommandBridge` discovers commands by calling `ExtensionRegistry.get_command_resources(scope)` and aggregating `list_commands()` from all results. Execution routes through `CommandBridge.execute(name, input, ctx)`, which looks up the cached `CommandEntry` and invokes its `handler`. Protocol servers construct a per-session `CommandBridge` and use it for both discovery and execution. + +**Advantages** + +- Single integration point: protocol servers delegate to `CommandBridge` instead of each having their own discovery logic +- Per-command execution: `handler` on `CommandEntry` allows different commands from the same capability to have different execution paths +- `CommandResource` protocol stays pure: discovery (`list_commands`/`get_command`) is unchanged, execution is added via the `CommandEntry` data model +- Backward compatible: `handler` defaults to `None`; existing `CommandEntry` consumers ignore it +- `compare=False` on `handler` field preserves existing `CommandEntry` equality semantics +- Cached name→entry index provides O(1) execution lookup after initial discovery + +**Disadvantages** + +- `CommandEntry` gains a callable field, making it a hybrid data/behavior object (though `compare=False` mitigates equality issues) +- `handler` is not serializable — `CommandEntry` cannot be sent over the wire with `handler` attached (protocol servers must convert to their own serializable types before transmission) +- Protocol servers need to be rewired to construct `CommandBridge` and route execution through it +- Per-session `CommandBridge` lifecycle adds a small memory overhead (one `CommandBridge` instance per active session) + +**Evaluation Against Criteria** + +| Criterion | Rating | Notes | +|-----------|--------|-------| +| Discovery completeness | Excellent | Queries all `CommandResource` capabilities via `ExtensionRegistry` | +| Execution path | Excellent | `handler` on `CommandEntry` provides direct execution path | +| Backward compatibility | Excellent | Additive only — `handler` defaults to `None`, no existing behavior change | +| Implementation effort | Good | ~200-300 LOC new code, ~100 LOC refactored across 4-5 files | +| Protocol server coupling | Excellent | Single `CommandBridge` integration point per protocol server | +| Type safety | Good | `Callable` type with full signature, `AgentContext` typed, no `Any` | +| Testability | Excellent | `CommandBridge` is a standalone class with clear inputs/outputs | + +**Effort Estimate** + +- Complexity: Medium +- Resources: 1 engineer, 2-3 days +- Dependencies: None — all building blocks (`ExtensionRegistry`, `CommandResource`, `CommandEntry`) already exist + +**Risk Assessment** + +| Risk | Likelihood | Impact | Mitigation | +|------|------------|--------|------------| +| Handler closures from destroyed scopes | Low | Medium | `CommandBridge` is per-session, discarded on session close; next `discover_commands()` rebuilds index | +| Command name collisions across capabilities | Medium | Low | First-wins in scope-priority order (TURN → AGENT → SESSION → POOL), log warning | +| `handler` not serializable | Low | Low | `handler` stays in-process; protocol servers convert to serializable types before transmission | +| `McpServerCap` emits `"prompts_changed"` not `"commands_changed"` | Medium | Medium | `CommandBridge.watch_changes()` forwards all three event kinds: `"commands_changed"`, `"skills_changed"`, `"prompts_changed"` | + +--- + +### Option 2: Add `execute_command()` to `CommandResource` Protocol + +**Description** + +Instead of adding `handler` to `CommandEntry`, add a new `execute_command(name: str, input: str, ctx: AgentContext) -> str` method to the `CommandResource` protocol. Protocol servers call `get_command_resources(scope)`, then for each `CommandResource` capability, call `list_commands()` for discovery and `execute_command(name, input, ctx)` for execution. + +**Advantages** + +- `CommandEntry` remains a pure data class — no callable field +- Execution is per-capability, matching the discovery pattern (`list_commands` / `get_command` on the capability, `execute_command` also on the capability) +- No new class needed — protocol servers call `CommandResource` methods directly + +**Disadvantages** + +- Couples discovery and execution: all commands from a single capability share one `execute_command` method, even if different commands need different execution paths +- Forces all `CommandResource` implementations to add `execute_command`, even those that are discovery-only (e.g., a capability that lists commands for display but doesn't support invocation) +- Protocol servers still need to iterate all `CommandResource` capabilities and match commands to capabilities for execution — no O(1) lookup without building a separate index +- No central deduplication or collision resolution — each protocol server must implement its own scope-priority logic +- Still requires rewiring all protocol servers, but without a centralized component to handle the wiring + +**Evaluation Against Criteria** + +| Criterion | Rating | Notes | +|-----------|--------|-------| +| Discovery completeness | Good | Queries all `CommandResource` capabilities | +| Execution path | Fair | Per-capability, not per-command; forces all implementors to add `execute_command` | +| Backward compatibility | Fair | Adding method to Protocol is a breaking change for existing implementations | +| Implementation effort | Fair | No new class, but more per-server wiring code | +| Protocol server coupling | Fair | Each server duplicates discovery + execution routing logic | +| Type safety | Good | Full type hints possible | +| Testability | Fair | No standalone component to unit test; logic spread across servers | + +**Effort Estimate** + +- Complexity: Medium +- Resources: 1 engineer, 3-4 days (more per-server wiring) +- Dependencies: None + +**Risk Assessment** + +| Risk | Likelihood | Impact | Mitigation | +|------|------------|--------|------------| +| Breaking existing `CommandResource` implementations | High | Medium | All implementors must add `execute_command` | +| Discovery-only capabilities forced to implement execution | Medium | Low | Can raise `NotImplementedError`, but violates interface contract | +| No centralized collision resolution | Medium | Low | Each server must implement its own | + +--- + +### Option 3: Protocol Servers Call `get_command_resources()` Directly + +**Description** + +No new components. Each protocol server (ACP, OpenCode) is modified to call `ExtensionRegistry.get_command_resources(scope)`, iterate all `CommandResource` capabilities, call `list_commands()` on each, and register the resulting `CommandEntry` objects. For execution, each protocol server maintains a mapping from command name to producing capability and calls `get_command(name)` on the capability to retrieve the entry, then has protocol-specific execution logic. + +**Advantages** + +- No new classes or fields — minimal new abstractions +- Each protocol server has full control over its command registration flow +- `CommandEntry` remains unchanged + +**Disadvantages** + +- Duplicates discovery + execution routing logic across 3+ protocol servers +- No execution path: `CommandEntry` still has no `handler`, so execution requires protocol-specific wiring per capability type (skill → load skill, MCP prompt → call `get_prompt`, custom → ???) +- Does not solve the execution gap at all — only partially solves the discovery gap +- Each server must independently handle collision resolution, scope priority, and change watching +- Violates DRY principle — same logic repeated in each server + +**Evaluation Against Criteria** + +| Criterion | Rating | Notes | +|-----------|--------|-------| +| Discovery completeness | Good | Queries all `CommandResource` capabilities | +| Execution path | Poor | No execution path added; requires per-capability-type wiring per server | +| Backward compatibility | Good | No changes to `CommandEntry` or `CommandResource` | +| Implementation effort | Fair | No new class, but more duplicated code per server | +| Protocol server coupling | Poor | Each server duplicates all logic | +| Type safety | Good | No new types needed | +| Testability | Poor | Logic spread across servers, no centralized component | + +**Effort Estimate** + +- Complexity: Medium-High (duplication across servers) +- Resources: 1 engineer, 4-5 days +- Dependencies: None + +**Risk Assessment** + +| Risk | Likelihood | Impact | Mitigation | +|------|------------|--------|------------| +| Execution gap not solved | High | High | Would require a follow-up RFC for execution | +| Logic duplication across servers | High | Medium | Extracted helpers, but still duplicated | +| Inconsistent behavior across protocols | Medium | Medium | Each server implements its own collision resolution | + +--- + +### Options Comparison Summary + +| Criterion | Option 1: CommandBridge + handler | Option 2: execute_command on Protocol | Option 3: Direct get_command_resources() | +|-----------|-----------------------------------|---------------------------------------|------------------------------------------| +| Discovery completeness | Excellent | Good | Good | +| Execution path | Excellent | Fair | Poor | +| Backward compatibility | Excellent | Fair | Good | +| Implementation effort | Good (2-3 days) | Fair (3-4 days) | Fair (4-5 days) | +| Protocol server coupling | Excellent | Fair | Poor | +| Type safety | Good | Good | Good | +| Testability | Excellent | Fair | Poor | +| **Overall** | **Recommended** | **Not recommended** | **Not recommended** | + +--- + +## Recommendation + +### Recommended Option + +**Option 1: `CommandBridge` with `handler` on `CommandEntry`** + +### Justification + +Option 1 scores highest across all evaluation criteria. It is the only option that fully solves both the discovery gap and the execution gap while maintaining backward compatibility. The `CommandBridge` provides a single, testable integration point that protocol servers delegate to, eliminating the duplicated discovery logic that Options 2 and 3 would perpetuate. + +The `handler` field on `CommandEntry` keeps execution per-command rather than per-capability, allowing a single capability to produce commands with different execution paths (e.g., `SkillManagerCap` produces both local skill commands and pass-through MCP prompt commands with different handlers). This is not possible with Option 2's per-capability `execute_command` method. + +Option 3 is ruled out because it does not solve the execution gap at all — it only partially addresses discovery and would require a follow-up RFC for execution, adding unnecessary latency. + +### Accepted Trade-offs + +1. `CommandEntry` becomes a hybrid data/behavior object: Acceptable because `compare=False` on `handler` preserves equality semantics, and `handler` is never serialized (protocol servers convert to their own types before transmission). +2. Per-session `CommandBridge` memory overhead: Acceptable because one additional lightweight object per active session is negligible. +3. Protocol servers must be rewired: Acceptable because the rewiring is additive (existing skill command paths remain as fallback), and the `CommandBridge` API is simpler than the current direct-access pattern. + +### Conditions + +- `create_skill_command()` in `skill_bridge.py` MUST NOT be modified — a separate `entry_to_slashed_command()` converter handles `CommandEntry`-based commands +- `CommandBridge` is per-session (Scope(SESSION)), not pool-level — this ensures handler closures reference valid session state +- AG-UI is explicitly excluded — its tool-based approach is orthogonal to slash commands + +--- + +## Technical Design + +### Architecture Overview + +``` +┌─────────────────────────────────────────────────────────────────────┐ +│ ExtensionRegistry │ +│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ +│ │ SkillMgr │ │ McpServer│ │ Custom │ │ ... │ │ +│ │ Cap │ │ Cap │ │ Cap │ │ │ │ +│ │ (CmdRes) │ │ (CmdRes) │ │ (CmdRes) │ │ │ │ +│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └──────────┘ │ +│ │ │ │ │ +│ └──────────────┴──────────────┘ │ +│ │ get_command_resources(scope) │ +└──────────────────────┼──────────────────────────────────────────────┘ + │ + ┌────────▼────────┐ + │ CommandBridge │ (per-session, Scope(SESSION)) + │ │ + │ discover_cmds()│──► list[CommandEntry] (with handler) + │ execute() │──► str (calls entry.handler) + │ watch_changes()│──► AsyncIterator[ChangeEvent] + │ entry_to_slash │──► SlashedCommand | None + └────────┬────────┘ + │ + ┌────────────┼────────────┐ + │ │ │ + ┌───────▼──────┐ ┌──▼─────────┐ ┌▼──────────────┐ + │ ACP Server │ │ OpenCode │ │ Other servers │ + │ │ │ Server │ │ (future) │ + │ ACPSession │ │ OpenCode │ │ │ + │ .bridge │ │ CmdBridge │ │ │ + └──────────────┘ └────────────┘ └───────────────┘ +``` + +### Key Components + +#### `CommandEntry` (modified) + +```python +@dataclass(frozen=True) +class CommandEntry: + name: str + description: str + skill_uri: str | None + source: str + handler: Callable[[str, AgentContext], Awaitable[str]] | None = field( + default=None, compare=False + ) +``` + +- `handler`: Optional async callable taking `(input: str, ctx: AgentContext) -> str` +- `compare=False`: Excludes `handler` from equality/hash comparisons, preserving existing `CommandEntry` comparison semantics +- `AgentContext`: The frozen dataclass defined in `src/agentpool/capabilities/agent_context.py` + +#### `CommandBridge` (new) + +```python +class CommandBridge: + def __init__(self, extension_registry: ExtensionRegistry, scope: Scope) -> None: ... + + def discover_commands(self) -> list[CommandEntry]: + """Query all CommandResource capabilities, aggregate commands, + de-duplicate by name (TURN → AGENT → SESSION → POOL priority), + build and cache a dict[str, CommandEntry] index for O(1) lookup.""" + + async def execute(self, name: str, input: str, ctx: AgentContext) -> str: + """Look up CommandEntry from cached index, invoke entry.handler. + Raises CommandNotFoundError if not found. + Raises CommandNotExecutableError if handler is None. + Exceptions from handler() propagate without wrapping.""" + + async def watch_changes(self) -> AsyncIterator[ChangeEvent]: + """Wrap extension_registry.merge_change_streams(), filtering for + 'commands_changed', 'skills_changed', 'prompts_changed' events. + Returns empty iterator when merge_change_streams returns None.""" + + @staticmethod + def entry_to_slashed_command(entry: CommandEntry) -> SlashedCommand | None: + """Convert CommandEntry to SlashedCommand. Returns None for + display-only entries (handler is None). MUST NOT modify + create_skill_command() in skill_bridge.py.""" +``` + +#### `CommandNotExecutableError`, `CommandNotFoundError` (new) + +```python +class CommandNotFoundError(Exception): + """Raised when CommandBridge.execute() cannot find a command by name.""" + +class CommandNotExecutableError(Exception): + """Raised when a CommandEntry has no handler (display-only command).""" +``` + +#### Protocol Server Integration + +**ACP Server:** +- `ACPSession` constructs a per-session `CommandBridge` from the session's `ExtensionRegistry` with `Scope(level=ScopeLevel.SESSION, session_id=...)` +- `_register_skill_commands()` uses `CommandBridge.discover_commands()` instead of reading `skills_registry.list_skills()` directly +- `_watch_skill_changes()` consumes `CommandBridge.watch_changes()` for all three event kinds +- Command execution routes through `CommandBridge.execute()` with fallback to existing manifest command path on `CommandNotFoundError` +- `ACPSkillBridge` retains `SkillCommand` → `SlashedCommand` conversion via unchanged `create_skill_command()`, delegates discovery to `CommandBridge` + +**OpenCode Server:** +- New `OpenCodeCommandBridge` class wraps `CommandBridge` for OpenCode-specific command conversion (NOT modifying `OpenCodeSkillBridge`) +- `GET /command` endpoint includes commands from `CommandBridge.discover_commands()` alongside existing commands +- Execution routes through `CommandBridge.execute()` with fallback to existing skill command path on `CommandNotFoundError` +- Change watcher consumes `CommandBridge.watch_changes()` for command list rebuilds + +### Data Model + +```python +# CommandEntry with handler (modified) +@dataclass(frozen=True) +class CommandEntry: + name: str + description: str + skill_uri: str | None + source: str + handler: Callable[[str, AgentContext], Awaitable[str]] | None = field( + default=None, compare=False + ) + +# New exceptions +class CommandNotFoundError(Exception): ... +class CommandNotExecutableError(Exception): ... + +# ChangeEvent kind additions +# "commands_changed" added to ChangeKind Literal (kind field is already str) +``` + +### Change Event Flow + +``` +SkillManagerCap ──► ChangeEvent(kind="skills_changed") +McpServerCap ──► ChangeEvent(kind="prompts_changed") +CustomCapability ──► ChangeEvent(kind="commands_changed") + │ + ▼ + ExtensionRegistry.merge_change_streams(scope) + │ + ▼ + CommandBridge.watch_changes() + (filters for 3 event kinds) + │ + ┌─────────┴─────────┐ + ▼ ▼ + ACP server rebuilds OpenCode server rebuilds + command list /command response +``` + +### Collision Resolution + +When multiple capabilities produce commands with the same name, `CommandBridge.discover_commands()` resolves duplicates by keeping the first occurrence in scope-priority order: + +``` +TURN (most specific) → AGENT → SESSION → POOL (least specific) +``` + +A warning is logged for each duplicate encountered. This ensures that session-scoped capabilities override pool-scoped ones, and turn-scoped capabilities override all others. + +--- + +## Security Considerations + +### Threat Analysis + +| Threat | Impact | Likelihood | Mitigation | +|--------|--------|------------|------------| +| Malicious capability registers harmful slash commands | Medium | Low | Capabilities are registered via `ExtensionRegistry` which is controlled by pool configuration; untrusted capabilities cannot self-register | +| Handler callable executes with elevated privileges | Medium | Low | `handler` receives `AgentContext` which carries the same permissions as the agent; no privilege escalation | +| Command name spoofing (custom capability overrides existing command) | Low | Medium | Scope-priority resolution ensures most specific scope wins; warning logged on collision | +| Stale handler references after session destruction | Low | Low | `CommandBridge` is per-session and discarded on session close; next `discover_commands()` rebuilds index | + +### Security Measures + +- [x] `CommandBridge` is constructed with a specific `Scope` — capabilities outside the scope are not visible +- [x] `handler` receives `AgentContext` — no direct access to `AgentPool` or `ExtensionRegistry` +- [x] Collision resolution logs warnings — operators can detect name spoofing attempts +- [x] Protocol servers convert `CommandEntry` to their own serializable types before transmission — `handler` never leaves the process + +### Compliance + +No regulatory or compliance requirements are affected. All changes are internal to the agent orchestration framework. + +--- + +## Implementation Plan + +### Phases + +#### Phase 1: Core Data Model Changes + +- **Scope**: Add `handler` field to `CommandEntry`, new exceptions, `ChangeKind` update +- **Deliverables**: Modified `resource_protocols.py`, new `command_bridge.py` (exceptions only), updated `change_event.py` +- **Dependencies**: None + +#### Phase 2: CommandBridge Implementation + +- **Scope**: Full `CommandBridge` class with `discover_commands()`, `execute()`, `watch_changes()`, `entry_to_slashed_command()` +- **Deliverables**: Complete `command_bridge.py` (~100-150 LOC), unit tests +- **Dependencies**: Phase 1 + +#### Phase 3: Update Existing CommandResource Implementations + +- **Scope**: `SkillManagerCap.list_commands()` and `McpServerCap.list_commands()` populate `handler` +- **Deliverables**: Modified `skill_manager_cap.py`, `mcp_server_cap.py`, integration tests +- **Dependencies**: Phase 1 + +#### Phase 4: ACP Server Integration + +- **Scope**: Rewire ACP server to use `CommandBridge` for discovery, execution, and change watching +- **Deliverables**: Modified `session_agent_mgmt.py`, `ACPSkillBridge` updated, integration tests +- **Dependencies**: Phases 2, 3 + +#### Phase 5: OpenCode Server Integration + +- **Scope**: Create `OpenCodeCommandBridge`, update `GET /command` endpoint, execution routing, change watcher +- **Deliverables**: New `OpenCodeCommandBridge` class, modified `agent_routes.py`, `server.py`, integration tests +- **Dependencies**: Phases 2, 3 + +#### Phase 6: Documentation and Validation + +- **Scope**: Docstrings, example capability, `AGENTS.md` update, full test suite, lint, type check +- **Deliverables**: Documentation, example, validation results +- **Dependencies**: Phases 4, 5 + +### Milestones + +| Milestone | Description | Target | Status | +|-----------|-------------|--------|--------| +| M1: Core types ready | `CommandEntry.handler`, exceptions, `ChangeKind` | Day 1 | Not Started | +| M2: CommandBridge complete | Full class with unit tests | Day 2 | Not Started | +| M3: Existing capabilities updated | `SkillManagerCap`, `McpServerCap` | Day 2 | Not Started | +| M4: ACP server integrated | End-to-end ACP tests pass | Day 3 | Not Started | +| M5: OpenCode server integrated | End-to-end OpenCode tests pass | Day 3 | Not Started | +| M6: Validation complete | Full suite + lint + type check | Day 3 | Not Started | + +### Rollback Strategy + +All changes are additive. Rollback procedure: + +1. Revert protocol server wiring (ACP: `session_agent_mgmt.py`, OpenCode: `agent_routes.py`, `server.py`) to direct `SkillsRegistry` access +2. Remove `CommandBridge` class and `command_bridge.py` +3. Remove `handler` field from `CommandEntry` (or leave as unused `None` default — no breaking change) +4. Revert `SkillManagerCap` and `McpServerCap` `list_commands()` to not populate `handler` + +No data migration or state cleanup is required — `CommandBridge` is stateless (rebuilt on each `discover_commands()` call). + +--- + +## Open Questions + +1. **Should `ExtensionRegistry.get_command_resources(scope)` guarantee deterministic ordering by scope specificity?** + + - Context: `CommandBridge.discover_commands()` relies on scope-priority ordering for collision resolution. If `get_command_resources()` returns capabilities in non-deterministic order, the collision resolution behavior is undefined. + - Owner: Implementation engineer (verify during Phase 2, Task 7.5) + - Status: Open — needs code verification of current `ExtensionRegistry` ordering behavior + +2. **Should `entry_to_slashed_command()` be a static method on `CommandBridge` or a standalone function?** + + - Context: It's currently specified as a static method, but a standalone function in `command_bridge.py` would be equally valid and easier to mock in tests. + - Owner: Implementation engineer + - Status: Open — defer to implementation preference + +3. **Should the `"commands_changed"` event be emitted by `SkillManagerCap` in addition to `"skills_changed"`?** + + - Context: `SkillManagerCap` currently emits `"skills_changed"`. Adding `"commands_changed"` would be redundant since `CommandBridge.watch_changes()` already forwards `"skills_changed"`. But it would make the event model more consistent. + - Owner: Implementation engineer + - Status: Open — leaning towards NOT emitting (avoid redundancy, `CommandBridge` handles the mapping) + +--- + +## Decision Record + +> Complete this section after RFC review is concluded. + +### Decision + +**Status**: PENDING REVIEW + +**Date**: TBD + +**Approvers**: +- [Reviewer 1] +- [Reviewer 2] + +### Decision Summary + +[To be completed after review] + +### Key Discussion Points + +1. The `handler` field on `CommandEntry` vs. `execute_command` on `CommandResource` — the per-command execution model was preferred over per-capability +2. Creating `OpenCodeCommandBridge` vs. generalizing `OpenCodeSkillBridge` — a new class was chosen to avoid breaking the shared `create_skill_command()` function +3. AG-UI exclusion — AG-UI's tool-based approach is orthogonal to slash commands and does not benefit from `CommandBridge` + +### Conditions of Approval + +- All existing tests pass without modification +- `create_skill_command()` is not modified +- A custom capability end-to-end test demonstrates the full discovery → execution flow + +### Dissenting Opinions + +None recorded. Oracle, Metis, and Momus reviews converged on the same recommendation after the review-revise loop (3 rounds, 46 tasks refined). + +--- + +## References + +### Related Documents + +- [OpenSpec Change: capability-command-bridge](../../../openspec/changes/capability-command-bridge/) + - `proposal.md` — What and why + - `design.md` — 5 design decisions, 6 risk entries + - `tasks.md` — 46 tasks across 7 sections + - `specs/` — 7 spec files (6 modified + 1 new `opencode-server`) +- [RFC-0016: Unified Skill-to-Slash Command Architecture](./RFC-0016-skill-slash-commands.md) +- [RFC-0032: ACP Slash Commands Protocol Compliance](./RFC-0032-acp-slash-commands-session-update.md) +- [RFC-0051: Extension Source Architecture](./RFC-0051-extension-source-architecture.md) + +### External Resources + +- [Agent Skills Spec](https://github.com/agentskills/agentskills) +- [Agent Client Protocol (ACP) Specification](https://agentclientprotocol.com/) + +### Appendix + +#### Review History + +This RFC was developed through an OpenSpec change that underwent 3 rounds of Oracle + Metis + Momus review: + +| Round | Reviewers | Findings | Outcome | +|-------|-----------|----------|---------| +| 1 | Oracle, Metis, Momus | 35 total (7 CRITICAL, 13 MAJOR, 15 MINOR) | NEEDS REVISION | +| 2 | — (fixes applied) | 21 fixes across 10 files | Fixes applied | +| 3 | Oracle (verification) | 4 remaining issues | NOT VERIFIED → fixes applied | +| 4 | Oracle (re-verification) | 0 issues | VERIFIED | +| 5 | Oracle (final check) | 0 issues | VERIFIED (3rd pass) | + +Final spec state: 7 spec files, 5 design decisions, 6 risk entries, 46 tasks across 7 sections. + +#### File Impact Summary + +| File | Change Type | Est. LOC | +|------|-------------|----------| +| `src/agentpool/capabilities/resource_protocols.py` | Modified (add `handler` to `CommandEntry`) | +15 | +| `src/agentpool/capabilities/command_bridge.py` | New (full `CommandBridge` class) | +150 | +| `src/agentpool/capabilities/change_event.py` | Modified (add `ChangeKind` literal) | +3 | +| `src/agentpool/capabilities/skill_manager_cap.py` | Modified (populate `handler` in `list_commands()`) | +20 | +| `src/agentpool/capabilities/mcp_server_cap.py` | Modified (populate `handler` in `list_commands()`) | +15 | +| `src/agentpool_server/acp_server/session_agent_mgmt.py` | Modified (use `CommandBridge`) | +30, -15 | +| `src/agentpool_server/opencode_server/` (new `OpenCodeCommandBridge`) | New + Modified | +40 | +| Tests (new + modified) | New | +200 | +| **Total** | | **~460 LOC** | diff --git a/examples/custom_command_capability.py b/examples/custom_command_capability.py new file mode 100644 index 000000000..c128f2326 --- /dev/null +++ b/examples/custom_command_capability.py @@ -0,0 +1,105 @@ +"""Example: Custom capability that registers slash commands via CommandResource. + +This demonstrates the end-to-end flow: +1. A custom capability implements ``CommandResource`` to publish commands. +2. ``CommandBridge.discover_commands()`` finds them via ``ExtensionRegistry``. +3. Protocol servers (ACP, OpenCode) expose them as slash commands to clients. +4. When invoked, ``CommandBridge.execute()`` calls the ``CommandEntry.handler``. + +To use: register this capability in your AgentPool config or programmatically:: + + from examples.custom_command_capability import WeatherCommandCapability + + pool.extension_registry.register( + WeatherCommandCapability(), + scope=Scope(level=ScopeLevel.POOL), + ) + +The ``/weather`` command will then appear in ACP and OpenCode clients. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from agentpool.capabilities.resource_protocols import ( + CommandEntry, + CommandResource, +) + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Sequence + + from agentpool.capabilities.agent_context import AgentContext + from agentpool.capabilities.change_event import ChangeEvent + + +@dataclass +class WeatherCommandCapability(CommandResource): + """A minimal custom capability that provides a ``/weather`` slash command. + + This capability does NOT provide tools or instructions — it only + implements ``CommandResource`` to publish a command with a handler. + ``CommandBridge`` discovers it via ``ExtensionRegistry.get_command_resources()`` + and protocol servers expose it to clients. + """ + + name: str = "weather-cmd" + _commands: list[CommandEntry] | None = None + + def __post_init__(self) -> None: + """Build the command entries with handlers.""" + self._commands = [ + CommandEntry( + name="weather", + description="Get the current weather for a city", + skill_uri="weather://command", + source="custom", + handler=self._weather_handler, + ), + ] + + @staticmethod + async def _weather_handler( + input_text: str, + ctx: AgentContext, + ) -> str: + """Handle the ``/weather`` command. + + Args: + input_text: The user's input after the command name (e.g., "San Francisco"). + ctx: The agent context (provides access to host, registry, etc.). + + Returns: + A weather report string. + """ + city = input_text.strip() or "Unknown" + # In a real implementation, you would call a weather API here. + # The handler has access to ``ctx.host`` for MCP tools, storage, etc. + return f"🌤️ Weather for {city}: Sunny, 72°F (22°C)" + + # --- CommandResource protocol --- + + async def list_commands(self) -> Sequence[CommandEntry]: + """Return all commands provided by this capability.""" + return self._commands or [] + + async def get_command(self, name: str) -> CommandEntry | None: + """Look up a command by name.""" + for entry in self._commands or []: + if entry.name == name: + return entry + return None + + # --- Optional: ChangeObservable --- + + def on_change(self) -> AsyncIterator[ChangeEvent] | None: + """Emit change events when the command list changes. + + For a static capability, return ``None`` (no changes expected). + For a dynamic capability, yield ``ChangeEvent(kind="commands_changed")`` + when the command list is updated. + """ + return None diff --git a/src/agentpool/capabilities/change_event.py b/src/agentpool/capabilities/change_event.py index 787cbf7ab..766cd1af5 100644 --- a/src/agentpool/capabilities/change_event.py +++ b/src/agentpool/capabilities/change_event.py @@ -18,6 +18,7 @@ "prompts_changed", "resources_changed", "skills_changed", + "commands_changed", ] """Discriminator for which resource type changed in a capability.""" diff --git a/src/agentpool/capabilities/command_bridge.py b/src/agentpool/capabilities/command_bridge.py new file mode 100644 index 000000000..b7ae054d1 --- /dev/null +++ b/src/agentpool/capabilities/command_bridge.py @@ -0,0 +1,265 @@ +"""CommandBridge — unified command discovery, execution, and change watching. + +The :class:`CommandBridge` connects the :class:`ExtensionRegistry` (which tracks +capabilities at four scope levels) to protocol servers (ACP, OpenCode) that need +to expose slash commands to clients. + +Architecture +------------ + +1. **Discovery** — ``discover_commands(scope)`` queries + :meth:`ExtensionRegistry.get_command_resources` and aggregates + :meth:`CommandResource.list_commands` from all visible capabilities. + Commands are de-duplicated by name with most-specific-scope-first priority + (TURN → AGENT → SESSION → POOL). + +2. **Execution** — ``execute(name, input, ctx)`` looks up the command in a + cached name→entry index (built during discovery) and invokes + :attr:`CommandEntry.handler`. + +3. **Change watching** — ``watch_changes(scope)`` wraps + :meth:`ExtensionRegistry.merge_change_streams` and filters for + ``"commands_changed"``, ``"skills_changed"``, and ``"prompts_changed"`` + events. + +4. **Per-session lifecycle** — Each protocol session constructs its own + ``CommandBridge`` with ``Scope(level=ScopeLevel.SESSION, session_id=...)``. + The bridge references the registry but does not own its lifecycle. + +5. **Protocol conversion** — ``entry_to_slashed_command(entry)`` converts a + :class:`CommandEntry` to a :class:`slashed.Command` for protocol bridges + that use the ``slashed`` command store. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import logfire + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + from typing import Any + + from slashed import Command as SlashedCommand + + from agentpool.capabilities.agent_context import AgentContext + from agentpool.capabilities.change_event import ChangeEvent + from agentpool.capabilities.extension_registry import ExtensionRegistry, Scope + from agentpool.capabilities.resource_protocols import CommandEntry + + +class CommandNotFoundError(Exception): + """Raised when a command name is not found in the CommandBridge index.""" + + def __init__(self, name: str) -> None: + super().__init__(f"Command not found: {name!r}") + self.name = name + + +class CommandNotExecutableError(Exception): + """Raised when a command entry has no handler (display-only).""" + + def __init__(self, name: str) -> None: + super().__init__(f"Command {name!r} is not executable (no handler)") + self.name = name + + +class CommandBridge: + """Bridge between ExtensionRegistry and protocol servers for slash commands. + + Constructed per-session with a :class:`Scope` at SESSION level. Discovers + commands from all visible :class:`CommandResource` capabilities, caches + them in a name→entry index for O(1) execution lookup, and watches for + changes via :meth:`ExtensionRegistry.merge_change_streams`. + + Attributes: + _registry: The ExtensionRegistry to query. + _scope: The scope to query at (typically SESSION level). + _commands: Cached list of discovered CommandEntry objects. + _index: Cached dict[str, CommandEntry] for O(1) name lookup. + """ + + def __init__( + self, + registry: ExtensionRegistry, + scope: Scope, + ) -> None: + """Initialize the CommandBridge. + + Args: + registry: The ExtensionRegistry to query for CommandResource + capabilities. + scope: The scope at which to discover commands. Typically + ``Scope(level=ScopeLevel.SESSION, session_id=...)``. + """ + self._registry = registry + self._scope = scope + self._commands: list[CommandEntry] = [] + self._index: dict[str, CommandEntry] = {} + + @logfire.instrument("command_bridge.discover_commands") + async def discover_commands(self) -> list[CommandEntry]: + """Discover all commands visible at the bridge's scope. + + Queries :meth:`ExtensionRegistry.get_command_resources` and aggregates + ``list_commands()`` from all results. De-duplicates by name with + most-specific-scope-first priority (TURN → AGENT → SESSION → POOL). + Builds and caches a ``dict[str, CommandEntry]`` index for O(1) lookup. + + Returns: + List of unique CommandEntry objects (de-duplicated by name). + """ + resources = self._registry.get_command_resources(self._scope) + + # Aggregate commands from all CommandResource capabilities. + # get_command_resources returns in scope-specificity order + # (TURN → AGENT → SESSION → POOL), so the first occurrence of each + # name wins during de-duplication. + seen: set[str] = set() + commands: list[CommandEntry] = [] + index: dict[str, CommandEntry] = {} + + for cap in resources: + try: + cap_commands = await cap.list_commands() + except Exception: + logger = _get_logger() + logger.exception( + "Failed to list commands from capability", + capability=type(cap).__name__, + ) + continue + + for entry in cap_commands: + if entry.name in seen: + logger = _get_logger() + logger.debug( + "Duplicate command name, keeping first (more specific scope)", + name=entry.name, + ) + continue + seen.add(entry.name) + commands.append(entry) + index[entry.name] = entry + + self._commands = commands + self._index = index + return commands + + @logfire.instrument("command_bridge.execute {name}") + async def execute( + self, + name: str, + input: str, # noqa: A002 + ctx: AgentContext, + ) -> str: + """Execute a command by name via its handler. + + Looks up the :class:`CommandEntry` from the cached name→entry index. + Invokes ``entry.handler(input, ctx)`` if the handler is non-None. + + Args: + name: The command name to execute. + input: The raw input text (arguments) for the command. + ctx: The agent context for this execution. + + Returns: + The string result from the command handler. + + Raises: + CommandNotFoundError: If no command with ``name`` exists in the + cached index. + CommandNotExecutableError: If the command entry has no handler + (``handler is None``). + """ + entry = self._index.get(name) + if entry is None: + raise CommandNotFoundError(name) + if entry.handler is None: + raise CommandNotExecutableError(name) + # Exceptions from handler() propagate without wrapping. + return await entry.handler(input, ctx) + + async def watch_changes(self) -> AsyncIterator[ChangeEvent]: + """Watch for command list changes via ExtensionRegistry. + + Wraps :meth:`ExtensionRegistry.merge_change_streams` and filters for + ``"commands_changed"``, ``"skills_changed"``, and + ``"prompts_changed"`` events. Other event kinds (e.g., + ``"tools_changed"``, ``"resources_changed"``) are filtered out. + + Returns: + An async iterator yielding filtered ChangeEvent objects. If + ``merge_change_streams`` returns ``None``, returns an empty async + iterator. + + Yields: + ChangeEvent: A change event with a relevant kind. + """ + stream = self._registry.merge_change_streams(self._scope) + if stream is None: + return + async for event in stream: + if event.kind in ("commands_changed", "skills_changed", "prompts_changed"): + yield event + + @staticmethod + def entry_to_slashed_command( + entry: CommandEntry, + bridge: CommandBridge, + ) -> SlashedCommand | None: + """Convert a CommandEntry to a slashed Command. + + Creates a :class:`slashed.Command` whose executor calls + :meth:`CommandBridge.execute` with the entry's name. Returns ``None`` + for display-only entries (``handler is None``). + + This method MUST NOT modify ``create_skill_command()`` in + ``skill_bridge.py`` — it is a separate conversion path for + ``CommandEntry``-based commands. + + Args: + entry: The CommandEntry to convert. + bridge: The CommandBridge to use for execution. + + Returns: + A SlashedCommand if the entry has a handler, ``None`` otherwise. + """ + if entry.handler is None: + return None + + from slashed import Command as SlashedCommand + + async def execute_entry( + ctx: Any, + args: list[str], + kwargs: dict[str, str], + ) -> None: + """Execute the command entry via CommandBridge.""" + input_text = " ".join(args) + # Extract AgentContext from the command context's data field. + agent_ctx: AgentContext | None = None + if hasattr(ctx, "data") and ctx.data is not None: + agent_ctx = ctx.data + if agent_ctx is None: + msg = "No AgentContext available in command context" + raise RuntimeError(msg) + result = await bridge.execute(entry.name, input_text, agent_ctx) + if hasattr(ctx, "print"): + await ctx.print(result) + + return SlashedCommand.from_raw( + execute_entry, + name=entry.name, + description=entry.description, + category="capability", + ) + + +def _get_logger() -> Any: + """Get the module logger (deferred to avoid import-time side effects).""" + from agentpool.log import get_logger + + return get_logger(__name__) diff --git a/src/agentpool/capabilities/extension_registry.py b/src/agentpool/capabilities/extension_registry.py index ee6ff0cfb..2c90285b4 100644 --- a/src/agentpool/capabilities/extension_registry.py +++ b/src/agentpool/capabilities/extension_registry.py @@ -365,17 +365,52 @@ def get_command_resources( ) -> list[CommandResource]: """Get visible capabilities implementing ``CommandResource``. + Returns capabilities in scope-specificity order (most specific first: + turn-level → agent-level → session-level → pool-level). Within each + level, capabilities are returned in registration order. This ordering + enables ``CommandBridge`` to de-duplicate by name with most-specific + scope winning. + Args: scope: The scope to query. Returns: - List of capabilities implementing ``CommandResource``. + List of capabilities implementing ``CommandResource`` in + scope-specificity order (TURN → AGENT → SESSION → POOL). """ from agentpool.capabilities.resource_protocols import CommandResource - return [ - cap for cap in self.get_visible_capabilities(scope) if isinstance(cap, CommandResource) - ] + result: list[CommandResource] = [] + + # Turn level (most specific) + if scope.level.value >= ScopeLevel.TURN.value: + session_map = self._turn.get(scope.session_id, {}) + agent_map = session_map.get(scope.agent_name, {}) + result.extend( + cap for cap in agent_map.get(scope.turn_id, []) if isinstance(cap, CommandResource) + ) + + # Agent level + if scope.level.value >= ScopeLevel.AGENT.value: + agent_map = self._agent.get(scope.session_id, {}) + result.extend( + cap + for cap in agent_map.get(scope.agent_name, []) + if isinstance(cap, CommandResource) + ) + + # Session level + if scope.level.value >= ScopeLevel.SESSION.value: + result.extend( + cap + for cap in self._session.get(scope.session_id, []) + if isinstance(cap, CommandResource) + ) + + # Pool level (least specific) + result.extend(cap for cap in self._pool if isinstance(cap, CommandResource)) + + return result def get_observable_capabilities( self, diff --git a/src/agentpool/capabilities/mcp_server_cap.py b/src/agentpool/capabilities/mcp_server_cap.py index 38b97b571..b485da278 100644 --- a/src/agentpool/capabilities/mcp_server_cap.py +++ b/src/agentpool/capabilities/mcp_server_cap.py @@ -41,6 +41,7 @@ from pydantic_ai.toolsets import AbstractToolset + from agentpool.capabilities.agent_context import AgentContext from agentpool.mcp_server.client import MCPClient from agentpool.mcp_server.session_pool import SessionConnectionPool from agentpool_config.mcp_server import MCPServerConfig @@ -488,7 +489,8 @@ async def list_commands(self) -> Sequence[CommandEntry]: """List MCP prompts as commands. Maps MCP prompts to ``CommandEntry`` descriptors. Each prompt - becomes a command with ``source="remote"``. + becomes a command with ``source="remote"`` and a ``handler`` + that calls ``get_prompt`` on the MCP server. Returns: Sequence of ``CommandEntry`` descriptors. @@ -499,15 +501,45 @@ async def list_commands(self) -> Sequence[CommandEntry]: except Exception: # noqa: BLE001 logger.warning("Failed to list prompts for commands", exc_info=True) return [] - return [ - CommandEntry( - name=p.name, - description=p.description or "", - skill_uri=f"skill://{self._name}/{p.name}", - source="remote", + entries: list[CommandEntry] = [] + for p in prompts: + prompt_name = p.name + + async def _prompt_handler( + user_input: str, + ctx: AgentContext, + _name: str = prompt_name, + ) -> str: + """Call the MCP prompt and return the result as a string.""" + del ctx # MCP prompt execution does not need AgentContext. + mcp_client = await self._ensure_client() + arguments: dict[str, str] | None = None + if user_input: + parts = user_input.split() + if parts: + arguments = {str(i): v for i, v in enumerate(parts)} + result = await mcp_client.get_prompt(_name, arguments) + # Extract text from GetPromptResult messages. + texts: list[str] = [] + for msg in result.messages: + content = msg.content + text: str | None = getattr(content, "text", None) + if text is not None: + texts.append(text) + else: + texts.append(str(content)) + return "\n".join(texts) if texts else "" + + entries.append( + CommandEntry( + name=p.name, + description=p.description or "", + skill_uri=f"skill://{self._name}/{p.name}", + source="remote", + handler=_prompt_handler, + ) ) - for p in prompts - ] + return entries async def get_command(self, name: str) -> CommandEntry | None: """Get a specific command by name. @@ -527,11 +559,39 @@ async def get_command(self, name: str) -> CommandEntry | None: return None for p in prompts: if p.name == name: + prompt_name = p.name + + async def _prompt_handler( + user_input: str, + ctx: AgentContext, + _pname: str = prompt_name, + ) -> str: + """Call the MCP prompt and return the result as a string.""" + del ctx # MCP prompt execution does not need AgentContext. + mcp_client = await self._ensure_client() + arguments: dict[str, str] | None = None + if user_input: + parts = user_input.split() + if parts: + arguments = {str(i): v for i, v in enumerate(parts)} + result = await mcp_client.get_prompt(_pname, arguments) + # Extract text from GetPromptResult messages. + texts: list[str] = [] + for msg in result.messages: + content = msg.content + text: str | None = getattr(content, "text", None) + if text is not None: + texts.append(text) + else: + texts.append(str(content)) + return "\n".join(texts) if texts else "" + return CommandEntry( name=p.name, description=p.description or "", skill_uri=f"skill://{self._name}/{p.name}", source="remote", + handler=_prompt_handler, ) return None diff --git a/src/agentpool/capabilities/resource_protocols.py b/src/agentpool/capabilities/resource_protocols.py index ce645ce83..bb11fb301 100644 --- a/src/agentpool/capabilities/resource_protocols.py +++ b/src/agentpool/capabilities/resource_protocols.py @@ -15,11 +15,12 @@ if TYPE_CHECKING: - from collections.abc import AsyncIterator, Sequence + from collections.abc import AsyncIterator, Awaitable, Callable, Sequence from pathlib import PurePosixPath from upath import UPath + from agentpool.capabilities.agent_context import AgentContext from agentpool.capabilities.change_event import ChangeEvent @@ -100,12 +101,21 @@ class CommandEntry: description: Short description of what the command does. skill_uri: URI of the skill backing this command, if any. source: Where the command comes from — ``"local"`` or ``"remote"``. + handler: Optional async callable invoked with ``(input, ctx)`` to + execute the command. When ``None``, the command is display-only + and cannot be executed via ``CommandBridge.execute()``. Uses + ``compare=False`` to exclude the callable from equality/hash + comparisons, preserving existing ``CommandEntry`` comparison + semantics. """ name: str description: str = "" skill_uri: str = "" source: str = "local" + handler: Callable[[str, AgentContext], Awaitable[str]] | None = field( + default=None, compare=False + ) # ---- Protocols ---- diff --git a/src/agentpool/capabilities/skill_manager_cap.py b/src/agentpool/capabilities/skill_manager_cap.py index f7606da6b..c9ab71b1b 100644 --- a/src/agentpool/capabilities/skill_manager_cap.py +++ b/src/agentpool/capabilities/skill_manager_cap.py @@ -52,6 +52,7 @@ from pydantic_ai.capabilities import AbstractCapability from pydantic_ai.models import ModelRequestContext + from agentpool.capabilities.agent_context import AgentContext from agentpool.capabilities.mcp_server_cap import McpServerCap from agentpool.skills.skill import Skill from agentpool.skills.skill_tool_manager import SkillToolManager @@ -548,16 +549,37 @@ async def list_commands(self) -> Sequence[CommandEntry]: for name, skill in self._local_skills.items(): if not skill.user_invocable: continue + + # Capture ``name`` and ``skill`` by value for the closure. + skill_obj = skill + + async def _skill_handler( + user_input: str, + ctx: AgentContext, + _skill: Skill = skill_obj, + ) -> str: + """Load skill instructions and concatenate with user input.""" + del ctx # Skill execution does not need AgentContext. + try: + instructions = _skill.load_instructions() + except (ValueError, OSError): + return user_input + if not instructions: + return user_input + return f"{instructions}\n\n{user_input}" + entries.append( CommandEntry( name=name, description=skill.description, skill_uri=f"skill://{name}", source="local", + handler=_skill_handler, ) ) # Remote commands from child McpServerCap instances. + # PASS THROUGH — do NOT re-wrap or override child handlers. for child in self._children: if isinstance(child, CommandResource): try: @@ -587,11 +609,30 @@ async def get_command(self, name: str) -> CommandEntry | None: if name in self._local_skills: skill = self._local_skills[name] if skill.user_invocable: + # Capture ``skill`` by value for the closure. + skill_obj = skill + + async def _skill_handler( + user_input: str, + ctx: AgentContext, + _skill: Skill = skill_obj, + ) -> str: + """Load skill instructions and concatenate with user input.""" + del ctx # Skill execution does not need AgentContext. + try: + instructions = _skill.load_instructions() + except (ValueError, OSError): + return user_input + if not instructions: + return user_input + return f"{instructions}\n\n{user_input}" + return CommandEntry( name=name, description=skill.description, skill_uri=f"skill://{name}", source="local", + handler=_skill_handler, ) # Remote. diff --git a/src/agentpool_server/acp_server/session.py b/src/agentpool_server/acp_server/session.py index 4e045276a..3762b3c0e 100644 --- a/src/agentpool_server/acp_server/session.py +++ b/src/agentpool_server/acp_server/session.py @@ -236,6 +236,18 @@ def __post_init__(self) -> None: self._skill_change_task: asyncio.Task[None] | None = None self._skill_register_lock = asyncio.Lock() + # CommandBridge: discovers commands from all CommandResource capabilities + from agentpool.capabilities.command_bridge import CommandBridge + from agentpool.capabilities.extension_registry import Scope, ScopeLevel + + self._command_bridge: CommandBridge | None = None + hctx = self.host_context + if hctx.extension_registry is not None: + self._command_bridge = CommandBridge( + registry=hctx.extension_registry, + scope=Scope(level=ScopeLevel.SESSION, session_id=self.session_id), + ) + # CRITICAL: Initialize requests and acp_env BEFORE agent mutation self.notifications = ACPNotifications(client=self.client, session_id=self.session_id) self.requests = ACPRequests(client=self.client, session_id=self.session_id) @@ -278,8 +290,11 @@ async def permission_callback( # Register global commands from manifest.commands (e.g., static commands like start_eval) self._register_manifest_commands() - # Register pool-level skills as slash commands - self._register_skill_commands() + # Register commands from CommandBridge + skills (async — scheduled as task + # because __post_init__ is synchronous but discover_commands() is async) + self._command_register_task: asyncio.Task[None] | None = asyncio.create_task( + self._register_skill_commands() + ) # Subscribe to dynamic skill changes from ExtensionRegistry self._start_skill_change_watcher() diff --git a/src/agentpool_server/acp_server/session_agent_mgmt.py b/src/agentpool_server/acp_server/session_agent_mgmt.py index e00355a2b..4c64ce463 100644 --- a/src/agentpool_server/acp_server/session_agent_mgmt.py +++ b/src/agentpool_server/acp_server/session_agent_mgmt.py @@ -26,6 +26,8 @@ from slashed import CommandStore from acp.schema import AvailableCommand + from agentpool.capabilities.agent_context import AgentContext as CapabilityAgentContext + from agentpool.capabilities.command_bridge import CommandBridge logger = get_logger(__name__) @@ -58,6 +60,7 @@ class ACPSessionAgentMgmtMixin: _skill_bridge: Any # ACPSkillBridge _skill_change_task: asyncio.Task[None] | None _skill_register_lock: asyncio.Lock + _command_bridge: CommandBridge | None _update_callbacks: list[Callable[[], None]] _remote_commands: list[AvailableCommand] client_info: Any # Implementation | None @@ -118,16 +121,43 @@ def _register_manifest_commands(self) -> None: self._notify_command_update() self.log.info("Registered manifest commands", count=cmd_count) - def _register_skill_commands(self) -> None: - """Register pool-level and client-side skills as slash commands. - - Builds SkillCommand objects from the skills registry, feeds them - through ACPSkillBridge, and registers the resulting SlashedCommand - objects in command_store with replace=True for idempotent updates. - Also removes stale commands that are no longer present or invocable. + async def _register_skill_commands(self) -> None: + """Register commands from CommandBridge and skill commands as slash commands. + + Discovers commands from ALL CommandResource capabilities via + ``CommandBridge.discover_commands()``, converts them to + ``SlashedCommand`` objects, and registers them in ``command_store`` + with ``replace=True``. Also registers skill commands via + ``ACPSkillBridge`` for backward compatibility. When a command name + exists in both ``CommandBridge`` and ``ACPSkillBridge``, the + ``CommandBridge`` version is preferred (more comprehensive). + Also removes stale skill commands that are no longer present. """ from agentpool.skills.command import SkillCommand + # --- Phase 1: Discover commands from CommandBridge --- + bridge_names: set[str] = set() + if self._command_bridge is not None: + from agentpool.capabilities.command_bridge import CommandBridge + + try: + entries = await self._command_bridge.discover_commands() + except Exception: + self.log.exception("Failed to discover commands from CommandBridge") + entries = [] + + for entry in entries: + slashed_cmd = CommandBridge.entry_to_slashed_command(entry, self._command_bridge) + if slashed_cmd is not None: + self.command_store.register_command(slashed_cmd, replace=True) + bridge_names.add(entry.name) + self.log.debug( + "Registered CommandBridge command in command_store", + name=entry.name, + source=entry.source, + ) + + # --- Phase 2: Register skill commands via ACPSkillBridge (backward compat) --- ctx = self.host_context skills_registry = ctx.skills_registry skills = skills_registry.list_skills() @@ -152,72 +182,77 @@ def _register_skill_commands(self) -> None: stale_names = old_names - new_names for stale in stale_names: self._skill_bridge.handle_change(stale, None) - self.command_store.unregister_command(stale) + # Don't unregister from command_store if CommandBridge has it + if stale not in bridge_names: + self.command_store.unregister_command(stale) self.log.debug("Unregistered stale skill command", name=stale) # Add/update commands through the bridge for cmd in new_cmds: self._skill_bridge.handle_change(cmd.name, cmd) - # Register all bridge commands in command_store with replace=True + # Register all bridge commands in command_store with replace=True, + # but skip ones already registered by CommandBridge (de-duplication) for slashed_cmd in self._skill_bridge.get_commands(): + if slashed_cmd.name in bridge_names: + self.log.debug( + "Skipping skill command (CommandBridge version preferred)", + name=slashed_cmd.name, + ) + continue self.command_store.register_command(slashed_cmd, replace=True) self.log.debug( "Registered skill command in command_store", name=slashed_cmd.name, ) - if new_cmds or stale_names: + if new_cmds or stale_names or bridge_names: self._notify_command_update() self.log.info( - "Synced skill commands", + "Synced commands", + bridge_count=len(bridge_names), added=len(new_cmds), removed=len(stale_names), ) def _start_skill_change_watcher(self) -> None: - """Start watching for dynamic skill changes from ExtensionRegistry.""" - ctx = self.host_context - if ctx.extension_registry is None: + """Start watching for command/skill/prompt changes via CommandBridge.""" + if self._command_bridge is None: return self._skill_change_task = asyncio.create_task( - self._watch_skill_changes(), name=f"skill_watcher_{self.session_id}" + self._watch_skill_changes(), name=f"command_watcher_{self.session_id}" ) async def _watch_skill_changes(self) -> None: - """Watch for skill change events and rebuild skill commands. + """Watch for command/skill/prompt change events and rebuild commands. - Subscribes to ExtensionRegistry.merge_change_streams() for the - POOL scope. When a skills_changed event arrives, rebuilds skill - commands and sends an update to the client. + Consumes ``CommandBridge.watch_changes()`` which filters for + ``"commands_changed"``, ``"skills_changed"``, and + ``"prompts_changed"`` events from the ``ExtensionRegistry``. + When any of these events arrive, rebuilds the command list and + sends an ``AvailableCommandsUpdate`` to the client. """ - from agentpool.capabilities.extension_registry import Scope, ScopeLevel - - ctx = self.host_context - if ctx.extension_registry is None: - return - - stream = ctx.extension_registry.merge_change_streams(Scope(level=ScopeLevel.POOL)) - if stream is None: - self.log.debug("No skill change streams to watch") + if self._command_bridge is None: + self.log.debug("No CommandBridge — change watcher disabled") return try: - async for event in stream: - if event.kind != "skills_changed": - continue - self.log.info("Skill change detected, rebuilding skill commands") + async for event in self._command_bridge.watch_changes(): + self.log.info( + "Command/skill/prompt change detected, rebuilding commands", + kind=event.kind, + ) try: async with self._skill_register_lock: - self._register_skill_commands() + await self._register_skill_commands() await self.send_available_commands_update() except Exception: - self.log.exception("Failed to rebuild skill commands after change") + self.log.exception("Failed to rebuild commands after change") except asyncio.CancelledError: - self.log.debug("Skill change watcher cancelled") + self.log.debug("Command change watcher cancelled") raise except Exception: - self.log.exception("Skill change watcher error") + self.log.exception("Command change watcher error") async def init_client_skills(self) -> None: """Discover and load skills from client-side .claude/skills directory.""" @@ -228,7 +263,7 @@ async def init_client_skills(self) -> None: skills = self.host_context.skills_registry.list_skills() self.log.info("Collected client-side skills", skill_count=len(skills)) # Bridge newly discovered skills into command_store - self._register_skill_commands() + await self._register_skill_commands() await self.send_available_commands_update() except Exception as e: self.log.exception("Failed to discover client-side skills", error=e) @@ -252,7 +287,7 @@ async def switch_active_agent(self, agent_name: str) -> None: # Remove session-specific mutations from old agent before switching if isinstance(self.agent, Agent) and self.get_cwd_context in self.agent.sys_prompts.prompts: - self.agent.sys_prompts.prompts.remove(self.get_cwd_context) # pyright: ignore[reportArgumentType] # ty: ignore[invalid-argument-type] + self.agent.sys_prompts.prompts.remove(self.get_cwd_context) # pyright: ignore[reportArgumentType] # Create new session agent via SessionPool (pool-level agents removed) ctx = self.host_context @@ -270,7 +305,7 @@ async def switch_active_agent(self, agent_name: str) -> None: self.agent.env = self.acp_env self.agent._input_provider = self.input_provider if isinstance(self.agent, Agent): - self.agent.sys_prompts.prompts.append(self.get_cwd_context) # pyright: ignore[reportArgumentType] # ty: ignore[invalid-argument-type] + self.agent.sys_prompts.prompts.append(self.get_cwd_context) # pyright: ignore[reportArgumentType] # Reconnect signal with suppress(Exception): @@ -315,13 +350,56 @@ async def _register_prompt_hub_commands(self) -> None: self.log.info("Registered hub prompts as slash commands", cmd_count=cmd_count) await self.send_available_commands_update() # Send updated command list to client + def _build_command_agent_context(self) -> CapabilityAgentContext: + """Construct a capabilities.AgentContext from the session's current state. + + This is used for ``CommandBridge.execute()`` which expects a + :class:`~agentpool.capabilities.agent_context.AgentContext` (not + the ``agents.AgentContext`` used by ``command_store``). + + Returns: + A minimal ``AgentContext`` with the session's host context, + extension registry, and scope information. + """ + from agentpool.capabilities.agent_context import AgentContext as CapabilityAgentContext + from agentpool.capabilities.runloop_delegation import RunLoopDelegationService + from agentpool.host.context import RunScope + from agentpool.host.registry import AgentRegistry + from agentpool.orchestrator.session_controller import SessionState + + hctx = self.host_context + registry = AgentRegistry() + delegation: RunLoopDelegationService = RunLoopDelegationService( + registry=registry, + host=hctx, + session_id=self.session_id, + ) + session = SessionState( + session_id=self.session_id, + agent_name=self.agent.name, + ) + scope = RunScope(session_id=self.session_id) + return CapabilityAgentContext( + agent_registry=registry, + delegation=delegation, + session=session, + scope=scope, + host=hctx, + extension_registry=hctx.extension_registry, + ) + @logfire.instrument(r"Execute Slash Command {command_text}") async def execute_slash_command(self, command_text: str) -> None: """Execute any slash command with unified handling. + Routes execution through ``CommandBridge.execute()`` first. If + ``CommandNotFoundError`` is raised, falls back to the existing + ``command_store.execute_command()`` path (manifest commands, + debug commands, etc.). If ``CommandNotExecutableError`` is raised, + sends an error toast to the client. + Args: command_text: Full command text (including slash) - session: ACP session context """ from agentpool_server.acp_server.session import SLASH_PATTERN @@ -342,7 +420,31 @@ async def execute_slash_command(self, command_text: str) -> None: await self.notifications.send_agent_text(error_msg) return - # Create context with session data + # --- Phase 1: Try CommandBridge.execute() first --- + if self._command_bridge is not None: + from agentpool.capabilities.command_bridge import ( + CommandNotExecutableError, + CommandNotFoundError, + ) + + try: + agent_ctx = self._build_command_agent_context() + result = await self._command_bridge.execute(command_name, args, agent_ctx) + except CommandNotFoundError: + pass # Fall through to command_store fallback + except CommandNotExecutableError: + await self._send_toast( + message=f"Command `/{command_name}` is not executable", + level="error", + ) + await anyio.sleep(0.05) + return + else: + await self.notifications.send_agent_text(result) + await anyio.sleep(0.05) # Allow network buffers to flush + return + + # --- Phase 2: Fallback to command_store.execute_command() --- agent_context = self.agent.get_context(data=self) cmd_ctx = self.command_store.create_context( data=agent_context, diff --git a/src/agentpool_server/opencode_server/command_bridge.py b/src/agentpool_server/opencode_server/command_bridge.py new file mode 100644 index 000000000..950ebcf7d --- /dev/null +++ b/src/agentpool_server/opencode_server/command_bridge.py @@ -0,0 +1,121 @@ +"""OpenCode command bridge for exposing CommandResource commands as slashed Commands. + +The :class:`OpenCodeCommandBridge` wraps :class:`CommandBridge` to provide +OpenCode-specific command conversion. It discovers commands from ALL +``CommandResource`` capabilities registered in the ``ExtensionRegistry``, +converts them to ``slashed.Command`` instances, and exposes them alongside +existing skill and MCP prompt commands. + +This class is separate from :class:`OpenCodeSkillBridge` and does NOT modify +``create_skill_command()`` — skill-based commands continue to flow through the +existing ``SkillCommand`` → ``SlashedCommand`` path. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import logfire + + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from slashed import Command as SlashedCommand + + from agentpool.capabilities.agent_context import AgentContext + from agentpool.capabilities.change_event import ChangeEvent + from agentpool.capabilities.command_bridge import CommandBridge + from agentpool.capabilities.extension_registry import ExtensionRegistry + + +class OpenCodeCommandBridge: + """Bridge wrapping :class:`CommandBridge` for OpenCode-specific command conversion. + + Constructed per-server (not per-session) with a SESSION-level scope. Discovers + commands from all :class:`CommandResource` capabilities via the inner + :class:`CommandBridge`, converts each :class:`CommandEntry` to a + :class:`slashed.Command`, and delegates execution to the inner bridge. + + This class does NOT replace :class:`OpenCodeSkillBridge` — both coexist. + Skill commands flow through the existing ``SkillCommand`` path; capability + commands flow through this bridge. + """ + + def __init__(self, registry: ExtensionRegistry, session_id: str) -> None: + """Initialize the OpenCode command bridge. + + Args: + registry: The ExtensionRegistry to query for CommandResource + capabilities. + session_id: The session ID for scoping command discovery. + """ + from agentpool.capabilities.command_bridge import CommandBridge + from agentpool.capabilities.extension_registry import Scope, ScopeLevel + + self._bridge: CommandBridge = CommandBridge( + registry=registry, + scope=Scope(level=ScopeLevel.SESSION, session_id=session_id), + ) + + @logfire.instrument("opencode_command_bridge.discover_commands") + async def discover_commands(self) -> list[SlashedCommand]: + """Discover all commands and convert to SlashedCommand instances. + + Calls the inner :meth:`CommandBridge.discover_commands` to collect + :class:`CommandEntry` objects from all visible ``CommandResource`` + capabilities, then converts each entry via + :meth:`CommandBridge.entry_to_slashed_command`. Entries without a + handler (display-only) are filtered out. + + Returns: + List of :class:`SlashedCommand` instances from capability commands. + """ + from agentpool.capabilities.command_bridge import CommandBridge + + entries = await self._bridge.discover_commands() + commands: list[SlashedCommand] = [] + for entry in entries: + slashed = CommandBridge.entry_to_slashed_command(entry, self._bridge) + if slashed is not None: + commands.append(slashed) + return commands + + @logfire.instrument("opencode_command_bridge.execute {name}") + async def execute(self, name: str, input: str, ctx: AgentContext) -> str: # noqa: A002 + """Execute a command by name via the inner CommandBridge. + + Delegates to :meth:`CommandBridge.execute`. Raises + :class:`CommandNotFoundError` if the command is not in the bridge's + index, or :class:`CommandNotExecutableError` if the entry has no handler. + + Args: + name: The command name to execute. + input: The raw input text (arguments) for the command. + ctx: The agent context for this execution. + + Returns: + The string result from the command handler. + """ + return await self._bridge.execute(name, input, ctx) + + async def watch_changes(self) -> AsyncIterator[ChangeEvent]: + """Watch for command list changes via the inner CommandBridge. + + Delegates to :meth:`CommandBridge.watch_changes`, which filters for + ``"commands_changed"``, ``"skills_changed"``, and + ``"prompts_changed"`` events. + + Yields: + ChangeEvent: A change event with a relevant kind. + """ + async for event in self._bridge.watch_changes(): + yield event + + def get_command_bridge(self) -> CommandBridge: + """Return the inner CommandBridge for direct access. + + Returns: + The wrapped :class:`CommandBridge` instance. + """ + return self._bridge diff --git a/src/agentpool_server/opencode_server/routes/agent_routes.py b/src/agentpool_server/opencode_server/routes/agent_routes.py index acf4b6a41..b8c76c2bf 100644 --- a/src/agentpool_server/opencode_server/routes/agent_routes.py +++ b/src/agentpool_server/opencode_server/routes/agent_routes.py @@ -207,32 +207,64 @@ async def list_skills(state: StateDep) -> list[SkillInfo]: @router.get("/command") -async def list_commands(state: StateDep) -> list[Command]: +async def list_commands(state: StateDep) -> list[Command]: # noqa: PLR0915 """List available slash commands. Commands include: + - Capability commands from OpenCodeCommandBridge (if available) - MCP prompts as commands - Skill commands from skill_bridge (if available) or skill_provider + + De-duplication by name: if a command name exists in both the + OpenCodeCommandBridge and existing sources, the CommandBridge version + is preferred. """ commands: list[Command] = [] + seen_names: set[str] = set() + + # 1. Add capability commands from OpenCodeCommandBridge (highest priority) + if state.command_bridge is not None: + try: + slashed_commands = await state.command_bridge.discover_commands() + for slashed in slashed_commands: + if slashed.name in seen_names: + continue + seen_names.add(slashed.name) + commands.append( + Command( + name=slashed.name, + description=slashed.description, + source="command", + template="", + hints=[], + ) + ) + except Exception: + logger.exception("Failed to discover capability commands") - # Add MCP prompts as commands (source="mcp") + # 2. Add MCP prompts as commands (source="mcp") try: prompts = await state.agent.list_prompts() - commands.extend([ - Command(name=p.name, description=p.description, source="mcp", hints=[]) for p in prompts - ]) + for p in prompts: + if p.name in seen_names: + continue + seen_names.add(p.name) + commands.append(Command(name=p.name, description=p.description, source="mcp", hints=[])) except Exception: # noqa: BLE001 pass - # Add skill commands from skill_bridge if available + # 3. Add skill commands from skill_bridge if available logger.debug( "list_commands debug", skill_bridge_exists=state.skill_bridge is not None, skill_provider_exists=state.pool.skill_provider is not None, + command_bridge_exists=state.command_bridge is not None, ) if state.skill_bridge is not None: for skill_cmd in state.skill_bridge.get_skill_commands(): + if skill_cmd.name in seen_names: + continue + seen_names.add(skill_cmd.name) # For virtual skills (from MCP), fetch instructions from resolver template = "" if state.pool.skill_resolver is not None: @@ -272,6 +304,9 @@ async def list_commands(state: StateDep) -> list[Command]: skill_count=len(provider_skills), ) for entry in provider_skills: + if entry.name in seen_names: + continue + seen_names.add(entry.name) # Use provider's read_skill for content try: template = await provider.read_skill(entry.name) or "" diff --git a/src/agentpool_server/opencode_server/routes/session_routes.py b/src/agentpool_server/opencode_server/routes/session_routes.py index c44f41598..03f0beb50 100644 --- a/src/agentpool_server/opencode_server/routes/session_routes.py +++ b/src/agentpool_server/opencode_server/routes/session_routes.py @@ -254,6 +254,13 @@ async def _execute_slashed_command( # noqa: PLR0915 try: await command.execute(cmd_ctx, args, {}) except Exception as e: + # Let CommandNotFoundError propagate so execute_command can + # fall back to MCP prompts for commands no longer in the + # CommandBridge index. + from agentpool.capabilities.command_bridge import CommandNotFoundError + + if isinstance(e, CommandNotFoundError): + raise raise HTTPException(status_code=500, detail=f"Command execution failed: {e}") from e # Get command output @@ -2116,7 +2123,21 @@ async def execute_command( # noqa: PLR0915 "Both slashed command and prompt exist for '%s'. Using slashed command.", request.command, ) - return await _execute_slashed_command(state, session_id, request) + try: + return await _execute_slashed_command(state, session_id, request) + except Exception as exc: + # If CommandBridge raised CommandNotFoundError, fall through + # to MCP prompts (the command may have been removed from the + # bridge's index but still exists as an MCP prompt). + from agentpool.capabilities.command_bridge import CommandNotFoundError + + if isinstance(exc, CommandNotFoundError): + logger.info( + "Command not in CommandBridge index, falling back to MCP prompts", + command=request.command, + ) + else: + raise # Fall back to MCP prompts (existing code remains unchanged) session_agent = state.agent diff --git a/src/agentpool_server/opencode_server/server.py b/src/agentpool_server/opencode_server/server.py index 4d875cdc2..073c71192 100644 --- a/src/agentpool_server/opencode_server/server.py +++ b/src/agentpool_server/opencode_server/server.py @@ -266,6 +266,63 @@ async def _watch_mcp_tool_changes() -> None: state._mcp_tool_change_task = asyncio.create_task(_watch_mcp_tool_changes()) + # Set up OpenCodeCommandBridge for capability command discovery. + # This is separate from the skill bridge above — it discovers commands + # from ALL CommandResource capabilities (not just skills) via the + # ExtensionRegistry, converts them to SlashedCommand instances, and + # registers them in the CommandStore for execution. + extension_registry = state.pool.extension_registry + if extension_registry is not None: + from agentpool_server.opencode_server.command_bridge import ( + OpenCodeCommandBridge, + ) + + oc_command_bridge = OpenCodeCommandBridge( + registry=extension_registry, + session_id="opencode-server", + ) + state.command_bridge = oc_command_bridge + + # Watch for command changes via CommandBridge.watch_changes(). + # Handles "commands_changed", "skills_changed", and "prompts_changed" + # events — all three trigger a command list rebuild. + async def _watch_command_changes() -> None: + """Watch for command change events and rebuild CommandStore.""" + # Initial discovery happens here (inside the async watcher task). + try: + initial_commands = await oc_command_bridge.discover_commands() + if state.command_store is not None: + for slashed_cmd in initial_commands: + state.command_store.register_command(slashed_cmd, replace=True) + logger.info( + "OpenCode command bridge initial discovery complete", + capability_command_count=len(initial_commands), + ) + except Exception: + logger.exception("Failed to discover initial capability commands") + + async for _event in oc_command_bridge.watch_changes(): + logger.info( + "Command change detected, rebuilding capability commands", + kind=_event.kind, + capability=_event.capability_name, + ) + try: + new_commands = await oc_command_bridge.discover_commands() + if state.command_store is not None: + # Re-register all capability commands (replace=True + # handles both new and existing entries). + for slashed_cmd in new_commands: + state.command_store.register_command(slashed_cmd, replace=True) + logger.debug( + "Capability commands rebuilt", + command_count=len(new_commands), + ) + except Exception: + logger.exception("Failed to rebuild capability commands after change") + + state._command_change_task = asyncio.create_task(_watch_command_changes()) + # Set up todo change callback to broadcast events async def on_todo_change(tracker: TodoTracker) -> None: """Broadcast todo updates to all active sessions.""" @@ -418,6 +475,16 @@ async def check_for_updates() -> None: except Exception: logger.exception("Error during skill change task cleanup") state._skill_change_task = None + # Cancel command change watcher + if state._command_change_task is not None: + state._command_change_task.cancel() + try: + await state._command_change_task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("Error during command change task cleanup") + state._command_change_task = None # Cancel MCP tool change watcher if state._mcp_tool_change_task is not None: state._mcp_tool_change_task.cancel() diff --git a/src/agentpool_server/opencode_server/state.py b/src/agentpool_server/opencode_server/state.py index 6dc34040d..13ac4b5ec 100644 --- a/src/agentpool_server/opencode_server/state.py +++ b/src/agentpool_server/opencode_server/state.py @@ -87,8 +87,10 @@ class ServerState: event_managers: dict[str, Any] = field(default_factory=dict) auth_service: ProviderAuthService = field(default_factory=create_default_auth_service) skill_bridge: Any = field(default=None) + command_bridge: Any = field(default=None) command_store: CommandStore | None = field(default=None) _skill_change_task: Any = field(default=None, repr=False) + _command_change_task: Any = field(default=None, repr=False) _mcp_tool_change_task: Any = field(default=None, repr=False) session_pool_integration: Any = field(default=None) session_controller: SessionController | None = field(default=None) diff --git a/tests/acp_server/test_acp_skill_commands.py b/tests/acp_server/test_acp_skill_commands.py index 13028e8e6..44f6afa90 100644 --- a/tests/acp_server/test_acp_skill_commands.py +++ b/tests/acp_server/test_acp_skill_commands.py @@ -181,6 +181,7 @@ def _make_minimal_session( session = MagicMock() session.host_context = host_context session._skill_bridge = skill_bridge + session._command_bridge = None # No ExtensionRegistry in unit tests session.command_store = CommandStore() session.command_store._initialize_sync() session._notify_command_update = MagicMock() @@ -190,10 +191,11 @@ def _make_minimal_session( def _call_register_skill_commands(session: MagicMock) -> None: """Invoke the real _register_skill_commands method on a mock session.""" - # We need to call the unbound method from ACPSession on our mock + import asyncio + from agentpool_server.acp_server.session import ACPSession - ACPSession._register_skill_commands(session) # type: ignore[arg-type] + asyncio.run(ACPSession._register_skill_commands(session)) # type: ignore[arg-type] def _skill_commands_from_store(store: CommandStore) -> list[SlashedCommand]: diff --git a/tests/acp_server/test_acp_skill_e2e.py b/tests/acp_server/test_acp_skill_e2e.py index f8c364a5a..f645a93b7 100644 --- a/tests/acp_server/test_acp_skill_e2e.py +++ b/tests/acp_server/test_acp_skill_e2e.py @@ -80,25 +80,35 @@ def _make_mock_acp_agent() -> MagicMock: return mock_acp_agent -def _create_session( +async def _create_session( pool: AgentPool, agent: Agent, *, mock_client: AsyncMock | None = None, mock_acp_agent: MagicMock | None = None, ) -> ACPSession: - """Create a real ACPSession with mocked transport layer.""" + """Create a real ACPSession with mocked transport layer. + + Awaits the async command registration task scheduled by __post_init__ + so skill/CommandBridge commands are registered before the caller + inspects the session. + """ if mock_client is None: mock_client = AsyncMock() if mock_acp_agent is None: mock_acp_agent = _make_mock_acp_agent() - return ACPSession( + session = ACPSession( session_id="test-session-e2e", agent=agent, cwd="/tmp", client=mock_client, acp_agent=mock_acp_agent, ) + # Wait for the async _register_skill_commands task to complete + if session._command_register_task is not None: + with contextlib.suppress(Exception): + await session._command_register_task + return session def _skill_cmd_names(store: CommandStore) -> set[str]: @@ -150,7 +160,9 @@ async def test_e2e_pool_skills_in_available_commands_update( skill_b = _make_skill(name="skill-b", description="Skill B") _mock_skills_on_pool(pool, [skill_a, skill_b]) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) session.notifications.update_commands = AsyncMock() # type: ignore[method-assign] await session.send_available_commands_update() @@ -176,7 +188,9 @@ async def test_e2e_init_client_skills_sends_update( pool, agent = pool_and_agent _mock_skills_on_pool(pool, []) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Simulate client skills being discovered client_skill = _make_skill(name="client-skill", description="Client skill") @@ -226,7 +240,9 @@ def _merge_streams(_scope: Any) -> Any: pool._host_context = None # type: ignore[attr-defined] pool._extension_registry = mock_ext_registry # type: ignore[attr-defined] - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Replace update mock BEFORE triggering the event session.notifications.update_commands = AsyncMock() # type: ignore[method-assign] @@ -288,7 +304,9 @@ async def test_e2e_execute_skill_command_injects_instructions( ) _mock_skills_on_pool(pool, [skill]) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Mock notifications to avoid real ACP communication session.notifications.send_agent_text = AsyncMock() # type: ignore[method-assign] @@ -329,7 +347,9 @@ async def _empty_stream() -> Any: mock_ext_registry.merge_change_streams.return_value = _empty_stream() pool._extension_registry = mock_ext_registry # type: ignore[attr-defined] - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Verify watcher task was created assert session._skill_change_task is not None, "Skill change watcher task was not created" @@ -358,7 +378,9 @@ async def test_e2e_non_invocable_skills_never_in_commands( ) _mock_skills_on_pool(pool, [invocable, non_invocable]) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Check command_store: non-invocable should not be registered store_names = _skill_cmd_names(session.command_store) @@ -407,7 +429,9 @@ async def test_regression_manifest_commands_still_work( return_value={"manifest-cmd": manifest_cmd} ) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Both manifest and skill commands should be in the store all_names = _all_cmd_names(session.command_store) @@ -425,7 +449,9 @@ async def test_regression_mcp_prompts_as_commands_still_works( pool, agent = pool_and_agent _mock_skills_on_pool(pool, []) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Mock agent.list_prompts to return a fake MCP prompt mock_prompt = MagicMock() @@ -475,7 +501,9 @@ async def test_regression_send_update_includes_both_manifest_and_skill_commands( return_value={"manifest-cmd": manifest_cmd} ) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) session.notifications.update_commands = AsyncMock() # type: ignore[method-assign] await session.send_available_commands_update() @@ -515,7 +543,9 @@ async def test_regression_command_name_collision_last_registered_wins( return_value={shared_name: manifest_cmd} ) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # __post_init__ calls _register_manifest_commands() first, # then _register_skill_commands() @@ -552,7 +582,9 @@ async def test_regression_load_skills_false_pool_skills_still_registered( pool_skill = _make_skill(name="pool-skill", description="A pool-level skill") _mock_skills_on_pool(pool, [pool_skill]) - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) # Pool skills should be registered (from __post_init__ -> _register_skill_commands) store_names = _skill_cmd_names(session.command_store) @@ -591,7 +623,9 @@ async def test_regression_50_plus_skills_registration_performance( import time start = time.monotonic() - session = _create_session(pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent) + session = await _create_session( + pool, agent, mock_client=mock_client, mock_acp_agent=mock_acp_agent + ) elapsed = time.monotonic() - start # All 50 skills should be registered diff --git a/tests/acp_server/test_acp_skill_lifecycle.py b/tests/acp_server/test_acp_skill_lifecycle.py index 99f61e6c7..178fc013f 100644 --- a/tests/acp_server/test_acp_skill_lifecycle.py +++ b/tests/acp_server/test_acp_skill_lifecycle.py @@ -82,6 +82,18 @@ def _make_mock_session( session._skill_change_task = None session._skill_register_lock = asyncio.Lock() session._remote_commands = [] + + # Create CommandBridge if extension_registry is provided + if extension_registry is not None: + from agentpool.capabilities.command_bridge import CommandBridge + from agentpool.capabilities.extension_registry import Scope, ScopeLevel + + session._command_bridge = CommandBridge( + registry=extension_registry, + scope=Scope(level=ScopeLevel.SESSION, session_id="test-session-id"), + ) + else: + session._command_bridge = None session._update_callbacks = [] session.log = MagicMock() session.fs = MagicMock() @@ -212,7 +224,7 @@ async def test_pool_and_client_skills_coexist_without_duplicates() -> None: client_skill = _make_skill(name="client-skill", description="Client-level skill") # Pool skill pre-registered via _register_skill_commands in __post_init__ session = _make_mock_session(skills_list=[pool_skill, client_skill]) - session._register_skill_commands() + await session._register_skill_commands() # Both skills should be in command_store assert session.command_store.get_command("pool-skill") is not None @@ -220,7 +232,7 @@ async def test_pool_and_client_skills_coexist_without_duplicates() -> None: # Re-register with same skills (simulating init_client_skills re-registering) # _register_skill_commands uses replace=True, so no duplicates - session._register_skill_commands() + await session._register_skill_commands() pool_cmds = [c for c in session.command_store.list_commands() if c.name == "pool-skill"] client_cmds = [c for c in session.command_store.list_commands() if c.name == "client-skill"] @@ -246,7 +258,7 @@ async def test_watch_skill_changes_calls_register_on_skills_changed_event() -> N mock_ext_registry.merge_change_streams.return_value = mock_stream session = _make_mock_session(extension_registry=mock_ext_registry) - session._register_skill_commands = MagicMock() + session._register_skill_commands = AsyncMock() session.send_available_commands_update = AsyncMock() await session._watch_skill_changes() @@ -257,7 +269,7 @@ async def test_watch_skill_changes_calls_register_on_skills_changed_event() -> N @pytest.mark.unit async def test_watch_skill_changes_ignores_non_skills_changed_events() -> None: - """T8.2: _watch_skill_changes() ignores non-skills_changed events.""" + """T8.2: _watch_skill_changes() ignores non-relevant events.""" tools_event = ChangeEvent(capability_name="test-cap", kind="tools_changed") resources_event = ChangeEvent(capability_name="test-cap", kind="resources_changed") skills_event = ChangeEvent(capability_name="test-cap", kind="skills_changed") @@ -267,7 +279,7 @@ async def test_watch_skill_changes_ignores_non_skills_changed_events() -> None: mock_ext_registry.merge_change_streams.return_value = mock_stream session = _make_mock_session(extension_registry=mock_ext_registry) - session._register_skill_commands = MagicMock() + session._register_skill_commands = AsyncMock() session.send_available_commands_update = AsyncMock() await session._watch_skill_changes() @@ -304,7 +316,7 @@ async def test_watch_skill_changes_handles_none_stream_no_crash() -> None: mock_ext_registry.merge_change_streams.return_value = None session = _make_mock_session(extension_registry=mock_ext_registry) - session._register_skill_commands = MagicMock() + session._register_skill_commands = AsyncMock() # Should return without error await session._watch_skill_changes() @@ -327,7 +339,7 @@ async def test_concurrent_register_skill_commands_serialized_by_lock() -> None: # _register_skill_commands itself is sync. We test that calling # it multiple times with replace=True doesn't produce duplicates. for _ in range(5): - session._register_skill_commands() + await session._register_skill_commands() cmds = [c for c in session.command_store.list_commands() if c.name == "concurrent-skill"] assert len(cmds) == 1 @@ -350,7 +362,7 @@ async def test_execute_slash_command_finds_and_executes_skill() -> None: instructions="Do something useful", ) session = _make_mock_session(skills_list=[skill]) - session._register_skill_commands() + await session._register_skill_commands() # Mock command_store.execute_command to verify it's called session.command_store.execute_command = AsyncMock() diff --git a/tests/capabilities/test_command_bridge.py b/tests/capabilities/test_command_bridge.py new file mode 100644 index 000000000..43a5cb8bd --- /dev/null +++ b/tests/capabilities/test_command_bridge.py @@ -0,0 +1,785 @@ +"""Tests for the capability-command-bridge feature. + +Covers: + 1. CommandEntry handler field (task 1.4) + 2. CommandBridge.discover_commands() (task 2.6) + 3. CommandBridge.execute() (task 2.7) + 4. CommandBridge.watch_changes() (task 2.8) + 5. entry_to_slashed_command() (task 2.9) + 6. SkillManagerCap handler tests (task 3.3) + 7. McpServerCap handler tests (task 3.4) + 8. SkillManagerCap pass-through test (task 3.5) + 9. Backward compatibility test (task 3.6) +""" + +from __future__ import annotations + +import asyncio +from pathlib import PurePosixPath +from typing import TYPE_CHECKING, Any, Self, cast +from unittest.mock import MagicMock + +import pytest + +from agentpool.capabilities.agent_context import AgentContext +from agentpool.capabilities.change_event import ChangeEvent +from agentpool.capabilities.command_bridge import ( + CommandBridge, + CommandNotExecutableError, + CommandNotFoundError, +) +from agentpool.capabilities.extension_registry import ( + ExtensionRegistry, + Scope, + ScopeLevel, +) +from agentpool.capabilities.resource_protocols import ( + CommandEntry, +) +from agentpool.capabilities.skill_manager_cap import SkillManagerCap +from agentpool.skills.skill import Skill + + +if TYPE_CHECKING: + from agentpool.mcp_server.client import MCPClient + + +# ---- Shared fixtures and helpers ---- + + +def _make_agent_context() -> AgentContext: + """Create a minimal AgentContext with MagicMock fields for testing.""" + return AgentContext( + agent_registry=MagicMock(), + delegation=MagicMock(), + session=MagicMock(), + scope=MagicMock(), + host=MagicMock(), + extension_registry=None, + ) + + +class FakeCommandResource: + """Fake capability implementing CommandResource for testing.""" + + def __init__( + self, + name: str = "fake-cmd", + commands: list[CommandEntry] | None = None, + ) -> None: + self._name = name + self._commands = commands or [] + + def get_serialization_name(self) -> str: + return self._name + + async def list_commands(self) -> list[CommandEntry]: + return list(self._commands) + + async def get_command(self, name: str) -> CommandEntry | None: + return next((c for c in self._commands if c.name == name), None) + + +class FakeChangeObservable: + """Fake capability implementing ChangeObservable for testing.""" + + def __init__( + self, + name: str = "fake-obs", + events: list[ChangeEvent] | None = None, + ) -> None: + self._name = name + self._events = events or [] + + def get_serialization_name(self) -> str: + return self._name + + def on_change(self) -> asyncio.Queue[ChangeEvent] | None: + if not self._events: + return None + queue: asyncio.Queue[ChangeEvent | None] = asyncio.Queue() + + async def _gen() -> Any: + for ev in self._events: + await queue.put(ev) + await queue.put(None) # sentinel + + # Push events into queue immediately via a task wrapper. + # We return a simple async generator instead. + async def _iterator() -> Any: + for ev in self._events: + yield ev + + return _iterator() + + +# ===================================================================== +# 1. CommandEntry handler field tests (task 1.4) +# ===================================================================== + + +@pytest.mark.unit +async def test_command_entry_with_handler_populated() -> None: + """CommandEntry with handler is not None and is callable.""" + + async def handler(input_text: str, ctx: AgentContext) -> str: + return f"result: {input_text}" + + entry = CommandEntry( + name="test-cmd", + description="A test command", + handler=handler, + ) + assert entry.handler is not None + assert callable(entry.handler) + + +@pytest.mark.unit +async def test_command_entry_with_handler_none_default() -> None: + """CommandEntry handler defaults to None.""" + entry = CommandEntry(name="display-only") + assert entry.handler is None + + +@pytest.mark.unit +async def test_command_entry_handler_returns_expected_result() -> None: + """Handler is callable and returns expected result when invoked.""" + + async def handler(input_text: str, ctx: AgentContext) -> str: + return f"processed({input_text})" + + entry = CommandEntry(name="echo", handler=handler) + ctx = _make_agent_context() + assert entry.handler is not None + result = await entry.handler("hello", ctx) + assert result == "processed(hello)" + + +@pytest.mark.unit +async def test_command_entry_compare_false_handler() -> None: + """Two CommandEntry with same fields but different handlers are equal.""" + + async def handler_a(input_text: str, ctx: AgentContext) -> str: + return "a" + + async def handler_b(input_text: str, ctx: AgentContext) -> str: + return "b" + + entry_a = CommandEntry( + name="cmd", + description="desc", + skill_uri="skill://cmd", + source="local", + handler=handler_a, + ) + entry_b = CommandEntry( + name="cmd", + description="desc", + skill_uri="skill://cmd", + source="local", + handler=handler_b, + ) + # compare=False means handler is excluded from equality. + assert entry_a == entry_b + # But handlers are different objects. + assert entry_a.handler is not entry_b.handler + + +# ===================================================================== +# 2. CommandBridge.discover_commands() tests (task 2.6) +# ===================================================================== + + +@pytest.mark.unit +async def test_discover_commands_multiple_capabilities() -> None: + """Multiple capabilities each return commands — all discovered.""" + + async def handler_a(input_text: str, ctx: AgentContext) -> str: + return "a" + + async def handler_b(input_text: str, ctx: AgentContext) -> str: + return "b" + + cap_a = FakeCommandResource( + "cap-a", + [CommandEntry(name="cmd-a", description="A", handler=handler_a)], + ) + cap_b = FakeCommandResource( + "cap-b", + [CommandEntry(name="cmd-b", description="B", handler=handler_b)], + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(cap_a, scope) # type: ignore[arg-type] + registry.register(cap_b, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + commands = await bridge.discover_commands() + names = [c.name for c in commands] + assert "cmd-a" in names + assert "cmd-b" in names + assert len(commands) == 2 + + +@pytest.mark.unit +async def test_discover_commands_duplicate_at_different_scopes_turn_wins() -> None: + """Same command name at TURN and POOL — TURN wins (most specific first).""" + + async def turn_handler(input_text: str, ctx: AgentContext) -> str: + return "turn" + + async def pool_handler(input_text: str, ctx: AgentContext) -> str: + return "pool" + + turn_cap = FakeCommandResource( + "turn-cap", + [CommandEntry(name="dup", description="turn", handler=turn_handler)], + ) + pool_cap = FakeCommandResource( + "pool-cap", + [CommandEntry(name="dup", description="pool", handler=pool_handler)], + ) + registry = ExtensionRegistry() + pool_scope = Scope(level=ScopeLevel.POOL) + turn_scope = Scope( + level=ScopeLevel.TURN, + session_id="s1", + agent_name="a1", + turn_id="t1", + ) + registry.register(pool_cap, pool_scope) # type: ignore[arg-type] + registry.register(turn_cap, turn_scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, turn_scope) + commands = await bridge.discover_commands() + assert len(commands) == 1 + assert commands[0].name == "dup" + assert commands[0].description == "turn" + + +@pytest.mark.unit +async def test_discover_commands_empty_registry() -> None: + """No capabilities → empty list.""" + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + bridge = CommandBridge(registry, scope) + commands = await bridge.discover_commands() + assert commands == [] + + +@pytest.mark.unit +async def test_discover_commands_builds_index_for_execute() -> None: + """After discover_commands(), execute() can find commands by name.""" + + async def handler(input_text: str, ctx: AgentContext) -> str: + return "ok" + + cap = FakeCommandResource( + "cap", + [CommandEntry(name="indexed", description="test", handler=handler)], + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(cap, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + await bridge.discover_commands() + ctx = _make_agent_context() + result = await bridge.execute("indexed", "input", ctx) + assert result == "ok" + + +# ===================================================================== +# 3. CommandBridge.execute() tests (task 2.7) +# ===================================================================== + + +@pytest.mark.unit +async def test_execute_handler_found_returns_result() -> None: + """execute(name, input, ctx) returns handler output.""" + + async def handler(input_text: str, ctx: AgentContext) -> str: + return f"handled:{input_text}" + + cap = FakeCommandResource( + "cap", + [CommandEntry(name="my-cmd", description="test", handler=handler)], + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(cap, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + await bridge.discover_commands() + ctx = _make_agent_context() + result = await bridge.execute("my-cmd", "hello", ctx) + assert result == "handled:hello" + + +@pytest.mark.unit +async def test_execute_handler_none_raises_not_executable() -> None: + """Handler is None → raises CommandNotExecutableError.""" + cap = FakeCommandResource( + "cap", + [CommandEntry(name="display-only", description="no handler")], + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(cap, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + await bridge.discover_commands() + ctx = _make_agent_context() + with pytest.raises(CommandNotExecutableError) as exc_info: + await bridge.execute("display-only", "input", ctx) + assert exc_info.value.name == "display-only" + + +@pytest.mark.unit +async def test_execute_command_not_found_raises() -> None: + """Command not found → raises CommandNotFoundError.""" + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + bridge = CommandBridge(registry, scope) + await bridge.discover_commands() + ctx = _make_agent_context() + with pytest.raises(CommandNotFoundError) as exc_info: + await bridge.execute("nonexistent", "input", ctx) + assert exc_info.value.name == "nonexistent" + + +@pytest.mark.unit +async def test_execute_handler_exception_propagates_unwrapped() -> None: + """If handler raises ValueError, it propagates as ValueError (not wrapped).""" + + async def handler(input_text: str, ctx: AgentContext) -> str: + msg = "boom" + raise ValueError(msg) + + cap = FakeCommandResource( + "cap", + [CommandEntry(name="error-cmd", description="test", handler=handler)], + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(cap, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + await bridge.discover_commands() + ctx = _make_agent_context() + with pytest.raises(ValueError, match="boom"): + await bridge.execute("error-cmd", "input", ctx) + + +# ===================================================================== +# 4. CommandBridge.watch_changes() tests (task 2.8) +# ===================================================================== + + +@pytest.mark.unit +async def test_watch_changes_commands_changed_forwarded() -> None: + """'commands_changed' event is forwarded by watch_changes().""" + event = ChangeEvent( + capability_name="cap", + kind="commands_changed", + source_uri="skill://cap", + ) + obs = FakeChangeObservable("obs", [event]) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(obs, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + results: list[ChangeEvent] = [] + async for ev in bridge.watch_changes(): + results.append(ev) + break # Only one event expected + assert len(results) == 1 + assert results[0].kind == "commands_changed" + + +@pytest.mark.unit +async def test_watch_changes_skills_changed_forwarded() -> None: + """'skills_changed' event is forwarded by watch_changes().""" + event = ChangeEvent( + capability_name="cap", + kind="skills_changed", + source_uri="skill://cap", + ) + obs = FakeChangeObservable("obs", [event]) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(obs, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + results: list[ChangeEvent] = [] + async for ev in bridge.watch_changes(): + results.append(ev) + break + assert len(results) == 1 + assert results[0].kind == "skills_changed" + + +@pytest.mark.unit +async def test_watch_changes_prompts_changed_forwarded() -> None: + """'prompts_changed' event is forwarded by watch_changes().""" + event = ChangeEvent( + capability_name="cap", + kind="prompts_changed", + source_uri="mcp://cap", + ) + obs = FakeChangeObservable("obs", [event]) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(obs, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + results: list[ChangeEvent] = [] + async for ev in bridge.watch_changes(): + results.append(ev) + break + assert len(results) == 1 + assert results[0].kind == "prompts_changed" + + +@pytest.mark.unit +async def test_watch_changes_tools_changed_filtered_out() -> None: + """'tools_changed' event is filtered out (not forwarded).""" + tool_event = ChangeEvent( + capability_name="cap", + kind="tools_changed", + source_uri="mcp://cap", + ) + cmd_event = ChangeEvent( + capability_name="cap", + kind="commands_changed", + source_uri="skill://cap", + ) + obs = FakeChangeObservable("obs", [tool_event, cmd_event]) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + registry.register(obs, scope) # type: ignore[arg-type] + + bridge = CommandBridge(registry, scope) + results = [ev async for ev in bridge.watch_changes()] + # tools_changed should be filtered, only commands_changed forwarded. + assert len(results) == 1 + assert results[0].kind == "commands_changed" + + +@pytest.mark.unit +async def test_watch_changes_merge_returns_none_empty_iterator() -> None: + """merge_change_streams returns None → empty async iterator (no items yielded).""" + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + bridge = CommandBridge(registry, scope) + results = [ev async for ev in bridge.watch_changes()] + assert results == [] + + +# ===================================================================== +# 5. entry_to_slashed_command() tests (task 2.9) +# ===================================================================== + + +@pytest.mark.unit +async def test_entry_to_slashed_command_with_handler() -> None: + """Entry with handler → returns SlashedCommand (not None).""" + from slashed import Command as SlashedCommand + + async def handler(input_text: str, ctx: AgentContext) -> str: + return "ok" + + entry = CommandEntry( + name="my-cmd", + description="A command", + handler=handler, + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + bridge = CommandBridge(registry, scope) + + result = CommandBridge.entry_to_slashed_command(entry, bridge) + assert result is not None + assert isinstance(result, SlashedCommand) + assert result.name == "my-cmd" + + +@pytest.mark.unit +async def test_entry_to_slashed_command_without_handler_returns_none() -> None: + """Entry without handler (handler=None) → returns None.""" + entry = CommandEntry( + name="display-only", + description="No handler", + ) + registry = ExtensionRegistry() + scope = Scope(level=ScopeLevel.POOL) + bridge = CommandBridge(registry, scope) + + result = CommandBridge.entry_to_slashed_command(entry, bridge) + assert result is None + + +# ===================================================================== +# 6. SkillManagerCap handler tests (task 3.3) +# ===================================================================== + + +@pytest.mark.unit +async def test_skill_manager_cap_list_commands_returns_entries_with_handlers() -> None: + """SkillManagerCap.list_commands() returns entries with callable handlers.""" + skill = Skill( + name="my-skill", + description="A test skill", + skill_path=PurePosixPath("skill://local/my-skill"), + instructions="Skill instructions here.", + ) + cap = SkillManagerCap(local_skills={"my-skill": skill}) + commands = await cap.list_commands() + assert len(commands) == 1 + entry = commands[0] + assert entry.name == "my-skill" + assert entry.description == "A test skill" + assert entry.source == "local" + assert entry.skill_uri == "skill://my-skill" + assert entry.handler is not None + assert callable(entry.handler) + + +@pytest.mark.unit +async def test_skill_manager_cap_handler_loads_skill_and_concatenates_input() -> None: + """Handler loads skill content and concatenates with input when invoked.""" + skill = Skill( + name="concat-skill", + description="Concatenation test", + skill_path=PurePosixPath("skill://local/concat-skill"), + instructions="INSTRUCTIONS_CONTENT", + ) + cap = SkillManagerCap(local_skills={"concat-skill": skill}) + commands = await cap.list_commands() + assert len(commands) == 1 + entry = commands[0] + assert entry.handler is not None + ctx = _make_agent_context() + result = await entry.handler("USER_INPUT", ctx) # type: ignore[misc] + assert "INSTRUCTIONS_CONTENT" in result + assert "USER_INPUT" in result + # Instructions come first, then user input. + assert result.index("INSTRUCTIONS_CONTENT") < result.index("USER_INPUT") + + +# ===================================================================== +# 7. McpServerCap handler tests (task 3.4) +# ===================================================================== + + +class MockMcpClient: + """Mock MCPClient for testing McpServerCap without real connections.""" + + def __init__( + self, + prompts: list[Any] | None = None, + prompt_results: dict[str, Any] | None = None, + ) -> None: + self._prompts = prompts or [] + self._prompt_results = prompt_results or {} + + async def list_prompts(self) -> list[Any]: + return list(self._prompts) + + async def get_prompt(self, name: str, arguments: dict[str, str] | None) -> Any: + result = self._prompt_results.get(name) + if result is None: + # Return empty messages + return MagicMock(messages=[]) + return result + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *args: object) -> None: + pass + + +def _make_mcp_config(client_id: str = "test-server") -> MagicMock: + """Create a fake MCP server config for McpServerCap.""" + config = MagicMock() + config.client_id = client_id + return config + + +@pytest.mark.unit +async def test_mcp_server_cap_list_commands_returns_entries_with_handlers() -> None: + """McpServerCap.list_commands() returns entries with callable handlers.""" + from agentpool.capabilities.mcp_server_cap import McpServerCap + + # Create a mock prompt object. + mock_prompt = MagicMock() + mock_prompt.name = "greet" + mock_prompt.description = "A greeting prompt" + + # Create a mock prompt result. + mock_content = MagicMock() + mock_content.text = "Hello from MCP!" + mock_message = MagicMock() + mock_message.content = mock_content + mock_result = MagicMock() + mock_result.messages = [mock_message] + + mock_client = MockMcpClient( + prompts=[mock_prompt], + prompt_results={"greet": mock_result}, + ) + + cap = McpServerCap(config=_make_mcp_config(), client=cast("MCPClient", mock_client)) + + commands = await cap.list_commands() + assert len(commands) == 1 + entry = commands[0] + assert entry.name == "greet" + assert entry.description == "A greeting prompt" + assert entry.source == "remote" + assert entry.handler is not None + assert callable(entry.handler) + + +@pytest.mark.unit +async def test_mcp_server_cap_handler_calls_get_prompt() -> None: + """Handler calls get_prompt(name, arguments) when invoked.""" + from agentpool.capabilities.mcp_server_cap import McpServerCap + + mock_prompt = MagicMock() + mock_prompt.name = "summarize" + mock_prompt.description = "Summarize text" + + mock_content = MagicMock() + mock_content.text = "Summary result" + mock_message = MagicMock() + mock_message.content = mock_content + mock_result = MagicMock() + mock_result.messages = [mock_message] + + mock_client = MockMcpClient( + prompts=[mock_prompt], + prompt_results={"summarize": mock_result}, + ) + + cap = McpServerCap(config=_make_mcp_config(), client=cast("MCPClient", mock_client)) + + commands = await cap.list_commands() + assert len(commands) == 1 + entry = commands[0] + assert entry.handler is not None + ctx = _make_agent_context() + result = await entry.handler("some text", ctx) # type: ignore[misc] + assert result == "Summary result" + + +# ===================================================================== +# 8. SkillManagerCap pass-through test (task 3.5) +# ===================================================================== + + +class MockCommandResourceForPassthrough: + """Mock McpServerCap-like child that implements CommandResource.""" + + def __init__(self) -> None: + self._name = "mock-mcp" + + async def _remote_handler(input_text: str, ctx: AgentContext) -> str: + return "remote-result" + + self._commands = [ + CommandEntry( + name="remote-cmd", + description="Remote command", + skill_uri="skill://mock-mcp/remote-cmd", + source="remote", + handler=_remote_handler, + ) + ] + + def get_serialization_name(self) -> str: + return self._name + + def get_toolset(self) -> Any: + return None + + def get_instructions(self) -> str | None: + return None + + async def list_commands(self) -> list[CommandEntry]: + return list(self._commands) + + async def get_command(self, name: str) -> CommandEntry | None: + return next((c for c in self._commands if c.name == name), None) + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *args: object) -> None: + pass + + +@pytest.mark.unit +async def test_skill_manager_cap_passes_through_mcp_commands_unchanged() -> None: + """SkillManagerCap passes through McpServerCap commands unchanged.""" + child = MockCommandResourceForPassthrough() + # The child's original handler object. + original_handler = child._commands[0].handler + + cap = SkillManagerCap( + local_skills={}, + children=[child], # type: ignore[list-item] + ) + commands = await cap.list_commands() + # Should have the remote command passed through. + assert len(commands) == 1 + entry = commands[0] + assert entry.name == "remote-cmd" + assert entry.source == "remote" + # The handler should be the same object, not re-wrapped. + assert entry.handler is original_handler + + +# ===================================================================== +# 9. Backward compatibility test (task 3.6) +# ===================================================================== + + +@pytest.mark.unit +async def test_command_entry_without_handler_still_works() -> None: + """Existing code that creates CommandEntry without handler still works.""" + entry = CommandEntry( + name="legacy-cmd", + description="A legacy display-only command", + skill_uri="skill://legacy", + source="local", + ) + assert entry.handler is None + assert entry.name == "legacy-cmd" + assert entry.description == "A legacy display-only command" + assert entry.skill_uri == "skill://legacy" + assert entry.source == "local" + + +@pytest.mark.unit +async def test_command_entry_equality_without_handler() -> None: + """CommandEntry equality without handler works as before.""" + entry_a = CommandEntry( + name="cmd", + description="desc", + skill_uri="skill://cmd", + source="local", + ) + entry_b = CommandEntry( + name="cmd", + description="desc", + skill_uri="skill://cmd", + source="local", + ) + assert entry_a == entry_b + + # Different name → not equal. + entry_c = CommandEntry(name="other", description="desc") + assert entry_a != entry_c