feat(drone): 188-test suite, interactive backup commands, type fix (#46)

* feat(backup): stderr routing, error masking fix, mtime pre-check, ignore patterns split

- Route all error/warning prints to stderr via error()/warning() (backup_core, google_drive_sync)
- Fix route_command error masking — show real errors instead of "Unknown command"
- Add versioned backup mtime pre-check to skip unchanged files
- Extract ignore patterns from config_handler to editable JSON (FPLAN-0037)
  - New: ignore_patterns.py loads from json_templates/ignore_patterns.json
  - config_handler.py v3.0.0 re-exports for backwards compatibility
  - 132 patterns, 14/14 tests pass

Co-Authored-By: @backup <backup@aipass>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(prax): dashboard cleanup — remove devpulse section, fix broken paths

Phase 1 dashboard cleanup (FPLAN-0040):
- Remove devpulse section from template, all hardcoded defaults, and refresh pipeline
- Fix broken paths: old aipass_os/dev_central/devpulse → prax via Path(__file__)
- Fix registry: BRANCH_REGISTRY.json → AIPASS_REGISTRY.json with repo root walk-up
- Fix relative path resolution: registry stores relative paths, now resolved against repo root
- Fix agent_status_writer subprocess import: aipass.devpulse → aipass.prax
- Update help text and docstrings: drone @devpulse → drone @prax
- Add devpulse to DEPRECATED_SECTIONS in template_pusher and differ
- Push-template verified: all 15 branches updated, devpulse section removed

Co-Authored-By: @prax <prax@aipass>

* feat(drone): add 188-test suite, interactive commands for backup, type fix

- Add 5 test files (188 tests) covering executor, resolver, router,
  registry_handler, and discovery — built via 3-round audit process
- Add snapshot/versioned to interactive command tuple for backup Rich output
- Fix type error in router_handler.py:119 (command: Optional[str] → str)
- Fix pytest.ini inline comments breaking pytest 9.x
- Document interactive commands in README

Co-Authored-By: @drone <drone@aipass>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: @backup <backup@aipass>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: @prax <prax@aipass>
Co-authored-by: @drone <drone@aipass>

src/aipass/drone/README.md

@@ -99,6 +99,24 @@ drone/
 
 ---
 
+## Interactive Commands
+
+By default, drone captures subprocess output (`capture_output=True`) with a 30s timeout. This is safe for AI-to-AI routing but strips Rich colors, buffers progress bars, and kills long-running commands.
+
+Commands in the interactive tuple bypass capture and inherit the terminal directly — enabling live Rich output, colors, and no timeout. Only add commands here when Patrick needs full terminal experience.
+
+**Current interactive commands** (in `apps/drone.py`):
+
+| Command      | Reason                                      |
+|--------------|---------------------------------------------|
+| `monitor`    | Prax real-time monitoring (live TUI)        |
+| `snapshot`   | Backup snapshot (Rich progress bars)        |
+| `versioned`  | Backup versioned (Rich progress, long-running) |
+
+To add a new command: edit the `interactive` tuple in `_handle_target()` in `apps/drone.py`.
+
+---
+
 ## Integration Points
 
 ### Depends On
@@ -113,4 +131,4 @@ drone/
 
 ---
 
-**Last Updated:** 2026-03-08
+**Last Updated:** 2026-03-14

src/aipass/drone/apps/drone.py

@@ -210,7 +210,7 @@ def _handle_target(args: List[str]) -> int:
     cmd_args = rest[1:]
 
     # Long-running interactive commands bypass capture + timeout
-    interactive = command in ("monitor",)
+    interactive = command in ("monitor", "snapshot", "versioned")
 
     try:
         result = route_command(

src/aipass/drone/apps/handlers/router_handler.py

@@ -116,5 +116,5 @@ def execute_branch_command(
         stderr=result.stderr,
         exit_code=result.exit_code,
         branch=branch_name,
-        command=command,
+        command=command or "",
     )

src/aipass/drone/pytest.ini

@@ -9,10 +9,10 @@ python_classes = Test*
 
 # Command-line options (always applied)
 addopts =
-    -v                          # Verbose output
-    --tb=short                  # Short traceback format
-    --strict-markers            # Raise error on unknown markers
-    -ra                         # Show summary of all test outcomes
+    -v
+    --tb=short
+    --strict-markers
+    -ra
 
 # Test markers (for categorizing tests)
 markers =

src/aipass/drone/tests/test_discovery.py

@@ -0,0 +1,640 @@
+# =================== AIPass ====================
+# Name: test_discovery.py
+# Description: Tests for module and command discovery
+# Version: 1.0.0
+# Created: 2026-03-14
+# Modified: 2026-03-14
+# =============================================
+
+"""Tests for module and command discovery.
+
+Covers handler-layer functions (scan_modules_directory, parse_help_for_commands,
+get_help, get_module_introspective) and orchestration-layer functions
+(discover_modules, get_help, get_system_help) from the discovery module.
+"""
+
+import subprocess
+import types
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from aipass.drone.apps.handlers.discovery_handler import (
+    HelpResult,
+    scan_modules_directory,
+    parse_help_for_commands,
+    get_help as handler_get_help,
+    get_entry_point,
+    discover_modules as handler_discover_modules,
+)
+from aipass.drone.apps.handlers.exceptions import (
+    BranchNotFoundError,
+    CommandExecutionError,
+)
+from aipass.drone.apps.handlers.module_registry_handler import (
+    get_module_introspective,
+)
+
+
+# =============================================================================
+# scan_modules_directory tests
+# =============================================================================
+
+class TestScanModulesDirectory:
+    """Tests for scan_modules_directory()."""
+
+    def test_finds_py_files(self, temp_test_dir: Path):
+        """Should return stems of .py files in apps/modules/."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+        (modules_dir / "alpha.py").write_text("# alpha", encoding="utf-8")
+        (modules_dir / "beta.py").write_text("# beta", encoding="utf-8")
+        (modules_dir / "gamma.py").write_text("# gamma", encoding="utf-8")
+
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert result == ["alpha", "beta", "gamma"]
+
+    def test_skips_init_and_main(self, temp_test_dir: Path):
+        """Should exclude __init__.py and __main__.py from results."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+        (modules_dir / "__init__.py").write_text("", encoding="utf-8")
+        (modules_dir / "__main__.py").write_text("", encoding="utf-8")
+        (modules_dir / "real_module.py").write_text("# code", encoding="utf-8")
+
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert "__init__" not in result
+        assert "__main__" not in result
+        assert result == ["real_module"]
+
+    def test_ignores_pycache_and_non_py(self, temp_test_dir: Path):
+        """Should ignore __pycache__ directories and non-.py files."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+        (modules_dir / "__pycache__").mkdir()
+        (modules_dir / "__pycache__" / "cached.pyc").write_bytes(b"\x00")
+        (modules_dir / "notes.txt").write_text("not python", encoding="utf-8")
+        (modules_dir / "data.json").write_text("{}", encoding="utf-8")
+        (modules_dir / "valid.py").write_text("# ok", encoding="utf-8")
+
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert result == ["valid"]
+
+    def test_returns_empty_for_empty_directory(self, temp_test_dir: Path):
+        """Should return empty list when apps/modules/ exists but has no .py files."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert result == []
+
+    def test_returns_empty_when_no_modules_dir(self, temp_test_dir: Path):
+        """Should return empty list when apps/modules/ does not exist."""
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert result == []
+
+    def test_results_are_sorted(self, temp_test_dir: Path):
+        """Returned module names should be sorted alphabetically."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+        for name in ["zebra", "apple", "mango"]:
+            (modules_dir / f"{name}.py").write_text("", encoding="utf-8")
+
+        result = scan_modules_directory(str(temp_test_dir))
+
+        assert result == sorted(result)
+        assert result == ["apple", "mango", "zebra"]
+
+
+# =============================================================================
+# parse_help_for_commands tests
+# =============================================================================
+
+class TestParseHelpForCommands:
+    """Tests for parse_help_for_commands()."""
+
+    def test_extracts_commands_from_commands_section(self):
+        """Should extract command names from a 'Commands:' section."""
+        help_text = (
+            "Usage: tool [OPTIONS] COMMAND\n"
+            "\n"
+            "Commands:\n"
+            "  init        Initialize a project\n"
+            "  build       Build the project\n"
+            "  deploy      Deploy to production\n"
+            "\n"
+            "Options:\n"
+            "  --help      Show this message\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert "init" in result
+        assert "build" in result
+        assert "deploy" in result
+
+    def test_extracts_from_subcommands_section(self):
+        """Should recognize 'Subcommands:' as a section marker."""
+        help_text = (
+            "MyTool v1.0\n"
+            "\n"
+            "Subcommands:\n"
+            "  status      Show status\n"
+            "  config      Configure settings\n"
+            "\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert "status" in result
+        assert "config" in result
+
+    def test_extracts_from_available_commands_section(self):
+        """Should recognize 'Available Commands:' as a section marker."""
+        help_text = (
+            "Usage: app\n"
+            "\n"
+            "Available Commands:\n"
+            "  run         Run the app\n"
+            "  test        Run tests\n"
+            "\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert "run" in result
+        assert "test" in result
+
+    def test_skips_options_flags(self):
+        """Should not include lines starting with - as commands."""
+        help_text = (
+            "Commands:\n"
+            "  start       Start the server\n"
+            "  -v          Verbose mode\n"
+            "  --debug     Debug mode\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert "start" in result
+        assert "-v" not in result
+        assert "--debug" not in result
+
+    def test_empty_string_returns_empty_list(self):
+        """Empty help text should return empty list."""
+        result = parse_help_for_commands("")
+
+        assert result == []
+
+    def test_no_commands_section_returns_empty(self):
+        """Help text without a commands section should return empty list."""
+        help_text = (
+            "Usage: tool [OPTIONS]\n"
+            "\n"
+            "Options:\n"
+            "  --help      Show help\n"
+            "  --version   Show version\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert result == []
+
+    def test_section_ends_at_blank_line(self):
+        """Commands section should end at the first blank line after it."""
+        help_text = (
+            "Commands:\n"
+            "  alpha       First command\n"
+            "  beta        Second command\n"
+            "\n"
+            "Not a command section anymore:\n"
+            "  gamma       Should not appear\n"
+        )
+
+        result = parse_help_for_commands(help_text)
+
+        assert "alpha" in result
+        assert "beta" in result
+        # "gamma" is in a new section after the blank line ended the Commands
+        # section, and "Not a command section anymore:" is not a recognized marker
+        assert "gamma" not in result
+
+
+# =============================================================================
+# get_entry_point tests
+# =============================================================================
+
+class TestGetEntryPoint:
+    """Tests for get_entry_point()."""
+
+    def test_returns_path_when_entry_exists(self, temp_test_dir: Path):
+        """Should return the entry point Path when apps/{name}.py exists."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        entry = apps_dir / "mybranch.py"
+        entry.write_text("# entry", encoding="utf-8")
+
+        result = get_entry_point(str(temp_test_dir), "mybranch")
+
+        assert result is not None
+        assert result == entry
+
+    def test_returns_none_when_missing(self, temp_test_dir: Path):
+        """Should return None when the entry point file does not exist."""
+        result = get_entry_point(str(temp_test_dir), "nonexistent")
+
+        assert result is None
+
+
+# =============================================================================
+# handler get_help tests
+# =============================================================================
+
+class TestHandlerGetHelp:
+    """Tests for discovery_handler.get_help()."""
+
+    def test_returns_help_result_for_valid_entry(self, temp_test_dir: Path):
+        """Should return HelpResult when entry point exists and subprocess runs."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        entry = apps_dir / "testbranch.py"
+        entry.write_text("# entry", encoding="utf-8")
+
+        mock_result = MagicMock()
+        mock_result.stdout = b"Usage: testbranch\n\nCommands:\n  run   Run it\n  stop  Stop it\n"
+        mock_result.stderr = b""
+
+        with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result):
+            result = handler_get_help(str(temp_test_dir), "testbranch")
+
+        assert isinstance(result, HelpResult)
+        assert result.branch == "testbranch"
+        assert result.command is None
+        assert "run" in result.commands_found
+        assert "stop" in result.commands_found
+
+    def test_returns_help_for_specific_command(self, temp_test_dir: Path):
+        """Should pass command argument through and set it on result."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "testbranch.py").write_text("# entry", encoding="utf-8")
+
+        mock_result = MagicMock()
+        mock_result.stdout = b"Help for the run command\n"
+        mock_result.stderr = b""
+
+        with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result):
+            result = handler_get_help(str(temp_test_dir), "testbranch", command="run")
+
+        assert result.command == "run"
+        assert "Help for the run command" in result.text
+
+    def test_raises_when_entry_point_missing(self, temp_test_dir: Path):
+        """Should raise CommandExecutionError when no entry point exists."""
+        with pytest.raises(CommandExecutionError, match="Entry point not found"):
+            handler_get_help(str(temp_test_dir), "nonexistent")
+
+    def test_raises_on_timeout(self, temp_test_dir: Path):
+        """Should raise CommandExecutionError when subprocess times out."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "slowbranch.py").write_text("# entry", encoding="utf-8")
+
+        with patch(
+            "aipass.drone.apps.handlers.discovery_handler.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd="test", timeout=10),
+        ):
+            with pytest.raises(CommandExecutionError, match="timed out"):
+                handler_get_help(str(temp_test_dir), "slowbranch")
+
+    def test_raises_on_oserror(self, temp_test_dir: Path):
+        """Should raise CommandExecutionError when subprocess raises OSError."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "broken.py").write_text("# entry", encoding="utf-8")
+
+        with patch(
+            "aipass.drone.apps.handlers.discovery_handler.subprocess.run",
+            side_effect=OSError("No such file or directory"),
+        ):
+            with pytest.raises(CommandExecutionError, match="OS error"):
+                handler_get_help(str(temp_test_dir), "broken")
+
+    def test_falls_back_to_stderr(self, temp_test_dir: Path):
+        """When stdout is empty, should use stderr for help text."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "testbranch.py").write_text("# entry", encoding="utf-8")
+
+        mock_result = MagicMock()
+        mock_result.stdout = b""
+        mock_result.stderr = b"Usage printed to stderr\n"
+
+        with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result):
+            result = handler_get_help(str(temp_test_dir), "testbranch")
+
+        assert "Usage printed to stderr" in result.text
+
+
+# =============================================================================
+# handler discover_modules tests
+# =============================================================================
+
+class TestHandlerDiscoverModules:
+    """Tests for discovery_handler.discover_modules()."""
+
+    def test_uses_help_parsing_when_entry_point_exists(self, temp_test_dir: Path):
+        """Should prefer parsed help text when entry point exists and returns commands."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
+
+        modules_dir = apps_dir / "modules"
+        modules_dir.mkdir()
+        (modules_dir / "fallback.py").write_text("", encoding="utf-8")
+
+        mock_result = MagicMock()
+        mock_result.stdout = b"Commands:\n  alpha   Do alpha\n  beta    Do beta\n"
+        mock_result.stderr = b""
+
+        with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result):
+            result = handler_discover_modules(str(temp_test_dir), "mybranch")
+
+        assert "alpha" in result
+        assert "beta" in result
+        # fallback module should NOT appear when help parsing succeeded
+        assert "fallback" not in result
+
+    def test_falls_back_to_scan_when_no_entry_point(self, temp_test_dir: Path):
+        """Should fall back to scan_modules_directory when no entry point."""
+        modules_dir = temp_test_dir / "apps" / "modules"
+        modules_dir.mkdir(parents=True)
+        (modules_dir / "scanner.py").write_text("", encoding="utf-8")
+        (modules_dir / "parser.py").write_text("", encoding="utf-8")
+
+        result = handler_discover_modules(str(temp_test_dir), "nonexistent")
+
+        assert result == ["parser", "scanner"]
+
+    def test_falls_back_to_scan_on_timeout(self, temp_test_dir: Path):
+        """Should fall back to scan_modules_directory when subprocess times out."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
+
+        modules_dir = apps_dir / "modules"
+        modules_dir.mkdir()
+        (modules_dir / "fallback_timeout.py").write_text("", encoding="utf-8")
+
+        with patch(
+            "aipass.drone.apps.handlers.discovery_handler.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd="test", timeout=10),
+        ):
+            result = handler_discover_modules(str(temp_test_dir), "mybranch")
+
+        assert result == ["fallback_timeout"]
+
+    def test_falls_back_to_scan_when_help_empty(self, temp_test_dir: Path):
+        """Should fall back to scan when help text yields no commands."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir(parents=True)
+        (apps_dir / "mybranch.py").write_text("# entry", encoding="utf-8")
+
+        modules_dir = apps_dir / "modules"
+        modules_dir.mkdir()
+        (modules_dir / "fallback_cmd.py").write_text("", encoding="utf-8")
+
+        mock_result = MagicMock()
+        mock_result.stdout = b"No commands here, just text.\n"
+        mock_result.stderr = b""
+
+        with patch("aipass.drone.apps.handlers.discovery_handler.subprocess.run", return_value=mock_result):
+            result = handler_discover_modules(str(temp_test_dir), "mybranch")
+
+        assert result == ["fallback_cmd"]
+
+
+# =============================================================================
+# orchestration layer: discovery.discover_modules tests
+# =============================================================================
+
+class TestOrchestrationDiscoverModules:
+    """Tests for discovery.discover_modules() orchestration."""
+
+    @patch("aipass.drone.apps.modules.discovery.resolve_branch")
+    @patch("aipass.drone.apps.handlers.discovery_handler.discover_modules")
+    def test_resolves_and_delegates(self, mock_handler_discover, mock_resolve):
+        """Should resolve branch path then delegate to handler."""
+        from aipass.drone.apps.modules.discovery import discover_modules
+
+        mock_resolve.return_value = "/fake/path/to/branch"
+        mock_handler_discover.return_value = ["cmd_a", "cmd_b"]
+
+        result = discover_modules("@testbranch")
+
+        mock_resolve.assert_called_once_with("@testbranch")
+        mock_handler_discover.assert_called_once_with("/fake/path/to/branch", "testbranch")
+        assert result == ["cmd_a", "cmd_b"]
+
+    @patch("aipass.drone.apps.modules.discovery.resolve_branch")
+    def test_raises_on_invalid_branch(self, mock_resolve):
+        """Should propagate BranchNotFoundError for unknown branches."""
+        from aipass.drone.apps.modules.discovery import discover_modules
+
+        mock_resolve.side_effect = BranchNotFoundError("Branch '@nope' not found")
+
+        with pytest.raises(BranchNotFoundError):
+            discover_modules("@nope")
+
+
+# =============================================================================
+# orchestration layer: discovery.get_help tests
+# =============================================================================
+
+class TestOrchestrationGetHelp:
+    """Tests for discovery.get_help() orchestration."""
+
+    @patch("aipass.drone.apps.modules.discovery.resolve_branch")
+    @patch("aipass.drone.apps.handlers.discovery_handler.get_help")
+    def test_returns_help_result(self, mock_handler_help, mock_resolve):
+        """Should return HelpResult for a valid branch."""
+        from aipass.drone.apps.modules.discovery import get_help
+
+        mock_resolve.return_value = "/fake/branch"
+        mock_handler_help.return_value = HelpResult(
+            branch="valid",
+            command=None,
+            text="Some help text",
+            commands_found=["run"],
+        )
+
+        result = get_help("@valid")
+
+        assert isinstance(result, HelpResult)
+        assert result.branch == "valid"
+        assert result.text == "Some help text"
+
+    @patch("aipass.drone.apps.modules.discovery.resolve_branch")
+    def test_raises_on_invalid_branch(self, mock_resolve):
+        """Should propagate BranchNotFoundError for unknown branches."""
+        from aipass.drone.apps.modules.discovery import get_help
+
+        mock_resolve.side_effect = BranchNotFoundError("not found")
+
+        with pytest.raises(BranchNotFoundError):
+            get_help("@nonexistent")
+
+
+# =============================================================================
+# orchestration layer: discovery.get_system_help tests
+# =============================================================================
+
+class TestOrchestrationGetSystemHelp:
+    """Tests for discovery.get_system_help() orchestration."""
+
+    @patch("aipass.drone.apps.modules.discovery.list_branches")
+    @patch("aipass.drone.apps.handlers.discovery_handler.get_system_help")
+    def test_aggregates_across_branches(self, mock_sys_help, mock_list):
+        """Should pass active branches to handler and return results."""
+        from aipass.drone.apps.modules.discovery import get_system_help
+
+        mock_list.return_value = ["@alpha", "@beta"]
+        mock_sys_help.return_value = {
+            "alpha": HelpResult(branch="alpha", command=None, text="alpha help", commands_found=[]),
+            "beta": HelpResult(branch="beta", command=None, text="beta help", commands_found=[]),
+        }
+
+        result = get_system_help()
+
+        mock_list.assert_called_once_with(status="active")
+        assert "alpha" in result
+        assert "beta" in result
+
+
+# =============================================================================
+# get_module_introspective tests
+# =============================================================================
+
+class TestGetModuleIntrospective:
+    """Tests for module_registry_handler.get_module_introspective()."""
+
+    @patch("aipass.drone.apps.handlers.module_registry_handler._MODULE_REGISTRY", {"testmod": "fake.module.path"})
+    def test_returns_introspective_output(self):
+        """Should call get_introspective() on the module adapter."""
+        fake_mod = types.ModuleType("fake.module.path")
+        fake_mod.get_introspective = lambda: "Introspective info for testmod"  # type: ignore[attr-defined]
+
+        with patch("aipass.drone.apps.handlers.module_registry_handler.importlib.import_module", return_value=fake_mod):
+            result = get_module_introspective("testmod")
+
+        assert result == "Introspective info for testmod"
+
+    @patch("aipass.drone.apps.handlers.module_registry_handler._MODULE_REGISTRY", {"testmod": "fake.module.path"})
+    def test_falls_back_to_help(self):
+        """Should fall back to get_help(None) if get_introspective is missing."""
+        fake_mod = types.ModuleType("fake.module.path")
+        # Only attach get_help, no get_introspective
+        fake_mod.get_help = lambda cmd: "Help fallback text"  # type: ignore[attr-defined]
+
+        with patch("aipass.drone.apps.handlers.module_registry_handler.importlib.import_module", return_value=fake_mod):
+            result = get_module_introspective("testmod")
+
+        assert result == "Help fallback text"
+
+    def test_returns_empty_for_unknown_module(self):
+        """Should return empty string for unregistered module name."""
+        result = get_module_introspective("totally_unknown_module_xyz")
+
+        assert result == ""
+
+
+# =============================================================================
+# handle_command routing tests
+# =============================================================================
+
+class TestHandleCommand:
+    """Tests for discovery.handle_command() routing."""
+
+    @patch("aipass.drone.apps.modules.discovery.discover_modules")
+    def test_modules_command_requires_arg(self, mock_discover):
+        """'modules' command with no args should return False."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        result = handle_command("modules", [])
+
+        assert result is False
+        mock_discover.assert_not_called()
+
+    @patch("aipass.drone.apps.modules.discovery.discover_modules", return_value=["cmd1"])
+    def test_modules_command_succeeds(self, mock_discover):
+        """'modules' command with a target should return True."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        result = handle_command("modules", ["@mybranch"])
+
+        assert result is True
+        mock_discover.assert_called_once_with("@mybranch")
+
+    def test_help_command_requires_arg(self):
+        """'help' command with no args should return False."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        result = handle_command("help", [])
+
+        assert result is False
+
+    @patch("aipass.drone.apps.modules.discovery.get_help")
+    def test_help_command_with_branch(self, mock_get_help):
+        """'help' command with a branch target should return True."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        mock_get_help.return_value = HelpResult(
+            branch="branch", command=None, text="help text", commands_found=[],
+        )
+
+        result = handle_command("help", ["@branch"])
+
+        assert result is True
+        mock_get_help.assert_called_once_with("@branch", None)
+
+    @patch("aipass.drone.apps.modules.discovery.get_help")
+    def test_help_command_with_subcommand(self, mock_get_help):
+        """'help' command with branch and subcommand passes command through."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        mock_get_help.return_value = HelpResult(
+            branch="branch", command="subcmd", text="subcmd help", commands_found=[],
+        )
+
+        result = handle_command("help", ["@branch", "subcmd"])
+
+        assert result is True
+        mock_get_help.assert_called_once_with("@branch", "subcmd")
+
+    @patch("aipass.drone.apps.modules.discovery.get_system_help")
+    def test_system_command(self, mock_sys_help):
+        """'system' command with no args should return True."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        mock_sys_help.return_value = {
+            "alpha": HelpResult(branch="alpha", command=None, text="alpha help", commands_found=[]),
+        }
+
+        result = handle_command("system", [])
+
+        assert result is True
+        mock_sys_help.assert_called_once()
+
+    def test_unknown_command_returns_false(self):
+        """Unknown command should return False."""
+        from aipass.drone.apps.modules.discovery import handle_command
+
+        result = handle_command("nonexistent_command", [])
+
+        assert result is False

