diff --git a/docs/memory/architecture.md b/docs/memory/architecture.md index 57b739ac7..0e5034b9f 100644 --- a/docs/memory/architecture.md +++ b/docs/memory/architecture.md @@ -402,6 +402,7 @@ Services that run continuously in the backend process: | **Monitoring Service** | `monitoring_service.py` | Fleet-wide health checks on configurable interval. (MON-001) | | **Scheduler Service** | `scheduler_service.py` | APScheduler-based cron job execution. Async fire-and-forget with DB polling for status. On each cron-triggered fire, optionally invokes the agent's executable `~/.trinity/pre-check` (interpreter chosen by shebang) via the backend's `POST /api/internal/agents/{name}/pre-check` (which `docker exec`s into the agent container). Empty stdout + exit 0 records a skipped execution and does not invoke Claude (SCHED-COND-001, #454). | | **Capacity Maintenance** | `capacity_manager.py` | Calls `CapacityManager.run_maintenance()` every 60s — expires stale queued tasks (>24h) and drains orphans after restart. (BACKLOG-001 / CAPACITY-CONSOLIDATE #428) | +| **Audit Retention** | `audit_retention_service.py` | Daily APScheduler job at 04:15 UTC that DELETEs `audit_log` rows past the retention window. Configured via `AUDIT_LOG_RETENTION_DAYS` (default 365, floored at 365 — the `audit_log_no_delete` trigger refuses younger rows). Pruning ages out hash-chain history past the cutoff by design. (#552) | The **agent server** also runs a 15-min `auto_sync` heartbeat loop (gated by `GIT_SYNC_AUTO` env var; default-on for non-source-mode GitHub-template diff --git a/src/backend/database.py b/src/backend/database.py index f7e2f8506..e5bac9ec7 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -1833,6 +1833,10 @@ def get_audit_stats(self, start_time: str = None, end_time: str = None): """Aggregate counts by event_type and actor_type for the dashboard.""" return self._audit_ops.get_audit_stats(start_time=start_time, end_time=end_time) + def prune_audit_log(self, retention_days: int) -> int: + """Delete audit_log entries older than ``retention_days``. Returns count removed.""" + return self._audit_ops.prune_audit_log(retention_days) + # Global database manager instance db = DatabaseManager() diff --git a/src/backend/db/audit.py b/src/backend/db/audit.py index f1c2f37f0..5de93c926 100644 --- a/src/backend/db/audit.py +++ b/src/backend/db/audit.py @@ -252,6 +252,39 @@ def get_audit_stats( "by_actor_type": by_actor_type, } + # --------------------------------------------------------------------- + # Retention + # --------------------------------------------------------------------- + + def prune_audit_log(self, retention_days: int) -> int: + """Delete entries older than ``retention_days``. Returns rows removed. + + The append-only trigger ``audit_log_no_delete`` blocks DELETEs of + rows whose ``timestamp > datetime('now', '-365 days')``. Callers + must not pass ``retention_days < 365`` — the trigger would raise + on every candidate row and the bulk DELETE would abort. + + Note (architectural invariant #16): we intentionally use SQLite's + ``datetime('now', ?)`` here — not ``iso_cutoff()`` — so the prune + WHERE filter and the trigger's WHEN clause apply the *same* + format-mismatched comparison. Aligning with the trigger avoids + IntegrityError on the day-of-cutoff boundary. Fixing the trigger + to use ISO-Z form is tracked separately. + """ + if retention_days < 365: + raise ValueError( + "retention_days must be >= 365 (audit_log_no_delete trigger floor)" + ) + with get_db_connection() as conn: + cursor = conn.cursor() + cursor.execute( + "DELETE FROM audit_log WHERE timestamp < datetime('now', ?)", + (f"-{int(retention_days)} days",), + ) + removed = cursor.rowcount + conn.commit() + return int(removed) + # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- diff --git a/src/backend/main.py b/src/backend/main.py index 6cf888b12..1281ed526 100644 --- a/src/backend/main.py +++ b/src/backend/main.py @@ -101,6 +101,9 @@ # Import log archive service from services.log_archive_service import log_archive_service +# Import audit retention service (#552) +from services.audit_retention_service import audit_retention_service + # Import operator queue sync service from services.operator_queue_service import operator_queue_service, set_websocket_manager as set_opqueue_sync_ws_manager from services.sync_health_service import sync_health_service @@ -366,6 +369,13 @@ async def lifespan(app: FastAPI): except Exception as e: print(f"Error starting log archive service: {e}") + # Initialize audit retention service (#552) + try: + audit_retention_service.start() + print("Audit retention service started") + except Exception as e: + print(f"Error starting audit retention service: {e}") + # PERF-269: Stagger background services to reduce SQLite write contention # Start operator queue sync service (OPS-001) — polls every 5s try: @@ -550,6 +560,13 @@ async def _capacity_maintenance_loop(): except Exception as e: print(f"Error stopping log archive service: {e}") + # Shutdown audit retention service (#552) + try: + audit_retention_service.stop() + print("Audit retention service stopped") + except Exception as e: + print(f"Error stopping audit retention service: {e}") + # Shutdown cleanup service try: cleanup_service.stop() diff --git a/src/backend/services/audit_retention_service.py b/src/backend/services/audit_retention_service.py new file mode 100644 index 000000000..de2823dc4 --- /dev/null +++ b/src/backend/services/audit_retention_service.py @@ -0,0 +1,107 @@ +""" +Audit Log Retention Service. + +Daily APScheduler job that prunes ``audit_log`` rows past the retention +window. Closes the gap left after #20 / Phase 4 of +``docs/requirements/AUDIT_TRAIL_ARCHITECTURE.md``: the append-only +contract is enforced by SQLite triggers, but nothing was deleting old +entries. + +Configuration (env vars): + +- ``AUDIT_LOG_RETENTION_DAYS`` (default ``365``) — minimum age before a + row is eligible for deletion. Floored at 365 because the + ``audit_log_no_delete`` trigger refuses younger rows. +- ``AUDIT_RETENTION_ENABLED`` (default ``true``) — set to ``false`` to + disable the daily prune. +- ``AUDIT_RETENTION_HOUR`` (default ``4``) — UTC hour to run. Defaults + to one hour after log archival to spread the nightly DB writes. + +Hash chain note: pruning DELETEs entries, which breaks the SHA-256 +``previous_hash``/``entry_hash`` chain across the cutoff. Verification +via ``POST /api/audit-log/verify`` should be scoped to ranges *within* +the retention window. This is documented behavior — pruning ages out +unverifiable history rather than maintaining a chain over deleted rows. +""" + +from __future__ import annotations + +import logging +import os +from typing import Any, Dict + +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger + +from database import db +from services.platform_audit_service import platform_audit_service + +logger = logging.getLogger(__name__) + +# Trigger floor — see ``audit_log_no_delete`` in db/schema.py +_RETENTION_FLOOR_DAYS = 365 + +AUDIT_LOG_RETENTION_DAYS = max( + int(os.getenv("AUDIT_LOG_RETENTION_DAYS", str(_RETENTION_FLOOR_DAYS))), + _RETENTION_FLOOR_DAYS, +) +AUDIT_RETENTION_ENABLED = os.getenv("AUDIT_RETENTION_ENABLED", "true").lower() == "true" +AUDIT_RETENTION_HOUR = int(os.getenv("AUDIT_RETENTION_HOUR", "4")) + + +class AuditRetentionService: + """Daily prune of expired audit_log rows.""" + + def __init__(self) -> None: + self.scheduler = AsyncIOScheduler() + + def start(self) -> None: + if not AUDIT_RETENTION_ENABLED: + logger.info("Audit retention disabled (AUDIT_RETENTION_ENABLED=false)") + return + + self.scheduler.add_job( + self.prune, + CronTrigger(hour=AUDIT_RETENTION_HOUR, minute=15), + id="audit_log_retention", + name="Daily audit_log retention prune", + replace_existing=True, + misfire_grace_time=3600, + ) + self.scheduler.start() + logger.info( + "Audit retention scheduler started: daily at %02d:15 UTC (retention=%dd)", + AUDIT_RETENTION_HOUR, + AUDIT_LOG_RETENTION_DAYS, + ) + + def stop(self) -> None: + if self.scheduler.running: + self.scheduler.shutdown(wait=False) + logger.info("Audit retention scheduler stopped") + + async def prune(self) -> Dict[str, Any]: + """Run a single prune cycle. Returns summary for tests/manual triggers.""" + retention_days = AUDIT_LOG_RETENTION_DAYS + try: + removed = db.prune_audit_log(retention_days) + except Exception as exc: + logger.exception("audit_log prune failed: %s", exc) + return {"removed": 0, "retention_days": retention_days, "error": str(exc)} + + if getattr(platform_audit_service, "_hash_chain_enabled", False) and removed: + logger.warning( + "audit_log prune removed %d rows while hash chain is enabled — " + "verification ranges spanning the cutoff will fail by design", + removed, + ) + + logger.info( + "audit_log prune complete: removed=%d retention_days=%d", + removed, + retention_days, + ) + return {"removed": removed, "retention_days": retention_days} + + +audit_retention_service = AuditRetentionService() diff --git a/tests/unit/test_audit_retention_prune.py b/tests/unit/test_audit_retention_prune.py new file mode 100644 index 000000000..e2b757049 --- /dev/null +++ b/tests/unit/test_audit_retention_prune.py @@ -0,0 +1,188 @@ +""" +Unit tests for audit_log retention prune (#552). + +Covers ``PlatformAuditOperations.prune_audit_log`` — the service-level +scheduler is a thin APScheduler wrapper around it. + +What we pin: +- Old rows (> retention) are removed. +- Young rows are kept (the SQLite ``audit_log_no_delete`` trigger + protects them; if prune ever tries, sqlite raises and the test + fails loudly). +- ``retention_days < 365`` is rejected before touching the DB. +- Empty table prune returns 0. +""" + +from __future__ import annotations + +import importlib.util +import sqlite3 +import sys +from pathlib import Path + +import pytest + +pytestmark = pytest.mark.unit + +_BACKEND = Path(__file__).resolve().parent.parent.parent / "src" / "backend" + + +def _load(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +_AUDIT_LOG_DDL = """ +CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT UNIQUE NOT NULL, + event_type TEXT NOT NULL, + event_action TEXT NOT NULL, + actor_type TEXT NOT NULL, + actor_id TEXT, + actor_email TEXT, + actor_ip TEXT, + mcp_key_id TEXT, + mcp_key_name TEXT, + mcp_scope TEXT, + target_type TEXT, + target_id TEXT, + timestamp TEXT NOT NULL, + details TEXT, + request_id TEXT, + source TEXT NOT NULL, + endpoint TEXT, + previous_hash TEXT, + entry_hash TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')) +) +""" + +_NO_UPDATE_TRIGGER = """ +CREATE TRIGGER audit_log_no_update BEFORE UPDATE ON audit_log +BEGIN + SELECT RAISE(ABORT, 'Audit log entries cannot be modified'); +END +""" + +_NO_DELETE_TRIGGER = """ +CREATE TRIGGER audit_log_no_delete BEFORE DELETE ON audit_log +WHEN OLD.timestamp > datetime('now', '-365 days') +BEGIN + SELECT RAISE(ABORT, 'Audit log entries cannot be deleted within retention period'); +END +""" + + +@pytest.fixture +def audit_ops(tmp_path, monkeypatch): + """Build a tmp DB with the audit_log table + triggers, return ``PlatformAuditOperations``.""" + db_path = tmp_path / "trinity.db" + monkeypatch.setenv("TRINITY_DB_PATH", str(db_path)) + + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + conn.executescript(_AUDIT_LOG_DDL) + conn.executescript(_NO_UPDATE_TRIGGER) + conn.executescript(_NO_DELETE_TRIGGER) + conn.commit() + conn.close() + + # Force-reload db/connection.py so it picks up the new TRINITY_DB_PATH. + sys.modules.pop("_arp_db_connection", None) + _load("_arp_db_connection", _BACKEND / "db" / "connection.py") + + db_pkg = type(sys)("db") + db_pkg.__path__ = [str(_BACKEND / "db")] + monkeypatch.setitem(sys.modules, "db", db_pkg) + monkeypatch.setitem(sys.modules, "db.connection", sys.modules["_arp_db_connection"]) + + # Load db/audit.py as `db.audit` so its relative `from .connection import ...` + # resolves against the `db` package we just registered. + spec = importlib.util.spec_from_file_location( + "db.audit", str(_BACKEND / "db" / "audit.py") + ) + audit_mod = importlib.util.module_from_spec(spec) + sys.modules["db.audit"] = audit_mod + spec.loader.exec_module(audit_mod) + return audit_mod.PlatformAuditOperations() + + +def _insert(db_path: Path, *, event_id: str, days_ago: int) -> None: + """Insert a minimal audit_log row whose timestamp is ``days_ago`` days old.""" + conn = sqlite3.connect(str(db_path)) + conn.execute( + """ + INSERT INTO audit_log + (event_id, event_type, event_action, actor_type, + timestamp, source, created_at) + VALUES (?, ?, ?, ?, datetime('now', ?), ?, datetime('now')) + """, + (event_id, "test", "noop", "system", f"-{days_ago} days", "api"), + ) + conn.commit() + conn.close() + + +def _count(db_path: Path) -> int: + conn = sqlite3.connect(str(db_path)) + n = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] + conn.close() + return int(n) + + +def test_prune_removes_only_old_rows(audit_ops, tmp_path): + db_path = tmp_path / "trinity.db" + + _insert(db_path, event_id="old-400d", days_ago=400) + _insert(db_path, event_id="old-380d", days_ago=380) + _insert(db_path, event_id="recent-30d", days_ago=30) + _insert(db_path, event_id="recent-1d", days_ago=1) + assert _count(db_path) == 4 + + removed = audit_ops.prune_audit_log(365) + + assert removed == 2 + assert _count(db_path) == 2 + + # Confirm the survivors are the recent ones. + conn = sqlite3.connect(str(db_path)) + rows = {r[0] for r in conn.execute("SELECT event_id FROM audit_log").fetchall()} + conn.close() + assert rows == {"recent-30d", "recent-1d"} + + +def test_prune_empty_table_returns_zero(audit_ops): + assert audit_ops.prune_audit_log(365) == 0 + + +def test_prune_below_floor_raises(audit_ops, tmp_path): + db_path = tmp_path / "trinity.db" + _insert(db_path, event_id="anything", days_ago=10) + + with pytest.raises(ValueError, match="retention_days must be >= 365"): + audit_ops.prune_audit_log(364) + + # No rows touched. + assert _count(db_path) == 1 + + +def test_prune_does_not_violate_no_delete_trigger(audit_ops, tmp_path): + """The WHERE clause and trigger WHEN clause must agree on the cutoff — + otherwise sqlite raises an IntegrityError on boundary rows.""" + db_path = tmp_path / "trinity.db" + + # Seed many rows clustered around the cutoff. If prune ever picks a row + # the trigger considers protected, sqlite raises and rowcount is -1 + # (or the entire DELETE aborts). Test fails loudly on either path. + for d in (366, 380, 400, 720): + _insert(db_path, event_id=f"old-{d}", days_ago=d) + for d in (10, 100, 200, 360): + _insert(db_path, event_id=f"new-{d}", days_ago=d) + + removed = audit_ops.prune_audit_log(365) + assert removed == 4 + assert _count(db_path) == 4