slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #259 from AIOSAI/system/devpulse-s90-aimail-contacts-based-resolution-dplan-0121-ph

Browse Source 
feat(system): S90: ai_mail contacts-based resolution (DPLAN-0121 Phase 5) — replaces env var branch detection
This commit is contained in:
AIPass
2026-04-11 23:57:14 -07:00
committed by GitHub
parent 79f072d031 46d80093d1
commit 081c827e1f
19 changed files with 1025 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/ai_mail/.seedgo/bypass.json
+58 -3
View File
@@ -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": {
src/aipass/ai_mail/apps/handlers/email/delivery.py
+78
View File
@@ -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', ''))
src/aipass/ai_mail/apps/handlers/email/reply.py
+78
View File
@@ -133,6 +133,12 @@ def send_reply(
break
if not target_branch:
# Fallback: cross-project delivery via reply_path stored at receive time
stored_reply_path = original_email.get("reply_path")
if stored_reply_path:
return _deliver_via_reply_path(
stored_reply_path, reply_email_data, from_branch_path, original_email
)
return False, f"Could not find branch for {reply_destination}", None
# Deliver the reply (pass email address, not path)
@@ -161,6 +167,78 @@ def send_reply(
return True, f"Reply sent to {reply_destination}, original closed", reply_id
def _deliver_via_reply_path(
reply_path: str,
reply_email_data: Dict,
from_branch_path: Path,
original_email: Dict,
) -> Tuple[bool, str, Optional[str]]:
"""Deliver reply directly to an external project's inbox via stored reply_path.
Used when the recipient is not in the AIPass registry (cross-project reply).
Writes directly to the inbox.json at the stored path, then saves to sent/
and closes the original.
Args:
reply_path: Absolute path to the target inbox.json (stored at receive time).
reply_email_data: The reply email dict to deliver.
from_branch_path: Path to the replying branch (for sent folder and close).
original_email: The original email being replied to (for close).
Returns:
Tuple of (success, message, reply_id or None)
"""
inbox_file = Path(reply_path)
if not inbox_file.exists():
return False, f"reply_path inbox not found: {reply_path}", None
try:
with open(inbox_file, "r", encoding="utf-8") as f:
inbox_data = json.load(f)
except Exception as e:
logger.warning("[reply] _deliver_via_reply_path read failed %s: %s", reply_path, e)
return False, f"Failed to read target inbox: {e}", None
reply_id = str(uuid.uuid4())[:8]
reply_email_data["id"] = reply_id
inbox_data.setdefault("messages", []).insert(0, reply_email_data)
inbox_data["total_messages"] = len(inbox_data["messages"])
new_count = sum(1 for m in inbox_data["messages"]
if m.get("status") == "new" or not m.get("read", False))
inbox_data["unread_count"] = new_count
try:
with open(inbox_file, "w", encoding="utf-8") as f:
json.dump(inbox_data, f, indent=2, ensure_ascii=False)
except Exception as e:
logger.warning("[reply] _deliver_via_reply_path write failed %s: %s", reply_path, e)
return False, f"Failed to write to target inbox: {e}", None
logger.info("[reply] Cross-project reply delivered to %s", reply_path)
# Save to sender's sent folder
sent_folder = from_branch_path / ".ai_mail.local" / "sent"
sent_folder.mkdir(parents=True, exist_ok=True)
sent_file = sent_folder / f"{reply_id}.json"
try:
with open(sent_file, "w", encoding="utf-8") as f:
json.dump(reply_email_data, f, indent=2)
except Exception as e:
logger.warning("[reply] failed to save sent copy: %s", e)
# Auto-close the original email
from aipass.ai_mail.apps.handlers.email.inbox_cleanup import mark_as_closed_and_archive
original_id = original_email.get("id")
if original_id:
close_success, close_msg = mark_as_closed_and_archive(from_branch_path, original_id)
if not close_success:
return True, f"Reply sent (warning: original not closed: {close_msg})", reply_id
destination = reply_email_data.get("to", reply_path)
return True, f"Reply sent to {destination} via reply_path, original closed", reply_id
if __name__ == "__main__":
from aipass.cli.apps.modules import console
console.print("\n" + "="*70)
src/aipass/ai_mail/apps/handlers/users/branch_detection.py
+117 -37
View File
@@ -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__":
src/aipass/ai_mail/apps/modules/email.py
+49 -7
View File
@@ -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 <message_id>")
error("Usage: drone @ai_mail view <message_id> | 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()
src/aipass/ai_mail/tests/test_delivery.py
+64
View File
@@ -19,6 +19,7 @@ import aipass.ai_mail.apps.handlers.email.delivery as delivery_mod
from aipass.ai_mail.apps.handlers.email.delivery import (
_migrate_inbox_format,
_is_private_branch_email,
_resolve_reply_path,
deliver_email_to_branch,
)
@@ -429,3 +430,66 @@ def test_deliver_path_input_unresolvable(repo_root, noop_inbox_lock):
assert success is False
assert "Could not resolve path" in error
# ---- _resolve_reply_path() tests ------------------------------
def test_resolve_reply_path_no_env(monkeypatch):
"""Returns empty string when AIPASS_CALLER_CWD is not set."""
monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False)
assert _resolve_reply_path() == ""
def test_resolve_reply_path_inbox_in_cwd(tmp_path, monkeypatch):
"""Returns inbox path when .ai_mail.local/inbox.json exists in caller CWD."""
inbox_dir = tmp_path / ".ai_mail.local"
inbox_dir.mkdir(parents=True)
inbox_file = inbox_dir / "inbox.json"
inbox_file.write_text("{}", encoding="utf-8")
monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
assert _resolve_reply_path() == str(inbox_file)
def test_resolve_reply_path_inbox_in_parent(tmp_path, monkeypatch):
"""Returns inbox path when .ai_mail.local/inbox.json exists in a parent of caller CWD."""
inbox_dir = tmp_path / ".ai_mail.local"
inbox_dir.mkdir(parents=True)
inbox_file = inbox_dir / "inbox.json"
inbox_file.write_text("{}", encoding="utf-8")
nested_cwd = tmp_path / "src" / "feature"
nested_cwd.mkdir(parents=True)
monkeypatch.setenv("AIPASS_CALLER_CWD", str(nested_cwd))
assert _resolve_reply_path() == str(inbox_file)
def test_resolve_reply_path_no_inbox(tmp_path, monkeypatch):
"""Returns empty string when no .ai_mail.local/inbox.json found in tree."""
monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path))
assert _resolve_reply_path() == ""
def test_deliver_stores_reply_path_from_env(tmp_path, repo_root, noop_inbox_lock, monkeypatch):
"""reply_path env detection is stored on delivered message."""
branches = _setup_branch(tmp_path)
caller_project = tmp_path / "external_project"
inbox_dir = caller_project / ".ai_mail.local"
inbox_dir.mkdir(parents=True)
inbox_file = inbox_dir / "inbox.json"
inbox_file.write_text("{}", encoding="utf-8")
monkeypatch.setenv("AIPASS_CALLER_CWD", str(caller_project))
with patch.object(delivery_mod, "get_all_branches", return_value=branches):
success, _ = deliver_email_to_branch("@target", _make_email_data())
assert success is True
inbox_file_target = Path(branches[0]["path"]) / ".ai_mail.local" / "inbox.json"
with open(inbox_file_target, "r", encoding="utf-8") as f:
inbox = json.load(f)
msg = inbox["messages"][0]
assert "reply_path" in msg
assert msg["reply_path"] == str(inbox_file)
src/aipass/ai_mail/tests/test_send_identity.py
+123
View File
@@ -41,6 +41,7 @@ from aipass.ai_mail.apps.handlers.users.branch_detection import (
find_branch_root,
get_branch_info_from_registry,
_lookup_branch_by_name,
_find_caller_registry,
)
from aipass.ai_mail.apps.handlers.email.send import resolve_sender_info
from aipass.ai_mail.apps.handlers.email.send_args import parse_send_args
@@ -634,3 +635,125 @@ class TestAntiRegression:
assert result["reply_to"] == "@flow"
assert result["recipients"] == ["@spawn"]
assert result["mode"] == "direct"
# ─── _find_caller_registry() tests ───────────────────────
class TestFindCallerRegistry:
"""Tests for the caller registry fallback used for external project branches."""
def test_returns_none_when_no_env(self, clean_env):
"""Returns None when AIPASS_CALLER_CWD is not set."""
result = _find_caller_registry()
assert result is None
def test_returns_none_when_no_registry_in_tree(self, clean_env, tmp_path):
"""Returns None when no AIPASS_REGISTRY.json found under AIPASS_CALLER_CWD."""
os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
result = _find_caller_registry()
assert result is None
def test_finds_registry_in_cwd(self, clean_env, tmp_path):
"""Returns registry path when AIPASS_REGISTRY.json exists in caller CWD."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text('{"branches": []}', encoding="utf-8")
os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH",
tmp_path / "other" / "AIPASS_REGISTRY.json"):
result = _find_caller_registry()
assert result == registry
def test_finds_registry_in_parent(self, clean_env, tmp_path):
"""Returns registry path when found in a parent directory of caller CWD."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text('{"branches": []}', encoding="utf-8")
nested = tmp_path / "src" / "vera"
nested.mkdir(parents=True)
os.environ["AIPASS_CALLER_CWD"] = str(nested)
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH",
tmp_path / "other" / "AIPASS_REGISTRY.json"):
result = _find_caller_registry()
assert result == registry
def test_skips_aipass_registry_itself(self, clean_env, tmp_path):
"""Returns None when found registry is the same as BRANCH_REGISTRY_PATH."""
registry = tmp_path / "AIPASS_REGISTRY.json"
registry.write_text('{"branches": []}', encoding="utf-8")
os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry):
result = _find_caller_registry()
assert result is None
# ─── Caller registry fallback tests ──────────────────────
class TestCallerRegistryFallback:
"""Tests for branch lookup fallback to external project registry."""
def test_lookup_by_name_falls_back_to_caller_registry(self, clean_env, tmp_path):
"""_lookup_branch_by_name finds external branch via caller registry."""
caller_registry = tmp_path / "AIPASS_REGISTRY.json"
caller_registry.write_text(json.dumps({"branches": [
{"name": "VERA", "path": str(tmp_path / "vera"), "email": "@vera", "status": "active"}
]}), encoding="utf-8")
empty_aipass = tmp_path / "other" / "AIPASS_REGISTRY.json"
empty_aipass.parent.mkdir(parents=True)
empty_aipass.write_text('{"branches": []}', encoding="utf-8")
os.environ["AIPASS_CALLER_CWD"] = str(tmp_path)
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", empty_aipass):
result = _lookup_branch_by_name("vera")
assert result is not None
assert result["email"] == "@vera"
assert result["name"] == "VERA"
def test_lookup_by_name_prefers_aipass_registry(self, clean_env, tmp_path):
"""_lookup_branch_by_name returns AIPass result first when branch exists in both."""
aipass_registry = tmp_path / "AIPASS_REGISTRY.json"
aipass_registry.write_text(json.dumps({"branches": [
{"name": "SPAWN", "path": str(tmp_path / "spawn"), "email": "@spawn-aipass", "status": "active"}
]}), encoding="utf-8")
caller_dir = tmp_path / "external"
caller_dir.mkdir()
caller_registry = caller_dir / "AIPASS_REGISTRY.json"
caller_registry.write_text(json.dumps({"branches": [
{"name": "SPAWN", "path": str(tmp_path / "other_spawn"), "email": "@spawn-external", "status": "active"}
]}), encoding="utf-8")
os.environ["AIPASS_CALLER_CWD"] = str(caller_dir)
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", aipass_registry):
result = _lookup_branch_by_name("spawn")
assert result is not None
assert result["email"] == "@spawn-aipass"
def test_get_branch_info_falls_back_to_caller_registry(self, clean_env, tmp_path):
"""get_branch_info_from_registry finds external branch path via caller registry."""
vera_dir = tmp_path / "vera_studio" / "src" / "vera"
vera_dir.mkdir(parents=True)
caller_registry = tmp_path / "vera_studio" / "AIPASS_REGISTRY.json"
caller_registry.write_text(json.dumps({"branches": [
{"name": "VERA", "path": str(vera_dir), "email": "@vera", "status": "active"}
]}), encoding="utf-8")
empty_aipass = tmp_path / "AIPASS_REGISTRY.json"
empty_aipass.write_text('{"branches": []}', encoding="utf-8")
os.environ["AIPASS_CALLER_CWD"] = str(tmp_path / "vera_studio" / "src" / "vera")
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", empty_aipass):
result = get_branch_info_from_registry(vera_dir)
assert result is not None
assert result["email"] == "@vera"
src/aipass/cli/.seedgo/bypass.json
+15
View File
@@ -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",
src/aipass/cli/README.md
+2 -2
View File
@@ -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
src/aipass/cli/apps/handlers/init/bootstrap.py
+118 -20
View File
@@ -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/<agent>/).
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,
}
src/aipass/cli/apps/modules/display.py
+6 -2
View File
@@ -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
src/aipass/cli/apps/modules/init_project.py
+14 -6
View File
@@ -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()
src/aipass/cli/tests/test_bootstrap.py
+171 -5
View File
@@ -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"]
src/aipass/drone/apps/handlers/json/json_handler.py
+28 -24
View File
@@ -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(
src/aipass/drone/tests/test_registry_handler.py
+5 -1
View File
@@ -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")
src/aipass/drone/tests/test_resolver.py
+10
View File
@@ -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:
src/aipass/prax/apps/handlers/monitoring/branch_detector.py
+55 -26
View File
@@ -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):
src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py
+16 -29
View File
@@ -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')
src/aipass/prax/apps/modules/monitor.py
+18 -138
View File
@@ -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