src/aipass/drone/tests/test_executor.py

@@ -0,0 +1,403 @@
+"""Tests for the subprocess executor module."""
+
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from aipass.drone.apps.handlers.exceptions import CommandExecutionError
+from aipass.drone.apps.handlers.executor import execute_command
+
+
+# ---------------------------------------------------------------------------
+# 1. Captured mode — returns stdout/stderr as strings
+# ---------------------------------------------------------------------------
+
+
+class TestCapturedMode:
+    """Tests for captured (non-interactive) execution mode."""
+
+    def test_captured_stdout(self, temp_test_dir: Path):
+        """Captured mode returns stdout as a decoded string."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "print('hello world')"],
+            cwd=str(temp_test_dir),
+        )
+        assert result.stdout.strip() == "hello world"
+        assert result.exit_code == 0
+
+    def test_captured_stderr(self, temp_test_dir: Path):
+        """Captured mode returns stderr as a decoded string."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "import sys; sys.stderr.write('err msg\\n')"],
+            cwd=str(temp_test_dir),
+        )
+        assert "err msg" in result.stderr
+        assert result.exit_code == 0
+
+    def test_captured_both_streams(self, temp_test_dir: Path):
+        """Both stdout and stderr are captured simultaneously."""
+        code = (
+            "import sys; "
+            "print('out'); "
+            "sys.stderr.write('err\\n')"
+        )
+        result = execute_command(
+            sys.executable, ["-c", code], cwd=str(temp_test_dir)
+        )
+        assert "out" in result.stdout
+        assert "err" in result.stderr
+
+
+# ---------------------------------------------------------------------------
+# 2. Captured mode — timeout enforcement
+# ---------------------------------------------------------------------------
+
+
+class TestTimeout:
+    """Timeout enforcement in captured mode."""
+
+    def test_timeout_raises_command_execution_error(self, temp_test_dir: Path):
+        """Exceeding the timeout raises CommandExecutionError."""
+        with pytest.raises(CommandExecutionError, match="timed out"):
+            execute_command(
+                sys.executable,
+                ["-c", "import time; time.sleep(10)"],
+                cwd=str(temp_test_dir),
+                timeout=1,
+            )
+
+    def test_timeout_chains_original_exception(self, temp_test_dir: Path):
+        """CommandExecutionError wraps the original TimeoutExpired."""
+        with pytest.raises(CommandExecutionError) as exc_info:
+            execute_command(
+                sys.executable,
+                ["-c", "import time; time.sleep(10)"],
+                cwd=str(temp_test_dir),
+                timeout=1,
+            )
+        assert isinstance(exc_info.value.__cause__, subprocess.TimeoutExpired)
+
+
+# ---------------------------------------------------------------------------
+# 3. Interactive mode — no capture
+# ---------------------------------------------------------------------------
+
+
+class TestInteractiveMode:
+    """Tests for interactive execution mode."""
+
+    def test_interactive_stdout_is_empty(self, temp_test_dir: Path):
+        """Interactive mode does not capture stdout."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "print('hello')"],
+            cwd=str(temp_test_dir),
+            interactive=True,
+        )
+        assert result.stdout == ""
+
+    def test_interactive_stderr_is_empty(self, temp_test_dir: Path):
+        """Interactive mode does not capture stderr."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "import sys; sys.stderr.write('err\\n')"],
+            cwd=str(temp_test_dir),
+            interactive=True,
+        )
+        assert result.stderr == ""
+
+    def test_interactive_exit_code_propagates(self, temp_test_dir: Path):
+        """Interactive mode still returns the process exit code."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "raise SystemExit(7)"],
+            cwd=str(temp_test_dir),
+            interactive=True,
+        )
+        assert result.exit_code == 7
+
+
+# ---------------------------------------------------------------------------
+# 4. Interactive mode — no timeout
+# ---------------------------------------------------------------------------
+
+
+class TestInteractiveNoTimeout:
+    """Interactive mode must not pass a timeout to subprocess.run."""
+
+    def test_no_timeout_kwarg(self, temp_test_dir: Path):
+        """subprocess.run is called without a timeout arg in interactive mode."""
+        with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run:
+            mock_run.return_value = subprocess.CompletedProcess(
+                args=[], returncode=0
+            )
+            execute_command(
+                sys.executable,
+                ["-c", "pass"],
+                cwd=str(temp_test_dir),
+                interactive=True,
+            )
+            call_kwargs = mock_run.call_args.kwargs
+            assert "timeout" not in call_kwargs
+
+
+# ---------------------------------------------------------------------------
+# 5. FileNotFoundError wraps to CommandExecutionError
+# ---------------------------------------------------------------------------
+
+
+class TestFileNotFoundWrapping:
+    """FileNotFoundError from a missing executable wraps correctly."""
+
+    def test_missing_executable_raises(self, temp_test_dir: Path):
+        """Non-existent executable raises CommandExecutionError."""
+        with pytest.raises(CommandExecutionError, match="Executable not found"):
+            execute_command(
+                "this_executable_does_not_exist_xyz",
+                [],
+                cwd=str(temp_test_dir),
+            )
+
+    def test_missing_executable_chains_cause(self, temp_test_dir: Path):
+        """The original FileNotFoundError is chained."""
+        with pytest.raises(CommandExecutionError) as exc_info:
+            execute_command(
+                "this_executable_does_not_exist_xyz",
+                [],
+                cwd=str(temp_test_dir),
+            )
+        assert isinstance(exc_info.value.__cause__, FileNotFoundError)
+
+
+# ---------------------------------------------------------------------------
+# 6. OSError wraps to CommandExecutionError
+# ---------------------------------------------------------------------------
+
+
+class TestOSErrorWrapping:
+    """Generic OSError wraps to CommandExecutionError."""
+
+    def test_oserror_wraps(self, temp_test_dir: Path):
+        """An OSError from subprocess.run becomes CommandExecutionError."""
+        with patch(
+            "aipass.drone.apps.handlers.executor.subprocess.run",
+            side_effect=OSError("mock OS failure"),
+        ):
+            with pytest.raises(CommandExecutionError, match="OS error"):
+                execute_command(
+                    sys.executable,
+                    ["-c", "pass"],
+                    cwd=str(temp_test_dir),
+                )
+
+    def test_oserror_chains_cause(self, temp_test_dir: Path):
+        """The original OSError is preserved as __cause__."""
+        with patch(
+            "aipass.drone.apps.handlers.executor.subprocess.run",
+            side_effect=OSError("mock OS failure"),
+        ):
+            with pytest.raises(CommandExecutionError) as exc_info:
+                execute_command(
+                    sys.executable,
+                    ["-c", "pass"],
+                    cwd=str(temp_test_dir),
+                )
+            assert isinstance(exc_info.value.__cause__, OSError)
+
+
+# ---------------------------------------------------------------------------
+# 7. KeyboardInterrupt in interactive mode returns exit code 130
+# ---------------------------------------------------------------------------
+
+
+class TestKeyboardInterrupt:
+    """KeyboardInterrupt handling differs by mode."""
+
+    def test_interactive_returns_130(self, temp_test_dir: Path):
+        """Interactive mode catches Ctrl+C and returns exit code 130."""
+        with patch(
+            "aipass.drone.apps.handlers.executor.subprocess.run",
+            side_effect=KeyboardInterrupt,
+        ):
+            result = execute_command(
+                sys.executable,
+                ["-c", "pass"],
+                cwd=str(temp_test_dir),
+                interactive=True,
+            )
+            assert result.exit_code == 130
+            assert result.stdout == ""
+            assert result.stderr == ""
+
+    def test_captured_mode_reraises_keyboard_interrupt(self, temp_test_dir: Path):
+        """Captured mode does NOT catch KeyboardInterrupt — it propagates."""
+        with patch(
+            "aipass.drone.apps.handlers.executor.subprocess.run",
+            side_effect=KeyboardInterrupt,
+        ):
+            with pytest.raises(KeyboardInterrupt):
+                execute_command(
+                    sys.executable,
+                    ["-c", "pass"],
+                    cwd=str(temp_test_dir),
+                    interactive=False,
+                )
+
+
+# ---------------------------------------------------------------------------
+# 8. Exit codes propagate correctly
+# ---------------------------------------------------------------------------
+
+
+class TestExitCodes:
+    """Exit codes from the subprocess are faithfully returned."""
+
+    def test_exit_code_zero(self, temp_test_dir: Path):
+        """Successful command returns exit code 0."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "pass"],
+            cwd=str(temp_test_dir),
+        )
+        assert result.exit_code == 0
+
+    def test_exit_code_one(self, temp_test_dir: Path):
+        """Failed command returns exit code 1."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "raise SystemExit(1)"],
+            cwd=str(temp_test_dir),
+        )
+        assert result.exit_code == 1
+
+    def test_exit_code_nonzero_arbitrary(self, temp_test_dir: Path):
+        """Arbitrary non-zero exit code propagates."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "raise SystemExit(42)"],
+            cwd=str(temp_test_dir),
+        )
+        assert result.exit_code == 42
+
+    def test_exit_code_syntax_error(self, temp_test_dir: Path):
+        """A Python syntax error produces non-zero exit code and stderr output."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "def"],
+            cwd=str(temp_test_dir),
+        )
+        assert result.exit_code != 0
+        assert "SyntaxError" in result.stderr
+
+
+# ---------------------------------------------------------------------------
+# 9. Custom env vars are merged with os.environ
+# ---------------------------------------------------------------------------
+
+
+class TestEnvMerging:
+    """Custom env dict merges with the process environment."""
+
+    def test_custom_env_var_visible(self, temp_test_dir: Path):
+        """A custom env var is available inside the subprocess."""
+        result = execute_command(
+            sys.executable,
+            ["-c", "import os; print(os.environ['AIPASS_TEST_VAR'])"],
+            cwd=str(temp_test_dir),
+            env={"AIPASS_TEST_VAR": "sentinel_value_123"},
+        )
+        assert result.stdout.strip() == "sentinel_value_123"
+
+    def test_existing_env_preserved(self, temp_test_dir: Path):
+        """Existing environment variables are still present when custom env is set."""
+        import os
+
+        expected_path = os.environ.get("PATH", "")
+        result = execute_command(
+            sys.executable,
+            ["-c", "import os; print(os.environ.get('PATH', ''))"],
+            cwd=str(temp_test_dir),
+            env={"AIPASS_TEST_VAR": "x"},
+        )
+        assert result.stdout.strip() == expected_path
+
+    def test_no_env_uses_inherited(self, temp_test_dir: Path):
+        """When env=None, the subprocess inherits the parent environment."""
+        import os
+
+        expected_path = os.environ.get("PATH", "")
+        result = execute_command(
+            sys.executable,
+            ["-c", "import os; print(os.environ.get('PATH', ''))"],
+            cwd=str(temp_test_dir),
+            env=None,
+        )
+        assert result.stdout.strip() == expected_path
+
+    def test_custom_env_overrides_existing(self, temp_test_dir: Path):
+        """Custom env values override existing environment variables."""
+        # We pick a var that definitely exists, then override it
+        result = execute_command(
+            sys.executable,
+            ["-c", "import os; print(os.environ['HOME'])"],
+            cwd=str(temp_test_dir),
+            env={"HOME": "/tmp/overridden"},
+        )
+        assert result.stdout.strip() == "/tmp/overridden"
+
+
+# ---------------------------------------------------------------------------
+# 10. shell=False is always used (security)
+# ---------------------------------------------------------------------------
+
+
+class TestShellSecurity:
+    """Verify shell=False is always passed to subprocess.run."""
+
+    def test_captured_mode_shell_false(self, temp_test_dir: Path):
+        """Captured mode calls subprocess.run with shell=False."""
+        with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run:
+            mock_run.return_value = subprocess.CompletedProcess(
+                args=[], returncode=0, stdout=b"", stderr=b""
+            )
+            execute_command(
+                sys.executable,
+                ["-c", "pass"],
+                cwd=str(temp_test_dir),
+            )
+            call_kwargs = mock_run.call_args.kwargs
+            assert call_kwargs["shell"] is False
+
+    def test_interactive_mode_shell_false(self, temp_test_dir: Path):
+        """Interactive mode calls subprocess.run with shell=False."""
+        with patch("aipass.drone.apps.handlers.executor.subprocess.run") as mock_run:
+            mock_run.return_value = subprocess.CompletedProcess(
+                args=[], returncode=0
+            )
+            execute_command(
+                sys.executable,
+                ["-c", "pass"],
+                cwd=str(temp_test_dir),
+                interactive=True,
+            )
+            call_kwargs = mock_run.call_args.kwargs
+            assert call_kwargs["shell"] is False
+
+    def test_shell_injection_prevented(self, temp_test_dir: Path):
+        """Shell metacharacters are NOT interpreted (shell=False)."""
+        # If shell=True were used, this would execute `echo pwned` too.
+        # With shell=False, the entire string is passed as one arg and
+        # Python will fail to parse it — confirming no shell expansion.
+        result = execute_command(
+            sys.executable,
+            ["-c", "import sys; print(sys.argv[1])", "hello; echo pwned"],
+            cwd=str(temp_test_dir),
+        )
+        # The semicolon is treated as literal text, not a shell separator
+        assert result.stdout.strip() == "hello; echo pwned"
+        assert result.exit_code == 0

src/aipass/drone/tests/test_registry_handler.py

@@ -0,0 +1,534 @@
+"""Tests for registry loading and credential verification.
+
+Covers:
+- load_registry() — valid JSON, missing file, corrupt JSON, metadata parsing
+- get_all_branches() — filtering, empty results
+- find_registry() — directory walk-up, no registry found
+- _verify_registry_credential() — ID match, ID missing, ID mismatch
+"""
+
+import json
+import os
+import shutil
+import tempfile
+from pathlib import Path
+from typing import Generator
+from unittest.mock import patch
+
+import pytest
+
+from aipass.drone.apps.handlers.registry_handler import (
+    _first_registry_in,
+    _verify_registry_credential,
+    find_registry,
+    get_all_branches,
+    get_registry_path,
+    load_registry,
+    reset_registry_path,
+    set_registry_path,
+)
+from aipass.drone.apps.handlers.exceptions import (
+    RegistryCorruptError,
+    RegistryMismatchError,
+    RegistryNotFoundError,
+    RegistryPermissionError,
+)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def registry_dir() -> Generator[Path, None, None]:
+    """Isolated temp directory for registry tests; cleaned up after."""
+    d = Path(tempfile.mkdtemp(prefix="reg_test_"))
+    yield d
+    shutil.rmtree(d, ignore_errors=True)
+
+
+@pytest.fixture(autouse=True)
+def _reset_global_path():
+    """Ensure the module-level _registry_path is reset between tests."""
+    reset_registry_path()
+    yield
+    reset_registry_path()
+
+
+def _write_registry(directory: Path, data: dict, name: str = "AIPASS_REGISTRY.json") -> Path:
+    """Helper: write a registry JSON file and return its path."""
+    p = directory / name
+    p.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    return p
+
+
+def _write_passport(directory: Path, passport: dict) -> Path:
+    """Helper: write a .trinity/passport.json and return its path."""
+    trinity = directory / ".trinity"
+    trinity.mkdir(parents=True, exist_ok=True)
+    p = trinity / "passport.json"
+    p.write_text(json.dumps(passport, indent=2), encoding="utf-8")
+    return p
+
+
+def _minimal_registry(*, metadata_id: str | None = None, branches: list | None = None) -> dict:
+    """Return a minimal valid registry dict."""
+    meta = {"version": "1.0.0"}
+    if metadata_id is not None:
+        meta["id"] = metadata_id
+    if branches is None:
+        branches = [
+            {
+                "name": "Alpha",
+                "path": "alpha",
+                "type": "library",
+                "status": "active",
+                "description": "Alpha branch",
+            }
+        ]
+    return {"metadata": meta, "branches": branches}
+
+
+# ===================================================================
+# 1. load_registry() — valid registry
+# ===================================================================
+
+class TestLoadRegistry:
+
+    def test_branches_normalised_to_dict(self, registry_dir: Path):
+        """List-format branches are converted to a dict keyed by lowercased name."""
+        reg = _minimal_registry()
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+
+        assert isinstance(result, dict)
+        assert "branches" in result
+        assert "metadata" in result
+        assert isinstance(result["branches"], dict)
+        assert "alpha" in result["branches"]
+        assert result["branches"]["alpha"]["name"] == "alpha"
+
+    def test_relative_paths_resolved(self, registry_dir: Path):
+        """Relative branch paths are resolved against registry location."""
+        reg = _minimal_registry()
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+        branch_path = Path(result["branches"]["alpha"]["path"])
+
+        # The resolved path should be absolute
+        assert branch_path.is_absolute()
+        # It should be relative to the registry directory
+        assert str(branch_path).startswith(str(registry_dir))
+
+    def test_absolute_path_not_re_resolved(self, registry_dir: Path):
+        """A branch with an already-absolute path is NOT re-resolved against registry_dir."""
+        abs_path = "/opt/custom/my_branch"
+        reg = _minimal_registry(
+            branches=[
+                {
+                    "name": "Absolute",
+                    "path": abs_path,
+                    "type": "library",
+                    "status": "active",
+                }
+            ]
+        )
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+        branch_path = result["branches"]["absolute"]["path"]
+
+        assert branch_path == abs_path
+
+    def test_branches_with_empty_name_skipped(self, registry_dir: Path):
+        """load_registry() skips branch entries with no name field."""
+        reg = _minimal_registry(
+            branches=[
+                {
+                    "name": "Good",
+                    "path": "good",
+                    "type": "library",
+                    "status": "active",
+                },
+                {
+                    "name": "",
+                    "path": "empty_name",
+                    "type": "library",
+                    "status": "active",
+                },
+                {
+                    "path": "missing_name",
+                    "type": "library",
+                    "status": "active",
+                },
+            ]
+        )
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+
+        assert len(result["branches"]) == 1
+        assert "good" in result["branches"]
+
+    def test_dict_format_branches_preserved(self, registry_dir: Path):
+        """If branches is already a dict, it stays a dict."""
+        reg = {
+            "metadata": {"version": "1.0.0"},
+            "branches": {
+                "beta": {
+                    "name": "beta",
+                    "path": "/tmp/beta",
+                    "status": "active",
+                }
+            },
+        }
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+        assert result["branches"]["beta"]["name"] == "beta"
+
+    # ---------------------------------------------------------------
+    # 2. load_registry() — missing file
+    # ---------------------------------------------------------------
+
+    def test_missing_file_raises_not_found(self, registry_dir: Path):
+        """load_registry() raises RegistryNotFoundError when file is absent."""
+        set_registry_path(registry_dir / "NONEXISTENT_REGISTRY.json")
+
+        with pytest.raises(RegistryNotFoundError):
+            load_registry()
+
+    # ---------------------------------------------------------------
+    # 3. load_registry() — corrupt / invalid JSON
+    # ---------------------------------------------------------------
+
+    def test_corrupt_json_raises_corrupt_error(self, registry_dir: Path):
+        """Invalid JSON content triggers RegistryCorruptError."""
+        bad_file = registry_dir / "BAD_REGISTRY.json"
+        bad_file.write_text("{not valid json!!!", encoding="utf-8")
+        set_registry_path(bad_file)
+
+        with pytest.raises(RegistryCorruptError):
+            load_registry()
+
+    def test_non_object_json_raises_corrupt_error(self, registry_dir: Path):
+        """Top-level JSON that isn't a dict triggers RegistryCorruptError."""
+        bad_file = registry_dir / "LIST_REGISTRY.json"
+        bad_file.write_text(json.dumps([1, 2, 3]), encoding="utf-8")
+        set_registry_path(bad_file)
+
+        with pytest.raises(RegistryCorruptError):
+            load_registry()
+
+    def test_missing_branches_key_raises_corrupt_error(self, registry_dir: Path):
+        """Registry dict without 'branches' triggers RegistryCorruptError."""
+        no_branches = {"metadata": {"version": "1.0.0"}}
+        _write_registry(registry_dir, no_branches)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        with pytest.raises(RegistryCorruptError, match="missing 'branches'"):
+            load_registry()
+
+    def test_branches_wrong_type_raises_corrupt_error(self, registry_dir: Path):
+        """branches field that is neither list nor dict triggers RegistryCorruptError."""
+        bad = {"metadata": {"version": "1.0.0"}, "branches": "not-a-list-or-dict"}
+        _write_registry(registry_dir, bad)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        with pytest.raises(RegistryCorruptError):
+            load_registry()
+
+    # ---------------------------------------------------------------
+    # 4. load_registry() — permission error
+    # ---------------------------------------------------------------
+
+    def test_permission_error_raises_registry_permission_error(self, registry_dir: Path):
+        """PermissionError when reading the file triggers RegistryPermissionError."""
+        reg_file = _write_registry(registry_dir, _minimal_registry())
+        set_registry_path(reg_file)
+
+        with patch("builtins.open", side_effect=PermissionError("access denied")):
+            with pytest.raises(RegistryPermissionError, match="Permission denied"):
+                load_registry()
+
+
+# ===================================================================
+# 4 & 5. get_all_branches()
+# ===================================================================
+
+class TestGetAllBranches:
+
+    def test_returns_list_of_branch_dicts(self, registry_dir: Path):
+        """get_all_branches() returns a list of dicts for active branches."""
+        reg = _minimal_registry()
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        branches = get_all_branches()
+
+        assert isinstance(branches, list)
+        assert len(branches) == 1
+        assert branches[0]["name"] == "alpha"
+
+    def test_filters_by_status(self, registry_dir: Path):
+        """Branches with non-matching status are excluded."""
+        reg = _minimal_registry(
+            branches=[
+                {"name": "Active", "path": "a", "status": "active", "type": "lib"},
+                {"name": "Archived", "path": "b", "status": "archived", "type": "lib"},
+            ]
+        )
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        active = get_all_branches(status="active")
+        assert len(active) == 1
+        assert active[0]["name"] == "active"
+
+        archived = get_all_branches(status="archived")
+        assert len(archived) == 1
+        assert archived[0]["name"] == "archived"
+
+    def test_filters_by_branch_type(self, registry_dir: Path):
+        """branch_type filter limits results."""
+        reg = _minimal_registry(
+            branches=[
+                {"name": "Lib", "path": "l", "status": "active", "type": "library"},
+                {"name": "App", "path": "a", "status": "active", "type": "app"},
+            ]
+        )
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        libs = get_all_branches(branch_type="library")
+        assert len(libs) == 1
+        assert libs[0]["type"] == "library"
+
+    def test_returns_empty_list_when_no_registry(self, registry_dir: Path):
+        """When no registry file exists, get_all_branches returns []."""
+        set_registry_path(registry_dir / "NONEXISTENT_REGISTRY.json")
+
+        result = get_all_branches()
+        assert result == []
+
+    def test_returns_empty_when_no_matching_branches(self, registry_dir: Path):
+        """All branches filtered out yields an empty list."""
+        reg = _minimal_registry(
+            branches=[
+                {"name": "Only", "path": "o", "status": "archived", "type": "lib"},
+            ]
+        )
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = get_all_branches(status="active")
+        assert result == []
+
+
+# ===================================================================
+# 6 & 7. find_registry() / _first_registry_in()
+# ===================================================================
+
+class TestFindRegistry:
+
+    def test_first_registry_in_finds_file(self, registry_dir: Path):
+        """_first_registry_in returns path when *_REGISTRY.json exists."""
+        _write_registry(registry_dir, _minimal_registry())
+
+        hit = _first_registry_in(registry_dir)
+        assert hit is not None
+        assert hit.name == "AIPASS_REGISTRY.json"
+
+    def test_first_registry_in_returns_none_when_empty(self, registry_dir: Path):
+        """_first_registry_in returns None in an empty directory."""
+        assert _first_registry_in(registry_dir) is None
+
+    def test_first_registry_in_alphabetical(self, registry_dir: Path):
+        """When multiple registries exist, alphabetically first wins."""
+        _write_registry(registry_dir, _minimal_registry(), name="A_REGISTRY.json")
+        _write_registry(registry_dir, _minimal_registry(), name="Z_REGISTRY.json")
+
+        hit = _first_registry_in(registry_dir)
+        assert hit is not None
+        assert hit.name == "A_REGISTRY.json"
+
+    def test_find_registry_from_child_dir(self, registry_dir: Path, monkeypatch):
+        """find_registry() walks up from cwd to find registry in ancestor."""
+        _write_registry(registry_dir, _minimal_registry())
+        child = registry_dir / "level1" / "level2"
+        child.mkdir(parents=True)
+
+        monkeypatch.chdir(child)
+        result = find_registry()
+        assert result.exists()
+        assert result.name == "AIPASS_REGISTRY.json"
+        assert result.parent == registry_dir
+
+    def test_find_registry_returns_path_when_no_registry(self, registry_dir: Path, monkeypatch):
+        """find_registry() returns a fallback path ending with AIPASS_REGISTRY.json when nothing found."""
+        empty_child = registry_dir / "empty_sub"
+        empty_child.mkdir()
+
+        monkeypatch.chdir(empty_child)
+        # find_registry never returns None -- it returns a conventional fallback path
+        result = find_registry()
+        assert isinstance(result, Path)
+        assert result.name == "AIPASS_REGISTRY.json"
+
+
+# ===================================================================
+# 8, 9, 10. _verify_registry_credential()
+# ===================================================================
+
+class TestVerifyRegistryCredential:
+
+    def test_passes_when_ids_match(self, registry_dir: Path, monkeypatch):
+        """No error when passport.citizenship.registry_id == registry.metadata.id."""
+        shared_id = "reg-abc-123"
+        registry_data = _minimal_registry(metadata_id=shared_id)
+        registry_path = _write_registry(registry_dir, registry_data)
+        _write_passport(registry_dir, {
+            "citizenship": {"registry_id": shared_id}
+        })
+
+        monkeypatch.chdir(registry_dir)
+        # Should not raise
+        _verify_registry_credential(registry_path, registry_data)
+
+    def test_passes_when_registry_id_missing(self, registry_dir: Path, monkeypatch):
+        """No error when registry has no metadata.id (migration period)."""
+        registry_data = _minimal_registry()  # no metadata_id
+        registry_path = _write_registry(registry_dir, registry_data)
+        _write_passport(registry_dir, {
+            "citizenship": {"registry_id": "some-id"}
+        })
+
+        monkeypatch.chdir(registry_dir)
+        _verify_registry_credential(registry_path, registry_data)
+
+    def test_passes_when_passport_id_missing(self, registry_dir: Path, monkeypatch):
+        """No error when passport has no citizenship.registry_id (migration period)."""
+        shared_id = "reg-xyz-789"
+        registry_data = _minimal_registry(metadata_id=shared_id)
+        registry_path = _write_registry(registry_dir, registry_data)
+        _write_passport(registry_dir, {
+            "citizenship": {}  # no registry_id
+        })
+
+        monkeypatch.chdir(registry_dir)
+        _verify_registry_credential(registry_path, registry_data)
+
+    def test_passes_when_no_passport_file(self, registry_dir: Path, monkeypatch):
+        """No error when passport.json does not exist at all."""
+        shared_id = "reg-000"
+        registry_data = _minimal_registry(metadata_id=shared_id)
+        registry_path = _write_registry(registry_dir, registry_data)
+        # No passport written
+
+        monkeypatch.chdir(registry_dir)
+        _verify_registry_credential(registry_path, registry_data)
+
+    def test_raises_on_id_mismatch(self, registry_dir: Path, monkeypatch):
+        """RegistryMismatchError raised when IDs differ -- security-critical."""
+        registry_data = _minimal_registry(metadata_id="registry-AAA")
+        registry_path = _write_registry(registry_dir, registry_data)
+        _write_passport(registry_dir, {
+            "citizenship": {"registry_id": "registry-BBB"}
+        })
+
+        monkeypatch.chdir(registry_dir)
+        with pytest.raises(RegistryMismatchError, match="mismatch"):
+            _verify_registry_credential(registry_path, registry_data)
+
+    def test_mismatch_error_contains_both_ids(self, registry_dir: Path, monkeypatch):
+        """Error message includes both the passport and registry IDs for debugging."""
+        reg_id = "registry-PROD"
+        passport_id = "registry-DEV"
+        registry_data = _minimal_registry(metadata_id=reg_id)
+        registry_path = _write_registry(registry_dir, registry_data)
+        _write_passport(registry_dir, {
+            "citizenship": {"registry_id": passport_id}
+        })
+
+        monkeypatch.chdir(registry_dir)
+        with pytest.raises(RegistryMismatchError) as exc_info:
+            _verify_registry_credential(registry_path, registry_data)
+
+        msg = str(exc_info.value)
+        assert passport_id in msg
+        assert reg_id in msg
+
+    def test_passes_when_metadata_key_absent(self, registry_dir: Path, monkeypatch):
+        """No error when the entire metadata dict is absent."""
+        registry_data = {"branches": []}  # no metadata at all
+        registry_path = registry_dir / "FAKE_REGISTRY.json"
+
+        monkeypatch.chdir(registry_dir)
+        _verify_registry_credential(registry_path, registry_data)
+
+
+# ===================================================================
+# 11. Metadata parsing
+# ===================================================================
+
+class TestMetadataParsing:
+
+    def test_metadata_id_preserved(self, registry_dir: Path, monkeypatch):
+        """Registry metadata.id field is available after loading."""
+        reg = _minimal_registry(metadata_id="my-unique-id")
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        # chdir to temp so _verify_registry_credential won't find the
+        # real project passport (which has a different registry_id).
+        monkeypatch.chdir(registry_dir)
+        result = load_registry()
+        assert result["metadata"]["id"] == "my-unique-id"
+
+    def test_metadata_version_preserved(self, registry_dir: Path):
+        """Registry metadata.version field survives loading."""
+        reg = _minimal_registry()
+        _write_registry(registry_dir, reg)
+        set_registry_path(registry_dir / "AIPASS_REGISTRY.json")
+
+        result = load_registry()
+        assert result["metadata"]["version"] == "1.0.0"
+
+
+# ===================================================================
+# Registry path management
+# ===================================================================
+
+class TestRegistryPathManagement:
+
+    def test_set_and_get_registry_path(self, registry_dir: Path):
+        """set_registry_path() overrides get_registry_path()."""
+        custom = registry_dir / "CUSTOM_REGISTRY.json"
+        set_registry_path(custom)
+
+        assert get_registry_path() == custom
+
+    def test_reset_clears_override(self, registry_dir: Path):
+        """reset_registry_path() clears any previously set override."""
+        set_registry_path(registry_dir / "CUSTOM_REGISTRY.json")
+        reset_registry_path()
+
+        # After reset, get_registry_path falls through to env / find
+        result = get_registry_path()
+        assert result != registry_dir / "CUSTOM_REGISTRY.json"
+
+    def test_env_var_takes_precedence_after_reset(self, registry_dir: Path):
+        """AIPASS_REGISTRY env var is used when no explicit path is set."""
+        env_path = str(registry_dir / "ENV_REGISTRY.json")
+        reset_registry_path()
+
+        with patch.dict(os.environ, {"AIPASS_REGISTRY": env_path}):
+            result = get_registry_path()
+            assert result == Path(env_path)

