feat(audit): daily retention prune for audit_log entries past 365d (#552)

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

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()

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
     # ---------------------------------------------------------------------

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()

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()

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