feat(system): S90: ai_mail bidirectional cross-project email (DPLAN-0121 Phase 3) + reply-path delivery

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

src/aipass/ai_mail/.seedgo/bypass.json

@@ -78,7 +78,7 @@
     {
       "file": "apps/handlers/email/reply.py",
       "standard": "handlers",
-      "reason": "Imports users.branch_detection.get_branch_info_from_registry and registry.read.get_all_branches — same-branch cross-handler imports for sender identity and branch listing on reply."
+      "reason": "Same-branch cross-handler lazy imports: users.branch_detection.get_branch_info_from_registry, registry.read.get_all_branches (sender identity + branch listing), email.delivery.deliver_email_to_branch, email.inbox_cleanup.mark_as_closed_and_archive (reply delivery + close). All lazy-imported inside functions to avoid circular imports."
     },
     {
       "file": "apps/handlers/email/send.py",
@@ -234,6 +234,31 @@
       "file": "apps/handlers/users/branch_detection.py",
       "standard": "handlers",
       "reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036."
+    },
+    {
+      "file": "tests/test_delivery.py",
+      "standard": "architecture",
+      "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
+    },
+    {
+      "file": "tests/test_delivery.py",
+      "standard": "encapsulation",
+      "reason": "Unit tests must import handlers directly to exercise internal functions (_migrate_inbox_format, _is_private_branch_email, _resolve_reply_path). Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_send_identity.py",
+      "standard": "architecture",
+      "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
+    },
+    {
+      "file": "tests/test_send_identity.py",
+      "standard": "encapsulation",
+      "reason": "Unit tests must import handlers directly to exercise internal detection functions. Module entry-point-only rule does not apply to tests."
+    },
+    {
+      "file": "tests/test_send_identity.py",
+      "standard": "documentation",
+      "reason": "Test helper functions (_load_active_source, fixtures) are private/internal — docstring requirement does not apply to test infrastructure."
     }
   ],
   "notes": {

src/aipass/ai_mail/apps/handlers/email/delivery.py

@@ -14,6 +14,7 @@ Independent handler - no module dependencies.
 """
 
 import json
+import os
 import uuid
 import subprocess
 from pathlib import Path
@@ -130,6 +131,27 @@ def _is_private_branch_email(email: str) -> bool:
     return False
 
 
+def _resolve_reply_path() -> str:
+    """Detect the caller's ai_mail inbox path from AIPASS_CALLER_CWD env var.
+
+    Used to store a reply_path on delivered messages so cross-project replies
+    can bypass registry lookup and write directly to the sender's inbox.
+
+    Returns the absolute path to the caller's inbox.json, or empty string if
+    AIPASS_CALLER_CWD is not set or no inbox directory is found.
+    """
+    caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+    if not caller_cwd:
+        return ""
+    # Check the CWD itself and up to 5 parent directories for .ai_mail.local/
+    candidate = Path(caller_cwd)
+    for path in [candidate] + list(candidate.parents)[:5]:
+        inbox = path / ".ai_mail.local" / "inbox.json"
+        if inbox.exists():
+            return str(inbox)
+    return ""
+
+
 def deliver_email_to_branch(
     to_branch: str,
     email_data: Dict,
@@ -251,6 +273,12 @@ def deliver_email_to_branch(
             if email_data.get('dispatched_to'):
                 message["dispatched_to"] = email_data['dispatched_to']
 
+            # Store reply_path for cross-project replies.
+            # Pass-through from email_data, or auto-detect from AIPASS_CALLER_CWD.
+            reply_path = email_data.get("reply_path") or _resolve_reply_path()
+            if reply_path:
+                message["reply_path"] = reply_path
+
             # Prepend message to inbox (newest first)
             inbox_data["messages"].insert(0, message)
             inbox_data["total_messages"] = len(inbox_data["messages"])

src/aipass/ai_mail/apps/handlers/email/reply.py

@@ -133,6 +133,12 @@ def send_reply(
             break
 
     if not target_branch:
+        # Fallback: cross-project delivery via reply_path stored at receive time
+        stored_reply_path = original_email.get("reply_path")
+        if stored_reply_path:
+            return _deliver_via_reply_path(
+                stored_reply_path, reply_email_data, from_branch_path, original_email
+            )
         return False, f"Could not find branch for {reply_destination}", None
 
     # Deliver the reply (pass email address, not path)
@@ -161,6 +167,78 @@ def send_reply(
     return True, f"Reply sent to {reply_destination}, original closed", reply_id
 
 
+def _deliver_via_reply_path(
+    reply_path: str,
+    reply_email_data: Dict,
+    from_branch_path: Path,
+    original_email: Dict,
+) -> Tuple[bool, str, Optional[str]]:
+    """Deliver reply directly to an external project's inbox via stored reply_path.
+
+    Used when the recipient is not in the AIPass registry (cross-project reply).
+    Writes directly to the inbox.json at the stored path, then saves to sent/
+    and closes the original.
+
+    Args:
+        reply_path: Absolute path to the target inbox.json (stored at receive time).
+        reply_email_data: The reply email dict to deliver.
+        from_branch_path: Path to the replying branch (for sent folder and close).
+        original_email: The original email being replied to (for close).
+
+    Returns:
+        Tuple of (success, message, reply_id or None)
+    """
+    inbox_file = Path(reply_path)
+    if not inbox_file.exists():
+        return False, f"reply_path inbox not found: {reply_path}", None
+
+    try:
+        with open(inbox_file, "r", encoding="utf-8") as f:
+            inbox_data = json.load(f)
+    except Exception as e:
+        logger.warning("[reply] _deliver_via_reply_path read failed %s: %s", reply_path, e)
+        return False, f"Failed to read target inbox: {e}", None
+
+    reply_id = str(uuid.uuid4())[:8]
+    reply_email_data["id"] = reply_id
+
+    inbox_data.setdefault("messages", []).insert(0, reply_email_data)
+    inbox_data["total_messages"] = len(inbox_data["messages"])
+    new_count = sum(1 for m in inbox_data["messages"]
+                    if m.get("status") == "new" or not m.get("read", False))
+    inbox_data["unread_count"] = new_count
+
+    try:
+        with open(inbox_file, "w", encoding="utf-8") as f:
+            json.dump(inbox_data, f, indent=2, ensure_ascii=False)
+    except Exception as e:
+        logger.warning("[reply] _deliver_via_reply_path write failed %s: %s", reply_path, e)
+        return False, f"Failed to write to target inbox: {e}", None
+
+    logger.info("[reply] Cross-project reply delivered to %s", reply_path)
+
+    # Save to sender's sent folder
+    sent_folder = from_branch_path / ".ai_mail.local" / "sent"
+    sent_folder.mkdir(parents=True, exist_ok=True)
+    sent_file = sent_folder / f"{reply_id}.json"
+    try:
+        with open(sent_file, "w", encoding="utf-8") as f:
+            json.dump(reply_email_data, f, indent=2)
+    except Exception as e:
+        logger.warning("[reply] failed to save sent copy: %s", e)
+
+    # Auto-close the original email
+    from aipass.ai_mail.apps.handlers.email.inbox_cleanup import mark_as_closed_and_archive
+    original_id = original_email.get("id")
+    if original_id:
+        close_success, close_msg = mark_as_closed_and_archive(from_branch_path, original_id)
+        if not close_success:
+            return True, f"Reply sent (warning: original not closed: {close_msg})", reply_id
+
+    destination = reply_email_data.get("to", reply_path)
+    return True, f"Reply sent to {destination} via reply_path, original closed", reply_id
+
+
 if __name__ == "__main__":
     from aipass.cli.apps.modules import console
     console.print("\n" + "="*70)

src/aipass/ai_mail/apps/handlers/users/branch_detection.py

@@ -31,6 +31,31 @@ from aipass.ai_mail.apps.handlers.paths import find_repo_root
 BRANCH_REGISTRY_PATH = find_repo_root() / "AIPASS_REGISTRY.json"
 
 
+def _find_caller_registry() -> Optional[Path]:
+    """Find the caller's AIPASS_REGISTRY.json by walking up from AIPASS_CALLER_CWD.
+
+    Used to resolve external project branches that aren't in the AIPass registry.
+    Skips the AIPass registry itself to avoid redundant double-lookup.
+
+    Returns:
+        Path to the caller's registry file, or None if not found or same as main registry.
+    """
+    caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+    if not caller_cwd:
+        return None
+    candidate = Path(caller_cwd)
+    aipass_registry = BRANCH_REGISTRY_PATH.resolve()
+    for path in [candidate] + list(candidate.parents)[:10]:
+        registry = path / "AIPASS_REGISTRY.json"
+        if registry.exists():
+            try:
+                if registry.resolve() != aipass_registry:
+                    return registry
+            except Exception as e:
+                logger.warning("[identity] _find_caller_registry() resolve failed for %s: %s", registry, e)
+    return None
+
+
 def _get_branches_list(registry: dict) -> list:
     """Normalize branches from registry to a list of dicts.
 
@@ -111,23 +136,31 @@ def _lookup_branch_by_name(branch_name: str) -> Optional[Dict]:
     Returns:
         Dict with branch info from registry, or None if not found
     """
-    if not BRANCH_REGISTRY_PATH.exists():
-        return None
+    name_lower = branch_name.lower()
 
-    try:
-        with open(BRANCH_REGISTRY_PATH, 'r', encoding='utf-8') as f:
-            registry = json.load(f)
+    if BRANCH_REGISTRY_PATH.exists():
+        try:
+            with open(BRANCH_REGISTRY_PATH, 'r', encoding='utf-8') as f:
+                registry = json.load(f)
+            for branch in _get_branches_list(registry):
+                if branch.get("name", "").lower() == name_lower:
+                    return branch
+        except Exception as e:
+            logger.warning("[identity] _lookup_branch_by_name(%s) failed: %s", branch_name, e)
 
-        name_lower = branch_name.lower()
-        for branch in _get_branches_list(registry):
-            if branch.get("name", "").lower() == name_lower:
-                return branch
+    # Fallback: caller's registry (external project branches not in AIPass registry)
+    caller_registry = _find_caller_registry()
+    if caller_registry:
+        try:
+            with open(caller_registry, 'r', encoding='utf-8') as f:
+                registry = json.load(f)
+            for branch in _get_branches_list(registry):
+                if branch.get("name", "").lower() == name_lower:
+                    return branch
+        except Exception as e:
+            logger.warning("[identity] _lookup_branch_by_name(%s) caller registry %s failed: %s", branch_name, caller_registry, e)
 
-        return None
-
-    except Exception as e:
-        logger.warning("[identity] _lookup_branch_by_name(%s) failed: %s", branch_name, e)
-        return None
+    return None
 
 
 def find_branch_root(start_path: Path) -> Optional[Path]:
@@ -170,32 +203,43 @@ def get_branch_info_from_registry(branch_path: Path) -> Optional[Dict]:
     Returns:
         Dict with branch info from registry, or None if not found
     """
-    if not BRANCH_REGISTRY_PATH.exists():
-        return None
+    branch_path_resolved = branch_path.resolve()
 
-    try:
-        with open(BRANCH_REGISTRY_PATH, 'r', encoding='utf-8') as f:
-            registry = json.load(f)
+    if BRANCH_REGISTRY_PATH.exists():
+        try:
+            with open(BRANCH_REGISTRY_PATH, 'r', encoding='utf-8') as f:
+                registry = json.load(f)
+            registry_dir = BRANCH_REGISTRY_PATH.parent
+            for branch in _get_branches_list(registry):
+                reg_path = Path(branch["path"])
+                if not reg_path.is_absolute():
+                    reg_path = (registry_dir / reg_path).resolve()
+                else:
+                    reg_path = reg_path.resolve()
+                if reg_path == branch_path_resolved:
+                    return branch
+        except Exception as e:
+            logger.warning("[identity] get_branch_info_from_registry(%s) failed: %s", branch_path, e)
 
-        registry_dir = BRANCH_REGISTRY_PATH.parent
-        branch_path_resolved = branch_path.resolve()
+    # Fallback: caller's registry (external project branches not in AIPass registry)
+    caller_registry = _find_caller_registry()
+    if caller_registry:
+        try:
+            with open(caller_registry, 'r', encoding='utf-8') as f:
+                registry = json.load(f)
+            registry_dir = caller_registry.parent
+            for branch in _get_branches_list(registry):
+                reg_path = Path(branch["path"])
+                if not reg_path.is_absolute():
+                    reg_path = (registry_dir / reg_path).resolve()
+                else:
+                    reg_path = reg_path.resolve()
+                if reg_path == branch_path_resolved:
+                    return branch
+        except Exception as e:
+            logger.warning("[identity] get_branch_info_from_registry(%s) caller registry failed: %s", branch_path, e)
 
-        # Search registry for matching path
-        for branch in _get_branches_list(registry):
-            reg_path = Path(branch["path"])
-            # Resolve relative paths against registry location, not CWD
-            if not reg_path.is_absolute():
-                reg_path = (registry_dir / reg_path).resolve()
-            else:
-                reg_path = reg_path.resolve()
-            if reg_path == branch_path_resolved:
-                return branch
-
-        return None
-
-    except Exception as e:
-        logger.warning("[identity] get_branch_info_from_registry(%s) failed: %s", branch_path, e)
-        return None
+    return None
 
 
 if __name__ == "__main__":

src/aipass/ai_mail/tests/test_delivery.py

@@ -19,6 +19,7 @@ import aipass.ai_mail.apps.handlers.email.delivery as delivery_mod
 from aipass.ai_mail.apps.handlers.email.delivery import (
     _migrate_inbox_format,
     _is_private_branch_email,
+    _resolve_reply_path,
     deliver_email_to_branch,
 )
 
@@ -429,3 +430,66 @@ def test_deliver_path_input_unresolvable(repo_root, noop_inbox_lock):
 
     assert success is False
     assert "Could not resolve path" in error
+
+
+# ---- _resolve_reply_path() tests ------------------------------
+
+
+def test_resolve_reply_path_no_env(monkeypatch):
+    """Returns empty string when AIPASS_CALLER_CWD is not set."""
+    monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False)
+    assert _resolve_reply_path() == ""
+
+
+def test_resolve_reply_path_inbox_in_cwd(tmp_path, monkeypatch):
+    """Returns inbox path when .ai_mail.local/inbox.json exists in caller CWD."""
+    inbox_dir = tmp_path / ".ai_mail.local"
+    inbox_dir.mkdir(parents=True)
+    inbox_file = inbox_dir / "inbox.json"
+    inbox_file.write_text("{}", encoding="utf-8")
+
+    monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+    assert _resolve_reply_path() == str(inbox_file)
+
+
+def test_resolve_reply_path_inbox_in_parent(tmp_path, monkeypatch):
+    """Returns inbox path when .ai_mail.local/inbox.json exists in a parent of caller CWD."""
+    inbox_dir = tmp_path / ".ai_mail.local"
+    inbox_dir.mkdir(parents=True)
+    inbox_file = inbox_dir / "inbox.json"
+    inbox_file.write_text("{}", encoding="utf-8")
+
+    nested_cwd = tmp_path / "src" / "feature"
+    nested_cwd.mkdir(parents=True)
+    monkeypatch.setenv("AIPASS_CALLER_CWD", str(nested_cwd))
+
+    assert _resolve_reply_path() == str(inbox_file)
+
+
+def test_resolve_reply_path_no_inbox(tmp_path, monkeypatch):
+    """Returns empty string when no .ai_mail.local/inbox.json found in tree."""
+    monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
+    assert _resolve_reply_path() == ""
+
+
+def test_deliver_stores_reply_path_from_env(tmp_path, repo_root, noop_inbox_lock, monkeypatch):
+    """reply_path env detection is stored on delivered message."""
+    branches = _setup_branch(tmp_path)
+    caller_project = tmp_path / "external_project"
+    inbox_dir = caller_project / ".ai_mail.local"
+    inbox_dir.mkdir(parents=True)
+    inbox_file = inbox_dir / "inbox.json"
+    inbox_file.write_text("{}", encoding="utf-8")
+
+    monkeypatch.setenv("AIPASS_CALLER_CWD", str(caller_project))
+
+    with patch.object(delivery_mod, "get_all_branches", return_value=branches):
+        success, _ = deliver_email_to_branch("@target", _make_email_data())
+
+    assert success is True
+    inbox_file_target = Path(branches[0]["path"]) / ".ai_mail.local" / "inbox.json"
+    with open(inbox_file_target, "r", encoding="utf-8") as f:
+        inbox = json.load(f)
+    msg = inbox["messages"][0]
+    assert "reply_path" in msg
+    assert msg["reply_path"] == str(inbox_file)