src/aipass/drone/tests/test_resolver.py

@@ -0,0 +1,392 @@
+"""Tests for branch resolution (@name -> path).
+
+Covers resolve_branch(), list_branches(), branch_exists(), get_branch_info(),
+normalize helpers, and handle_command() routing.
+"""
+
+import json
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+
+from aipass.drone.apps.handlers.exceptions import BranchNotFoundError
+from aipass.drone.apps.handlers.registry_handler import (
+    reset_registry_path,
+    set_registry_path,
+)
+from aipass.drone.apps.modules.resolver import (
+    branch_exists,
+    get_branch_info,
+    handle_command,
+    list_branches,
+    normalize_branch_arg,
+    normalize_branch_name,
+    resolve_branch,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _write_registry(
+    registry_path: Path,
+    branches: List[Dict[str, Any]],
+    *,
+    metadata: Dict[str, Any] | None = None,
+) -> Path:
+    """Write a registry file with the given branches list."""
+    registry = {
+        "metadata": metadata or {"version": "1.0.0"},
+        "branches": branches,
+    }
+    registry_path.write_text(json.dumps(registry, indent=2))
+    return registry_path
+
+
+def _make_branch(
+    name: str,
+    path: str,
+    *,
+    status: str = "active",
+    branch_type: str = "library",
+    description: str = "",
+) -> Dict[str, Any]:
+    """Build a branch entry dict."""
+    return {
+        "name": name,
+        "path": path,
+        "profile": branch_type,
+        "type": branch_type,
+        "description": description or f"{name} branch",
+        "email": f"@{name.lower()}",
+        "status": status,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def registry_dir(temp_test_dir: Path):
+    """Set registry path for the duration of a test, then reset."""
+    registry_path = temp_test_dir / "AIPASS_REGISTRY.json"
+    set_registry_path(registry_path)
+    yield temp_test_dir
+    reset_registry_path()
+
+
+@pytest.fixture
+def populated_registry(registry_dir: Path) -> Path:
+    """Create a registry with several branches of varying status/type."""
+    branch_alpha = registry_dir / "alpha"
+    branch_beta = registry_dir / "beta"
+    branch_gamma = registry_dir / "gamma"
+
+    registry_path = registry_dir / "AIPASS_REGISTRY.json"
+    _write_registry(
+        registry_path,
+        [
+            _make_branch("ALPHA", str(branch_alpha), status="active", branch_type="library"),
+            _make_branch("BETA", str(branch_beta), status="active", branch_type="service"),
+            _make_branch("GAMMA", str(branch_gamma), status="archived", branch_type="library"),
+        ],
+    )
+    return registry_path
+
+
+@pytest.fixture
+def single_branch_registry(registry_dir: Path, temp_test_dir: Path) -> Path:
+    """Registry with exactly one active branch — uses the conftest sample layout."""
+    branch_path = temp_test_dir / "test_branch"
+    registry_path = registry_dir / "AIPASS_REGISTRY.json"
+    _write_registry(
+        registry_path,
+        [
+            _make_branch("TEST_BRANCH", str(branch_path)),
+        ],
+    )
+    return registry_path
+
+
+@pytest.fixture
+def empty_registry(registry_dir: Path) -> Path:
+    """Registry with an empty branches list."""
+    registry_path = registry_dir / "AIPASS_REGISTRY.json"
+    _write_registry(registry_path, [])
+    return registry_path
+
+
+# ---------------------------------------------------------------------------
+# normalize helpers
+# ---------------------------------------------------------------------------
+
+class TestNormalizeBranchName:
+    def test_strips_at_prefix(self):
+        assert normalize_branch_name("@FOO") == "FOO"
+
+    def test_no_prefix_unchanged(self):
+        assert normalize_branch_name("FOO") == "FOO"
+
+    def test_empty_string(self):
+        assert normalize_branch_name("") == ""
+
+    def test_double_at_strips_one(self):
+        assert normalize_branch_name("@@FOO") == "@FOO"
+
+
+class TestNormalizeBranchArg:
+    def test_strips_at_and_lowercases(self):
+        assert normalize_branch_arg("@FOO") == "foo"
+
+    def test_no_prefix_lowercases(self):
+        assert normalize_branch_arg("FOO") == "foo"
+
+    def test_multiple_at_stripped(self):
+        # lstrip removes all leading '@' characters
+        assert normalize_branch_arg("@@FOO") == "foo"
+
+
+# ---------------------------------------------------------------------------
+# resolve_branch
+# ---------------------------------------------------------------------------
+
+class TestResolveBranch:
+    def test_resolve_with_at_prefix(self, populated_registry):
+        path = resolve_branch("@ALPHA")
+        assert Path(path).name == "alpha"
+
+    def test_resolve_without_at_prefix(self, populated_registry):
+        path = resolve_branch("ALPHA")
+        assert Path(path).name == "alpha"
+
+    def test_returns_absolute_path(self, populated_registry):
+        path = resolve_branch("@ALPHA")
+        assert Path(path).is_absolute()
+
+    def test_case_insensitive_lookup(self, populated_registry):
+        """Resolver lowercases the name, registry normalizes names to lower."""
+        path_upper = resolve_branch("@ALPHA")
+        path_lower = resolve_branch("@alpha")
+        path_mixed = resolve_branch("@Alpha")
+        assert path_upper == path_lower == path_mixed
+
+    def test_invalid_branch_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            resolve_branch("@NONEXISTENT")
+
+    def test_invalid_branch_without_prefix_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            resolve_branch("NONEXISTENT")
+
+    def test_empty_name_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            resolve_branch("")
+
+    def test_at_only_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            resolve_branch("@")
+
+    def test_resolve_each_branch(self, populated_registry):
+        """Every branch in the registry should resolve."""
+        for name in ("ALPHA", "BETA", "GAMMA"):
+            path = resolve_branch(f"@{name}")
+            assert Path(path).name == name.lower()
+
+    def test_empty_registry_raises(self, empty_registry):
+        with pytest.raises(BranchNotFoundError):
+            resolve_branch("@ALPHA")
+
+
+# ---------------------------------------------------------------------------
+# branch_exists
+# ---------------------------------------------------------------------------
+
+class TestBranchExists:
+    def test_exists_for_valid_branch(self, populated_registry):
+        assert branch_exists("@ALPHA") is True
+
+    def test_exists_without_prefix(self, populated_registry):
+        assert branch_exists("ALPHA") is True
+
+    def test_not_exists_for_unknown(self, populated_registry):
+        assert branch_exists("@NOPE") is False
+
+    def test_case_insensitive(self, populated_registry):
+        assert branch_exists("@alpha") is True
+        assert branch_exists("@Alpha") is True
+
+    def test_empty_registry(self, empty_registry):
+        assert branch_exists("@ALPHA") is False
+
+    def test_archived_branch_exists(self, populated_registry):
+        """branch_exists uses get_branch_by_name — no status filter."""
+        assert branch_exists("@GAMMA") is True
+
+
+# ---------------------------------------------------------------------------
+# get_branch_info
+# ---------------------------------------------------------------------------
+
+class TestGetBranchInfo:
+    def test_contains_expected_keys(self, populated_registry):
+        info = get_branch_info("@ALPHA")
+        assert "name" in info
+        assert "path" in info
+        assert "status" in info
+        assert info["status"] == "active"
+        assert Path(info["path"]).name == "alpha"
+
+    def test_name_is_lowercased_in_result(self, populated_registry):
+        info = get_branch_info("@ALPHA")
+        assert info["name"] == "alpha"
+
+    def test_path_matches_resolve(self, populated_registry):
+        info = get_branch_info("@BETA")
+        path = resolve_branch("@BETA")
+        assert info["path"] == path
+
+    def test_case_insensitive(self, populated_registry):
+        info_upper = get_branch_info("@ALPHA")
+        info_lower = get_branch_info("@alpha")
+        assert info_upper == info_lower
+
+    def test_invalid_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            get_branch_info("@NONEXISTENT")
+
+    def test_without_prefix_raises_for_invalid(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            get_branch_info("NONEXISTENT")
+
+    def test_empty_registry_raises(self, empty_registry):
+        with pytest.raises(BranchNotFoundError):
+            get_branch_info("@ALPHA")
+
+
+# ---------------------------------------------------------------------------
+# list_branches
+# ---------------------------------------------------------------------------
+
+class TestListBranches:
+    def test_default_status_active(self, populated_registry):
+        """Default status='active' should exclude archived branches."""
+        result = list_branches()
+        names_lower = [b.lower() for b in result]
+        assert "@alpha" in names_lower
+        assert "@beta" in names_lower
+        assert "@gamma" not in names_lower  # archived
+
+    def test_all_active_returned(self, populated_registry):
+        result = list_branches()
+        assert len(result) == 2
+
+    def test_filter_by_archived_status(self, populated_registry):
+        result = list_branches(status="archived")
+        assert len(result) == 1
+        assert result[0].lower() == "@gamma"
+
+    def test_filter_by_type(self, populated_registry):
+        result = list_branches(branch_type="service")
+        assert len(result) == 1
+        assert result[0].lower() == "@beta"
+
+    def test_filter_by_type_and_status(self, populated_registry):
+        result = list_branches(branch_type="library", status="active")
+        assert len(result) == 1
+        assert result[0].lower() == "@alpha"
+
+    def test_entries_have_at_prefix(self, populated_registry):
+        result = list_branches()
+        for entry in result:
+            assert entry.startswith("@")
+
+    def test_empty_registry(self, empty_registry):
+        result = list_branches()
+        assert result == []
+
+    def test_no_matching_type(self, populated_registry):
+        result = list_branches(branch_type="nonexistent_type")
+        assert result == []
+
+
+# ---------------------------------------------------------------------------
+# handle_command routing
+# ---------------------------------------------------------------------------
+
+class TestHandleCommand:
+    def test_resolve_command_success(self, populated_registry):
+        assert handle_command("resolve", ["@ALPHA"]) is True
+
+    def test_resolve_command_no_args(self, populated_registry):
+        assert handle_command("resolve", []) is False
+
+    def test_exists_command_success(self, populated_registry):
+        assert handle_command("exists", ["@ALPHA"]) is True
+
+    def test_exists_command_no_args(self, populated_registry):
+        assert handle_command("exists", []) is False
+
+    def test_info_command_success(self, populated_registry):
+        assert handle_command("info", ["@ALPHA"]) is True
+
+    def test_info_command_no_args(self, populated_registry):
+        assert handle_command("info", []) is False
+
+    def test_list_command(self, populated_registry):
+        assert handle_command("list", []) is True
+
+    def test_resolve_nonexistent_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            handle_command("resolve", ["@NONEXISTENT"])
+
+    def test_info_nonexistent_raises(self, populated_registry):
+        with pytest.raises(BranchNotFoundError):
+            handle_command("info", ["@NONEXISTENT"])
+
+    def test_unknown_command(self, populated_registry):
+        assert handle_command("bogus", []) is False
+
+
+# ---------------------------------------------------------------------------
+# sample_registry conftest fixture
+# ---------------------------------------------------------------------------
+
+class TestWithSampleRegistry:
+    """Tests using the sample_registry fixture from conftest.py."""
+
+    def test_resolve_via_sample(self, sample_registry):
+        set_registry_path(sample_registry)
+        try:
+            path = resolve_branch("@TEST_BRANCH")
+            assert "test_branch" in path
+        finally:
+            reset_registry_path()
+
+    def test_branch_exists_via_sample(self, sample_registry):
+        set_registry_path(sample_registry)
+        try:
+            assert branch_exists("@TEST_BRANCH") is True
+            assert branch_exists("@MISSING") is False
+        finally:
+            reset_registry_path()
+
+    def test_list_branches_via_sample(self, sample_registry):
+        set_registry_path(sample_registry)
+        try:
+            result = list_branches()
+            assert len(result) == 1
+            assert result[0].lower() == "@test_branch"
+        finally:
+            reset_registry_path()
+
+    def test_get_info_via_sample(self, sample_registry):
+        set_registry_path(sample_registry)
+        try:
+            info = get_branch_info("@TEST_BRANCH")
+            assert info["name"] == "test_branch"
+            assert info["status"] == "active"
+        finally:
+            reset_registry_path()

src/aipass/drone/tests/test_router.py

@@ -0,0 +1,630 @@
+"""Tests for command routing — router module and router_handler."""
+
+import json
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from aipass.drone.apps.handlers.exceptions import (
+    BranchNotFoundError,
+    CommandExecutionError,
+)
+from aipass.drone.apps.handlers.executor import CommandResult
+from aipass.drone.apps.handlers.router_handler import (
+    detect_caller_branch_name,
+    execute_branch_command,
+    find_entry_point,
+)
+from aipass.drone.apps.modules.router import handle_command, route_command, route_all
+
+
+# ---------------------------------------------------------------------------
+# find_entry_point
+# ---------------------------------------------------------------------------
+
+
+class TestFindEntryPoint:
+    """Tests for find_entry_point() in router_handler."""
+
+    def test_locates_existing_entry_point(self, temp_test_dir: Path):
+        """find_entry_point returns the correct Path when apps/{name}.py exists."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir()
+        entry_file = apps_dir / "mybranch.py"
+        entry_file.write_text("# entry point stub")
+
+        result = find_entry_point(str(temp_test_dir), "mybranch")
+
+        assert result == entry_file
+        assert result.exists()
+
+    def test_raises_when_entry_point_missing(self, temp_test_dir: Path):
+        """find_entry_point raises CommandExecutionError if file doesn't exist."""
+        with pytest.raises(CommandExecutionError, match="Entry point not found"):
+            find_entry_point(str(temp_test_dir), "nonexistent")
+
+    def test_returns_path_under_apps_subdirectory(self, temp_test_dir: Path):
+        """Returned path is always branch_path / apps / {name}.py."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir()
+        entry_file = apps_dir / "test_branch.py"
+        entry_file.write_text("")
+
+        result = find_entry_point(str(temp_test_dir), "test_branch")
+
+        assert result.parent.name == "apps"
+        assert result.name == "test_branch.py"
+
+
+# ---------------------------------------------------------------------------
+# execute_branch_command
+# ---------------------------------------------------------------------------
+
+
+class TestExecuteBranchCommand:
+    """Tests for execute_branch_command() in router_handler."""
+
+    @pytest.fixture
+    def branch_dir(self, temp_test_dir: Path) -> Path:
+        """Create a fake branch directory with a valid entry point."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir()
+        entry = apps_dir / "fakebranch.py"
+        entry.write_text("# stub")
+        return temp_test_dir
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_valid_command_returns_command_result(
+        self, mock_exec, branch_dir: Path
+    ):
+        """A valid command returns a CommandResult with correct fields."""
+        mock_exec.return_value = CommandResult(
+            stdout="ok\n", stderr="", exit_code=0, branch="", command=""
+        )
+
+        result = execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="status",
+        )
+
+        assert isinstance(result, CommandResult)
+        assert result.exit_code == 0
+        assert result.stdout == "ok\n"
+        assert result.branch == "fakebranch"
+        assert result.command == "status"
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_introspection_with_command_none(
+        self, mock_exec, branch_dir: Path
+    ):
+        """When command=None the entry point is invoked with no command args."""
+        mock_exec.return_value = CommandResult(
+            stdout="introspect output", stderr="", exit_code=0, branch="", command=""
+        )
+
+        result = execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command=None,
+        )
+
+        # Args passed to execute_command should only be the relative entry point
+        args_list = mock_exec.call_args.kwargs["args"]
+        assert args_list == [str(Path("apps") / "fakebranch.py")]
+        assert result.command == ""
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_interactive_flag_passed_through(
+        self, mock_exec, branch_dir: Path
+    ):
+        """interactive=True is forwarded to execute_command."""
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="monitor",
+            interactive=True,
+        )
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs.get("interactive") is True
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_sets_aipass_caller_cwd_env(
+        self, mock_exec, branch_dir: Path
+    ):
+        """AIPASS_CALLER_CWD is set in the env dict passed to execute_command."""
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="test",
+        )
+
+        call_kwargs = mock_exec.call_args.kwargs
+        env = call_kwargs.get("env", {})
+        assert "AIPASS_CALLER_CWD" in env
+        assert env["AIPASS_CALLER_CWD"] == str(Path.cwd())
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_timeout_propagated_to_executor(
+        self, mock_exec, branch_dir: Path
+    ):
+        """Timeout value is forwarded to execute_command."""
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="slow",
+            timeout=120,
+        )
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs.get("timeout") == 120
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_args_appended_to_command(
+        self, mock_exec, branch_dir: Path
+    ):
+        """Extra args are appended after the command in the args list."""
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="deploy",
+            args=["--force", "--env=prod"],
+        )
+
+        args_list = mock_exec.call_args.kwargs["args"]
+        # Verify the args list ends with the command and its arguments in exact order
+        assert args_list[-3:] == ["deploy", "--force", "--env=prod"]
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_uses_sys_executable(
+        self, mock_exec, branch_dir: Path
+    ):
+        """execute_command is called with sys.executable as the executable."""
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        execute_branch_command(
+            branch_path=str(branch_dir),
+            branch_name="fakebranch",
+            command="test",
+        )
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs.get("executable") == sys.executable
+
+    def test_raises_when_entry_point_missing(self, temp_test_dir: Path):
+        """execute_branch_command raises CommandExecutionError if entry point missing."""
+        with pytest.raises(CommandExecutionError, match="Entry point not found"):
+            execute_branch_command(
+                branch_path=str(temp_test_dir),
+                branch_name="does_not_exist",
+                command="test",
+            )
+
+
+# ---------------------------------------------------------------------------
+# route_command (integration-level, mocking handler layer)
+# ---------------------------------------------------------------------------
+
+
+class TestRouteCommand:
+    """Tests for route_command() in the router module."""
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_valid_branch_and_command(self, mock_resolve, mock_exec):
+        """route_command resolves target and delegates to execute_branch_command."""
+        mock_resolve.return_value = "/fake/path/to/branch"
+        mock_exec.return_value = CommandResult(
+            stdout="done", stderr="", exit_code=0, branch="mybranch", command="status"
+        )
+
+        result = route_command("@mybranch", "status")
+
+        mock_resolve.assert_called_once_with("@mybranch")
+        mock_exec.assert_called_once()
+        assert result.exit_code == 0
+        assert result.stdout == "done"
+
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_invalid_branch_raises_branch_not_found(self, mock_resolve):
+        """route_command propagates BranchNotFoundError from resolver."""
+        mock_resolve.side_effect = BranchNotFoundError("Branch '@ghost' not found")
+
+        with pytest.raises(BranchNotFoundError, match="not found"):
+            route_command("@ghost", "status")
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_timeout_forwarded(self, mock_resolve, mock_exec):
+        """route_command passes timeout through to execute_branch_command."""
+        mock_resolve.return_value = "/fake/path"
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="b", command="c"
+        )
+
+        route_command("@somebranch", "cmd", timeout=90)
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs["timeout"] == 90
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_interactive_forwarded(self, mock_resolve, mock_exec):
+        """route_command passes interactive flag through."""
+        mock_resolve.return_value = "/fake/path"
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="b", command="c"
+        )
+
+        route_command("@somebranch", "monitor", interactive=True)
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs["interactive"] is True
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_introspection_no_command(self, mock_resolve, mock_exec):
+        """route_command with command=None triggers introspection."""
+        mock_resolve.return_value = "/fake/path"
+        mock_exec.return_value = CommandResult(
+            stdout="info", stderr="", exit_code=0, branch="b", command=""
+        )
+
+        result = route_command("@somebranch", None)
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs["command"] is None
+        assert result.stdout == "info"
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_args_forwarded(self, mock_resolve, mock_exec):
+        """route_command forwards args list to execute_branch_command."""
+        mock_resolve.return_value = "/fake/path"
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="b", command="c"
+        )
+
+        route_command("@mybranch", "deploy", args=["--env=staging"])
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs["args"] == ["--env=staging"]
+
+    @patch(
+        "aipass.drone.apps.modules.router.execute_branch_command"
+    )
+    @patch("aipass.drone.apps.modules.router.resolve_branch")
+    def test_branch_name_stripped_and_lowered(self, mock_resolve, mock_exec):
+        """route_command strips @ prefix and lowercases for branch_name."""
+        mock_resolve.return_value = "/fake/path"
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="mybranch", command="test"
+        )
+
+        route_command("@MyBranch", "test")
+
+        call_kwargs = mock_exec.call_args.kwargs
+        assert call_kwargs["branch_name"] == "mybranch"
+
+
+# ---------------------------------------------------------------------------
+# route_all
+# ---------------------------------------------------------------------------
+
+
+class TestRouteAll:
+    """Tests for route_all() in the router module."""
+
+    @patch("aipass.drone.apps.modules.router.route_command")
+    @patch("aipass.drone.apps.modules.router.list_branches")
+    def test_routes_to_all_active_branches(self, mock_list, mock_route):
+        """route_all dispatches the command to every active branch."""
+        mock_list.return_value = ["@alpha", "@beta"]
+        mock_route.return_value = CommandResult(
+            stdout="ok", stderr="", exit_code=0, branch="", command="status"
+        )
+
+        results = route_all("status")
+
+        assert len(results) == 2
+        assert "alpha" in results
+        assert "beta" in results
+        assert mock_route.call_count == 2
+
+    @patch("aipass.drone.apps.modules.router.route_command")
+    @patch("aipass.drone.apps.modules.router.list_branches")
+    def test_captures_failure_per_branch(self, mock_list, mock_route):
+        """When a branch raises, route_all records exit_code=-1 for it."""
+        mock_list.return_value = ["@ok_branch", "@bad_branch"]
+        mock_route.side_effect = [
+            CommandResult(stdout="ok", stderr="", exit_code=0, branch="ok_branch", command="cmd"),
+            RuntimeError("boom"),
+        ]
+
+        results = route_all("cmd")
+
+        assert results["ok_branch"].exit_code == 0
+        assert results["bad_branch"].exit_code == -1
+        assert "boom" in results["bad_branch"].stderr
+
+
+# ---------------------------------------------------------------------------
+# detect_caller_branch_name
+# ---------------------------------------------------------------------------
+
+
+class TestDetectCallerBranchName:
+    """Tests for detect_caller_branch_name() in router_handler."""
+
+    def test_v1_passport_branch_info(self, temp_test_dir: Path):
+        """Detects branch name from v1 passport format (branch_info.branch_name)."""
+        trinity = temp_test_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text(json.dumps({
+            "branch_info": {"branch_name": "alpha"}
+        }))
+
+        result = detect_caller_branch_name(temp_test_dir)
+        assert result == "alpha"
+
+    def test_v2_passport_identity_name(self, temp_test_dir: Path):
+        """Detects branch name from v2 passport format (identity.name)."""
+        trinity = temp_test_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text(json.dumps({
+            "identity": {"name": "beta"}
+        }))
+
+        result = detect_caller_branch_name(temp_test_dir)
+        assert result == "beta"
+
+    def test_v1_takes_precedence_over_v2(self, temp_test_dir: Path):
+        """When both v1 and v2 keys exist, v1 branch_info.branch_name wins."""
+        trinity = temp_test_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text(json.dumps({
+            "branch_info": {"branch_name": "v1name"},
+            "identity": {"name": "v2name"},
+        }))
+
+        result = detect_caller_branch_name(temp_test_dir)
+        assert result == "v1name"
+
+    def test_no_passport_returns_none(self, temp_test_dir: Path):
+        """Returns None when no .trinity/passport.json exists."""
+        result = detect_caller_branch_name(temp_test_dir)
+        assert result is None
+
+    def test_corrupt_passport_returns_none(self, temp_test_dir: Path):
+        """Returns None and doesn't crash when passport.json is invalid JSON."""
+        trinity = temp_test_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text("{{{not valid json!!!")
+
+        result = detect_caller_branch_name(temp_test_dir)
+        assert result is None
+
+    def test_walks_up_from_subdirectory(self, temp_test_dir: Path):
+        """Finds passport.json in a parent directory when cwd is a subdirectory."""
+        trinity = temp_test_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text(json.dumps({
+            "branch_info": {"branch_name": "found_it"}
+        }))
+
+        sub = temp_test_dir / "deep" / "nested" / "dir"
+        sub.mkdir(parents=True)
+
+        result = detect_caller_branch_name(sub)
+        assert result == "found_it"
+
+
+# ---------------------------------------------------------------------------
+# AIPASS_CALLER_BRANCH env var
+# ---------------------------------------------------------------------------
+
+
+class TestCallerBranchEnvVar:
+    """Tests that execute_branch_command sets AIPASS_CALLER_BRANCH."""
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_sets_caller_branch_from_passport(self, mock_exec, temp_test_dir: Path):
+        """AIPASS_CALLER_BRANCH is set when passport.json exists in cwd."""
+        # Set up branch entry point
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir()
+        entry = apps_dir / "testbranch.py"
+        entry.write_text("# stub")
+
+        # Set up passport in cwd
+        cwd_dir = temp_test_dir / "caller_cwd"
+        cwd_dir.mkdir()
+        trinity = cwd_dir / ".trinity"
+        trinity.mkdir()
+        passport = trinity / "passport.json"
+        passport.write_text(json.dumps({
+            "branch_info": {"branch_name": "caller_branch"}
+        }))
+
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        with patch("aipass.drone.apps.handlers.router_handler.Path") as mock_path_cls:
+            # Make Path.cwd() return our fake cwd
+            mock_path_cls.cwd.return_value = cwd_dir
+            # But keep Path(branch_path) / ... working for find_entry_point
+            mock_path_cls.side_effect = Path
+
+            execute_branch_command(
+                branch_path=str(temp_test_dir),
+                branch_name="testbranch",
+                command="status",
+            )
+
+        env = mock_exec.call_args.kwargs["env"]
+        assert env["AIPASS_CALLER_BRANCH"] == "caller_branch"
+
+    @patch("aipass.drone.apps.handlers.router_handler.execute_command")
+    def test_no_caller_branch_without_passport(self, mock_exec, temp_test_dir: Path):
+        """AIPASS_CALLER_BRANCH is absent when no passport.json and no env var."""
+        apps_dir = temp_test_dir / "apps"
+        apps_dir.mkdir()
+        entry = apps_dir / "testbranch.py"
+        entry.write_text("# stub")
+
+        # Use a cwd with no passport
+        cwd_dir = temp_test_dir / "empty_cwd"
+        cwd_dir.mkdir()
+
+        mock_exec.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="", command=""
+        )
+
+        with patch("aipass.drone.apps.handlers.router_handler.Path") as mock_path_cls:
+            mock_path_cls.cwd.return_value = cwd_dir
+            mock_path_cls.side_effect = Path
+
+            with patch.dict(os.environ, {}, clear=False):
+                # Remove AIPASS_BRANCH_NAME if present
+                os.environ.pop("AIPASS_BRANCH_NAME", None)
+                execute_branch_command(
+                    branch_path=str(temp_test_dir),
+                    branch_name="testbranch",
+                    command="status",
+                )
+
+        env = mock_exec.call_args.kwargs["env"]
+        assert "AIPASS_CALLER_BRANCH" not in env
+
+
+# ---------------------------------------------------------------------------
+# handle_command
+# ---------------------------------------------------------------------------
+
+
+class TestHandleCommand:
+    """Tests for handle_command() in the router module."""
+
+    def test_route_no_args_returns_false(self):
+        """handle_command('route', []) returns False — not enough args."""
+        result = handle_command("route", [])
+        assert result is False
+
+    @patch("aipass.drone.apps.modules.router.route_command")
+    def test_route_with_target_and_command(self, mock_route):
+        """handle_command('route', ['@branch', 'cmd']) delegates to route_command."""
+        mock_route.return_value = CommandResult(
+            stdout="output", stderr="", exit_code=0, branch="branch", command="cmd"
+        )
+
+        result = handle_command("route", ["@branch", "cmd"])
+
+        assert result is True
+        mock_route.assert_called_once_with("@branch", "cmd", args=None)
+
+    @patch("aipass.drone.apps.modules.router.route_command")
+    def test_route_with_extra_args(self, mock_route):
+        """handle_command('route', ['@b', 'cmd', '--flag']) passes extra args."""
+        mock_route.return_value = CommandResult(
+            stdout="", stderr="", exit_code=0, branch="b", command="cmd"
+        )
+
+        result = handle_command("route", ["@b", "cmd", "--flag"])
+
+        assert result is True
+        mock_route.assert_called_once_with("@b", "cmd", args=["--flag"])
+
+    @patch("aipass.drone.apps.modules.router.route_command")
+    def test_route_nonzero_exit_returns_false(self, mock_route):
+        """handle_command('route', ...) returns False when exit_code != 0."""
+        mock_route.return_value = CommandResult(
+            stdout="", stderr="err", exit_code=1, branch="b", command="cmd"
+        )
+
+        result = handle_command("route", ["@b", "cmd"])
+
+        assert result is False
+
+    def test_route_all_no_args_returns_false(self):
+        """handle_command('route_all', []) returns False — missing command."""
+        result = handle_command("route_all", [])
+        assert result is False
+
+    @patch("aipass.drone.apps.modules.router.route_all")
+    def test_route_all_delegates(self, mock_route_all):
+        """handle_command('route_all', ['status']) delegates to route_all."""
+        mock_route_all.return_value = {
+            "a": CommandResult(stdout="", stderr="", exit_code=0, branch="a", command="status"),
+        }
+
+        result = handle_command("route_all", ["status"])
+
+        assert result is True
+        mock_route_all.assert_called_once_with("status", args=None)
+
+    @patch("aipass.drone.apps.modules.router.route_all")
+    def test_route_all_with_extra_args(self, mock_route_all):
+        """handle_command('route_all', ['cmd', '--v']) passes extra args."""
+        mock_route_all.return_value = {
+            "x": CommandResult(stdout="", stderr="", exit_code=0, branch="x", command="cmd"),
+        }
+
+        result = handle_command("route_all", ["cmd", "--v"])
+
+        assert result is True
+        mock_route_all.assert_called_once_with("cmd", args=["--v"])
+
+    @patch("aipass.drone.apps.modules.router.route_all")
+    def test_route_all_partial_failure_returns_false(self, mock_route_all):
+        """handle_command('route_all', ...) returns False if any branch fails."""
+        mock_route_all.return_value = {
+            "ok": CommandResult(stdout="", stderr="", exit_code=0, branch="ok", command="c"),
+            "bad": CommandResult(stdout="", stderr="err", exit_code=1, branch="bad", command="c"),
+        }
+
+        result = handle_command("route_all", ["c"])
+
+        assert result is False
+
+    def test_unknown_command_returns_false(self):
+        """handle_command('unknown_cmd', []) returns False."""
+        result = handle_command("unknown_cmd", [])
+        assert result is False

