feat: prax external log routing + dashboard template push + memory shebang cleanup
This commit is contained in:
AIOSAI
@@ -1,5 +1,3 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
# ===================AIPASS====================
|
||||
# META DATA HEADER
|
||||
# Name: __init__.py - Symbolic Memory Handler Package
|
||||
|
||||
@@ -496,6 +496,16 @@
|
||||
"standard": "cli_flags",
|
||||
"file": "apps/__init__.py",
|
||||
"reason": "Package marker \u2014 not a CLI entry point. --version is in apps/prax.py."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_logging.py",
|
||||
"standard": "architecture",
|
||||
"reason": "Test files live in tests/ by convention, not in the 3-layer apps/ structure. False positive for test directory."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_logging.py",
|
||||
"standard": "log_structure",
|
||||
"reason": "Test assertions reference _AIPASS_PKG_ROOT which resolves to /home/ paths at runtime. Not log config \u2014 test path constants."
|
||||
}
|
||||
],
|
||||
"notes": {
|
||||
@@ -518,4 +528,4 @@
|
||||
"trigger_integration": "Several handlers optionally import trigger.modules.core for event firing. All have graceful ImportError fallbacks. This is cross-branch integration, not an architectural violation."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -98,6 +98,7 @@ def get_caller_info() -> tuple:
|
||||
|
||||
_AIPASS_PKG_ROOT = Path(__file__).resolve().parents[4] # logging/ → handlers/ → apps/ → prax/ → aipass/
|
||||
_SRC_ROOT = _AIPASS_PKG_ROOT.parent # aipass/ → src/ (contains branches outside aipass namespace)
|
||||
_REPO_ROOT = _SRC_ROOT.parent # src/ → AIPass repo root
|
||||
|
||||
|
||||
def detect_branch_from_path(module_path: str) -> Optional[str]:
|
||||
@@ -151,3 +152,42 @@ def detect_branch_from_path(module_path: str) -> Optional[str]:
|
||||
logger.info("Path %s is not relative to src root", module_path)
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def detect_external_project(module_path: str) -> Optional[tuple]:
|
||||
"""Detect if a module path belongs to an external project (outside AIPass repo).
|
||||
|
||||
Walks up from the module path looking for a project root (.git or pyproject.toml).
|
||||
Returns the project name and root path if found and the path is outside the AIPass repo.
|
||||
|
||||
Returns:
|
||||
(project_name, project_root) or None if path is inside AIPass or unresolvable
|
||||
"""
|
||||
if not module_path:
|
||||
return None
|
||||
|
||||
path = Path(module_path).resolve()
|
||||
|
||||
# If the path is inside the AIPass repo, it's not external
|
||||
try:
|
||||
path.relative_to(_REPO_ROOT)
|
||||
return None
|
||||
except ValueError:
|
||||
logger.info("Path %s is external to AIPass repo", module_path)
|
||||
|
||||
# Also check AIPASS_CALLER_CWD for cross-project dispatch
|
||||
import os
|
||||
|
||||
caller_cwd = os.environ.get("AIPASS_CALLER_CWD")
|
||||
search_path = path if path.exists() else (Path(caller_cwd) if caller_cwd else None)
|
||||
|
||||
if not search_path:
|
||||
return None
|
||||
|
||||
# Walk up to find project root
|
||||
for candidate in [search_path] + list(search_path.parents):
|
||||
if (candidate / ".git").exists() or (candidate / "pyproject.toml").exists():
|
||||
project_name = candidate.name.lower().replace(" ", "_").replace("-", "_")
|
||||
return (project_name, candidate)
|
||||
|
||||
return None
|
||||
|
||||
@@ -31,7 +31,11 @@ from aipass.prax.apps.handlers.config.load import (
|
||||
)
|
||||
|
||||
# Import introspection functions
|
||||
from aipass.prax.apps.handlers.logging.introspection import get_calling_module_path, detect_branch_from_path
|
||||
from aipass.prax.apps.handlers.logging.introspection import (
|
||||
get_calling_module_path,
|
||||
detect_branch_from_path,
|
||||
detect_external_project,
|
||||
)
|
||||
|
||||
from aipass.prax.apps.handlers.json import json_handler
|
||||
|
||||
@@ -126,39 +130,80 @@ def setup_individual_logger(
|
||||
import os
|
||||
|
||||
branch_name: Optional[str] = caller_branch
|
||||
resolved_module_path: Optional[str] = caller_path
|
||||
if not branch_name:
|
||||
module_path = caller_path or get_calling_module_path()
|
||||
branch_path = detect_branch_from_path(module_path) if module_path else None
|
||||
resolved_module_path = caller_path or get_calling_module_path()
|
||||
branch_path = detect_branch_from_path(resolved_module_path) if resolved_module_path else None
|
||||
branch_name = Path(branch_path).name if branch_path else None
|
||||
|
||||
# Environment variable fallback (set by drone for dispatched commands)
|
||||
if not branch_name:
|
||||
branch_name = os.environ.get("AIPASS_BRANCH_NAME") or "unknown_branch"
|
||||
branch_name = os.environ.get("AIPASS_BRANCH_NAME")
|
||||
|
||||
# External project detection: route logs to the caller's own project
|
||||
external_project = None
|
||||
if not branch_name:
|
||||
check_path = resolved_module_path or os.environ.get("AIPASS_CALLER_CWD")
|
||||
if check_path:
|
||||
external_project = detect_external_project(check_path)
|
||||
|
||||
if not branch_name and not external_project:
|
||||
branch_name = "unknown_branch"
|
||||
|
||||
# Create formatter (shared by all handlers)
|
||||
formatter = logging.Formatter(log_config["log_format"], log_config["date_format"])
|
||||
|
||||
# HANDLER 1: System-wide log (central aggregation)
|
||||
system_log_file = get_system_logs_dir() / f"{branch_name}_{module_name}.log"
|
||||
system_limits = log_config["system_logs"]
|
||||
system_max_bytes = lines_to_bytes(system_limits["max_lines"])
|
||||
system_handler = _safe_rotating_handler(system_log_file, system_max_bytes, system_limits["backup_count"])
|
||||
system_handler.setFormatter(formatter)
|
||||
logger.addHandler(system_handler)
|
||||
if external_project:
|
||||
# EXTERNAL PROJECT: route both handlers to the caller's project directory
|
||||
project_name, project_root = external_project
|
||||
ext_system_logs = project_root / "system_logs"
|
||||
ext_system_logs.mkdir(parents=True, exist_ok=True)
|
||||
ext_local_logs = project_root / "logs"
|
||||
ext_local_logs.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 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"])
|
||||
local_handler.setFormatter(formatter)
|
||||
logger.addHandler(local_handler)
|
||||
# HANDLER 1: Project system log
|
||||
system_log_file = ext_system_logs / f"{project_name}_{module_name}.log"
|
||||
system_limits = log_config["system_logs"]
|
||||
system_max_bytes = lines_to_bytes(system_limits["max_lines"])
|
||||
system_handler = _safe_rotating_handler(system_log_file, system_max_bytes, system_limits["backup_count"])
|
||||
system_handler.setFormatter(formatter)
|
||||
logger.addHandler(system_handler)
|
||||
|
||||
if _system_logger:
|
||||
_system_logger.info(
|
||||
f"Logger created for {module_name} → system: {system_log_file} ({system_limits['max_lines']} lines), local: {module_log_file} ({local_limits['max_lines']} lines)"
|
||||
)
|
||||
# HANDLER 2: Project local log
|
||||
module_log_file = ext_local_logs / 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"])
|
||||
local_handler.setFormatter(formatter)
|
||||
logger.addHandler(local_handler)
|
||||
|
||||
if _system_logger:
|
||||
_system_logger.info(
|
||||
f"Logger created for {module_name} → external project '{project_name}': {system_log_file}, {module_log_file}"
|
||||
)
|
||||
else:
|
||||
# INTERNAL: route to AIPass system_logs + branch-local logs
|
||||
# HANDLER 1: System-wide log (central aggregation)
|
||||
system_log_file = get_system_logs_dir() / f"{branch_name}_{module_name}.log"
|
||||
system_limits = log_config["system_logs"]
|
||||
system_max_bytes = lines_to_bytes(system_limits["max_lines"])
|
||||
system_handler = _safe_rotating_handler(system_log_file, system_max_bytes, system_limits["backup_count"])
|
||||
system_handler.setFormatter(formatter)
|
||||
logger.addHandler(system_handler)
|
||||
|
||||
# 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"])
|
||||
local_handler.setFormatter(formatter)
|
||||
logger.addHandler(local_handler)
|
||||
|
||||
if _system_logger:
|
||||
_system_logger.info(
|
||||
f"Logger created for {module_name} → system: {system_log_file} ({system_limits['max_lines']} lines), local: {module_log_file} ({local_limits['max_lines']} lines)"
|
||||
)
|
||||
|
||||
# HANDLER 3: Terminal output (if enabled)
|
||||
if _terminal_output_enabled and _terminal_module_available:
|
||||
|
||||
@@ -10,7 +10,7 @@
|
||||
"description": "Branch dashboard template - v3 schema with write-through sections"
|
||||
}
|
||||
},
|
||||
"last_push": "2026-05-10 00:00:07",
|
||||
"last_push": "2026-05-16 21:34:52",
|
||||
"last_push_branches": [
|
||||
"AI_MAIL",
|
||||
"AIPASS",
|
||||
|
||||
@@ -142,6 +142,150 @@ class TestDetectBranchFromPath:
|
||||
assert detect_branch_from_path(deep_path) == "flow"
|
||||
|
||||
|
||||
# =============================================
|
||||
# detect_external_project
|
||||
# =============================================
|
||||
|
||||
|
||||
class TestDetectExternalProject:
|
||||
"""Tests for detect_external_project() — identifies external project paths."""
|
||||
|
||||
def test_none_for_empty_path(self, mock_prax_infrastructure):
|
||||
"""Returns None for empty string input."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import detect_external_project
|
||||
|
||||
assert detect_external_project("") is None
|
||||
|
||||
def test_none_for_internal_aipass_path(self, mock_prax_infrastructure):
|
||||
"""Returns None for paths inside the AIPass repo."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import (
|
||||
detect_external_project,
|
||||
_REPO_ROOT,
|
||||
)
|
||||
|
||||
internal_path = str(_REPO_ROOT / "src" / "aipass" / "flow" / "apps" / "handler.py")
|
||||
assert detect_external_project(internal_path) is None
|
||||
|
||||
def test_detects_external_project_with_git(self, mock_prax_infrastructure, tmp_path):
|
||||
"""Detects project root via .git directory for external paths."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import detect_external_project
|
||||
|
||||
# Create a fake external project with .git
|
||||
project_dir = tmp_path / "MyProject"
|
||||
project_dir.mkdir()
|
||||
(project_dir / ".git").mkdir()
|
||||
module_file = project_dir / "src" / "module.py"
|
||||
module_file.parent.mkdir(parents=True)
|
||||
module_file.touch()
|
||||
|
||||
result = detect_external_project(str(module_file))
|
||||
assert result is not None
|
||||
project_name, project_root = result
|
||||
assert project_name == "myproject"
|
||||
assert project_root == project_dir
|
||||
|
||||
def test_detects_external_project_with_pyproject(self, mock_prax_infrastructure, tmp_path):
|
||||
"""Detects project root via pyproject.toml for external paths."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import detect_external_project
|
||||
|
||||
project_dir = tmp_path / "Compass"
|
||||
project_dir.mkdir()
|
||||
(project_dir / "pyproject.toml").touch()
|
||||
module_file = project_dir / "nav" / "router.py"
|
||||
module_file.parent.mkdir(parents=True)
|
||||
module_file.touch()
|
||||
|
||||
result = detect_external_project(str(module_file))
|
||||
assert result is not None
|
||||
project_name, project_root = result
|
||||
assert project_name == "compass"
|
||||
assert project_root == project_dir
|
||||
|
||||
def test_normalizes_project_name(self, mock_prax_infrastructure, tmp_path):
|
||||
"""Project names are lowercased with hyphens/spaces converted to underscores."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import detect_external_project
|
||||
|
||||
project_dir = tmp_path / "Backup-System"
|
||||
project_dir.mkdir()
|
||||
(project_dir / ".git").mkdir()
|
||||
module_file = project_dir / "main.py"
|
||||
module_file.touch()
|
||||
|
||||
result = detect_external_project(str(module_file))
|
||||
assert result is not None
|
||||
assert result[0] == "backup_system"
|
||||
|
||||
def test_none_when_no_project_root_found(self, mock_prax_infrastructure, tmp_path):
|
||||
"""Returns None when no .git or pyproject.toml found walking up."""
|
||||
from aipass.prax.apps.handlers.logging.introspection import detect_external_project
|
||||
|
||||
# Create a file in tmp_path without .git or pyproject.toml
|
||||
# tmp_path itself might have parents with .git — use a nested isolated dir
|
||||
isolated = tmp_path / "no_project" / "deep" / "nested"
|
||||
isolated.mkdir(parents=True)
|
||||
module_file = isolated / "orphan.py"
|
||||
module_file.touch()
|
||||
|
||||
# tmp_path's parents may have .git, so this test may find one
|
||||
# The key assertion is: it doesn't return an AIPass-internal result
|
||||
result = detect_external_project(str(module_file))
|
||||
if result is not None:
|
||||
_, root = result
|
||||
# Should NOT be the AIPass repo root
|
||||
from aipass.prax.apps.handlers.logging.introspection import _REPO_ROOT
|
||||
|
||||
assert root != _REPO_ROOT
|
||||
|
||||
|
||||
# =============================================
|
||||
# setup_individual_logger external routing
|
||||
# =============================================
|
||||
|
||||
|
||||
class TestSetupExternalRouting:
|
||||
"""Tests for external project log routing in setup_individual_logger."""
|
||||
|
||||
def test_external_project_routes_to_project_system_logs(self, mock_prax_infrastructure, tmp_path):
|
||||
"""External callers get logs routed to their project's system_logs/ dir."""
|
||||
from aipass.prax.apps.handlers.logging import setup
|
||||
|
||||
project_dir = tmp_path / "AIPL"
|
||||
project_dir.mkdir()
|
||||
(project_dir / ".git").mkdir()
|
||||
fake_module = str(project_dir / "src" / "polyglot.py")
|
||||
env_patch = {"AIPASS_BRANCH_NAME": "", "AIPASS_CALLER_CWD": ""}
|
||||
|
||||
with patch.object(setup, "detect_external_project", return_value=("aipl", project_dir)):
|
||||
with patch.object(setup, "detect_branch_from_path", return_value=None):
|
||||
with patch.dict("os.environ", env_patch):
|
||||
setup._captured_loggers.clear()
|
||||
result = setup.setup_individual_logger("polyglot", caller_path=fake_module)
|
||||
|
||||
assert result is not None
|
||||
assert (project_dir / "system_logs").exists()
|
||||
assert (project_dir / "logs").exists()
|
||||
|
||||
def test_external_project_does_not_write_to_aipass_system_logs(self, mock_prax_infrastructure, tmp_path):
|
||||
"""External routing must NOT create files in AIPass system_logs."""
|
||||
from aipass.prax.apps.handlers.logging import setup
|
||||
|
||||
project_dir = tmp_path / "Compass"
|
||||
project_dir.mkdir()
|
||||
(project_dir / ".git").mkdir()
|
||||
fake_module = str(project_dir / "nav.py")
|
||||
get_calls = []
|
||||
env_patch = {"AIPASS_BRANCH_NAME": "", "AIPASS_CALLER_CWD": ""}
|
||||
|
||||
with patch.object(setup, "detect_external_project", return_value=("compass", project_dir)):
|
||||
with patch.object(setup, "detect_branch_from_path", return_value=None):
|
||||
with patch.dict("os.environ", env_patch):
|
||||
with patch.object(setup, "get_system_logs_dir", side_effect=lambda: get_calls.append(1)):
|
||||
setup._captured_loggers.clear()
|
||||
setup.setup_individual_logger("navigator", caller_path=fake_module)
|
||||
|
||||
assert len(get_calls) == 0
|
||||
|
||||
|
||||
# =============================================
|
||||
# get_caller_info
|
||||
# =============================================
|
||||
@@ -234,6 +378,7 @@ class TestLinesToBytes:
|
||||
mock_config_mod = MagicMock()
|
||||
|
||||
def real_lines_to_bytes(num_lines: int, avg_line_length: int = 200) -> int:
|
||||
"""Convert line count to byte estimate for test assertions."""
|
||||
return num_lines * avg_line_length
|
||||
|
||||
mock_config_mod.lines_to_bytes = real_lines_to_bytes
|
||||
|
||||
Reference in New Issue
Block a user