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

feat(system): S90: drone JSON fix + cli AIPASS_HOME auto-detect + mailbox + prax monitor refactor + test isolation

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-11 18:00:43 -07:00
co-authored by @devpulse
parent 35928fc911
commit e7950e76f9
8 changed files with 313 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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/apps/handlers/init/bootstrap.py
+95 -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,40 @@ 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.
Args:
aipass_home: Optional AIPass installation root to add as env.AIPASS_HOME.
"""
data: dict = {
"hooks": {
"UserPromptSubmit": [
{
"matcher": "",
"hooks": [
{
"type": "command",
"command": "cat .aipass/aipass_global_prompt.md 2>/dev/null || true",
}
],
}
]
}
}
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 +490,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 +589,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 +604,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 +654,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 +673,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 +728,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/init_project.py
+4 -3
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()
src/aipass/cli/tests/test_bootstrap.py
+149 -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"
@@ -312,7 +314,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 +369,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 +451,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 +553,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/modules/monitor.py
+7 -103
View File
@@ -46,6 +46,7 @@ Architecture:
- log_watcher.py → Log stream processing
"""
import signal
import sys
import argparse
import threading
@@ -70,15 +71,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 +153,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)
# =============================================================================
@@ -560,103 +560,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