diff --git a/src/aipass/memory/apps/handlers/symbolic/__init__.py b/src/aipass/memory/apps/handlers/symbolic/__init__.py index a4b26e00..98ef0721 100644 --- a/src/aipass/memory/apps/handlers/symbolic/__init__.py +++ b/src/aipass/memory/apps/handlers/symbolic/__init__.py @@ -1,5 +1,3 @@ -#!/usr/bin/env python3 - # ===================AIPASS==================== # META DATA HEADER # Name: __init__.py - Symbolic Memory Handler Package diff --git a/src/aipass/prax/.seedgo/bypass.json b/src/aipass/prax/.seedgo/bypass.json index c7c5a616..a35312f5 100644 --- a/src/aipass/prax/.seedgo/bypass.json +++ b/src/aipass/prax/.seedgo/bypass.json @@ -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." } } -} +} \ No newline at end of file diff --git a/src/aipass/prax/apps/handlers/logging/introspection.py b/src/aipass/prax/apps/handlers/logging/introspection.py index 0f63c009..2765f611 100755 --- a/src/aipass/prax/apps/handlers/logging/introspection.py +++ b/src/aipass/prax/apps/handlers/logging/introspection.py @@ -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 diff --git a/src/aipass/prax/apps/handlers/logging/setup.py b/src/aipass/prax/apps/handlers/logging/setup.py index bcda2b5d..54867b51 100755 --- a/src/aipass/prax/apps/handlers/logging/setup.py +++ b/src/aipass/prax/apps/handlers/logging/setup.py @@ -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: diff --git a/src/aipass/prax/templates/.dashboard_version.json b/src/aipass/prax/templates/.dashboard_version.json index 631b7223..40236e44 100644 --- a/src/aipass/prax/templates/.dashboard_version.json +++ b/src/aipass/prax/templates/.dashboard_version.json @@ -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", diff --git a/src/aipass/prax/tests/test_logging.py b/src/aipass/prax/tests/test_logging.py index 815deef4..ae227eea 100644 --- a/src/aipass/prax/tests/test_logging.py +++ b/src/aipass/prax/tests/test_logging.py @@ -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