feat(drone): rename routing → drone and add CLI entry point

BREAKING: aipass.routing renamed to aipass.drone. Routing IS drone —
the command routing engine IS the CLI command center. No reason to
keep them separate.

Adds `drone` CLI entry point:
  drone systems          — list registered branches
  drone @branch command  — route command to branch
  drone @branch --help   — get branch help
  drone --version        — show version

406 tests pass (20 new CLI tests). All imports updated.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
AIOSAI
2026-03-02 22:14:32 -08:00
co-authored by Claude Opus 4.6
parent a46c0e150c
commit 866b0596aa
15 changed files with 438 additions and 29 deletions
+1
View File
@@ -44,6 +44,7 @@ dev = [
]
[project.scripts]
drone = "aipass.drone.cli:main"
seedgo = "seedgo.cli:main"
[tool.hatch.build.targets.wheel]
+1 -1
View File
@@ -1,7 +1,7 @@
"""AIPass — Orchestration framework for autonomous AI agent ecosystems.
Modules:
aipass.routing — Command routing & discovery (v1.0.0 — BUILT)
aipass.drone — Command routing & discovery (v1.0.0 — BUILT)
aipass.paths — Root resolution, AIPASS_ROOT (PLACEHOLDER)
aipass.prax — Visibility & monitoring layer (PLACEHOLDER)
aipass.cortex — Branch creation & templates (PLACEHOLDER)
@@ -1,5 +1,5 @@
"""
AIPass Routing & Discovery Module.
AIPass Drone — Command routing & discovery module.
Provides symbolic addressing for multi-agent systems. Resolves @branch names
to absolute paths at runtime.
@@ -26,17 +26,17 @@ Registry Management:
get_registry_path() -> Path # Get current registry path
Example:
>>> from aipass.routing import resolve_branch, register_branch
>>> from aipass.drone import resolve_branch, register_branch
>>> register_branch("my_agent", "/path/to/agent", "agent")
>>> path = resolve_branch("@my_agent")
>>> print(path)
/path/to/agent
>>> from aipass.routing import route_command
>>> from aipass.drone import route_command
>>> result = route_command("@my_agent", "status")
>>> print(result.stdout)
>>> from aipass.routing import route_all
>>> from aipass.drone import route_all
>>> results = route_all("status")
>>> for branch, r in results.items():
... print(f"{branch}: exit={r.exit_code}")
+135
View File
@@ -0,0 +1,135 @@
"""
Drone CLI — command-line interface for aipass.drone.
Entry point: `drone` (wired via pyproject.toml console_scripts to cli.main).
Usage:
drone Show available commands
drone --help Show help
drone --version Show version
drone systems List registered branches
drone @branch command [args] Route command to branch
drone @branch --help Show help for branch
Zero external dependencies — stdlib only (sys).
"""
from __future__ import annotations
import sys
import aipass
from aipass.drone import (
CommandExecutionError,
BranchNotFoundError,
get_help,
list_branches,
route_command,
)
_HELP_TEXT = """\
drone — AIPass branch command router
Usage:
drone Show this help
drone --help Show this help
drone --version Show version
drone systems List registered branches
drone @branch command [args] Route command to branch
drone @branch --help Show help for branch
Examples:
drone systems
drone @flow status
drone @flow run --verbose
drone @flow --help
"""
def main() -> None:
"""Entry point for the `drone` CLI command.
Parses sys.argv manually and dispatches to the appropriate handler.
Exits with the command's exit code on branch commands, 0 on success,
1 on error.
"""
args = sys.argv[1:]
# No args or explicit --help
if not args or args == ["--help"]:
print(_HELP_TEXT, end="")
sys.exit(0)
# --version
if args == ["--version"]:
print(aipass.__version__)
sys.exit(0)
# systems — list registered branches
if args[0] == "systems":
_cmd_systems()
return
# @branch ... — route to a branch
if args[0].startswith("@"):
_cmd_branch(args)
return
# Unknown command
print(f"drone: unknown command '{args[0]}'", file=sys.stderr)
print("Run 'drone --help' for usage.", file=sys.stderr)
sys.exit(1)
def _cmd_systems() -> None:
"""Handle `drone systems` — list registered branches."""
branches = list_branches()
if not branches:
print("No branches registered.")
sys.exit(0)
print(f"Registered branches ({len(branches)}):")
for name in sorted(branches):
print(f" {name}")
sys.exit(0)
def _cmd_branch(args: list[str]) -> None:
"""Handle `drone @branch command [args]` or `drone @branch --help`."""
target = args[0]
rest = args[1:]
# drone @branch --help
if not rest or rest == ["--help"]:
try:
result = get_help(target)
if result.text:
print(result.text, end="")
else:
print(f"No help available for {target}.")
except BranchNotFoundError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
except CommandExecutionError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
sys.exit(0)
# drone @branch command [args...]
command = rest[0]
cmd_args = rest[1:]
try:
result = route_command(target, command, args=cmd_args if cmd_args else None)
except BranchNotFoundError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
except CommandExecutionError as exc:
print(f"drone: {exc}", file=sys.stderr)
sys.exit(1)
if result.stdout:
print(result.stdout, end="")
if result.stderr:
print(result.stderr, end="", file=sys.stderr)
sys.exit(result.exit_code)
@@ -1,5 +1,5 @@
"""
Routing module custom exceptions.
Drone module custom exceptions.
Defines the exception hierarchy for routing and branch resolution errors.
"""
@@ -79,7 +79,7 @@ def save_registry(registry: Dict[str, Any]) -> None:
registry["metadata"] = {}
registry["metadata"]["last_updated"] = datetime.now(timezone.utc).isoformat()
registry["metadata"]["managed_by"] = "aipass.routing"
registry["metadata"]["managed_by"] = "aipass.drone"
try:
with open(registry_path, "w", encoding="utf-8") as f:
@@ -107,7 +107,7 @@ def initialize_registry() -> None:
"branches": {},
"metadata": {
"last_updated": datetime.now(timezone.utc).isoformat(),
"managed_by": "aipass.routing",
"managed_by": "aipass.drone",
},
}
@@ -1,5 +1,5 @@
"""
Command routing logic for the AIPass routing module.
Command routing logic for the AIPass drone module.
Routes commands to branch entry points by resolving symbolic @branch names,
locating the branch's apps/{name}.py entry point, and executing via subprocess.
+273
View File
@@ -0,0 +1,273 @@
"""
Tests for the drone CLI (aipass.drone.cli).
Covers:
- drone / drone --help → shows help text
- drone --version → shows version
- drone systems → lists branches
- drone @branch command → routes correctly
- drone @branch --help → shows branch help
- unknown commands → error message + exit 1
- no args → shows help
"""
from __future__ import annotations
from unittest.mock import patch
import aipass
from aipass.drone.cli import main
from aipass.drone import CommandResult, HelpResult
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _run_cli(*args: str) -> tuple[int, str, str]:
"""
Call main() with the given argv, capturing stdout/stderr and exit code.
Returns:
(exit_code, stdout_text, stderr_text)
"""
stdout_lines: list[str] = []
stderr_lines: list[str] = []
class _Cap:
def __init__(self, buf: list[str]) -> None:
self._buf = buf
def write(self, s: str) -> int:
self._buf.append(s)
return len(s)
def flush(self) -> None:
pass
exit_code = 0
def _exit(code: int = 0) -> None:
nonlocal exit_code
exit_code = code
raise SystemExit(code)
with (
patch("sys.argv", ["drone", *args]),
patch("sys.stdout", _Cap(stdout_lines)),
patch("sys.stderr", _Cap(stderr_lines)),
patch("sys.exit", side_effect=_exit),
):
try:
main()
except SystemExit:
pass
return exit_code, "".join(stdout_lines), "".join(stderr_lines)
# ---------------------------------------------------------------------------
# Help / version
# ---------------------------------------------------------------------------
class TestHelpAndVersion:
"""drone --help and drone --version."""
def test_no_args_shows_help(self):
"""drone with no args prints help and exits 0."""
code, out, _ = _run_cli()
assert code == 0
assert "drone" in out
assert "Usage" in out or "usage" in out.lower() or "@branch" in out
def test_help_flag_shows_help(self):
"""drone --help prints help and exits 0."""
code, out, _ = _run_cli("--help")
assert code == 0
assert "drone" in out
def test_version_flag_shows_version(self):
"""drone --version prints the aipass version and exits 0."""
code, out, _ = _run_cli("--version")
assert code == 0
assert aipass.__version__ in out
# ---------------------------------------------------------------------------
# systems
# ---------------------------------------------------------------------------
class TestSystems:
"""drone systems — list registered branches."""
def test_systems_lists_branches(self):
"""drone systems calls list_branches and prints them."""
with patch("aipass.drone.cli.list_branches", return_value=["@flow", "@devpulse"]):
code, out, _ = _run_cli("systems")
assert code == 0
assert "@flow" in out
assert "@devpulse" in out
def test_systems_empty_registry(self):
"""drone systems with no branches prints a 'no branches' message."""
with patch("aipass.drone.cli.list_branches", return_value=[]):
code, out, _ = _run_cli("systems")
assert code == 0
assert "No branches" in out or "no branches" in out.lower() or out.strip() != ""
def test_systems_shows_count(self):
"""drone systems output mentions the number of branches."""
branches = ["@alpha", "@beta", "@gamma"]
with patch("aipass.drone.cli.list_branches", return_value=branches):
code, out, _ = _run_cli("systems")
assert code == 0
# Should show 3 branches somewhere in output
assert "3" in out or all(b in out for b in branches)
# ---------------------------------------------------------------------------
# @branch routing
# ---------------------------------------------------------------------------
class TestBranchRouting:
"""drone @branch command [args] — route commands."""
def _make_result(self, stdout="", stderr="", exit_code=0) -> CommandResult:
return CommandResult(stdout=stdout, stderr=stderr, exit_code=exit_code, branch="flow", command="status")
def test_routes_command_to_branch(self):
"""drone @flow status calls route_command with correct args."""
result = self._make_result(stdout="flow is running\n")
with patch("aipass.drone.cli.route_command", return_value=result) as mock_route:
code, out, _ = _run_cli("@flow", "status")
mock_route.assert_called_once_with("@flow", "status", args=None)
assert code == 0
assert "flow is running" in out
def test_routes_command_with_extra_args(self):
"""drone @flow run --verbose passes args correctly."""
result = self._make_result(stdout="verbose output\n")
with patch("aipass.drone.cli.route_command", return_value=result) as mock_route:
code, out, _ = _run_cli("@flow", "run", "--verbose")
mock_route.assert_called_once_with("@flow", "run", args=["--verbose"])
assert code == 0
def test_exits_with_command_exit_code(self):
"""drone exits with the same code as the routed command."""
result = self._make_result(exit_code=2, stderr="something failed\n")
with patch("aipass.drone.cli.route_command", return_value=result):
code, _, err = _run_cli("@flow", "status")
assert code == 2
assert "something failed" in err
def test_branch_not_found_exits_1(self):
"""BranchNotFoundError prints error and exits 1."""
from aipass.drone import BranchNotFoundError
with patch("aipass.drone.cli.route_command", side_effect=BranchNotFoundError("@ghost not found")):
code, _, err = _run_cli("@ghost", "status")
assert code == 1
assert "ghost" in err or "not found" in err.lower()
def test_command_execution_error_exits_1(self):
"""CommandExecutionError prints error and exits 1."""
from aipass.drone import CommandExecutionError
with patch("aipass.drone.cli.route_command", side_effect=CommandExecutionError("entry point missing")):
code, _, err = _run_cli("@flow", "status")
assert code == 1
assert "entry point missing" in err
def test_stdout_printed(self):
"""Command stdout is printed to stdout."""
result = self._make_result(stdout="hello world\n")
with patch("aipass.drone.cli.route_command", return_value=result):
code, out, _ = _run_cli("@flow", "status")
assert "hello world" in out
def test_stderr_printed_to_stderr(self):
"""Command stderr is printed to stderr."""
result = self._make_result(stderr="warning: something\n", exit_code=0)
with patch("aipass.drone.cli.route_command", return_value=result):
code, _, err = _run_cli("@flow", "status")
assert "warning: something" in err
# ---------------------------------------------------------------------------
# @branch --help
# ---------------------------------------------------------------------------
class TestBranchHelp:
"""drone @branch --help — show branch help."""
def test_branch_help_calls_get_help(self):
"""drone @flow --help calls get_help and prints text."""
help_result = HelpResult(branch="flow", command=None, text="flow help text\n", commands_found=["status"])
with patch("aipass.drone.cli.get_help", return_value=help_result) as mock_help:
code, out, _ = _run_cli("@flow", "--help")
mock_help.assert_called_once_with("@flow")
assert code == 0
assert "flow help text" in out
def test_branch_no_args_shows_help(self):
"""drone @flow with no command also calls get_help."""
help_result = HelpResult(branch="flow", command=None, text="flow help\n", commands_found=[])
with patch("aipass.drone.cli.get_help", return_value=help_result):
code, out, _ = _run_cli("@flow")
assert code == 0
assert "flow help" in out
def test_branch_not_found_on_help_exits_1(self):
"""BranchNotFoundError on --help exits 1."""
from aipass.drone import BranchNotFoundError
with patch("aipass.drone.cli.get_help", side_effect=BranchNotFoundError("@ghost not found")):
code, _, err = _run_cli("@ghost", "--help")
assert code == 1
def test_command_execution_error_on_help_exits_1(self):
"""CommandExecutionError on --help exits 1."""
from aipass.drone import CommandExecutionError
with patch("aipass.drone.cli.get_help", side_effect=CommandExecutionError("no entry")):
code, _, err = _run_cli("@flow", "--help")
assert code == 1
# ---------------------------------------------------------------------------
# Unknown commands
# ---------------------------------------------------------------------------
class TestUnknownCommands:
"""Unknown top-level commands show an error."""
def test_unknown_command_exits_1(self):
"""An unrecognised command exits with code 1."""
code, _, err = _run_cli("frobulate")
assert code == 1
def test_unknown_command_shows_error_message(self):
"""An unrecognised command prints an error to stderr."""
code, _, err = _run_cli("frobulate")
assert "frobulate" in err or "unknown" in err.lower()
def test_unknown_command_suggests_help(self):
"""Error message mentions --help."""
code, _, err = _run_cli("frobulate")
assert "--help" in err
@@ -1,5 +1,5 @@
"""
Unit tests for routing core functionality.
Unit tests for drone core functionality.
Tests all Phase 1 routing functions with >80% coverage target.
"""
@@ -9,7 +9,7 @@ from pathlib import Path
import pytest
from aipass.routing import (
from aipass.drone import (
BranchAlreadyExistsError,
BranchNotFoundError,
InvalidPathError,
@@ -1,5 +1,5 @@
"""
Integration tests for the AIPass routing module (Phase 3).
Integration tests for the AIPass drone module (Phase 3).
End-to-end tests that exercise the full workflow using real temp directories
with proper branch file structures. All subprocess calls are mocked so no
@@ -13,7 +13,7 @@ from unittest.mock import MagicMock, patch
import pytest
from aipass.routing import (
from aipass.drone import (
BranchNotFoundError,
CommandExecutionError,
CommandResult,
@@ -31,7 +31,7 @@ from aipass.routing import (
route_command,
set_registry_path,
)
from aipass.routing.discovery import get_system_help
from aipass.drone.discovery import get_system_help
# ---------------------------------------------------------------------------
@@ -55,7 +55,7 @@ def _make_branch(root: Path, name: str) -> Path:
Layout:
root/{name}/
apps/
{name}.py ← entry point
{name}.py <- entry point
modules/
status.py
info.py
@@ -237,8 +237,8 @@ class TestHelpResult:
assert result.commands_found == []
def test_help_result_importable_from_package(self):
"""HelpResult is importable directly from aipass.routing."""
from aipass.routing import HelpResult as HR
"""HelpResult is importable directly from aipass.drone."""
from aipass.drone import HelpResult as HR
assert HR is HelpResult
@@ -325,8 +325,8 @@ class TestRouteAll:
assert kwargs.get("timeout") == 5
def test_route_all_importable_from_package(self):
"""route_all is importable from aipass.routing."""
from aipass.routing import route_all as ra
"""route_all is importable from aipass.drone."""
from aipass.drone import route_all as ra
assert callable(ra)
@@ -479,7 +479,7 @@ class TestRegistryPersistence:
data = json.load(fh)
assert "last_updated" in data["metadata"]
assert data["metadata"]["managed_by"] == "aipass.routing"
assert data["metadata"]["managed_by"] == "aipass.drone"
reset_registry_path()
@@ -1,5 +1,5 @@
"""
Unit tests for Phase 2 routing functionality.
Unit tests for Phase 2 drone functionality.
Tests command routing, subprocess execution, and module discovery with >80%
coverage on new code. All subprocess calls are mocked — no real processes run.
@@ -10,7 +10,7 @@ from unittest.mock import MagicMock, patch
import pytest
from aipass.routing import (
from aipass.drone import (
BranchNotFoundError,
CommandExecutionError,
CommandResult,
@@ -22,8 +22,8 @@ from aipass.routing import (
route_command,
set_registry_path,
)
from aipass.routing.executor import execute_command
from aipass.routing.router import _find_entry_point
from aipass.drone.executor import execute_command
from aipass.drone.router import _find_entry_point
# ---------------------------------------------------------------------------
@@ -526,21 +526,21 @@ class TestPublicAPIImports:
"""Verify all Phase 2 symbols are importable from the top-level package."""
def test_route_command_importable(self):
from aipass.routing import route_command
from aipass.drone import route_command
assert callable(route_command)
def test_discover_modules_importable(self):
from aipass.routing import discover_modules
from aipass.drone import discover_modules
assert callable(discover_modules)
def test_get_help_importable(self):
from aipass.routing import get_help
from aipass.drone import get_help
assert callable(get_help)
def test_command_result_importable(self):
from aipass.routing import CommandResult
from aipass.drone import CommandResult
assert CommandResult is not None
def test_version_updated(self):
from aipass.routing import __version__
from aipass.drone import __version__
assert __version__ == "1.0.0"