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

chore(prax): json_structure compliance + stderr fix + dead code cleanup (#65)

Browse Source 
- Wire json_handler.log_operation() into all 51 modules/handlers (1%→100%)
- Fix stderr_routing: remove "Invalid JSON: 0" noise from dashboard.py
- Archive 9 dead handler files to .archive/ (never imported by anything):
  registry/{ignore,meta_ops,reader,statistics}.py
  watcher/reporter.py
  json/{initialize,load,log,save}.py

Seedgo audit: 93%→98% (23 of 24 categories at 100%)
Log_Structure remaining issue is a seedgo checker bug (PR #63)

Co-authored-by: @prax <prax@aipass>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIPass
2026-03-17 16:44:28 -07:00
committed by GitHub
co-authored by @prax Claude Opus 4.6
parent 1f08c0737f
commit eb527fe0b2
55 changed files with 173 additions and 1296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/prax/apps/handlers/central/reader.py
+3
View File
@@ -20,6 +20,7 @@ from pathlib import Path
from typing import Dict
from aipass.prax.apps.handlers.config.load import _find_repo_root
from aipass.prax.apps.handlers.json import json_handler
def read_all_centrals() -> Dict:
@@ -47,4 +48,6 @@ def read_all_centrals() -> Dict:
except (json.JSONDecodeError, OSError):
continue
json_handler.log_operation("central_data_read", {"services_found": len(centrals)})
return centrals
src/aipass/prax/apps/handlers/config/ignore_patterns.py
+2
View File
@@ -31,6 +31,7 @@ from pathlib import Path
from typing import Set
from aipass.prax.apps.handlers.config.load import PRAX_ROOT
from aipass.prax.apps.handlers.json import json_handler
# =============================================
# CONFIGURATION
@@ -73,6 +74,7 @@ def load_ignore_patterns_from_config() -> Set[str]:
config = json.load(f)
patterns = config.get('config', {}).get('ignore_patterns', [])
if patterns:
json_handler.log_operation("ignore_patterns_loaded", {"pattern_count": len(patterns)})
return set(patterns)
except Exception:
# Silently fall back to defaults - logging not available at this level
src/aipass/prax/apps/handlers/config/load.py
+5 -1
View File
@@ -32,6 +32,8 @@ import logging
from pathlib import Path
from typing import Dict, Any
from aipass.prax.apps.handlers.json import json_handler
# =============================================
# CONFIGURATION
# =============================================
@@ -174,12 +176,14 @@ def load_log_config() -> Dict[str, Any]:
system_logs = config.get('config', {}).get('system_logs', DEFAULT_SYSTEM_LOGS)
local_logs = config.get('config', {}).get('local_logs', DEFAULT_LOCAL_LOGS)
return {
result = {
'system_logs': system_logs,
'local_logs': local_logs,
'log_format': config.get('config', {}).get('log_format', LOG_FORMAT),
'date_format': config.get('config', {}).get('date_format', DATE_FORMAT)
}
json_handler.log_operation("config_loaded", {"source": str(PRAX_LOGGER_CONFIG_FILE)})
return result
except (json.JSONDecodeError, OSError) as e:
logging.debug(f"Log config load error (using defaults): {e}")
src/aipass/prax/apps/handlers/dashboard/agent_status_writer.py
+9
View File
@@ -26,6 +26,8 @@ from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# CONSTANTS
@@ -304,6 +306,13 @@ def push_agent_status_dashboard() -> bool:
"agent_status", section_data, branch_paths
)
json_handler.log_operation("agent_status_written", {
"branches_targeted": len(branch_paths),
"branches_updated": success_count,
"active_agents": section_data.get("agent_count", 0),
"stale_agents": len(section_data.get("stale_agents", [])),
})
return success_count > 0
except Exception:
src/aipass/prax/apps/handlers/dashboard/operations.py
+10 -1
View File
@@ -18,6 +18,8 @@ from datetime import datetime
from pathlib import Path
from typing import Dict
from aipass.prax.apps.handlers.json import json_handler
# Resolve prax root from this file's location
_PRAX_ROOT = Path(__file__).resolve().parents[3] # .../prax/
@@ -341,7 +343,14 @@ def write_section(branch_path: Path, section_name: str, section_data: Dict) -> b
dashboard["quick_status"] = _calculate_quick_status_standalone(dashboard["sections"])
# Save
return save_dashboard(branch_path, dashboard)
saved = save_dashboard(branch_path, dashboard)
json_handler.log_operation("section_updated", {
"section": section_name,
"branch": branch_path.name,
})
return saved
except Exception:
return False
src/aipass/prax/apps/handlers/dashboard/refresh.py
+8
View File
@@ -24,6 +24,8 @@ from .operations import create_fresh_dashboard, save_dashboard
# Cross-handler imports for central reader
from ..central.reader import read_all_centrals
from aipass.prax.apps.handlers.json import json_handler
# Sections managed by the refresh path — everything else is write-through only
REFRESH_MANAGED_SECTIONS = {"ai_mail", "flow", "memory_bank", "commons_activity"}
@@ -312,6 +314,12 @@ def refresh_all_dashboards() -> Dict:
else:
status = "error"
json_handler.log_operation("dashboard_refreshed", {
"status": status,
"branches_updated": branches_updated,
"branches_failed": branches_failed,
})
return {
"status": status,
"branches_updated": branches_updated,
src/aipass/prax/apps/handlers/dashboard/status.py
+11 -1
View File
@@ -17,6 +17,8 @@ import json
from pathlib import Path
from typing import Dict, List
from aipass.prax.apps.handlers.json import json_handler
def _find_repo_root() -> Path:
"""Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
@@ -66,7 +68,7 @@ def calculate_quick_status(sections: Dict) -> Dict:
if mentions:
summary_parts.append(f"{mentions} mentions")
return {
result = {
"new_mail": new_mail,
"opened_mail": opened_mail,
"active_plans": active_plans,
@@ -75,6 +77,14 @@ def calculate_quick_status(sections: Dict) -> Dict:
"summary": ", ".join(summary_parts) if summary_parts else "All clear"
}
json_handler.log_operation("status_calculated", {
"action_required": action_required,
"new_mail": new_mail,
"active_plans": active_plans,
})
return result
def get_branch_paths() -> List[Path]:
"""
src/aipass/prax/apps/handlers/dashboard/template_differ.py
+9
View File
@@ -26,6 +26,8 @@ import json
from pathlib import Path
from typing import Dict, Any, Optional
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# PATH RESOLUTION
# =============================================================================
@@ -235,6 +237,13 @@ def diff_dashboard_template(branch_name: Optional[str] = None) -> Dict[str, Any]
if status in result["summary"]:
result["summary"][status] += 1
json_handler.log_operation("template_diffed", {
"branch_filter": branch_name,
"branches_scanned": len(result["branches"]),
"needs_update": result["summary"].get("needs_update", 0),
"up_to_date": result["summary"].get("up_to_date", 0),
})
return result
src/aipass/prax/apps/handlers/dashboard/template_pusher.py
+9
View File
@@ -29,6 +29,8 @@ from pathlib import Path
from datetime import datetime
from typing import Dict, Any, List
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# PATH RESOLUTION
# =============================================================================
@@ -385,6 +387,13 @@ def push_dashboard_template(dry_run: bool = False) -> Dict[str, Any]:
if not dry_run and branches_updated_list:
_update_version_file(branches_updated_list)
json_handler.log_operation("template_pushed", {
"dry_run": dry_run,
"branches_scanned": result["branches_scanned"],
"branches_updated": result["branches_updated"],
"branches_created": result["branches_created"],
})
return result
src/aipass/prax/apps/handlers/discovery/filtering.py
+2
View File
@@ -16,6 +16,7 @@ from pathlib import Path
# Import from prax config
from aipass.prax.apps.handlers.config.ignore_patterns import load_ignore_patterns_from_config
from aipass.prax.apps.handlers.json import json_handler
def should_ignore_path(path: Path) -> bool:
"""Check if path should be ignored based on patterns from config
@@ -34,6 +35,7 @@ def should_ignore_path(path: Path) -> bool:
# Check against ignore patterns
for part in path_parts:
if part in ignore_patterns:
json_handler.log_operation("discovery_filtered", {"ignored_path": str(path)})
return True
return False
src/aipass/prax/apps/handlers/discovery/scanner.py
+3
View File
@@ -26,6 +26,7 @@ from aipass.prax.apps.handlers.config.load import (
# Import filtering
from aipass.prax.apps.handlers.discovery.filtering import should_ignore_path
from aipass.prax.apps.handlers.json import json_handler
def scan_directory_safely(directory: Path, modules: Dict, max_depth: int = 10):
"""Safely scan directory with depth limit
@@ -82,4 +83,6 @@ def discover_python_modules() -> Dict[str, Dict[str, Any]]:
# Scan entire ecosystem recursively
scan_directory_safely(ECOSYSTEM_ROOT, modules)
json_handler.log_operation("modules_scanned", {"modules_found": len(modules)})
return modules
src/aipass/prax/apps/handlers/discovery/watcher.py
+2
View File
@@ -36,6 +36,7 @@ from aipass.prax.apps.handlers.registry.save import save_module_registry
# Import filtering
from aipass.prax.apps.handlers.discovery.filtering import should_ignore_path
from aipass.prax.apps.handlers.json import json_handler
# Trigger integration - graceful fallback if trigger not available
try:
@@ -120,6 +121,7 @@ def start_file_watcher():
new_observer.start()
_observer = new_observer
json_handler.log_operation("discovery_watcher_event", {"action": "started", "watch_root": str(ECOSYSTEM_ROOT)})
def stop_file_watcher():
src/aipass/prax/apps/handlers/json/initialize.py
-95
View File
@@ -1,95 +0,0 @@
# =================== AIPass ====================
# Name: initialize.py
# Description: JSON Structure Initialization Handler
# Version: 1.0.0
# Created: 2025-11-07
# Modified: 2026-03-09
# =============================================
"""
JSON Structure Initialization Handler
Initializes complete 3-file JSON structure for a module.
Creates config, data, and log files with standard structure.
Features:
- Create all 3 JSON files at once
- Accept optional initial config and data
- Creates initial log entry
- Returns success status
Usage:
from aipass.prax.apps.handlers.json.initialize import initialize_json_structure
initial_config = {"enabled": True, "max_items": 100}
initial_data = {"count": 0, "last_run": None}
success = initialize_json_structure("my_module", json_dir, initial_config, initial_data)
"""
from pathlib import Path
from typing import Dict, Any, Optional
# Import other JSON handlers
from aipass.prax.apps.handlers.json.save import save_config, save_data
from aipass.prax.apps.handlers.json.log import log_operation
# =============================================
# CONFIGURATION
# =============================================
MODULE_NAME = "json_initialize"
# =============================================
# HANDLER FUNCTION
# =============================================
def initialize_json_structure(
module_name: str,
json_dir: Path,
initial_config: Optional[Dict[str, Any]] = None,
initial_data: Optional[Dict[str, Any]] = None
) -> bool:
"""Initialize complete 3-file JSON structure for a module
Args:
module_name: Name of the module
json_dir: Directory where JSON files are stored
initial_config: Initial config data (optional, uses empty dict if None)
initial_data: Initial data (optional, uses empty dict if None)
Returns:
True if initialization successful, False otherwise
Creates:
- {module_name}_config.json with standard structure
- {module_name}_data.json with standard structure
- {module_name}_log.json with initialization entry
Example:
>>> config = {"enabled": True, "debug": False}
>>> data = {"counter": 0, "items": []}
>>> success = initialize_json_structure("my_module", json_dir, config, data)
>>> if success:
>>> print("Module JSON structure initialized")
"""
try:
# Create config
if initial_config:
save_config(module_name, json_dir, initial_config)
else:
save_config(module_name, json_dir, {})
# Create data
if initial_data:
save_data(module_name, json_dir, initial_data)
else:
save_data(module_name, json_dir, {})
# Create initial log entry
log_operation(module_name, json_dir, "Module initialized", True, {"status": "ready"})
return True
except Exception as e:
return False
src/aipass/prax/apps/handlers/json/load.py
-135
View File
@@ -1,135 +0,0 @@
# =================== AIPass ====================
# Name: load.py
# Description: JSON Load Handler
# Version: 1.0.0
# Created: 2025-11-07
# Modified: 2026-03-09
# =============================================
"""
JSON Load Handler
Universal JSON loading utilities for 3-file pattern (config/data/log).
Provides standardized loading with graceful degradation.
Features:
- Load config files with standard structure
- Load data files with standard structure
- Graceful fallback to defaults on error
- Auto-creates default structure if missing
Usage:
from aipass.prax.apps.handlers.json.load import load_config, load_data
config = load_config("my_module", json_dir)
data = load_data("my_module", json_dir)
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any
# =============================================
# CONFIGURATION
# =============================================
MODULE_NAME = "json_load"
# Standard JSON structures
DEFAULT_CONFIG_STRUCTURE = {
"module_name": "",
"version": "1.0.0",
"timestamp": "",
"config": {}
}
DEFAULT_DATA_STRUCTURE = {
"module_name": "",
"timestamp": "",
"data": {}
}
# =============================================
# HANDLER FUNCTIONS
# =============================================
def load_config(module_name: str, json_dir: Path) -> Dict[str, Any]:
"""Load module config file with standard structure
Args:
module_name: Name of the module (e.g., "branch_create")
json_dir: Directory where JSON files are stored
Returns:
Config dict with standard structure:
{
"module_name": "...",
"version": "1.0.0",
"timestamp": "2025-11-07T...",
"config": {...}
}
Returns default structure if file doesn't exist or on error.
Example:
>>> config = load_config("my_module", Path("/path/to/json"))
>>> settings = config.get("config", {})
"""
config_file = json_dir / f"{module_name}_config.json"
try:
if config_file.exists():
with open(config_file, 'r', encoding='utf-8') as f:
return json.load(f)
else:
# Return default structure if file doesn't exist
default = DEFAULT_CONFIG_STRUCTURE.copy()
default["module_name"] = module_name
default["timestamp"] = datetime.now().isoformat()
return default
except Exception:
default = DEFAULT_CONFIG_STRUCTURE.copy()
default["module_name"] = module_name
default["timestamp"] = datetime.now().isoformat()
return default
def load_data(module_name: str, json_dir: Path) -> Dict[str, Any]:
"""Load module data file with standard structure
Args:
module_name: Name of the module
json_dir: Directory where JSON files are stored
Returns:
Data dict with standard structure:
{
"module_name": "...",
"timestamp": "2025-11-07T...",
"data": {...}
}
Returns default structure if file doesn't exist or on error.
Example:
>>> data = load_data("my_module", Path("/path/to/json"))
>>> runtime_data = data.get("data", {})
"""
data_file = json_dir / f"{module_name}_data.json"
try:
if data_file.exists():
with open(data_file, 'r', encoding='utf-8') as f:
return json.load(f)
else:
# Return default structure
default = DEFAULT_DATA_STRUCTURE.copy()
default["module_name"] = module_name
default["timestamp"] = datetime.now().isoformat()
return default
except Exception:
default = DEFAULT_DATA_STRUCTURE.copy()
default["module_name"] = module_name
default["timestamp"] = datetime.now().isoformat()
return default
src/aipass/prax/apps/handlers/json/log.py
-117
View File
@@ -1,117 +0,0 @@
# =================== AIPass ====================
# Name: log.py
# Description: JSON Log Handler
# Version: 1.0.0
# Created: 2025-11-07
# Modified: 2026-03-09
# =============================================
"""
JSON Log Handler
Universal logging utility for module operations.
Maintains last 100 log entries per module in JSON format.
Features:
- Log module operations to _log.json file
- Keep last 100 entries (newest first)
- Include timestamp, operation, success, details, error
- Graceful error handling
Usage:
from aipass.prax.apps.handlers.json.log import log_operation
log_operation("my_module", json_dir, "save_config", success=True, details="Config saved")
log_operation("my_module", json_dir, "load_data", success=False, error="File not found")
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Any, Optional
# =============================================
# CONFIGURATION
# =============================================
MODULE_NAME = "json_log"
MAX_LOG_ENTRIES = 100
# =============================================
# HANDLER FUNCTION
# =============================================
def log_operation(
module_name: str,
json_dir: Path,
operation: str,
success: bool = True,
details: Any = None,
error: Optional[str] = None
) -> bool:
"""Log module operation to log file
Args:
module_name: Name of the module
json_dir: Directory where JSON files are stored
operation: Operation description (e.g., "save_config", "load_data")
success: Whether operation succeeded (default True)
details: Additional details (any JSON-serializable data)
error: Error message if failed (optional)
Returns:
True if log successful, False otherwise
Log entries are stored as array in {module_name}_log.json:
[
{
"timestamp": "2025-11-07T...",
"operation": "save_config",
"success": true,
"details": "Config saved successfully",
"error": null
},
...
]
Only last 100 entries are kept (newest first).
Example:
>>> log_operation("my_module", json_dir, "save_config", True, "Saved 10 items")
>>> log_operation("my_module", json_dir, "load_data", False, error="File not found")
"""
log_file = json_dir / f"{module_name}_log.json"
try:
# Ensure directory exists
json_dir.mkdir(parents=True, exist_ok=True)
# Load existing log
if log_file.exists():
with open(log_file, 'r', encoding='utf-8') as f:
log_entries = json.load(f)
else:
log_entries = []
# Create new entry
entry = {
"timestamp": datetime.now().isoformat(),
"operation": operation,
"success": success,
"details": details,
"error": error
}
# Add to log (newest first)
log_entries.insert(0, entry)
# Keep only last 100 entries
log_entries = log_entries[:MAX_LOG_ENTRIES]
# Save log
with open(log_file, 'w', encoding='utf-8') as f:
json.dump(log_entries, f, indent=2, ensure_ascii=False)
return True
except Exception:
return False
src/aipass/prax/apps/handlers/json/save.py
-129
View File
@@ -1,129 +0,0 @@
# =================== AIPass ====================
# Name: save.py
# Description: JSON Save Handler
# Version: 1.0.0
# Created: 2025-11-07
# Modified: 2026-03-09
# =============================================
"""
JSON Save Handler
Universal JSON saving utilities for 3-file pattern (config/data/log).
Provides standardized saving with automatic timestamp updates.
Features:
- Save config files with standard structure
- Save data files with standard structure
- Auto-update timestamps
- Create directories if missing
- Graceful error handling
Usage:
from aipass.prax.apps.handlers.json.save import save_config, save_data
save_config("my_module", json_dir, {"setting": "value"})
save_data("my_module", json_dir, {"runtime": "data"})
"""
import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any
# =============================================
# CONFIGURATION
# =============================================
MODULE_NAME = "json_save"
# =============================================
# HANDLER FUNCTIONS
# =============================================
def save_config(module_name: str, json_dir: Path, config_data: Dict[str, Any]) -> bool:
"""Save module config file with standard structure
Args:
module_name: Name of the module
json_dir: Directory where JSON files are stored
config_data: Config data dict (will be wrapped in standard structure)
Returns:
True if successful, False otherwise
The config is saved with this structure:
{
"module_name": "...",
"version": "1.0.0",
"timestamp": "2025-11-07T...",
"config": {...} # your config_data goes here
}
Example:
>>> settings = {"enabled": True, "max_items": 100}
>>> success = save_config("my_module", json_dir, settings)
"""
config_file = json_dir / f"{module_name}_config.json"
try:
# Ensure directory exists
json_dir.mkdir(parents=True, exist_ok=True)
# Wrap data in standard structure
output = {
"module_name": module_name,
"version": config_data.get("version", "1.0.0"),
"timestamp": datetime.now().isoformat(),
"config": config_data
}
with open(config_file, 'w', encoding='utf-8') as f:
json.dump(output, f, indent=2, ensure_ascii=False)
return True
except Exception:
return False
def save_data(module_name: str, json_dir: Path, data: Dict[str, Any]) -> bool:
"""Save module data file with standard structure
Args:
module_name: Name of the module
json_dir: Directory where JSON files are stored
data: Data dict (will be wrapped in standard structure)
Returns:
True if successful, False otherwise
The data is saved with this structure:
{
"module_name": "...",
"timestamp": "2025-11-07T...",
"data": {...} # your data goes here
}
Example:
>>> runtime_data = {"last_run": "2025-11-07", "count": 42}
>>> success = save_data("my_module", json_dir, runtime_data)
"""
data_file = json_dir / f"{module_name}_data.json"
try:
# Ensure directory exists
json_dir.mkdir(parents=True, exist_ok=True)
# Wrap data in standard structure
output = {
"module_name": module_name,
"timestamp": datetime.now().isoformat(),
"data": data
}
with open(data_file, 'w', encoding='utf-8') as f:
json.dump(output, f, indent=2, ensure_ascii=False)
return True
except Exception:
return False
src/aipass/prax/apps/handlers/logging/direct.py
+2
View File
@@ -43,6 +43,7 @@ from aipass.prax.apps.handlers.config.load import (
lines_to_bytes
)
from aipass.prax.apps.handlers.logging.introspection import detect_branch_from_path
from aipass.prax.apps.handlers.json import json_handler
# Use original stdlib getLogger (not the prax-overridden version).
# This is intentional: direct.py creates raw stdlib loggers for
@@ -228,6 +229,7 @@ def get_direct_logger() -> DirectLogger:
DirectLogger instance bound to the caller's module/branch.
"""
module_name, branch_path = _get_direct_caller_info()
json_handler.log_operation("direct_log_created", {"module": module_name, "branch_path": branch_path})
return DirectLogger(module_name, branch_path)
src/aipass/prax/apps/handlers/logging/introspection.py
+5 -1
View File
@@ -16,6 +16,8 @@ Used by logger_setup.py to route logs to correct files.
from pathlib import Path
from typing import Optional
from aipass.prax.apps.handlers.json import json_handler
def get_calling_module() -> str:
"""Detect calling module from stack trace
@@ -99,7 +101,9 @@ def detect_branch_from_path(module_path: str) -> Optional[str]:
relative = path.relative_to(_AIPASS_PKG_ROOT)
# relative is like: flow/apps/module.py → parts[0] = "flow"
if len(relative.parts) >= 2:
return relative.parts[0]
branch = relative.parts[0]
json_handler.log_operation("introspection_resolved", {"module_path": module_path, "branch": branch})
return branch
except ValueError:
pass
src/aipass/prax/apps/handlers/logging/lifecycle.py
+3
View File
@@ -18,6 +18,7 @@ import sys
import time
from typing import Dict, Any, Callable
from aipass.prax.apps.handlers.json import json_handler
from aipass.prax.apps.handlers.logging.setup import (
setup_system_logger,
)
@@ -95,6 +96,8 @@ def run_initialize(module_name: str) -> Dict[str, Any]:
"consolidated_logger": True
})
json_handler.log_operation("lifecycle_event", {"event": "initialized", "modules_count": len(modules)})
return {
"modules_count": len(modules),
"system_logs_dir": str(get_system_logs_dir()),
src/aipass/prax/apps/handlers/logging/log_watchdog.py
+3
View File
@@ -26,6 +26,8 @@ from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Tuple
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# CONSTANTS
@@ -125,6 +127,7 @@ def scan_log_files() -> List[Dict[str, Any]]:
# Sort by line count descending (biggest problems first)
results.sort(key=lambda x: x["lines"], reverse=True)
json_handler.log_operation("log_watchdog_check", {"files_scanned": len(results)})
return results
src/aipass/prax/apps/handlers/logging/monitoring.py
+2
View File
@@ -21,6 +21,7 @@ from pathlib import Path
from typing import Callable, Dict, Any
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.prax.apps.handlers.json import json_handler
def run_monitoring_loop(
@@ -41,6 +42,7 @@ def run_monitoring_loop(
KeyboardInterrupt: When user presses Ctrl+C (caught and handled gracefully)
"""
module_name = "prax_logger"
json_handler.log_operation("monitoring_configured", {"interval": interval, "status_interval": status_interval})
try:
logger.info(f"[{module_name}] Logger capture active - monitoring all modules")
src/aipass/prax/apps/handlers/logging/operations.py
+2
View File
@@ -21,6 +21,7 @@ from typing import Dict, Optional
# Import from prax config
from aipass.prax.apps.handlers.config.load import PRAX_JSON_DIR
from aipass.prax.apps.handlers.logging.direct import get_direct_logger
from aipass.prax.apps.handlers.json import json_handler
_logger = get_direct_logger()
@@ -62,6 +63,7 @@ def log_operation(message: str, data: Optional[Dict] = None):
# Save log
with open(LOG_FILE, 'w', encoding='utf-8') as f:
json.dump(log_entries, f, indent=2, ensure_ascii=False)
json_handler.log_operation("log_operation_performed", {"message": message})
def create_config_file():
"""Create default config file if it doesn't exist"""
src/aipass/prax/apps/handlers/logging/override.py
+3
View File
@@ -31,6 +31,8 @@ from aipass.prax.apps.handlers.logging.setup import setup_individual_logger
# Import introspection
from aipass.prax.apps.handlers.logging.introspection import get_calling_module
from aipass.prax.apps.handlers.json import json_handler
# Store original logging functions for restoration
_original_getLogger = logging.getLogger
_original_basicConfig = logging.basicConfig
@@ -94,6 +96,7 @@ def install_logger_override():
logging.getLogger = enhanced_getLogger
# sys.stderr used intentionally: logging system is being replaced, cannot use prax logger here
sys.stderr.write("[prax] Global logger override installed\n")
json_handler.log_operation("logger_override_applied", {"action": "install"})
def restore_original_logger():
"""Restore original getLogger function
src/aipass/prax/apps/handlers/logging/setup.py
+4
View File
@@ -33,6 +33,8 @@ from aipass.prax.apps.handlers.logging.introspection import (
detect_branch_from_path
)
from aipass.prax.apps.handlers.json import json_handler
# Global state for logging system
_system_logger: Optional[logging.Logger] = None
_captured_loggers: Dict[str, logging.Logger] = {}
@@ -188,6 +190,8 @@ def setup_system_logger() -> logging.Logger:
_system_logger.info("Prax system logger initialized successfully")
_system_logger.info(f"System logger writing to: {system_log_file} + {local_log_file}")
json_handler.log_operation("logger_setup_complete", {"system_log": str(system_log_file), "local_log": str(local_log_file)})
return _system_logger
def get_captured_loggers_count() -> int:
src/aipass/prax/apps/handlers/logging/terminal/filtering.py
+5 -1
View File
@@ -20,6 +20,8 @@ from typing import Set, Optional
# Import from prax config
from aipass.prax.apps.handlers.config.load import PRAX_JSON_DIR
from aipass.prax.apps.handlers.json import json_handler
# Module constants
MODULE_NAME = "prax_terminal"
CONFIG_FILE = PRAX_JSON_DIR / f"{MODULE_NAME}_config.json"
@@ -64,4 +66,6 @@ def should_display_terminal(module_name: str, filtered_modules: Optional[Set[str
if filtered_modules is None:
filtered_modules = load_filtered_modules()
return module_name not in filtered_modules
result = module_name not in filtered_modules
json_handler.log_operation("terminal_filter_applied", {"module": module_name, "displayed": result})
return result
src/aipass/prax/apps/handlers/logging/terminal/formatting.py
+4
View File
@@ -24,6 +24,8 @@ from aipass.prax.apps.handlers.config.load import DEFAULT_LOG_LEVEL
# Import filtering
from aipass.prax.apps.handlers.logging.terminal.filtering import should_display_terminal
from aipass.prax.apps.handlers.json import json_handler
def detect_branch_from_logger_name(logger_name: str) -> Optional[str]:
"""Detect branch from logger name
@@ -112,4 +114,6 @@ def create_terminal_handler() -> logging.StreamHandler:
formatter = TerminalFormatter()
handler.setFormatter(formatter)
json_handler.log_operation("terminal_formatted", {"level": str(DEFAULT_LOG_LEVEL)})
return handler
src/aipass/prax/apps/handlers/monitoring/branch_detector.py
+2
View File
@@ -24,6 +24,7 @@ import logging
from functools import lru_cache
from aipass.prax import logger
from aipass.prax.apps.handlers.json import json_handler
# logger imported from aipass.prax
@@ -47,6 +48,7 @@ class BranchDetector:
self.known_branches: Set[str] = set()
self._repo_root: Optional[Path] = None
self._load_registry()
json_handler.log_operation("branch_detected", {"known_branches": len(self.known_branches)})
def _find_repo_root(self) -> Path:
"""Walk up from this file to find repo root (contains AIPASS_REGISTRY.json)."""
src/aipass/prax/apps/handlers/monitoring/event_queue.py
+4
View File
@@ -16,6 +16,8 @@ from datetime import datetime
from typing import Optional
import threading
from aipass.prax.apps.handlers.json import json_handler
@dataclass(order=True)
class MonitoringEvent:
"""Unified event structure for all monitoring sources"""
@@ -54,6 +56,8 @@ class MonitoringQueue:
if not self.running:
return False
json_handler.log_operation("event_queued", {"event_type": event.event_type, "branch": event.branch})
# Simple deduplication
if not self._is_duplicate(event):
try:
src/aipass/prax/apps/handlers/monitoring/file_watcher_integration.py
+3
View File
@@ -66,6 +66,8 @@ except ImportError as e:
MonitoringEvent = None # type: ignore[assignment, misc]
global_queue = None # type: ignore[assignment]
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# BRANCH REGISTRY LOADER
@@ -247,6 +249,7 @@ class FileWatcherManager:
if self.observer:
self.running = True
json_handler.log_operation("file_watcher_started", {"branches": len(self.branch_paths)})
logger.info("File watcher started successfully")
return True
else:
src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py
+2
View File
@@ -38,6 +38,7 @@ except ImportError:
from aipass.prax.apps.handlers.monitoring.event_queue import MonitoringEvent, MonitoringQueue
from aipass.prax.apps.handlers.monitoring.branch_detector import detect_branch_from_path
from aipass.prax.apps.handlers.monitoring.monitoring_filters import should_monitor, get_priority
from aipass.prax.apps.handlers.json import json_handler
class MonitoringFileHandler(FileSystemEventHandler):
@@ -59,6 +60,7 @@ class MonitoringFileHandler(FileSystemEventHandler):
super().__init__()
self._event_queue = event_queue
self._command_indicator_files = command_indicator_files or {}
json_handler.log_operation("file_change_handled", {"has_queue": event_queue is not None})
# Track last command emitted per file to avoid duplicate separators
self._last_file_command: Dict[str, str] = {}
# Track JSONL file positions for incremental reading
src/aipass/prax/apps/handlers/monitoring/interactive_filter.py
+4
View File
@@ -10,6 +10,8 @@
from typing import List, Tuple, Optional
from aipass.prax.apps.handlers.json import json_handler
def parse_command(cmd: str) -> Tuple[Optional[str], List[str]]:
"""Parse user command into action and arguments.
@@ -30,6 +32,8 @@ def parse_command(cmd: str) -> Tuple[Optional[str], List[str]]:
command = parts[0].lower()
args = parts[1:] if len(parts) > 1 else []
json_handler.log_operation("filter_applied", {"command": command, "args": args})
# Normalize aliases
if command in ['exit', 'q']:
command = 'quit'
src/aipass/prax/apps/handlers/monitoring/log_watcher.py
+3
View File
@@ -46,6 +46,8 @@ except ImportError:
trigger = None # type: ignore[assignment]
HAS_TRIGGER = False
from aipass.prax.apps.handlers.json import json_handler
# Logger
# logger imported from aipass.prax
@@ -521,6 +523,7 @@ def start_log_watcher(event_queue: MonitoringQueue) -> Any:
_log_observer = observer
json_handler.log_operation("log_watcher_started", {"log_dir": str(get_system_logs_dir())})
logger.info(f"Log watcher started, monitoring: {get_system_logs_dir()}")
return observer
src/aipass/prax/apps/handlers/monitoring/module_tracker.py
+3
View File
@@ -12,6 +12,8 @@ from typing import Dict, List, Optional
from datetime import datetime
import re
from aipass.prax.apps.handlers.json import json_handler
class ModuleTracker:
"""Track active modules and their execution"""
@@ -22,6 +24,7 @@ class ModuleTracker:
def track_start(self, module_name: str, command: str, pid: Optional[int] = None):
"""Track module start"""
json_handler.log_operation("module_tracked", {"module": module_name, "command": command})
self.active_modules[module_name] = {
'command': command,
'pid': pid,
src/aipass/prax/apps/handlers/monitoring/monitoring_filters.py
+3
View File
@@ -21,6 +21,8 @@ Based on backup_system/config_handler.py's excellent pattern organization.
from pathlib import Path
from typing import List, Set, Dict, Optional, Any
from aipass.prax.apps.handlers.json import json_handler
# =============================================
# MONITORING PATTERNS
# =============================================
@@ -303,6 +305,7 @@ def should_monitor(path: Path) -> bool:
should_monitor(Path("/home/user/.cache/data")) # False
should_monitor(Path("/home/user/FLOW.id.json")) # True (ALWAYS)
"""
json_handler.log_operation("monitoring_started", {"path": str(path)})
import os
path_str = str(path)
src/aipass/prax/apps/handlers/monitoring/unified_stream.py
+4
View File
@@ -28,6 +28,8 @@ except ImportError:
from rich.console import Console
console = Console()
from aipass.prax.apps.handlers.json import json_handler
# Thread safety
_print_lock = Lock()
@@ -113,6 +115,8 @@ def print_event(event_type: str, branch: str, message: str, level: str = 'info',
level: Event level (info, warning, error, critical)
pid: Optional process ID for the active agent
"""
json_handler.log_operation("stream_output", {"event_type": event_type, "branch": branch, "level": level})
with _print_lock:
# Timestamp
timestamp = datetime.now().strftime("%H:%M:%S")
src/aipass/prax/apps/handlers/registry/ignore.py
-127
View File
@@ -1,127 +0,0 @@
# =================== AIPass ====================
# Name: ignore.py
# Description: Registry Ignore Pattern Handler
# Version: 1.0.0
# Created: 2025-11-06
# Modified: 2026-03-09
# =============================================
"""
Registry Ignore Handler
Manages patterns for excluding files from:
- Branch creation (don't copy template-internal files)
- Registry regeneration (don't track template-internal files)
Loads patterns from registry_ignore.json in template directory.
"""
import json
from pathlib import Path
from typing import List, Set
from fnmatch import fnmatch
def load_ignore_patterns(template_dir: Path) -> dict:
"""
Load ignore patterns from registry_ignore.json
Args:
template_dir: Path to template directory
Returns:
Dict with ignore_files and ignore_patterns lists
Returns empty lists if file not found
"""
ignore_file = template_dir / "registry_ignore.json"
if not ignore_file.exists():
# Return empty patterns if file doesn't exist
return {
"ignore_files": [],
"ignore_patterns": []
}
try:
with open(ignore_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return {
"ignore_files": data.get("ignore_files", []),
"ignore_patterns": data.get("ignore_patterns", [])
}
except Exception:
return {
"ignore_files": [],
"ignore_patterns": []
}
def should_ignore(
file_path: Path,
template_dir: Path,
ignore_files: List[str],
ignore_patterns: List[str]
) -> bool:
"""
Check if a file should be ignored based on patterns
Args:
file_path: Path to file to check
template_dir: Template directory (for relative path calculation)
ignore_files: List of exact filenames to ignore
ignore_patterns: List of glob patterns to ignore
Returns:
True if file should be ignored, False otherwise
"""
filename = file_path.name
# Check exact filename matches
if filename in ignore_files:
return True
# Check glob patterns
for pattern in ignore_patterns:
if fnmatch(filename, pattern):
return True
# Check if any parent directory matches patterns
try:
relative_path = file_path.relative_to(template_dir)
for part in relative_path.parts:
for pattern in ignore_patterns:
if fnmatch(part, pattern):
return True
except ValueError:
# file_path not relative to template_dir
pass
return False
def get_ignored_files(template_dir: Path) -> Set[str]:
"""
Get set of all ignored filenames for quick lookup
Args:
template_dir: Path to template directory
Returns:
Set of filenames to ignore
"""
patterns = load_ignore_patterns(template_dir)
return set(patterns.get("ignore_files", []))
def get_ignore_patterns(template_dir: Path) -> List[str]:
"""
Get list of ignore glob patterns
Args:
template_dir: Path to template directory
Returns:
List of glob patterns to ignore
"""
patterns = load_ignore_patterns(template_dir)
return patterns.get("ignore_patterns", [])
src/aipass/prax/apps/handlers/registry/load.py
+2
View File
@@ -30,6 +30,7 @@ from pathlib import Path
from typing import Dict, Any
from aipass.prax.apps.handlers.config.load import PRAX_ROOT
from aipass.prax.apps.handlers.json import json_handler
# =============================================
# CONFIGURATION
@@ -76,6 +77,7 @@ def load_module_registry() -> Dict[str, Dict[str, Any]]:
with open(REGISTRY_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
modules = data.get('modules', {})
json_handler.log_operation("registry_loaded", {"module_count": len(modules)})
return modules
except Exception:
# Silently return empty dict - logging not available at this level
src/aipass/prax/apps/handlers/registry/meta_ops.py
-360
View File
@@ -1,360 +0,0 @@
# =================== AIPass ====================
# Name: meta_ops.py
# Description: Metadata Operations Handler
# Version: 1.0.0
# Created: 2025-11-04
# Modified: 2026-03-09
# =============================================
"""
Metadata Operations Handler
Functions for branch and template metadata:
- Load template registry
- Load branch metadata
- Generate metadata for existing branches
"""
import json
import hashlib
from pathlib import Path
from datetime import datetime
from typing import Dict, Optional
from aipass.prax import logger
# =============================================================================
# CONSTANTS
# =============================================================================
def _find_repo_root() -> Path:
"""Walk up from this file to find repo root (contains AIPASS_REGISTRY.json)."""
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
if (parent / "AIPASS_REGISTRY.json").exists():
return parent
return Path.cwd()
def _get_template_dir() -> Path:
"""Lazily resolve template directory (package-relative)."""
return _find_repo_root() / "src" / "aipass" / "cortex" / "templates" / "branch_template"
# Files that get renamed during branch creation
FILE_RENAMES = {
"PROJECT.json": "{BRANCHNAME}.json",
"LOCAL..json": "{BRANCHNAME}.local.json",
"OBSERVATIONS.json": "{BRANCHNAME}.observations.json",
"AI_MAIL.json": "{BRANCHNAME}.ai_mail.json",
"BRANCH.ID.json": "{BRANCHNAME}.id.json",
"BRANCH.py": "{branchname}.py",
}
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def calculate_file_hash(file_path: Path) -> str:
"""
Calculate SHA-256 hash of file content
Args:
file_path: Path to file
Returns:
Hex string of file hash (first 12 characters for readability)
"""
if not file_path.is_file():
return ""
try:
sha256 = hashlib.sha256()
with open(file_path, 'rb') as f:
# Read in chunks for large files
for chunk in iter(lambda: f.read(8192), b''):
sha256.update(chunk)
# Return first 12 chars of hash (enough for uniqueness)
return sha256.hexdigest()[:12]
except Exception:
return ""
# =============================================================================
# TEMPLATE REGISTRY OPERATIONS
# =============================================================================
def load_template_registry() -> Optional[Dict]:
"""
Load template_registry.json from template directory
Returns:
Template registry dict or None if not found/error
"""
registry_path = _get_template_dir() / "template_registry.json"
if not registry_path.exists():
return None
try:
with open(registry_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception:
return None
# =============================================================================
# BRANCH METADATA OPERATIONS
# =============================================================================
def load_branch_meta(branch_dir: Path) -> Optional[Dict]:
"""
Load .branch_meta.json from branch directory
Args:
branch_dir: Path to branch directory
Returns:
Branch metadata dict or None if not found/error
"""
meta_path = branch_dir / ".branch_meta.json"
if not meta_path.exists():
# This is normal for old branches that predate ID tracking
return None
try:
with open(meta_path, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
logger.error(f"Loading .branch_meta.json: {e}")
return None
def heal_branch_meta(
branch_dir: Path,
branch_meta: Optional[Dict],
template_registry: Dict,
template_version: str
) -> Optional[Dict]:
"""
Auto-heal branch metadata if format is outdated or missing
Handles:
- Missing .branch_meta.json (returns None - caller should regenerate)
- Old format: name→id mapping (auto-converts to new id→file_info format)
- Corrupted/invalid data (regenerates from scratch)
Args:
branch_dir: Path to branch directory
branch_meta: Loaded metadata (or None if missing)
template_registry: Template registry for regeneration
template_version: Current template version
Returns:
Healed metadata dict, or None if should regenerate from scratch
"""
# If no metadata exists, signal caller to regenerate
if branch_meta is None:
logger.info("No branch_meta - treating all template files as potential additions")
return None
# Check if file_tracking exists and needs healing
if "file_tracking" not in branch_meta:
logger.info("Missing file_tracking - regenerating metadata")
return None
file_tracking = branch_meta.get("file_tracking", {})
if not file_tracking:
# Empty tracking is fine
return branch_meta
# Detect old format: first value is string (name→id) vs dict (id→file_info)
first_value = next(iter(file_tracking.values()))
if isinstance(first_value, str):
# OLD FORMAT DETECTED - Auto-heal to new format AND remap IDs
logger.info("Old branch_meta format detected - auto-healing...")
# Build hash→template_id lookup for ID remapping
hash_to_template_id = {}
for file_id, file_info in template_registry.get("files", {}).items():
if file_info.get("content_hash"):
hash_to_template_id[file_info["content_hash"]] = file_id
# Old format: {"filename.py": "f001"}
# New format: {"f001": {"current_name": "filename.py", "content_hash": "abc123"}}
# Invert the mapping: name→id becomes id→file_info
healed_tracking = {}
id_remapping = {} # Track old_id → new_id for reporting
for filename, old_file_id in file_tracking.items():
# Calculate content hash if file exists
file_path = branch_dir / filename
content_hash = None
if file_path.exists() and file_path.is_file():
content_hash = calculate_file_hash(file_path)
# Try to remap ID using content hash
new_file_id = old_file_id # Default to old ID
# Skip remapping for empty files (hash: e3b0c44298fc = empty file)
# Multiple empty files share same hash, causing false matches
if content_hash and content_hash != "e3b0c44298fc" and content_hash in hash_to_template_id:
new_file_id = hash_to_template_id[content_hash]
if new_file_id != old_file_id:
id_remapping[old_file_id] = new_file_id
healed_tracking[new_file_id] = {
"current_name": filename,
"content_hash": content_hash
}
# Update metadata with healed format
branch_meta["file_tracking"] = healed_tracking
# Save healed version
save_branch_meta(branch_dir, branch_meta)
if id_remapping:
logger.info(f"Branch metadata auto-healed and saved ({len(id_remapping)} IDs remapped)")
else:
logger.info("Branch metadata auto-healed and saved")
return branch_meta
# Format is already correct - but check if IDs need remapping
logger.info("Checking for ID reassignments...")
# Build hash→template_id lookup
hash_to_template_id = {}
for file_id, file_info in template_registry.get("files", {}).items():
if file_info.get("content_hash"):
hash_to_template_id[file_info["content_hash"]] = file_id
# Check each tracked file for ID reassignment
remapped_tracking = {}
id_remapping = {}
for current_id, file_info in file_tracking.items():
content_hash = file_info.get("content_hash")
# Try to remap ID using content hash
new_id = current_id # Default to current ID
# Skip remapping for empty files (hash: e3b0c44298fc = empty file)
# Multiple empty files share same hash, causing false matches
if content_hash and content_hash != "e3b0c44298fc" and content_hash in hash_to_template_id:
new_id = hash_to_template_id[content_hash]
if new_id != current_id:
id_remapping[current_id] = new_id
remapped_tracking[new_id] = file_info
# If IDs were remapped, save updated metadata
if id_remapping:
logger.warning(f"Detected ID reassignments - remapping {len(id_remapping)} files...")
branch_meta["file_tracking"] = remapped_tracking
branch_meta["last_updated"] = datetime.now().isoformat()
save_branch_meta(branch_dir, branch_meta)
logger.info("Branch metadata updated with current template IDs")
else:
logger.info("All IDs current - no remapping needed")
# Format is already correct
return branch_meta
def save_branch_meta(branch_dir: Path, metadata: Dict) -> bool:
"""
Save .branch_meta.json to branch directory
Args:
branch_dir: Path to branch directory
metadata: Metadata dict to save
Returns:
True if successful, False otherwise
"""
meta_path = branch_dir / ".branch_meta.json"
try:
with open(meta_path, 'w', encoding='utf-8') as f:
json.dump(metadata, f, indent=2, ensure_ascii=False)
return True
except Exception as e:
logger.error(f"Saving .branch_meta.json: {e}")
return False
def generate_branch_meta_for_existing_branch(
target_path: Path,
branch_name: str,
template_registry: Dict
) -> Optional[Dict]:
"""
Generate .branch_meta.json for existing branch that predates ID tracking
Scans the branch directory and maps existing files to template IDs with content hashes.
Args:
target_path: Path to existing branch
branch_name: Branch name for placeholder substitution
template_registry: Loaded template registry
Returns:
Dict with metadata structure, or None if failed
"""
# Build reverse lookup: template_filename -> (template_id, file_info)
template_name_to_info = {}
# Map files
for file_id, file_info in template_registry.get("files", {}).items():
template_name_to_info[file_info["current_name"]] = (file_id, file_info)
# Map directories
for dir_id, dir_info in template_registry.get("directories", {}).items():
template_name_to_info[dir_info["current_name"]] = (dir_id, dir_info)
# Build file tracking with new structure: file_id -> {current_name, path, content_hash}
file_tracking = {}
branch_upper = branch_name.upper().replace("-", "_")
# Map files (handle projects placeholder substitution)
for template_name, (template_id, template_info) in template_name_to_info.items():
# Handle placeholder patterns
if "projects" in template_name:
actual_name = template_name.replace("projects", branch_upper)
else:
# Check if this file gets renamed by FILE_RENAMES pattern
if template_name in FILE_RENAMES:
actual_name = FILE_RENAMES[template_name].replace("{BRANCHNAME}", branch_upper)
else:
actual_name = template_name
# Check if file/directory exists in branch
file_path = target_path / actual_name
if file_path.exists():
# Calculate content hash for files (not directories)
content_hash = ""
if file_path.is_file():
content_hash = calculate_file_hash(file_path)
# Use new structure matching template_registry.json
file_tracking[template_id] = {
"current_name": actual_name,
"path": actual_name, # Relative path from branch root
"content_hash": content_hash,
"has_branch_placeholder": "projects" in template_name or "PROJECTS" in template_name
}
# Create metadata structure
meta_data = {
"template_version": template_registry.get("metadata", {}).get("version", "1.0.0"),
"branch_created": "unknown", # Can't determine for existing branches
"last_updated": datetime.now().isoformat(),
"file_tracking": file_tracking
}
return meta_data
src/aipass/prax/apps/handlers/registry/reader.py
-115
View File
@@ -1,115 +0,0 @@
# =================== AIPass ====================
# Name: reader.py
# Description: Branch Registry Reader Handler
# Version: 0.1.0
# Created: 2025-11-15
# Modified: 2026-03-09
# =============================================
"""
Branch Registry Reader Handler
Reads BRANCH_REGISTRY.json and returns branch information.
Used by branch watcher to discover which branches to monitor.
"""
import json
from pathlib import Path
from typing import List, Dict, Any
# =============================================================================
# CONSTANTS
# =============================================================================
def _find_repo_root() -> Path:
"""Walk up from this file to find the repo root (contains AIPASS_REGISTRY.json)."""
current = Path(__file__).resolve().parent
for parent in [current] + list(current.parents):
if (parent / "AIPASS_REGISTRY.json").exists():
return parent
return Path.cwd()
_branch_registry_path_cache: Path | None = None
def _get_branch_registry_path() -> Path:
"""Lazily resolve AIPASS_REGISTRY.json path."""
global _branch_registry_path_cache
if _branch_registry_path_cache is None:
_branch_registry_path_cache = _find_repo_root() / "AIPASS_REGISTRY.json"
return _branch_registry_path_cache
# =============================================================================
# CORE FUNCTIONS
# =============================================================================
def read_registry() -> List[Dict[str, Any]] | None:
"""
Read BRANCH_REGISTRY.json and return list of branches
Returns:
List of branch dictionaries, or None on error
"""
try:
if not _get_branch_registry_path().exists():
return None
with open(_get_branch_registry_path(), 'r', encoding='utf-8') as f:
data = json.load(f)
branches = data.get('branches', [])
return branches
except json.JSONDecodeError:
return None
except Exception:
return None
def get_branch_paths(branch_names: List[str] | None = None) -> List[Path] | None:
"""
Get paths for specified branches (or all branches if None)
Args:
branch_names: List of branch names to get paths for, or None for all
Returns:
List of Path objects, or None on error
"""
branches = read_registry()
if branches is None:
return None
paths = []
for branch in branches:
# Filter by branch names if specified
if branch_names:
if branch.get('name') not in branch_names:
continue
branch_path = branch.get('path')
if branch_path:
paths.append(Path(branch_path))
return paths
def get_branch_info(branch_name: str) -> Dict[str, Any] | None:
"""
Get detailed information for a specific branch
Args:
branch_name: Name of the branch
Returns:
Branch dictionary, or None if not found
"""
branches = read_registry()
if branches is None:
return None
for branch in branches:
if branch.get('name') == branch_name:
return branch
return None
src/aipass/prax/apps/handlers/registry/save.py
+3
View File
@@ -32,6 +32,7 @@ from datetime import datetime, timezone
from typing import Dict, Any
from aipass.prax.apps.handlers.config.load import PRAX_ROOT, ECOSYSTEM_ROOT
from aipass.prax.apps.handlers.json import json_handler
# =============================================
# CONFIGURATION
@@ -92,6 +93,8 @@ def save_module_registry(modules: Dict[str, Dict[str, Any]]) -> bool:
with open(REGISTRY_FILE, 'w', encoding='utf-8') as f:
json.dump(registry_structure, f, indent=2, ensure_ascii=False)
json_handler.log_operation("registry_saved", {"total_modules": len(modules)})
return True
except Exception:
src/aipass/prax/apps/handlers/registry/statistics.py
-97
View File
@@ -1,97 +0,0 @@
# =================== AIPass ====================
# Name: statistics.py
# Description: Registry Statistics Handler
# Version: 1.0.0
# Created: 2025-11-07
# Modified: 2026-03-09
# =============================================
"""
Registry Statistics Handler
Extracts and returns statistics about the Prax module registry.
Returns total module count and other registry metadata.
Features:
- Reads prax_registry.json statistics section
- Returns total_modules count
- Includes registry_exists flag
- Graceful error handling
Usage:
from aipass.prax.apps.handlers.registry.statistics import get_registry_statistics
stats = get_registry_statistics()
print(f"Total modules: {stats['total_modules']}")
"""
import json
from pathlib import Path
from typing import Dict, Any
from aipass.prax.apps.handlers.config.load import PRAX_ROOT
# =============================================
# CONFIGURATION
# =============================================
MODULE_NAME = "statistics"
PRAX_JSON_DIR = PRAX_ROOT / "prax_json"
REGISTRY_FILE = PRAX_JSON_DIR / "prax_registry.json"
# =============================================
# HANDLER FUNCTION
# =============================================
def get_registry_statistics() -> Dict[str, Any]:
"""Get statistics about the module registry
Returns:
Dict containing registry statistics:
{
"total_modules": 42,
"registry_exists": True,
"last_updated": "2025-11-07T...",
"scan_location": "src/aipass"
}
If registry doesn't exist or error occurs:
{
"total_modules": 0,
"registry_exists": False
}
Example:
>>> stats = get_registry_statistics()
>>> if stats["registry_exists"]:
>>> print(f"Registry has {stats['total_modules']} modules")
"""
if not REGISTRY_FILE.exists():
return {
"total_modules": 0,
"registry_exists": False
}
try:
with open(REGISTRY_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
# Extract statistics section if present
stats = data.get('statistics', {})
# Add registry_exists flag
stats['registry_exists'] = True
# Ensure total_modules is present (calculate if missing)
if 'total_modules' not in stats:
stats['total_modules'] = len(data.get('modules', {}))
return stats
except Exception as e:
# Silently return error info - logging not available at this level
return {
"total_modules": 0,
"registry_exists": False,
"error": str(e)
}
src/aipass/prax/apps/handlers/status/sync.py
+3
View File
@@ -26,6 +26,7 @@ from typing import Dict
from aipass.prax.apps.handlers.config.load import _find_repo_root
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.prax.apps.handlers.json import json_handler
# ---------------------------------------------------------------------------
@@ -150,6 +151,8 @@ def sync_status() -> Dict:
"STATUS.md synced: %d branches, %d missing", len(synced), len(missing)
)
json_handler.log_operation("status_synced", {"branches_synced": len(synced), "branches_missing": len(missing)})
return {
"status": "ok",
"branches_synced": synced,
src/aipass/prax/apps/handlers/watcher/monitor.py
+4
View File
@@ -17,6 +17,8 @@ complete visibility into branch modifications.
from pathlib import Path
from typing import List, Callable, Optional, TYPE_CHECKING, Any
from aipass.prax.apps.handlers.json import json_handler
# =============================================================================
# WATCHDOG IMPORT (external dependency)
# =============================================================================
@@ -161,6 +163,8 @@ def start_monitoring(branch_paths: List[tuple], callback: Callable) -> Any:
observer.start()
json_handler.log_operation("watcher_monitor_event", {"branches_monitored": len(branch_paths)})
return observer
src/aipass/prax/apps/handlers/watcher/reporter.py
-114
View File
@@ -1,114 +0,0 @@
# =================== AIPass ====================
# Name: reporter.py
# Description: File Change Reporter Handler
# Version: 0.2.0
# Created: 2025-11-15
# Modified: 2026-03-09
# =============================================
"""
File Change Reporter Handler
Formats and displays file change events in real-time using CLI service provider.
Shows branch name, action type, and file path with Rich formatting.
"""
from pathlib import Path
from datetime import datetime
from aipass.cli.apps.modules import console, header
# =============================================================================
# FORMATTING
# =============================================================================
# Action symbols and colors (Rich markup)
ACTION_FORMAT = {
'CREATED': {'symbol': '+', 'color': 'green'},
'MODIFIED': {'symbol': '~', 'color': 'yellow'},
'DELETED': {'symbol': '-', 'color': 'red'},
'MOVED': {'symbol': '→', 'color': 'blue'}
}
# =============================================================================
# REPORTER FUNCTIONS
# =============================================================================
def format_change(branch_name: str, action: str, file_path: str) -> str:
"""
Format a file change event for display with Rich markup
Args:
branch_name: Name of the branch
action: Type of action (CREATED, MODIFIED, DELETED, MOVED)
file_path: Path to the file
Returns:
Formatted string with Rich markup
"""
timestamp = datetime.now().strftime('%H:%M:%S')
fmt = ACTION_FORMAT.get(action, {'symbol': '?', 'color': 'white'})
symbol = fmt['symbol']
color = fmt['color']
# Make path relative to branch for cleaner output
try:
path_obj = Path(file_path)
relative_path = file_path
for part in path_obj.parts:
if part == 'aipass':
idx = path_obj.parts.index(part)
if idx + 1 < len(path_obj.parts):
relative_path = str(Path(*path_obj.parts[idx+1:]))
break
except Exception:
relative_path = file_path
# Rich markup format
return f"[dim]\\[{timestamp}][/dim] [{color}]\\[{branch_name.upper():12}][/{color}] [{color}]{symbol}[/{color}] {relative_path}"
def report_change(branch_name: str, action: str, file_path: str) -> None:
"""
Report a file change to terminal and log using CLI service
Args:
branch_name: Name of the branch
action: Type of action (CREATED, MODIFIED, DELETED, MOVED)
file_path: Path to the file
"""
formatted = format_change(branch_name, action, file_path)
console.print(formatted)
def print_header() -> None:
"""Print monitoring header using CLI service"""
console.print()
console.print("─" * 80)
console.print("[bold cyan]🔍 BRANCH WATCHER[/bold cyan] [dim]- Real-time File Monitoring[/dim]")
console.print("─" * 80)
console.print()
console.print("[cyan]Legend:[/cyan]")
console.print(" [green]+[/green] CREATED New file")
console.print(" [yellow]~[/yellow] MODIFIED File changed")
console.print(" [red]-[/red] DELETED File removed")
console.print(" [blue]→[/blue] MOVED File renamed/moved")
console.print()
console.print("[dim]Press Ctrl+C to stop monitoring[/dim]")
console.print()
console.print("─" * 80)
console.print()
def print_footer(total_changes: int) -> None:
"""
Print monitoring summary using CLI service
Args:
total_changes: Total number of changes detected
"""
console.print()
console.print("─" * 80)
console.print(f"[cyan]Monitoring stopped.[/cyan] Total changes: [bold]{total_changes}[/bold]")
console.print("─" * 80)
console.print()
src/aipass/prax/apps/modules/agent_status_module.py
+3
View File
@@ -17,6 +17,7 @@ import sys
from typing import List
from aipass.cli.apps.modules import console, error
from aipass.prax.apps.handlers.json import json_handler
def print_introspection():
@@ -81,6 +82,8 @@ def handle_command(command: str, args: List[str]) -> bool:
push_agent_status_dashboard,
)
json_handler.log_operation("agent_status_push_executed", {"args": args})
if '--help' in args:
print_help()
return True
src/aipass/prax/apps/modules/dashboard.py
+2 -2
View File
@@ -66,6 +66,7 @@ from aipass.prax.apps.handlers.dashboard.template_pusher import (
from aipass.prax.apps.handlers.dashboard.template_differ import (
diff_dashboard_template
)
from aipass.prax.apps.handlers.json import json_handler
# ============================================
@@ -391,8 +392,6 @@ def _handle_diff_template(args: List[str]) -> None:
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"]
@@ -467,6 +466,7 @@ def handle_command(command: str, args: List[str]) -> bool:
return True
subcmd = args[0]
json_handler.log_operation("dashboard_command", {"subcommand": subcmd})
if subcmd == "status":
print_status()
src/aipass/prax/apps/modules/discover_module.py
+2
View File
@@ -17,6 +17,7 @@ from typing import List
from aipass.cli.apps.modules import console
from aipass.prax.apps.handlers.discovery.scanner import discover_python_modules
from aipass.prax.apps.handlers.json import json_handler
def print_help():
@@ -57,6 +58,7 @@ def handle_command(command: str, args: List[str]) -> bool:
print_introspection()
return True
json_handler.log_operation("discover_command_executed", {"command": command})
console.print("🔍 Discovering Python modules...")
modules = discover_python_modules()
console.print(f"✅ Discovered {len(modules)} modules")
src/aipass/prax/apps/modules/init_module.py
+2
View File
@@ -17,6 +17,7 @@ from typing import List
from aipass.prax.apps.modules.logger import initialize_logging_system, system_logger as logger
from aipass.cli.apps.modules import console, error
from aipass.prax.apps.handlers.json import json_handler
def print_introspection():
@@ -74,6 +75,7 @@ def handle_command(command: str, args: List[str]) -> bool:
return True
try:
json_handler.log_operation("init_command_executed", {"command": command})
console.print("🚀 Initializing PRAX logging system...")
initialize_logging_system()
console.print("✅ PRAX logging system initialized")
src/aipass/prax/apps/modules/log_audit_module.py
+3
View File
@@ -19,6 +19,7 @@ from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, error
from aipass.prax.apps.handlers.json import json_handler
def print_introspection():
@@ -113,6 +114,8 @@ def handle_command(command: str, args: List[str]) -> bool:
log_health_summary,
)
json_handler.log_operation("log_audit_executed", {"mode": args[0] if args else "audit"})
if args and args[0] == 'enforce':
console.print("\n[bold cyan]Enforcing log limits...[/bold cyan]")
actions = enforce_log_limits()
src/aipass/prax/apps/modules/logger.py
+2
View File
@@ -67,6 +67,7 @@ from aipass.prax.apps.handlers.logging.direct import (
direct_log,
DirectLogger
)
from aipass.prax.apps.handlers.json import json_handler
# Module constants
MODULE_NAME = "prax_logger"
@@ -288,6 +289,7 @@ def handle_command(_command: str, args: list) -> bool:
Logger is a service module with no user-facing commands.
All interaction happens through the system_logger API.
"""
json_handler.log_operation("logger_handle_command", {"args": args})
if not args:
print_introspection()
return True
src/aipass/prax/apps/modules/monitor_module.py
+3
View File
@@ -59,6 +59,8 @@ from aipass.prax.apps.modules.logger import system_logger as logger
# CLI services (display/output formatting)
from aipass.cli.apps.modules import console, header, error
from aipass.prax.apps.handlers.json import json_handler
# Monitoring handlers (connected subsystems)
from aipass.prax.apps.handlers.monitoring import (
print_event, # unified_stream.py
@@ -207,6 +209,7 @@ def handle_command(command: str, args: List[str]) -> bool:
global _monitoring_active, _event_queue, _module_tracker
global _display_thread, _file_watcher_thread, _log_watcher_thread
json_handler.log_operation("monitor_started", {"args": args})
logger.info(f"Starting unified monitoring (args: {args})")
# Initialize monitoring subsystems
src/aipass/prax/apps/modules/run_module.py
+2
View File
@@ -17,6 +17,7 @@ from typing import List
from aipass.prax.apps.modules.logger import start_continuous_logging
from aipass.cli.apps.modules import console
from aipass.prax.apps.handlers.json import json_handler
def print_help():
@@ -53,6 +54,7 @@ def handle_command(command: str, args: List[str]) -> bool:
print_introspection()
return True
json_handler.log_operation("run_command_executed", {"command": command})
console.print("🚀 Starting PRAX continuous logging mode...")
start_continuous_logging()
return True
src/aipass/prax/apps/modules/shutdown_module.py
+2
View File
@@ -17,6 +17,7 @@ from typing import List
from aipass.prax.apps.modules.logger import shutdown_logging_system
from aipass.cli.apps.modules import console
from aipass.prax.apps.handlers.json import json_handler
def print_help():
@@ -53,6 +54,7 @@ def handle_command(command: str, args: List[str]) -> bool:
print_introspection()
return True
json_handler.log_operation("shutdown_command_executed", {"command": command})
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
+3
View File
@@ -18,6 +18,7 @@ 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, success, error, warning
from aipass.prax.apps.handlers.json import json_handler
def print_help():
@@ -48,6 +49,8 @@ def handle_command(command: str, args: List[str]) -> bool:
print_introspection()
return True
json_handler.log_operation("status_checked", {"subcommand": args[0] if args else "default"})
# --- sub-command routing ------------------------------------------------
if args and args[0] in ("--help", "help"):
print_help()
src/aipass/prax/apps/modules/terminal_module.py
+3
View File
@@ -17,6 +17,7 @@ from typing import List
from aipass.prax.apps.modules.logger import enable_terminal_output, disable_terminal_output, system_logger as logger
from aipass.cli.apps.modules import console, error
from aipass.prax.apps.handlers.json import json_handler
def print_introspection():
@@ -93,6 +94,8 @@ def handle_command(command: str, args: List[str]) -> bool:
print_help()
return True # Command was handled, even if validation failed
json_handler.log_operation("terminal_command", {"action": args[0]})
if args[0] == 'enable':
enable_terminal_output()
console.print("✅ Terminal output enabled")