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

Merge pull request #59 from AIOSAI/refactor/prax-simplify-logging-two-tier

Browse Source 
refactor(prax): simplify logging to two-tier model
This commit is contained in:
AIPass
2026-03-17 00:48:43 -07:00
committed by GitHub
parent 617f96cb66 e19f4b7ff1
commit efd462e03d
27 changed files with 151 additions and 649 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/prax/README.md
+16 -8
View File
@@ -2,7 +2,7 @@
**Purpose:** System-wide logging, real-time monitoring, and dashboard for AIPass.
**Module:** `aipass.prax`
**Last Updated:** 2026-03-08
**Last Updated:** 2026-03-17
---
@@ -31,7 +31,7 @@ logger.warning("Disk usage high")
logger.error("Connection failed")
```
Logs auto-route to `system_logs/<branch>_<module>.log` based on the calling module. No configuration needed — prax detects the caller via stack introspection.
Logs auto-route via two-tier placement: `system_logs/<branch>_<module>.log` (central aggregation) and `<branch>/logs/` (branch-local). No configuration needed — prax detects the caller via stack introspection.
For handlers or plugins that need to bypass the event pipeline:
@@ -73,7 +73,9 @@ quit # Exit
| `drone @prax log-audit` | Audit log file sizes and health |
| `drone @prax terminal enable\|disable` | Enable or disable terminal output |
| `drone @prax dashboard` | Show system dashboard |
| `drone @prax dashboard refresh --all` | Refresh dashboard data from centrals |
| `drone @prax agent-status` | Show agent status overview |
| `drone @prax status sync` | Sync STATUS.md from all branch STATUS.local.md |
## Architecture
@@ -91,23 +93,29 @@ prax/
│ │ ├── log_audit_module.py # Log file audit
│ │ ├── run_module.py # Continuous logging mode
│ │ ├── shutdown_module.py # Logging system shutdown
│ │ ├── status_module.py # System status display
│ │ ├── status_module.py # System status / STATUS sync
│ │ └── terminal_module.py # Terminal output toggle
│ └── handlers/
│ ├── central/ # Central file reader
│ ├── config/ # Configuration loading
│ ├── dashboard/ # Dashboard refresh and operations
│ ├── discovery/ # Module scanning and filtering
│ ├── logging/ # Log setup, rotation, introspection
│ ├── monitoring/ # Event queue, branch detection, stream output
│ ├── discovery/ # Module scanning and filtering
│ ├── config/ # Configuration loading
│ └── registry/ # Module registry management
│ ├── registry/ # Module registry management
│ ├── status/ # STATUS sync handler
│ └── watcher/ # File and log watchers
├── docs/ # Documentation
├── templates/ # Dashboard templates
└── tests/ # Test suite
```
## How It Works
1. **Auto-routing** — When any module calls `logger.info()`, prax inspects the call stack to identify the caller and routes the log entry to the appropriate file.
2. **Dual output** — Each log entry goes to both a system-wide log (`system_logs/`) and a branch-local log (`{branch}/logs/`), both with rotation.
2. **Two-tier logging** — Each log entry goes to both `system_logs/` (central aggregation) and `<branch>/logs/` (branch-local), both with rotation. No nested hierarchical placement.
3. **Mission Control** — A multi-threaded monitoring console that watches file changes (via inotify), log events, and agent activity across all branches simultaneously.
4. **Dashboard** — Aggregates data from central files and branch status into per-branch dashboard views. Supports manual refresh from centrals.
---
@@ -124,4 +132,4 @@ prax/
---
*Last Updated: 2026-03-08*
*Last Updated: 2026-03-17*
src/aipass/prax/apps/handlers/config/load.py
+3 -39
View File
@@ -73,8 +73,9 @@ def get_module_logs_dir(module_name: str) -> Path:
"""Get the branch-root logs directory for a module.
Returns ECOSYSTEM_ROOT / module_name / "logs", creating it if needed.
For hierarchical placement (logs at the caller's level), use
get_hierarchical_logs_dir() instead.
This is the primary local log directory resolver for the two-tier
model (system_logs/ for central aggregation + branch-root logs/
for local debugging).
Args:
module_name: Module name (e.g., "flow", "prax", "trigger")
@@ -86,43 +87,6 @@ def get_module_logs_dir(module_name: str) -> Path:
logs_dir.mkdir(parents=True, exist_ok=True)
return logs_dir
def get_hierarchical_logs_dir(caller_path: str) -> Path:
"""Get the logs directory at the caller's level in the code hierarchy.
Resolves a logs/ directory as a sibling of the caller's parent directory.
Logs live where the code lives:
handlers/dispatch/wake.py → handlers/dispatch/logs/
modules/email.py → modules/logs/
apps/branch.py → apps/logs/
Falls back to branch-root logs/ if the caller is outside the
ecosystem or path resolution fails.
Args:
caller_path: Absolute path to the calling Python file
Returns:
Path to the logs directory (created if it doesn't exist)
"""
try:
caller = Path(caller_path).resolve()
caller_dir = caller.parent
# Verify caller is inside the ecosystem
try:
caller.relative_to(ECOSYSTEM_ROOT)
except ValueError:
# Outside ecosystem — fall back to branch root
return get_module_logs_dir("prax")
logs_dir = caller_dir / "logs"
logs_dir.mkdir(parents=True, exist_ok=True)
return logs_dir
except Exception:
return get_module_logs_dir("prax")
# Config file
PRAX_LOGGER_CONFIG_FILE = PRAX_JSON_DIR / "prax_logger_config.json"
src/aipass/prax/apps/handlers/dashboard/template_pusher.py
+4 -2
View File
@@ -106,7 +106,7 @@ def _replace_placeholders(template: dict, branch_name: str) -> dict:
Returns:
New dict with placeholders replaced
"""
def _walk(val):
def _walk(val: Any) -> Any:
if isinstance(val, str):
return val.replace("{{BRANCHNAME}}", branch_name)
elif isinstance(val, list):
@@ -115,7 +115,9 @@ def _replace_placeholders(template: dict, branch_name: str) -> dict:
return {k: _walk(v) for k, v in val.items()}
return val
return _walk(copy.deepcopy(template))
result = _walk(copy.deepcopy(template))
assert isinstance(result, dict)
return result
# =============================================================================
src/aipass/prax/apps/handlers/discovery/watcher.py
+2 -1
View File
@@ -42,6 +42,7 @@ try:
from aipass.trigger.apps.modules.core import trigger
_HAS_TRIGGER = True
except ImportError:
trigger = None # type: ignore[assignment]
_HAS_TRIGGER = False
# Global observer instance
@@ -91,7 +92,7 @@ class PythonFileWatcher(FileSystemEventHandler):
# Fire trigger event for module discovery
if _HAS_TRIGGER:
try:
trigger.fire('module_discovered',
trigger.fire('module_discovered', # type: ignore[union-attr]
module_name=module_name,
file_path=str(py_file),
relative_path=str(relative_path)
src/aipass/prax/apps/handlers/logging/cli_commands.py
-85
View File
@@ -1,85 +0,0 @@
# =================== AIPass ====================
# Name: cli_commands.py
# Description: CLI Command Handlers
# Version: 1.0.0
# Created: 2025-11-10
# Modified: 2026-03-09
# =============================================
"""
PRAX CLI Command Handlers
Command handlers for prax_logger CLI commands (init, status, test, run).
These are called by modules/prax_logger.py main() function.
"""
# NOTE: These handlers will be used by modules/prax_logger.py once it's created.
# They reference functions that will be exported from that module.
# For now, this file serves as the handler structure.
def handle_init(args):
"""Handle init command
Initializes the prax logging system:
- Creates config file if missing
- Discovers all Python modules
- Sets up system logger
- Installs logger override
- Starts file watcher
"""
# Implementation will import from modules.prax_logger
# from aipass.prax.apps.modules.prax_logger import initialize_logging_system
# initialize_logging_system()
pass
def handle_status(args):
"""Handle status command
Displays current system status:
- Total modules discovered
- Individual loggers created
- System logs directory
- Registry file location
- File watcher status
- Logger override status
"""
# Implementation will import from modules.prax_logger
# from aipass.prax.apps.modules.prax_logger import get_system_status
# status = get_system_status()
# print("\n" + "="*60)
# print("PRAX LOGGING SYSTEM STATUS")
# print("="*60)
# for key, value in status.items():
# print(f"{key:.<40} {value}")
# print("="*60 + "\n")
pass
def handle_test(args):
"""Handle test command
Runs system self-test:
1. Initialize logging system
2. Check system status
3. Test logger capture
4. Check log files created
5. Check module registry
6. Test file watcher
7. Clean shutdown
"""
# Implementation will import from modules.prax_logger
# Full test implementation goes here
pass
def handle_run(args):
"""Handle run command
Starts continuous logging in background mode:
- Enables terminal output
- Initializes logging system
- Runs until Ctrl+C
- Displays status updates every 5 minutes
"""
# Implementation will import from modules.prax_logger
# from aipass.prax.apps.modules.prax_logger import start_continuous_logging
# start_continuous_logging()
pass
src/aipass/prax/apps/handlers/logging/setup.py
+7 -22
View File
@@ -22,7 +22,6 @@ from logging.handlers import RotatingFileHandler
from aipass.prax.apps.handlers.config.load import (
get_system_logs_dir,
get_module_logs_dir,
get_hierarchical_logs_dir,
DEFAULT_LOG_LEVEL,
load_log_config,
lines_to_bytes
@@ -46,7 +45,8 @@ try:
from aipass.prax.apps.handlers.logging.terminal.filtering import should_display_terminal
_terminal_module_available = True
except ImportError:
pass # Terminal module not available yet
create_terminal_handler = None # type: ignore[assignment]
should_display_terminal = None # type: ignore[assignment]
def _safe_rotating_handler(log_file: Path, max_bytes: int, backup_count: int) -> logging.Handler:
"""Create RotatingFileHandler — self-heals missing directories, never crashes."""
@@ -118,24 +118,9 @@ def setup_individual_logger(module_name: str) -> logging.Logger:
system_handler.setFormatter(formatter)
logger.addHandler(system_handler)
# HANDLER 2: Hierarchical local log (logs live where the code lives)
if module_path:
hierarchical_logs_dir = get_hierarchical_logs_dir(module_path)
# Detect fallback: caller was in ecosystem but resolution fell back
expected_logs = Path(module_path).resolve().parent / "logs"
if hierarchical_logs_dir.resolve() != expected_logs.resolve() and _system_logger:
_system_logger.warning(
f"Log placement fallback for {module_name}: "
f"caller outside ecosystem, logs redirected to {hierarchical_logs_dir}"
)
else:
hierarchical_logs_dir = get_module_logs_dir(branch_name)
if _system_logger:
_system_logger.warning(
f"Log placement fallback for {module_name}: "
f"stack introspection failed, using branch root {hierarchical_logs_dir}"
)
module_log_file = hierarchical_logs_dir / f"{module_name}.log"
# HANDLER 2: Branch-root local log (two-tier: system_logs/ + branch logs/)
local_logs_dir = get_module_logs_dir(branch_name)
module_log_file = local_logs_dir / f"{module_name}.log"
local_limits = log_config['local_logs']
local_max_bytes = lines_to_bytes(local_limits['max_lines'])
local_handler = _safe_rotating_handler(module_log_file, local_max_bytes, local_limits['backup_count'])
@@ -147,8 +132,8 @@ def setup_individual_logger(module_name: str) -> logging.Logger:
# HANDLER 3: Terminal output (if enabled)
if _terminal_output_enabled and _terminal_module_available:
if should_display_terminal(module_name):
terminal_handler = create_terminal_handler()
if should_display_terminal(module_name): # type: ignore[misc]
terminal_handler = create_terminal_handler() # type: ignore[misc]
logger.addHandler(terminal_handler)
# Store for reuse
src/aipass/prax/apps/handlers/monitoring/INTEGRATION_EXAMPLE.py
-220
View File
@@ -1,220 +0,0 @@
# =================== AIPass ====================
# Name: INTEGRATION_EXAMPLE.py
# Description: Example: File Watcher Integration with monitor_module.py
# Version: 1.0.0
# Created: 2026-03-09
# Modified: 2026-03-09
# =============================================
"""
Example: File Watcher Integration with monitor_module.py
This shows how to integrate the file watcher with the monitoring module.
This code would go in monitor_module.py's handle_command() function.
"""
import sys
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.prax.apps.handlers.monitoring import (
start_file_watcher,
stop_file_watcher,
is_file_watcher_running,
get_file_watcher_stats,
FileWatcherManager,
MonitoringEvent,
global_queue,
print_event,
)
def example_monitor_with_file_watcher(branch_filter=None):
"""
Example monitoring loop with file watcher integration
Args:
branch_filter: Optional list of branches to watch (e.g., ['PRAX', 'CLI'])
None = watch all branches (may hit inotify limits)
"""
logger.info("Starting PRAX Monitor with File Watcher")
logger.info("=" * 60)
# Start file watcher
if branch_filter:
logger.info(f"Starting file watcher for branches: {', '.join(branch_filter)}")
watcher = FileWatcherManager(branch_filter=branch_filter)
success = watcher.start()
else:
logger.info("Starting file watcher for all branches")
success = start_file_watcher()
if not success:
logger.error("Failed to start file watcher")
return
# Show stats
stats = get_file_watcher_stats()
logger.info(f"Watching {stats['branches_watched']} branches:")
for branch in stats['branch_names']:
logger.info(f" - {branch}")
logger.info("Monitoring active - press Ctrl+C to stop")
logger.info("-" * 60)
try:
# Main monitoring loop
while True:
# Get next event from queue (blocks for 0.5 seconds)
event = global_queue.dequeue(timeout=0.5)
if event:
# Handle different event types
if event.event_type == 'file':
# File change event
print_event(event.event_type, event.branch, event.message, event.level)
elif event.event_type == 'log':
# Log event (from log monitor - future)
print_event(event.event_type, event.branch, event.message, event.level)
elif event.event_type == 'module':
# Module execution event (from module tracker)
print_event(event.event_type, event.branch, event.message, event.level)
# You can also handle events directly:
# logger.info(f"[{event.branch}] {event.action}: {event.message}")
except KeyboardInterrupt:
logger.info("-" * 60)
logger.info("Stopping monitor...")
finally:
# Cleanup
stop_file_watcher()
logger.info("File watcher stopped")
logger.info("Monitor stopped")
def example_filtered_monitoring():
"""
Example: Monitor only PRAX and CLI branches
Recommended approach to avoid inotify limits
"""
example_monitor_with_file_watcher(branch_filter=['PRAX', 'CLI'])
def example_all_branches_monitoring():
"""
Example: Monitor all branches
WARNING: May hit inotify limits on systems with many branches
"""
example_monitor_with_file_watcher(branch_filter=None)
def example_custom_event_handling():
"""
Example: Custom event handling logic
"""
logger.info("Custom Event Handling Example")
logger.info("=" * 60)
# Start watcher for specific branches
watcher = FileWatcherManager(branch_filter=['PRAX'])
watcher.start()
logger.info("Watching PRAX branch for 30 seconds...")
logger.info("Try modifying a file in src/aipass/prax/")
import time
start_time = time.time()
event_count = 0
try:
while time.time() - start_time < 30:
event = global_queue.dequeue(timeout=0.5)
if event:
event_count += 1
# Custom handling based on action
if event.action == 'created':
logger.info(f"NEW FILE: {event.message}")
logger.info(f" Branch: {event.branch}")
logger.info(f" Time: {event.timestamp}")
elif event.action == 'modified':
logger.info(f"MODIFIED: {event.message}")
elif event.action == 'deleted':
logger.info(f"DELETED: {event.message}")
elif event.action == 'moved':
logger.info(f"MOVED: {event.message}")
except KeyboardInterrupt:
logger.info("Stopping...")
finally:
watcher.stop()
logger.info(f"Captured {event_count} events in total")
# For integration into monitor_module.py's handle_command():
"""
def handle_command(command: str, args: List[str]) -> bool:
if command != 'monitor':
return False
# Parse branch filter from args
branch_filter = None
if args:
branch_arg = args[0]
if branch_arg.lower() != 'all':
branch_filter = [b.strip().upper() for b in branch_arg.split(',')]
# Start file watcher
if branch_filter:
watcher = FileWatcherManager(branch_filter=branch_filter)
watcher.start()
else:
start_file_watcher()
# Main monitoring loop
try:
while True:
event = global_queue.dequeue(timeout=0.5)
if event:
print_event(event)
except KeyboardInterrupt:
pass
finally:
stop_file_watcher()
return True
"""
if __name__ == '__main__':
import sys
if len(sys.argv) > 1:
if sys.argv[1] == 'custom':
example_custom_event_handling()
elif sys.argv[1] == 'all':
example_all_branches_monitoring()
else:
# Parse branch list
branches = [b.strip().upper() for b in sys.argv[1].split(',')]
example_monitor_with_file_watcher(branch_filter=branches)
else:
# Default: monitor PRAX only
logger.info("Usage:")
logger.info(" python3 INTEGRATION_EXAMPLE.py # Monitor PRAX only")
logger.info(" python3 INTEGRATION_EXAMPLE.py PRAX,CLI # Monitor specific branches")
logger.info(" python3 INTEGRATION_EXAMPLE.py all # Monitor all branches (may fail)")
logger.info(" python3 INTEGRATION_EXAMPLE.py custom # Custom event handling demo")
logger.info("Running default: PRAX only")
example_monitor_with_file_watcher(branch_filter=['PRAX'])
src/aipass/prax/apps/handlers/monitoring/MONITOR_MODULE_INTEGRATION.py
-203
View File
@@ -1,203 +0,0 @@
# =================== AIPass ====================
# Name: MONITOR_MODULE_INTEGRATION.py
# Description: EXAMPLE: How to integrate log_watcher into monitor_module.py
# Version: 1.0.0
# Created: 2026-03-09
# Modified: 2026-03-09
# =============================================
"""
EXAMPLE: How to integrate log_watcher into monitor_module.py
This is a reference implementation showing the minimal changes needed
to add log monitoring to monitor_module.py's handle_command() function.
Copy the relevant sections into monitor_module.py as needed.
"""
import sys
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
# Import monitoring components
from aipass.prax.apps.handlers.monitoring import (
start_log_watcher, # NEW: Log watcher integration
stop_log_watcher, # NEW: Log watcher cleanup
is_log_watcher_active, # NEW: Status check
MonitoringQueue,
MonitoringEvent,
print_event,
FilterState,
should_display_event,
)
# Global state
_log_observer = None
_event_queue = None
def handle_command(command: str, args: List[str]) -> bool:
"""
Example handle_command with log watcher integration
This shows the minimal changes to add real-time log monitoring
to the existing monitor_module.py structure.
"""
if command != 'monitor':
return False
global _log_observer, _event_queue
logger.info(f"Starting unified monitoring (args: {args})")
logger.info("PRAX Mission Control - Unified Monitoring")
# Initialize event queue
_event_queue = MonitoringQueue()
logger.info("Event queue initialized")
# Start log watcher - NEW INTEGRATION
try:
_log_observer = start_log_watcher(_event_queue)
logger.info("Log watcher started")
logger.info("Monitoring: system_logs/*.log")
except Exception as e:
logger.error(f"Failed to start log watcher: {e}")
return False
logger.info("Monitoring active - type 'quit' to exit")
# Initialize filter state
filter_state = FilterState()
# Parse branch filters from args
if args:
branches = args[0].split(',') if args[0] != 'all' else []
if branches:
filter_state.watched_branches = {b.strip().upper() for b in branches}
logger.info(f"Filtering branches: {', '.join(filter_state.watched_branches)}")
try:
# Main event loop
while True:
# Dequeue next event (with timeout to allow Ctrl+C)
event = _event_queue.dequeue(timeout=0.5) if _event_queue else None
if event:
# Apply filters
if should_display_event(event.event_type, event.branch, event.level, filter_state):
# Handle command separator events
if event.event_type == 'command':
logger.info(f"Command: {event.message}")
# Handle log events
elif event.event_type == 'log':
# Format timestamp
timestamp = event.timestamp.strftime('%H:%M:%S')
# Branch column (right-aligned, fixed width)
branch_col = f"[{event.branch:>8}]"
# Display event via logger
logger.info(f"{timestamp} {branch_col} {event.message}")
# Check for keyboard input (simplified - use proper input handling)
# TODO: Add interactive command handling here
except KeyboardInterrupt:
logger.info("Stopping monitoring...")
finally:
# Cleanup
if _log_observer:
stop_log_watcher()
logger.info("Log watcher stopped")
if _event_queue:
_event_queue.stop()
logger.info("Event queue stopped")
logger.info("Monitoring stopped")
return True
def example_with_interactive_commands():
"""
Example showing interactive command handling
This is a more complete version with command input handling.
Requires threading to read both events and keyboard input.
"""
import threading
import select
def input_handler(running_flag):
"""Thread for reading keyboard input"""
while running_flag[0]:
# Use select for non-blocking input on Unix
if select.select([sys.stdin], [], [], 0.5)[0]:
try:
user_input = sys.stdin.readline().strip()
if user_input in ['quit', 'exit']:
running_flag[0] = False
elif user_input == 'help':
logger.info("Commands: help, status, quit")
elif user_input == 'status':
logger.info(f"Log watcher: {'active' if is_log_watcher_active() else 'inactive'}")
logger.info(f"Queue size: {_event_queue.size() if _event_queue else 0}")
except Exception:
pass
# Running flag for threads
running = [True]
# Start input handler thread
input_thread = threading.Thread(target=input_handler, args=(running,))
input_thread.daemon = True
input_thread.start()
# Main event loop
while running[0]:
event = _event_queue.dequeue(timeout=0.5) if _event_queue else None
if event:
# Display event (same as above)
pass
# Example of filter adjustment
def example_filter_adjustment():
"""Show how to dynamically adjust filters"""
filter_state = FilterState()
# Watch specific branches
filter_state.watched_branches = {'SEED', 'FLOW', 'PRAX'}
# Apply filter
event = MonitoringEvent(
priority=1,
event_type='log',
branch='SEED',
level='error',
message='ERROR: Something went wrong'
)
if should_display_event(event.event_type, event.branch, event.level, filter_state):
logger.info("Event passed filters")
if __name__ == '__main__':
logger.info("Monitor Module Integration Examples")
logger.info("This file shows examples of integrating log_watcher.py")
logger.info("into monitor_module.py's handle_command() function.")
logger.info("Key changes:")
logger.info(" 1. Import start_log_watcher, stop_log_watcher")
logger.info(" 2. Initialize MonitoringQueue")
logger.info(" 3. Start log watcher with queue")
logger.info(" 4. Main loop dequeues and displays events")
logger.info(" 5. Cleanup on exit")
logger.info("See code above for full implementation details.")
src/aipass/prax/apps/handlers/monitoring/file_watcher_integration.py
+11 -14
View File
@@ -34,7 +34,7 @@ Linux Limitations:
import json
import logging
from pathlib import Path
from typing import List, Tuple, Optional, TYPE_CHECKING, Any
from typing import List, Tuple, Optional, Any
from datetime import datetime
from aipass.prax import logger
@@ -48,26 +48,23 @@ from aipass.prax import logger
try:
# Import file watcher handler
from aipass.prax.apps.handlers.watcher.monitor import (
BranchFileHandler,
start_monitoring,
stop_monitoring,
WATCHDOG_AVAILABLE,
Observer
)
from aipass.prax.apps.handlers.monitoring.event_queue import (
MonitoringEvent,
MonitoringQueue,
global_queue
)
from aipass.prax.apps.handlers.monitoring.branch_detector import (
detect_branch_from_path
)
except ImportError as e:
logger.error(f"Import error in file_watcher_integration: {e}")
WATCHDOG_AVAILABLE = False
start_monitoring = None # type: ignore[assignment]
stop_monitoring = None # type: ignore[assignment]
MonitoringEvent = None # type: ignore[assignment, misc]
global_queue = None # type: ignore[assignment]
# =============================================================================
@@ -174,7 +171,7 @@ def file_event_callback(branch_name: str, event_type: str, file_path: str):
priority = priority_map.get(action, 3)
# Create monitoring event
event = MonitoringEvent(
event = MonitoringEvent( # type: ignore[misc]
priority=priority,
timestamp=datetime.now(),
event_type='file',
@@ -185,7 +182,7 @@ def file_event_callback(branch_name: str, event_type: str, file_path: str):
)
# Enqueue to global queue (thread-safe)
success = global_queue.enqueue(event)
success = global_queue.enqueue(event) # type: ignore[union-attr]
if not success:
logger.info(f"Failed to enqueue file event: {branch_name} {action} {file_path}")
@@ -208,7 +205,7 @@ class FileWatcherManager:
- Connect file events to monitoring queue
"""
def __init__(self, queue: MonitoringQueue | None = None, branch_filter: List[str] | None = None):
def __init__(self, queue: Any = None, branch_filter: List[str] | None = None):
"""
Initialize file watcher manager
@@ -246,7 +243,7 @@ class FileWatcherManager:
# Start monitoring with callback
logger.info(f"Starting file watcher for {len(self.branch_paths)} branches")
self.observer = start_monitoring(self.branch_paths, file_event_callback)
self.observer = start_monitoring(self.branch_paths, file_event_callback) # type: ignore[misc]
if self.observer:
self.running = True
@@ -263,7 +260,7 @@ class FileWatcherManager:
if self.observer:
logger.info("Stopping file watcher")
stop_monitoring(self.observer)
stop_monitoring(self.observer) # type: ignore[misc]
self.observer = None
self.running = False
@@ -380,7 +377,7 @@ if __name__ == '__main__':
event_count = 0
while time.time() - start_time < 10:
event = global_queue.dequeue(timeout=0.5)
event = global_queue.dequeue(timeout=0.5) # type: ignore[union-attr]
if event:
event_count += 1
print(f"Event #{event_count}: {event.branch} - {event.action} - {event.message}")
src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py
+2 -2
View File
@@ -19,7 +19,6 @@ Extracted from monitor_module.py to maintain 3-layer architecture:
"""
import json as _json
import threading
from pathlib import Path
from typing import Optional, Dict
@@ -32,6 +31,7 @@ try:
from aipass.trigger.apps.modules.core import trigger
_trigger_available = True
except ImportError:
trigger = None # type: ignore[assignment]
_trigger_available = False
# Monitoring subsystem imports
@@ -120,7 +120,7 @@ class MonitoringFileHandler(FileSystemEventHandler):
if not _trigger_available:
return
try:
trigger.fire(event_name, **kwargs)
trigger.fire(event_name, **kwargs) # type: ignore[union-attr]
except Exception as e:
logger.warning(f"[monitor] trigger.fire('{event_name}') failed: {e}")
src/aipass/prax/apps/handlers/monitoring/log_watcher.py
+6 -6
View File
@@ -23,10 +23,9 @@ Based on: apps/handlers/discovery/watcher.py (production-ready log tailing)
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional, Dict, Any, Union
from datetime import datetime
from typing import Optional, Dict, Any
import re
import logging
from aipass.prax import logger
from watchdog.observers import Observer as WatchdogObserver
@@ -41,9 +40,10 @@ from aipass.prax.apps.handlers.monitoring.branch_detector import detect_branch_f
# Trigger integration - graceful fallback if trigger not available
try:
from trigger import trigger
from aipass.trigger.apps.modules.core import trigger
HAS_TRIGGER = True
except ImportError:
trigger = None # type: ignore[assignment]
HAS_TRIGGER = False
# Logger
@@ -162,7 +162,7 @@ class LogFileWatcher(FileSystemEventHandler):
# Log error but don't crash watcher
logger.info(f"Error reading log file {file_path}: {e}")
def _should_display_log(self, log_line: str) -> bool:
def _should_display_log(self, _log_line: str) -> bool:
"""Check if log line should be displayed. No filtering — show everything."""
return True
@@ -409,7 +409,7 @@ class LogFileWatcher(FileSystemEventHandler):
if len(parts) >= 2:
module_name = parts[1].strip()
trigger.fire('error_detected',
trigger.fire('error_detected', # type: ignore[union-attr]
branch=branch,
message=clean_message,
error_hash=_generate_error_hash(module_name, clean_message),
src/aipass/prax/apps/handlers/status/sync.py
+1 -2
View File
@@ -22,7 +22,6 @@ Usage (internal):
import json
import re
from datetime import datetime
from pathlib import Path
from typing import Dict
from aipass.prax.apps.handlers.config.load import _find_repo_root
@@ -93,7 +92,7 @@ def sync_status() -> Dict:
if not status_file.exists():
missing.append(branch_email)
logger.debug("STATUS.local.md missing for %s", branch_email)
logger.info("STATUS.local.md missing for %s", branch_email)
continue
try:
src/aipass/prax/apps/json_templates/__init__.py
-1
View File
@@ -1 +0,0 @@
# JSON Templates package - Default JSON file templates
src/aipass/prax/apps/json_templates/default/config.json
-9
View File
@@ -1,9 +0,0 @@
{
"module_name": "{{MODULE_NAME}}",
"version": "1.0.0",
"timestamp": "2025-11-13",
"config": {
"auto_save": true,
"enabled": true
}
}
src/aipass/prax/apps/json_templates/default/data.json
-8
View File
@@ -1,8 +0,0 @@
{
"module_name": "{{MODULE_NAME}}",
"created": "2025-11-13",
"last_updated": "2025-11-13",
"operations_total": 0,
"operations_successful": 0,
"operations_failed": 0
}
src/aipass/prax/apps/json_templates/default/log.json
-1
View File
@@ -1 +0,0 @@
[]
src/aipass/prax/apps/modules/agent_status_module.py
+4 -1
View File
@@ -16,7 +16,6 @@ Pushes agent_status section to all branch dashboards showing active/stale agents
import sys
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, error
@@ -73,6 +72,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'agent-status-push':
return False
if not args:
print_introspection()
return True
from aipass.prax.apps.handlers.dashboard.agent_status_writer import (
build_agent_status_section,
push_agent_status_dashboard,
src/aipass/prax/apps/modules/dashboard.py
+28 -5
View File
@@ -15,6 +15,23 @@ DASHBOARD.local.json files. Each service manages only its own section.
Run directly or via: python3 apps/modules/dashboard.py
"""
__all__ = [
"write_section",
"update_section",
"get_dashboard_path",
"calculate_quick_status",
"get_branch_paths",
"resolve_branch_path",
"refresh_all_dashboards",
"refresh_single_dashboard",
"push_dashboard_template",
"get_template_status",
"diff_dashboard_template",
"DASHBOARD_TEMPLATE",
"handle_command",
"main",
]
import json
import sys
from pathlib import Path
@@ -25,10 +42,8 @@ from aipass.cli.apps.modules import console, error, warning
# Import handlers
from aipass.prax.apps.handlers.dashboard.operations import (
load_dashboard,
save_dashboard,
update_section as handler_update_section,
write_section,
write_section, # re-exported: used by ai_mail.apps.handlers.email.dashboard_sync
get_dashboard_path,
)
from aipass.prax.apps.handlers.dashboard.status import (
@@ -373,7 +388,11 @@ def _handle_diff_template(args: List[str]) -> None:
console.print(f" Needs update: {summary.get('needs_update', 0)}")
console.print(f" Up to date: {summary.get('up_to_date', 0)}")
console.print(f" Missing: {summary.get('missing', 0)}")
console.print(f" Invalid JSON: {summary.get('invalid_json', 0)}")
invalid_json_count = summary.get('invalid_json', 0)
if invalid_json_count:
error(f"Invalid JSON: {invalid_json_count}")
else:
console.print(f" Invalid JSON: {invalid_json_count}")
for branch_diff in result.get("branches", []):
status = branch_diff["status"]
@@ -443,7 +462,11 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != "dashboard":
return False
subcmd = args[0] if args else ""
if not args:
print_introspection()
return True
subcmd = args[0]
if subcmd == "status":
print_status()
src/aipass/prax/apps/modules/discover_module.py
+5 -2
View File
@@ -15,8 +15,7 @@ Implements the 'discover' command using handle_command interface.
import sys
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header, success, error
from aipass.cli.apps.modules import console
from aipass.prax.apps.handlers.discovery.scanner import discover_python_modules
@@ -54,6 +53,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'discover':
return False
if not args:
print_introspection()
return True
console.print("🔍 Discovering Python modules...")
modules = discover_python_modules()
console.print(f"✅ Discovered {len(modules)} modules")
src/aipass/prax/apps/modules/init_module.py
+4
View File
@@ -69,6 +69,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'init':
return False
if not args:
print_introspection()
return True
try:
console.print("🚀 Initializing PRAX logging system...")
initialize_logging_system()
src/aipass/prax/apps/modules/log_audit_module.py
+4
View File
@@ -103,6 +103,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'log-audit':
return False
if not args:
print_introspection()
return True
from aipass.prax.apps.handlers.logging.log_watchdog import (
scan_log_files,
enforce_log_limits,
src/aipass/prax/apps/modules/logger.py
+23 -2
View File
@@ -20,6 +20,25 @@ Provides:
- Status and control functions
"""
__all__ = [
"system_logger",
"get_system_logger",
"SystemLogger",
"get_direct_logger",
"direct_log",
"DirectLogger",
"initialize_logging_system",
"shutdown_logging_system",
"start_continuous_logging",
"get_system_status",
"enable_terminal_output",
"disable_terminal_output",
"print_introspection",
"handle_command",
"MODULE_NAME",
"DATA_FILE",
]
import sys
from typing import Dict, Any
@@ -39,7 +58,6 @@ from aipass.prax.apps.handlers.logging.override import (
)
from aipass.prax.apps.handlers.discovery.watcher import (
start_file_watcher,
stop_file_watcher,
is_file_watcher_active
)
from aipass.prax.apps.handlers.registry.load import load_module_registry
@@ -264,10 +282,13 @@ def print_introspection():
console.print()
def handle_command(command: str, args: list) -> bool:
def handle_command(_command: str, args: list) -> bool:
"""Handle commands routed by the entry point.
Logger is a service module with no user-facing commands.
All interaction happens through the system_logger API.
"""
if not args:
print_introspection()
return True
return False
src/aipass/prax/apps/modules/monitor_module.py
+9 -6
View File
@@ -57,13 +57,12 @@ from typing import List, Optional
from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services (display/output formatting)
from aipass.cli.apps.modules import console, header, success, error, warning
from aipass.cli.apps.modules import console, header, error
# Monitoring handlers (connected subsystems)
from aipass.prax.apps.handlers.monitoring import (
print_event, # unified_stream.py
print_command_separator, # unified_stream.py - command headers
parse_command, # interactive_filter.py
MonitoringQueue, # event_queue.py
ModuleTracker, # module_tracker.py
)
@@ -201,6 +200,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'monitor':
return False
if not args:
print_introspection()
return True
global _monitoring_active, _event_queue, _module_tracker
global _display_thread, _file_watcher_thread, _log_watcher_thread
@@ -343,7 +346,7 @@ def _log_watcher_worker():
if _event_queue is None:
logger.error("[monitor] Event queue not initialized for log watcher")
return
observer = start_log_watcher(_event_queue)
_observer = start_log_watcher(_event_queue)
try:
while _monitoring_active:
@@ -533,10 +536,10 @@ if __name__ == "__main__":
sys.exit(0)
# Prepare arguments for handle_command
cmd_args = []
_cmd_args = []
if args.branches:
cmd_args = [args.branches]
_cmd_args = [args.branches]
# Execute monitor command
handled = handle_command('monitor', cmd_args)
handled = handle_command('monitor', _cmd_args)
sys.exit(0 if handled else 1)
src/aipass/prax/apps/modules/run_module.py
+6 -3
View File
@@ -13,11 +13,10 @@ Implements the 'run' command using handle_command interface.
"""
import sys
from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import start_continuous_logging, system_logger as logger
from aipass.cli.apps.modules import console, header, success, error
from aipass.prax.apps.modules.logger import start_continuous_logging
from aipass.cli.apps.modules import console
def print_help():
@@ -50,6 +49,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'run':
return False
if not args:
print_introspection()
return True
console.print("🚀 Starting PRAX continuous logging mode...")
start_continuous_logging()
return True
src/aipass/prax/apps/modules/shutdown_module.py
+6 -3
View File
@@ -13,11 +13,10 @@ Implements the 'shutdown' command using handle_command interface.
"""
import sys
from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import shutdown_logging_system, system_logger as logger
from aipass.cli.apps.modules import console, header, success, error
from aipass.prax.apps.modules.logger import shutdown_logging_system
from aipass.cli.apps.modules import console
def print_help():
@@ -50,6 +49,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'shutdown':
return False
if not args:
print_introspection()
return True
console.print("🛑 Shutting down PRAX logging system...")
shutdown_logging_system()
console.print("✅ PRAX logging system shutdown complete")
src/aipass/prax/apps/modules/status_module.py
+5 -2
View File
@@ -13,12 +13,11 @@ Implements the 'status' command using handle_command interface.
"""
import sys
from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import get_system_status, system_logger as logger
from aipass.prax.apps.handlers.status.sync import sync_status
from aipass.cli.apps.modules import console, header, success, error, warning
from aipass.cli.apps.modules import console, success, error, warning
def print_help():
@@ -45,6 +44,10 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'status':
return False
if not args:
print_introspection()
return True
# --- sub-command routing ------------------------------------------------
if args and args[0] in ("--help", "help"):
print_help()
src/aipass/prax/apps/modules/terminal_module.py
+5 -2
View File
@@ -13,7 +13,6 @@ Implements the 'terminal' command using handle_command interface.
"""
import sys
from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import enable_terminal_output, disable_terminal_output, system_logger as logger
@@ -85,8 +84,12 @@ def handle_command(command: str, args: List[str]) -> bool:
if command != 'terminal':
return False
if not args:
print_introspection()
return True
try:
if not args or args[0] not in ['enable', 'disable']:
if args[0] not in ['enable', 'disable']:
print_help()
return True # Command was handled, even if validation failed