Merge pull request #38 from AIOSAI/feat/registry-glob-and-system-fixes

feat(system): registry glob discovery, drone stderr fix, pyright OOM fix
This commit is contained in:
AIPass
2026-03-14 11:52:06 -07:00
committed by GitHub
67 changed files with 1395 additions and 310 deletions
+5
View File
@@ -3,6 +3,8 @@
**This prompt is your guide.** The patterns shown here are exact. Don't guess command syntax — the examples ARE the API.
**If a command or workflow seems obvious but isn't documented here, flag it.** Don't silently guess — ask or investigate with `--help`. Missing instructions are a prompt bug, not a knowledge gap.
**USER NAME:** Patrick
If no user name is set above, ask on first interaction.
@@ -52,6 +54,9 @@ drone systems # List all registered branches
drone @seedgo audit aipass # Run standards audit on all branches
drone @seedgo standards_query aipass_standards # List all standards (then query by name)
drone @prax monitor # Real-time monitoring (interactive)
drone @flow create . "Subject" # Create FPLAN (execution plan)
drone @flow create . "Subject" dplan # Create DPLAN (design/planning doc)
drone @flow list # List active plans
```
## Dispatch — Wake a Branch
+4 -12
View File
@@ -1,22 +1,14 @@
# AIPass — Project Prompt
Project-level instructions loaded by Claude Code. Persists in context for the entire conversation.
Details live in README, --help, .trinity/ memories, and the Global Prompt (.aipass/aipass_global_prompt.md).
Project-level instructions. Persists in context for the entire conversation.
## Startup
On any greeting, silently read these files from CWD and run the commands — no narration, no announcing steps. Just do it and respond with the status.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `STATUS.local.md`, `README.md`
**Run:** `git status`, `drone systems`
## Navigation
- 15 branches under `src/aipass/` (+ commons at `src/commons/`, skills at `src/skills/`)
- `drone @branch --help` for commands. `drone systems` for branch list. README.md for architecture.
**Read:** `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`
**Run:** `git status`
## Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`. If compaction hits before you save, it's gone. Details in your branch prompt.
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`. If compaction hits before you save, it's gone.
+7 -1
View File
@@ -2,5 +2,11 @@
"extraPaths": ["src"],
"pythonVersion": "3.10",
"reportMissingImports": "error",
"reportAttributeAccessIssue": "error"
"reportAttributeAccessIssue": "error",
"exclude": [
"**/.venv",
"**/node_modules",
"**/.archive",
"**/.sorting_unprocessed"
]
}
+3 -3
View File
@@ -33,7 +33,7 @@ _UPDATE_SECTION = None # type: ignore
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services for display
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
# =============================================================================
# CONSTANTS & CONFIG
@@ -228,14 +228,14 @@ def main():
remaining_args = args[1:] if len(args) > 1 else []
if not modules:
console.print("❌ ERROR: No modules found")
error("No modules found")
return 1
# Route command
if route_command(command, remaining_args, modules):
return 0
else:
console.print(f"❌ ERROR: Unknown command: {command}")
error(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
@@ -21,7 +21,7 @@ from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services for formatting
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
from rich.panel import Panel
# Import handlers
@@ -65,7 +65,7 @@ def handle_ping(verbose: bool = False) -> bool:
except Exception as e:
logger.error(f"Ping failed: {e}")
if verbose:
console.print(f"Ping failed: {e}")
error(f"Ping failed: {e}")
return False
@@ -88,7 +88,7 @@ def handle_status() -> bool:
return True
except Exception as e:
logger.error(f"Status check failed: {e}")
console.print(f"Error getting status: {e}")
error(f"Error getting status: {e}")
return False
@@ -113,7 +113,7 @@ def handle_registry() -> bool:
return True
except Exception as e:
logger.error(f"Registry read failed: {e}")
console.print(f"Error reading registry: {e}")
error(f"Error reading registry: {e}")
return False
+3 -3
View File
@@ -18,7 +18,7 @@ from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
from aipass.ai_mail.apps.handlers.dispatch.status import (
load_dispatch_log,
check_pid_status,
@@ -93,7 +93,7 @@ def handle_command(command: str, args: List[str]) -> bool:
elif subcommand == "wake":
return _orchestrate_wake(args[1:])
else:
console.print(f"[red]Unknown dispatch subcommand: {subcommand}[/red]")
error(f"Unknown dispatch subcommand: {subcommand}")
print_help()
return False
@@ -175,7 +175,7 @@ def _orchestrate_wake(args: List[str]) -> bool:
i += 1
if not filtered:
console.print("[red]Missing branch argument[/red]")
error("Missing branch argument")
return False
branch_email = filtered[0]
+12 -9
View File
@@ -28,7 +28,7 @@ _AI_MAIL_DIR = Path(__file__).resolve().parents[2]
_REPO_ROOT = _AI_MAIL_DIR.parents[2]
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, success
from aipass.trigger.apps.modules.core import trigger
# Handlers - business logic providers
@@ -120,7 +120,7 @@ def handle_send(args: List[str]) -> bool:
parsed = parse_send_args(args)
if parsed["mode"] == "error":
console.print(f"[red]{parsed['error']}[/red]")
error(parsed['error'])
console.print(" Multiple: send @branch1 @branch2 \"Subject\" \"Message\"")
return False
@@ -204,7 +204,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
pass
return True
else:
console.print(f"[red]Failed to deliver: {error_msg}[/red]")
error(f"Failed to deliver: {error_msg}")
dispatch_send_error(to_branch, subject, error_msg, deliver_email_to_branch)
return False
except BrokenPipeError:
@@ -212,7 +212,7 @@ def _send_direct(to_branch, subject, message, auto_execute=False,
return True
except Exception as e:
logger.error(f"[email] Send failed: {e}")
console.print(f"[red]Error: {e}[/red]")
error(f"Error: {e}")
dispatch_send_error(to_branch, subject, str(e), deliver_email_to_branch)
return False
@@ -226,10 +226,13 @@ def _send_broadcast(subject, message, user_info, auto_execute, no_memory_save, r
branches, create_email_file, load_email_file, deliver_email_to_branch,
_delivery_callback, log_operation, update_central)
if isinstance(results, str):
console.print("[red]Failed to load email file for broadcast[/red]")
error("Failed to load email file for broadcast")
return False
for name, success, err in results:
console.print(f" {'[green]OK[/green]' if success else '[red]FAIL[/red]'} {name}" + (f" ({err})" if not success else ""))
for name, ok, err in results:
if ok:
console.print(f" [green]OK[/green] {name}")
else:
error(f"FAIL {name} ({err})")
console.print(f"\nBroadcast complete: {success_count}/{total} delivered")
return ok
@@ -241,7 +244,7 @@ def handle_inbox(args: List[str]) -> bool:
first_arg = args[0] if args else None
ok, info = resolve_inbox_target(first_arg, _REPO_ROOT, get_branch_by_email, get_current_user)
if not ok:
console.print(f"[red]{info['error']}[/red]")
error(info['error'])
return False
inbox_file = info["inbox_file"]
@@ -269,7 +272,7 @@ def handle_inbox(args: List[str]) -> bool:
return True
except Exception as e:
logger.error(f"[email] Inbox view failed: {e}")
console.print(f"[red]Error: {e}[/red]")
error(f"Error: {e}")
return False
+4 -12
View File
@@ -25,7 +25,7 @@ from typing import Any, List
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services for formatted output
from aipass.cli.apps.modules import console, header
from aipass.cli.apps.modules import console, header, error
from rich.panel import Panel
from rich.table import Table
from rich.columns import Columns
@@ -102,9 +102,7 @@ def print_introspection():
modules = discover_modules()
if not modules:
console.print("[red]No modules discovered[/red]")
console.print()
console.print("[dim]Run 'python3 api.py --help' for usage information[/dim]")
error("No modules discovered", suggestion="Run 'python3 api.py --help' for usage information")
console.print()
return
@@ -254,9 +252,7 @@ def main():
if not modules:
logger.error("No modules found")
console.print()
console.print("[red]ERROR: No modules found[/red]")
console.print()
error("No modules found")
return 1
# Extract command and remaining args (matching seed pattern)
@@ -274,11 +270,7 @@ def main():
return 0
else:
logger.warning(f"Unknown command: {command}")
console.print()
console.print(f"[red]ERROR: Unknown command: {command}[/red]")
console.print()
console.print("Run [dim]python3 api.py --help[/dim] for available commands")
console.print()
error(f"Unknown command: {command}", suggestion="Run 'python3 api.py --help' for available commands")
return 1
if __name__ == "__main__":
+5 -5
View File
@@ -19,7 +19,7 @@ import argparse
from pathlib import Path
from typing import Any, List
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
from aipass.prax import logger
# Explicit module imports (replaces dynamic discover_modules)
@@ -172,7 +172,7 @@ def main():
modules = get_modules()
if not modules:
console.print("❌ ERROR: No modules found")
error("No modules found")
return 1
# 5. Handle 'all' command: snapshot → versioned → drive-sync
@@ -187,7 +187,7 @@ def main():
force=args.force, test=False, limit=0
)
if not route_command(snapshot_args, modules):
console.print("[red]Snapshot failed - aborting[/red]")
error("Snapshot failed - aborting")
return 1
console.print()
@@ -201,7 +201,7 @@ def main():
force=args.force, test=False, limit=0
)
if not route_command(versioned_args, modules):
console.print("[red]Versioned backup failed - aborting[/red]")
error("Versioned backup failed - aborting")
return 1
console.print()
@@ -215,7 +215,7 @@ def main():
force=args.force, test=False, limit=args.limit
)
if not route_command(sync_args, modules):
console.print("[red]Drive sync failed[/red]")
error("Drive sync failed")
return 1
console.print()
+4 -1
View File
@@ -21,7 +21,7 @@ PATTERN (from Prax):
"""
# Rich console (primary service - like Prax logger)
from aipass.cli.apps.modules.display import console
from aipass.cli.apps.modules.display import console, err_console
# Display functions
from aipass.cli.apps.modules.display import (
@@ -29,6 +29,7 @@ from aipass.cli.apps.modules.display import (
success,
error,
warning,
fatal,
section
)
@@ -41,12 +42,14 @@ from aipass.cli.apps.modules.templates import (
__all__ = [
# Rich console (primary service)
'console',
'err_console',
# Display
'header',
'success',
'error',
'warning',
'fatal',
'section',
# Templates
+26 -4
View File
@@ -35,6 +35,7 @@ from rich.columns import Columns
# Initialize Rich console (lowercase follows service instance pattern)
CONSOLE = Console() # Internal constant
console = CONSOLE # Primary export (lowercase service instance pattern)
err_console = Console(stderr=True) # Stderr console for error/warning output
# Trigger loaded lazily to avoid circular import
_trigger = None
@@ -325,9 +326,9 @@ def error(message: str, suggestion: str | None = None) -> None:
Example:
error('Branch not found', suggestion='Check branch name spelling')
"""
CONSOLE.print(f"❌ [red bold]{message}[/red bold]")
err_console.print(f"❌ [red bold]{message}[/red bold]")
if suggestion:
CONSOLE.print(f" [yellow]→ Try: {suggestion}[/yellow]")
err_console.print(f" [yellow]→ Try: {suggestion}[/yellow]")
def warning(message: str, details: str | None = None) -> None:
@@ -341,9 +342,28 @@ def warning(message: str, details: str | None = None) -> None:
Example:
warning('Branch already exists, skipping')
"""
CONSOLE.print(f"⚠️ [yellow]{message}[/yellow]")
err_console.print(f"⚠️ [yellow]{message}[/yellow]")
if details:
CONSOLE.print(f" [dim]{details}[/dim]")
err_console.print(f" [dim]{details}[/dim]")
def fatal(message: str, suggestion: str | None = None) -> None:
"""
Display error message with Rich styling and exit with code 1
Like error() but terminates the process. Use for unrecoverable failures.
Args:
message: Error message
suggestion: Optional suggestion for fixing
Example:
fatal('Config file missing', suggestion='Run aipass init first')
"""
err_console.print(f"❌ [red bold]{message}[/red bold]")
if suggestion:
err_console.print(f" [yellow]→ Try: {suggestion}[/yellow]")
sys.exit(1)
def section(title: str) -> None:
@@ -369,10 +389,12 @@ def section(title: str) -> None:
__all__ = [
'console', # Primary export (service instance pattern)
'CONSOLE', # Internal constant (kept for backward compatibility)
'err_console', # Stderr console for error/warning output
'header',
'success',
'error',
'warning',
'fatal',
'section',
]
+2 -4
View File
@@ -22,7 +22,7 @@ from typing import List, Any
from aipass.prax.apps.modules.logger import system_logger as logger
# Console
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def _header(text):
console.print(f"\n[bold cyan]{'='*70}[/bold cyan]")
@@ -186,9 +186,7 @@ def main():
return 0
else:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
console.print()
console.print("Run [dim]daemon --help[/dim] for available commands")
error(f"Unknown command: {command}", suggestion="Run 'daemon --help' for available commands")
console.print()
return 1
+2 -2
View File
@@ -20,7 +20,7 @@ from typing import List
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error as cli_error
def _header(text):
console.print(f"\n[bold cyan]{'='*70}[/bold cyan]")
@@ -31,7 +31,7 @@ def _success(text):
console.print(f"[green]OK:[/green] {text}")
def _error(text):
console.print(f"[red]ERROR:[/red] {text}")
cli_error(text)
from aipass.daemon.apps.handlers.actions.actions_registry import (
list_actions,
+3 -3
View File
@@ -21,7 +21,7 @@ from typing import List
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error as cli_error
def _header(text):
console.print(f"\n[bold cyan]{'='*70}[/bold cyan]")
@@ -32,7 +32,7 @@ def _success(text):
console.print(f"[green]OK:[/green] {text}")
def _error(text):
console.print(f"[red]ERROR:[/red] {text}")
cli_error(text)
# Handler imports
from aipass.daemon.apps.handlers.schedule.task_registry import (
@@ -178,7 +178,7 @@ def _handle_create(args: List[str]) -> bool:
try:
parsed = parser.parse_args(args)
except SystemExit:
console.print("[red]Usage: schedule create \"task\" --due <date> --to @branch [--message \"details\"][/red]")
_error('Usage: schedule create "task" --due <date> --to @branch [--message "details"]')
return False
if not parsed.task:
+2 -2
View File
@@ -19,7 +19,7 @@ from typing import Dict, Any, List
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def _header(text):
console.print(f"\n[bold cyan]{'='*70}[/bold cyan]")
@@ -163,7 +163,7 @@ def handle_command(command: str, args: list) -> bool:
except Exception as e:
logger.error(f"[DAEMON] Error generating update digest: {e}", exc_info=True)
console.print(f"[red]Error: {e}[/red]")
error(f"Error: {e}")
return True
@@ -35,6 +35,7 @@ When a task belongs to a specialist, send it there. Don't burn context doing the
drone @ai_mail send @target "Subject" "Body" --dispatch # Task email
drone @ai_mail dispatch wake @target # Wake branch
drone @flow create . "Subject" # Create FPLAN
drone @flow create . "Subject" dplan # Create DPLAN (dplan template)
drone @flow list # Active plans
drone systems # All branches
```
+1
View File
@@ -0,0 +1 @@
# Temporary home for aipass init — will port to CLI branch later.
+183
View File
@@ -0,0 +1,183 @@
"""
aipass init — Bootstrap an AIPass project in any directory.
Temporary home in devpulse. Will port to CLI branch once proven.
Usage:
python -m aipass.devpulse.apps.init_project [target_dir]
What it does:
1. Generates a UUID for the project registry
2. Creates *_REGISTRY.json with metadata.id
3. Creates .trinity/ (passport with registry_id)
4. Creates .aipass/ (aipass_local_prompt.md)
5. Creates AIPASS.md (project prompt)
"""
import json
import re
import sys
import uuid
from datetime import date
from pathlib import Path
def _sanitize_name(raw: str) -> str:
"""Sanitize a project name for use in filenames.
Replaces non-alphanumeric characters (except underscore/hyphen) with
underscores and strips leading/trailing underscores.
"""
return re.sub(r"[^A-Z0-9_-]", "_", raw.upper()).strip("_")
def init_project(target: Path, project_name: str | None = None) -> dict:
"""Initialize an AIPass project in the target directory.
Args:
target: Directory to initialize
project_name: Name for the registry (defaults to directory name)
Returns:
dict with created files and registry_id
"""
target = target.resolve()
if not target.exists():
target.mkdir(parents=True)
raw_name = project_name or target.name
name = _sanitize_name(raw_name)
if not name:
raise ValueError(
f"Cannot derive project name from '{raw_name}'. "
"Pass a project name explicitly."
)
registry_id = str(uuid.uuid4())
today = date.today().isoformat()
created = []
# 1. Registry
registry_filename = f"{name}_REGISTRY.json"
registry_path = target / registry_filename
if registry_path.exists():
raise FileExistsError(f"Registry already exists: {registry_path}")
registry_data = {
"metadata": {
"id": registry_id,
"name": name,
"version": "1.0.0",
"created": today,
"last_updated": today,
"total_branches": 0,
},
"branches": [],
}
registry_path.write_text(
json.dumps(registry_data, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
created.append(str(registry_path))
# 2. .trinity/
trinity_dir = target / ".trinity"
trinity_dir.mkdir(exist_ok=True)
passport = {
"document_metadata": {
"document_type": "project_identity",
"document_name": f"{name}.PASSPORT",
"version": "1.0.0",
"created": today,
"last_updated": today,
},
"identity": {
"project_name": name,
"role": "project_root",
"purpose": "",
},
"citizenship": {
"registered": True,
"registry_id": registry_id,
"registry_name": name,
},
}
passport_path = trinity_dir / "passport.json"
if passport_path.exists():
raise FileExistsError(
f"Passport already exists: {passport_path}. "
"Remove .trinity/passport.json to re-initialize."
)
passport_path.write_text(
json.dumps(passport, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
created.append(str(passport_path))
# local.json + observations.json
for filename in ("local.json", "observations.json"):
p = trinity_dir / filename
if not p.exists():
p.write_text("{}\n", encoding="utf-8")
created.append(str(p))
# 3. .aipass/
aipass_dir = target / ".aipass"
aipass_dir.mkdir(exist_ok=True)
prompt_path = aipass_dir / "aipass_local_prompt.md"
if not prompt_path.exists():
prompt_path.write_text(
f"# {name} — Local Prompt\n\nInjected every turn. Add project-specific context here.\n",
encoding="utf-8",
)
created.append(str(prompt_path))
# 4. AIPASS.md
aipass_md = target / "AIPASS.md"
if not aipass_md.exists():
aipass_md.write_text(
"# AIPass — Project Prompt\n\n"
"Project-level instructions. Persists in context for the entire conversation.\n\n"
"## Startup\n\n"
"On any greeting, silently read these files from CWD and run the commands "
"— no narration, no announcing steps. Just do it and respond with the status.\n\n"
"**Read:** `.trinity/passport.json`, `.trinity/local.json`, "
"`.trinity/observations.json`, `README.md`\n"
"**Run:** `git status`\n\n"
"## Memories\n\n"
"Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`. "
"If compaction hits before you save, it's gone.\n",
encoding="utf-8",
)
created.append(str(aipass_md))
return {
"registry_id": registry_id,
"registry_file": registry_filename,
"project_name": name,
"target": str(target),
"created_files": created,
}
def main():
"""CLI entry point."""
target = Path(sys.argv[1]) if len(sys.argv) > 1 else Path.cwd()
name = sys.argv[2] if len(sys.argv) > 2 else None
try:
result = init_project(target, name)
except (FileExistsError, ValueError, OSError) as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
print(f"Initialized AIPass project: {result['project_name']}")
print(f"Registry: {result['registry_file']} (id: {result['registry_id'][:8]}...)")
print(f"Created {len(result['created_files'])} files:")
for f in result["created_files"]:
print(f" {f}")
if __name__ == "__main__":
main()
+22
View File
@@ -0,0 +1,22 @@
# DPLAN-003 Working Directory
Research, mapping, and planning files for "AIPass as Operating System."
Parent plan: `AIPass/DPLAN-003_aipass_as_operating_system_2026-03-13.md`
## Scope
**In scope:** `src/aipass/` branches only (drone, seedgo, prax, cli, flow, ai_mail, api, trigger, spawn, devpulse, backup, daemon, memory).
**Out of scope:** `src/commons/`, `src/skills/` — these are separate and not part of this refactor.
## Files
| File | Purpose | Status |
|------|---------|--------|
| `registry_discovery_map.md` | Every find_registry() call in src/aipass/ | Done |
| `portability_audit.md` | Full investigation results (session 24) | Done |
| `credential_model.md` | Registry credential design (UUID now, macaroons later) | Stage 1 Done |
| `registry_refactor_plan.md` | Shared find_registry() design, migration steps | Pending |
| `aipass_init_spec.md` | CLI owns init — `drone @cli aipass init` (entry point later) | Pending |
| `aipass_help_spec.md` | CLI owns help — `drone @cli aipass help` (entry point later) | Pending |
@@ -0,0 +1,156 @@
# DPLAN-003: Registry Credential Model
## The Idea
Registries get a unique token. Passports carry that token. Access is identity-based, not filesystem-based. No walk-up needed — your credential proves which registry is yours.
## Why
Current system finds registries by walking up directories. Works for one project, breaks with multiple. If two AIPass projects exist on one machine, a citizen launched from the wrong directory finds the wrong registry. Credentials solve this — your passport carries proof of membership.
## Prior Art (Research)
| System | Pattern | Fit |
|--------|---------|-----|
| **Macaroons** (Google Research) | Token IS the credential. Delegatable with caveats. Offline verification. DeepMind validated for AI agent delegation (2026). | Highest |
| **Vault Namespaces** | Project = namespace. Token scoped to namespace. Mini-registry per project. | High |
| **AWS STS / Token Vending** | Agent presents project ID, gets scoped credential. Credential itself is the boundary. | High |
| **K8s Namespace + ServiceAccount** | Token carries project scope as claim. RBAC composable. | High |
| **SPIFFE/SPIRE** | Process-level attestation without static secrets. | Medium |
| **direnv** | Auto-set env vars on directory entry. Zero-friction UX. | UX pattern |
Full research: agent output from session 25.
## Design: Two Stages
### Stage 1: UUID Match (Manual, Now)
Simple. Prove the concept works before adding crypto.
**Registry gets an ID:**
```json
{
"metadata": {
"id": "a1b2c3d4-...",
"name": "AIPASS",
"version": "1.0.0",
"last_updated": "2026-03-13",
"total_branches": 15
},
"branches": [...]
}
```
**Passports get the matching ID:**
```json
{
"citizenship": {
"registered": true,
"registry_id": "a1b2c3d4-...",
"registry_name": "AIPASS",
"citizen_number": 7
}
}
```
**Lookup flow:**
1. Walk up from CWD, find `*_REGISTRY.json`
2. Read its `metadata.id`
3. Check citizen's `citizenship.registry_id` matches
4. If mismatch → error ("citizen belongs to registry X, found registry Y")
5. If match → proceed
**Spawn changes:**
- `aipass init` (or manual setup) generates the UUID for the registry
- Spawn reads registry UUID and injects into new passports via `{{REGISTRY_ID}}` placeholder
- Existing 15 branches get the UUID added to their passports (one-time migration)
### Stage 2: Macaroon Tokens (Future, When Cross-Project Needed)
Upgrade path when we need delegation and cross-project access.
**Root token:** Created at `aipass init`. HMAC-based. Stored at `~/.secrets/aipass/projects/<uuid>.token`
**Citizen token:** Attenuated copy in passport. Can prove membership but can't mint new citizens.
**Agent token:** Further attenuated. Carries:
- Registry ID (which project)
- Scope (full access — agents do the real work in AIPass)
- Expiry (session-scoped, dies when agent dies)
- Issuer (which citizen spawned this agent)
Agents are NOT read-only. They're the builders — they write code, run tests, modify files. The token proves they belong to a project, it doesn't restrict what they do within it. Scope restrictions would be role-based (e.g., "can't modify other branches' files") not capability-based.
**Verification:** Local HMAC check. No daemon needed for basic validation. Daemon is optional enhancement for audit logging and revocation.
**direnv integration:** Entering a project directory auto-sets `AIPASS_PROJECT_TOKEN` in env. Agents inherit it.
## What Changes (Stage 1)
| Component | Change |
|-----------|--------|
| `AIPASS_REGISTRY.json` | Add `metadata.id` (UUID) |
| Passport template | Add `citizenship.registry_id` placeholder |
| Spawn `build_replacements_dict()` | Read registry UUID, add `{{REGISTRY_ID}}` |
| Spawn `add_to_registry()` | No change (branch entries stay the same) |
| Drone `find_registry()` | Optional: verify passport.registry_id matches found registry |
| All 15 passports | One-time: add `registry_id` field |
## What Does NOT Change
- Registry filename stays `*_REGISTRY.json`
- Walk-up discovery still works (credential is verification layer on top, not replacement)
- Branch structure unchanged
- No daemon needed
- No new dependencies
## Resolved Questions
1. **Registry ID location** → `metadata.id` — it's the project's identity, not the citizen's.
2. **UUID4 vs hash** → UUID4 (random). Simple, guaranteed unique, no inputs needed.
3. **Verification mode** → Hard error on mismatch. Fail loudly — that's the AIPass way.
4. **Where does init live?** → Temporary: `src/aipass/devpulse/apps/init_project.py`. Future: CLI branch.
## Bugs, Quirks, and Findings
Discovered during testing. Reference for future work.
### init_project.py (10 edge cases tested)
- **Spaces in dir name** → registry filename gets spaces (`MY COOL PROJECT_REGISTRY.json`). Fixed: `_sanitize_name()` replaces non-alphanumeric with `_`.
- **Root path `/`** → `Path("/").name` is empty string, creates `_REGISTRY.json`. Fixed: validation rejects empty name.
- **Permission errors** → raw traceback instead of clean message. Fixed: `main()` catches `OSError`.
- **Passport overwrite on re-init** → if registry deleted but `.trinity/` survives, re-init would silently overwrite passport with new UUID. Fixed: passport guarded with `exists()` check.
- **Double init** → correctly blocked by `FileExistsError` on registry file.
- **Deep nested paths** → `mkdir(parents=True)` handles correctly.
- **UUID uniqueness** → 5 runs, 5 unique UUIDs. No collisions.
- **JSON validity** → all generated files parse clean.
### Drone isolation (6 scenarios tested)
- **Sibling projects** → PASS. Two projects in same parent dir, fully isolated.
- **Nested project** → PASS. Inner registry wins over outer. No bleed-through.
- **Deep subdir walk-up** → PASS. Finds nearest ancestor registry correctly.
- **Cross-project contamination** → PASS. Branches are registry-scoped; modules are global.
- **Empty directory (no registry)** → CONCERN. Drone silently falls back to AIPass source registry via `__file__` walk-up. Any dir on this machine without a registry sees production branches. This is by design in `find_registry()` but will be dangerous with multi-project. Credential verification would catch this — citizen's registry_id won't match the fallback registry.
- **`drone @ai_mail` from mock project** → correctly returns "Branch not found in registry" (empty project has no branches).
### Registry/passport gitignore
- `AIPASS_REGISTRY.json` and all `.trinity/passport.json` files are gitignored. UUID migration is local-only. This is correct for now — credentials are machine-specific, not repo state. But means `aipass init` must run on every clone/install. Future: consider whether UUID should be in-repo or machine-local.
### AI mail after migration
- Send/receive works fine after UUID migration. AI mail's 4 internal `find_registry()` copies still hardcode `AIPASS_REGISTRY.json` — functional for now since that filename exists, but won't find `*_REGISTRY.json` in other projects.
### Drone built-in modules vs branches
- `@drone` and `@seedgo` are hardcoded as "modules" in `module_registry.py`, always visible everywhere. Other branches (`@ai_mail`, `@spawn`, etc.) are registry-scoped. This distinction matters: modules are global services, branches are project citizens.
## Decision Log
- **2026-03-14:** Patrick proposed credential-based registry access. Agents do the real work — tokens prove membership, not restrict capability. Manual first, `aipass init` later.
- **2026-03-14:** Research confirmed macaroons as best-fit pattern (Google Research + DeepMind 2026 validation). Stage 1 = UUID match, Stage 2 = macaroon upgrade.
- **2026-03-14:** Scope limited to `src/aipass/` branches only. Commons and skills excluded.
- **2026-03-14:** All 4 open questions resolved. `init_project.py` built and tested — creates registry with UUID, passport with matching registry_id, .trinity/, .aipass/, AIPASS.md. Tested with temp directory — drone isolation confirmed (only built-in modules visible, not AIPass branches).
- **2026-03-14:** UUID migration executed (FPLAN-0030). Registry + 13 passports updated. Registry and passports are gitignored — UUID is machine-local, not repo state.
- **2026-03-14:** Drone verification added (`_verify_registry_credential` in load_registry). Drone dispatched and fixed error propagation — new `RegistryMismatchError` class separates "not found" (fallback OK) from "mismatch" (hard error). Spawn templates updated with `{{REGISTRY_ID}}` placeholder.
- **2026-03-14:** Stage 1 complete. Credential model is end-to-end: init creates credentialed registries, drone verifies on load, spawn injects into new passports, mismatch = hard error. Remaining: drone CLI stderr surfacing (DPLAN-0031), `aipass init` CLI command (future).
- **2026-03-14:** FPLAN-0032 executed — CLI stderr standardization Phase 1+2 done. CLI owns err_console, error()/warning() go to stderr. Drone imports from CLI instead of creating its own Console(stderr=True). Credential mismatch errors now properly surface to terminal via stderr. The stderr issue that blocked error visibility during DPLAN-003 testing is resolved.
- **2026-03-14:** CLI confirmed as owner of credential model + project commands. CLI owns: `aipass init` (create project), `aipass help` (project help), display API (error/warning/fatal/err_console). Drone owns verification (registry_handler). Spawn owns injection (passport templates). Access via `drone @cli aipass init/help` for now; `aipass init` entry point is future icing.
- **2026-03-14:** Seedgo `stderr_routing` standard created (24th standard). Checker bug found — was scanning all files but not displaying module/handler results. Seedgo fixed display + aggregation. Agent-verified across 3 branches. FPLAN-0033 parked for Phase 3 migration (343 error prints, 10 branches).
@@ -0,0 +1,31 @@
# Portability Audit — Session 24 Results
## Summary
| Tool | Registry Discovery | CWD-Aware | Portable | Hardcoded |
|------|-------------------|-----------|----------|-----------|
| Drone | Walk-up + env var | No (uses registry) | Yes | Registry filename |
| Spawn | Walk-up + env var | No (uses registry) | Partial | Template location |
| Prax | Walk-up (no env) | No (sys logs at repo) | Partial | System logs dir |
| AI_Mail | Walk-up (no env) | No (inbox per branch) | Yes | Inbox location |
| Flow | Walk-up (no env) | Yes (plan creation) | Hybrid | Plan registry |
## Key Findings
- All tools use walk-up strategy to find `AIPASS_REGISTRY.json`
- Registry-relative path resolution already works (move registry + dirs = works)
- `AIPASS_REGISTRY` env var supported by drone and spawn
- System logs hardcoded to `{repo_root}/system_logs/`
- Spawn templates hardcoded to `{spawn_package}/templates/`
- Walk-up doesn't stop at project boundaries — finds nearest registry up the tree
## The Core Fix
Change `find_registry()` to:
1. Walk up from CWD looking for `*_REGISTRY.json` (glob, not hardcoded name)
2. Stop at first match — that's the project boundary
3. If none found, return error ("No AIPass project. Run `aipass init`")
## Source
Full investigation transcript: background agent session 24, 42 tool calls across drone/spawn/ai_mail/flow/prax.
@@ -0,0 +1,46 @@
# Registry Discovery Map — FPLAN-0029 Phase 1
## Primary Implementations (5)
| File | Function | Strategy |
|------|----------|----------|
| `src/aipass/spawn/apps/handlers/registry.py:34-79` | `find_registry(start_path)` | Env var → project markers (.git/pyproject.toml) → walk-up → cwd fallback |
| `src/aipass/drone/apps/handlers/registry_handler.py:36-61` | `find_registry()` | Walk-up from __file__ → walk-up from cwd → parents[4] fallback |
| `src/aipass/drone/apps/handlers/registry_handler.py:64-81` | `get_registry_path()` | Global override → env var → find_registry() |
| `src/commons/apps/handlers/database/db.py:343-378` | `_find_branch_registry()` | Env var AIPASS_ROOT → walk-up (10 limit) → ~/.aipass/ fallback |
| `src/commons/apps/handlers/identity/identity_ops.py:33-67` | `_find_branch_registry_path()` | Same as db.py |
## Local Reimplementations (23+)
All do walk-up from `__file__` looking for `AIPASS_REGISTRY.json`:
### ai_mail (4 files)
- `apps/handlers/users/branch_detection.py:26-38`
- `apps/handlers/registry/read.py:31-41`
- `apps/handlers/email/format.py:21-31`
- `apps/handlers/dispatch/daemon.py:36-55`
### memory (3 files)
- `apps/handlers/dashboard_push.py:44-54`
- `apps/handlers/monitor/detector.py:36-83`
- `apps/handlers/monitor/memory_watcher.py:363-385`
### prax (2 files)
- `apps/handlers/registry/reader.py:24-39`
- `apps/handlers/dashboard/agent_status_writer.py:33-44`
### seedgo (3 files)
- `apps/handlers/audit/discovery.py:54-62`
- `apps/handlers/diagnostics/discovery.py:21-29`
- `apps/handlers/readme/readme_ops.py:29-37`
### + 11 more across other branches
## Hardcoded String Count
100+ references to `"AIPASS_REGISTRY.json"` across codebase.
## Phase 1 Plan
1. Create `commons.registry.find_registry()` — one function, glob for `*_REGISTRY.json`
2. All 23+ modules import from commons
3. Drone/spawn wrappers call commons internally
4. Test: temp dir with TEST_REGISTRY.json → drone systems finds it
+23 -24
View File
@@ -17,8 +17,8 @@ import sys
from typing import List
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.drone.apps.modules import BranchNotFoundError, CommandExecutionError
from aipass.cli.apps.modules import console, err_console
from aipass.drone.apps.modules import BranchNotFoundError, CommandExecutionError, RegistryError
from aipass.drone.apps.modules.discovery import get_help
from aipass.drone.apps.modules.resolver import list_branches
from aipass.drone.apps.modules.router import route_command
@@ -159,7 +159,7 @@ def _handle_module(name: str, args: List[str]) -> int:
try:
result = route_module_command(name, command, cmd_args)
except (ImportError, AttributeError) as exc:
console.print(f"drone: module @{name} is registered but not available: {exc}", stderr=True)
err_console.print(f"drone: module @{name} is registered but not available: {exc}")
return 1
if result.get("stdout"):
@@ -183,16 +183,13 @@ def _handle_target(args: List[str]) -> int:
if not rest:
try:
result = route_command(target)
except BranchNotFoundError as exc:
console.print(f"drone: {exc}", stderr=True)
return 1
except CommandExecutionError as exc:
console.print(f"drone: {exc}", stderr=True)
except (BranchNotFoundError, CommandExecutionError, RegistryError) as exc:
err_console.print(f"drone: {exc}")
return 1
if result.stdout:
console.print(result.stdout, end="", highlight=False)
if result.stderr:
console.print(result.stderr, end="", highlight=False, stderr=True)
err_console.print(result.stderr, end="", highlight=False)
return result.exit_code
# --help = show help
@@ -203,11 +200,8 @@ def _handle_target(args: List[str]) -> int:
console.print(result.text, end="", highlight=False)
else:
console.print(f"No help available for {target}.")
except BranchNotFoundError as exc:
console.print(f"drone: {exc}", stderr=True)
return 1
except CommandExecutionError as exc:
console.print(f"drone: {exc}", stderr=True)
except (BranchNotFoundError, CommandExecutionError, RegistryError) as exc:
err_console.print(f"drone: {exc}")
return 1
return 0
@@ -224,17 +218,14 @@ def _handle_target(args: List[str]) -> int:
args=cmd_args if cmd_args else None,
interactive=interactive,
)
except BranchNotFoundError as exc:
console.print(f"drone: {exc}", stderr=True)
return 1
except CommandExecutionError as exc:
console.print(f"drone: {exc}", stderr=True)
except (BranchNotFoundError, CommandExecutionError, RegistryError) as exc:
err_console.print(f"drone: {exc}")
return 1
if result.stdout:
console.print(result.stdout, end="", highlight=False)
if result.stderr:
console.print(result.stderr, end="", highlight=False, stderr=True)
err_console.print(result.stderr, end="", highlight=False)
return result.exit_code
@@ -248,7 +239,11 @@ def main() -> int:
# No args -> introspection
if not args:
show_introspection()
try:
show_introspection()
except RegistryError as exc:
err_console.print(f"drone: {exc}")
return 1
return 0
# --version
@@ -265,15 +260,19 @@ def main() -> int:
# systems — list branches and modules
if command == "systems":
return _handle_systems()
try:
return _handle_systems()
except RegistryError as exc:
err_console.print(f"drone: {exc}")
return 1
# @target — route to branch or module
if command.startswith("@"):
return _handle_target(args)
# Unknown command
console.print(f"drone: unknown command '{command}'", stderr=True)
console.print("Run 'drone --help' for usage.", stderr=True)
err_console.print(f"drone: unknown command '{command}'")
err_console.print("Run 'drone --help' for usage.")
return 1
@@ -43,6 +43,15 @@ class RegistryNotFoundError(RegistryError):
pass
class RegistryMismatchError(RegistryError):
"""Raised when registry credential doesn't match caller's passport.
NOT recoverable — unlike RegistryNotFoundError (no file, use fallback),
a mismatch means the wrong registry was found. Must error, not fall back.
"""
pass
class RegistryCorruptError(RegistryError):
"""Raised when the registry file is corrupted or invalid JSON."""
pass
@@ -9,7 +9,7 @@
"""
Handler for registry file operations.
Handles loading, parsing, and normalizing AIPASS_REGISTRY.json.
Handles loading, parsing, and normalizing *_REGISTRY.json files.
All file I/O and data transformation for the registry lives here.
"""
@@ -21,6 +21,7 @@ from typing import Any, Dict, List, Optional
from aipass.prax import logger
from .exceptions import (
RegistryCorruptError,
RegistryMismatchError,
RegistryNotFoundError,
RegistryPermissionError,
)
@@ -33,8 +34,18 @@ from .exceptions import (
_registry_path: Optional[Path] = None
def _first_registry_in(directory: Path) -> Optional[Path]:
"""Return the first *_REGISTRY.json in *directory*, or None.
When multiple matches exist, the alphabetically-first name wins
so the result is deterministic across platforms.
"""
matches = sorted(directory.glob("*_REGISTRY.json"))
return matches[0] if matches else None
def find_registry() -> Path:
"""Find AIPASS_REGISTRY.json by walking up from this file's location.
"""Find a *_REGISTRY.json by walking up from this file's location.
Search order:
1. Explicitly set path via set_registry_path()
@@ -42,23 +53,33 @@ def find_registry() -> Path:
3. Walk up from drone package location
4. Walk up from cwd
5. Default: package-relative path
"""
# Walk up from this file (works for pip editable installs)
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
candidate = parent / "AIPASS_REGISTRY.json"
if candidate.exists():
return candidate
# Walk up from cwd (works for regular installs)
The first directory that contains any *_REGISTRY.json is treated
as the project boundary. If that directory holds more than one
match, the alphabetically-first file is returned.
"""
# Walk up from cwd FIRST — this is where the user is working
cwd = Path.cwd()
for parent in [cwd] + list(cwd.parents):
candidate = parent / "AIPASS_REGISTRY.json"
if candidate.exists():
return candidate
hit = _first_registry_in(parent)
if hit is not None:
return hit
# Fallback — use package-relative path (no filesystem assumptions)
return Path(__file__).resolve().parents[4] / "AIPASS_REGISTRY.json"
# Walk up from this file (fallback for pip editable installs)
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
hit = _first_registry_in(parent)
if hit is not None:
return hit
# Fallback — use package-relative path; glob there too
fallback_dir = Path(__file__).resolve().parents[4]
hit = _first_registry_in(fallback_dir)
if hit is not None:
return hit
# Ultimate fallback: return a conventional name so the caller
# gets a clear "not found" path in the error message.
return fallback_dir / "AIPASS_REGISTRY.json"
def get_registry_path() -> Path:
@@ -81,6 +102,49 @@ def get_registry_path() -> Path:
return find_registry()
def _verify_registry_credential(registry_path: Path, registry_data: Dict[str, Any]) -> None:
"""Verify that the registry matches the caller's passport credential.
Compares registry metadata.id against the nearest passport's
citizenship.registry_id. Only raises when BOTH sides have an ID
and they don't match — silent pass otherwise.
"""
try:
registry_id = registry_data.get("metadata", {}).get("id")
if not registry_id:
return
# Walk up from CWD looking for .trinity/passport.json
cwd = Path.cwd()
passport_path = None
for parent in [cwd] + list(cwd.parents):
candidate = parent / ".trinity" / "passport.json"
if candidate.is_file():
passport_path = candidate
break
if passport_path is None:
return
with open(passport_path, "r", encoding="utf-8") as f:
passport = json.load(f)
passport_id = passport.get("citizenship", {}).get("registry_id")
if not passport_id:
return
if passport_id != registry_id:
raise RegistryMismatchError(
f"Registry mismatch: citizen belongs to registry "
f"'{passport_id}' but found registry '{registry_id}' "
f"at {registry_path}"
)
except RegistryMismatchError:
raise
except Exception:
pass # Verification should never crash drone
def set_registry_path(path: str | Path) -> None:
"""Set a custom registry path."""
global _registry_path
@@ -113,7 +177,7 @@ def load_registry() -> Dict[str, Any]:
if not registry_path.exists():
raise RegistryNotFoundError(
f"Registry not found at {registry_path}. "
"Create an AIPASS_REGISTRY.json in your project root."
"Create a *_REGISTRY.json file in your project root."
)
try:
@@ -154,6 +218,8 @@ def load_registry() -> Dict[str, Any]:
elif not isinstance(branches_raw, dict):
raise RegistryCorruptError("Registry 'branches' must be a list or dict")
_verify_registry_credential(registry_path, data)
return data
@@ -8,6 +8,7 @@ from aipass.drone.apps.handlers.exceptions import (
InvalidPathError,
RegistryCorruptError,
RegistryError,
RegistryMismatchError,
RegistryNotFoundError,
RegistryPermissionError,
RoutingError,
@@ -23,6 +24,7 @@ __all__ = [
"InvalidPathError",
"RegistryCorruptError",
"RegistryError",
"RegistryMismatchError",
"RegistryNotFoundError",
"RegistryPermissionError",
"RoutingError",
+14
View File
@@ -27,6 +27,20 @@
"subject": "Build STATUS board — per-branch STATUS.local.md with Prax-managed central aggregation",
"date_closed": "2026-03-10",
"location": "flow"
},
{
"plan_id": "FPLAN-0032",
"type": "FPLAN",
"subject": "CLI stderr standardization — Phase 1 add err_console to display.py",
"date_closed": "2026-03-14",
"location": "flow"
},
{
"plan_id": "FPLAN-0031",
"type": "FPLAN",
"subject": "Drone stderr and error propagation — investigation and long-term fix",
"date_closed": "2026-03-14",
"location": "flow"
}
]
}
+3 -3
View File
@@ -32,7 +32,7 @@ signal.signal(signal.SIGPIPE, signal.SIG_DFL)
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services for formatted output
from aipass.cli.apps.modules import console, header, success, error
from aipass.cli.apps.modules import console, header, success, error, warning
# =============================================================================
# MODULE DISCOVERY
@@ -151,7 +151,7 @@ def main():
return 0
else:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
error(f"Unknown command: {command}")
console.print()
console.print("Run [dim]drone @flow --help[/dim] for available commands")
console.print()
@@ -287,7 +287,7 @@ def print_module_help(command: str, modules: List[Any]):
if not target_module:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
error(f"Unknown command: {command}")
console.print()
console.print("Run [dim]drone @flow --help[/dim] for available commands")
console.print()
+4 -4
View File
@@ -34,7 +34,7 @@ from aipass.prax import logger
from aipass.flow.apps.handlers.json import json_handler
# CLI services for display and error handling
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# Internal: Registry handlers
from aipass.flow.apps.handlers.registry.load_registry import load_registry
@@ -90,7 +90,7 @@ def _display_messages(messages: List[Dict[str, Any]]):
console.print(format_plan_error(error_text, plan_num, details=details))
elif msg_type == "warning":
console.print(f"[yellow]{msg['text']}[/yellow]")
warning(msg['text'])
elif msg_type == "dim":
console.print(f"[dim]{msg['text']}[/dim]")
@@ -102,7 +102,7 @@ def _display_messages(messages: List[Dict[str, Any]]):
console.print(f"[green]{msg['text']}[/green]")
elif msg_type == "error_text":
console.print(f"[red]{msg['text']}[/red]")
error(msg['text'])
elif msg_type == "header":
console.print(format_plan_deletion_header(msg["plan_key"], msg["plan_info"]))
@@ -119,7 +119,7 @@ def _display_messages(messages: List[Dict[str, Any]]):
console.print(f" * FPLAN-{plan['plan_num']}: {plan['subject']}")
elif msg_type == "confirm_warning":
console.print(f"\n[bold red]WARNING: This will close all {msg['count']} plans![/bold red]")
error(f"WARNING: This will close all {msg['count']} plans!")
elif msg_type == "closing_all":
console.print(f"\n[bold]Closing all {msg['count']} plan(s)...[/bold]")
+2 -2
View File
@@ -44,7 +44,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.flow.apps.handlers.json import json_handler
# CLI services for display
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, warning
# Registry handlers (cross-domain - OK for modules)
from aipass.flow.apps.handlers.registry.load_registry import load_registry
@@ -195,7 +195,7 @@ def create_plan(
if msg_type == "dim":
console.print(f"[dim]{msg['text']}[/dim]")
elif msg_type == "warning":
console.print(f"[yellow]{msg['text']}[/yellow]")
warning(msg['text'])
elif msg_type == "display":
console.print(msg["text"])
return ok, num, loc, tmpl, error
+5 -5
View File
@@ -20,7 +20,7 @@ from typing import List
# Infrastructure imports (module does the logging)
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header, success, error
from aipass.cli.apps.modules import console, header, success, error, warning
# Handler imports (local handlers in handlers/dplan/)
from aipass.flow.apps.handlers.dplan.create import create_plan
@@ -458,7 +458,7 @@ def _handle_close(args: List[str]) -> bool:
console.print(f"[dim] Memory Bank archival running in background[/dim]")
except Exception as e:
logger.warning(f"[dev_flow] Failed to spawn background processing: {e}")
console.print(f"[yellow] Background archival failed to start - will retry on next close[/yellow]")
warning("Background archival failed to start - will retry on next close")
# Step 3/3: Done
console.print(f"[dim][3/3][/dim] Finalizing...")
@@ -476,7 +476,7 @@ def _handle_close_all() -> bool:
open_plans = get_open_plans()
if not open_plans:
console.print("\n[yellow]No open plans to close[/yellow]\n")
warning("No open plans to close")
return True
console.print(f"\n[bold yellow]Found {len(open_plans)} open plan(s) to close:[/bold yellow]")
@@ -511,7 +511,7 @@ def _handle_close_all() -> bool:
else:
failure_count += 1
logger.warning(f"[dev_flow] Failed to close DPLAN-{p['number']:03d}: {err}")
console.print(f"[red] Failed: {err}[/red]")
error(f" Failed: {err}")
# Push dashboard
try:
@@ -530,7 +530,7 @@ def _handle_close_all() -> bool:
console.print(f"\n[dim]Background processing started for {success_count} plan(s)[/dim]")
except Exception as e:
logger.warning(f"[dev_flow] Failed to spawn background processing: {e}")
console.print(f"\n[yellow]Background processing failed to start[/yellow]")
warning("Background processing failed to start")
console.print("\n" + "=" * 60)
console.print("[bold green]CLOSE ALL COMPLETE[/bold green]")
@@ -25,7 +25,7 @@ from pathlib import Path
FLOW_ROOT = Path(__file__).resolve().parents[2]
# External: CLI console (Rich display) and Prax logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.prax.apps.modules.logger import system_logger as logger
MODULE_NAME = "dplan_post_close_runner"
@@ -58,7 +58,7 @@ def handle_command(command: str, args: list) -> bool:
# Run the post-close processing directly (foreground)
if not _acquire_lock():
console.print("[yellow]Another instance is already running[/yellow]")
warning("Another instance is already running")
return True
try:
@@ -66,7 +66,7 @@ def handle_command(command: str, args: list) -> bool:
console.print(f"[green]Processing complete:[/green] {result.get('processed', 0)} processed, {result.get('errors', 0)} errors")
except Exception as e:
logger.error(f"[{MODULE_NAME}] Background processing failed: {e}")
console.print(f"[red]Processing failed: {e}[/red]")
error(f"Processing failed: {e}")
finally:
_release_lock()
+4 -4
View File
@@ -43,7 +43,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.flow.apps.handlers.json import json_handler
# CLI services for display
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# Registry handlers
from aipass.flow.apps.handlers.registry.load_registry import load_registry
@@ -149,13 +149,13 @@ def list_plans(filter_type: str = "open") -> bool:
# Module handles display
if result.get("empty") and result.get("success"):
console.print("[yellow]No plans found in registry[/yellow]")
warning("No plans found in registry")
return True
if not result.get("success"):
error_msg = result.get("error", "Unknown error")
try:
console.print(f"[red]ERROR: {error_msg}[/red]")
error(f"ERROR: {error_msg}")
except BrokenPipeError:
pass
return False
@@ -204,7 +204,7 @@ def handle_command(command: str, args: List[str]) -> bool:
if filter_arg in ["open", "closed", "all"]:
filter_type = filter_arg
else:
console.print(f"[yellow]Unknown filter '{filter_arg}', defaulting to 'open'[/yellow]")
warning(f"Unknown filter '{filter_arg}', defaulting to 'open'")
console.print("[dim]Valid filters: open, closed, all[/dim]")
# STEP 2: Execute workflow
@@ -28,7 +28,7 @@ _PKG_ROOT = Path(__file__).resolve().parents[3] # file.py → modules/ → apps
FLOW_ROOT = _PKG_ROOT / "flow"
# External: CLI console (Rich display) and Prax logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.prax.apps.modules.logger import system_logger as logger
MODULE_NAME = "post_close_runner"
@@ -63,7 +63,7 @@ def handle_command(command: str, args: list) -> bool:
# Run the post-close processing directly (foreground)
if not _acquire_lock():
console.print("[yellow]Another instance is already running[/yellow]")
warning("Another instance is already running")
return True
try:
@@ -71,7 +71,7 @@ def handle_command(command: str, args: list) -> bool:
console.print("[green]Processing complete[/green]")
except Exception as e:
logger.error(f"[{MODULE_NAME}] Background processing failed: {e}")
console.print(f"[red]Processing failed: {e}[/red]")
error(f"Processing failed: {e}")
finally:
_release_lock()
@@ -53,7 +53,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.flow.apps.handlers.json import json_handler
# CLI services for display
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# Registry handlers
from aipass.flow.apps.handlers.registry.load_registry import load_registry
@@ -115,11 +115,11 @@ def start_monitoring():
# Module handles display
status = result.get("status", "")
if status == "already_running":
console.print("[yellow]Monitor is already running[/yellow]")
warning("Monitor is already running")
elif status == "started":
console.print(f"[green]OK[/green] {result['message']}")
elif status == "error":
console.print(f"[red]{result['message']}[/red]")
error(result['message'])
return result.get("success", False)
@@ -135,7 +135,7 @@ def stop_monitoring():
if status == "stopped":
console.print("[green]OK[/green] Monitor stopped")
elif status == "not_running":
console.print("[yellow]Monitor is not running[/yellow]")
warning("Monitor is not running")
return result.get("success", False)
@@ -252,7 +252,7 @@ def handle_command(command: str, args: List[str]) -> bool:
return True
else:
console.print(f"[red]Unknown subcommand: {subcommand}[/red]")
error(f"Unknown subcommand: {subcommand}")
console.print()
console.print("Available commands:")
console.print(" • scan - One-time scan and heal registry")
+2 -2
View File
@@ -33,7 +33,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.flow.apps.handlers.json import json_handler
# CLI services for display and error handling
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, warning
# Internal: Registry handlers
from aipass.flow.apps.handlers.registry.load_registry import load_registry
@@ -84,7 +84,7 @@ def _display_messages(messages: List[Dict[str, Any]]):
console.print(format_restore_error(error_type, plan_key, details=details))
elif msg_type == "warning":
console.print(f"[yellow]{msg['text']}[/yellow]")
warning(msg['text'])
elif msg_type == "dim":
console.print(f"[dim]{msg['text']}[/dim]")
+9 -11
View File
@@ -30,7 +30,7 @@ from rich import box
from rich.table import Table
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
# =============================================================================
# INFRASTRUCTURE SETUP
@@ -55,7 +55,7 @@ def print_introspection():
# Discover modules
modules = discover_modules()
console.print(f"[yellow]Discovered Modules:[/yellow] {len(modules)}")
console.print(f"[bold cyan]Discovered Modules:[/bold cyan] {len(modules)}")
console.print()
for module in modules:
@@ -123,12 +123,12 @@ def print_help():
console.print("[bold cyan]USAGE:[/bold cyan]")
console.print()
console.print(" [yellow]Via Drone (recommended):[/yellow]")
console.print(" [dim]drone @memory search \"error handling\"[/dim]")
console.print(" [bold]Via Drone (recommended):[/bold]")
console.print(" [dim]drone @memory search \"performance patterns\"[/dim]")
console.print(" [dim]drone @memory status[/dim]")
console.print(" [dim]drone @memory rollover[/dim]")
console.print()
console.print(" [yellow]Direct execution:[/yellow]")
console.print(" [bold]Direct execution:[/bold]")
console.print(" [dim]python3 -m aipass.memory.apps.memory search \"query\"[/dim]")
console.print(" [dim]python3 -m aipass.memory.apps.memory rollover[/dim]")
console.print()
@@ -141,7 +141,7 @@ def print_help():
console.print(" [cyan]--type TYPE[/cyan] Filter by memory type (observations, local)")
console.print(" [cyan]--n N[/cyan] Number of results (default: 5)")
console.print()
console.print(" [yellow]Example:[/yellow]")
console.print(" [bold]Example:[/bold]")
console.print(" [dim]drone @memory search \"registry bugs\" --branch SEED --n 10[/dim]")
console.print()
console.print("-" * 70)
@@ -149,7 +149,7 @@ def print_help():
console.print("[bold cyan]WATCH MODE:[/bold cyan]")
console.print()
console.print(" [yellow]Start memory file watcher:[/yellow]")
console.print(" [bold]Start memory file watcher:[/bold]")
console.print(" [dim]drone @memory watch[/dim]")
console.print(" [dim]Monitors all branches, auto-rolls when limit exceeded[/dim]")
console.print(" [dim]Press Ctrl+C to stop[/dim]")
@@ -275,7 +275,7 @@ def start_watch() -> None:
result = start_memory_watcher()
if not result.get('success'):
console.print(f"[red]x Failed to start watcher: {result.get('error')}[/red]")
error(f"Failed to start watcher: {result.get('error')}")
return
console.print(f"[green]>[/green] Watching {result.get('count', 0)} branch directories")
@@ -332,9 +332,7 @@ def main():
return # Module handled it successfully
else:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
console.print()
console.print("Run [dim]python3 -m aipass.memory.apps.memory --help[/dim] for available commands")
error(f"Unknown command: {command}", suggestion="Run 'python3 -m aipass.memory.apps.memory --help' for available commands")
console.print()
return
+11 -12
View File
@@ -27,7 +27,7 @@ from rich.panel import Panel
from rich import box
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# =============================================================================
# INFRASTRUCTURE SETUP
@@ -139,7 +139,7 @@ def run_rollover() -> bool:
result = _handler_execute_rollover()
if not result.get('success') and result.get('error'):
console.print(f"[red]x[/red] {result['error']}")
error(result['error'])
return False
triggers_count = result.get('triggers_count', 0)
@@ -168,9 +168,8 @@ def run_rollover() -> bool:
if failed:
console.print()
console.print("[red]Failed operations:[/red]")
for fail in failed:
console.print(f" [red]x[/red] {fail['trigger']} - {fail['stage']}: {fail['error']}")
error(f"{fail['trigger']} - {fail['stage']}: {fail['error']}")
return success_count > 0
@@ -201,11 +200,11 @@ def sync_line_counts() -> None:
if result['success']:
console.print(f"[green]>[/green] Updated {result['updated']} files")
if result['failed'] > 0:
console.print(f"[yellow]![/yellow] {result['failed']} files failed:")
for branch, mem_type, error in result.get('failures', []):
console.print(f" [red]x[/red] {branch}.{mem_type}: {error}")
warning(f"{result['failed']} files failed")
for branch, mem_type, err_msg in result.get('failures', []):
error(f"{branch}.{mem_type}: {err_msg}")
else:
console.print("[red]x[/red] Failed to sync line counts")
error("Failed to sync line counts")
console.print()
@@ -235,7 +234,7 @@ def show_status() -> None:
stats_result = detector.get_rollover_stats()
if not stats_result['success']:
console.print(f"[red]x[/red] Failed to get status: {stats_result.get('error', 'Unknown error')}")
error(f"Failed to get status: {stats_result.get('error', 'Unknown error')}")
logger.error(f"[rollover] Failed to get status: {stats_result.get('error')}")
return
@@ -249,7 +248,7 @@ def show_status() -> None:
# Per-branch details
if stats['branches']:
console.print("[yellow]Branch Details:[/yellow]")
console.print("[bold cyan]Branch Details:[/bold cyan]")
console.print()
for branch_name, branch_stats in stats['branches'].items():
@@ -294,7 +293,7 @@ def check_triggers() -> None:
triggers_result = detector.check_all_branches()
if not triggers_result['success']:
console.print(f"[red]x[/red] Failed to check triggers: {triggers_result.get('error', 'Unknown error')}")
error(f"Failed to check triggers: {triggers_result.get('error', 'Unknown error')}")
logger.error(f"[rollover] Failed to check triggers: {triggers_result.get('error')}")
return
@@ -304,7 +303,7 @@ def check_triggers() -> None:
console.print("[green]>[/green] No files need rollover")
return
console.print(f"[yellow]Found {len(triggers)} files ready for rollover:[/yellow]")
console.print(f"[bold cyan]Found {len(triggers)} files ready for rollover:[/bold cyan]")
console.print()
for trigger in triggers:
+8 -17
View File
@@ -26,7 +26,7 @@ from rich.panel import Panel
from rich import box
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# =============================================================================
# INFRASTRUCTURE SETUP
@@ -63,8 +63,7 @@ def handle_command(command: str, args: List[str]) -> bool:
if command == 'search':
if not args:
console.print("[red]Error:[/red] Search query required")
console.print("Usage: search <query> [--branch BRANCH] [--type TYPE] [--n N]")
error("Search query required", suggestion="Usage: search <query> [--branch BRANCH] [--type TYPE] [--n N]")
return True
# Parse arguments
@@ -85,7 +84,7 @@ def handle_command(command: str, args: List[str]) -> bool:
try:
n_results = int(args[i + 1])
except ValueError:
console.print(f"[red]Error:[/red] Invalid number: {args[i + 1]}")
error(f"Invalid number: {args[i + 1]}")
return True
i += 2
else:
@@ -94,7 +93,7 @@ def handle_command(command: str, args: List[str]) -> bool:
query = ' '.join(query_parts)
if not query:
console.print("[red]Error:[/red] Search query required")
error("Search query required")
return True
show_search_results(query, branch=branch, memory_type=memory_type, n_results=n_results)
@@ -126,7 +125,7 @@ def print_help() -> None:
console.print()
console.print("[bold]EXAMPLES:[/bold]")
console.print(" # Search all branches")
console.print(" [dim]drone @memory search \"error handling patterns\"[/dim]")
console.print(" [dim]drone @memory search \"performance patterns\"[/dim]")
console.print()
console.print(" # Search specific branch")
console.print(" [dim]drone @memory search \"registry bugs\" --branch SEED[/dim]")
@@ -191,7 +190,7 @@ def show_search_results(
)
if not result['success']:
console.print(f"[red]x[/red] {result.get('error', 'Unknown error')}")
error(result.get('error', 'Unknown error'))
return False
collections_searched = result.get('collections_searched', 0)
@@ -203,19 +202,11 @@ def show_search_results(
console.print()
if not filtered_results and total_results == 0:
console.print("[yellow]No matching memories found[/yellow]")
console.print()
console.print("[dim]Try:[/dim]")
console.print(" * Different search terms")
console.print(" * Broader query without filters")
console.print(" * Check if memories have been rolled over (drone @memory status)")
warning("No matching memories found", details="Try different search terms, broader query without filters, or check if memories have been rolled over (drone @memory status)")
return True
if not filtered_results:
console.print("[yellow]No relevant memories found[/yellow]")
console.print()
console.print("[dim]The search found some results but none were relevant enough (>40% similarity).[/dim]")
console.print("[dim]Try more specific search terms related to your AIPass work.[/dim]")
warning("No relevant memories found", details="Results found but none relevant enough (>40% similarity). Try more specific search terms.")
return True
for i, item in enumerate(filtered_results, 1):
@@ -17,7 +17,7 @@ import sys
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def print_introspection():
@@ -102,7 +102,7 @@ def handle_command(command: str, args: List[str]) -> bool:
if result:
console.print("[green]✅ Pushed to all branch dashboards[/green]\n")
else:
console.print("[red]❌ Push failed — check logs[/red]\n")
error("Push failed — check logs")
return True
+14 -14
View File
@@ -21,7 +21,7 @@ from pathlib import Path
from typing import Dict, List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
# Import handlers
from aipass.prax.apps.handlers.dashboard.operations import (
@@ -192,7 +192,7 @@ def print_status():
try:
branches = get_branch_paths()
except Exception as e:
console.print(f"[red]Error loading branches: {e}[/red]")
error(f"Error loading branches: {e}")
return
console.print()
@@ -254,13 +254,13 @@ def _handle_refresh(args: List[str]) -> None:
if result["status"] == "success":
console.print(f"[green]Refreshed {result['branches_updated']} branches[/green]")
elif result["status"] == "partial":
console.print(f"[yellow]Refreshed {result['branches_updated']} branches, {result['branches_failed']} failed[/yellow]")
warning(f"Refreshed {result['branches_updated']} branches, {result['branches_failed']} failed")
for err in result.get("errors", []):
console.print(f" [red]{err}[/red]")
error(str(err))
else:
console.print(f"[red]Refresh failed[/red]")
error("Refresh failed")
for err in result.get("errors", []):
console.print(f" [red]{err}[/red]")
error(str(err))
return
# @branch arg: refresh specific branch
@@ -268,14 +268,14 @@ def _handle_refresh(args: List[str]) -> None:
try:
branch_path = _resolve_branch_path(args[0])
except FileNotFoundError as e:
console.print(f"[red]{e}[/red]")
error(str(e))
return
console.print(f"[dim]Refreshing {branch_path.name.upper()} dashboard...[/dim]")
result = refresh_single_dashboard(branch_path)
if result["status"] == "success":
console.print(f"[green]Refreshed {result['branch']}[/green]")
else:
console.print(f"[red]Failed: {result.get('error', 'unknown')}[/red]")
error(f"Failed: {result.get('error', 'unknown')}")
return
# No args: refresh current branch (detect from CWD)
@@ -296,7 +296,7 @@ def _handle_refresh(args: List[str]) -> None:
if result["status"] == "success":
console.print(f"[green]Refreshed {result['branch']}[/green]")
else:
console.print(f"[red]Failed: {result.get('error', 'unknown')}[/red]")
error(f"Failed: {result.get('error', 'unknown')}")
def _handle_push_template(args: List[str]) -> None:
@@ -334,9 +334,9 @@ def _handle_push_template(args: List[str]) -> None:
if result["errors"]:
console.print()
console.print(f"[red]Errors ({len(result['errors'])}):[/red]")
error(f"Errors ({len(result['errors'])})")
for err in result["errors"]:
console.print(f" [red]! {err}[/red]")
error(f"! {err}")
if not result["changes"] and not result["errors"]:
console.print()
@@ -362,13 +362,13 @@ def _handle_diff_template(args: List[str]) -> None:
if idx + 1 < len(args):
branch_name = args[idx + 1]
else:
console.print("[red]--branch requires a branch name[/red]")
error("--branch requires a branch name")
return
result = diff_dashboard_template(branch_name=branch_name)
if "error" in result:
console.print(f"[red]Error: {result['error']}[/red]")
error(result['error'])
return
summary = result.get("summary", {})
@@ -491,7 +491,7 @@ def main():
remaining_args = args[1:]
if not handle_command(command, remaining_args):
console.print(f"[red]Unknown command: {command}[/red]")
error(f"Unknown command: {command}")
print_help()
+2 -2
View File
@@ -16,7 +16,7 @@ import sys
from typing import List
from aipass.prax.apps.modules.logger import initialize_logging_system, system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def print_introspection():
@@ -77,7 +77,7 @@ def handle_command(command: str, args: List[str]) -> bool:
except Exception as e:
logger.error(f"Error in init command: {e}")
console.print(f"[red]❌ ERROR: {e}[/red]")
error(str(e))
return True
@@ -18,7 +18,7 @@ import sys
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def print_introspection():
@@ -71,7 +71,7 @@ def _display_audit(files: list, summary: dict) -> None:
if summary['healthy']:
console.print("[green] Status: HEALTHY — all logs within limits[/green]")
else:
console.print(f"[red] Status: {summary['oversized_count']} oversized, {summary['critical_count']} critical[/red]")
error(f"Status: {summary['oversized_count']} oversized, {summary['critical_count']} critical")
# Show oversized files
oversized = [f for f in files if f["status"] != "ok"]
@@ -389,7 +389,7 @@ def _interactive_loop():
elif cmd == 'status':
_print_status()
else:
console.print(f"[red]Unknown command: {cmd}[/red]")
error(f"Unknown command: {cmd}")
console.print("[dim]Type 'help' for available commands[/dim]")
except KeyboardInterrupt:
@@ -17,7 +17,7 @@ from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import enable_terminal_output, disable_terminal_output, system_logger as logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
def print_introspection():
@@ -101,7 +101,7 @@ def handle_command(command: str, args: List[str]) -> bool:
except Exception as e:
logger.error(f"Error in terminal command: {e}")
console.print(f"[red]❌ ERROR: {e}[/red]")
error(str(e))
return True
+5 -5
View File
@@ -25,7 +25,7 @@ from typing import List, Callable
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services
from aipass.cli.apps.modules import console, header, success, error
from aipass.cli.apps.modules import console, header, success, error, warning
# =============================================================================
# MODULE DISCOVERY
@@ -63,7 +63,7 @@ def discover_command_modules() -> List[Callable]:
command_handlers.append(module.handle_command)
except Exception as e:
console.print(f"Warning: Failed to load module {module_file.name}: {e}")
warning(f"Failed to load module {module_file.name}: {e}")
return command_handlers
@@ -121,7 +121,7 @@ def route_command(command: str, args: List[str], handlers: List[Callable]) -> bo
if handler(command, args):
return True
except Exception as e:
console.print(f"❌ ERROR: Handler failed: {e}")
error(f"Handler failed: {e}")
return False
return False
@@ -177,14 +177,14 @@ Examples:
handlers = discover_command_modules()
if not handlers:
console.print("❌ ERROR: No command modules discovered")
error("No command modules discovered")
return 1
# Route command to appropriate handler
if route_command(parsed_args.command, parsed_args.args, handlers):
return 0
else:
console.print(f"❌ ERROR: Unknown command: {parsed_args.command}")
error(f"Unknown command: {parsed_args.command}")
return 1
if __name__ == "__main__":
@@ -0,0 +1,61 @@
# Stderr Routing Standard
**Status:** Active
**Date:** 2026-03-13
**Origin:** FPLAN-0032 (CLI stderr standardization)
## What This Covers
Error and warning output must route to stderr via CLI display functions, not stay on stdout via raw `console.print()` with colored markup.
---
## The Problem
CLI `display.py` now has `err_console = Console(stderr=True)`. The functions `error()`, `warning()`, and `fatal()` route through it automatically. But branches still using `console.print('[red]Error...[/red]')` send errors to stdout — breaking piping, redirection, and machine-readable output.
## Correct Pattern
```python
from aipass.cli.apps.modules import error, warning, fatal
error('Branch not found', suggestion='Check spelling') # stderr
warning('Template mismatch', details='Expected v2') # stderr
fatal('Config missing') # stderr + sys.exit(1)
```
## Wrong Pattern
```python
console.print('[red]Error: Branch not found[/red]') # stdout!
console.print('[yellow]Warning: mismatch[/yellow]') # stdout!
Console(stderr=True) # Don't create your own
```
## What the Checker Catches
1. `console.print()` with `[red]`/`[bold red]` markup — use `error()` or `fatal()`
2. `console.print()` with `[yellow]`/`[bold yellow]` markup — use `warning()`
3. Custom `Console(stderr=True)` creation — import `err_console` from CLI
## CLI Display Exports
```python
from aipass.cli.apps.modules import (
error, # ❌ message + optional suggestion → stderr
warning, # ⚠ message + optional details → stderr
fatal, # ❌ message + sys.exit(1) → stderr
err_console, # Raw stderr Console (rare, prefer functions above)
)
```
## Scope
- All branches except CLI (which defines these functions)
- Handlers are exempt (CLI separation standard covers them separately)
- `if __name__ == '__main__':` blocks are excluded
## Reference
- `src/aipass/cli/apps/modules/display.py` — error(), warning(), fatal() definitions
- FPLAN-0032 — Phase 1+2: CLI stderr standardization (closed)
- FPLAN-0033 — Phase 3: branch migration (gates on this standard)
@@ -0,0 +1,303 @@
# =================== AIPass ====================
# Name: stderr_routing_check.py
# Description: Stderr Routing Standards Checker
# Version: 1.0.0
# Created: 2026-03-13
# Modified: 2026-03-13
# =============================================
"""
Stderr Routing Standards Checker
Validates that error/warning output uses CLI display functions (error(), warning(),
fatal()) which route to stderr, instead of raw console.print() with red/yellow markup
that stays on stdout.
CORRECT:
from aipass.cli.apps.modules import error, warning, fatal
error('Branch not found', suggestion='Check spelling')
warning('Template mismatch', details='Expected v2')
WRONG:
console.print('[red]Error: Branch not found[/red]')
console.print('[yellow]Warning: mismatch[/yellow]')
"""
import re
from pathlib import Path
from typing import Dict, List
AUDIT_SCOPE = "all_files"
def is_bypassed(file_path: str, standard: str, line: int | None = None, bypass_rules: list | None = None) -> bool:
"""Check if a violation should be bypassed."""
if not bypass_rules:
return False
for rule in bypass_rules:
if rule.get('standard') and rule.get('standard') != standard:
continue
rule_file = rule.get('file', '')
if rule_file and rule_file not in file_path:
continue
rule_lines = rule.get('lines', [])
if rule_lines and line is not None:
if line in rule_lines:
return True
elif not rule_lines:
return True
return False
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check if module routes error/warning output to stderr via CLI display functions.
Returns:
dict: {passed, checks, score, standard}
"""
checks: List[Dict] = []
path = Path(module_path)
if is_bypassed(module_path, 'stderr_routing', bypass_rules=bypass_rules):
return {
'passed': True,
'checks': [{'name': 'Bypassed', 'passed': True, 'message': 'Standard bypassed via .seedgo/bypass.json'}],
'score': 100,
'standard': 'STDERR_ROUTING'
}
if not path.exists():
return {
'passed': False,
'checks': [{'name': 'File exists', 'passed': False, 'message': f'File not found: {module_path}'}],
'score': 0,
'standard': 'STDERR_ROUTING'
}
# CLI branch is exempt — it defines these functions
if '/cli/apps/' in module_path:
return {
'passed': True,
'checks': [{'name': 'Stderr routing', 'passed': True, 'message': 'CLI branch exempt (defines display functions)'}],
'score': 100,
'standard': 'STDERR_ROUTING'
}
try:
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
lines = content.split('\n')
except Exception as e:
return {
'passed': False,
'checks': [{'name': 'File readable', 'passed': False, 'message': f'Error reading file: {e}'}],
'score': 0,
'standard': 'STDERR_ROUTING'
}
filename = path.name
# Check 1: console.print() with error-like red markup
error_prints = _find_error_prints(lines, module_path, bypass_rules)
if error_prints:
checks.append({
'name': 'Error output routing',
'passed': False,
'message': f'{filename}: {len(error_prints)} error print(s) on lines {error_prints[:5]} — use error() or fatal() instead'
})
elif _has_any_output(content):
checks.append({
'name': 'Error output routing',
'passed': True,
'message': 'No error-like console.print() with red markup'
})
# Check 2: console.print() with warning-like yellow markup
warning_prints = _find_warning_prints(lines, module_path, bypass_rules)
if warning_prints:
checks.append({
'name': 'Warning output routing',
'passed': False,
'message': f'{filename}: {len(warning_prints)} warning print(s) on lines {warning_prints[:5]} — use warning() instead'
})
elif _has_any_output(content):
checks.append({
'name': 'Warning output routing',
'passed': True,
'message': 'No warning-like console.print() with yellow markup'
})
# Check 3: Custom Console(stderr=True) — should import err_console
custom_stderr = _find_custom_stderr_console(lines, module_path, bypass_rules)
if custom_stderr:
checks.append({
'name': 'Stderr console creation',
'passed': False,
'message': f'{filename}: Custom Console(stderr=True) on lines {custom_stderr[:3]} — import err_console from aipass.cli.apps.modules'
})
# No checks applied = no output patterns = skip
if not checks:
return {
'passed': True,
'checks': [{'name': 'Stderr routing', 'passed': True, 'message': 'No error/warning output patterns (skipped)'}],
'score': 100,
'standard': 'STDERR_ROUTING'
}
passed_checks = sum(1 for c in checks if c['passed'])
total_checks = len(checks)
score = int((passed_checks / total_checks * 100)) if total_checks > 0 else 0
return {
'passed': score >= 75,
'checks': checks,
'score': score,
'standard': 'STDERR_ROUTING'
}
def _has_any_output(content: str) -> bool:
"""Check if file has any console output at all."""
return 'console.print(' in content or 'err_console.print(' in content
def _in_skip_context(line: str, in_docstring: bool, in_main_block: bool) -> bool:
"""Check if we should skip this line."""
stripped = line.strip()
if in_docstring or in_main_block:
return True
if stripped.startswith('#'):
return True
return False
def _find_error_prints(lines: List[str], module_path: str, bypass_rules: list | None) -> List[int]:
"""Find console.print() calls with red markup containing error-like content."""
violations = []
in_docstring = False
in_main_block = False
main_block_indent = 0
# Patterns: [red], [bold red], [red bold] in console.print()
red_pattern = re.compile(r'console\.print\(.*\[(?:bold\s+)?red(?:\s+bold)?\]', re.IGNORECASE)
# Also catch: console.print(f"❌ ..." or console.print("Error: ...")
error_msg_pattern = re.compile(r'console\.print\([f"\']*[^)]*(?:error|failed|fatal|cannot|unable|invalid|❌)', re.IGNORECASE)
for i, line in enumerate(lines, 1):
stripped = line.strip()
current_indent = len(line) - len(line.lstrip()) if stripped else 0
# Track __main__ blocks
if "if __name__ ==" in stripped and "__main__" in stripped:
in_main_block = True
main_block_indent = current_indent
continue
if in_main_block and stripped and current_indent <= main_block_indent:
in_main_block = False
# Track docstrings
for quote in ('"""', "'''"):
if line.count(quote) % 2 == 1:
in_docstring = not in_docstring
if _in_skip_context(line, in_docstring, in_main_block):
continue
# Skip if in a string context (the console.print itself is in a string)
if 'console.print(' not in stripped:
continue
# Check it's not inside a string literal
before = line.split('console.print(')[0]
if before.count("'") % 2 == 1 or before.count('"') % 2 == 1:
continue
# Skip bypassed lines
if is_bypassed(module_path, 'stderr_routing', line=i, bypass_rules=bypass_rules):
continue
# Match red markup patterns
if red_pattern.search(stripped):
violations.append(i)
elif error_msg_pattern.search(stripped):
# Only flag if it looks like an error message, not just mentioning the word
# Must be the console.print argument, not a variable name
violations.append(i)
return violations
def _find_warning_prints(lines: List[str], module_path: str, bypass_rules: list | None) -> List[int]:
"""Find console.print() calls with yellow markup containing warning-like content."""
violations = []
in_docstring = False
in_main_block = False
main_block_indent = 0
# Patterns: [yellow] in console.print() that look like warnings
yellow_pattern = re.compile(r'console\.print\(.*\[(?:bold\s+)?yellow(?:\s+bold)?\]', re.IGNORECASE)
warning_msg_pattern = re.compile(r'console\.print\([f"\']*[^)]*(?:warning|⚠)', re.IGNORECASE)
for i, line in enumerate(lines, 1):
stripped = line.strip()
current_indent = len(line) - len(line.lstrip()) if stripped else 0
if "if __name__ ==" in stripped and "__main__" in stripped:
in_main_block = True
main_block_indent = current_indent
continue
if in_main_block and stripped and current_indent <= main_block_indent:
in_main_block = False
for quote in ('"""', "'''"):
if line.count(quote) % 2 == 1:
in_docstring = not in_docstring
if _in_skip_context(line, in_docstring, in_main_block):
continue
if 'console.print(' not in stripped:
continue
before = line.split('console.print(')[0]
if before.count("'") % 2 == 1 or before.count('"') % 2 == 1:
continue
if is_bypassed(module_path, 'stderr_routing', line=i, bypass_rules=bypass_rules):
continue
if yellow_pattern.search(stripped):
violations.append(i)
elif warning_msg_pattern.search(stripped):
violations.append(i)
return violations
def _find_custom_stderr_console(lines: List[str], module_path: str, bypass_rules: list | None) -> List[int]:
"""Find custom Console(stderr=True) creation — should use err_console import."""
violations = []
in_docstring = False
stderr_pattern = re.compile(r'Console\s*\(.*stderr\s*=\s*True', re.IGNORECASE)
for i, line in enumerate(lines, 1):
stripped = line.strip()
for quote in ('"""', "'''"):
if line.count(quote) % 2 == 1:
in_docstring = not in_docstring
if in_docstring or stripped.startswith('#'):
continue
if is_bypassed(module_path, 'stderr_routing', line=i, bypass_rules=bypass_rules):
continue
if stderr_pattern.search(stripped):
violations.append(i)
return violations
@@ -0,0 +1,73 @@
# =================== AIPass ====================
# Name: stderr_routing_content.py
# Description: Stderr Routing Standards Content Handler
# Version: 1.0.0
# Created: 2026-03-13
# Modified: 2026-03-13
# =============================================
"""
Stderr Routing Standards Content Handler
Provides formatted stderr routing standards content.
Module orchestrates, handler implements.
"""
def get_stderr_routing_standards() -> str:
"""Return formatted stderr routing standards content with Rich markup."""
lines = [
"[bold red]STDERR ROUTING STANDARD[/bold red]",
"",
"[yellow]POLICY:[/yellow] Error and warning output MUST route to stderr",
"via CLI display functions. Normal output stays on stdout.",
"",
"[bold cyan]WHY:[/bold cyan]",
" CLI display.py has [dim]err_console = Console(stderr=True)[/dim].",
" [dim]error()[/dim], [dim]warning()[/dim], and [dim]fatal()[/dim] route through it.",
" Branches using [dim]console.print('[red]Error...[/red]')[/dim] bypass this",
" and send errors to stdout, breaking piping and redirection.",
"",
"\u2500" * 70,
"",
"[bold cyan]CORRECT PATTERN:[/bold cyan]",
" [dim]from aipass.cli.apps.modules import error, warning, fatal[/dim]",
"",
" [green]\u2713[/green] [dim]error('Branch not found', suggestion='Check spelling')[/dim]",
" [green]\u2713[/green] [dim]warning('Template mismatch', details='Expected v2')[/dim]",
" [green]\u2713[/green] [dim]fatal('Config missing') # stderr + sys.exit(1)[/dim]",
"",
"[bold cyan]WRONG PATTERN:[/bold cyan]",
" [red]\u2717[/red] [dim]console.print('[red]Error: Branch not found[/red]')[/dim]",
" [red]\u2717[/red] [dim]console.print('[yellow]Warning: mismatch[/yellow]')[/dim]",
" [red]\u2717[/red] [dim]Console(stderr=True) # Don't create your own[/dim]",
"",
"\u2500" * 70,
"",
"[bold cyan]WHAT THE CHECKER CATCHES:[/bold cyan]",
" 1. [dim]console.print()[/dim] with [red][red]/[bold red][/red] markup",
" \u2192 Should use [dim]error()[/dim] or [dim]fatal()[/dim]",
" 2. [dim]console.print()[/dim] with [yellow][yellow]/[bold yellow][/yellow] markup",
" \u2192 Should use [dim]warning()[/dim]",
" 3. Custom [dim]Console(stderr=True)[/dim] creation",
" \u2192 Import [dim]err_console[/dim] from CLI instead",
"",
"[bold cyan]CLI DISPLAY EXPORTS:[/bold cyan]",
" [dim]from aipass.cli.apps.modules import ([/dim]",
" [dim] error, # \u2718 message + optional suggestion \u2192 stderr[/dim]",
" [dim] warning, # \u26a0 message + optional details \u2192 stderr[/dim]",
" [dim] fatal, # \u2718 message + sys.exit(1) \u2192 stderr[/dim]",
" [dim] err_console, # Raw stderr Console (rare, prefer functions)[/dim]",
" [dim])[/dim]",
"",
"[bold cyan]SCOPE:[/bold cyan]",
" All branches except CLI (which defines these functions).",
" Handlers are exempt from CLI output standards (separate concern).",
"",
"[bold cyan]REFERENCE:[/bold cyan]",
" [dim]See: src/aipass/cli/apps/modules/display.py (error, warning, fatal)[/dim]",
" [dim]See: FPLAN-0032 (Phase 1+2: CLI stderr standardization)[/dim]",
" [dim]See: FPLAN-0033 (Phase 3: branch migration — gates on this standard)[/dim]",
]
return "\n".join(lines)
@@ -233,6 +233,16 @@ def print_branch_summary(audit_result: Dict, system_averages: Dict[str, int] | N
for issue in violation.get('issues', []):
console.print(f" [dim]• {issue}[/dim]")
# Show STDERR_ROUTING violations (console.print red/yellow → use error()/warning())
stderr_routing_violations = audit_result.get('stderr_routing_violations', [])
if stderr_routing_violations:
console.print()
console.print(f" [bold red]STDERR_ROUTING VIOLATIONS ({len(stderr_routing_violations)} files):[/bold red]")
for violation in stderr_routing_violations:
console.print(f" [red]✗[/red] {violation['path']} [dim](score: {violation['score']}%)[/dim]")
for issue in violation.get('issues', []):
console.print(f" [dim]• {issue}[/dim]")
# Show SHEBANG violations (no #!/... in pip packages)
shebang_violations = audit_result.get('shebang_violations', [])
if shebang_violations:
@@ -143,7 +143,14 @@ def audit_branch(branch: Dict[str, str], bypass_rules: list, pack_path: Path = N
v, s = _run_all_files(checker, name, all_files, bypass_rules)
all_violations[name] = v
if s:
scores[name] = int(sum(s) / len(s))
avg_score = int(sum(s) / len(s))
scores[name] = avg_score
# Update results to reflect all-files findings
all_failed = []
for vi in v:
all_failed.extend({"name": name, "passed": False, "message": iss} for iss in vi.get("issues", []))
if all_failed:
results[name] = {"passed": avg_score >= 75, "checks": all_failed, "score": avg_score, "standard": name.upper()}
# Log structure post-checks (audit-level, not in any checker)
if "log_structure" in scores:
@@ -51,14 +51,20 @@ def _is_branch_private(branch_name: str) -> bool:
def _find_registry() -> Path:
"""
Find AIPASS_REGISTRY.json by walking up from this file's location.
Works regardless of install path depth.
Find *_REGISTRY.json by walking up from this file's location.
Uses glob to match any registry file (aligned with drone's registry_handler).
"""
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
candidate = parent / "AIPASS_REGISTRY.json"
if candidate.exists():
return candidate
matches = sorted(parent.glob("*_REGISTRY.json"))
if matches:
return matches[0]
# Fallback: walk up from CWD (matches drone's search order)
cwd = Path.cwd()
for parent in [cwd] + list(cwd.parents):
matches = sorted(parent.glob("*_REGISTRY.json"))
if matches:
return matches[0]
return Path.cwd() / "AIPASS_REGISTRY.json"
@@ -32,6 +32,7 @@ from aipass.prax import logger
# CLI services (display/output formatting)
from aipass.cli import console, header
from aipass.cli.apps.modules import error, warning
# JSON handler for tracking
from aipass.seedgo.apps.handlers.json import json_handler
@@ -204,21 +205,18 @@ def handle_command(command: str, args: List[str]) -> bool:
# Validate pack name
packs = _discover_packs()
if pack_name not in packs:
console.print(f"\n[red]Unknown pack:[/red] '{pack_name}'")
console.print()
console.print("[yellow]Available packs:[/yellow]")
for name in packs:
console.print(f" [cyan]{name}[/cyan]")
console.print()
console.print(f"[dim]Usage: drone @seedgo audit {next(iter(packs), '<pack>')}[/dim]")
console.print()
available = ", ".join(packs.keys())
error(
f"Unknown pack: '{pack_name}'",
suggestion=f"Available packs: {available}. Usage: drone @seedgo audit {next(iter(packs), '<pack>')}"
)
return True
pack_path = packs[pack_name]
# Handle --show-bypasses mode (placeholder — bypass audit merged into audit per D11)
if show_bypasses:
console.print("[yellow]--show-bypasses not yet implemented in seedgo[/yellow]")
warning("--show-bypasses not yet implemented in seedgo")
return True
# =========================================================================
+2 -2
View File
@@ -22,6 +22,7 @@ from typing import List, Any
from aipass.prax import logger
from aipass.cli import console, header
from aipass.cli.apps.modules import error
VERSION = "2.0.0"
SEEDGO_ROOT = Path(__file__).parent
@@ -228,8 +229,7 @@ def main() -> int:
if route_command(command, remaining, modules):
return 0
console.print(f"[red]Unknown command:[/red] {command}")
console.print("Run 'seedgo --help' for usage")
error(f"Unknown command: {command}", suggestion="Run 'seedgo --help' for usage")
return 1
@@ -8,10 +8,13 @@
"""Placeholder replacement engine for agent templates."""
import json
import re
from datetime import datetime
from pathlib import Path
from aipass.spawn.apps.handlers.registry import find_registry
def replace_placeholders(content, replacements):
"""Replace {{PLACEHOLDER}} patterns in content string."""
@@ -37,6 +40,16 @@ def build_replacements_dict(target_dir, branch_name, **overrides):
lower = branch_name.lower().replace("-", "_")
now = datetime.now()
# Read registry ID — never crash spawn if registry is missing
registry_id = ""
try:
registry_path = find_registry()
if registry_path.exists():
data = json.loads(registry_path.read_text(encoding="utf-8"))
registry_id = data.get("metadata", {}).get("id", "")
except Exception:
registry_id = ""
replacements = {
"BRANCHNAME": upper,
"branchname": lower,
@@ -50,6 +63,7 @@ def build_replacements_dict(target_dir, branch_name, **overrides):
"TRAITS": overrides.get("traits", ""),
"PURPOSE_BRIEF": overrides.get("purpose", "New agent - purpose TBD"),
"CITIZEN_NUMBER": str(overrides.get("citizen_number", 0)),
"REGISTRY_ID": registry_id,
"KEY_CAPABILITIES": "",
"DEPENDS_ON": "",
"PROVIDES_TO": "",
+32 -25
View File
@@ -1,12 +1,12 @@
# =================== AIPass ====================
# Name: registry.py
# Description: AIPASS_REGISTRY.json CRUD operations
# Description: *_REGISTRY.json discovery and CRUD operations
# Version: 1.0.0
# Created: 2026-03-05
# Modified: 2026-03-07
# =============================================
"""AIPASS_REGISTRY.json CRUD operations."""
"""*_REGISTRY.json discovery and CRUD operations."""
import json
import os
@@ -31,49 +31,56 @@ def _branches_as_list(branches):
return []
def _glob_registry(directory):
"""Find the first *_REGISTRY.json in a directory (sorted for consistency).
Args:
directory: Path to search in
Returns:
Path to the registry file, or None if not found
"""
matches = sorted(directory.glob("*_REGISTRY.json"))
return matches[0] if matches else None
def find_registry(start_path=None):
"""
Find AIPASS_REGISTRY.json — consistent with drone's resolution.
Find *_REGISTRY.json — walks up from __file__ and start_path/cwd.
The first *_REGISTRY.json found while walking up IS the project boundary.
If multiple exist in the same directory, picks the first alphabetically.
Priority:
1. AIPASS_REGISTRY environment variable
2. Project root (directory with pyproject.toml or .git) — walk up from __file__
3. Project root — walk up from start_path/cwd
4. Walk up from __file__ for any existing registry
5. Last resort: cwd
2. Walk up from __file__ — first dir containing *_REGISTRY.json
3. Walk up from start_path/cwd — first dir containing *_REGISTRY.json
4. Last resort: cwd / AIPASS_REGISTRY.json (backwards compat)
Args:
start_path: Directory to start searching from
Returns:
Path to AIPASS_REGISTRY.json
Path to *_REGISTRY.json
"""
# Check environment variable first (same as drone's config.py)
env_path = os.environ.get("AIPASS_REGISTRY")
if env_path:
return Path(env_path)
# Walk up from package location — find project root first
pkg_dir = Path(__file__).resolve().parent
for parent in [pkg_dir] + list(pkg_dir.parents):
if (parent / "pyproject.toml").exists() or (parent / ".git").exists():
candidate = parent / "AIPASS_REGISTRY.json"
if candidate.exists():
return candidate
# Project root found but no registry — create here
return candidate
# Walk up from start_path or cwd
# Walk up from start_path or cwd FIRST — user's location takes priority
current = Path(start_path).resolve() if start_path else Path.cwd()
for parent in [current] + list(current.parents):
if (parent / "pyproject.toml").exists() or (parent / ".git").exists():
return parent / "AIPASS_REGISTRY.json"
found = _glob_registry(parent)
if found:
return found
# Fallback: any existing registry walking up from package
# Walk up from package location (fallback for editable installs)
pkg_dir = Path(__file__).resolve().parent
for parent in [pkg_dir] + list(pkg_dir.parents):
candidate = parent / "AIPASS_REGISTRY.json"
if candidate.exists():
return candidate
found = _glob_registry(parent)
if found:
return found
# Last resort: cwd
return Path.cwd() / "AIPASS_REGISTRY.json"
+8 -8
View File
@@ -14,7 +14,7 @@ All implementation logic lives in apps/handlers/delete_ops.py.
from aipass.prax import logger
# CLI service: from cli.apps.modules import console (via aipass namespace)
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.spawn.apps.handlers.delete_ops import delete_branch
@@ -65,7 +65,7 @@ def handle_delete(args: list[str]) -> int:
Returns exit code (0=success, 1=failure).
"""
if not args:
console.print("[yellow]Usage: drone @spawn delete <@branch> [--yes] [--dry-run][/yellow]")
warning("Usage: drone @spawn delete <@branch> [--yes] [--dry-run]")
console.print()
console.print(" [green]@branch[/green] Branch to archive and deregister")
console.print(" [green]--yes[/green] Skip confirmation prompt")
@@ -79,7 +79,7 @@ def handle_delete(args: list[str]) -> int:
targets = [a for a in args if not a.startswith("--")]
if not targets:
console.print("[red]Error: specify a branch name (e.g. @api)[/red]")
error("specify a branch name (e.g. @api)")
return 1
branch_name = targets[0].lstrip("@").lower()
@@ -92,7 +92,7 @@ def handle_delete(args: list[str]) -> int:
)
except Exception as exc:
logger.error(f"[delete] Unexpected error deleting {branch_name}: {exc}")
console.print(f"[red]Error deleting {branch_name}: {exc}[/red]")
error(f"Error deleting {branch_name}: {exc}")
return 1
_print_summary(result, dry_run)
@@ -113,7 +113,7 @@ def _print_summary(result: dict, dry_run: bool) -> None:
if success:
console.print(f"[green]Delete {mode}{branch}[/green]")
else:
console.print(f"[red]Delete FAILED {mode}{branch}[/red]")
error(f"Delete FAILED {mode}{branch}")
archive_path = result.get("archive_path", "")
if archive_path:
@@ -121,8 +121,8 @@ def _print_summary(result: dict, dry_run: bool) -> None:
console.print(f" Registry updated: {result.get('registry_updated', False)}")
error = result.get("error", "")
if error:
console.print(f" [red]Error: {error}[/red]")
err_msg = result.get("error", "")
if err_msg:
error(err_msg)
console.print()
+6 -6
View File
@@ -15,7 +15,7 @@ All implementation logic lives in apps/handlers/passport_ops.py.
import argparse
from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.spawn.apps.handlers.passport_ops import grant_passport
@@ -58,13 +58,13 @@ def handle_passport(args: list[str]) -> int:
Returns exit code (0=success, 1=failure).
"""
if not args:
console.print("[yellow]Usage: drone @spawn passport <@dirname> [--role ...] [--purpose ...][/yellow]")
warning("Usage: drone @spawn passport <@dirname> [--role ...] [--purpose ...]")
console.print()
console.print(" [dim]Grant birthright citizenship to a directory[/dim]")
console.print()
console.print(" [green]@dirname[/green] Directory to grant citizenship (created if needed)")
console.print(" [yellow]--role[/yellow] Role description for the passport")
console.print(" [yellow]--purpose[/yellow] Purpose description")
warning("--role", details="Role description for the passport")
warning("--purpose", details="Purpose description")
return 1
parser = argparse.ArgumentParser(prog="spawn passport", add_help=False)
@@ -96,9 +96,9 @@ def handle_passport(args: list[str]) -> int:
console.print(f" Files: {result['files_copied']}")
console.print(f" Registry: {'updated' if result['registry_updated'] else 'not updated'}")
if result["validation_issues"]:
console.print(f" [yellow]Warnings: {len(result['validation_issues'])} unreplaced placeholders[/yellow]")
warning(f"{len(result['validation_issues'])} unreplaced placeholders")
console.print()
return 0
else:
console.print(f"[red]Error: {result['error']}[/red]")
error(result['error'])
return 1
@@ -14,7 +14,7 @@ All implementation logic lives in apps/handlers/sync_registry_ops.py.
from aipass.prax import logger
# CLI service: from cli.apps.modules import console (via aipass namespace)
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.spawn.apps.handlers.sync_registry_ops import sync_registry
@@ -64,7 +64,7 @@ def handle_sync_registry(args: list[str]) -> int:
Returns exit code (0=success, 1=failure).
"""
if args and args[0] in ["--help", "-h"]:
console.print("[yellow]Usage: drone @spawn sync-registry [--fix][/yellow]")
warning("Usage: drone @spawn sync-registry [--fix]")
console.print()
console.print(" [green](no args)[/green] Report mismatches between registry and filesystem")
console.print(" [green]--fix[/green] Auto-repair: remove stale, add unregistered")
@@ -76,7 +76,7 @@ def handle_sync_registry(args: list[str]) -> int:
result = sync_registry(fix=fix)
except Exception as exc:
logger.error(f"[sync-registry] Unexpected error: {exc}")
console.print(f"[red]Error: {exc}[/red]")
error(str(exc))
return 1
_print_summary(result)
@@ -106,7 +106,7 @@ def _print_summary(result: dict) -> None:
# Stale
if stale:
console.print(f" [red]Stale ({len(stale)}):[/red] [dim]registered but missing/invalid[/dim]")
error(f"Stale ({len(stale)}): registered but missing/invalid")
for name in sorted(stale):
console.print(f" {name}")
else:
@@ -114,7 +114,7 @@ def _print_summary(result: dict) -> None:
# Unregistered
if unregistered:
console.print(f" [yellow]Unregistered ({len(unregistered)}):[/yellow] [dim]on disk but not in registry[/dim]")
warning(f"Unregistered ({len(unregistered)}): on disk but not in registry")
for name in sorted(unregistered):
console.print(f" {name}")
else:
@@ -14,7 +14,7 @@ All implementation logic lives in apps/handlers/sync_templates_ops.py.
from aipass.prax import logger
# CLI service: from cli.apps.modules import console (via aipass namespace)
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.spawn.apps.handlers.sync_templates_ops import sync_templates
@@ -66,7 +66,7 @@ def handle_sync_templates(args: list[str]) -> int:
Returns exit code (0=success, 1=failure).
"""
if args and args[0] in ["--help", "-h"]:
console.print("[yellow]Usage: drone @spawn sync-templates [--status|--sync|--dry-run][/yellow]")
warning("Usage: drone @spawn sync-templates [--status|--sync|--dry-run]")
console.print()
console.print(" [green](no args)[/green] Report which managed template files are stale")
console.print(" [green]--status[/green] Same as no args")
@@ -81,7 +81,7 @@ def handle_sync_templates(args: list[str]) -> int:
result = sync_templates(sync=sync, dry_run=dry_run)
except Exception as exc:
logger.error(f"[sync-templates] Unexpected error: {exc}")
console.print(f"[red]Error: {exc}[/red]")
error(str(exc))
return 1
_print_summary(result, dry_run)
@@ -119,7 +119,7 @@ def _print_summary(result: dict, dry_run: bool) -> None:
console.print(f" {name}")
if stale:
console.print(f" [yellow]Stale ({len(stale)}):[/yellow]")
warning(f"Stale ({len(stale)}):")
for name in stale:
console.print(f" {name}")
@@ -129,7 +129,7 @@ def _print_summary(result: dict, dry_run: bool) -> None:
console.print(f" {name}")
if errors:
console.print(f" [red]Errors ({len(errors)}):[/red]")
error(f"Errors ({len(errors)}):")
for err in errors:
console.print(f" {err}")
+8 -10
View File
@@ -14,7 +14,7 @@ All implementation logic lives in apps/handlers/update_ops.py.
from aipass.prax import logger
# CLI service: from cli.apps.modules import console (via aipass namespace)
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error, warning
from aipass.spawn.apps.handlers.update_ops import (
update_branch,
@@ -72,7 +72,7 @@ def handle_update(args: list[str]) -> int:
Returns exit code (0=success, 1=failure).
"""
if not args:
console.print("[yellow]Usage: drone @spawn update <@branch|class --all> [--dry-run] [--trace][/yellow]")
warning("Usage: drone @spawn update <@branch|class --all> [--dry-run] [--trace]")
console.print()
console.print(" [green]@branch[/green] Update a single branch (uses its own class)")
console.print(" [green]builder --all[/green] Update all builder-class branches")
@@ -100,9 +100,7 @@ def handle_update(args: list[str]) -> int:
if "--all" in args:
if citizen_class is None:
classes = ", ".join(get_available_classes())
console.print(f"[red]Error: --all requires a citizen class[/red]")
console.print(f"[dim]Specify a class: drone @spawn update <class> --all[/dim]")
console.print(f"[dim]Available classes: {classes}[/dim]")
error("--all requires a citizen class", suggestion=f"drone @spawn update <class> --all (available: {classes})")
return 1
results = update_all(dry_run=dry_run, trace=trace, citizen_class=citizen_class)
@@ -110,7 +108,7 @@ def handle_update(args: list[str]) -> int:
return 0 if all(r.get("success") for r in results) else 1
if not targets:
console.print("[red]Error: specify a branch name (e.g. @prax) or use <class> --all[/red]")
error("specify a branch name (e.g. @prax) or use <class> --all")
return 1
# Take the first target, strip leading @
@@ -120,7 +118,7 @@ def handle_update(args: list[str]) -> int:
result = update_branch(branch_name, dry_run=dry_run, trace=trace)
except Exception as exc:
logger.error(f"[update] Unexpected error updating {branch_name}: {exc}")
console.print(f"[red]Error updating {branch_name}: {exc}[/red]")
error(f"Error updating {branch_name}: {exc}")
return 1
_print_branch_summary(result, dry_run)
@@ -141,7 +139,7 @@ def _print_branch_summary(result: dict, dry_run: bool) -> None:
if success:
console.print(f"[green]Update {mode}{branch}[/green]")
else:
console.print(f"[red]Update FAILED {mode}{branch}[/red]")
error(f"Update FAILED {mode}{branch}")
console.print(f" Additions: {result.get('additions', 0)}")
console.print(f" Renames: {result.get('renames', 0)}")
@@ -151,7 +149,7 @@ def _print_branch_summary(result: dict, dry_run: bool) -> None:
errs = result.get("errors", [])
if errs:
console.print(f" [red]Errors: {len(errs)}[/red]")
error(f"Errors: {len(errs)}")
for e in errs:
console.print(f" - {e}")
@@ -223,5 +221,5 @@ def _print_all_summary(results: list[dict], dry_run: bool) -> None:
f"!{total_skip} py-skipped")
if total_err:
console.print(f" [red]{total_err} errors across all branches[/red]")
error(f"{total_err} errors across all branches")
console.print()
+14 -17
View File
@@ -17,7 +17,7 @@ import sys
import argparse
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header
from aipass.cli.apps.modules import console, header, error, warning
def print_help():
@@ -52,13 +52,13 @@ def print_help():
console.print()
console.print("[bold cyan]OPTIONS:[/bold cyan]")
console.print()
console.print(" [yellow]--role[/yellow] Agent role description")
console.print(" [yellow]--traits[/yellow] Agent personality traits")
console.print(" [yellow]--purpose[/yellow] Agent purpose (brief)")
console.print(" [yellow]--template[/yellow] Custom template directory")
console.print(" [yellow]--registry[/yellow] Path to AIPASS_REGISTRY.json")
console.print(" [yellow]--dry-run[/yellow] Preview changes without modifying files")
console.print(" [yellow]--trace[/yellow] Enable verbose logging")
warning("--role", details="Agent role description")
warning("--traits", details="Agent personality traits")
warning("--purpose", details="Agent purpose (brief)")
warning("--template", details="Custom template directory")
warning("--registry", details="Path to AIPASS_REGISTRY.json")
warning("--dry-run", details="Preview changes without modifying files")
warning("--trace", details="Enable verbose logging")
console.print()
@@ -68,8 +68,7 @@ def handle_create(args):
from aipass.spawn.apps.modules.core import validate_class, get_default_class
if not args:
console.print("[red]Error: target path required[/red]")
console.print("[dim]Usage: drone @spawn create [class] <target_path> [--role ...][/dim]")
error("target path required", suggestion="drone @spawn create [class] <target_path> [--role ...]")
return 1
# Check if first arg is a citizen class
@@ -79,7 +78,7 @@ def handle_create(args):
citizen_class = args[0]
remaining_args = args[1:]
if not remaining_args:
console.print("[red]Error: target path required after class name[/red]")
error("target path required after class name")
return 1
parser = argparse.ArgumentParser(prog="spawn create", add_help=False)
@@ -110,11 +109,11 @@ def handle_create(args):
console.print(f" Files: {result['files_copied']}")
console.print(f" Registry: {'updated' if result['registry_updated'] else 'not updated'}")
if result["validation_issues"]:
console.print(f" [yellow]Warnings: {len(result['validation_issues'])} unreplaced placeholders[/yellow]")
warning(f"{len(result['validation_issues'])} unreplaced placeholders")
console.print()
return 0
else:
console.print(f"[red]Error: {result['error']}[/red]")
error(result['error'])
return 1
@@ -183,12 +182,10 @@ def main():
}
if command in stub_commands:
console.print(f"[yellow]'{command}' is not yet implemented.[/yellow]")
console.print(f"[dim]Planned: {stub_commands[command]}[/dim]")
warning(f"'{command}' is not yet implemented.", details=f"Planned: {stub_commands[command]}")
return 1
console.print(f"[red]Unknown command: {command}[/red]")
console.print("[dim]Run 'drone @spawn --help' for available commands[/dim]")
error(f"Unknown command: {command}", suggestion="Run 'drone @spawn --help' for available commands")
return 1
@@ -0,0 +1,38 @@
{
"document_metadata": {
"document_type": "branch_identity",
"document_name": "{{BRANCHNAME}}.PASSPORT",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": ["identity", "passport", "branch_profile"]
},
"branch_info": {
"branch_name": "{{BRANCHNAME}}",
"alias": "",
"path": "{{CWD}}",
"module": "{{MODULE}}",
"created": "{{DATE}}"
},
"identity": {
"citizen_class": "birthright",
"role": "{{ROLE}}",
"purpose": "{{PURPOSE_BRIEF}}",
"what_i_do": [],
"what_i_dont_do": []
},
"principles": [
"Code is truth - fail honestly",
"Memory persists - context survives",
"Simple solutions over complex architecture"
],
"citizenship": {
"registered": true,
"registry_id": "{{REGISTRY_ID}}",
"registry_path": ".aipass/registry.json",
"communications": true,
"memory": true
}
}
@@ -0,0 +1,38 @@
{
"document_metadata": {
"document_type": "branch_identity",
"document_name": "{{BRANCHNAME}}.PASSPORT",
"version": "1.0.0",
"schema_version": "1.0.0",
"created": "{{DATE}}",
"last_updated": "{{DATE}}",
"managed_by": "{{BRANCHNAME}}",
"tags": ["identity", "passport", "branch_profile"]
},
"branch_info": {
"branch_name": "{{BRANCHNAME}}",
"alias": "",
"path": "{{CWD}}",
"module": "{{MODULE}}",
"created": "{{DATE}}"
},
"identity": {
"citizen_class": "builder",
"role": "{{ROLE}}",
"purpose": "",
"what_i_do": [],
"what_i_dont_do": []
},
"principles": [
"Code is truth - fail honestly",
"Memory persists - context survives",
"Simple solutions over complex architecture"
],
"citizenship": {
"registered": true,
"registry_id": "{{REGISTRY_ID}}",
"registry_path": ".aipass/registry.json",
"communications": true,
"memory": true
}
}
+2 -2
View File
@@ -185,7 +185,7 @@ def handle_command(command: str, args: list) -> bool:
Returns:
True if command was handled, False otherwise
"""
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
# Handle module-name routing (drone @trigger core <subcmd>)
if command == "core":
@@ -203,7 +203,7 @@ def handle_command(command: str, args: list) -> bool:
if command == "fire":
if not args:
console.print("[red]Usage: drone @trigger fire <event> [key=value ...][/red]")
error("Usage: drone @trigger fire <event> [key=value ...]")
return True
event_name = args[0]
# Parse key=value pairs from remaining args
+5 -7
View File
@@ -147,7 +147,7 @@ def handle_command(command: str, args: list) -> bool:
Returns:
True if command was handled, False otherwise
"""
from aipass.cli.apps.modules import console
from aipass.cli.apps.modules import console, error
if command != "errors":
return False
@@ -170,8 +170,7 @@ def handle_command(command: str, args: list) -> bool:
if sub in routes:
return routes[sub](console, rest)
console.print(f"[red]Unknown subcommand: {sub}[/red]")
console.print("Run [dim]drone @trigger errors help[/dim] for available commands")
error(f"Unknown subcommand: {sub}", suggestion="Run 'drone @trigger errors help' for available commands")
return True
@@ -229,16 +228,15 @@ def _cmd_list(console, args: list) -> bool:
def _cmd_detail(console, args: list) -> bool:
"""Show full error details for a fingerprint or ID."""
from rich.panel import Panel
from aipass.cli.apps.modules import error
if not args:
console.print("[red]Missing error ID or fingerprint[/red]")
console.print("Usage: drone @trigger errors detail <id_or_fingerprint>")
error("Missing error ID or fingerprint", suggestion="Usage: drone @trigger errors detail <id_or_fingerprint>")
return True
entry = _find_by_id_or_fp(args[0])
if not entry:
console.print(f"[red]Error not found:[/red] {args[0]}")
console.print(" [dim]Try a fingerprint prefix, full fingerprint, or short ID[/dim]")
error(f"Error not found: {args[0]}", suggestion="Try a fingerprint prefix, full fingerprint, or short ID")
return True
st = entry.get("status", "?")
+2 -4
View File
@@ -27,7 +27,7 @@ from typing import List, Any
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services for formatted output
from aipass.cli.apps.modules import console, header
from aipass.cli.apps.modules import console, header, error
# =============================================================================
# MODULE DISCOVERY
@@ -209,9 +209,7 @@ def main():
return 0
else:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
console.print()
console.print("Run [dim]python3 trigger.py --help[/dim] for available commands")
error(f"Unknown command: {command}", suggestion="Run 'python3 trigger.py --help' for available commands")
console.print()
return 1