src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py

@@ -9,7 +9,7 @@
 """
 Agent Status Dashboard Write-Through Handler
 
-Pushes the 'agent_status' section to branch dashboards via DevPulse write_section().
+Pushes the 'agent_status' section to branch dashboards via prax write_section().
 Scans dispatch lock files and /proc to detect active and stale agents, then pushes
 to ALL branch dashboards (agent status is system-wide info).
 
@@ -244,7 +244,7 @@ def _write_section_to_all_branches(section_name: str, section_data: Dict,
         script = (
             "import sys, json\n"
             "from pathlib import Path\n"
-            "from aipass.devpulse.apps.modules.dashboard import write_section\n"
+            "from aipass.prax.apps.modules.dashboard import write_section\n"
             "data = json.loads(sys.stdin.read())\n"
             "section_name = data['section_name']\n"
             "section_data = data['section_data']\n"

src/aipass/prax/apps/handlers/dashboard/operations.py

@@ -14,13 +14,12 @@ All business logic for dashboard file operations.
 """
 
 import json
-import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Dict
 
-AIPASS_ROOT = Path.home() / "aipass_core"
-sys.path.insert(0, str(AIPASS_ROOT))
+# Resolve prax root from this file's location
+_PRAX_ROOT = Path(__file__).resolve().parents[3]  # .../prax/
 
 
 def get_dashboard_path(branch_path: Path) -> Path:
