From 262bb7bb58128b28924a6468aaddc8fed34c5d83 Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Sat, 11 Apr 2026 18:15:37 -0700 Subject: [PATCH] feat(system): S90: ai_mail bidirectional cross-project email (DPLAN-0121 Phase 3) + reply-path delivery Co-Authored-By: @devpulse --- src/aipass/ai_mail/.seedgo/bypass.json | 27 +++- .../ai_mail/apps/handlers/email/delivery.py | 28 ++++ .../ai_mail/apps/handlers/email/reply.py | 78 +++++++++++ .../apps/handlers/users/branch_detection.py | 118 +++++++++++------ src/aipass/ai_mail/tests/test_delivery.py | 64 +++++++++ .../ai_mail/tests/test_send_identity.py | 123 ++++++++++++++++++ .../handlers/monitoring/branch_detector.py | 13 +- .../handlers/monitoring/filesystem_handler.py | 19 ++- src/aipass/prax/apps/modules/monitor.py | 46 ++----- 9 files changed, 433 insertions(+), 83 deletions(-) diff --git a/src/aipass/ai_mail/.seedgo/bypass.json b/src/aipass/ai_mail/.seedgo/bypass.json index 2975c0a3..c4773590 100644 --- a/src/aipass/ai_mail/.seedgo/bypass.json +++ b/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": { diff --git a/src/aipass/ai_mail/apps/handlers/email/delivery.py b/src/aipass/ai_mail/apps/handlers/email/delivery.py index dbb9cde8..ff3f6614 100644 --- a/src/aipass/ai_mail/apps/handlers/email/delivery.py +++ b/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"]) diff --git a/src/aipass/ai_mail/apps/handlers/email/reply.py b/src/aipass/ai_mail/apps/handlers/email/reply.py index adebebbd..f32b33ca 100644 --- a/src/aipass/ai_mail/apps/handlers/email/reply.py +++ b/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) diff --git a/src/aipass/ai_mail/apps/handlers/users/branch_detection.py b/src/aipass/ai_mail/apps/handlers/users/branch_detection.py index 4d4068c2..225d7037 100644 --- a/src/aipass/ai_mail/apps/handlers/users/branch_detection.py +++ b/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__": diff --git a/src/aipass/ai_mail/tests/test_delivery.py b/src/aipass/ai_mail/tests/test_delivery.py index 9634a912..b5921060 100644 --- a/src/aipass/ai_mail/tests/test_delivery.py +++ b/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) diff --git a/src/aipass/ai_mail/tests/test_send_identity.py b/src/aipass/ai_mail/tests/test_send_identity.py index bcb7465e..63df4168 100755 --- a/src/aipass/ai_mail/tests/test_send_identity.py +++ b/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" diff --git a/src/aipass/prax/apps/handlers/monitoring/branch_detector.py b/src/aipass/prax/apps/handlers/monitoring/branch_detector.py index 16c883bf..cc27f3ec 100644 --- a/src/aipass/prax/apps/handlers/monitoring/branch_detector.py +++ b/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' diff --git a/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py b/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py index 406e51b9..fd71447c 100644 --- a/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py +++ b/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') diff --git a/src/aipass/prax/apps/modules/monitor.py b/src/aipass/prax/apps/modules/monitor.py index 263720e2..1e94485a 100755 --- a/src/aipass/prax/apps/modules/monitor.py +++ b/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():