test: add TDD bug-capture test for #991 — AuditService TOCTOU race

Add a Behave scenario that proves the TOCTOU race condition in
AuditService._ensure_session() described in bug #991. The method checks
`if self._session is None` then creates an engine, runs create_all,
builds a sessionmaker, and assigns self._session — all without holding
a threading.Lock. Concurrent threads each see _session as None and
create duplicate engines, leaking all but the last.

The test launches 10 threads synchronized via threading.Barrier that
all call _ensure_session() simultaneously. Race detection uses
unittest.mock.patch on create_engine within the audit_service module
to count actual invocations — this is deterministic and avoids the
timing-dependent was_none proxy. Without the fix, create_engine is
called multiple times (mock.call_count > 1). The assertion
`engine_create_count == 1` fails, proving the race exists.

Key design decisions:
- mock.patch wraps real create_engine (side_effect=_real_create_engine)
  so the database is still created but calls are counted deterministically
- Daemon threads prevent test process hangs; is_alive() check after join
  catches deadlocks
- Engine/session cleanup handler prevents SQLAlchemy resource leaks
- Thread completion assertion checks total == thread_count (not > 0)
- Tags at Feature level per project convention (@tdd_expected_fail
  @tdd_bug @tdd_bug_991)

Tagged per the TDD Bug Fix Workflow in CONTRIBUTING.md. When #991
is fixed by adding a threading.Lock, the @tdd_expected_fail tag must
be removed and the test will run as a permanent regression guard.

Robot Framework test: N/A — purely unit-level thread-safety bug with
no integration boundaries.

Quality gates: lint ✅, typecheck ✅, unit_tests ✅ (462 features,
12233 scenarios), coverage_report ✅ (98%, threshold 97%).

ISSUES CLOSED: #1095

features/audit_session_race.feature

@@ -0,0 +1,27 @@
+# This test captures bug #991 — AuditService._ensure_session() TOCTOU race.
+#
+# The @tdd_expected_fail tag inverts the test result: the underlying assertion
+# fails (proving the bug exists) but CI reports the scenario as passed.
+# When #991 is fixed, the @tdd_expected_fail tag must be removed.
+#
+# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/991
+
+@tdd_expected_fail @tdd_bug @tdd_bug_991
+Feature: TDD Bug #991 — AuditService._ensure_session() TOCTOU race
+  AuditService._ensure_session() checks ``if self._session is None``, then
+  creates an engine, runs create_all, builds a sessionmaker, and assigns
+  self._session — all without a threading.Lock.  If two threads call
+  _ensure_session() concurrently, both see self._session is None, both
+  create separate engines and sessions, and the second assignment wins,
+  leaking the first engine/session.
+
+  This test proves the race exists by launching multiple threads that call
+  _ensure_session() simultaneously via a threading.Barrier and verifying
+  that create_engine is called exactly once (the correct, thread-safe
+  behavior).  Without the fix, create_engine is called multiple times.
+
+  Scenario: Concurrent _ensure_session() calls must create only one engine
+    Given an AuditService with no pre-injected session and a temp database
+    When 10 threads call _ensure_session concurrently through a barrier
+    Then create_engine should have been called exactly once
+    And all 10 threads should have received a valid session

features/steps/audit_session_race_steps.py