@@ -112,7 +111,7 @@ def create_fresh_dashboard(branch_path: Path) -> Dict:
         Fresh dashboard dict with warning and all sections
     """
     # Try loading from template file
-    template_file = Path.home() / "aipass_os" / "dev_central" / "devpulse" / "templates" / "DASHBOARD.template.json"
+    template_file = _PRAX_ROOT / "templates" / "DASHBOARD.template.json"
     if template_file.exists():
         try:
             template = json.loads(template_file.read_text())
@@ -137,7 +136,6 @@ def create_fresh_dashboard(branch_path: Path) -> Dict:
             "ai_mail": {"managed_by": "ai_mail", "new": 0, "opened": 0, "total": 0, "last_updated": ""},
             "flow": {"managed_by": "flow", "active_plans": 0, "recently_closed": [], "last_updated": ""},
             "memory_bank": {"managed_by": "memory_bank", "vectors_stored": 0, "notes": {}, "last_updated": ""},
-            "devpulse": {"managed_by": "devpulse", "summary": {}, "last_updated": ""},
             "commons_activity": {"managed_by": "the_commons", "mentions": 0, "new_posts_since_last_visit": 0, "new_comments_since_last_visit": 0, "last_updated": ""}
         }
     }
@@ -172,7 +170,6 @@ def ensure_dashboard_structure(branch_path: Path) -> Dict:
             "ai_mail": {"managed_by": "ai_mail", "new": 0, "opened": 0, "total": 0, "last_updated": ""},
             "flow": {"managed_by": "flow", "active_plans": 0, "recently_closed": [], "last_updated": ""},
             "memory_bank": {"managed_by": "memory_bank", "vectors_stored": 0, "notes": {}, "last_updated": ""},
-            "devpulse": {"managed_by": "devpulse", "summary": {}, "last_updated": ""},
             "commons_activity": {"managed_by": "the_commons", "mentions": 0, "new_posts_since_last_visit": 0, "new_comments_since_last_visit": 0, "last_updated": ""}
         }
     }
@@ -300,7 +297,7 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b
         - Saves back to disk
 
     Args:
-        branch_path: Path to branch root directory (e.g. Path("/home/aipass/aipass_os/dev_central/devpulse"))
+        branch_path: Path to branch root directory
         section_name: Section key (e.g. "ai_mail", "flow", "commons_activity")
         section_data: Dict of data for this section. Will be written as-is
             with an auto-added "last_updated" timestamp.
@@ -309,8 +306,8 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b
         True if saved successfully, False on any error
 
     Example:
-        >>> from aipass_os.dev_central.devpulse.apps.modules.dashboard import write_section
-        >>> write_section(Path("/home/aipass/aipass_os/flow"), "flow", {"active_plans": 2})
+        >>> from aipass.prax.apps.modules.dashboard import write_section
+        >>> write_section(branch_path, "flow", {"active_plans": 2})
         True
     """
     try:

