diff --git a/src/aipass/ai_mail/.seedgo/bypass.json b/src/aipass/ai_mail/.seedgo/bypass.json index 2975c0a3..12f188ad 100644 --- a/src/aipass/ai_mail/.seedgo/bypass.json +++ b/src/aipass/ai_mail/.seedgo/bypass.json @@ -68,7 +68,7 @@ { "file": "apps/handlers/email/delivery.py", "standard": "handlers", - "reason": "Imports json_utils.json_handler (load_json, save_json) — shared handler utility for JSON I/O, same-branch cross-handler import. Also imports registry.read.get_all_branches — consolidated from dual implementation per DPLAN-0036." + "reason": "Imports json_utils.json_handler (load_json, save_json) — shared handler utility for JSON I/O, same-branch cross-handler import. Also imports registry.read.get_all_branches — consolidated from dual implementation per DPLAN-0036. Also lazy-imports email.contacts.register_contact inside _auto_register_contact() and _auto_register_sender() for post-delivery contact registration (DPLAN-0121 Phase 5)." }, { "file": "apps/handlers/email/inbox_cleanup.py", @@ -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", @@ -233,7 +233,62 @@ { "file": "apps/handlers/users/branch_detection.py", "standard": "handlers", - "reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036." + "reason": "Imports paths.find_repo_root — shared utility consolidated from 8 copies per DPLAN-0036. Also lazy-imports email.contacts.get_contact inside _get_contact_info() for contacts-first sender detection (DPLAN-0121 Phase 5)." + }, + { + "file": "apps/handlers/email/contacts.py", + "standard": "handlers", + "reason": "Imports paths.find_repo_root — shared utility for CONTACTS_FILE path resolution. Same pattern as registry/read.py and other handlers per DPLAN-0036." + }, + { + "file": "apps/handlers/email/identity.py", + "standard": "handlers", + "reason": "bootstrap_aipass_identities() lazy-imports registry.read.get_all_branches and paths.find_repo_root inside function body to avoid circular imports. Same cross-handler pattern as reply.py." + }, + { + "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." + }, + { + "file": "tests/test_contacts.py", + "standard": "architecture", + "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule." + }, + { + "file": "tests/test_contacts.py", + "standard": "encapsulation", + "reason": "Unit tests must import handlers directly to exercise internal functions (_load_contacts, get_contact, register_contact, all_contacts). Module entry-point-only rule does not apply to tests." + }, + { + "file": "tests/test_identity.py", + "standard": "architecture", + "reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule." + }, + { + "file": "tests/test_identity.py", + "standard": "encapsulation", + "reason": "Unit tests must import handlers directly to exercise create_identity, read_identity. Module entry-point-only rule does not apply to tests." } ], "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..18c5bafe 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 @@ -32,6 +33,47 @@ _REPO_ROOT = find_repo_root() _INBOX_LOCK = None +def _auto_register_contact(email: str, branch_path: Path, inbox_file: Path) -> None: + """Auto-register a recipient in the contacts address book after successful delivery. + + Non-critical: failures are logged and silently ignored. + + Args: + email: Recipient email address (e.g., '@devpulse'). + branch_path: Resolved path to the branch root directory. + inbox_file: Path to the branch's inbox.json file. + """ + try: + from aipass.ai_mail.apps.handlers.email.contacts import register_contact + name_key = email.lstrip("@").lower() + register_contact(name_key, "AIPass", str(inbox_file)) + except Exception as e: + logger.warning("[delivery] _auto_register_contact(%s) failed: %s", email, e) + + +def _auto_register_sender(branch_name: str, caller_cwd: str) -> None: + """Auto-register a sender in contacts when called from an external project. + + Walks up from caller_cwd to find .ai_mail.local/inbox.json. + Non-critical: failures are logged and silently ignored. + + Args: + branch_name: Sender branch name or email (e.g., 'vera' or '@vera'). + caller_cwd: Working directory of the calling project. + """ + try: + candidate = Path(caller_cwd) + for path in [candidate] + list(candidate.parents)[:5]: + inbox_file = path / ".ai_mail.local" / "inbox.json" + if inbox_file.exists(): + from aipass.ai_mail.apps.handlers.email.contacts import register_contact + name_key = branch_name.lstrip("@").lower() + register_contact(name_key, "", str(inbox_file)) + return + except Exception as e: + logger.warning("[delivery] _auto_register_sender(%s) failed: %s", branch_name, e) + + def _get_inbox_lock(): """Lazy import inbox_lock context manager.""" global _INBOX_LOCK @@ -130,6 +172,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 +314,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"]) @@ -273,6 +342,15 @@ def deliver_email_to_branch( logger.warning("[delivery] failed to acquire inbox lock for %s: %s", to_branch, e) return False, f"Failed to acquire inbox lock: {e}" + # Auto-register recipient in contacts for future fast lookup + _auto_register_contact(to_branch, branch_path, inbox_file) + + # Auto-register sender if external project called with AIPASS_CALLER_BRANCH + caller_branch = os.environ.get("AIPASS_CALLER_BRANCH", "") + caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "") + if caller_branch and caller_cwd: + _auto_register_sender(caller_branch, caller_cwd) + # Send desktop notification for new email _send_desktop_notification(email_data['from'], to_branch, email_data['subject'], email_data.get('message', '')) 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..ff54fd8b 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,62 @@ from aipass.ai_mail.apps.handlers.paths import find_repo_root BRANCH_REGISTRY_PATH = find_repo_root() / "AIPASS_REGISTRY.json" +def _get_contact_info(branch_name: str) -> Optional[Dict]: + """Look up branch info from the contacts address book. + + Fastest path for sender detection — works for external projects that + have previously registered via contacts, bypassing registry/CWD walk. + + Args: + branch_name: Branch name or email (e.g., 'devpulse' or '@devpulse'). + + Returns: + Synthetic branch info dict compatible with registry format, or None. + """ + try: + from aipass.ai_mail.apps.handlers.email.contacts import get_contact + contact = get_contact(branch_name) + if not contact: + return None + inbox_path = Path(contact["inbox"]) + branch_path = inbox_path.parent.parent # .ai_mail.local -> branch root + name_key = branch_name.lstrip("@").lower() + return { + "name": name_key.upper(), + "email": "@" + name_key, + "path": str(branch_path), + "project": contact.get("project", ""), + } + except Exception as e: + logger.warning("[identity] _get_contact_info(%s) failed: %s", branch_name, e) + return None + + +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. @@ -72,6 +128,11 @@ def detect_branch_from_pwd() -> Optional[Dict]: # Primary: use explicit branch name passed by drone (works in Docker + local) caller_branch = os.environ.get("AIPASS_CALLER_BRANCH") if caller_branch: + # Try contacts first (fastest, works for external projects) + contact = _get_contact_info(caller_branch) + if contact: + return contact + # Fall back to registry lookup branch_info = _lookup_branch_by_name(caller_branch) if branch_info: return branch_info @@ -111,23 +172,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 +239,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/apps/modules/email.py b/src/aipass/ai_mail/apps/modules/email.py index fa1c20f1..deec0f65 100644 --- a/src/aipass/ai_mail/apps/modules/email.py +++ b/src/aipass/ai_mail/apps/modules/email.py @@ -119,7 +119,7 @@ def handle_command(command: str, args: List[str]) -> bool: if command in ("--help", "-h"): print_help() return True - valid = ["send", "email", "inbox", "view", "close", "reply", "sent", "contacts", "read"] + valid = ["send", "email", "inbox", "view", "close", "reply", "sent", "contacts", "read", "register"] if command not in valid: return False if args and args[0] in ['--help', '-h', 'help']: @@ -131,6 +131,7 @@ def handle_command(command: str, args: List[str]) -> bool: "inbox": handle_inbox, "view": handle_view, "close": handle_close, "reply": handle_reply, "read": handle_view, "sent": handle_sent, "contacts": handle_contacts, + "register": handle_register, } return dispatch[command](args) @@ -220,10 +221,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False, label = f"\\[dispatch: queued for daemon]" if auto_execute else "" console.print(f"[green]Email sent to {to_branch} {label}[/green]") if auto_execute: - try: - trigger.fire('email_dispatched', to=to_branch, subject=subject) - except Exception as e: - logger.warning("[email] trigger fire for email_dispatched failed: %s", e) + _fire_dispatch_trigger(to_branch, subject) return True else: error(f"Failed to deliver: {error_msg}") @@ -239,6 +237,14 @@ def _send_direct(to_branch, subject, message, auto_execute=False, return False +def _fire_dispatch_trigger(to_branch: str, subject: str) -> None: + """Fire email_dispatched trigger event if auto_execute enabled.""" + try: + trigger.fire('email_dispatched', to=to_branch, subject=subject) + except Exception as e: + logger.warning("[email] trigger fire for email_dispatched failed: %s", e) + + def _send_broadcast(subject, message, user_info, auto_execute, no_memory_save, reply_to, dispatched_to) -> bool: """Broadcast send to all branches - display wrapper.""" branches = get_all_branches() @@ -312,11 +318,26 @@ def handle_view(args: List[str]) -> bool: """View email content and mark as opened.""" json_handler.log_operation("view_email_initiated", {"args": args}) if not args: - error("Usage: drone @ai_mail view ") + error("Usage: drone @ai_mail view | drone @ai_mail view latest") return True try: branch_path = _resolve_branch_path() - success, message, email_data = mark_as_opened(branch_path, args[0]) + message_id = args[0] + + # Handle "latest" shortcut — get the most recent message ID + if message_id.lower() == "latest": + inbox_file = branch_path / ".ai_mail.local" / "inbox.json" + inbox_data = load_inbox(inbox_file) + if not inbox_data or not inbox_data.get("messages"): + error("Inbox is empty") + return True + # Get the most recent message (last in the list) + message_id = inbox_data["messages"][-1].get("id") + if not message_id: + error("Could not find latest message") + return True + + success, message, email_data = mark_as_opened(branch_path, message_id) if not success or email_data is None: error(message) return True @@ -451,6 +472,27 @@ def handle_contacts(args: List[str]) -> bool: return True +def handle_register(args: List[str]) -> bool: + """Register a branch in the contacts address book. + + Usage: drone @ai_mail register @branch /path/to/inbox [project] + """ + json_handler.log_operation("register_contact_initiated", {"args": args}) + if len(args) < 2: + error("Usage: register @branch /path/to/inbox [project]") + return True + branch_name = args[0].lstrip("@") + inbox_path = args[1] + project = args[2] if len(args) > 2 else "" + from aipass.ai_mail.apps.handlers.email.contacts import register_contact + ok = register_contact(branch_name, project, inbox_path) + if ok: + console.print(f"[green]Registered @{branch_name} -> {inbox_path}[/green]") + else: + error(f"Failed to register @{branch_name}") + return True + + def print_introspection(): """Display module introspection info.""" console.print() 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/cli/.seedgo/bypass.json b/src/aipass/cli/.seedgo/bypass.json index 57fb26e1..b3b333d7 100644 --- a/src/aipass/cli/.seedgo/bypass.json +++ b/src/aipass/cli/.seedgo/bypass.json @@ -10,6 +10,21 @@ "standard": "json_structure", "reason": "Pure Python bootstrap \u2014 no branch imports by design" }, + { + "file": "apps/handlers/init/bootstrap.py", + "standard": "silent_catch", + "reason": "Pure Python bootstrap \u2014 cannot import prax (circular: prax depends on cli). _detect_aipass_home() catches detection failures silently by design." + }, + { + "file": "apps/handlers/init/bootstrap.py", + "standard": "error_handling", + "reason": "Same silent catch as silent_catch standard \u2014 already bypassed (pure Python, no prax import allowed)." + }, + { + "file": "apps/handlers/init/bootstrap.py", + "standard": "log_visibility", + "reason": "Pure Python bootstrap \u2014 cannot import prax system_logger (circular: prax depends on cli). Uses stdlib logging as fallback." + }, { "file": "apps/modules/display.py", "standard": "silent_catch", diff --git a/src/aipass/cli/README.md b/src/aipass/cli/README.md index 3095422a..9889ee75 100644 --- a/src/aipass/cli/README.md +++ b/src/aipass/cli/README.md @@ -5,7 +5,7 @@ **Purpose:** Display and output formatting service for AIPass modules. Provides consistent terminal output — headers, success/error/warning messages, section breaks, and operation templates — so every module looks the same without duplicating Rich formatting code. **Module:** `aipass.cli` **Seedgo:** 100% -**Tests:** 151 passing (6 files, 5/5 modules covered) +**Tests:** 161 passing (6 files, 5/5 modules covered) **Last Updated:** 2026-04-11 ## Usage @@ -66,7 +66,7 @@ cli/ ├── cli_json/ # Auto-created JSON output (three-file pattern) ├── dropbox/ # Inbound file drop ├── logs/ # Branch-level logs -├── tests/ # 151 tests across 6 files +├── tests/ # 161 tests across 6 files │ ├── test_bootstrap.py # bootstrap.py handler tests │ ├── test_json_handler.py # json_handler tests │ ├── test_display.py # display module tests diff --git a/src/aipass/cli/apps/handlers/init/bootstrap.py b/src/aipass/cli/apps/handlers/init/bootstrap.py index 2e36a710..d90e64b4 100644 --- a/src/aipass/cli/apps/handlers/init/bootstrap.py +++ b/src/aipass/cli/apps/handlers/init/bootstrap.py @@ -21,6 +21,7 @@ Business logic for `aipass init`. Creates the project scaffold: 9. .claude/settings.json — Claude Code hooks configuration 10. hooks/ — directory for user hooks 11. src/ — directory where agents live + 12. .ai_mail.local/inbox.json — empty project mailbox Projects are NOT citizens — no .trinity/ directory. Identity lives in the registry JSON. Init is re-runnable: existing files are skipped, not errors. @@ -31,12 +32,16 @@ RULES: - No hardcoded paths """ +import importlib.util import json +import logging import re import uuid from datetime import date from pathlib import Path +logger = logging.getLogger(__name__) + def _sanitize_name(raw: str) -> str: """Sanitize a project name for use in filenames. @@ -47,6 +52,23 @@ def _sanitize_name(raw: str) -> str: return re.sub(r"[^A-Z0-9_-]", "_", raw.upper()).strip("_") +def _detect_aipass_home() -> str | None: + """Detect the AIPass installation root from the aipass package location. + + Returns the parent of the src/ directory (the repo root). + Returns None if detection fails. + """ + try: + spec = importlib.util.find_spec("aipass") + if spec and spec.origin: + # aipass/__init__.py lives at src/aipass/__init__.py + # parent = src/aipass/, parent.parent = src/, parent.parent.parent = AIPass root + return str(Path(spec.origin).resolve().parent.parent.parent) + except Exception as exc: + logger.info("AIPASS_HOME detection skipped: %s", exc) + return None + + # --------------------------------------------------------------------------- # Template content generators # --------------------------------------------------------------------------- @@ -387,23 +409,63 @@ def _gitignore() -> str: ) -def _claude_settings() -> str: - """Generate .claude/settings.json — minimal hooks for prompt injection.""" +def _claude_settings(aipass_home: str | None = None) -> str: + """Generate .claude/settings.json — minimal hooks for prompt injection. + + Installs two UserPromptSubmit hooks: + 1. Global prompt — injects .aipass/aipass_global_prompt.md from CWD. + 2. Local prompt — walks up from CWD to find .aipass/aipass_local_prompt.md + (branch-level prompt, e.g. inside src//). + + Args: + aipass_home: Optional AIPass installation root to add as env.AIPASS_HOME. + """ + _local_prompt_cmd = ( + "dir=$(pwd); " + "while [ \"$dir\" != \"/\" ]; do " + "if [ -f \"$dir/.aipass/aipass_local_prompt.md\" ]; then " + "cat \"$dir/.aipass/aipass_local_prompt.md\"; break; " + "fi; " + "dir=$(dirname \"$dir\"); " + "done" + ) + data: dict = { + "hooks": { + "UserPromptSubmit": [ + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": "cat .aipass/aipass_global_prompt.md 2>/dev/null || true", + } + ], + }, + { + "matcher": "", + "hooks": [ + { + "type": "command", + "command": _local_prompt_cmd, + } + ], + }, + ] + } + } + if aipass_home: + data["env"] = {"AIPASS_HOME": aipass_home} + return json.dumps(data, indent=2, ensure_ascii=False) + "\n" + + +def _inbox_json() -> str: + """Generate .ai_mail.local/inbox.json — empty project mailbox structure.""" return json.dumps( { - "hooks": { - "UserPromptSubmit": [ - { - "matcher": "", - "hooks": [ - { - "type": "command", - "command": "cat .aipass/aipass_global_prompt.md 2>/dev/null || true", - } - ], - } - ] - } + "mailbox": "inbox", + "total_messages": 0, + "unread_count": 0, + "messages": [], }, indent=2, ensure_ascii=False, @@ -451,6 +513,7 @@ def init_project(target: Path, project_name: str | None = None) -> dict: registry_id = str(uuid.uuid4()) today = date.today().isoformat() created = [] + aipass_home = _detect_aipass_home() # 1. Registry (skip if exists — init is re-runnable) registry_filename = f"{name}_REGISTRY.json" @@ -549,7 +612,7 @@ def init_project(target: Path, project_name: str | None = None) -> dict: settings_path = claude_dir / "settings.json" if not settings_path.exists(): - settings_path.write_text(_claude_settings(), encoding="utf-8") + settings_path.write_text(_claude_settings(aipass_home), encoding="utf-8") created.append(str(settings_path)) # 10. hooks/ directory @@ -564,12 +627,21 @@ def init_project(target: Path, project_name: str | None = None) -> dict: src_dir.mkdir() created.append(str(src_dir)) + # 12. .ai_mail.local/inbox.json — empty project mailbox + mail_dir = target / ".ai_mail.local" + mail_dir.mkdir(exist_ok=True) + inbox_path = mail_dir / "inbox.json" + if not inbox_path.exists(): + inbox_path.write_text(_inbox_json(), encoding="utf-8") + created.append(str(inbox_path)) + return { "registry_id": registry_id, "registry_file": registry_filename, "project_name": name, "target": str(target), "created_files": created, + "aipass_home": aipass_home, } @@ -605,6 +677,7 @@ def update_project(target: Path) -> dict: updated: list[str] = [] already_current: list[str] = [] skipped: list[str] = [] + aipass_home: str | None = None # Managed directories — create if missing (graceful recovery). aipass_dir = target / ".aipass" @@ -623,13 +696,27 @@ def update_project(target: Path) -> dict: else: already_current.append(str(global_prompt_path)) + # settings.json — smart merge: preserve existing AIPASS_HOME, detect if missing settings_path = claude_dir / "settings.json" - generated = _claude_settings() - if not settings_path.exists() or settings_path.read_text(encoding="utf-8") != generated: - settings_path.write_text(generated, encoding="utf-8") + if not settings_path.exists(): + aipass_home = _detect_aipass_home() + settings_path.write_text(_claude_settings(aipass_home), encoding="utf-8") updated.append(str(settings_path)) else: - already_current.append(str(settings_path)) + existing_content = settings_path.read_text(encoding="utf-8") + try: + existing_env = json.loads(existing_content).get("env", {}) + except json.JSONDecodeError as exc: + logger.info("settings.json parse failed, rebuilding: %s", exc) + existing_env = {} + # Preserve existing AIPASS_HOME; detect and add if missing + aipass_home = existing_env.get("AIPASS_HOME") or _detect_aipass_home() + generated = _claude_settings(aipass_home) + if existing_content != generated: + settings_path.write_text(generated, encoding="utf-8") + updated.append(str(settings_path)) + else: + already_current.append(str(settings_path)) claude_md_path = target / "CLAUDE.md" generated = _with_source(_claude_md(name), claude_md_path) @@ -664,10 +751,21 @@ def update_project(target: Path) -> dict: ): skipped.append(skip_name) + # Mailbox — create if missing, never overwrite existing + mail_dir = target / ".ai_mail.local" + mail_dir.mkdir(exist_ok=True) + inbox_path = mail_dir / "inbox.json" + if not inbox_path.exists(): + inbox_path.write_text(_inbox_json(), encoding="utf-8") + updated.append(str(inbox_path)) + else: + skipped.append(str(inbox_path)) + return { "project_name": name, "target": str(target), "updated_files": updated, "already_current": already_current, "skipped_files": skipped, + "aipass_home": aipass_home, } diff --git a/src/aipass/cli/apps/modules/display.py b/src/aipass/cli/apps/modules/display.py index b46fde8e..eb5c8a84 100755 --- a/src/aipass/cli/apps/modules/display.py +++ b/src/aipass/cli/apps/modules/display.py @@ -35,10 +35,14 @@ from aipass.cli.apps.handlers.json import json_handler # NOTE: Cannot import prax here — circular import (prax depends on cli) # Silent catches in this file are bypassed via .seedgo/bypass.json +# Detect if output is a TTY (interactive terminal) vs piped/redirected +# When piped, disable force_terminal to let Rich auto-detect and strip ANSI codes +_IS_TTY = sys.stdout.isatty() + # Initialize Rich console (lowercase follows service instance pattern) -CONSOLE = Console(force_terminal=True) # Internal constant — force_terminal ensures ANSI colors even when piped +CONSOLE = Console(force_terminal=_IS_TTY) # TTY=colors, piped=plain console = CONSOLE # Primary export (lowercase service instance pattern) -err_console = Console(stderr=True, force_terminal=True) # Stderr console for error/warning output +err_console = Console(stderr=True, force_terminal=sys.stderr.isatty()) # Stderr auto-detect # Trigger loaded lazily to avoid circular import _TRIGGER = None diff --git a/src/aipass/cli/apps/modules/init_project.py b/src/aipass/cli/apps/modules/init_project.py index 1f908b21..11be5785 100644 --- a/src/aipass/cli/apps/modules/init_project.py +++ b/src/aipass/cli/apps/modules/init_project.py @@ -108,7 +108,7 @@ def print_help(): console.print("[bold cyan]WHAT INIT CREATES:[/bold cyan]") console.print() - files_text = """[bold]Project scaffold (11 items):[/bold] + files_text = """[bold]Project scaffold (12 items):[/bold] [green]1.[/green] [yellow]{NAME}_REGISTRY.json[/yellow] Project registry with UUID [green]2.[/green] [yellow].aipass/aipass_global_prompt.md[/yellow] Global prompt (injected every turn) @@ -118,9 +118,10 @@ def print_help(): [green]6.[/green] [yellow]README.md[/yellow] Getting started guide [green]7.[/green] [yellow]STATUS.local.md[/yellow] Project status [green]8.[/green] [yellow].gitignore[/yellow] Standard AIPass ignores - [green]9.[/green] [yellow].claude/settings.json[/yellow] Claude Code hooks + [green]9.[/green] [yellow].claude/settings.json[/yellow] Claude Code hooks + AIPASS_HOME [green]10.[/green] [yellow]hooks/[/yellow] User hooks directory - [green]11.[/green] [yellow]src/[/yellow] Agent directories live here""" + [green]11.[/green] [yellow]src/[/yellow] Agent directories live here + [green]12.[/green] [yellow].ai_mail.local/inbox.json[/yellow] Empty project mailbox""" console.print(Panel(files_text, border_style="green", padding=(1, 2), box=box.ROUNDED)) console.print() @@ -259,6 +260,12 @@ def _handle_init(args: List[str]) -> bool: success(f"Created {len(result['created_files'])} files") + if result.get("aipass_home"): + console.print() + console.print(f"[bold cyan]AIPASS_HOME:[/bold cyan] [yellow]{result['aipass_home']}[/yellow]") + console.print(f"[dim]For terminal usage, add to your shell profile:[/dim]") + console.print(f" [green]export AIPASS_HOME={result['aipass_home']}[/green]") + json_handler.log_operation("aipass_init", { "project_name": result["project_name"], "target": result["target"], @@ -461,7 +468,7 @@ def _print_init_help(): console.print("[bold cyan]WHAT IT CREATES:[/bold cyan]") console.print() - files_text = """[bold]Project scaffold (11 items):[/bold] + files_text = """[bold]Project scaffold (12 items):[/bold] [green]1.[/green] [yellow]{NAME}_REGISTRY.json[/yellow] Project registry with UUID [green]2.[/green] [yellow].aipass/aipass_global_prompt.md[/yellow] Global prompt (injected every turn) @@ -471,9 +478,10 @@ def _print_init_help(): [green]6.[/green] [yellow]README.md[/yellow] Getting started guide [green]7.[/green] [yellow]STATUS.local.md[/yellow] Project status [green]8.[/green] [yellow].gitignore[/yellow] Standard AIPass ignores - [green]9.[/green] [yellow].claude/settings.json[/yellow] Claude Code hooks + [green]9.[/green] [yellow].claude/settings.json[/yellow] Claude Code hooks + AIPASS_HOME [green]10.[/green] [yellow]hooks/[/yellow] User hooks directory - [green]11.[/green] [yellow]src/[/yellow] Agent directories live here""" + [green]11.[/green] [yellow]src/[/yellow] Agent directories live here + [green]12.[/green] [yellow].ai_mail.local/inbox.json[/yellow] Empty project mailbox""" console.print(Panel(files_text, border_style="green", padding=(1, 2), box=box.ROUNDED)) console.print() diff --git a/src/aipass/cli/tests/test_bootstrap.py b/src/aipass/cli/tests/test_bootstrap.py index 4a0473ea..effb11ce 100644 --- a/src/aipass/cli/tests/test_bootstrap.py +++ b/src/aipass/cli/tests/test_bootstrap.py @@ -91,6 +91,7 @@ def test_init_project_creates_all_expected_files(tmp_path): target / "STATUS.local.md", target / ".gitignore", target / ".claude" / "settings.json", + target / ".ai_mail.local" / "inbox.json", ] for f in expected_files: assert f.exists(), f"Expected file not created: {f}" @@ -105,8 +106,8 @@ def test_init_project_creates_all_expected_files(tmp_path): # No local prompt at project level (belongs in agent dirs only) assert not (target / ".aipass" / "aipass_local_prompt.md").exists() - # 9 files + 2 directories = 11 created_files entries - assert len(result["created_files"]) == 11 + # 10 files + 2 directories = 12 created_files entries + assert len(result["created_files"]) == 12 def test_init_project_return_dict_structure(tmp_path): @@ -122,6 +123,7 @@ def test_init_project_return_dict_structure(tmp_path): "project_name", "target", "created_files", + "aipass_home", } assert result["project_name"] == "ALPHA" assert result["registry_file"] == "ALPHA_REGISTRY.json" @@ -276,6 +278,28 @@ def test_init_project_claude_settings_content(tmp_path): assert "UserPromptSubmit" in data["hooks"] +def test_init_project_settings_has_two_hooks(tmp_path): + """.claude/settings.json has both global and local prompt hooks.""" + target = tmp_path / "proj" + target.mkdir() + + init_project(target, project_name="alpha") + + settings_path = target / ".claude" / "settings.json" + data = json.loads(settings_path.read_text(encoding="utf-8")) + hooks = data["hooks"]["UserPromptSubmit"] + assert len(hooks) == 2, f"Expected 2 UserPromptSubmit hooks, got {len(hooks)}" + + # First hook: global prompt + global_cmd = hooks[0]["hooks"][0]["command"] + assert "aipass_global_prompt.md" in global_cmd + + # Second hook: local prompt walk-up + local_cmd = hooks[1]["hooks"][0]["command"] + assert "aipass_local_prompt.md" in local_cmd + assert "dirname" in local_cmd, "Local prompt hook should walk up with dirname" + + def test_init_project_global_prompt_content(tmp_path): """Global prompt contains project name and AIPass terminology.""" target = tmp_path / "proj" @@ -312,7 +336,7 @@ def test_init_project_auto_creates_target_dir(tmp_path): assert target.is_dir() assert result["project_name"] == "NESTED" - assert len(result["created_files"]) == 11 + assert len(result["created_files"]) == 12 def test_init_project_defaults_name_from_directory(tmp_path): @@ -367,6 +391,10 @@ def test_init_project_skips_existing_optional_files(tmp_path): src_dir = target / "src" src_dir.mkdir() + mail_dir = target / ".ai_mail.local" + mail_dir.mkdir() + (mail_dir / "inbox.json").write_text("{}\n", encoding="utf-8") + result = init_project(target, project_name="eta") # Only registry should be in created_files (everything else pre-existed) @@ -445,6 +473,7 @@ def test_update_project_return_dict_structure(tmp_path): "updated_files", "already_current", "skipped_files", + "aipass_home", } assert result["project_name"] == "UPD" assert result["target"] == str(target.resolve()) @@ -546,11 +575,148 @@ def test_update_project_creates_missing_managed_dirs(tmp_path): def test_update_project_skipped_files_count(tmp_path): - """update_project always skips exactly 4 user-owned files.""" + """update_project skips 4 user-owned files + existing mailbox = 5 total.""" target = tmp_path / "proj" target.mkdir() init_project(target, project_name="count") result = update_project(target) - assert len(result["skipped_files"]) == 4 + # 4 user-owned (registry, README, STATUS, .gitignore) + inbox.json = 5 + assert len(result["skipped_files"]) == 5 + + +# --------------------------------------------------------------------------- +# DPLAN-0121: AIPASS_HOME + mailbox tests +# --------------------------------------------------------------------------- + + +def test_init_project_creates_mailbox(tmp_path): + """init_project creates .ai_mail.local/inbox.json.""" + target = tmp_path / "proj" + target.mkdir() + + init_project(target, project_name="mail") + + assert (target / ".ai_mail.local" / "inbox.json").exists() + + +def test_init_project_mailbox_json_contents(tmp_path): + """inbox.json has valid empty mailbox structure.""" + target = tmp_path / "proj" + target.mkdir() + + init_project(target, project_name="mail") + + data = json.loads((target / ".ai_mail.local" / "inbox.json").read_text(encoding="utf-8")) + assert data["mailbox"] == "inbox" + assert data["total_messages"] == 0 + assert data["unread_count"] == 0 + assert data["messages"] == [] + + +def test_init_project_mailbox_not_overwritten_on_rerun(tmp_path): + """Re-running init skips existing inbox.json.""" + target = tmp_path / "proj" + target.mkdir() + init_project(target, project_name="mail") + + inbox = target / ".ai_mail.local" / "inbox.json" + inbox.write_text('{"custom": true}\n', encoding="utf-8") + + init_project(target, project_name="mail") + + assert json.loads(inbox.read_text(encoding="utf-8")) == {"custom": True} + + +def test_init_project_returns_aipass_home(tmp_path): + """init_project return dict includes aipass_home key.""" + target = tmp_path / "proj" + target.mkdir() + + result = init_project(target, project_name="home") + + assert "aipass_home" in result + assert result["aipass_home"] is None or isinstance(result["aipass_home"], str) + + +def test_init_project_settings_has_aipass_home_when_detected(tmp_path): + """When AIPASS_HOME is detected, settings.json includes env.AIPASS_HOME.""" + target = tmp_path / "proj" + target.mkdir() + + result = init_project(target, project_name="env") + + if result["aipass_home"] is None: + pytest.skip("AIPASS_HOME not detectable in this environment") + + settings = json.loads((target / ".claude" / "settings.json").read_text(encoding="utf-8")) + assert "env" in settings + assert settings["env"]["AIPASS_HOME"] == result["aipass_home"] + + +def test_update_project_creates_mailbox_if_missing(tmp_path): + """update_project creates inbox.json if it does not exist.""" + import shutil + + target = tmp_path / "proj" + target.mkdir() + init_project(target, project_name="newmail") + + # Remove the entire mailbox directory to simulate missing mailbox + shutil.rmtree(target / ".ai_mail.local") + + result = update_project(target) + + inbox = target / ".ai_mail.local" / "inbox.json" + assert inbox.exists() + assert str(inbox) in result["updated_files"] + + +def test_update_project_skips_existing_mailbox(tmp_path): + """update_project never overwrites an existing inbox.json.""" + target = tmp_path / "proj" + target.mkdir() + init_project(target, project_name="keepmail") + + inbox = target / ".ai_mail.local" / "inbox.json" + inbox.write_text( + '{"mailbox":"inbox","total_messages":5,"unread_count":2,"messages":["x"]}\n', + encoding="utf-8", + ) + + result = update_project(target) + + assert str(inbox) in result["skipped_files"] + assert json.loads(inbox.read_text(encoding="utf-8"))["total_messages"] == 5 + + +def test_update_project_returns_aipass_home(tmp_path): + """update_project return dict includes aipass_home key.""" + target = tmp_path / "proj" + target.mkdir() + init_project(target, project_name="uhome") + + result = update_project(target) + + assert "aipass_home" in result + assert result["aipass_home"] is None or isinstance(result["aipass_home"], str) + + +def test_update_project_adds_aipass_home_if_missing(tmp_path): + """update_project injects AIPASS_HOME into settings.json if env section is absent.""" + target = tmp_path / "proj" + target.mkdir() + init_project(target, project_name="addenv") + + settings_path = target / ".claude" / "settings.json" + data = json.loads(settings_path.read_text(encoding="utf-8")) + data.pop("env", None) + settings_path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") + + result = update_project(target) + + if result["aipass_home"] is not None: + new_data = json.loads(settings_path.read_text(encoding="utf-8")) + assert new_data.get("env", {}).get("AIPASS_HOME") == result["aipass_home"] + assert str(settings_path) in result["updated_files"] diff --git a/src/aipass/drone/apps/handlers/json/json_handler.py b/src/aipass/drone/apps/handlers/json/json_handler.py index a9943e64..c01599d4 100644 --- a/src/aipass/drone/apps/handlers/json/json_handler.py +++ b/src/aipass/drone/apps/handlers/json/json_handler.py @@ -77,9 +77,9 @@ def _atomic_write_json(path: Path, data: Any) -> None: with os.fdopen(fd, "w", encoding="utf-8") as fh: json.dump(data, fh, indent=2, ensure_ascii=False) os.replace(tmp_path, str(path)) - except Exception as exc: + except BaseException as exc: logger.warning("_atomic_write_json: failed for %s: %s", path, exc) - # Clean up temp file on failure + # Clean up temp file on failure — BaseException covers KeyboardInterrupt try: os.unlink(tmp_path) except OSError as cleanup_exc: @@ -310,34 +310,38 @@ def log_operation( if module_name is None: module_name = _get_caller_module_name() - ensure_module_jsons(module_name) + try: + ensure_module_jsons(module_name) - # Read rotation limit from config - config = load_json(module_name, "config") - max_entries = 100 - if config and "config" in config: - max_entries = config["config"].get("max_log_entries", 100) + # Read rotation limit from config + config = load_json(module_name, "config") + max_entries = 100 + if config and "config" in config: + max_entries = config["config"].get("max_log_entries", 100) - # Load existing log - log = load_json(module_name, "log") - if log is None: - log = [] + # Load existing log + log = load_json(module_name, "log") + if log is None: + log = [] - # Build entry - entry: dict[str, Any] = { - "timestamp": datetime.now().isoformat(), - "operation": operation, - } - if data: - entry["data"] = data + # Build entry + entry: dict[str, Any] = { + "timestamp": datetime.now().isoformat(), + "operation": operation, + } + if data: + entry["data"] = data - log.append(entry) + log.append(entry) - # FIFO rotation — keep only the most recent entries - if len(log) > max_entries: - log = log[-max_entries:] + # FIFO rotation — keep only the most recent entries + if len(log) > max_entries: + log = log[-max_entries:] - return save_json(module_name, "log", log) + return save_json(module_name, "log", log) + except Exception as exc: + logger.warning("log_operation: failed for %s/%s, skipping: %s", module_name, operation, exc) + return False def increment_counter( diff --git a/src/aipass/drone/tests/test_registry_handler.py b/src/aipass/drone/tests/test_registry_handler.py index 32fef4bc..6bc058ee 100644 --- a/src/aipass/drone/tests/test_registry_handler.py +++ b/src/aipass/drone/tests/test_registry_handler.py @@ -265,8 +265,12 @@ class TestLoadRegistry: class TestGetAllBranches: + @pytest.fixture(autouse=True) + def _isolate_home(self, monkeypatch): + """Prevent real AIPASS_HOME from leaking into test results.""" + monkeypatch.delenv("AIPASS_HOME", raising=False) + 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") diff --git a/src/aipass/drone/tests/test_resolver.py b/src/aipass/drone/tests/test_resolver.py index 4936d872..07e54007 100644 --- a/src/aipass/drone/tests/test_resolver.py +++ b/src/aipass/drone/tests/test_resolver.py @@ -272,6 +272,11 @@ class TestGetBranchInfo: # --------------------------------------------------------------------------- class TestListBranches: + @pytest.fixture(autouse=True) + def _isolate_home(self, monkeypatch): + """Prevent real AIPASS_HOME from leaking into test results.""" + monkeypatch.delenv("AIPASS_HOME", raising=False) + def test_default_status_active(self, populated_registry): """Default status='active' should exclude archived branches.""" result = list_branches() @@ -356,6 +361,11 @@ class TestHandleCommand: class TestWithSampleRegistry: """Tests using the sample_registry fixture from conftest.py.""" + @pytest.fixture(autouse=True) + def _isolate_home(self, monkeypatch): + """Prevent real AIPASS_HOME from leaking into test results.""" + monkeypatch.delenv("AIPASS_HOME", raising=False) + def test_resolve_via_sample(self, sample_registry): set_registry_path(sample_registry) try: diff --git a/src/aipass/prax/apps/handlers/monitoring/branch_detector.py b/src/aipass/prax/apps/handlers/monitoring/branch_detector.py index 1e7592ca..c4c221de 100644 --- a/src/aipass/prax/apps/handlers/monitoring/branch_detector.py +++ b/src/aipass/prax/apps/handlers/monitoring/branch_detector.py @@ -127,8 +127,8 @@ class BranchDetector: for item in project_dir.iterdir(): if item.is_file() and item.name.endswith('_REGISTRY.json'): return item.stem.replace('_REGISTRY', '') - except (OSError, PermissionError): - pass + except (OSError, PermissionError) as e: + logger.info(f"[branch_detector] Cannot read project dir {project_dir}: {e}") return project_part.upper() return project_part.upper() @@ -198,15 +198,21 @@ class BranchDetector: # Internal AIPass: path contains -projects-aipass-src-aipass- if '-projects-aipass-src-aipass-' in folder_lower: - # Simple hyphen-to-slash decode works (no hyphens in AIPass branch names) - project_path = '/' + project_folder.replace('-', '/') + # Strip leading dash before decode — avoids double slash and preserves + # normalization that treats - and _ as equivalent (handles ai_mail→ai-mail). + name_part = project_folder[1:] if project_folder.startswith('-') else project_folder + project_path = '/' + name_part.replace('-', '/') for registered_path, branch_name in self.branch_map.items(): reg_norm = registered_path.replace('_', '/') proj_norm = project_path.replace('_', '/') if reg_norm == proj_norm or registered_path == project_path: return f"AIPASS/{branch_name}{sub_suffix}" - # Fallback: last segment after aipass- + # Fallback: scan segments for known branch names (handles multi-word: ai_mail) segs = [s for s in project_folder.split('-') if s] + for n in range(min(3, len(segs)), 0, -1): + candidate = '_'.join(segs[-n:]).upper() + if candidate in self.known_branches: + return f"AIPASS/{candidate}{sub_suffix}" if segs: return f"AIPASS/{segs[-1].upper()}{sub_suffix}" return None @@ -257,6 +263,7 @@ class BranchDetector: try: rel = path.relative_to(projects_base) except ValueError: + logger.info(f"[branch_detector] Path not under ~/Projects/: {path}") return None parts = rel.parts @@ -280,8 +287,8 @@ class BranchDetector: if item.is_file() and item.name.endswith('_REGISTRY.json'): project_name = item.stem.replace('_REGISTRY', '') break - except (OSError, PermissionError): - pass + except (OSError, PermissionError) as e: + logger.info(f"[branch_detector] Cannot scan project dir {project_dir}: {e}") if not project_name: return None # Not an external AIPass project self._external_project_cache[project_dir_name] = project_name @@ -337,6 +344,26 @@ class BranchDetector: if path_str in self.log_map: return self.log_map[path_str] + # Strategy 0: External AIPass project files get priority over bare registry lookups. + # AIPL branches (e.g. POLYGLOT) are registered in AIPASS_REGISTRY.json with + # absolute paths, so Strategy 2 would return bare 'POLYGLOT' before we can + # add the project prefix. Check external paths first to return 'AIPL/POLYGLOT TESTS'. + _repo_root = self._find_repo_root() + _projects_base = Path.home() / 'Projects' + _path_str_lower = path_str.lower() + _projects_str = str(_projects_base).lower() + _repo_str = str(_repo_root).lower() + _is_external = ( + _path_str_lower.startswith(_projects_str + '/') + and not _path_str_lower.startswith(_repo_str + '/') + ) + + if _is_external: + result = self._detect_from_external_project_path(path) + if result: + self.log_map[path_str] = result + return result + # Strategy 1: Exact match if path_str in self.branch_map: result = self.branch_map[path_str] @@ -351,11 +378,12 @@ class BranchDetector: self.log_map[path_str] = result return result - # Strategy 2.5: External AIPass project files (AIPL, Vera-Studio, etc.) - result = self._detect_from_external_project_path(path) - if result: - self.log_map[path_str] = result - return result + # Strategy 2.5: External AIPass project files (fallback for non-~/Projects/ paths) + if not _is_external: + result = self._detect_from_external_project_path(path) + if result: + self.log_map[path_str] = result + return result # Strategy 3: Claude Code project files if '.claude/projects/' in path_str: @@ -430,8 +458,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() @@ -444,10 +477,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' @@ -455,7 +484,7 @@ class BranchDetector: logger.info(f"Error detecting branch from log {log_file}: {e}") return 'UNKNOWN' - def detect_from_module(self, module_name: str) -> str: + def detect_from_module(self, dotted_name: str) -> str: """ Detect branch from Python module name. @@ -464,30 +493,30 @@ class BranchDetector: - seedgo.core.validator -> SEEDGO Args: - module_name: Python module dotted name + dotted_name: Python module dotted name (e.g. 'aipass.prax.apps') Returns: Branch name in uppercase """ try: # Check cache - if module_name in self.module_map: - return self.module_map[module_name] + if dotted_name in self.module_map: + return self.module_map[dotted_name] # Split on dots and check first part - parts = module_name.split('.') + parts = dotted_name.split('.') if parts: first_part = parts[0].upper() if first_part in self.known_branches: - self.module_map[module_name] = first_part + self.module_map[dotted_name] = first_part return first_part - logger.info(f"Could not detect branch from module: {module_name}") + logger.info(f"Could not detect branch from module: {dotted_name}") return 'UNKNOWN' except Exception as e: - logger.error(f"Error detecting branch from module {module_name}: {e}") + logger.error(f"Error detecting branch from module {dotted_name}: {e}") return 'UNKNOWN' def reload_registry(self): diff --git a/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py b/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py index 406e51b9..8a44c110 100644 --- a/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py +++ b/src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py @@ -28,15 +28,6 @@ from aipass.prax.apps.modules.logger import get_direct_logger logger = get_direct_logger() -# Trigger integration (graceful fallback if unavailable) -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 from aipass.prax.apps.handlers.monitoring.branch_detector import detect_branch_from_path @@ -80,10 +71,12 @@ class MonitoringFileHandler(FileSystemEventHandler): # -- public property so the module can swap queues after construction ------ @property def event_queue(self) -> Optional[MonitoringQueue]: + """The monitoring event queue used to publish file events.""" return self._event_queue @event_queue.setter def event_queue(self, queue: Optional[MonitoringQueue]): + """Replace the event queue (e.g. after handler construction).""" self._event_queue = queue # ========================================================================= @@ -91,49 +84,33 @@ class MonitoringFileHandler(FileSystemEventHandler): # ========================================================================= def on_created(self, event): + """Handle file creation events.""" if not event.is_directory: self._handle_event('created', event.src_path) - self._fire_trigger('file_created', path=event.src_path) def on_modified(self, event): + """Handle file modification events.""" if not event.is_directory: self._handle_event('modified', event.src_path) - self._fire_trigger('file_modified', path=event.src_path) def on_deleted(self, event): + """Handle file deletion events.""" if not event.is_directory: self._handle_event('deleted', event.src_path) - self._fire_trigger('file_deleted', path=event.src_path) def on_moved(self, event): + """Handle file move/rename events.""" if not event.is_directory: # dest_path can be bytes or str, normalize to str for comparison dest_path_str = event.dest_path.decode() if isinstance(event.dest_path, bytes) else event.dest_path src_path_str = event.src_path.decode() if isinstance(event.src_path, bytes) else event.src_path if 'Trash' in dest_path_str or '.local/share/Trash' in dest_path_str: - # Moved to Trash = deletion self._handle_event('deleted', src_path_str) - self._fire_trigger('file_deleted', path=src_path_str) elif '.tmp.' in src_path_str or src_path_str.endswith('.tmp'): # Atomic write: tmp file moved to real file = modification self._handle_event('modified', dest_path_str) - self._fire_trigger('file_modified', path=dest_path_str) else: self._handle_event('moved', dest_path_str) - self._fire_trigger('file_moved', src_path=src_path_str, dest_path=dest_path_str) - - # ========================================================================= - # TRIGGER INTEGRATION - # ========================================================================= - - 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] - except Exception as e: - logger.warning(f"[monitor] trigger.fire('{event_name}') failed: {e}") # ========================================================================= # AGENT ACTIVITY PARSING (Claude Code JSONL sessions) @@ -209,6 +186,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 ae600b7e..1e94485a 100755 --- a/src/aipass/prax/apps/modules/monitor.py +++ b/src/aipass/prax/apps/modules/monitor.py @@ -9,43 +9,16 @@ """ 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 import sys import argparse import threading @@ -70,15 +43,9 @@ from aipass.prax.apps.handlers.monitoring import ( ModuleTracker, # module_tracker.py ) from aipass.prax.apps.handlers.monitoring.event_queue import MonitoringEvent -# NOTE: FileSystemEventHandler implementation lives in: -# aipass.prax.apps.handlers.monitoring.filesystem_handler.MonitoringFileHandler -# It handles trigger events for file_created/file_deleted/file_modified/file_moved +from aipass.prax.apps.modules.monitor_info import print_introspection as _print_introspection, print_help -# ============================================================================= -# UTILITY FUNCTIONS -# ============================================================================= - # ============================================================================= # PID CACHE - Maps branch names to active agent PIDs from dispatch lock files # ============================================================================= @@ -158,6 +125,11 @@ _file_watcher_thread: Optional[threading.Thread] = None _log_watcher_thread: Optional[threading.Thread] = None +def print_introspection(): + """Display module introspection - shows connected handlers and architecture.""" + _print_introspection() + + # ============================================================================= # CORE COMMAND HANDLER (Required for auto-discovery) # ============================================================================= @@ -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(): @@ -560,103 +536,7 @@ def _print_status(): console.print() -# ============================================================================= -# INTROSPECTION (Module metadata and handler connections) -# ============================================================================= - -def print_introspection(): - """Display module introspection - shows connected handlers and architecture""" - console.print() - console.print("[bold cyan]PRAX Monitor Module[/bold cyan]") - console.print() - console.print("[yellow]Purpose:[/yellow]") - console.print(" Mission Control for autonomous branch monitoring") - console.print(" Unified console for file changes, logs, and module activity") - console.print() - - console.print("[yellow]Connected Handlers (apps/handlers/monitoring/):[/yellow]") - console.print() - console.print(" [cyan]1. unified_stream.py[/cyan]") - console.print(" [dim]→ print_event() - Terminal output formatting[/dim]") - console.print() - console.print(" [cyan]2. branch_detector.py[/cyan]") - console.print(" [dim]→ detect_branch_from_path() - Path-to-branch mapping[/dim]") - console.print() - console.print(" [cyan]3. interactive_filter.py[/cyan]") - console.print(" [dim]→ FilterState, parse_command() - Runtime filtering[/dim]") - console.print() - console.print(" [cyan]4. monitoring_filters.py[/cyan]") - console.print(" [dim]→ should_monitor(), get_priority() - Event filtering[/dim]") - console.print() - console.print(" [cyan]5. event_queue.py[/cyan]") - console.print(" [dim]→ MonitoringEvent, MonitoringQueue - Event buffering[/dim]") - console.print() - console.print(" [cyan]6. module_tracker.py[/cyan]") - console.print(" [dim]→ ModuleTracker - Module execution tracking[/dim]") - console.print() - console.print(" [cyan]7. file watcher (threaded)[/cyan]") - console.print(" [dim]→ Real-time file change detection using watchdog[/dim]") - console.print(" [green]STATUS: Active - monitors ECOSYSTEM_ROOT recursively[/green]") - console.print() - console.print(" [cyan]8. log monitor (threaded)[/cyan]") - console.print(" [dim]→ Log stream processing from SYSTEM_LOGS_DIR[/dim]") - console.print(" [green]STATUS: Active - watches *.log files for new entries[/green]") - console.print() - - console.print("[dim]Run 'drone @prax monitor --help' for usage[/dim]") - console.print() - - -# ============================================================================= -# HELP OUTPUT (Drone-compliant command documentation) -# ============================================================================= - -def print_help(): - """Drone-compliant help output - command syntax and examples""" - console.print() - console.print("[bold cyan]PRAX Monitor - Unified Branch Monitoring[/bold cyan]") - console.print() - - console.print("[yellow]Commands:[/yellow]") - console.print() - console.print(" [cyan]drone @prax monitor[/cyan]") - console.print(" Show module introspection") - console.print() - console.print(" [cyan]drone @prax monitor run[/cyan]") - console.print(" Start monitoring all branches") - console.print() - console.print(" [cyan]drone @prax monitor run all[/cyan]") - console.print(" Explicit all-branches monitoring") - console.print() - console.print(" [cyan]drone @prax monitor run [branches][/cyan]") - console.print(" Monitor specific branches (comma-separated)") - console.print(" Example: drone @prax monitor run seedgo,cli,flow") - console.print() - console.print(" [cyan]drone @prax monitor --help[/cyan]") - console.print(" Show this help") - console.print() - - console.print("[yellow]Interactive Mode Commands:[/yellow]") - console.print() - console.print(" [cyan]help[/cyan] Show available commands") - console.print(" [cyan]status[/cyan] Display current monitoring state") - console.print(" [cyan]filter [branches][/cyan] Adjust branch filter") - console.print(" [cyan]quit/exit[/cyan] Stop monitoring") - console.print() - - console.print("[yellow]Examples:[/yellow]") - console.print() - console.print(" [dim]# Monitor all branches[/dim]") - console.print(" $ drone @prax monitor run") - console.print() - console.print(" [dim]# Monitor specific branches[/dim]") - console.print(" $ drone @prax monitor run seedgo,cli,flow") - console.print() - - -# ============================================================================= # MAIN BLOCK (Standalone execution support) -# ============================================================================= if __name__ == "__main__": # Show introspection when run without arguments