src/aipass/ai_mail/tests/test_send_identity.py

@@ -41,6 +41,7 @@ from aipass.ai_mail.apps.handlers.users.branch_detection import (
     find_branch_root,
     get_branch_info_from_registry,
     _lookup_branch_by_name,
+    _find_caller_registry,
 )
 from aipass.ai_mail.apps.handlers.email.send import resolve_sender_info
 from aipass.ai_mail.apps.handlers.email.send_args import parse_send_args
@@ -634,3 +635,125 @@ class TestAntiRegression:
         assert result["reply_to"] == "@flow"
         assert result["recipients"] == ["@spawn"]
         assert result["mode"] == "direct"
+
+
+# ─── _find_caller_registry() tests ───────────────────────
+
+
+class TestFindCallerRegistry:
+    """Tests for the caller registry fallback used for external project branches."""
+
+    def test_returns_none_when_no_env(self, clean_env):
+        """Returns None when AIPASS_CALLER_CWD is not set."""
+        result = _find_caller_registry()
+        assert result is None
+
+    def test_returns_none_when_no_registry_in_tree(self, clean_env, tmp_path):
+        """Returns None when no AIPASS_REGISTRY.json found under AIPASS_CALLER_CWD."""
+        os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
+        result = _find_caller_registry()
+        assert result is None
+
+    def test_finds_registry_in_cwd(self, clean_env, tmp_path):
+        """Returns registry path when AIPASS_REGISTRY.json exists in caller CWD."""
+        registry = tmp_path / "AIPASS_REGISTRY.json"
+        registry.write_text('{"branches": []}', encoding="utf-8")
+
+        os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH",
+                   tmp_path / "other" / "AIPASS_REGISTRY.json"):
+            result = _find_caller_registry()
+
+        assert result == registry
+
+    def test_finds_registry_in_parent(self, clean_env, tmp_path):
+        """Returns registry path when found in a parent directory of caller CWD."""
+        registry = tmp_path / "AIPASS_REGISTRY.json"
+        registry.write_text('{"branches": []}', encoding="utf-8")
+        nested = tmp_path / "src" / "vera"
+        nested.mkdir(parents=True)
+
+        os.environ["AIPASS_CALLER_CWD"] = str(nested)
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH",
+                   tmp_path / "other" / "AIPASS_REGISTRY.json"):
+            result = _find_caller_registry()
+
+        assert result == registry
+
+    def test_skips_aipass_registry_itself(self, clean_env, tmp_path):
+        """Returns None when found registry is the same as BRANCH_REGISTRY_PATH."""
+        registry = tmp_path / "AIPASS_REGISTRY.json"
+        registry.write_text('{"branches": []}', encoding="utf-8")
+
+        os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry):
+            result = _find_caller_registry()
+
+        assert result is None
+
+
+# ─── Caller registry fallback tests ──────────────────────
+
+
+class TestCallerRegistryFallback:
+    """Tests for branch lookup fallback to external project registry."""
+
+    def test_lookup_by_name_falls_back_to_caller_registry(self, clean_env, tmp_path):
+        """_lookup_branch_by_name finds external branch via caller registry."""
+        caller_registry = tmp_path / "AIPASS_REGISTRY.json"
+        caller_registry.write_text(json.dumps({"branches": [
+            {"name": "VERA", "path": str(tmp_path / "vera"), "email": "@vera", "status": "active"}
+        ]}), encoding="utf-8")
+
+        empty_aipass = tmp_path / "other" / "AIPASS_REGISTRY.json"
+        empty_aipass.parent.mkdir(parents=True)
+        empty_aipass.write_text('{"branches": []}', encoding="utf-8")
+
+        os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", empty_aipass):
+            result = _lookup_branch_by_name("vera")
+
+        assert result is not None
+        assert result["email"] == "@vera"
+        assert result["name"] == "VERA"
+
+    def test_lookup_by_name_prefers_aipass_registry(self, clean_env, tmp_path):
+        """_lookup_branch_by_name returns AIPass result first when branch exists in both."""
+        aipass_registry = tmp_path / "AIPASS_REGISTRY.json"
+        aipass_registry.write_text(json.dumps({"branches": [
+            {"name": "SPAWN", "path": str(tmp_path / "spawn"), "email": "@spawn-aipass", "status": "active"}
+        ]}), encoding="utf-8")
+
+        caller_dir = tmp_path / "external"
+        caller_dir.mkdir()
+        caller_registry = caller_dir / "AIPASS_REGISTRY.json"
+        caller_registry.write_text(json.dumps({"branches": [
+            {"name": "SPAWN", "path": str(tmp_path / "other_spawn"), "email": "@spawn-external", "status": "active"}
+        ]}), encoding="utf-8")
+
+        os.environ["AIPASS_CALLER_CWD"] = str(caller_dir)
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", aipass_registry):
+            result = _lookup_branch_by_name("spawn")
+
+        assert result is not None
+        assert result["email"] == "@spawn-aipass"
+
+    def test_get_branch_info_falls_back_to_caller_registry(self, clean_env, tmp_path):
+        """get_branch_info_from_registry finds external branch path via caller registry."""
+        vera_dir = tmp_path / "vera_studio" / "src" / "vera"
+        vera_dir.mkdir(parents=True)
+
+        caller_registry = tmp_path / "vera_studio" / "AIPASS_REGISTRY.json"
+        caller_registry.write_text(json.dumps({"branches": [
+            {"name": "VERA", "path": str(vera_dir), "email": "@vera", "status": "active"}
+        ]}), encoding="utf-8")
+
+        empty_aipass = tmp_path / "AIPASS_REGISTRY.json"
+        empty_aipass.write_text('{"branches": []}', encoding="utf-8")
+
+        os.environ["AIPASS_CALLER_CWD"] = str(tmp_path / "vera_studio" / "src" / "vera")
+        with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", empty_aipass):
+            result = get_branch_info_from_registry(vera_dir)
+
+        assert result is not None
+        assert result["email"] == "@vera"