src/aipass/prax/apps/handlers/dashboard/refresh.py

@@ -25,11 +25,20 @@ from .operations import create_fresh_dashboard, save_dashboard
 from ..central.reader import read_all_centrals
 
 # Sections managed by the refresh path — everything else is write-through only
-REFRESH_MANAGED_SECTIONS = {"ai_mail", "flow", "memory_bank", "devpulse", "commons_activity"}
+REFRESH_MANAGED_SECTIONS = {"ai_mail", "flow", "memory_bank", "commons_activity"}
+
+
+def _find_repo_root() -> Path:
+    """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
+    current = Path(__file__).resolve().parent
+    for parent in [current] + list(current.parents):
+        if (parent / "AIPASS_REGISTRY.json").exists():
+            return parent
+    return Path.cwd()
+
 
 # Infrastructure
-AIPASS_ROOT = Path.home()
-BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json"
+BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
 
 
 def _load_branch_paths() -> List[Path]:
@@ -45,6 +54,7 @@ def _load_branch_paths() -> List[Path]:
     if not BRANCH_REGISTRY.exists():
         raise FileNotFoundError(f"Branch registry not found: {BRANCH_REGISTRY}")
 
+    repo_root = _find_repo_root()
     data = json.loads(BRANCH_REGISTRY.read_text())
     branches = data.get("branches", [])
 
@@ -52,7 +62,8 @@ def _load_branch_paths() -> List[Path]:
     for branch in branches:
         path_str = branch.get("path")
         if path_str:
-            path = Path(path_str)
+            raw = Path(path_str)
+            path = raw if raw.is_absolute() else repo_root / raw
             if path.exists():
                 paths.append(path)
 
@@ -139,29 +150,6 @@ def _extract_memory_bank_section(centrals: Dict, branch_path: Path) -> Dict:
     }
 
 
-def _extract_devpulse_section(centrals: Dict, branch_name: str) -> Dict:
-    """Extract devpulse section from DEVPULSE.central.json, including dplan_counts."""
-    dp_data = centrals.get("devpulse")
-    if not dp_data:
-        return {"managed_by": "devpulse", "summary": {}, "dplan_counts": {}, "recent_activity": ""}
-
-    summaries = dp_data.get("branch_summaries", {})
-    branch_summary = summaries.get(branch_name, {})
-
-    # Include dplan_counts from central dplan_summary (Phase 4 enrichment)
-    dplan_summary = dp_data.get("dplan_summary", {})
-    dplan_counts = dplan_summary.get("dplan_counts", {})
-    recent_activity = dplan_summary.get("recent_activity", "")
-
-    return {
-        "managed_by": "devpulse",
-        "summary": branch_summary,
-        "dplan_counts": dplan_counts,
-        "recent_activity": recent_activity,
-        "last_updated": dp_data.get("last_updated", datetime.now().isoformat())
-    }
-
-
 def _extract_commons_section(centrals: Dict, branch_name: str) -> Optional[Dict]:
     """
     Extract commons_activity section from COMMONS.central.json.
