diff --git a/src/aipass/prax/__init__.py b/src/aipass/prax/__init__.py index 2e63903a..d63f693d 100644 --- a/src/aipass/prax/__init__.py +++ b/src/aipass/prax/__init__.py @@ -1,5 +1,10 @@ """Prax - Monitoring and logging for AIPass.""" +try: + from aipass.prax.apps.modules.logger import append_jsonl +except Exception: + append_jsonl = None # type: ignore[assignment] + try: from aipass.prax.apps.modules.logger import system_logger as logger except Exception: @@ -7,7 +12,7 @@ except Exception: # Provides no-op info/warning/error so callers keep running. import logging as _logging - class _NullLogger: + class NullLogger: """Fallback logger when prax SystemLogger fails to import.""" def __init__(self): @@ -25,4 +30,4 @@ except Exception: def error(self, message, *args, **kwargs): self._logger.error(message, *args, **kwargs) - logger = _NullLogger() + logger = NullLogger() diff --git a/src/aipass/prax/apps/handlers/logging/jsonl_writer.py b/src/aipass/prax/apps/handlers/logging/jsonl_writer.py new file mode 100644 index 00000000..9724e19c --- /dev/null +++ b/src/aipass/prax/apps/handlers/logging/jsonl_writer.py @@ -0,0 +1,104 @@ +# =================== AIPass ==================== +# Name: jsonl_writer.py +# Description: JSONL append with size-based rotation +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= + +""" +PRAX JSONL Writer + +Sanctioned path for structured JSONL appending with size-based rotation. + +Standalone — zero dependency on prax's logging pipeline, event system, or +stack introspection. Safe to call from any branch, including those where +importing the full prax logger would cause import recursion (e.g. @trigger +event handlers). + +Usage (from any branch): + from aipass.prax.apps.handlers.logging.jsonl_writer import append_jsonl + + append_jsonl(Path("logs/operations.jsonl"), {"op": "backup", "files": 42}) + +Or via the package shortcut: + from aipass.prax import append_jsonl +""" + +import json +import logging +import os +from pathlib import Path +from typing import Any, Dict, Union + +from aipass.prax.apps.handlers.json import json_handler + +logger = logging.getLogger(__name__) + +JSONL_MAX_BYTES = 500_000 # 500 KB per file +JSONL_BACKUP_COUNT = 1 + + +def append_jsonl( + filepath: Union[str, Path], + data: Dict[str, Any], + *, + max_bytes: int = JSONL_MAX_BYTES, + backup_count: int = JSONL_BACKUP_COUNT, +) -> None: + """Append a JSON object as a single line, rotating when the file exceeds max_bytes. + + Rotation: when the file reaches max_bytes, rename it to .1 (overwriting any + previous .1) and start fresh. Only 1 backup is kept by default — matching + prax's RotatingFileHandler behavior. + + Auto-creates parent directories if missing. + """ + filepath = Path(filepath) + filepath.parent.mkdir(parents=True, exist_ok=True) + + _maybe_rotate(filepath, max_bytes, backup_count) + + line = json.dumps(data, default=str, ensure_ascii=False) + "\n" + with open(filepath, "a", encoding="utf-8") as f: + f.write(line) + + json_handler.log_operation("jsonl_append", {"file": str(filepath)}) + + +def _rotate_with_backup(filepath: Path) -> None: + """Rename file to .1 backup; fall back to unlink on failure.""" + backup = filepath.parent / f"{filepath.name}.1" + try: + os.replace(str(filepath), str(backup)) + except OSError as exc: + logger.warning("JSONL rotation rename failed for %s: %s — unlinking instead", filepath, exc) + _unlink_safe(filepath) + + +def _unlink_safe(filepath: Path) -> None: + """Remove file, logging on failure.""" + try: + filepath.unlink() + except OSError as exc: + logger.warning("Failed to unlink oversized JSONL %s: %s", filepath, exc) + + +def _maybe_rotate(filepath: Path, max_bytes: int, backup_count: int) -> None: + """Rotate the file if it exceeds max_bytes.""" + if not filepath.exists(): + return + + try: + size = filepath.stat().st_size + except OSError as exc: + logger.warning("Cannot stat %s for rotation check: %s", filepath, exc) + return + + if size < max_bytes: + return + + if backup_count >= 1: + _rotate_with_backup(filepath) + else: + _unlink_safe(filepath) diff --git a/src/aipass/prax/apps/handlers/logging/log_watchdog.py b/src/aipass/prax/apps/handlers/logging/log_watchdog.py index d612b031..e4cf1af3 100644 --- a/src/aipass/prax/apps/handlers/logging/log_watchdog.py +++ b/src/aipass/prax/apps/handlers/logging/log_watchdog.py @@ -28,6 +28,7 @@ Two modes: import logging import sys +import time from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Tuple @@ -72,6 +73,9 @@ BRANCH_WARN_SIZE_MB = 1.0 BRANCH_CRITICAL_SIZE_MB = 10.0 BRANCH_DEFAULT_MAX_LINES = 5000 +# Sweep — stale log cleanup +SWEEP_MAX_AGE_DAYS = 30 + # ============================================================================= # SCANNING @@ -416,6 +420,80 @@ def branch_log_health_summary() -> Dict[str, Any]: } +# ============================================================================= +# SWEEP — stale log cleanup (30-day policy) +# ============================================================================= + + +def _file_age_days(filepath: Path) -> float: + """Return file age in days based on mtime.""" + try: + mtime = filepath.stat().st_mtime + except OSError as exc: + logger.warning("Cannot stat %s for age check: %s", filepath, exc) + return 0.0 + return (time.time() - mtime) / 86400.0 + + +def _sweep_directory(directory: Path, patterns: List[str], max_age_days: int) -> List[Dict[str, Any]]: + """Delete files matching patterns that are older than max_age_days.""" + removed: List[Dict[str, Any]] = [] + if not directory.exists(): + return removed + + for pattern in patterns: + for filepath in sorted(directory.glob(pattern)): + age = _file_age_days(filepath) + if age < max_age_days: + continue + size_kb = _get_file_size_kb(filepath) + try: + filepath.unlink() + removed.append( + { + "path": str(filepath), + "name": filepath.name, + "age_days": round(age, 1), + "size_kb": round(size_kb, 1), + } + ) + except OSError as exc: + logger.warning("Sweep failed to delete %s: %s", filepath, exc) + + return removed + + +def sweep_stale_logs(max_age_days: int = SWEEP_MAX_AGE_DAYS) -> Dict[str, Any]: + """Delete log files older than max_age_days across system_logs/ and branch logs/. + + Scans system_logs/ for *.log and *.log.1, and all branch logs/ directories + for *.log, *.log.1, *.jsonl, and *.jsonl.1. + + Returns a summary with counts and the list of removed files. + """ + all_removed: List[Dict[str, Any]] = [] + + system_patterns = ["*.log", "*.log.1"] + all_removed.extend(_sweep_directory(_get_system_logs_dir(), system_patterns, max_age_days)) + + branch_patterns = ["*.log", "*.log.1", "*.jsonl", "*.jsonl.1"] + eco_root = _get_ecosystem_root() + if eco_root.exists(): + for branch_dir in sorted(eco_root.iterdir()): + logs_dir = branch_dir / "logs" + if branch_dir.is_dir() and logs_dir.is_dir(): + all_removed.extend(_sweep_directory(logs_dir, branch_patterns, max_age_days)) + + json_handler.log_operation("log_sweep", {"max_age_days": max_age_days, "files_removed": len(all_removed)}) + + return { + "max_age_days": max_age_days, + "files_removed": len(all_removed), + "total_reclaimed_kb": round(sum(f["size_kb"] for f in all_removed), 1), + "removed": all_removed, + } + + # ============================================================================= # CLI ENTRY POINT (for testing) # ============================================================================= diff --git a/src/aipass/prax/apps/modules/log_audit.py b/src/aipass/prax/apps/modules/log_audit.py index 4046c8f1..26f017cf 100644 --- a/src/aipass/prax/apps/modules/log_audit.py +++ b/src/aipass/prax/apps/modules/log_audit.py @@ -63,6 +63,7 @@ def print_help(): console.print() console.print(" [cyan]audit[/cyan] Show log health summary + any oversized files") console.print(" [cyan]enforce[/cyan] Truncate all oversized files to 1000 lines") + console.print(" [cyan]sweep[/cyan] Delete log files older than 30 days") console.print() console.print("[yellow]Usage:[/yellow]") console.print() @@ -72,6 +73,9 @@ def print_help(): console.print(" [dim]# Truncate all oversized files to 1000 lines[/dim]") console.print(" $ drone @prax log-audit enforce") console.print() + console.print(" [dim]# Delete log files older than 30 days[/dim]") + console.print(" $ drone @prax log-audit sweep") + console.print() def _display_audit(files: list, summary: dict) -> None: @@ -183,6 +187,10 @@ def handle_command(command: str, args: List[str]) -> bool: _run_branch_enforce() return True + if subcmd == "sweep": + _run_sweep() + return True + error(f"Unknown log-audit subcommand: {subcmd}") print_help() return True @@ -234,6 +242,30 @@ def _run_branch_enforce(): logger.info("[log-audit] Enforced branch log limits on %d files", len(actions)) +def sweep_stale_logs(): + """Public re-export of the watchdog sweep for module-layer access.""" + from aipass.prax.apps.handlers.logging.log_watchdog import sweep_stale_logs as _sweep + + return _sweep() + + +def _run_sweep(): + """Execute stale log sweep and display results.""" + from aipass.prax.apps.handlers.logging.log_watchdog import sweep_stale_logs + + console.print("\n[bold cyan]Sweeping stale logs (>30 days)...[/bold cyan]") + result = sweep_stale_logs() + + if not result["files_removed"]: + console.print("[green]No stale logs found — nothing to delete[/green]\n") + return + + for entry in result["removed"]: + console.print(f" [red]DELETED[/red] {entry['name']}: {entry['age_days']} days old, {entry['size_kb']} KB") + console.print(f"\n Removed {result['files_removed']} file(s), reclaimed {result['total_reclaimed_kb']} KB\n") + logger.info("[log-audit] Sweep removed %d stale files", result["files_removed"]) + + if __name__ == "__main__": if len(sys.argv) == 1: print_introspection() diff --git a/src/aipass/prax/apps/modules/logger.py b/src/aipass/prax/apps/modules/logger.py index 062d8bca..e4d6b0d0 100755 --- a/src/aipass/prax/apps/modules/logger.py +++ b/src/aipass/prax/apps/modules/logger.py @@ -36,16 +36,13 @@ __all__ = [ "handle_command", "MODULE_NAME", "DATA_FILE", + "append_jsonl", ] import logging import threading from typing import Dict, Any -# Stdlib logger for except-block compliance (seedgo requires variable named 'logger') -# SystemLogger methods shadow this with local 'logger = get_system_logger()' which is fine -logger = logging.getLogger(__name__) - # NOTE: CLI imports are done lazily inside functions to avoid circular dependency. # CLI imports prax logger, so prax logger must not import CLI at module level. @@ -62,8 +59,13 @@ from aipass.prax.apps.handlers.discovery.watcher import start_file_watcher, is_f from aipass.prax.apps.handlers.registry.load import load_module_registry from aipass.prax.apps.handlers.config.load import get_system_logs_dir, get_module_logs_dir, PRAX_JSON_DIR from aipass.prax.apps.handlers.logging.direct import get_direct_logger, direct_log, DirectLogger +from aipass.prax.apps.handlers.logging.jsonl_writer import append_jsonl from aipass.prax.apps.handlers.json import json_handler +# Stdlib logger for except-block compliance (seedgo requires variable named 'logger') +# SystemLogger methods shadow this with local 'logger = get_system_logger()' which is fine +logger = logging.getLogger(__name__) + # Module constants MODULE_NAME = "prax_logger" DATA_FILE = PRAX_JSON_DIR / f"{MODULE_NAME}_data.json" diff --git a/src/aipass/prax/tests/test_jsonl_writer.py b/src/aipass/prax/tests/test_jsonl_writer.py new file mode 100644 index 00000000..8a48e811 --- /dev/null +++ b/src/aipass/prax/tests/test_jsonl_writer.py @@ -0,0 +1,151 @@ +# =================== AIPass ==================== +# Name: test_jsonl_writer.py +# Description: Tests for PRAX JSONL writer with rotation +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= + +""" +Tests for the JSONL writer — append_jsonl with size-based rotation. + +Tests verify: basic append, auto-rotation at size cap, backup creation, +directory auto-creation, and the package-level export. +""" + +import json +import sys +from pathlib import Path + + +def _get_append_jsonl(): + """Import append_jsonl after conftest mocks are active.""" + mod_name = "aipass.prax.apps.modules.logger" + sys.modules.pop(mod_name, None) + from aipass.prax.apps.modules.logger import append_jsonl + + return append_jsonl + + +class TestAppendJsonl: + """Core append behavior.""" + + def test_creates_file_and_appends(self, tmp_path): + """Verify a new file is created and data appended as JSON line.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + append_jsonl(target, {"key": "value"}) + + assert target.exists() + lines = target.read_text().strip().split("\n") + assert len(lines) == 1 + assert json.loads(lines[0]) == {"key": "value"} + + def test_appends_multiple_lines(self, tmp_path): + """Verify successive appends produce multiple JSON lines.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + append_jsonl(target, {"n": 1}) + append_jsonl(target, {"n": 2}) + append_jsonl(target, {"n": 3}) + + lines = target.read_text().strip().split("\n") + assert len(lines) == 3 + assert json.loads(lines[2])["n"] == 3 + + def test_creates_parent_directories(self, tmp_path): + """Verify missing parent directories are auto-created.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "deep" / "nested" / "dir" / "test.jsonl" + + append_jsonl(target, {"created": True}) + + assert target.exists() + assert json.loads(target.read_text().strip())["created"] is True + + def test_handles_non_serializable_with_default_str(self, tmp_path): + """Verify non-serializable types fall back to str().""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + append_jsonl(target, {"path": Path("/some/path")}) + + line = json.loads(target.read_text().strip()) + assert line["path"] == "/some/path" + + +class TestRotation: + """Size-based rotation behavior.""" + + def test_rotates_when_exceeding_max_bytes(self, tmp_path): + """Verify file is rotated to .1 when it exceeds max_bytes.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + target.write_text("x" * 500 + "\n") + + append_jsonl(target, {"after": "rotation"}, max_bytes=400) + + backup = tmp_path / "test.jsonl.1" + assert backup.exists() + assert "x" * 500 in backup.read_text() + + content = target.read_text().strip() + assert json.loads(content)["after"] == "rotation" + + def test_no_rotation_under_limit(self, tmp_path): + """Verify no rotation occurs when file is under max_bytes.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + append_jsonl(target, {"small": True}, max_bytes=10000) + + backup = tmp_path / "test.jsonl.1" + assert not backup.exists() + + def test_backup_overwritten_on_second_rotation(self, tmp_path): + """Verify second rotation overwrites the previous .1 backup.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + backup = tmp_path / "test.jsonl.1" + + target.write_text("first_content\n") + append_jsonl(target, {"round": 1}, max_bytes=10) + + assert backup.exists() + assert "first_content" in backup.read_text() + + target.write_text("second_content_padded_long\n") + append_jsonl(target, {"round": 2}, max_bytes=10) + + assert "second_content" in backup.read_text() + assert "first_content" not in backup.read_text() + + def test_zero_backup_count_deletes_instead(self, tmp_path): + """Verify backup_count=0 deletes the oversized file instead of rotating.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + target.write_text("x" * 500 + "\n") + + append_jsonl(target, {"fresh": True}, max_bytes=100, backup_count=0) + + backup = tmp_path / "test.jsonl.1" + assert not backup.exists() + assert json.loads(target.read_text().strip())["fresh"] is True + + +class TestDefaultRotation: + """Verify default rotation kicks in at the right size.""" + + def test_no_rotation_under_default_cap(self, tmp_path): + """Verify file stays intact under the 500KB default cap.""" + append_jsonl = _get_append_jsonl() + target = tmp_path / "test.jsonl" + + target.write_text("x" * 400_000 + "\n") + append_jsonl(target, {"still": "ok"}) + + assert not (tmp_path / "test.jsonl.1").exists() diff --git a/src/aipass/prax/tests/test_sweep.py b/src/aipass/prax/tests/test_sweep.py new file mode 100644 index 00000000..5167f48a --- /dev/null +++ b/src/aipass/prax/tests/test_sweep.py @@ -0,0 +1,203 @@ +# =================== AIPass ==================== +# Name: test_sweep.py +# Description: Tests for stale log sweep in log_audit +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= + +""" +Tests for sweep_stale_logs — the 30-day stale log cleanup policy. + +Verifies: age-based deletion, pattern matching (.log, .jsonl, .1 siblings), +directory scanning across system_logs/ and branch logs/. +All imports go through the module layer (log_audit). +""" + +import os +import sys +import time +from pathlib import Path +from unittest.mock import patch, MagicMock + + +def _make_old_file(path: Path, age_days: int) -> None: + """Create a file and backdate its mtime.""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("stale log content\n") + old_time = time.time() - (age_days * 86400) + os.utime(path, (old_time, old_time)) + + +def _get_sweep(): + """Import sweep_stale_logs from the module layer after conftest mocks are active.""" + mod_name = "aipass.prax.apps.modules.log_audit" + sys.modules.pop(mod_name, None) + from aipass.prax.apps.modules.log_audit import sweep_stale_logs + + return sweep_stale_logs + + +def _ensure_watchdog_mock(monkeypatch): + """Inject a mock for log_watchdog so handle_command('sweep') works.""" + mock_watchdog = MagicMock() + mock_watchdog.sweep_stale_logs = MagicMock( + return_value={ + "max_age_days": 30, + "files_removed": 1, + "total_reclaimed_kb": 12.5, + "removed": [ + {"path": "/fake/logs/old.log", "name": "old.log", "age_days": 45.2, "size_kb": 12.5}, + ], + } + ) + monkeypatch.setitem( + sys.modules, + "aipass.prax.apps.handlers.logging.log_watchdog", + mock_watchdog, + ) + return mock_watchdog + + +class TestSweepIntegration: + """Integration: sweep across system_logs and branch logs.""" + + def test_deletes_old_system_log(self, tmp_path): + """Verify sweep deletes old files from system_logs/.""" + sweep_stale_logs = _get_sweep() + + sys_logs = tmp_path / "system_logs" + sys_logs.mkdir() + _make_old_file(sys_logs / "old_module.log", 45) + + with ( + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_system_logs_dir", + return_value=sys_logs, + ), + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_ecosystem_root", + return_value=tmp_path / "src" / "aipass", + ), + ): + result = sweep_stale_logs() + + assert result["files_removed"] == 1 + assert not (sys_logs / "old_module.log").exists() + + def test_deletes_old_branch_jsonl(self, tmp_path): + """Verify sweep deletes old .jsonl files from branch logs/.""" + sweep_stale_logs = _get_sweep() + + eco = tmp_path / "src" / "aipass" + branch_logs = eco / "testbranch" / "logs" + branch_logs.mkdir(parents=True) + _make_old_file(branch_logs / "ops.jsonl", 35) + + with ( + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_system_logs_dir", + return_value=tmp_path / "system_logs", + ), + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_ecosystem_root", + return_value=eco, + ), + ): + result = sweep_stale_logs() + + assert result["files_removed"] == 1 + assert not (branch_logs / "ops.jsonl").exists() + + def test_keeps_fresh_files(self, tmp_path): + """Verify sweep leaves files younger than 30 days untouched.""" + sweep_stale_logs = _get_sweep() + + sys_logs = tmp_path / "system_logs" + sys_logs.mkdir() + fresh = sys_logs / "recent.log" + fresh.write_text("fresh content\n") + + with ( + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_system_logs_dir", + return_value=sys_logs, + ), + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_ecosystem_root", + return_value=tmp_path / "src" / "aipass", + ), + ): + result = sweep_stale_logs() + + assert result["files_removed"] == 0 + assert fresh.exists() + + def test_deletes_rotation_siblings(self, tmp_path): + """Verify sweep also removes stale .log.1 rotation backups.""" + sweep_stale_logs = _get_sweep() + + sys_logs = tmp_path / "system_logs" + sys_logs.mkdir() + _make_old_file(sys_logs / "module.log", 40) + _make_old_file(sys_logs / "module.log.1", 40) + + with ( + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_system_logs_dir", + return_value=sys_logs, + ), + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_ecosystem_root", + return_value=tmp_path / "src" / "aipass", + ), + ): + result = sweep_stale_logs() + + assert result["files_removed"] == 2 + assert not (sys_logs / "module.log").exists() + assert not (sys_logs / "module.log.1").exists() + + def test_returns_structured_summary(self, tmp_path): + """Verify sweep returns summary with counts and reclaimed size.""" + sweep_stale_logs = _get_sweep() + + sys_logs = tmp_path / "system_logs" + sys_logs.mkdir() + _make_old_file(sys_logs / "stale.log", 60) + + with ( + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_system_logs_dir", + return_value=sys_logs, + ), + patch( + "aipass.prax.apps.handlers.logging.log_watchdog._get_ecosystem_root", + return_value=tmp_path / "src" / "aipass", + ), + ): + result = sweep_stale_logs() + + assert result["max_age_days"] == 30 + assert result["files_removed"] == 1 + assert result["total_reclaimed_kb"] >= 0 + assert len(result["removed"]) == 1 + entry = result["removed"][0] + assert entry["name"] == "stale.log" + assert entry["age_days"] > 50 + + +class TestSweepCommand: + """Test the 'sweep' subcommand routing in handle_command.""" + + def test_sweep_subcommand_routes(self, monkeypatch, capsys): + """Verify 'drone @prax log-audit sweep' routes to _run_sweep.""" + _ensure_watchdog_mock(monkeypatch) + + mod_name = "aipass.prax.apps.modules.log_audit" + sys.modules.pop(mod_name, None) + from aipass.prax.apps.modules.log_audit import handle_command + + result = handle_command("log-audit", ["sweep"]) + + assert result is True