feat(security): add audit logging for apply

benchmarks/security_audit_bench.py

@@ -0,0 +1,136 @@
+"""ASV benchmarks for SEC7 - Audit logging overhead.
+
+Measures the performance of recording, querying, and pruning
+audit log entries to establish baselines for write overhead.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.application.services.audit_service import AuditService
+from cleveragents.config.settings import Settings
+from cleveragents.infrastructure.database.models import AuditLogModel, Base
+
+
+def _make_service() -> AuditService:
+    """Create a service backed by in-memory SQLite."""
+    Settings._instance = None
+    settings = Settings()
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    return AuditService(settings=settings, session=session)
+
+
+def _seed_entries(service: AuditService, count: int) -> None:
+    """Insert *count* entries for benchmarking."""
+    for i in range(count):
+        service.record(
+            event_type="plan_applied",
+            plan_id=f"plan-{i}",
+            project_name="bench-project",
+            details={"index": i},
+        )
+
+
+class TimeRecordSingle:
+    """Benchmark recording a single audit event."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.service = _make_service()
+
+    def teardown(self) -> None:
+        pass
+
+    def time_record_single_event(self) -> None:
+        self.service.record(
+            event_type="plan_applied",
+            plan_id="bench-plan",
+            project_name="bench-project",
+            details={"files_changed": 5},
+        )
+
+
+class TimeRecordBatch:
+    """Benchmark recording a batch of audit events."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        self.service = _make_service()
+
+    def teardown(self) -> None:
+        pass
+
+    def time_record_100_events(self) -> None:
+        for i in range(100):
+            self.service.record(
+                event_type="plan_applied",
+                plan_id=f"plan-{i}",
+                details={"index": i},
+            )
+
+
+class TimeQueryEntries:
+    """Benchmark querying audit log entries."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.service = _make_service()
+        _seed_entries(self.service, 500)
+
+    def teardown(self) -> None:
+        pass
+
+    def time_list_all(self) -> None:
+        self.service.list_entries(limit=100)
+
+    def time_list_by_plan(self) -> None:
+        self.service.list_entries(plan_id="plan-42")
+
+    def time_list_by_project(self) -> None:
+        self.service.list_entries(project_name="bench-project")
+
+    def time_list_by_type(self) -> None:
+        self.service.list_entries(event_type="plan_applied")
+
+    def time_get_single(self) -> None:
+        self.service.get_entry(250)
+
+    def time_count(self) -> None:
+        self.service.count()
+
+
+class TimePrune:
+    """Benchmark pruning old audit entries."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.service = _make_service()
+        # Insert entries backdated to 60 days ago
+        old_time = (datetime.now(tz=UTC) - timedelta(days=60)).strftime(
+            "%Y-%m-%dT%H:%M:%S.%f"
+        )
+        for i in range(200):
+            row = AuditLogModel(
+                event_type="plan_applied",
+                plan_id=f"plan-{i}",
+                details="{}",
+                created_at=old_time,
+            )
+            self.service._session.add(row)
+        self.service._session.commit()
+
+    def teardown(self) -> None:
+        pass
+
+    def time_prune_200_entries(self) -> None:
+        self.service.prune(retention_days=30)