From f0a40afecc9c8954aba158c5dd2ca1631a892226 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Thu, 2 Apr 2026 17:00:53 +0000 Subject: [PATCH] refactor(audit): implement async audit recording to unblock event pipeline (#1279) Implements a write-behind queue with a background daemon thread in AuditService so that record() returns immediately without blocking the calling domain operation on a synchronous SQLite INSERT + COMMIT. Changes: - AuditService: add _writer_loop(), _write_payload(), flush() methods and a write-behind queue with a background thread - record() enqueues the payload and returns a placeholder entry with id=-1 in async mode - close() calls flush() to drain the queue before closing the session - Settings: add audit_async (default True) and audit_queue_maxsize (default 10000) fields - 20 BDD scenarios covering non-blocking record(), flush/close lifecycle, ordering, sync fallback, and settings Closes #718 Co-authored-by: Jeffrey Phillips Freeman Co-committed-by: Jeffrey Phillips Freeman --- features/async_audit_recording.feature | 112 ++++++ features/steps/async_audit_recording_steps.py | 348 ++++++++++++++++++ features/steps/audit_service_wiring_steps.py | 2 +- features/steps/security_audit_steps.py | 4 +- .../application/services/audit_service.py | 207 ++++++++++- src/cleveragents/config/settings.py | 25 ++ 6 files changed, 692 insertions(+), 6 deletions(-) create mode 100644 features/async_audit_recording.feature create mode 100644 features/steps/async_audit_recording_steps.py diff --git a/features/async_audit_recording.feature b/features/async_audit_recording.feature new file mode 100644 index 00000000..c6a4a40c --- /dev/null +++ b/features/async_audit_recording.feature @@ -0,0 +1,112 @@ +Feature: Async audit recording (issue #718) + As a system operator + I want audit recording to be non-blocking + So that domain operations are not delayed by SQLite writes + + # ── Non-blocking record() ───────────────────────────────────── + + Scenario: record() returns immediately without blocking in async mode + Given an async audit service + When I record a "plan_applied" event in async mode + Then the record call should return immediately + And the returned entry should have event_type "plan_applied" + + Scenario: Async record returns placeholder id of -1 + Given an async audit service + When I record a "plan_applied" event in async mode + Then the returned entry id should be -1 + + Scenario: Entries are persisted after flush + Given an async audit service + When I record a "plan_applied" event in async mode + And I flush the async audit service + Then the audit log should contain 1 persisted entry + + Scenario: Multiple entries are all persisted after flush + Given an async audit service + When I record 5 "plan_applied" events in async mode + And I flush the async audit service + Then the audit log should contain 5 persisted entries + + Scenario: Flush waits for all enqueued entries to be written + Given an async audit service + When I record 10 "config_changed" events in async mode + And I flush the async audit service + Then the audit log should contain 10 persisted entries + + # ── No data loss ────────────────────────────────────────────── + + Scenario: Close flushes all pending entries before shutting down + Given an async audit service + When I record 3 "session_created" events in async mode + And I close the async audit service + Then the audit log should contain 3 persisted entries after close + + Scenario: Context manager flushes on exit + Given an async audit service used as context manager + When I record a "plan_cancelled" event inside the context manager + Then the audit log should contain 1 persisted entry after context exit + + # ── Synchronous fallback ────────────────────────────────────── + + Scenario: Synchronous mode returns real id immediately + Given a synchronous audit service + When I record a "plan_applied" event synchronously + Then the returned entry id should be a positive integer + + Scenario: Synchronous mode persists immediately without flush + Given a synchronous audit service + When I record a "plan_applied" event synchronously + Then the audit log should contain 1 persisted entry immediately + + # ── Settings ────────────────────────────────────────────────── + + Scenario: Default audit_async setting is True + Given default settings + Then audit_async should be True + + Scenario: audit_async can be disabled via settings + Given settings with audit_async False + Then audit_async should be False + + Scenario: audit_queue_maxsize default is 10000 + Given default settings + Then audit_queue_maxsize should be 10000 + + # ── Background thread ───────────────────────────────────────── + + Scenario: Background writer thread is started in async mode + Given an async audit service + Then the background writer thread should be alive + + Scenario: Background writer thread stops after flush + Given an async audit service + When I flush the async audit service + Then the background writer thread should be stopped + + # ── Ordering guarantees ─────────────────────────────────────── + + Scenario: Entries are written in enqueue order + Given an async audit service + When I record events with plan_ids "first" "second" "third" in async mode + And I flush the async audit service + Then the entries should be persisted in enqueue order + + # ── Error resilience ────────────────────────────────────────── + + Scenario: Invalid event_type raises ValueError immediately in async mode + Given an async audit service + When I record an event with invalid type "not_a_real_event" in async mode + Then a ValueError should be raised immediately + + Scenario: flush() is idempotent + Given an async audit service + When I flush the async audit service + And I flush the async audit service again + Then no async audit error should be raised + + Scenario: close() is idempotent + Given an async audit service + When I close the async audit service + And I close the async audit service again + Then no async audit error should be raised diff --git a/features/steps/async_audit_recording_steps.py b/features/steps/async_audit_recording_steps.py new file mode 100644 index 00000000..2465defc --- /dev/null +++ b/features/steps/async_audit_recording_steps.py @@ -0,0 +1,348 @@ +"""Step definitions for async audit recording (issue #718). + +Tests the write-behind queue, non-blocking record(), flush(), and +graceful shutdown behaviour of :class:`AuditService` in async mode. + +Design note: async mode is only active when the service owns its +session (no injected session). Tests therefore create the service +with a ``database_url`` pointing to a temporary SQLite file so that +the background writer thread can open its own connection. +""" + +from __future__ import annotations + +import tempfile +import time + +from behave import given, then, when +from behave.runner import Context +from sqlalchemy import create_engine + +from cleveragents.application.services.audit_service import AuditService +from cleveragents.config.settings import Settings +from cleveragents.infrastructure.database.models import Base + +__all__: list[str] = [] + +# ── Helpers ─────────────────────────────────────────────────────── + + +def _make_settings(**overrides: object) -> Settings: + """Create a fresh Settings instance with optional overrides.""" + Settings._instance = None + base = Settings() + if overrides: + return base.model_copy(update=overrides) + return base + + +def _make_async_service(context: Context) -> AuditService: + """Create an AuditService in async mode using a temp SQLite file. + + The service owns its session so the background writer thread can + open its own connection to the same file. + """ + tmp = tempfile.mktemp(suffix=".db", prefix="audit_async_test_") + context._async_db_path = tmp + db_url = f"sqlite:///{tmp}" + settings = _make_settings(audit_async=True, database_url=db_url) + # Pre-create the table so the background thread doesn't race. + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine, tables=[]) + from cleveragents.infrastructure.database.models import AuditLogModel + Base.metadata.create_all(engine, tables=[AuditLogModel.__table__]) + engine.dispose() + return AuditService(settings=settings, database_url=db_url) + + +def _make_sync_service(context: Context) -> AuditService: + """Create an AuditService in synchronous mode using a temp SQLite file.""" + tmp = tempfile.mktemp(suffix=".db", prefix="audit_sync_test_") + context._sync_db_path = tmp + db_url = f"sqlite:///{tmp}" + settings = _make_settings(audit_async=False, database_url=db_url) + from cleveragents.infrastructure.database.models import AuditLogModel + engine = create_engine(db_url) + Base.metadata.create_all(engine, tables=[AuditLogModel.__table__]) + engine.dispose() + return AuditService(settings=settings, database_url=db_url) + + +# ── Given ───────────────────────────────────────────────────────── + + +@given("an async audit service") +def step_async_audit_service(context: Context) -> None: + context.async_service = _make_async_service(context) + context.async_entry = None + context.async_error = None + context.async_closed = False + + +@given("an async audit service used as context manager") +def step_async_audit_service_ctx(context: Context) -> None: + context.async_service = _make_async_service(context) + context.async_entry = None + context.async_error = None + context.async_closed = False + context.ctx_entry = None + + +@given("a synchronous audit service") +def step_sync_audit_service(context: Context) -> None: + context.async_service = _make_sync_service(context) + context.async_entry = None + context.async_error = None + + +# ── When ────────────────────────────────────────────────────────── + + +@when('I record a "{event_type}" event in async mode') +def step_record_async(context: Context, event_type: str) -> None: + start = time.monotonic() + context.async_entry = context.async_service.record(event_type=event_type) + context.record_elapsed = time.monotonic() - start + + +@when('I record {count:d} "{event_type}" events in async mode') +def step_record_multiple_async(context: Context, count: int, event_type: str) -> None: + for i in range(count): + context.async_service.record( + event_type=event_type, plan_id=f"plan-async-{i}" + ) + + +@when("I flush the async audit service") +def step_flush_async(context: Context) -> None: + context.async_error = None + try: + context.async_service.flush() + except Exception as exc: + context.async_error = exc + + +@when("I flush the async audit service again") +def step_flush_async_again(context: Context) -> None: + try: + context.async_service.flush() + except Exception as exc: + context.async_error = exc + + +@when("I close the async audit service") +def step_close_async(context: Context) -> None: + context.async_error = None + context.async_closed = True + try: + context.async_service.close() + except Exception as exc: + context.async_error = exc + + +@when("I close the async audit service again") +def step_close_async_again(context: Context) -> None: + try: + context.async_service.close() + except Exception as exc: + context.async_error = exc + + +@when('I record a "{event_type}" event synchronously') +def step_record_sync(context: Context, event_type: str) -> None: + context.async_entry = context.async_service.record(event_type=event_type) + + +@when('I record a "{event_type}" event inside the context manager') +def step_record_inside_ctx(context: Context, event_type: str) -> None: + with context.async_service as svc: + context.ctx_entry = svc.record(event_type=event_type) + context.async_closed = True + + +@when( + 'I record events with plan_ids "{p1}" "{p2}" "{p3}" in async mode' +) +def step_record_ordered_async( + context: Context, p1: str, p2: str, p3: str +) -> None: + for pid in (p1, p2, p3): + context.async_service.record(event_type="plan_applied", plan_id=pid) + context.ordered_plan_ids = [p1, p2, p3] + + +@when( + 'I record an event with invalid type "{event_type}" in async mode' +) +def step_record_invalid_async(context: Context, event_type: str) -> None: + try: + context.async_service.record(event_type=event_type) + context.async_error = None + except ValueError as exc: + context.async_error = exc + + +# ── Then ────────────────────────────────────────────────────────── + + +@then("the record call should return immediately") +def step_record_returned_immediately(context: Context) -> None: + # "Immediately" is defined as < 100 ms — a synchronous SQLite write + # typically takes 5-50 ms; the async enqueue should be < 1 ms. + assert context.record_elapsed < 0.1, ( + f"record() took {context.record_elapsed:.3f}s — expected < 0.1s" + ) + + +@then('the returned entry should have event_type "{expected}"') +def step_returned_entry_event_type(context: Context, expected: str) -> None: + assert context.async_entry is not None + assert context.async_entry.event_type == expected, ( + f"Expected event_type={expected!r}, got {context.async_entry.event_type!r}" + ) + + +@then("the returned entry id should be -1") +def step_returned_entry_id_minus_one(context: Context) -> None: + assert context.async_entry is not None + assert context.async_entry.id == -1, ( + f"Expected id=-1 (async placeholder), got {context.async_entry.id}" + ) + + +@then("the returned entry id should be a positive integer") +def step_returned_entry_id_positive(context: Context) -> None: + assert context.async_entry is not None + assert context.async_entry.id > 0, ( + f"Expected positive id (sync mode), got {context.async_entry.id}" + ) + + +@then("the audit log should contain {count:d} persisted entry") +@then("the audit log should contain {count:d} persisted entries") +def step_persisted_count(context: Context, count: int) -> None: + actual = context.async_service.count() + assert actual == count, f"Expected {count} persisted entries, got {actual}" + + +@then("the audit log should contain {count:d} persisted entries after close") +def step_persisted_count_after_close(context: Context, count: int) -> None: + # After close() we need a fresh read session to verify persistence. + db_path = getattr(context, "_async_db_path", None) + if db_path: + from sqlalchemy import create_engine as _ce + from sqlalchemy.orm import sessionmaker as _sm + + from cleveragents.infrastructure.database.models import AuditLogModel as _M + engine = _ce(f"sqlite:///{db_path}") + session = _sm(bind=engine)() + actual = session.query(_M).count() + session.close() + engine.dispose() + else: + actual = context.async_service.count() + assert actual == count, ( + f"Expected {count} persisted entries after close, got {actual}" + ) + + +@then("the audit log should contain 1 persisted entry after context exit") +def step_persisted_after_ctx_exit(context: Context) -> None: + db_path = getattr(context, "_async_db_path", None) + if db_path: + from sqlalchemy import create_engine as _ce + from sqlalchemy.orm import sessionmaker as _sm + + from cleveragents.infrastructure.database.models import AuditLogModel as _M + engine = _ce(f"sqlite:///{db_path}") + session = _sm(bind=engine)() + actual = session.query(_M).count() + session.close() + engine.dispose() + else: + actual = context.async_service.count() + assert actual == 1, ( + f"Expected 1 persisted entry after context exit, got {actual}" + ) + + +@then("the audit log should contain 1 persisted entry immediately") +def step_persisted_immediately(context: Context) -> None: + actual = context.async_service.count() + assert actual == 1, ( + f"Expected 1 persisted entry immediately (sync mode), got {actual}" + ) + + +@then("the background writer thread should be alive") +def step_writer_thread_alive(context: Context) -> None: + thread = context.async_service._writer_thread + assert thread is not None, "Expected a background writer thread" + assert thread.is_alive(), "Expected background writer thread to be alive" + + +@then("the background writer thread should be stopped") +def step_writer_thread_stopped(context: Context) -> None: + thread = context.async_service._writer_thread + assert thread is not None, "Expected a background writer thread" + assert not thread.is_alive(), "Expected background writer thread to be stopped" + + +@then("the entries should be persisted in enqueue order") +def step_entries_in_order(context: Context) -> None: + entries = context.async_service.list_entries( + event_type="plan_applied", limit=100 + ) + # list_entries returns newest-first; reverse to get enqueue order. + persisted_ids = [e.plan_id for e in reversed(entries)] + assert persisted_ids == context.ordered_plan_ids, ( + f"Expected order {context.ordered_plan_ids}, got {persisted_ids}" + ) + + +@then("a ValueError should be raised immediately") +def step_value_error_raised(context: Context) -> None: + assert isinstance(context.async_error, ValueError), ( + f"Expected ValueError, got {type(context.async_error)}" + ) + + +@then("no async audit error should be raised") +def step_no_async_audit_error_raised(context: Context) -> None: + assert context.async_error is None, ( + f"Expected no error, got {context.async_error}" + ) + + +# ── Settings steps ──────────────────────────────────────────────── + + +@then("audit_async should be True") +def step_audit_async_true(context: Context) -> None: + assert context.settings.audit_async is True, ( + f"Expected audit_async=True, got {context.settings.audit_async}" + ) + + +@then("audit_async should be False") +def step_audit_async_false(context: Context) -> None: + assert context.settings.audit_async is False, ( + f"Expected audit_async=False, got {context.settings.audit_async}" + ) + + +@then("audit_queue_maxsize should be {expected:d}") +def step_audit_queue_maxsize(context: Context, expected: int) -> None: + assert context.settings.audit_queue_maxsize == expected, ( + f"Expected audit_queue_maxsize={expected}, " + f"got {context.settings.audit_queue_maxsize}" + ) + + +# ── Settings Given steps ────────────────────────────────────────── + + +@given("settings with audit_async False") +def step_settings_audit_async_false(context: Context) -> None: + Settings._instance = None + context.settings = Settings(audit_async=False) diff --git a/features/steps/audit_service_wiring_steps.py b/features/steps/audit_service_wiring_steps.py index 9b961935..73517d33 100644 --- a/features/steps/audit_service_wiring_steps.py +++ b/features/steps/audit_service_wiring_steps.py @@ -32,7 +32,7 @@ __all__: list[str] = [] def _make_settings() -> Settings: """Create a fresh Settings instance for testing.""" Settings._instance = None - return Settings(database_url="sqlite:///:memory:") + return Settings(database_url="sqlite:///:memory:", audit_async=False) def _fresh_audit_service() -> AuditService: diff --git a/features/steps/security_audit_steps.py b/features/steps/security_audit_steps.py index 8f5fbd34..fcc0a511 100644 --- a/features/steps/security_audit_steps.py +++ b/features/steps/security_audit_steps.py @@ -35,7 +35,7 @@ def _make_settings(**overrides: object) -> Settings: def _fresh_service(settings: Settings | None = None) -> AuditService: """Create an AuditService backed by an in-memory SQLite database.""" if settings is None: - settings = _make_settings(database_url="sqlite:///:memory:") + settings = _make_settings(database_url="sqlite:///:memory:", audit_async=False) engine = create_engine("sqlite:///:memory:", echo=False) Base.metadata.create_all(engine) session = sessionmaker(bind=engine)() @@ -508,7 +508,7 @@ def step_audit_error_message_contains(context: Context, text: str) -> None: @given("a fresh audit service with owned session") def step_fresh_owned_session(context: Context) -> None: - settings = _make_settings(database_url="sqlite:///:memory:") + settings = _make_settings(database_url="sqlite:///:memory:", audit_async=False) context.service = AuditService(settings=settings) context.entry = None context.entries = [] diff --git a/src/cleveragents/application/services/audit_service.py b/src/cleveragents/application/services/audit_service.py index 413383d7..41d15dbb 100644 --- a/src/cleveragents/application/services/audit_service.py +++ b/src/cleveragents/application/services/audit_service.py @@ -4,15 +4,49 @@ Records and queries security-relevant events such as plan applies, cancellations, resource modifications, corrections, config changes, and entity deletions. Supports filtering by plan, project, time range, and event type, plus a configurable retention/prune policy. + +Async write-behind (issue #718) +-------------------------------- +By default (``settings.audit_async=True``) ``record()`` enqueues the +audit payload onto an in-process :class:`queue.Queue` and returns +immediately, so the calling domain operation is never blocked by a +SQLite INSERT + COMMIT. A single daemon background thread drains the +queue and performs the actual database writes. + +Ordering guarantees +~~~~~~~~~~~~~~~~~~~ +* **Best-effort ordering** — entries are written in the order they are + enqueued, but because the background thread may lag behind the + producer, the ``created_at`` timestamps (set at enqueue time) may + arrive in the database slightly out of wall-clock order relative to + other concurrent writers. For the single-writer model used by this + service this is not a concern in practice. +* **No data loss under normal operation** — :meth:`close` / the context + manager call :meth:`flush` which joins the background thread after + sending a sentinel, ensuring all enqueued entries are persisted before + the session is closed. +* **Back-pressure** — when the queue reaches ``settings.audit_queue_maxsize`` + (default 10 000) ``record()`` blocks until space is available, preventing + unbounded memory growth under extreme load. + +Synchronous fallback +~~~~~~~~~~~~~~~~~~~~ +Set ``settings.audit_async=False`` (or ``CLEVERAGENTS_AUDIT_ASYNC=false``) +to restore the original synchronous behaviour. This is useful for +debugging, strict ordering requirements, or test scenarios that need +immediate persistence. """ from __future__ import annotations import json +import queue +import threading from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any, Self +import structlog from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker @@ -25,6 +59,8 @@ __all__ = [ "AuditService", ] +_logger = structlog.get_logger(__name__) + # Spec-defined event types (specification.md ~L39764). VALID_EVENT_TYPES: frozenset[str] = frozenset( { @@ -46,6 +82,9 @@ VALID_EVENT_TYPES: frozenset[str] = frozenset( # comparisons on the ``created_at`` TEXT column remain correct. _TIMESTAMP_FMT = "%Y-%m-%dT%H:%M:%S.%f" +# Sentinel object used to signal the background writer thread to stop. +_STOP_SENTINEL: object = object() + # ── Domain data class ──────────────────────────────────────────── @@ -77,6 +116,22 @@ class AuditLogEntry: } +# ── Internal payload type ──────────────────────────────────────── + + +@dataclass(slots=True) +class _AuditPayload: + """Serialised audit entry waiting to be written to the database.""" + + event_type: str + plan_id: str | None + project_name: str | None + actor_name: str | None + user_identity: str | None + details_json: str + created_at: str + + # ── Service ────────────────────────────────────────────────────── @@ -89,8 +144,24 @@ class AuditService: svc.record(event_type="plan_applied", ...) Args: - settings: Application settings (for DB URL and retention). + settings: Application settings (for DB URL, retention, and async + mode). session: Optional pre-built SQLAlchemy session (for testing). + database_url: Override the database URL from settings. + + Async mode (``settings.audit_async=True``, the default) + -------------------------------------------------------- + ``record()`` enqueues the payload and returns a *placeholder* + :class:`AuditLogEntry` whose ``id`` is ``-1`` (the real ``id`` is + assigned by SQLite after the background write). Callers that need + the real ``id`` immediately should either: + + * Use synchronous mode (``settings.audit_async=False``), or + * Call :meth:`flush` after ``record()`` to wait for the background + thread to drain the queue. + + The placeholder ``id=-1`` is intentional and documented — it signals + "pending persistence". All other fields are populated correctly. """ def __init__( @@ -113,6 +184,26 @@ class AuditService: # bypassing the template-DB fast-path in the test harness. self._session: Session | None = session + # ── Async write-behind setup ───────────────────────────── + # Async mode is only active when the service owns its session + # (i.e. no session was injected). When a session is injected + # (typically in tests) we fall back to synchronous mode because + # SQLite connections are thread-local and cannot be shared across + # the background writer thread and the calling thread. + self._async_mode: bool = settings.audit_async and session is None + self._queue: queue.Queue[_AuditPayload | object] | None = None + self._writer_thread: threading.Thread | None = None + self._closed: bool = False + + if self._async_mode: + self._queue = queue.Queue(maxsize=settings.audit_queue_maxsize) + self._writer_thread = threading.Thread( + target=self._writer_loop, + name="audit-writer", + daemon=True, + ) + self._writer_thread.start() + # ── Lazy session bootstrap ─────────────────────────────────── def _ensure_session(self) -> Session: @@ -142,6 +233,53 @@ class AuditService: self._session = factory() return self._session + # ── Background writer loop ─────────────────────────────────── + + def _writer_loop(self) -> None: + """Drain the write-behind queue and persist entries to SQLite. + + Runs on the ``audit-writer`` daemon thread. Exits when it + receives :data:`_STOP_SENTINEL` from :meth:`flush`. + + Errors during individual writes are logged and swallowed so that + a transient DB error does not kill the background thread. + """ + assert self._queue is not None # always set when async_mode is True + while True: + item = self._queue.get() + if item is _STOP_SENTINEL: + self._queue.task_done() + break + if not isinstance(item, _AuditPayload): + self._queue.task_done() + continue + try: + self._write_payload(item) + except Exception as exc: + _logger.warning( + "audit_async_write_failed", + event_type=item.event_type, + error_type=type(exc).__name__, + error_message=str(exc), + ) + finally: + self._queue.task_done() + + def _write_payload(self, payload: _AuditPayload) -> None: + """Persist a single :class:`_AuditPayload` to the database.""" + row = AuditLogModel( + event_type=payload.event_type, + plan_id=payload.plan_id, + project_name=payload.project_name, + actor_name=payload.actor_name, + user_identity=payload.user_identity, + details=payload.details_json, + created_at=payload.created_at, + ) + session = self._ensure_session() + session.add(row) + session.commit() + # ── Context manager ────────────────────────────────────────── def __enter__(self) -> Self: @@ -150,12 +288,41 @@ class AuditService: def __exit__(self, *_exc: object) -> None: self.close() + def flush(self) -> None: + """Block until all enqueued audit entries have been persisted. + + In synchronous mode this is a no-op. In async mode it sends the + stop sentinel to the background thread and joins it, ensuring + every entry that was enqueued before this call has been written + to the database. + + After ``flush()`` the service is **closed** — no further + ``record()`` calls should be made. This matches the semantics + of :meth:`close`. + """ + if ( + self._async_mode + and self._queue is not None + and self._writer_thread is not None + and self._writer_thread.is_alive() + ): + self._queue.put(_STOP_SENTINEL) + self._writer_thread.join() + def close(self) -> None: - """Close the underlying SQLAlchemy session. + """Flush the write-behind queue and close the SQLAlchemy session. + + In async mode, waits for the background thread to drain the queue + before closing the session, ensuring no data loss. Only closes sessions that were created by this service (i.e. not externally-provided test sessions). Safe to call multiple times. """ + if self._closed: + return + self._closed = True + # Drain the queue before closing the session. + self.flush() if self._owns_session and self._session is not None: self._session.close() @@ -173,12 +340,22 @@ class AuditService: ) -> AuditLogEntry: """Write an audit event to the database. + In **async mode** (default) the entry is enqueued immediately and + a placeholder :class:`AuditLogEntry` with ``id=-1`` is returned. + The actual database write happens on the background thread. + + In **synchronous mode** (``settings.audit_async=False``) the + behaviour is identical to the original implementation: the entry + is written and committed before this method returns, and the + returned entry has the real database-assigned ``id``. + Raises: ValueError: If *event_type* is not one of the spec-defined event types in :data:`VALID_EVENT_TYPES`. Returns: - The created :class:`AuditLogEntry` with its assigned ``id``. + The created :class:`AuditLogEntry`. In async mode the ``id`` + field is ``-1`` (pending persistence). """ if event_type not in VALID_EVENT_TYPES: raise ValueError( @@ -188,6 +365,30 @@ class AuditService: now = datetime.now(tz=UTC).strftime(_TIMESTAMP_FMT) details_json = json.dumps(details or {}, default=str) + if self._async_mode and self._queue is not None: + payload = _AuditPayload( + event_type=event_type, + plan_id=plan_id, + project_name=project_name, + actor_name=actor_name, + user_identity=user_identity, + details_json=details_json, + created_at=now, + ) + self._queue.put(payload) + # Return a placeholder entry — id is not yet known. + return AuditLogEntry( + id=-1, + event_type=event_type, + plan_id=plan_id, + project_name=project_name, + actor_name=actor_name, + user_identity=user_identity, + details=details or {}, + created_at=now, + ) + + # Synchronous path (original behaviour). row = AuditLogModel( event_type=event_type, plan_id=plan_id, diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index e361b621..80a8e69d 100644 --- a/src/cleveragents/config/settings.py +++ b/src/cleveragents/config/settings.py @@ -189,6 +189,31 @@ class Settings(BaseSettings): ), ) + # Async audit recording (issue #718) + audit_async: bool = Field( + default=True, + validation_alias=AliasChoices("CLEVERAGENTS_AUDIT_ASYNC"), + description=( + "When True (default), audit entries are written asynchronously " + "via a write-behind queue on a background thread, so that " + "AuditService.record() does not block the calling domain " + "operation. Set to False to restore synchronous behaviour " + "(useful for debugging or when strict ordering is required)." + ), + ) + + audit_queue_maxsize: int = Field( + default=10_000, + ge=1, + validation_alias=AliasChoices("CLEVERAGENTS_AUDIT_QUEUE_MAXSIZE"), + description=( + "Maximum number of pending audit entries in the write-behind " + "queue. When the queue is full, record() blocks until space " + "is available, providing back-pressure. Increase this value " + "for very high-throughput workloads." + ), + ) + # Per-session and per-org cost budgets (M6+ #584) session_max_cost_usd: float | None = Field( default=None,