diff --git a/CHANGELOG.md b/CHANGELOG.md index b2b3b76a..ca50f028 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/src/aipass/prax/apps/handlers/logging/log_watchdog.py b/src/aipass/prax/apps/handlers/logging/log_watchdog.py index 20290203..d612b031 100644 --- a/src/aipass/prax/apps/handlers/logging/log_watchdog.py +++ b/src/aipass/prax/apps/handlers/logging/log_watchdog.py @@ -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 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) # ============================================================================= diff --git a/src/aipass/prax/apps/modules/log_audit.py b/src/aipass/prax/apps/modules/log_audit.py index 185c1da2..4046c8f1 100644 --- a/src/aipass/prax/apps/modules/log_audit.py +++ b/src/aipass/prax/apps/modules/log_audit.py @@ -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("[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 oversized files[/dim]") + 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() diff --git a/src/aipass/prax/tests/test_log_audit.py b/src/aipass/prax/tests/test_log_audit.py index c1c6bd03..3c1cc5a4 100644 --- a/src/aipass/prax/tests/test_log_audit.py +++ b/src/aipass/prax/tests/test_log_audit.py @@ -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() diff --git a/src/aipass/prax/tests/test_logging_handlers.py b/src/aipass/prax/tests/test_logging_handlers.py index 47d5a3e1..5cdfa65f 100644 --- a/src/aipass/prax/tests/test_logging_handlers.py +++ b/src/aipass/prax/tests/test_logging_handlers.py @@ -24,6 +24,7 @@ import json import logging import sys +from pathlib import Path from unittest.mock import MagicMock, patch @@ -188,7 +189,7 @@ def test_returns_path_when_external_caller_found(self, mock_prax_infrastructure) """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 @@ def test_truncate_handles_os_error(self, mock_prax_infrastructure, tmp_path): 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 # =============================================