src/aipass/prax/apps/handlers/monitoring/branch_detector.py

@@ -437,8 +437,13 @@ class BranchDetector:
                     self.log_map[name] = branch_name
                     return branch_name
 
-            # Fallback: split on underscore for branches not in registry
-            # Log files follow pattern: branch_operation.log
+            # Full path: use path detection before falling back to stem splitting.
+            # This ensures ai_mail/logs/mail_*.log resolves to AI_MAIL via branch_map
+            # rather than returning a truncated stem like MAIL.
+            if '/' in log_file:
+                return self.detect_from_path(log_file)
+
+            # Bare filename: fallback to stem splitting
             if '_' in name:
                 parts = name.split('_')
                 first_part = parts[0].upper()
@@ -451,10 +456,6 @@ class BranchDetector:
                 self.log_map[name] = name_upper
                 return name_upper
 
-            # If we have a full path, try path detection
-            if '/' in log_file:
-                return self.detect_from_path(log_file)
-
             logger.info(f"Could not detect branch from log: {log_file}")
             return 'UNKNOWN'
 

src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py

@@ -29,13 +29,13 @@ from aipass.prax.apps.modules.logger import get_direct_logger
 logger = get_direct_logger()
 
 # Trigger integration (graceful fallback if unavailable)
+_trigger_available = False
 try:
     from aipass.trigger.apps.modules.core import trigger
     _trigger_available = True
 except ImportError as e:
     logger.info(f"[monitor] trigger module not available, falling back: {e}")
     trigger = None  # type: ignore[assignment]
