Merge pull request #271 from AIOSAI/fix/prax-aipl-log-leak-dplan-0125-track-g

fix(prax): AIPL polyglot log leak into unknown_branch — path resolver fix (DPLAN-0125 Track G)
This commit is contained in:
AIPass
2026-04-14 19:59:59 -07:00
committed by GitHub
3 changed files with 95 additions and 13 deletions
+11 -1
View File
@@ -3,9 +3,19 @@
"version": "2.0.0",
"created": "2026-03-07T22:43:24.315842",
"description": "Standards bypass configuration for prax branch",
"last_updated": "2026-03-24"
"last_updated": "2026-04-14"
},
"bypass": [
{
"file": "tests/test_config.py",
"standard": "documentation",
"reason": "Test functions follow pytest naming conventions (test_*). Docstrings on test functions are non-standard in this codebase — only new tests added in DPLAN-0125 Track G have them. Pre-existing 36 functions comply with project test style."
},
{
"file": "tests/test_config.py",
"standard": "architecture",
"reason": "Test files live in tests/ by convention, not in the 3-layer apps/ structure. This is a false positive for the test directory."
},
{
"file": "apps/modules/logger.py",
"standard": "cli",
+57 -9
View File
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: load.py
# Description: Load Logging Configuration Handler
# Version: 1.0.1
# Version: 1.0.2
# Created: 2025-11-07
# Modified: 2026-03-09
# Modified: 2026-04-14
# =============================================
"""
@@ -27,12 +27,13 @@ Usage:
max_lines = system_logs['max_lines']
"""
import inspect
import json
import logging
import os
logger = logging.getLogger(__name__)
from pathlib import Path
from typing import Dict, Any
from typing import Dict, Any, Optional
from aipass.prax.apps.handlers.json import json_handler
@@ -78,26 +79,53 @@ def get_system_logs_dir() -> Path:
return _system_logs_dir_cache
def get_module_logs_dir(module_name: str) -> Path:
def _warn_routing(module_name: str, destination: object) -> None:
"""Log routing warning when a module's log path falls outside ECOSYSTEM_ROOT."""
try:
from aipass.prax.apps.modules.logger import get_direct_logger
get_direct_logger().warning(
"[get_module_logs_dir] '%s' not in ECOSYSTEM_ROOT; routing to %s",
module_name, destination,
)
except Exception as e:
logger.warning(
"[get_module_logs_dir] '%s' routing to %s (logger unavailable: %s)",
module_name, destination, e,
)
def get_module_logs_dir(module_name: Optional[str] = None) -> Path:
"""Get the branch-root logs directory for a module.
Checks ECOSYSTEM_ROOT (src/aipass/) first, then SRC_ROOT (src/) for
branches that live outside the aipass namespace (e.g., commons).
branches that live outside the aipass namespace (e.g., commons). For
cross-project dispatch, resolves paths relative to the caller's project
root via AIPASS_CALLER_CWD (set by drone, DPLAN-0121) instead of
ECOSYSTEM_ROOT. Falls back to system_logs/external/ for unknown modules —
never creates new directories inside the AIPass source tree.
This is the primary local log directory resolver for the two-tier
model (system_logs/ for central aggregation + branch-root logs/
for local debugging).
Args:
module_name: Module name (e.g., "flow", "prax", "commons")
module_name: Module name (e.g., "flow", "prax", "commons").
Auto-detected from the calling module if not provided.
Returns:
Path to the module's branch-root logs directory
Path to the module's logs directory
"""
# Auto-detect caller module name when not provided
if module_name is None:
frame = inspect.stack()[1]
module_name = Path(frame.filename).stem
test_log_dir = os.environ.get("AIPASS_TEST_LOG_DIR")
if test_log_dir:
p = Path(test_log_dir) / module_name
p.mkdir(parents=True, exist_ok=True)
return p
# Standard: src/aipass/{module}/logs
branch_dir = ECOSYSTEM_ROOT / module_name
if branch_dir.exists():
@@ -113,9 +141,29 @@ def get_module_logs_dir(module_name: str) -> Path:
logs_dir.mkdir(parents=True, exist_ok=True)
return logs_dir
# Default: create under ECOSYSTEM_ROOT (original behavior for new branches)
logs_dir = branch_dir / "logs"
# Cross-project dispatch: AIPASS_CALLER_CWD is set by drone router_handler
# (DPLAN-0121). Walk up from the caller's CWD to find the project root
# (.git or pyproject.toml), then log there rather than polluting ECOSYSTEM_ROOT
# with directories for unknown/external modules (e.g. AIPL polyglot agents).
caller_cwd = os.environ.get("AIPASS_CALLER_CWD")
if caller_cwd:
caller_path = Path(caller_cwd)
project_root = next(
(c for c in [caller_path, *caller_path.parents]
if (c / ".git").exists() or (c / "pyproject.toml").exists()),
None,
)
if project_root:
logs_dir = project_root / "logs" / module_name
logs_dir.mkdir(parents=True, exist_ok=True)
_warn_routing(module_name, logs_dir)
return logs_dir
# Final safe fallback: system_logs/external/ — never create unknown directories
# inside the AIPass source tree. Fixes AIPL polyglot log leak (DPLAN-0125 Track G).
logs_dir = get_system_logs_dir() / "external" / module_name
logs_dir.mkdir(parents=True, exist_ok=True)
_warn_routing(module_name, "system_logs/external/")
return logs_dir
# Config file
+27 -3
View File
@@ -149,12 +149,36 @@ class TestGetModuleLogsDir:
assert result.name == "logs"
assert result.exists()
def test_creates_logs_dir_for_new_branch(self, mock_prax_infrastructure, monkeypatch, tmp_path):
def test_unknown_module_routes_to_system_logs_external(self, mock_prax_infrastructure, monkeypatch, tmp_path):
"""Unknown modules must NOT create dirs in ECOSYSTEM_ROOT (log-leak regression)."""
load_mod = _fresh_import_load(monkeypatch, tmp_path)
result = load_mod.get_module_logs_dir("newbranch")
monkeypatch.setattr(load_mod, "_find_repo_root", lambda: tmp_path)
monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False)
result = load_mod.get_module_logs_dir("unknown_branch")
# Must route to system_logs/external/, NOT create src/aipass/unknown_branch/
assert result == tmp_path / "system_logs" / "external" / "unknown_branch"
assert result.exists()
assert result.is_dir()
assert result.name == "logs"
# Verify ECOSYSTEM_ROOT is not polluted
assert not (tmp_path / "unknown_branch").exists()
def test_aipass_caller_cwd_routes_to_caller_project(self, mock_prax_infrastructure, monkeypatch, tmp_path):
"""Regression: AIPASS_CALLER_CWD directs logs to caller project root, not ECOSYSTEM_ROOT."""
load_mod = _fresh_import_load(monkeypatch, tmp_path)
monkeypatch.setattr(load_mod, "_find_repo_root", lambda: tmp_path)
# Set up a mock caller project with a .git marker
caller_project = tmp_path / "caller_project"
caller_project.mkdir()
(caller_project / ".git").mkdir()
caller_cwd = str(caller_project / "src" / "polyglot")
monkeypatch.setenv("AIPASS_CALLER_CWD", caller_cwd)
result = load_mod.get_module_logs_dir("polyglot")
# Must resolve to caller project's logs/, not ECOSYSTEM_ROOT
assert result == caller_project / "logs" / "polyglot"
assert result.exists()
assert result.is_dir()
# ECOSYSTEM_ROOT must not be polluted
assert not (tmp_path / "polyglot").exists()
# =============================================