diff --git a/src/aipass/ai_mail/apps/handlers/dispatch/pending_work.py b/src/aipass/ai_mail/apps/handlers/dispatch/pending_work.py deleted file mode 100644 index 622a7429..00000000 --- a/src/aipass/ai_mail/apps/handlers/dispatch/pending_work.py +++ /dev/null @@ -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 diff --git a/src/aipass/ai_mail/apps/handlers/email/lock_utils.py b/src/aipass/ai_mail/apps/handlers/email/lock_utils.py deleted file mode 100644 index e694bfef..00000000 --- a/src/aipass/ai_mail/apps/handlers/email/lock_utils.py +++ /dev/null @@ -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 diff --git a/src/aipass/ai_mail/apps/handlers/monitoring/data_ops.py b/src/aipass/ai_mail/apps/handlers/monitoring/data_ops.py deleted file mode 100644 index 751bb6a2..00000000 --- a/src/aipass/ai_mail/apps/handlers/monitoring/data_ops.py +++ /dev/null @@ -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) diff --git a/src/aipass/ai_mail/apps/handlers/registry/validate.py b/src/aipass/ai_mail/apps/handlers/registry/validate.py deleted file mode 100644 index cc8d6c9d..00000000 --- a/src/aipass/ai_mail/apps/handlers/registry/validate.py +++ /dev/null @@ -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") diff --git a/src/aipass/ai_mail/apps/handlers/trigger/__init__.py b/src/aipass/ai_mail/apps/handlers/trigger/__init__.py index 8f5db295..c86c3795 100644 --- a/src/aipass/ai_mail/apps/handlers/trigger/__init__.py +++ b/src/aipass/ai_mail/apps/handlers/trigger/__init__.py @@ -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__ = [] diff --git a/src/aipass/ai_mail/apps/handlers/trigger/error_handler.py b/src/aipass/ai_mail/apps/handlers/trigger/error_handler.py deleted file mode 100644 index 757d22e4..00000000 --- a/src/aipass/ai_mail/apps/handlers/trigger/error_handler.py +++ /dev/null @@ -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) diff --git a/src/aipass/ai_mail/apps/handlers/users/__init__.py b/src/aipass/ai_mail/apps/handlers/users/__init__.py index 2aa0717d..f97d2da2 100644 --- a/src/aipass/ai_mail/apps/handlers/users/__init__.py +++ b/src/aipass/ai_mail/apps/handlers/users/__init__.py @@ -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', diff --git a/src/aipass/ai_mail/apps/handlers/users/config_generator.py b/src/aipass/ai_mail/apps/handlers/users/config_generator.py deleted file mode 100644 index 91040e79..00000000 --- a/src/aipass/ai_mail/apps/handlers/users/config_generator.py +++ /dev/null @@ -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") diff --git a/src/aipass/ai_mail/apps/handlers/users/load.py b/src/aipass/ai_mail/apps/handlers/users/load.py deleted file mode 100644 index c22644c4..00000000 --- a/src/aipass/ai_mail/apps/handlers/users/load.py +++ /dev/null @@ -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) diff --git a/src/aipass/ai_mail/apps/handlers/users/user.py b/src/aipass/ai_mail/apps/handlers/users/user.py index bc66f500..522089d8 100644 --- a/src/aipass/ai_mail/apps/handlers/users/user.py +++ b/src/aipass/ai_mail/apps/handlers/users/user.py @@ -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"), diff --git a/src/aipass/ai_mail/tests/test_user_paths.py b/src/aipass/ai_mail/tests/test_user_paths.py new file mode 100644 index 00000000..2aa31343 --- /dev/null +++ b/src/aipass/ai_mail/tests/test_user_paths.py @@ -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 == {} diff --git a/src/aipass/cli/README.md b/src/aipass/cli/README.md index aea25a6c..d878ca5b 100644 --- a/src/aipass/cli/README.md +++ b/src/aipass/cli/README.md @@ -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* diff --git a/src/aipass/cli/apps/handlers/json/json_handler.py b/src/aipass/cli/apps/handlers/json/json_handler.py index aba1ef16..2b427c2b 100755 --- a/src/aipass/cli/apps/handlers/json/json_handler.py +++ b/src/aipass/cli/apps/handlers/json/json_handler.py @@ -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") diff --git a/src/aipass/cli/apps/modules/display.py b/src/aipass/cli/apps/modules/display.py index f93f78ce..1658b595 100755 --- a/src/aipass/cli/apps/modules/display.py +++ b/src/aipass/cli/apps/modules/display.py @@ -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() diff --git a/src/aipass/cli/apps/modules/init_project.py b/src/aipass/cli/apps/modules/init_project.py index 111449aa..8d24c9af 100644 --- a/src/aipass/cli/apps/modules/init_project.py +++ b/src/aipass/cli/apps/modules/init_project.py @@ -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 diff --git a/src/aipass/cli/apps/modules/templates.py b/src/aipass/cli/apps/modules/templates.py index 943881e6..baee3523 100755 --- a/src/aipass/cli/apps/modules/templates.py +++ b/src/aipass/cli/apps/modules/templates.py @@ -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()