@@ -0,0 +1,210 @@
+"""Step definitions for bug #991 — AuditService._ensure_session() TOCTOU race.
+
+This module provides the Behave step implementations for the TDD bug-capture
+test that proves bug #991 exists.  The bug is a Time-Of-Check-to-Time-Of-Use
+(TOCTOU) race in ``AuditService._ensure_session()``: the method checks
+``if self._session is None`` and then creates an engine, runs ``create_all``,
+builds a ``sessionmaker``, and assigns ``self._session`` — all without holding
+a ``threading.Lock``.  Concurrent callers can each see ``_session is None``
+and create duplicate engines, leaking all but the last.
+
+The test uses ``@tdd_expected_fail`` so that CI passes while the bug is
+unfixed.  The underlying assertion ("create_engine called exactly once")
+fails because the race allows multiple threads to trigger creation.  The
+tag inversion causes the scenario to be reported as passed.
+
+Race detection uses ``unittest.mock.patch`` on ``create_engine`` within the
+``audit_service`` module to count actual calls.  This is deterministic and
+avoids the timing-dependent ``was_none`` proxy that cannot distinguish
+between "thread observed None then raced" and "thread correctly entered the
+creation branch".
+
+When bug #991 is fixed (by adding a ``threading.Lock``), the assertion will
+pass and the ``@tdd_expected_fail`` tag must be removed from the feature file.
+
+See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/991
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import tempfile
+import threading
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine as _real_create_engine
+
+from cleveragents.application.services.audit_service import AuditService
+from cleveragents.config.settings import Settings
+
+
+def _make_settings(**overrides: str) -> Settings:
+    """Create a Settings instance with optional field overrides.
+
+    Resets the Settings singleton so a fresh instance is constructed.
+    This follows the established project pattern used in environment.py
+    and dozens of other step files.
+    """
+    Settings._instance = None
+    base = Settings()
+    if overrides:
+        return base.model_copy(update=overrides)
+    return base
+
+
+# ── Given ─────────────────────────────────────────────────────────
+
+
+@given("an AuditService with no pre-injected session and a temp database")
+def step_audit_service_no_session(context: Context) -> None:
+    """Create an AuditService that will use _ensure_session() for lazy init.
+
+    A file-based temp SQLite database is used (not :memory:) because each
+    ``create_engine("sqlite:///:memory:")`` call produces an independent
+    in-memory database, which would mask the race condition.  A shared
+    file ensures all threads contend on the same database.
+    """
+    fd, db_path = tempfile.mkstemp(suffix=".db", prefix="test_race_991_")
+    os.close(fd)
+    # Remove the empty file so create_engine creates it fresh
+    os.unlink(db_path)
+    context.db_path = db_path
+    db_url = f"sqlite:///{db_path}"
+
+    settings = _make_settings(database_url=db_url)
+    # Do NOT pass session= so that _ensure_session() performs lazy init
+    context.service = AuditService(settings=settings, database_url=db_url)
+    context.thread_sessions: list[object] = []
+    context.thread_errors: list[Exception] = []
+    context.engine_create_count = 0
+
+    # Register cleanup to dispose engine/session and remove temp database
+    # files (M4: prevent SQLAlchemy resource leaks).
+    def _cleanup_db(path: str = db_path) -> None:
+        svc: AuditService = context.service
+        if svc._session is not None:
+            svc._session.close()
+            if svc._session.bind is not None:
+                svc._session.bind.dispose()
+        for suffix in ("", "-journal", "-wal", "-shm"):
+            with contextlib.suppress(OSError):
+                os.unlink(path + suffix)
+
+    context._cleanup_handlers.append(_cleanup_db)
+
+
+# ── When ──────────────────────────────────────────────────────────
+
+
+@when("{n:d} threads call _ensure_session concurrently through a barrier")
+def step_concurrent_ensure_session(context: Context, n: int) -> None:
+    """Launch *n* threads that all call ``_ensure_session()`` simultaneously.
+
+    A ``threading.Barrier`` synchronises the threads so they all enter
+    ``_ensure_session()`` at (approximately) the same instant, maximising
+    the chance of the TOCTOU race manifesting.
+
+    Race detection uses ``unittest.mock.patch`` on ``create_engine`` within
+    the ``audit_service`` module to count actual invocations.  The mock
+    wraps the real ``create_engine`` so the database is still created, but
+    we can inspect ``mock.call_args_list`` afterward for a deterministic race
+    indicator.
+    """
+    barrier = threading.Barrier(n)
+    sessions: list[object] = []
+    errors: list[Exception] = []
+    collect_lock = threading.Lock()
+
+    # Wrap create_engine so real engines are created but calls are counted
+    engine_mock = MagicMock(side_effect=_real_create_engine)
+
+    def worker() -> None:
+        """Thread worker: wait on barrier, then call _ensure_session()."""
+        try:
+            barrier.wait(timeout=10)
+            session = context.service._ensure_session()
+            with collect_lock:
+                sessions.append(session)
+        except Exception as exc:
+            with collect_lock:
+                errors.append(exc)
+
+    with patch(
+        "cleveragents.application.services.audit_service.create_engine",
+        engine_mock,
+    ):
+        threads = [threading.Thread(target=worker, daemon=True) for _ in range(n)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=30)
+
+        # M3: Verify no threads are still alive after join timeout
+        for t in threads:
+            assert not t.is_alive(), (
+                f"Thread {t.name} is still alive after join timeout — "
+                f"possible deadlock in _ensure_session()"
+            )
+
+        barrier_errors = [
+            err for err in errors if isinstance(err, threading.BrokenBarrierError)
+        ]
+        assert not barrier_errors, (
+            "Barrier synchronization failed while coordinating concurrent "
+            "_ensure_session() calls. This invalidates race setup. "
+            f"Barrier errors: {barrier_errors}; all errors: {errors}"
+        )
+
+        context.engine_create_count = len(engine_mock.call_args_list)
+
+    context.thread_sessions = sessions
+    context.thread_errors = errors
+    context.thread_count = n
+
+
+# ── Then ──────────────────────────────────────────────────────────
+
+
+@then("create_engine should have been called exactly once")
+def step_create_engine_called_once(context: Context) -> None:
+    """Assert create_engine was called exactly once.
+
+    With the bug present (no threading.Lock), multiple threads will enter
+    the ``if self._session is None`` branch and each call ``create_engine``,
+    so the mock's call list length will be > 1.
+
+    This assertion therefore **fails** while the bug exists — which is
+    the expected behavior for a ``@tdd_expected_fail`` test.
+
+    When the fix adds a lock, only the first thread calls create_engine
+    and the count is exactly 1.
+    """
+    count = context.engine_create_count
+    assert count == 1, (
+        f"create_engine was called {count} time(s), expected exactly 1. "
+        f"This confirms the TOCTOU race in _ensure_session() — multiple "
+        f"threads saw _session as None because there is no threading.Lock. "
+        f"Thread errors (if any): {context.thread_errors}"
+    )
+
+
+@then("all {n:d} threads should have received a valid session")
+def step_all_threads_got_session(context: Context, n: int) -> None:
+    """Verify that all *n* threads completed — either with a session or error.
+
+    The total of successful sessions plus errors must equal the number of
+    threads launched (M2: non-tautological assertion).  This confirms no
+    threads silently hung or crashed.  When the bug is fixed, all threads
+    will succeed with no errors.
+    """
+    total = len(context.thread_sessions) + len(context.thread_errors)
+    assert total == n, (
+        f"Expected {n} threads to complete but only {total} did "
+        f"({len(context.thread_sessions)} succeeded, "
+        f"{len(context.thread_errors)} errored). "
+        "Each worker should report exactly one outcome (session or error). "
+        "A mismatch indicates a hung thread or missing outcome capture."
+    )