@@ -277,7 +265,6 @@ def refresh_all_dashboards() -> Dict:
             dashboard["sections"]["ai_mail"] = _extract_ai_mail_section(centrals, branch_name)
             dashboard["sections"]["flow"] = _extract_flow_section(centrals, branch_name)
             dashboard["sections"]["memory_bank"] = _extract_memory_bank_section(centrals, branch_path)
-            dashboard["sections"]["devpulse"] = _extract_devpulse_section(centrals, branch_name)
 
             # Commons: preserve existing write-through data if no central file
             commons_section = _extract_commons_section(centrals, branch_name)
@@ -351,7 +338,6 @@ def refresh_single_dashboard(branch_path: Path) -> Dict:
         dashboard["sections"]["ai_mail"] = _extract_ai_mail_section(centrals, branch_name)
         dashboard["sections"]["flow"] = _extract_flow_section(centrals, branch_name)
         dashboard["sections"]["memory_bank"] = _extract_memory_bank_section(centrals, branch_path)
-        dashboard["sections"]["devpulse"] = _extract_devpulse_section(centrals, branch_name)
 
         # Commons: preserve existing write-through data if no central file
         commons_section = _extract_commons_section(centrals, branch_name)

src/aipass/prax/apps/handlers/dashboard/status.py

@@ -17,8 +17,17 @@ import json
 from pathlib import Path
 from typing import Dict, List
 
