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

prax: log watchdog covers branch logs/ dirs — .jsonl runaway growth caught (built by @prax)

Browse Source 
- Root cause: rotation .log-hardcoded; .jsonl files are raw open('a') appenders bypassing prax; watchdog only scanned system_logs
- New: scan_branch_log_files (WARN 1MB / CRITICAL 10MB unrotated), enforce_branch_log_limits (tail-keep 5000 lines), branch health summary, log-audit shows both scopes
- 11 new tests, prax suite 947 green (61/61 verified in touched files by devpulse)
- Offender writers routed to owners: @hooks engine.jsonl+telegram_delivery.jsonl, @backup operations.jsonl, @trigger medic_suppressed.log
This commit is contained in:
AIOSAI
2026-07-06 10:16:11 -07:00
parent f4f6943ed6
commit 47b5b2d871
5 changed files with 506 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+19
View File
@@ -9,6 +9,25 @@ PyPI version — not the changelog header.
---
## [2026-07-06]
### Fixed
- **prax log watchdog now covers branch `logs/` dirs — `.jsonl` runaway growth
caught.** Rotation was hardcoded to `.log` files, and several branches write
`.jsonl` logs via raw `open(path, "a")` appenders that bypass prax entirely —
`hooks/logs/engine.jsonl` had grown to 63 MB, `backup/logs/operations.jsonl`
to 31 MB, `trigger/logs/medic_suppressed.log` to 7 MB, all unrotated. The
log-watchdog safety net also only scanned `system_logs/*.log`. @prax extended
it: `scan_branch_log_files()` sweeps every `src/aipass/*/logs/` for `.log` +
`.jsonl` (WARN at 1 MB unrotated, CRITICAL at 10 MB),
`enforce_branch_log_limits()` truncates flagged files to the last 5000 lines,
and `drone @prax log-audit` now reports both system and branch scopes. 11 new
tests, full prax suite 947 green. The raw-appender writers themselves still
need per-owner caps — routed to @hooks, @backup, @trigger. (built by @prax)
---
## [2026-07-05]
### Fixed
src/aipass/prax/apps/handlers/logging/log_watchdog.py
+145 -6
View File
@@ -9,21 +9,24 @@
"""
System Log Size Watchdog
Scans the system_logs/ directory for oversized log files and enforces size limits.
Catches ALL log files regardless of how they were created — even those bypassing
PRAX's RotatingFileHandler (e.g., telegram bots using plain FileHandler).
Scans system_logs/ and all branch logs/ directories for oversized files and
enforces size limits. Catches ALL log files regardless of how they were created
— even those bypassing PRAX's RotatingFileHandler (e.g., raw open("a") appenders
writing .jsonl files, telegram bots using plain FileHandler).
This is the safety net: even if a branch misconfigures logging, the watchdog
prevents unbounded growth that caused the 2026-02-26 system crash (DPLAN-037).
Two scopes:
- system_logs/ (central) — scanned by scan_log_files()
- src/aipass/*/logs/ (branch-local) — scanned by scan_branch_log_files()
Two modes:
- audit: Report oversized files without changing anything
- enforce: Truncate oversized files to keep last max_lines
"""
import logging
logger = logging.getLogger(__name__)
import sys
from datetime import datetime
from pathlib import Path
@@ -31,6 +34,8 @@ from typing import Any, Dict, List, Tuple
from aipass.prax.apps.handlers.json import json_handler
logger = logging.getLogger(__name__)
# =============================================================================
# CONSTANTS
@@ -57,11 +62,16 @@ def _get_system_logs_dir() -> Path:
return _system_logs_dir_cache
# Thresholds
# Thresholds — system_logs (line-based)
WARN_THRESHOLD_LINES = 5000 # Fire warning at this line count
DEFAULT_MAX_LINES = 1000 # Truncate to this many lines (matches prax config)
CRITICAL_THRESHOLD_LINES = 10000 # Immediate action recommended
# Thresholds — branch logs (size-based, catches .jsonl and unrotated .log)
BRANCH_WARN_SIZE_MB = 1.0
BRANCH_CRITICAL_SIZE_MB = 10.0
BRANCH_DEFAULT_MAX_LINES = 5000
# =============================================================================
# SCANNING
@@ -277,6 +287,135 @@ def log_health_summary() -> Dict[str, Any]:
}
# =============================================================================
# BRANCH LOG SCANNING — covers .log and .jsonl in src/aipass/*/logs/
# =============================================================================
def _get_ecosystem_root() -> Path:
"""Find src/aipass/ directory."""
return _find_repo_root() / "src" / "aipass"
def _has_rotation_sibling(filepath: Path) -> bool:
"""Check if a file has a .1 rotation sibling."""
return (filepath.parent / f"{filepath.name}.1").exists()
def _classify_branch_log(log_file: Path, branch_name: str) -> Dict[str, Any]:
"""Build an audit record for a single branch log file."""
size_kb = _get_file_size_kb(log_file)
size_mb = size_kb / 1024.0
lines = _count_lines(log_file)
has_rotation = _has_rotation_sibling(log_file)
if size_mb >= BRANCH_CRITICAL_SIZE_MB:
status = "critical"
elif size_mb >= BRANCH_WARN_SIZE_MB and not has_rotation:
status = "warning"
else:
status = "ok"
return {
"path": str(log_file),
"name": log_file.name,
"branch": branch_name,
"lines": lines,
"size_kb": round(size_kb, 1),
"size_mb": round(size_mb, 1),
"has_rotation": has_rotation,
"status": status,
}
def scan_branch_log_files() -> List[Dict[str, Any]]:
"""
Scan all branch logs/ directories for files with unbounded growth.
Checks .log and .jsonl files. Flags unrotated files exceeding size
thresholds — the safety net for writers that bypass RotatingFileHandler.
"""
results: List[Dict[str, Any]] = []
eco_root = _get_ecosystem_root()
if not eco_root.exists():
return results
for branch_dir in sorted(eco_root.iterdir()):
logs_dir = branch_dir / "logs"
if not branch_dir.is_dir() or not logs_dir.is_dir():
continue
for log_file in sorted(logs_dir.glob("*.log")) + sorted(logs_dir.glob("*.jsonl")):
results.append(_classify_branch_log(log_file, branch_dir.name))
results.sort(key=lambda x: x["size_kb"], reverse=True)
json_handler.log_operation("branch_log_watchdog_check", {"files_scanned": len(results)})
return results
def get_oversized_branch_files() -> List[Dict[str, Any]]:
"""Get branch log files exceeding size thresholds."""
return [f for f in scan_branch_log_files() if f["status"] != "ok"]
def enforce_branch_log_limits(
max_lines: int = BRANCH_DEFAULT_MAX_LINES,
) -> List[Dict[str, Any]]:
"""
Truncate oversized branch log files (including .jsonl).
Only truncates files flagged as warning or critical by scan_branch_log_files().
"""
actions: List[Dict[str, Any]] = []
for file_info in get_oversized_branch_files():
filepath = Path(file_info["path"])
original, new = truncate_log_file(filepath, max_lines)
actions.append(
{
"name": file_info["name"],
"branch": file_info["branch"],
"original_lines": original,
"new_lines": new,
"size_mb": file_info["size_mb"],
"truncated": original != new,
}
)
return actions
def branch_log_health_summary() -> Dict[str, Any]:
"""Generate a health summary of branch logs."""
files = scan_branch_log_files()
if not files:
return {
"total_files": 0,
"oversized_count": 0,
"critical_count": 0,
"total_size_mb": 0.0,
"largest_file": None,
"healthy": True,
}
oversized = [f for f in files if f["status"] in ("warning", "critical")]
critical = [f for f in files if f["status"] == "critical"]
largest = files[0] if files else None
total_size = sum(f["size_mb"] for f in files)
return {
"total_files": len(files),
"oversized_count": len(oversized),
"critical_count": len(critical),
"total_size_mb": round(total_size, 1),
"largest_file": f"{largest['branch']}/{largest['name']}" if largest else None,
"largest_size_mb": largest["size_mb"] if largest else 0.0,
"healthy": len(oversized) == 0,
}
# =============================================================================
# CLI ENTRY POINT (for testing)
# =============================================================================
src/aipass/prax/apps/modules/log_audit.py
+64 -4
View File
@@ -75,9 +75,9 @@ def print_help():
def _display_audit(files: list, summary: dict) -> None:
"""Display audit results."""
"""Display system_logs/ audit results."""
console.print()
console.print("[bold cyan]System Log Audit[/bold cyan]")
console.print("[bold cyan]System Log Audit[/bold cyan] [dim](system_logs/)[/dim]")
console.print(f" Total files: {summary['total_files']}")
console.print(f" Total lines: {summary['total_lines']:,}")
if summary.get("largest_file"):
@@ -90,7 +90,6 @@ def _display_audit(files: list, summary: dict) -> None:
else:
error(f"Status: {summary['oversized_count']} oversized, {summary['critical_count']} critical")
# Show oversized files
oversized = [f for f in files if f["status"] != "ok"]
if oversized:
console.print()
@@ -101,8 +100,36 @@ def _display_audit(files: list, summary: dict) -> None:
f" [{status_color}]{f['status'].upper()}[/{status_color}] "
f"{f['name']}: {f['lines']:,} lines ({f['size_kb']} KB)"
)
console.print()
def _display_branch_audit(files: list, summary: dict) -> None:
"""Display branch logs/ audit results."""
console.print("[bold cyan]Branch Log Audit[/bold cyan] [dim](src/aipass/*/logs/)[/dim]")
console.print(f" Total files: {summary['total_files']}")
console.print(f" Total size: {summary['total_size_mb']} MB")
if summary.get("largest_file"):
console.print(f" Largest: {summary['largest_file']} ({summary.get('largest_size_mb', 0)} MB)")
if summary["healthy"]:
console.print("[green] Status: HEALTHY — no unbounded files[/green]")
else:
error(f"Status: {summary['oversized_count']} unbounded, {summary['critical_count']} critical")
oversized = [f for f in files if f["status"] != "ok"]
if oversized:
console.print()
console.print("[dim]Run 'drone @prax log-audit enforce' to truncate oversized files[/dim]")
console.print("[bold]Unbounded files (no rotation, exceeds size threshold):[/bold]")
for f in oversized:
status_color = "red" if f["status"] == "critical" else "yellow"
rotation = "[green]rotated[/green]" if f["has_rotation"] else "[red]unrotated[/red]"
console.print(
f" [{status_color}]{f['status'].upper()}[/{status_color}] "
f"{f['branch']}/{f['name']}: {f['size_mb']} MB, "
f"{f['lines']:,} lines, {rotation}"
)
console.print()
console.print("[dim]Run 'drone @prax log-audit enforce' to truncate[/dim]")
console.print()
@@ -140,10 +167,20 @@ def handle_command(command: str, args: List[str]) -> bool:
files = scan_log_files()
summary = log_health_summary()
_display_audit(files, summary)
from aipass.prax.apps.handlers.logging.log_watchdog import (
scan_branch_log_files,
branch_log_health_summary,
)
branch_files = scan_branch_log_files()
branch_summary = branch_log_health_summary()
_display_branch_audit(branch_files, branch_summary)
return True
if subcmd == "enforce":
_run_enforce()
_run_branch_enforce()
return True
error(f"Unknown log-audit subcommand: {subcmd}")
@@ -174,6 +211,29 @@ def _run_enforce():
logger.info("[log-audit] Enforced limits on %d files", len(actions))
def _run_branch_enforce():
"""Execute branch log enforcement and display results."""
from aipass.prax.apps.handlers.logging.log_watchdog import enforce_branch_log_limits
console.print("[bold cyan]Enforcing branch log limits...[/bold cyan]")
actions = enforce_branch_log_limits()
if not actions:
console.print("[green]All branch logs within limits — nothing to truncate[/green]\n")
return
for action in actions:
if action["truncated"]:
console.print(
f" [red]TRUNCATED[/red] {action['branch']}/{action['name']}: "
f"{action['size_mb']} MB, {action['original_lines']:,} → {action['new_lines']:,} lines"
)
else:
console.print(f" [green]OK[/green] {action['branch']}/{action['name']}: within limits")
console.print()
logger.info("[log-audit] Enforced branch log limits on %d files", len(actions))
if __name__ == "__main__":
if len(sys.argv) == 1:
print_introspection()
src/aipass/prax/tests/test_log_audit.py
+126 -11
View File
@@ -47,6 +47,43 @@ def _ensure_watchdog_mock(monkeypatch):
{"name": "error.log", "truncated": True, "original_lines": 2500, "new_lines": 1000},
]
)
mock_watchdog.scan_branch_log_files = MagicMock(
return_value=[
{
"name": "engine.jsonl",
"branch": "hooks",
"lines": 200000,
"size_kb": 64512.0,
"size_mb": 63.0,
"has_rotation": False,
"status": "critical",
"path": "/fake/hooks/logs/engine.jsonl",
},
]
)
mock_watchdog.branch_log_health_summary = MagicMock(
return_value={
"total_files": 5,
"oversized_count": 1,
"critical_count": 1,
"total_size_mb": 95.1,
"largest_file": "hooks/engine.jsonl",
"largest_size_mb": 63.0,
"healthy": False,
}
)
mock_watchdog.enforce_branch_log_limits = MagicMock(
return_value=[
{
"name": "engine.jsonl",
"branch": "hooks",
"original_lines": 200000,
"new_lines": 5001,
"size_mb": 63.0,
"truncated": True,
},
]
)
monkeypatch.setitem(
sys.modules,
"aipass.prax.apps.handlers.logging.log_watchdog",
@@ -64,9 +101,10 @@ def _fresh_import():
print_help,
print_introspection,
_display_audit,
_display_branch_audit,
)
return handle_command, print_help, print_introspection, _display_audit
return handle_command, print_help, print_introspection, _display_audit, _display_branch_audit
# =============================================
@@ -76,7 +114,7 @@ def _fresh_import():
def test_handle_command_help(mock_prax_infrastructure, monkeypatch):
"""--help flag returns True and displays help text."""
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", ["--help"])
assert result is True
@@ -85,7 +123,7 @@ def test_handle_command_help(mock_prax_infrastructure, monkeypatch):
def test_handle_command_help_h_flag(mock_prax_infrastructure, monkeypatch):
"""-h flag also triggers help with audit-related content."""
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", ["-h"])
assert result is True
@@ -95,7 +133,7 @@ def test_handle_command_help_h_flag(mock_prax_infrastructure, monkeypatch):
def test_handle_command_no_args_calls_introspection(mock_prax_infrastructure, monkeypatch):
"""No args prints introspection and returns True."""
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", [])
assert result is True
@@ -105,7 +143,7 @@ def test_handle_command_no_args_calls_introspection(mock_prax_infrastructure, mo
def test_handle_command_wrong_command(mock_prax_infrastructure, monkeypatch):
"""Wrong command name returns False."""
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("not-log-audit", [])
assert result is False
@@ -113,7 +151,7 @@ def test_handle_command_wrong_command(mock_prax_infrastructure, monkeypatch):
def test_print_help_runs(mock_prax_infrastructure, monkeypatch):
"""print_help runs without error and includes audit/enforce subcommands."""
_, print_help, _, _ = _fresh_import()
_, print_help, _, _, _ = _fresh_import()
print_help()
mock_prax_infrastructure.console.print.assert_called()
@@ -124,7 +162,7 @@ def test_print_help_runs(mock_prax_infrastructure, monkeypatch):
def test_print_introspection_runs(mock_prax_infrastructure, monkeypatch):
"""print_introspection runs without error."""
_, _, print_introspection, _ = _fresh_import()
_, _, print_introspection, _, _ = _fresh_import()
print_introspection()
calls = [str(c) for c in mock_prax_infrastructure.console.print.call_args_list]
@@ -133,7 +171,7 @@ def test_print_introspection_runs(mock_prax_infrastructure, monkeypatch):
def test_display_audit_healthy(mock_prax_infrastructure, monkeypatch):
"""_display_audit formats healthy summary correctly."""
_, _, _, _display_audit = _fresh_import()
_, _, _, _display_audit, _ = _fresh_import()
files = [{"name": "system.log", "lines": 200, "size_kb": 10, "status": "ok"}]
summary = {
@@ -154,7 +192,7 @@ def test_display_audit_healthy(mock_prax_infrastructure, monkeypatch):
def test_display_audit_oversized(mock_prax_infrastructure, monkeypatch):
"""_display_audit shows oversized files when present."""
_, _, _, _display_audit = _fresh_import()
_, _, _, _display_audit, _ = _fresh_import()
files = [
{"name": "system.log", "lines": 500, "size_kb": 45, "status": "ok"},
@@ -185,7 +223,7 @@ def test_display_audit_oversized(mock_prax_infrastructure, monkeypatch):
def test_handle_command_unknown_subcommand(mock_prax_infrastructure, monkeypatch):
"""Unknown subcommand shows error and help text."""
_ensure_watchdog_mock(monkeypatch)
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", ["bogus"])
assert result is True
@@ -200,9 +238,86 @@ def test_handle_command_unknown_subcommand(mock_prax_infrastructure, monkeypatch
def test_handle_command_audit_subcommand(mock_prax_infrastructure, monkeypatch):
"""'audit' subcommand calls scan_log_files and log_health_summary."""
mock_watchdog = _ensure_watchdog_mock(monkeypatch)
handle_command, _, _, _ = _fresh_import()
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", ["audit"])
assert result is True
mock_watchdog.scan_log_files.assert_called_once()
mock_watchdog.log_health_summary.assert_called_once()
mock_watchdog.scan_branch_log_files.assert_called_once()
mock_watchdog.branch_log_health_summary.assert_called_once()
def test_handle_command_enforce_calls_branch_enforce(mock_prax_infrastructure, monkeypatch):
"""'enforce' subcommand calls both system and branch enforcement."""
mock_watchdog = _ensure_watchdog_mock(monkeypatch)
handle_command, _, _, _, _ = _fresh_import()
result = handle_command("log-audit", ["enforce"])
assert result is True
mock_watchdog.enforce_log_limits.assert_called_once()
mock_watchdog.enforce_branch_log_limits.assert_called_once()
def test_display_branch_audit_healthy(mock_prax_infrastructure, monkeypatch):
"""_display_branch_audit shows healthy status when no unbounded files."""
_, _, _, _, _display_branch_audit = _fresh_import()
files = [
{
"name": "client.log",
"branch": "backup",
"lines": 200,
"size_kb": 40.0,
"size_mb": 0.04,
"has_rotation": True,
"status": "ok",
},
]
summary = {
"total_files": 1,
"oversized_count": 0,
"critical_count": 0,
"total_size_mb": 0.04,
"largest_file": "backup/client.log",
"largest_size_mb": 0.04,
"healthy": True,
}
_display_branch_audit(files, summary)
calls = [str(c) for c in mock_prax_infrastructure.console.print.call_args_list]
assert any("HEALTHY" in c for c in calls)
def test_display_branch_audit_critical(mock_prax_infrastructure, monkeypatch):
"""_display_branch_audit shows critical unbounded .jsonl files."""
_, _, _, _, _display_branch_audit = _fresh_import()
files = [
{
"name": "engine.jsonl",
"branch": "hooks",
"lines": 200000,
"size_kb": 64512.0,
"size_mb": 63.0,
"has_rotation": False,
"status": "critical",
"path": "/fake/hooks/logs/engine.jsonl",
},
]
summary = {
"total_files": 1,
"oversized_count": 1,
"critical_count": 1,
"total_size_mb": 63.0,
"largest_file": "hooks/engine.jsonl",
"largest_size_mb": 63.0,
"healthy": False,
}
_display_branch_audit(files, summary)
calls = [str(c) for c in mock_prax_infrastructure.console.print.call_args_list]
assert any("engine.jsonl" in c for c in calls)
assert any("hooks" in c for c in calls)
assert any("unrotated" in c for c in calls)
mock_prax_infrastructure.cli.error.assert_called()
src/aipass/prax/tests/test_logging_handlers.py
+152 -1
View File
@@ -24,6 +24,7 @@ import importlib # noqa: F401 — used inside test functions for dynamic module
import json
import logging
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -188,7 +189,7 @@ class TestGetCallingModulePath:
"""Returns the path from _find_external_caller_path when found."""
from aipass.prax.apps.handlers.logging import introspection
fake_path = "/home/user/src/aipass/flow/apps/flow.py"
fake_path = str(Path.home() / "src" / "aipass" / "flow" / "apps" / "flow.py")
with patch.object(introspection, "_find_external_caller_path", return_value=fake_path):
result = introspection.get_calling_module_path()
assert result == fake_path
@@ -342,6 +343,156 @@ class TestTruncateLogFile:
assert new == 0
# =============================================
# log_watchdog.py -- scan_branch_log_files
# =============================================
def _import_watchdog(tmp_path):
"""Fresh-import log_watchdog with mocked config."""
with patch.dict(
sys.modules,
{
"aipass.prax.apps.handlers.config.load": MagicMock(
PRAX_JSON_DIR=tmp_path / "prax_json",
),
},
):
sys.modules.pop("aipass.prax.apps.handlers.logging.log_watchdog", None)
import aipass.prax.apps.handlers.logging.log_watchdog as lw
return lw
class TestScanBranchLogFiles:
"""Tests for log_watchdog.py scan_branch_log_files()."""
def test_detects_large_jsonl(self, mock_prax_infrastructure, tmp_path):
"""Flags .jsonl files exceeding size threshold as critical."""
lw = _import_watchdog(tmp_path)
eco = tmp_path / "src" / "aipass"
branch_logs = eco / "hooks" / "logs"
branch_logs.mkdir(parents=True)
big_jsonl = branch_logs / "engine.jsonl"
big_jsonl.write_text(
"\n".join(f'{{"line": {i}, "padding": "{" " * 200}}}' for i in range(10000)) + "\n",
encoding="utf-8",
)
with patch.object(lw, "_get_ecosystem_root", return_value=eco):
with patch.object(lw, "BRANCH_WARN_SIZE_MB", 0.5):
results = lw.scan_branch_log_files()
assert len(results) == 1
assert results[0]["name"] == "engine.jsonl"
assert results[0]["branch"] == "hooks"
assert not results[0]["has_rotation"]
assert results[0]["status"] in ("warning", "critical")
def test_ignores_small_rotated_logs(self, mock_prax_infrastructure, tmp_path):
"""Small .log files with rotation siblings are status ok."""
lw = _import_watchdog(tmp_path)
eco = tmp_path / "src" / "aipass"
branch_logs = eco / "backup" / "logs"
branch_logs.mkdir(parents=True)
small_log = branch_logs / "client.log"
small_log.write_text("line1\nline2\n", encoding="utf-8")
rotation = branch_logs / "client.log.1"
rotation.write_text("old line\n", encoding="utf-8")
with patch.object(lw, "_get_ecosystem_root", return_value=eco):
results = lw.scan_branch_log_files()
assert len(results) == 1
assert results[0]["name"] == "client.log"
assert results[0]["has_rotation"] is True
assert results[0]["status"] == "ok"
def test_empty_ecosystem_returns_empty(self, mock_prax_infrastructure, tmp_path):
"""Returns empty when ecosystem root does not exist."""
lw = _import_watchdog(tmp_path)
nonexistent = tmp_path / "no_such_dir"
with patch.object(lw, "_get_ecosystem_root", return_value=nonexistent):
results = lw.scan_branch_log_files()
assert results == []
def test_scans_multiple_branches(self, mock_prax_infrastructure, tmp_path):
"""Scans logs/ across multiple branches."""
lw = _import_watchdog(tmp_path)
eco = tmp_path / "src" / "aipass"
for branch in ("hooks", "backup", "trigger"):
logs = eco / branch / "logs"
logs.mkdir(parents=True)
(logs / "test.log").write_text("line\n" * 10, encoding="utf-8")
with patch.object(lw, "_get_ecosystem_root", return_value=eco):
results = lw.scan_branch_log_files()
branches = {r["branch"] for r in results}
assert branches == {"hooks", "backup", "trigger"}
class TestBranchLogHealthSummary:
"""Tests for log_watchdog.py branch_log_health_summary()."""
def test_healthy_summary(self, mock_prax_infrastructure, tmp_path):
"""Returns healthy when all files are ok."""
lw = _import_watchdog(tmp_path)
eco = tmp_path / "src" / "aipass"
logs = eco / "prax" / "logs"
logs.mkdir(parents=True)
(logs / "test.log").write_text("line\n" * 5, encoding="utf-8")
with patch.object(lw, "_get_ecosystem_root", return_value=eco):
summary = lw.branch_log_health_summary()
assert summary["healthy"] is True
assert summary["total_files"] == 1
assert summary["oversized_count"] == 0
def test_empty_summary(self, mock_prax_infrastructure, tmp_path):
"""Returns healthy empty summary when no files found."""
lw = _import_watchdog(tmp_path)
nonexistent = tmp_path / "nope"
with patch.object(lw, "_get_ecosystem_root", return_value=nonexistent):
summary = lw.branch_log_health_summary()
assert summary["healthy"] is True
assert summary["total_files"] == 0
class TestEnforceBranchLogLimits:
"""Tests for log_watchdog.py enforce_branch_log_limits()."""
def test_truncates_oversized_jsonl(self, mock_prax_infrastructure, tmp_path):
"""Truncates .jsonl files that exceed size threshold."""
lw = _import_watchdog(tmp_path)
eco = tmp_path / "src" / "aipass"
logs = eco / "hooks" / "logs"
logs.mkdir(parents=True)
big_jsonl = logs / "engine.jsonl"
big_jsonl.write_text(
"\n".join(f'{{"line": {i}, "padding": "{" " * 200}}}' for i in range(10000)) + "\n",
encoding="utf-8",
)
with patch.object(lw, "_get_ecosystem_root", return_value=eco):
with patch.object(lw, "BRANCH_WARN_SIZE_MB", 0.5):
actions = lw.enforce_branch_log_limits(max_lines=1000)
assert len(actions) >= 1
action = actions[0]
assert action["truncated"] is True
assert action["branch"] == "hooks"
content = big_jsonl.read_text(encoding="utf-8")
assert "LOG TRUNCATED by PRAX watchdog" in content
# =============================================
# monitoring.py -- run_monitoring_loop
# =============================================