feat(cli): 100% seedgo compliance — all 24 standards (#66)
* fix(ai_mail): resolve relative mailbox_path bug + archive dead handlers Fix get_user_by_email() and get_all_users() returning relative mailbox_path values from registry instead of absolute paths (causing doubled paths in reply). Added path resolution matching get_current_user() pattern. Archive 7 unused handler files to apps/.archive/ (gitignored): - pending_work.py — planned feature, never wired - lock_utils.py — superseded by inbox_lock.py - data_ops.py — consumer (error_monitor) was archived - config_generator.py, users/load.py, registry/validate.py — scaffolded, never used - trigger/error_handler.py — superseded by trigger branch's event system Added test_user_paths.py (13 tests) covering absolute path resolution. Co-Authored-By: @ai_mail <ai_mail@aipass> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * feat(cli): achieve 100% seedgo compliance across all 24 standards Clean up json_handler (remove prax dependency, use portable Path resolution), wire json_handler into display.py, templates.py, and init_project.py using module import pattern. Restructure handle_command() with seedgo-compliant no-args introspection gate. Update README with init commands and architecture. Co-Authored-By: @cli <cli@aipass> --------- Co-authored-by: @ai_mail <ai_mail@aipass> Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: @cli <cli@aipass>
This commit is contained in:
AIPass
co-authored by
@cli
@ai_mail
Claude Opus 4.6
@cli
@ai_mail
Claude Opus 4.6
parent
eb527fe0b2
commit
d7f2587465
@@ -1,151 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: pending_work.py
|
||||
# Description: Pending Work Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-02-17
|
||||
# Modified: 2026-02-17
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Pending Work Handler
|
||||
|
||||
Read/write utils for .pending_work.json per branch.
|
||||
Tracks dispatch workflows, waiting_for states, and next_action queues.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
PENDING_WORK_FILENAME = ".pending_work.json"
|
||||
|
||||
|
||||
def _get_pending_path(branch_path: Path) -> Path:
|
||||
"""Get the pending work file path for a branch."""
|
||||
if branch_path == Path("/"):
|
||||
return Path.cwd() / ".ai_mail.local" / PENDING_WORK_FILENAME
|
||||
return branch_path / ".ai_mail.local" / PENDING_WORK_FILENAME
|
||||
|
||||
|
||||
def load_pending_work(branch_path: Path) -> Dict[str, Any]:
|
||||
"""
|
||||
Load pending work for a branch.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the branch directory
|
||||
|
||||
Returns:
|
||||
Pending work dict with 'workflows' array
|
||||
"""
|
||||
pending_path = _get_pending_path(branch_path)
|
||||
|
||||
if not pending_path.exists():
|
||||
return {"workflows": []}
|
||||
|
||||
try:
|
||||
with open(pending_path, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
if "workflows" not in data:
|
||||
data["workflows"] = []
|
||||
return data
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return {"workflows": []}
|
||||
|
||||
|
||||
def save_pending_work(branch_path: Path, data: Dict[str, Any]) -> bool:
|
||||
"""
|
||||
Save pending work for a branch.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the branch directory
|
||||
data: Pending work dict to save
|
||||
|
||||
Returns:
|
||||
True if saved successfully
|
||||
"""
|
||||
pending_path = _get_pending_path(branch_path)
|
||||
pending_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
data["last_updated"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
with open(pending_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def add_workflow(
|
||||
branch_path: Path,
|
||||
dispatch_id: str,
|
||||
dispatched_to: str,
|
||||
subject: str,
|
||||
waiting_for: Optional[str] = None,
|
||||
next_action: Optional[str] = None
|
||||
) -> bool:
|
||||
"""
|
||||
Add a workflow entry to a branch's pending work.
|
||||
|
||||
Adds a new dispatch workflow to track pending communications and actions.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the branch directory
|
||||
dispatch_id: Message ID of the dispatched email
|
||||
dispatched_to: Target branch email (e.g., @flow)
|
||||
subject: Subject of the dispatched email
|
||||
waiting_for: What the branch is waiting for (e.g., reply from @flow)
|
||||
next_action: What to do when the response arrives
|
||||
|
||||
Returns:
|
||||
True if added successfully
|
||||
"""
|
||||
data = load_pending_work(branch_path)
|
||||
|
||||
entry = {
|
||||
"dispatch_id": dispatch_id,
|
||||
"dispatched_to": dispatched_to,
|
||||
"subject": subject,
|
||||
"status": "waiting",
|
||||
"created": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
}
|
||||
if waiting_for:
|
||||
entry["waiting_for"] = waiting_for
|
||||
if next_action:
|
||||
entry["next_action"] = next_action
|
||||
|
||||
data["workflows"].append(entry)
|
||||
return save_pending_work(branch_path, data)
|
||||
|
||||
|
||||
def clear_workflow(branch_path: Path, dispatch_id: str) -> bool:
|
||||
"""
|
||||
Remove a completed workflow entry.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the branch directory
|
||||
dispatch_id: Message ID to remove
|
||||
|
||||
Returns:
|
||||
True if removed successfully
|
||||
"""
|
||||
data = load_pending_work(branch_path)
|
||||
data["workflows"] = [
|
||||
w for w in data["workflows"]
|
||||
if w.get("dispatch_id") != dispatch_id
|
||||
]
|
||||
return save_pending_work(branch_path, data)
|
||||
|
||||
|
||||
def has_pending_work(branch_path: Path) -> bool:
|
||||
"""
|
||||
Check if a branch has any pending workflows.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the branch directory
|
||||
|
||||
Returns:
|
||||
True if there are active workflows
|
||||
"""
|
||||
data = load_pending_work(branch_path)
|
||||
return len(data.get("workflows", [])) > 0
|
||||
@@ -1,202 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: lock_utils.py
|
||||
# Description: Dispatch Lock Handler
|
||||
# Version: 1.1.0
|
||||
# Created: 2026-02-09
|
||||
# Modified: 2026-02-09
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Dispatch Lock Handler
|
||||
|
||||
PID-based single instance lock per branch.
|
||||
Prevents multiple dispatch agents from spawning concurrently at the same branch.
|
||||
Uses atomic file creation (O_CREAT|O_EXCL) to avoid race conditions.
|
||||
"""
|
||||
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# Standard logging
|
||||
|
||||
# Lock file name - placed in branch's .ai_mail.local/ directory
|
||||
LOCK_FILENAME = ".dispatch.lock"
|
||||
|
||||
# Stale lock timeout in seconds (10 minutes)
|
||||
STALE_LOCK_TIMEOUT = 600
|
||||
|
||||
|
||||
def _get_lock_path(branch_path: Path) -> Path:
|
||||
"""Get the lock file path for a branch."""
|
||||
if branch_path == Path("/"):
|
||||
return Path.cwd() / ".ai_mail.local" / LOCK_FILENAME
|
||||
return branch_path / ".ai_mail.local" / LOCK_FILENAME
|
||||
|
||||
|
||||
def _is_pid_running(pid: int) -> bool:
|
||||
"""Check if a process with the given PID is still running."""
|
||||
try:
|
||||
os.kill(pid, 0)
|
||||
return True
|
||||
except PermissionError:
|
||||
# Process exists but we don't have permission to signal it
|
||||
return True
|
||||
except ProcessLookupError:
|
||||
return False
|
||||
|
||||
|
||||
def _is_lock_stale(lock_data: dict) -> bool:
|
||||
"""
|
||||
Check if a lock is stale (process dead or timeout exceeded).
|
||||
|
||||
Returns True if the lock should be considered stale and can be removed.
|
||||
"""
|
||||
pid = lock_data.get("pid")
|
||||
timestamp = lock_data.get("timestamp")
|
||||
|
||||
# No PID = stale
|
||||
if pid is None:
|
||||
return True
|
||||
|
||||
# Process no longer running = stale
|
||||
if not _is_pid_running(pid):
|
||||
return True # Process no longer running - stale
|
||||
|
||||
# Timeout check - if lock is older than STALE_LOCK_TIMEOUT, consider stale
|
||||
if timestamp:
|
||||
try:
|
||||
lock_time = datetime.fromisoformat(timestamp)
|
||||
elapsed = (datetime.now() - lock_time).total_seconds()
|
||||
if elapsed > STALE_LOCK_TIMEOUT:
|
||||
return True # Timeout exceeded - stale
|
||||
except (ValueError, TypeError) as e:
|
||||
pass # Unparseable timestamp - ignore
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def acquire_lock(branch_path: Path, pid: int) -> tuple[bool, str]:
|
||||
"""
|
||||
Attempt to acquire a dispatch lock for a branch.
|
||||
|
||||
Uses atomic file creation to prevent race conditions.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the target branch
|
||||
pid: PID of the agent being spawned
|
||||
|
||||
Returns:
|
||||
Tuple of (acquired: bool, message: str)
|
||||
If not acquired, message contains reason (e.g., existing PID info)
|
||||
"""
|
||||
lock_path = _get_lock_path(branch_path)
|
||||
|
||||
# Ensure parent directory exists
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Check for existing lock first
|
||||
if lock_path.exists():
|
||||
try:
|
||||
with open(lock_path, 'r', encoding='utf-8') as f:
|
||||
existing_lock = json.load(f)
|
||||
|
||||
if _is_lock_stale(existing_lock):
|
||||
# Remove stale lock and try again
|
||||
lock_path.unlink(missing_ok=True) # Remove stale lock
|
||||
else:
|
||||
# Active lock exists - bounce
|
||||
existing_pid = existing_lock.get("pid", "unknown")
|
||||
existing_sender = existing_lock.get("sender", "unknown")
|
||||
existing_time = existing_lock.get("timestamp", "unknown")
|
||||
msg = f"Branch already has active dispatch agent (PID: {existing_pid}, sender: {existing_sender}, since: {existing_time})"
|
||||
return False, msg
|
||||
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
# Corrupted lock file - remove it
|
||||
lock_path.unlink(missing_ok=True) # Corrupted lock - remove it
|
||||
|
||||
# Create lock file atomically using O_CREAT|O_EXCL
|
||||
lock_data = {
|
||||
"pid": pid,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"branch": str(branch_path)
|
||||
}
|
||||
|
||||
try:
|
||||
fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
|
||||
try:
|
||||
os.write(fd, json.dumps(lock_data, indent=2).encode('utf-8'))
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
return True, "Lock acquired"
|
||||
|
||||
except FileExistsError:
|
||||
# Race condition - another process created the lock between our check and create
|
||||
msg = "Lock acquisition failed - another dispatch just started"
|
||||
return False, msg
|
||||
|
||||
|
||||
def release_lock(branch_path: Path, pid: int | None = None) -> bool:
|
||||
"""
|
||||
Release a dispatch lock for a branch.
|
||||
|
||||
Args:
|
||||
branch_path: Path to the target branch
|
||||
pid: If provided, only release if lock belongs to this PID (safety check)
|
||||
|
||||
Returns:
|
||||
True if lock was released, False if not found or owned by different PID
|
||||
"""
|
||||
lock_path = _get_lock_path(branch_path)
|
||||
|
||||
if not lock_path.exists():
|
||||
return True # No lock = already released
|
||||
|
||||
# If PID specified, verify ownership before releasing
|
||||
if pid is not None:
|
||||
try:
|
||||
with open(lock_path, 'r', encoding='utf-8') as f:
|
||||
lock_data = json.load(f)
|
||||
if lock_data.get("pid") != pid:
|
||||
return False # Lock owned by different PID
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
pass # Corrupted lock file during release - continue anyway
|
||||
|
||||
try:
|
||||
lock_path.unlink(missing_ok=True)
|
||||
return True
|
||||
except OSError as e:
|
||||
return False # Failed to release lock
|
||||
|
||||
|
||||
def check_lock(branch_path: Path) -> dict | None:
|
||||
"""
|
||||
Check if a branch has an active dispatch lock.
|
||||
|
||||
Returns:
|
||||
Lock data dict if active lock exists, None otherwise.
|
||||
Automatically cleans up stale locks.
|
||||
"""
|
||||
lock_path = _get_lock_path(branch_path)
|
||||
|
||||
if not lock_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(lock_path, 'r', encoding='utf-8') as f:
|
||||
lock_data = json.load(f)
|
||||
|
||||
if _is_lock_stale(lock_data):
|
||||
# Auto-cleanup stale lock
|
||||
lock_path.unlink(missing_ok=True) # Auto-cleaned stale lock
|
||||
return None
|
||||
|
||||
return lock_data
|
||||
|
||||
except (json.JSONDecodeError, OSError):
|
||||
# Corrupted - clean up
|
||||
lock_path.unlink(missing_ok=True)
|
||||
return None
|
||||
@@ -1,66 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: data_ops.py
|
||||
# Description: Error Monitor Data Operations Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2025-11-15
|
||||
# Modified: 2025-11-15
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Error Monitor Data Operations Handler
|
||||
|
||||
Independent handler for error tracking data persistence.
|
||||
Provides functions for loading and saving error monitor data.
|
||||
|
||||
Architecture:
|
||||
- No cross-domain imports (independent handler)
|
||||
- Provides: data loading, data saving
|
||||
- Used by: error_monitor module
|
||||
"""
|
||||
|
||||
# =============================================
|
||||
# IMPORTS
|
||||
# =============================================
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
|
||||
# =============================================
|
||||
# BUSINESS LOGIC
|
||||
# =============================================
|
||||
|
||||
def load_data(data_file: Path) -> Dict:
|
||||
"""
|
||||
Load error tracking data from file.
|
||||
|
||||
Args:
|
||||
data_file: Path to the error tracking data file
|
||||
|
||||
Returns:
|
||||
Dictionary containing error tracking data, or empty dict if file doesn't exist
|
||||
"""
|
||||
if not data_file.exists():
|
||||
return {}
|
||||
|
||||
try:
|
||||
# Direct file read for error tracking data (non-standard structure)
|
||||
with open(data_file, 'r', encoding='utf-8') as f:
|
||||
data = json.load(f)
|
||||
return data if isinstance(data, dict) else {}
|
||||
except Exception:
|
||||
return {}
|
||||
|
||||
|
||||
def save_data(data: Dict, data_file: Path) -> None:
|
||||
"""
|
||||
Save error tracking data to file.
|
||||
|
||||
Args:
|
||||
data: Dictionary containing error tracking data
|
||||
data_file: Path to the error tracking data file
|
||||
"""
|
||||
# Direct file write for error tracking data (non-standard structure)
|
||||
data_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(data_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, indent=2)
|
||||
@@ -1,240 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: validate.py
|
||||
# Description: Registry Validation Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2025-11-15
|
||||
# Modified: 2025-11-15
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Registry Validation Handler
|
||||
|
||||
Handles validation of branch registry data including:
|
||||
- Checking for email address collisions
|
||||
- Validating email derivation rules
|
||||
- Detecting unreachable branches
|
||||
|
||||
Handler Independence:
|
||||
- No module imports from ai_mail
|
||||
- Only uses Prax logger and standard library
|
||||
- Fully transportable and self-contained
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Tuple
|
||||
|
||||
|
||||
# Constants
|
||||
MODULE_NAME = "registry.validate"
|
||||
|
||||
|
||||
def check_email_collisions(branches: List[Dict]) -> Tuple[bool, List[Dict]]:
|
||||
"""
|
||||
Check for email address collisions in branch list.
|
||||
|
||||
Args:
|
||||
branches: List of branch dicts with keys: name, path, email
|
||||
|
||||
Returns:
|
||||
Tuple of (has_collisions: bool, collisions: List[Dict])
|
||||
Collision dict format:
|
||||
{
|
||||
"email": "@email",
|
||||
"branch1": "First Branch Name",
|
||||
"branch2": "Second Branch Name"
|
||||
}
|
||||
"""
|
||||
email_map = {}
|
||||
collisions = []
|
||||
|
||||
for branch in branches:
|
||||
email = branch["email"]
|
||||
if email in email_map:
|
||||
# Collision detected
|
||||
collisions.append({
|
||||
"email": email,
|
||||
"branch1": email_map[email],
|
||||
"branch2": branch["name"]
|
||||
})
|
||||
else:
|
||||
email_map[email] = branch["name"]
|
||||
|
||||
has_collisions = len(collisions) > 0
|
||||
|
||||
return has_collisions, collisions
|
||||
|
||||
|
||||
def get_collision_report(collisions: List[Dict]) -> str:
|
||||
"""
|
||||
Generate human-readable collision report.
|
||||
|
||||
Args:
|
||||
collisions: List of collision dicts from check_email_collisions()
|
||||
|
||||
Returns:
|
||||
Formatted report string
|
||||
"""
|
||||
if not collisions:
|
||||
return "No collisions detected."
|
||||
|
||||
report_lines = [
|
||||
f"EMAIL ADDRESS COLLISIONS DETECTED: {len(collisions)}",
|
||||
"",
|
||||
"One or more branches are unreachable via AI_Mail!",
|
||||
"",
|
||||
"Collisions:"
|
||||
]
|
||||
|
||||
for collision in collisions:
|
||||
report_lines.extend([
|
||||
"",
|
||||
f" {collision['email']}",
|
||||
f" - {collision['branch1']}",
|
||||
f" - {collision['branch2']}",
|
||||
])
|
||||
|
||||
report_lines.extend([
|
||||
"",
|
||||
"Fix: Rename branches in BRANCH_REGISTRY.json to ensure unique email derivation",
|
||||
"See email derivation rules in registry/read.py"
|
||||
])
|
||||
|
||||
return "\n".join(report_lines)
|
||||
|
||||
|
||||
def validate_branch_data(branch: Dict) -> Tuple[bool, str]:
|
||||
"""
|
||||
Validate a single branch entry.
|
||||
|
||||
Args:
|
||||
branch: Branch dict with keys: name, path, email
|
||||
|
||||
Returns:
|
||||
Tuple of (is_valid: bool, error_message: str)
|
||||
error_message is empty string if valid
|
||||
"""
|
||||
# Check required fields
|
||||
required_fields = ["name", "path", "email"]
|
||||
for field in required_fields:
|
||||
if field not in branch:
|
||||
return False, f"Missing required field: {field}"
|
||||
if not branch[field]:
|
||||
return False, f"Empty value for field: {field}"
|
||||
|
||||
# Validate email format
|
||||
if not branch["email"].startswith("@"):
|
||||
return False, f"Email must start with @: {branch['email']}"
|
||||
|
||||
# Validate path format
|
||||
path = branch["path"]
|
||||
if not path.startswith("/"):
|
||||
return False, f"Path must be absolute: {path}"
|
||||
|
||||
return True, ""
|
||||
|
||||
|
||||
def validate_all_branches(branches: List[Dict]) -> Tuple[bool, List[str]]:
|
||||
"""
|
||||
Validate all branch entries.
|
||||
|
||||
Args:
|
||||
branches: List of branch dicts
|
||||
|
||||
Returns:
|
||||
Tuple of (all_valid: bool, error_messages: List[str])
|
||||
"""
|
||||
errors = []
|
||||
|
||||
for i, branch in enumerate(branches):
|
||||
is_valid, error_msg = validate_branch_data(branch)
|
||||
if not is_valid:
|
||||
errors.append(f"Branch {i} ({branch.get('name', 'UNKNOWN')}): {error_msg}")
|
||||
|
||||
all_valid = len(errors) == 0
|
||||
|
||||
return all_valid, errors
|
||||
|
||||
|
||||
def get_duplicate_names(branches: List[Dict]) -> List[str]:
|
||||
"""
|
||||
Find duplicate branch names.
|
||||
|
||||
Args:
|
||||
branches: List of branch dicts
|
||||
|
||||
Returns:
|
||||
List of duplicate branch names
|
||||
"""
|
||||
name_counts = {}
|
||||
duplicates = []
|
||||
|
||||
for branch in branches:
|
||||
name = branch.get("name", "")
|
||||
if name:
|
||||
name_counts[name] = name_counts.get(name, 0) + 1
|
||||
|
||||
for name, count in name_counts.items():
|
||||
if count > 1:
|
||||
duplicates.append(name)
|
||||
|
||||
return duplicates
|
||||
|
||||
|
||||
def get_duplicate_paths(branches: List[Dict]) -> List[str]:
|
||||
"""
|
||||
Find duplicate branch paths.
|
||||
|
||||
Args:
|
||||
branches: List of branch dicts
|
||||
|
||||
Returns:
|
||||
List of duplicate branch paths
|
||||
"""
|
||||
path_counts = {}
|
||||
duplicates = []
|
||||
|
||||
for branch in branches:
|
||||
path = branch.get("path", "")
|
||||
if path:
|
||||
path_counts[path] = path_counts.get(path, 0) + 1
|
||||
|
||||
for path, count in path_counts.items():
|
||||
if count > 1:
|
||||
duplicates.append(path)
|
||||
|
||||
return duplicates
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
from aipass.cli.apps.modules import console
|
||||
console.print("\n" + "="*70)
|
||||
console.print("AI_MAIL HANDLER: registry/validate.py")
|
||||
console.print("="*70)
|
||||
console.print("\nRegistry Validation Handler")
|
||||
console.print()
|
||||
console.print("FUNCTIONS PROVIDED:")
|
||||
console.print(" - check_email_collisions(branches) -> Tuple[bool, List[Dict]]")
|
||||
console.print(" - get_collision_report(collisions) -> str")
|
||||
console.print(" - validate_branch_data(branch) -> Tuple[bool, str]")
|
||||
console.print(" - validate_all_branches(branches) -> Tuple[bool, List[str]]")
|
||||
console.print(" - get_duplicate_names(branches) -> List[str]")
|
||||
console.print(" - get_duplicate_paths(branches) -> List[str]")
|
||||
console.print()
|
||||
console.print("TESTING:")
|
||||
|
||||
# Sample test data
|
||||
test_branches = [
|
||||
{"name": "AIPASS.admin", "path": "/", "email": "@admin"},
|
||||
{"name": "FLOW", "path": "src/aipass/flow", "email": "@flow"},
|
||||
{"name": "DRONE", "path": "src/aipass/drone", "email": "@drone"},
|
||||
]
|
||||
|
||||
has_collisions, collisions = check_email_collisions(test_branches)
|
||||
console.print(f"\nCollisions detected: {has_collisions}")
|
||||
console.print(f"Number of collisions: {len(collisions)}")
|
||||
|
||||
all_valid, errors = validate_all_branches(test_branches)
|
||||
console.print(f"\nAll branches valid: {all_valid}")
|
||||
console.print(f"Validation errors: {len(errors)}")
|
||||
|
||||
console.print("\n" + "="*70 + "\n")
|
||||
@@ -21,6 +21,6 @@ Contains handlers that respond to Trigger events.
|
||||
These handlers are registered by Trigger's event registry.
|
||||
"""
|
||||
|
||||
from .error_handler import handle_error_detected
|
||||
|
||||
__all__ = ['handle_error_detected']
|
||||
# error_handler.py moved to apps/.archive/trigger_error_handler(disabled).py
|
||||
# Trigger branch owns event handling — ai_mail provides delivery only.
|
||||
__all__ = []
|
||||
|
||||
@@ -1,172 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: error_handler.py
|
||||
# Description: Error Detected Event Consumer
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-02-02
|
||||
# Modified: 2026-02-02
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Error Detected Event Consumer
|
||||
|
||||
Handles error_detected events fired by Trigger's log_watcher.
|
||||
Delivers notifications to affected branches via AI_MAIL.
|
||||
|
||||
Event data from log_watcher.py:
|
||||
- branch: Target branch name (e.g., 'FLOW')
|
||||
- module: Module that logged the error
|
||||
- message: Error message text
|
||||
- log_path: Path to log file
|
||||
- error_hash: 8-char hash for deduplication
|
||||
- timestamp: When error occurred
|
||||
|
||||
Architecture:
|
||||
1. Trigger's log_watcher detects ERROR in branch logs
|
||||
2. log_watcher fires error_detected event
|
||||
3. This handler receives event, calls deliver_email_to_branch()
|
||||
4. Email delivered to affected branch inbox (auto_execute=True)
|
||||
5. Branch agent spawns and investigates
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
|
||||
|
||||
def _build_notification_message(
|
||||
error_hash: str,
|
||||
module: str,
|
||||
message: str,
|
||||
timestamp: str,
|
||||
log_path: str
|
||||
) -> str:
|
||||
"""
|
||||
Build error notification message with investigation instructions.
|
||||
|
||||
Args:
|
||||
error_hash: Unique error identifier (8-char)
|
||||
module: Module that logged the error
|
||||
message: Error message text
|
||||
timestamp: When error occurred
|
||||
log_path: Path to source log file
|
||||
|
||||
Returns:
|
||||
Formatted message string with investigation instructions
|
||||
"""
|
||||
return f"""Error detected - investigate and respond.
|
||||
|
||||
Error ID: {error_hash}
|
||||
Module: {module}
|
||||
Timestamp: {timestamp}
|
||||
Log file: {log_path}
|
||||
|
||||
Error message:
|
||||
{message}
|
||||
|
||||
---
|
||||
Investigation steps:
|
||||
1. Check the log file for context around this error
|
||||
2. Identify root cause
|
||||
|
||||
Decision tree:
|
||||
- Simple fix (typo, missing import, config issue):
|
||||
-> Fix it yourself, then report what you did to @devpulse
|
||||
- Complex/unclear (needs research, affects multiple files):
|
||||
-> Report findings only to @devpulse, recommend action, don't fix
|
||||
- Critical (data loss risk, security, system stability):
|
||||
-> Stop immediately, escalate to @devpulse with full context
|
||||
|
||||
Report to @devpulse:
|
||||
ai_mail send @devpulse "ERROR {error_hash} - [STATUS]" "Findings..."
|
||||
|
||||
Include: Error ID, severity (low/medium/high/critical), what you found, action taken or recommended.
|
||||
"""
|
||||
|
||||
|
||||
def handle_error_detected(
|
||||
branch: str | None = None,
|
||||
module: str | None = None,
|
||||
message: str | None = None,
|
||||
log_path: str | None = None,
|
||||
error_hash: str | None = None,
|
||||
timestamp: str | None = None,
|
||||
**kwargs: Any
|
||||
) -> None:
|
||||
"""
|
||||
Handle error_detected event - deliver notification to affected branch.
|
||||
|
||||
Called by Trigger when log_watcher detects an ERROR in branch logs.
|
||||
Sends email to affected branch with auto_execute=True so an
|
||||
investigation agent spawns automatically.
|
||||
|
||||
Args:
|
||||
branch: Target branch name (e.g., 'FLOW') - REQUIRED
|
||||
module: Module that logged the error - REQUIRED
|
||||
message: Error message text - REQUIRED
|
||||
log_path: Path to source log file
|
||||
error_hash: 8-char unique error identifier - REQUIRED
|
||||
timestamp: When error occurred (defaults to now)
|
||||
**kwargs: Additional event data (ignored)
|
||||
|
||||
Returns:
|
||||
None - handlers must not return values
|
||||
|
||||
Note:
|
||||
Handler follows silent failure pattern - all exceptions caught.
|
||||
NO logger imports (causes infinite recursion with trigger events).
|
||||
NO console.print() (handlers must be silent).
|
||||
"""
|
||||
try:
|
||||
# Validate required fields
|
||||
if not branch or not module or not message or not error_hash:
|
||||
return
|
||||
|
||||
# Import delivery handler here to avoid import-time failures
|
||||
try:
|
||||
from aipass.ai_mail.apps.handlers.email.delivery import deliver_email_to_branch
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
# Default timestamp to now if not provided
|
||||
if not timestamp:
|
||||
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
# Default log_path if not provided
|
||||
if not log_path:
|
||||
log_path = "unknown"
|
||||
|
||||
# Convert branch name to email format (FLOW -> @flow)
|
||||
branch_email = f"@{branch.lower()}"
|
||||
|
||||
# Build subject line
|
||||
subject = f"[ERROR] {module} - detected in logs"
|
||||
|
||||
# Build notification message
|
||||
notification_message = _build_notification_message(
|
||||
error_hash=error_hash,
|
||||
module=module,
|
||||
message=message,
|
||||
timestamp=timestamp,
|
||||
log_path=log_path
|
||||
)
|
||||
|
||||
# Build email data for delivery
|
||||
email_data = {
|
||||
'from': '@error_monitor',
|
||||
'from_name': 'Error Monitor',
|
||||
'to': branch_email,
|
||||
'subject': subject,
|
||||
'message': notification_message,
|
||||
'timestamp': timestamp,
|
||||
'auto_execute': True,
|
||||
'priority': 'normal',
|
||||
'reply_to': '@devpulse'
|
||||
}
|
||||
|
||||
# Deliver via inbox.json
|
||||
deliver_email_to_branch(branch_email, email_data)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning("[error_handler] Failed to deliver error notification to %s: %s", branch_email, e)
|
||||
@@ -1,15 +1,11 @@
|
||||
"""
|
||||
User Handlers - User Configuration and Management
|
||||
|
||||
Handles loading of user configuration files and user information for AI_Mail system.
|
||||
Handles user information retrieval for AI_Mail system.
|
||||
"""
|
||||
|
||||
from .load import (
|
||||
load_user_config,
|
||||
load_config,
|
||||
create_default_config,
|
||||
load_or_create_config
|
||||
)
|
||||
# load.py moved to apps/.archive/users_load(disabled).py
|
||||
# config_generator.py moved to apps/.archive/config_generator(disabled).py
|
||||
|
||||
from .user import (
|
||||
get_current_user,
|
||||
@@ -18,13 +14,6 @@ from .user import (
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Config loading
|
||||
'load_user_config',
|
||||
'load_config',
|
||||
'create_default_config',
|
||||
'load_or_create_config',
|
||||
|
||||
# User info
|
||||
'get_current_user',
|
||||
'get_user_by_email',
|
||||
'get_all_users',
|
||||
|
||||
@@ -1,260 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: config_generator.py
|
||||
# Description: Local Config Auto-Generation Handler
|
||||
# Version: 1.0.0
|
||||
# Created: 2025-11-18
|
||||
# Modified: 2025-11-18
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Local Config Auto-Generation Handler
|
||||
|
||||
Auto-generates user_config.json files for branches that use AI_MAIL.
|
||||
Creates ai_mail_config/ directory and populates with branch-specific config.
|
||||
"""
|
||||
|
||||
# =============================================
|
||||
# IMPORTS
|
||||
# =============================================
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
# =============================================
|
||||
# CONFIG GENERATION FUNCTIONS
|
||||
# =============================================
|
||||
|
||||
def generate_local_config(branch_info: Dict) -> Dict:
|
||||
"""
|
||||
Generate local user_config.json content for a branch.
|
||||
|
||||
Args:
|
||||
branch_info: Branch info dict from registry with keys:
|
||||
- name: Branch name (e.g., "SEED")
|
||||
- path: Branch path
|
||||
- email: Branch email address (e.g., "@seed")
|
||||
- description: Branch description
|
||||
|
||||
Returns:
|
||||
Dict with user_config.json structure
|
||||
"""
|
||||
branch_name = branch_info.get("name", "").lower()
|
||||
branch_email = branch_info.get("email", f"@{branch_name}")
|
||||
branch_path = Path(branch_info.get("path", ""))
|
||||
|
||||
# Generate display name from branch info
|
||||
display_name = generate_display_name(branch_info)
|
||||
|
||||
# Generate mailbox path (.ai_mail.local/ in branch directory)
|
||||
mailbox_path = str(branch_path / ".ai_mail.local")
|
||||
|
||||
config = {
|
||||
"version": "1.0.0",
|
||||
"current_user": branch_name,
|
||||
"users": {
|
||||
branch_name: {
|
||||
"name": branch_info.get("name", "").title(),
|
||||
"email_address": branch_email,
|
||||
"display_name": display_name,
|
||||
"mailbox_path": mailbox_path
|
||||
}
|
||||
},
|
||||
"settings": {
|
||||
"timestamp_format": "%Y-%m-%d %H:%M:%S",
|
||||
"max_inbox_display": 20,
|
||||
"max_sent_display": 20
|
||||
}
|
||||
}
|
||||
|
||||
return config
|
||||
|
||||
|
||||
def generate_display_name(branch_info: Dict) -> str:
|
||||
"""
|
||||
Generate user-friendly display name for branch.
|
||||
|
||||
Args:
|
||||
branch_info: Branch info dict from registry
|
||||
|
||||
Returns:
|
||||
Display name string (e.g., "Seed (Standards Branch)")
|
||||
"""
|
||||
name = branch_info.get("name", "Unknown").title()
|
||||
description = branch_info.get("description", "")
|
||||
|
||||
# If description is meaningful (not default), use it
|
||||
if description and description != "New branch - purpose TBD":
|
||||
return f"{name} ({description})"
|
||||
else:
|
||||
# Use profile as context if available
|
||||
profile = branch_info.get("profile", "")
|
||||
if profile and profile != "AIPass Workshop":
|
||||
return f"{name} ({profile})"
|
||||
else:
|
||||
# Just branch name
|
||||
return name
|
||||
|
||||
|
||||
def create_local_config_file(branch_info: Dict, force: bool = False) -> Path | None:
|
||||
"""
|
||||
Create local user_config.json file for a branch.
|
||||
|
||||
Saves to branch's [branch_name]_json/ directory (e.g., seed_json/user_config.json).
|
||||
Follows the pattern: all JSON files for a branch go in their [branch]_json/ folder.
|
||||
|
||||
Args:
|
||||
branch_info: Branch info dict from registry
|
||||
force: If True, overwrite existing config file
|
||||
|
||||
Returns:
|
||||
Path to created config file, or None if failed
|
||||
"""
|
||||
try:
|
||||
branch_path = Path(branch_info.get("path", ""))
|
||||
if not branch_path.exists():
|
||||
return None
|
||||
|
||||
# Get branch name for directory pattern
|
||||
branch_name = branch_info.get("name", "").lower()
|
||||
|
||||
# Config directory: [branch_name]_json/
|
||||
config_dir = branch_path / f"{branch_name}_json"
|
||||
|
||||
# Create directory if it doesn't exist
|
||||
if not config_dir.exists():
|
||||
config_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Config file path
|
||||
config_file = config_dir / "user_config.json"
|
||||
|
||||
# Check if already exists
|
||||
if config_file.exists() and not force:
|
||||
return config_file
|
||||
|
||||
# Generate config content
|
||||
config = generate_local_config(branch_info)
|
||||
|
||||
# Write config file
|
||||
with open(config_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(config, f, indent=2)
|
||||
|
||||
return config_file
|
||||
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def create_mailbox_directory(branch_path: Path) -> Path | None:
|
||||
"""
|
||||
Create .ai_mail.local/ mailbox directory for a branch.
|
||||
|
||||
Creates subdirectories: inbox/, sent/, deleted/
|
||||
|
||||
Args:
|
||||
branch_path: Path to branch directory
|
||||
|
||||
Returns:
|
||||
Path to mailbox directory, or None if failed
|
||||
"""
|
||||
try:
|
||||
mailbox_dir = branch_path / ".ai_mail.local"
|
||||
mailbox_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create subdirectories
|
||||
(mailbox_dir / "sent").mkdir(exist_ok=True)
|
||||
|
||||
# Create empty inbox.json if doesn't exist
|
||||
inbox_file = mailbox_dir / "inbox.json"
|
||||
if not inbox_file.exists():
|
||||
inbox_data = {
|
||||
"mailbox": "inbox",
|
||||
"total_messages": 0,
|
||||
"unread_count": 0,
|
||||
"messages": []
|
||||
}
|
||||
with open(inbox_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(inbox_data, f, indent=2)
|
||||
|
||||
# Create empty sent.json if doesn't exist
|
||||
sent_file = mailbox_dir / "sent.json"
|
||||
if not sent_file.exists():
|
||||
sent_data = {
|
||||
"mailbox": "sent",
|
||||
"total_messages": 0,
|
||||
"messages": []
|
||||
}
|
||||
with open(sent_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(sent_data, f, indent=2)
|
||||
|
||||
return mailbox_dir
|
||||
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def setup_branch_for_aimail(branch_info: Dict, force: bool = False) -> bool:
|
||||
"""
|
||||
Complete AI_MAIL setup for a branch.
|
||||
|
||||
Creates:
|
||||
- ai_mail_config/user_config.json
|
||||
- .ai_mail.local/ mailbox directory
|
||||
- .ai_mail.local/inbox.json
|
||||
- .ai_mail.local/sent.json
|
||||
|
||||
Args:
|
||||
branch_info: Branch info dict from registry
|
||||
force: If True, overwrite existing files
|
||||
|
||||
Returns:
|
||||
True if setup successful, False otherwise
|
||||
"""
|
||||
try:
|
||||
# Create config file
|
||||
config_file = create_local_config_file(branch_info, force=force)
|
||||
if not config_file:
|
||||
return False
|
||||
|
||||
# Create mailbox directory
|
||||
branch_path = Path(branch_info.get("path", ""))
|
||||
mailbox_dir = create_mailbox_directory(branch_path)
|
||||
if not mailbox_dir:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
from aipass.cli.apps.modules import console
|
||||
|
||||
console.print("\n" + "="*70)
|
||||
console.print("LOCAL CONFIG AUTO-GENERATION HANDLER")
|
||||
console.print("="*70)
|
||||
console.print("\nPURPOSE:")
|
||||
console.print(" Auto-generates user_config.json files for branches using AI_MAIL")
|
||||
console.print(" Creates ai_mail_config/ directory and mailbox structure")
|
||||
console.print()
|
||||
console.print("FUNCTIONS PROVIDED:")
|
||||
console.print(" - generate_local_config(branch_info) -> Dict")
|
||||
console.print(" - generate_display_name(branch_info) -> str")
|
||||
console.print(" - create_local_config_file(branch_info, force) -> Path | None")
|
||||
console.print(" - create_mailbox_directory(branch_path) -> Path | None")
|
||||
console.print(" - setup_branch_for_aimail(branch_info, force) -> bool")
|
||||
console.print()
|
||||
console.print("HANDLER CHARACTERISTICS:")
|
||||
console.print(" ✓ Independent - no module dependencies")
|
||||
console.print(" ✓ Can import Prax (service provider)")
|
||||
console.print(" ✓ Pure business logic")
|
||||
console.print(" ✗ CANNOT import parent modules")
|
||||
console.print()
|
||||
console.print("SETUP WORKFLOW:")
|
||||
console.print(" 1. Generate config from branch registry info")
|
||||
console.print(" 2. Create ai_mail_config/ directory")
|
||||
console.print(" 3. Write user_config.json with branch-specific settings")
|
||||
console.print(" 4. Create .ai_mail.local/ mailbox directory")
|
||||
console.print(" 5. Initialize inbox.json and sent.json")
|
||||
console.print()
|
||||
console.print("="*70 + "\n")
|
||||
@@ -1,144 +0,0 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: load.py
|
||||
# Description: User Config Loading Handler
|
||||
# Version: 1.2.0
|
||||
# Created: 2025-11-15
|
||||
# Modified: 2025-11-15
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
User Config Loading Handler
|
||||
|
||||
Handles loading of user configuration files for AI_Mail system.
|
||||
Provides unified config loading for user_config.json and module configs.
|
||||
|
||||
NEW in v1.2.0: Per-branch config support
|
||||
- Detects calling branch from PWD/CWD
|
||||
- Checks for local config at branch_path/ai_mail_config/user_config.json
|
||||
- Falls back to AI_MAIL's global config if no local config found
|
||||
"""
|
||||
|
||||
# =============================================
|
||||
# IMPORTS
|
||||
# =============================================
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
# =============================================
|
||||
# CONSTANTS
|
||||
# =============================================
|
||||
_AI_MAIL_ROOT = Path(__file__).resolve().parents[3] # ai_mail/
|
||||
AI_MAIL_JSON = _AI_MAIL_ROOT / ".ai_mail.local"
|
||||
USER_CONFIG_FILE = AI_MAIL_JSON / "user_config.json"
|
||||
|
||||
# Import branch detection (after constants defined)
|
||||
from .branch_detection import detect_branch_from_pwd, get_local_config_path
|
||||
|
||||
# =============================================
|
||||
# CONFIG LOADING FUNCTIONS
|
||||
# =============================================
|
||||
|
||||
def load_user_config() -> Dict:
|
||||
"""
|
||||
Load user configuration from user_config.json
|
||||
|
||||
NEW in v1.2.0: Per-branch config support
|
||||
- First checks if calling from a branch (detects via PWD)
|
||||
- Looks for local config at branch_path/ai_mail_config/user_config.json
|
||||
- Falls back to AI_MAIL's global config if no local config
|
||||
- Auto-generates local config if branch detected but config missing
|
||||
|
||||
Returns:
|
||||
Dict containing user configuration
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If no config found (neither local nor global)
|
||||
"""
|
||||
# Try to detect calling branch from PWD
|
||||
branch_info = detect_branch_from_pwd()
|
||||
|
||||
if branch_info:
|
||||
# Branch detected - check for local config
|
||||
branch_path = Path(branch_info["path"])
|
||||
branch_name = branch_info["name"]
|
||||
local_config_path = get_local_config_path(branch_path, branch_name)
|
||||
|
||||
if local_config_path.exists():
|
||||
# Local config exists - use it
|
||||
with open(local_config_path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
else:
|
||||
# No local config - offer to auto-generate
|
||||
# For now, fall through to global config
|
||||
pass
|
||||
|
||||
# No branch detected or no local config - use AI_MAIL's global config
|
||||
if not USER_CONFIG_FILE.exists():
|
||||
raise FileNotFoundError(f"User config not found: {USER_CONFIG_FILE}")
|
||||
|
||||
with open(USER_CONFIG_FILE, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def load_config(config_file: Path) -> Dict:
|
||||
"""
|
||||
Load generic configuration file with auto-healing
|
||||
|
||||
Creates default config if missing (for module configs).
|
||||
|
||||
Args:
|
||||
config_file: Path to configuration file
|
||||
|
||||
Returns:
|
||||
Dict containing configuration data
|
||||
|
||||
Raises:
|
||||
Exception: If config cannot be loaded or created
|
||||
"""
|
||||
if not config_file.exists():
|
||||
raise FileNotFoundError(f"Config file not found: {config_file}")
|
||||
|
||||
try:
|
||||
with open(config_file, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
except Exception as e:
|
||||
raise
|
||||
|
||||
|
||||
def create_default_config(config_file: Path, default_config: Dict) -> None:
|
||||
"""
|
||||
Create default configuration file
|
||||
|
||||
Args:
|
||||
config_file: Path where config should be created
|
||||
default_config: Default configuration dictionary
|
||||
|
||||
Raises:
|
||||
Exception: If config cannot be created
|
||||
"""
|
||||
# Ensure parent directory exists
|
||||
config_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
with open(config_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(default_config, f, indent=2)
|
||||
except Exception as e:
|
||||
raise
|
||||
|
||||
|
||||
def load_or_create_config(config_file: Path, default_config: Dict) -> Dict:
|
||||
"""
|
||||
Load config file, creating with defaults if missing
|
||||
|
||||
Args:
|
||||
config_file: Path to configuration file
|
||||
default_config: Default configuration to use if file doesn't exist
|
||||
|
||||
Returns:
|
||||
Dict containing configuration data
|
||||
"""
|
||||
if not config_file.exists():
|
||||
create_default_config(config_file, default_config)
|
||||
|
||||
return load_config(config_file)
|
||||
@@ -112,12 +112,16 @@ def get_user_by_email(email: str) -> Dict | None:
|
||||
|
||||
try:
|
||||
import json
|
||||
_repo_root = registry_path.parent
|
||||
|
||||
with open(registry_path, 'r', encoding='utf-8') as f:
|
||||
registry = json.load(f)
|
||||
|
||||
for branch in _get_branches_list(registry):
|
||||
if branch.get("email") == email:
|
||||
branch_path = Path(branch.get("path", ""))
|
||||
if branch_path and not branch_path.is_absolute():
|
||||
branch_path = (_repo_root / branch_path).resolve()
|
||||
return {
|
||||
"email_address": branch.get("email"),
|
||||
"display_name": branch.get("name"),
|
||||
@@ -147,11 +151,14 @@ def get_all_users() -> Dict[str, Dict]:
|
||||
with open(registry_path, 'r', encoding='utf-8') as f:
|
||||
registry = json.load(f)
|
||||
|
||||
_repo_root = registry_path.parent
|
||||
users = {}
|
||||
for branch in _get_branches_list(registry):
|
||||
email = branch.get("email", "")
|
||||
if email:
|
||||
branch_path = Path(branch.get("path", ""))
|
||||
if branch_path and not branch_path.is_absolute():
|
||||
branch_path = (_repo_root / branch_path).resolve()
|
||||
users[email] = {
|
||||
"email_address": email,
|
||||
"display_name": branch.get("name"),
|
||||
|
||||
@@ -0,0 +1,274 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: test_user_paths.py
|
||||
# Description: Tests for absolute mailbox_path resolution in user functions
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-03-17
|
||||
# Modified: 2026-03-17
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Tests for mailbox_path Absolute Resolution
|
||||
|
||||
Bug: get_user_by_email() and get_all_users() returned relative paths like
|
||||
"src/aipass/ai_mail/.ai_mail.local" instead of absolute paths.
|
||||
get_current_user() was already correct (resolved against _repo_root).
|
||||
|
||||
Fix: Both functions now resolve relative registry paths against _repo_root
|
||||
(the parent of BRANCH_REGISTRY_PATH), matching get_current_user()'s pattern.
|
||||
|
||||
These tests verify:
|
||||
1. get_user_by_email() returns an absolute mailbox_path
|
||||
2. get_all_users() returns absolute mailbox_path for every entry
|
||||
3. Paths are never doubled (no src/aipass/.../src/aipass/...)
|
||||
4. Absolute paths in the registry are preserved as-is
|
||||
"""
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from aipass.ai_mail.apps.handlers.users.user import get_user_by_email, get_all_users
|
||||
|
||||
|
||||
# ─── Fixtures ────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def relative_path_registry(tmp_path):
|
||||
"""Create a registry with relative paths (production format).
|
||||
|
||||
Production AIPASS_REGISTRY.json uses list format with relative paths:
|
||||
"path": "src/aipass/ai_mail"
|
||||
|
||||
The registry sits at tmp_path/AIPASS_REGISTRY.json, so _repo_root
|
||||
is tmp_path. Resolved paths should be tmp_path / "src/aipass/..." .
|
||||
|
||||
Returns (registry_path, expected_repo_root).
|
||||
"""
|
||||
registry = {"branches": [
|
||||
{
|
||||
"name": "AI_MAIL",
|
||||
"path": "src/aipass/ai_mail",
|
||||
"email": "@ai_mail",
|
||||
"status": "active",
|
||||
"description": "Agent-to-agent messaging system",
|
||||
},
|
||||
{
|
||||
"name": "SPAWN",
|
||||
"path": "src/aipass/spawn",
|
||||
"email": "@spawn",
|
||||
"status": "active",
|
||||
"description": "Branch spawner",
|
||||
},
|
||||
{
|
||||
"name": "TRIGGER",
|
||||
"path": "src/aipass/trigger",
|
||||
"email": "@trigger",
|
||||
"status": "active",
|
||||
"description": "Event trigger system",
|
||||
},
|
||||
]}
|
||||
registry_path = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry_path.write_text(json.dumps(registry, indent=2), encoding="utf-8")
|
||||
return registry_path, tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def absolute_path_registry(tmp_path):
|
||||
"""Create a registry where paths are already absolute.
|
||||
|
||||
Ensures absolute paths pass through without double-resolution.
|
||||
Returns (registry_path, branch_dir).
|
||||
"""
|
||||
branch_dir = tmp_path / "src" / "aipass" / "solo_branch"
|
||||
registry = {"branches": [
|
||||
{
|
||||
"name": "SOLO",
|
||||
"path": str(branch_dir),
|
||||
"email": "@solo",
|
||||
"status": "active",
|
||||
"description": "Branch with absolute path",
|
||||
},
|
||||
]}
|
||||
registry_path = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry_path.write_text(json.dumps(registry, indent=2), encoding="utf-8")
|
||||
return registry_path, branch_dir
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def dict_format_registry(tmp_path):
|
||||
"""Create a registry using dict format (legacy).
|
||||
|
||||
Tests that the dict->list normalization via _get_branches_list still
|
||||
produces absolute paths.
|
||||
"""
|
||||
registry = {"branches": {
|
||||
"devpulse": {
|
||||
"name": "DEVPULSE",
|
||||
"path": "src/aipass/devpulse",
|
||||
"email": "@devpulse",
|
||||
"status": "active",
|
||||
"description": "DevPulse branch",
|
||||
},
|
||||
"backup": {
|
||||
"name": "BACKUP",
|
||||
"path": "src/aipass/backup",
|
||||
"email": "@backup",
|
||||
"status": "active",
|
||||
"description": "Backup branch",
|
||||
},
|
||||
}}
|
||||
registry_path = tmp_path / "AIPASS_REGISTRY.json"
|
||||
registry_path.write_text(json.dumps(registry, indent=2), encoding="utf-8")
|
||||
return registry_path, tmp_path
|
||||
|
||||
|
||||
# ─── get_user_by_email() tests ───────────────────────────
|
||||
|
||||
|
||||
class TestGetUserByEmailPaths:
|
||||
"""Verify get_user_by_email() returns absolute mailbox_path values."""
|
||||
|
||||
def test_returns_absolute_mailbox_path(self, relative_path_registry):
|
||||
"""Relative registry paths must be resolved to absolute mailbox_path."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@ai_mail")
|
||||
assert result is not None
|
||||
mailbox = Path(result["mailbox_path"])
|
||||
assert mailbox.is_absolute(), (
|
||||
f"mailbox_path must be absolute, got: {result['mailbox_path']}"
|
||||
)
|
||||
|
||||
def test_path_rooted_at_repo_root(self, relative_path_registry):
|
||||
"""Resolved path should start from the repo root (registry parent)."""
|
||||
registry_path, repo_root = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@spawn")
|
||||
assert result is not None
|
||||
expected = str((repo_root / "src" / "aipass" / "spawn" / ".ai_mail.local").resolve())
|
||||
assert result["mailbox_path"] == expected
|
||||
|
||||
def test_no_doubled_relative_path(self, relative_path_registry):
|
||||
"""Path must not contain the relative prefix twice (the old bug)."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@trigger")
|
||||
assert result is not None
|
||||
path = result["mailbox_path"]
|
||||
# Count occurrences of the relative segment
|
||||
assert path.count("src/aipass/trigger") == 1, (
|
||||
f"Path contains doubled segment: {path}"
|
||||
)
|
||||
|
||||
def test_absolute_path_preserved(self, absolute_path_registry):
|
||||
"""Registry entries with absolute paths should not be re-rooted."""
|
||||
registry_path, branch_dir = absolute_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@solo")
|
||||
assert result is not None
|
||||
expected = str((branch_dir / ".ai_mail.local").resolve())
|
||||
assert result["mailbox_path"] == expected
|
||||
|
||||
def test_returns_none_for_unknown_email(self, relative_path_registry):
|
||||
"""Unknown email should return None, not crash."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@nonexistent_branch_xyz")
|
||||
assert result is None
|
||||
|
||||
def test_dict_format_returns_absolute_path(self, dict_format_registry):
|
||||
"""Dict-format registry should also produce absolute paths."""
|
||||
registry_path, repo_root = dict_format_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
result = get_user_by_email("@devpulse")
|
||||
assert result is not None
|
||||
mailbox = Path(result["mailbox_path"])
|
||||
assert mailbox.is_absolute(), (
|
||||
f"mailbox_path must be absolute (dict format), got: {result['mailbox_path']}"
|
||||
)
|
||||
expected = str((repo_root / "src" / "aipass" / "devpulse" / ".ai_mail.local").resolve())
|
||||
assert result["mailbox_path"] == expected
|
||||
|
||||
|
||||
# ─── get_all_users() tests ───────────────────────────────
|
||||
|
||||
|
||||
class TestGetAllUsersPaths:
|
||||
"""Verify get_all_users() returns absolute mailbox_path for every entry."""
|
||||
|
||||
def test_all_paths_are_absolute(self, relative_path_registry):
|
||||
"""Every user returned must have an absolute mailbox_path."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
assert len(users) == 3, f"Expected 3 users, got {len(users)}"
|
||||
for email, info in users.items():
|
||||
mailbox = Path(info["mailbox_path"])
|
||||
assert mailbox.is_absolute(), (
|
||||
f"mailbox_path for {email} must be absolute, got: {info['mailbox_path']}"
|
||||
)
|
||||
|
||||
def test_all_paths_end_with_ai_mail_local(self, relative_path_registry):
|
||||
"""Every mailbox_path should end with .ai_mail.local."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
for email, info in users.items():
|
||||
assert info["mailbox_path"].endswith(".ai_mail.local"), (
|
||||
f"mailbox_path for {email} should end with .ai_mail.local, "
|
||||
f"got: {info['mailbox_path']}"
|
||||
)
|
||||
|
||||
def test_no_doubled_paths_in_any_entry(self, relative_path_registry):
|
||||
"""No entry should have a doubled relative segment."""
|
||||
registry_path, _ = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
for email, info in users.items():
|
||||
path = info["mailbox_path"]
|
||||
# The relative prefix "src/aipass" should appear exactly once
|
||||
assert path.count("src/aipass") == 1, (
|
||||
f"Path for {email} contains doubled 'src/aipass': {path}"
|
||||
)
|
||||
|
||||
def test_paths_resolve_against_repo_root(self, relative_path_registry):
|
||||
"""Resolved paths should be rooted at the registry's parent dir."""
|
||||
registry_path, repo_root = relative_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
for email, info in users.items():
|
||||
assert info["mailbox_path"].startswith(str(repo_root)), (
|
||||
f"Path for {email} should start with repo root {repo_root}, "
|
||||
f"got: {info['mailbox_path']}"
|
||||
)
|
||||
|
||||
def test_absolute_paths_preserved(self, absolute_path_registry):
|
||||
"""Entries with absolute paths should pass through unchanged."""
|
||||
registry_path, branch_dir = absolute_path_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
assert "@solo" in users
|
||||
expected = str((branch_dir / ".ai_mail.local").resolve())
|
||||
assert users["@solo"]["mailbox_path"] == expected
|
||||
|
||||
def test_dict_format_all_absolute(self, dict_format_registry):
|
||||
"""Dict-format registry should produce absolute paths for all entries."""
|
||||
registry_path, _ = dict_format_registry
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", registry_path):
|
||||
users = get_all_users()
|
||||
assert len(users) == 2
|
||||
for email, info in users.items():
|
||||
mailbox = Path(info["mailbox_path"])
|
||||
assert mailbox.is_absolute(), (
|
||||
f"mailbox_path for {email} must be absolute (dict format), "
|
||||
f"got: {info['mailbox_path']}"
|
||||
)
|
||||
|
||||
def test_empty_registry_returns_empty_dict(self, tmp_path):
|
||||
"""Missing registry file should return empty dict, not crash."""
|
||||
fake_path = tmp_path / "nonexistent_registry.json"
|
||||
with patch("aipass.ai_mail.apps.handlers.users.branch_detection.BRANCH_REGISTRY_PATH", fake_path):
|
||||
users = get_all_users()
|
||||
assert users == {}
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
**Purpose:** Display and output formatting service for AIPass modules. Provides consistent terminal output — headers, success/error/warning messages, section breaks, and operation templates — so every module looks the same without duplicating Rich formatting code.
|
||||
**Module:** `aipass.cli`
|
||||
**Last Updated:** 2026-03-08
|
||||
**Last Updated:** 2026-03-17
|
||||
|
||||
## Usage
|
||||
|
||||
@@ -42,9 +42,16 @@ cli/
|
||||
│ ├── cli.py # Entry point
|
||||
│ ├── modules/
|
||||
│ │ ├── display.py # header, success, error, warning, section
|
||||
│ │ └── templates.py # operation_start, operation_complete
|
||||
│ │ ├── templates.py # operation_start, operation_complete
|
||||
│ │ └── init_project.py # aipass init command routing
|
||||
│ └── handlers/
|
||||
│ ├── init/ # Project bootstrap logic
|
||||
│ │ └── bootstrap.py
|
||||
│ └── json/ # JSON file management
|
||||
│ └── json_handler.py
|
||||
├── cli_json/ # Auto-created JSON output (three-file pattern)
|
||||
├── dropbox/ # Inbound file drop
|
||||
├── logs/ # Branch-level logs
|
||||
└── tests/
|
||||
```
|
||||
|
||||
@@ -54,9 +61,14 @@ cli/
|
||||
## Commands / Usage
|
||||
|
||||
```bash
|
||||
drone @cli --help # Show services and Rich formatting showcase
|
||||
drone @cli --version # Show version
|
||||
drone @cli help # Same as --help
|
||||
drone @cli --help # Show services and Rich formatting showcase
|
||||
drone @cli --version # Show version
|
||||
drone @cli help # Same as --help
|
||||
drone @cli aipass # Show aipass subcommands
|
||||
drone @cli aipass init # Bootstrap AIPass project in current dir
|
||||
drone @cli aipass init /path # Bootstrap in target directory
|
||||
drone @cli aipass init /path MyProject # Bootstrap with custom name
|
||||
drone @cli aipass init --help # Detailed init usage
|
||||
```
|
||||
|
||||
---
|
||||
@@ -72,7 +84,8 @@ drone @cli help # Same as --help
|
||||
- All modules — display formatting (headers, success/error/warning, section breaks)
|
||||
- All modules — operation templates (`operation_start`, `operation_complete`)
|
||||
- All modules — Rich console access
|
||||
- All users — `aipass init` project bootstrap command
|
||||
|
||||
---
|
||||
|
||||
*Last Updated: 2026-03-08*
|
||||
*Last Updated: 2026-03-17*
|
||||
|
||||
@@ -13,22 +13,12 @@ from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any, Optional
|
||||
import inspect
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
|
||||
def _find_repo_root() -> Path:
|
||||
"""Walk up from this file to find the repo root (contains pyproject.toml or AIPASS_REGISTRY.json)."""
|
||||
current = Path(__file__).resolve().parent
|
||||
for parent in [current] + list(current.parents):
|
||||
if (parent / "pyproject.toml").exists() or (parent / "AIPASS_REGISTRY.json").exists():
|
||||
return parent
|
||||
return Path.cwd()
|
||||
|
||||
|
||||
# Constants — resolved via repo root walk-up (portable across any machine)
|
||||
_REPO_ROOT = _find_repo_root()
|
||||
CLI_ROOT = _REPO_ROOT / "src" / "aipass" / "cli"
|
||||
CLI_JSON_DIR = CLI_ROOT / "cli_json"
|
||||
JSON_TEMPLATES_DIR = CLI_ROOT / "apps" / "json_templates"
|
||||
# Constants — resolved via __file__ (portable across any machine)
|
||||
_BRANCH_ROOT = Path(__file__).resolve().parents[3] # json/ -> handlers/ -> apps/ -> cli/
|
||||
_BRANCH_NAME = _BRANCH_ROOT.name
|
||||
JSON_DIR = _BRANCH_ROOT / f"{_BRANCH_NAME}_json"
|
||||
JSON_TEMPLATES_DIR = _BRANCH_ROOT / "apps" / "json_templates"
|
||||
|
||||
|
||||
def _get_caller_module_name() -> str:
|
||||
@@ -94,12 +84,12 @@ def validate_json_structure(data: Any, json_type: str) -> bool:
|
||||
def get_json_path(module_name: str, json_type: str) -> Path:
|
||||
"""Get path for module JSON file"""
|
||||
filename = f"{module_name}_{json_type}.json"
|
||||
return CLI_JSON_DIR / filename
|
||||
return JSON_DIR / filename
|
||||
|
||||
|
||||
def ensure_json_exists(module_name: str, json_type: str) -> bool:
|
||||
"""Ensure JSON file exists, create from template if missing"""
|
||||
CLI_JSON_DIR.mkdir(parents=True, exist_ok=True)
|
||||
JSON_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
json_path = get_json_path(module_name, json_type)
|
||||
|
||||
@@ -111,8 +101,9 @@ def ensure_json_exists(module_name: str, json_type: str) -> bool:
|
||||
if validate_json_structure(data, json_type):
|
||||
return True
|
||||
# If corrupted, fall through to regenerate
|
||||
except Exception as exc:
|
||||
logger.warning("JSON unreadable at %s, regenerating: %s", json_path, exc)
|
||||
except Exception:
|
||||
# If unreadable, fall through to regenerate
|
||||
pass
|
||||
|
||||
template = load_template(json_type, module_name)
|
||||
|
||||
@@ -257,7 +248,7 @@ if __name__ == "__main__":
|
||||
update_data_metrics("cli", test_metric="working")
|
||||
|
||||
console.print()
|
||||
console.print(f"[green]Check {CLI_JSON_DIR}/ for created files:[/green]")
|
||||
console.print(f"[green]Check {JSON_DIR}/ for created files:[/green]")
|
||||
console.print(" [dim]•[/dim] cli_config.json")
|
||||
console.print(" [dim]•[/dim] cli_data.json")
|
||||
console.print(" [dim]•[/dim] cli_log.json")
|
||||
|
||||
@@ -29,6 +29,8 @@ from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich.columns import Columns
|
||||
|
||||
from aipass.cli.apps.handlers.json import json_handler
|
||||
|
||||
# NOTE: Cannot import prax here — circular import (prax depends on cli)
|
||||
# from aipass.prax import logger
|
||||
|
||||
@@ -226,11 +228,15 @@ def handle_command(command: str, args: List[str]) -> bool:
|
||||
if command == "demo":
|
||||
run_demo()
|
||||
return True
|
||||
elif command in ["display", "show"]:
|
||||
if command not in ("display", "show"):
|
||||
return False
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
if args[0] == "demo":
|
||||
run_demo()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def run_demo():
|
||||
@@ -262,6 +268,9 @@ def run_demo():
|
||||
|
||||
CONSOLE.print("[bold green]✨ Rich library integration complete![/bold green]")
|
||||
CONSOLE.print("[dim]All display functions now use Rich for beautiful terminal output[/dim]")
|
||||
|
||||
json_handler.log_operation("display_demo")
|
||||
|
||||
CONSOLE.print()
|
||||
|
||||
|
||||
|
||||
@@ -22,6 +22,7 @@ from typing import List
|
||||
|
||||
from aipass.cli.apps.handlers.init.bootstrap import init_project
|
||||
from aipass.cli.apps.modules.display import console, success, error, header
|
||||
from aipass.cli.apps.handlers.json import json_handler
|
||||
|
||||
|
||||
# =============================================================================
|
||||
@@ -217,6 +218,13 @@ def _handle_init(args: List[str]) -> bool:
|
||||
console.print()
|
||||
|
||||
success(f"Created {len(result['created_files'])} files")
|
||||
|
||||
json_handler.log_operation("aipass_init", {
|
||||
"project_name": result["project_name"],
|
||||
"target": result["target"],
|
||||
"files_created": len(result["created_files"]),
|
||||
})
|
||||
|
||||
console.print()
|
||||
|
||||
return True
|
||||
|
||||
@@ -18,10 +18,11 @@ Uses Rich library for beautiful terminal output.
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, Any, Optional, List
|
||||
from typing import List
|
||||
|
||||
# Import console from CLI display module (using our own service!)
|
||||
from aipass.cli.apps.modules.display import console as CONSOLE
|
||||
from aipass.cli.apps.handlers.json import json_handler
|
||||
# NOTE: Cannot import prax here — circular import (prax depends on cli)
|
||||
# from aipass.prax import logger
|
||||
|
||||
@@ -116,21 +117,17 @@ def print_help():
|
||||
|
||||
def handle_command(command: str, args: List[str]) -> bool:
|
||||
"""Handle 'templates' and 'demo' commands"""
|
||||
# Handle demo as direct command
|
||||
if command == "demo":
|
||||
run_demo()
|
||||
return True
|
||||
|
||||
# Handle templates command
|
||||
if command == "templates":
|
||||
# Check for subcommand in args
|
||||
if args and args[0] == "demo":
|
||||
run_demo()
|
||||
else:
|
||||
# templates just shows introspection
|
||||
print_introspection()
|
||||
if command != "templates":
|
||||
return False
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
if args[0] == "demo":
|
||||
run_demo()
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@@ -161,6 +158,9 @@ def run_demo():
|
||||
|
||||
CONSOLE.print("[bold green]✨ Rich library integration complete![/bold green]")
|
||||
CONSOLE.print("[dim]Templates provide consistent operation patterns across all branches[/dim]")
|
||||
|
||||
json_handler.log_operation("templates_demo")
|
||||
|
||||
CONSOLE.print()
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user