-    _trigger_available = False
 
 # Monitoring subsystem imports
 from aipass.prax.apps.handlers.monitoring.event_queue import MonitoringEvent, MonitoringQueue
@@ -128,10 +128,11 @@ class MonitoringFileHandler(FileSystemEventHandler):
 
     def _fire_trigger(self, event_name: str, **kwargs):
         """Fire a trigger event for cross-module integration."""
-        if not _trigger_available:
-            return
         try:
-            trigger.fire(event_name, **kwargs)  # type: ignore[union-attr]
+            from aipass.trigger.apps.modules.core import trigger as _trigger
+            _trigger.fire(event_name, **kwargs)
+        except ImportError:
+            pass  # trigger not available — silently skip
         except Exception as e:
             logger.warning(f"[monitor] trigger.fire('{event_name}') failed: {e}")
 
@@ -209,6 +210,16 @@ class MonitoringFileHandler(FileSystemEventHandler):
             idx = parts.index('aipass')
             if idx + 1 < len(parts):
                 return parts[idx + 1].upper()
+        # External project under ~/Projects/: use branch_detector for full label
+        # e.g. ~/Projects/AIPL/src/polyglot/ → AIPL/POLYGLOT
+        projects_base = Path.home() / 'Projects'
+        try:
+            Path(cwd).relative_to(projects_base)
+            result = detect_branch_from_path(cwd)
+            if result and result != 'UNKNOWN':
+                return result
+        except ValueError:
+            logger.info(f"[monitor] CWD not under ~/Projects/: {cwd}")
         # Fallback: check src/{name} for branches outside aipass namespace
         if 'src' in parts:
             idx = parts.index('src')

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