-AIPASS_ROOT = Path.home()
-BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json"
+
+def _find_repo_root() -> Path:
+    """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
+    current = Path(__file__).resolve().parent
+    for parent in [current] + list(current.parents):
+        if (parent / "AIPASS_REGISTRY.json").exists():
+            return parent
+    return Path.cwd()
+
+
+BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
 
 
 def calculate_quick_status(sections: Dict) -> Dict:
@@ -81,13 +90,18 @@ def get_branch_paths() -> List[Path]:
     if not BRANCH_REGISTRY.exists():
         raise FileNotFoundError(f"Branch registry not found: {BRANCH_REGISTRY}")
 
+    repo_root = _find_repo_root()
     data = json.loads(BRANCH_REGISTRY.read_text())
-    return [Path(b.get("path", "")) for b in data.get("branches", [])]
+    paths = []
+    for b in data.get("branches", []):
+        raw = Path(b.get("path", ""))
+        paths.append(raw if raw.is_absolute() else repo_root / raw)
+    return paths
 
 
 def resolve_branch_path(branch_ref: str) -> Path:
     """
-    Resolve @branch reference to filesystem path via BRANCH_REGISTRY.json.
+    Resolve @branch reference to filesystem path via AIPASS_REGISTRY.json.
 
     Handler-layer function: performs file I/O to read registry and
     resolve branch name to its directory path.
@@ -104,12 +118,14 @@ def resolve_branch_path(branch_ref: str) -> Path:
     name = branch_ref.lstrip("@").upper()
 
     if not BRANCH_REGISTRY.exists():
-        raise FileNotFoundError("BRANCH_REGISTRY.json not found")
+        raise FileNotFoundError("AIPASS_REGISTRY.json not found")
 
+    repo_root = _find_repo_root()
     data = json.loads(BRANCH_REGISTRY.read_text())
     for branch in data.get("branches", []):
         if branch.get("name", "").upper() == name:
-            path = Path(branch["path"])
+            raw = Path(branch["path"])
+            path = raw if raw.is_absolute() else repo_root / raw
             if path.exists():
                 return path
             raise FileNotFoundError(f"Branch path does not exist: {path}")

src/aipass/prax/apps/handlers/dashboard/template_differ.py

@@ -23,32 +23,42 @@ Independence:
 """
 
 import json
-import sys
 from pathlib import Path
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any, Optional
+
+# =============================================================================
+# PATH RESOLUTION
+# =============================================================================
+
+_PRAX_ROOT = Path(__file__).resolve().parents[3]  # .../prax/
+
+
+def _find_repo_root() -> Path:
+    """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
+    current = Path(__file__).resolve().parent
+    for parent in [current] + list(current.parents):
+        if (parent / "AIPASS_REGISTRY.json").exists():
+            return parent
+    return Path.cwd()
 
-# Infrastructure setup
-AIPASS_ROOT = Path.home()
-sys.path.insert(0, str(AIPASS_ROOT))
 
 # =============================================================================
 # CONSTANTS
 # =============================================================================
 
-DEVPULSE_ROOT = AIPASS_ROOT / "aipass_os" / "dev_central" / "devpulse"
-TEMPLATE_DIR = DEVPULSE_ROOT / "templates"
+TEMPLATE_DIR = _PRAX_ROOT / "templates"
 TEMPLATE_FILE = TEMPLATE_DIR / "DASHBOARD.template.json"
-BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json"
+BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
 
 # Deprecated sections that should be flagged for removal
-DEPRECATED_SECTIONS = ["bulletin_board"]
+DEPRECATED_SECTIONS = ["bulletin_board", "devpulse"]
 
 # Deprecated quick_status keys that should be flagged
 DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins"]
 
 # Required sections (from template)
 REQUIRED_SECTIONS = [
-    "ai_mail", "flow", "memory_bank", "devpulse", "commons_activity"
+    "ai_mail", "flow", "memory_bank", "commons_activity"
 ]
 
 
@@ -107,7 +117,6 @@ def _diff_branch(branch_name: str, branch_path: Path, template: dict) -> Dict[st
         return result
 
     # Check for missing required sections
-    template_sections = template.get("sections", {})
     for section_name in REQUIRED_SECTIONS:
         if section_name not in sections:
             result["additions"].append(f"{section_name} section")
@@ -202,9 +211,12 @@ def diff_dashboard_template(branch_name: Optional[str] = None) -> Dict[str, Any]
                 "summary": {}
             }
 
+    repo_root = _find_repo_root()
+
     for branch in branches:
         bname = branch.get("name", "UNKNOWN").upper()
-        bpath = Path(branch.get("path", ""))
+        raw_path = Path(branch.get("path", ""))
+        bpath = raw_path if raw_path.is_absolute() else repo_root / raw_path
 
         if not bpath.exists():
             branch_diff = {

src/aipass/prax/apps/handlers/dashboard/template_pusher.py

@@ -25,27 +25,37 @@ Independence:
 
 import json
 import copy
-import sys
 from pathlib import Path
 from datetime import datetime
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any, List
+
+# =============================================================================
+# PATH RESOLUTION
+# =============================================================================
+
+_PRAX_ROOT = Path(__file__).resolve().parents[3]  # .../prax/
+
+
+def _find_repo_root() -> Path:
+    """Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
+    current = Path(__file__).resolve().parent
+    for parent in [current] + list(current.parents):
+        if (parent / "AIPASS_REGISTRY.json").exists():
+            return parent
+    return Path.cwd()
 
-# Infrastructure setup
-AIPASS_ROOT = Path.home()
-sys.path.insert(0, str(AIPASS_ROOT))
 
 # =============================================================================
 # CONSTANTS
 # =============================================================================
 
-DEVPULSE_ROOT = AIPASS_ROOT / "aipass_os" / "dev_central" / "devpulse"
-TEMPLATE_DIR = DEVPULSE_ROOT / "templates"
+TEMPLATE_DIR = _PRAX_ROOT / "templates"
 TEMPLATE_FILE = TEMPLATE_DIR / "DASHBOARD.template.json"
 VERSION_FILE = TEMPLATE_DIR / ".dashboard_version.json"
-BRANCH_REGISTRY = AIPASS_ROOT / "BRANCH_REGISTRY.json"
+BRANCH_REGISTRY = _find_repo_root() / "AIPASS_REGISTRY.json"
 
 # Deprecated sections to REMOVE during push
-DEPRECATED_SECTIONS = ["bulletin_board"]
+DEPRECATED_SECTIONS = ["bulletin_board", "devpulse"]
 
 # Deprecated quick_status keys to REMOVE during push
 DEPRECATED_QUICK_STATUS_KEYS = ["pending_bulletins"]
@@ -71,11 +81,6 @@ REQUIRED_SECTIONS = {
         "notes": {},
         "last_updated": ""
     },
-    "devpulse": {
-        "managed_by": "devpulse",
-        "summary": {},
-        "last_updated": ""
-    },
     "commons_activity": {
         "managed_by": "the_commons",
         "mentions": 0,
@@ -229,9 +234,12 @@ def push_dashboard_template(dry_run: bool = False) -> Dict[str, Any]:
     branches = [b for b in registry.get("branches", []) if b.get("status") == "active"]
     branches_updated_list: List[str] = []
 
+    repo_root = _find_repo_root()
+
     for branch in branches:
         branch_name = branch.get("name", "UNKNOWN").upper()
-        branch_path = Path(branch.get("path", ""))
+        raw_path = Path(branch.get("path", ""))
+        branch_path = raw_path if raw_path.is_absolute() else repo_root / raw_path
         result["branches_scanned"] += 1
 
         if not branch_path.exists():

src/aipass/prax/apps/modules/dashboard.py

@@ -79,11 +79,6 @@ DASHBOARD_TEMPLATE = {
             "notes": {},
             "last_updated": ""
         },
-        "devpulse": {
-            "managed_by": "devpulse",
-            "summary": {},
-            "last_updated": ""
-        },
         "commons_activity": {
             "managed_by": "the_commons",
             "mentions": 0,
@@ -115,7 +110,7 @@ def update_section(
     Update a specific section in branch dashboard (legacy wrapper).
 
     For new integrations, prefer write_section() from
-    aipass_os.dev_central.devpulse.apps.handlers.dashboard.operations
+    aipass.prax.apps.handlers.dashboard.operations
     which is self-contained and dependency-free.
 
     Args:
@@ -170,19 +165,19 @@ def print_help():
     console.print("  template-status  - Show template version and push info")
     console.print()
     console.print("[yellow]USAGE:[/yellow]")
-    console.print("  drone @devpulse dashboard status")
-    console.print("  drone @devpulse dashboard template")
-    console.print("  drone @devpulse dashboard refresh          # refresh current branch")
-    console.print("  drone @devpulse dashboard refresh @flow    # refresh specific branch")
-    console.print("  drone @devpulse dashboard refresh --all    # refresh all branches")
-    console.print("  drone @devpulse dashboard push-template             # push to all branches")
-    console.print("  drone @devpulse dashboard push-template --dry-run   # preview changes")
-    console.print("  drone @devpulse dashboard diff-template             # diff all branches")
-    console.print("  drone @devpulse dashboard diff-template --branch FLOW  # diff single branch")
-    console.print("  drone @devpulse dashboard template-status           # version info")
+    console.print("  drone @prax dashboard status")
+    console.print("  drone @prax dashboard template")
+    console.print("  drone @prax dashboard refresh          # refresh current branch")
+    console.print("  drone @prax dashboard refresh @flow    # refresh specific branch")
+    console.print("  drone @prax dashboard refresh --all    # refresh all branches")
+    console.print("  drone @prax dashboard push-template             # push to all branches")
+    console.print("  drone @prax dashboard push-template --dry-run   # preview changes")
+    console.print("  drone @prax dashboard diff-template             # diff all branches")
+    console.print("  drone @prax dashboard diff-template --branch FLOW  # diff single branch")
+    console.print("  drone @prax dashboard template-status           # version info")
     console.print()
     console.print("[yellow]PROGRAMMATIC (write-through API):[/yellow]")
-    console.print("  from aipass_os.dev_central.devpulse.apps.modules.dashboard import write_section")
+    console.print("  from aipass.prax.apps.modules.dashboard import write_section")
     console.print("  write_section(branch_path, 'ai_mail', {'new': 3, 'total': 5})")
     console.print()
 

src/aipass/prax/templates/.dashboard_version.json

@@ -10,35 +10,22 @@
       "description": "Branch dashboard template - v3 schema with write-through sections"
     }
   },
-  "last_push": "2026-02-25 18:23:16",
+  "last_push": "2026-03-14 21:30:50",
   "last_push_branches": [
-    ".VSCODE",
     "AI_MAIL",
     "API",
-    "ASSISTANT",
-    "BACKUP_SYSTEM",
+    "BACKUP",
     "CLI",
-    "CORTEX",
+    "COMMONS",
+    "DAEMON",
     "DEVPULSE",
-    "DEV_CENTRAL",
     "DRONE",
     "FLOW",
-    "GIT_MANAGER",
-    "GROWTH",
-    "MCP_SERVERS",
-    "MEMORY_BANK",
-    "NEXUS",
-    "PERMISSIONS",
+    "MEMORY",
     "PRAX",
-    "PROJECTS",
-    "SEED",
-    "SPEAKEASY",
-    "TEAM_1",
-    "TEAM_2",
-    "TEAM_3",
-    "TEST",
-    "THE_COMMONS",
-    "TRIGGER",
-    "VERA"
+    "SEEDGO",
+    "SKILLS",
+    "SPAWN",
+    "TRIGGER"
   ]
 }

src/aipass/prax/templates/DASHBOARD.template.json

@@ -30,11 +30,6 @@
       "notes": {},
       "last_updated": ""
     },
-    "devpulse": {
-      "managed_by": "devpulse",
-      "summary": {},
-      "last_updated": ""
-    },
     "commons_activity": {
       "managed_by": "the_commons",
       "mentions": 0,