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

#673 prax-side: append_jsonl (sanctioned .jsonl writer — 500KB/1-backup atomic os.replace rotation) + drone @prax log-audit sweep (30-day stale-log sweep over system + branch logs). Replaces the raw open('a') rotation-bypass. 15 tests. Offenders hooks/backup/trigger adopt next.

Browse Source
This commit is contained in:
AIOSAI
2026-07-10 18:00:12 -07:00
parent 948d2ed535
commit dfd6732f5b
7 changed files with 581 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/prax/__init__.py
+7 -2
View File
@@ -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()
src/aipass/prax/apps/handlers/logging/jsonl_writer.py
+104
View File
@@ -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)
src/aipass/prax/apps/handlers/logging/log_watchdog.py
+78
View File
@@ -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)
# =============================================================================
src/aipass/prax/apps/modules/log_audit.py
+32
View File
@@ -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()
src/aipass/prax/apps/modules/logger.py
+6 -4
View File
@@ -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"
src/aipass/prax/tests/test_jsonl_writer.py
+151
View File
@@ -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()
src/aipass/prax/tests/test_sweep.py
+203
View File
@@ -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