@@ -9,41 +9,13 @@
 """
 PRAX Monitor Module - Mission Control for Autonomous Branches
 
-Unified monitoring orchestrator that provides real-time visibility into:
-- File changes across all branches (file watcher)
-- Log events from all modules (log monitoring)
-- Branch activity and state changes
-- Module execution tracking
-- System health and status
-
-Purpose:
-    Single command interface for monitoring all autonomous branch activity.
-    Replaces fragmented monitoring with unified Mission Control console.
-    Enables multi-agent workflow visibility and coordination.
+Thin orchestration layer for real-time monitoring of file changes, log events,
+and agent activity across all AIPass branches. Delegates to handlers in
+apps/handlers/monitoring/ (unified_stream, branch_detector, event_queue, etc.)
 
 Usage:
     drone @prax monitor              # Show introspection
     drone @prax monitor run          # Monitor all branches
-    drone @prax monitor run seedgo,cli # Monitor specific branches
-
-Interactive Commands:
-    help                      # Show available commands
-    status                    # Display current monitoring state
-    filter [branches]         # Adjust branch filter
-    quit/exit                 # Stop monitoring
-
-Architecture:
-    This module is thin orchestration layer only. All implementation
-    delegated to specialized handlers in apps/handlers/monitoring/:
-
-    - unified_stream.py       → Terminal output formatting
-    - branch_detector.py      → Path-to-branch mapping
-    - interactive_filter.py   → Runtime filter adjustment
-    - monitoring_filters.py   → Event filtering logic
-    - event_queue.py          → Event buffering and deduplication
-    - module_tracker.py       → Module execution tracking
-    - filesystem_handler.py   → Real-time file change detection (FileSystemEventHandler)
-    - log_watcher.py          → Log stream processing
 """
 
 import signal
@@ -227,13 +199,17 @@ def _run_monitor(args: List[str]) -> bool:
     # Start monitoring threads
     _start_threads()
 
-    # Enter interactive mode
-    _interactive_loop()
+    try:
+        _interactive_loop()
+    except KeyboardInterrupt:
+        logger.info("[monitor] KeyboardInterrupt escaped interactive loop")
+        console.print("\n[yellow]Monitoring stopped.[/yellow]")
 
-    # Cleanup on exit
     _stop_threads()
 
-    return True
+    # sys.exit(0) prevents drone's post-execution json_handler from running
+    # after the monitor exits, avoiding a json.load crash on Ctrl+C.
+    sys.exit(0)
 
 
 def _start_threads():