fix(capsule): one transient error no longer permanently kills the audit trail

CapsuleStorage._ensure_db assigned self._engine BEFORE running create_all, and
gated re-entry on `if self._engine is None`. So any failure inside create_all (a
locked db file, a full disk, an engine pool bound to a since-closed event loop)
left the instance holding an engine but no session factory. Every later call then
short-circuited, did nothing, and _get_session_factory() raised "Database not
initialized -- call _ensure_db() first" for the entire life of the process, while
_ensure_db() was in fact being called every time. One transient error silently
and permanently disabled capsule persistence, and the error message actively
misled whoever read it.

Under "Capsule everything" (Trust Foundation layer 5, the immutable audit trail),
a silently dead audit log is a security failure, not an availability one.

Initialization is now atomic. Readiness keys off _session_factory, which is what
callers actually need; the engine and the factory publish together and only after
create_all succeeds; a failure disposes the partial engine and leaves both unset,
so the next call retries from scratch. A concurrent initializer is handled by
keeping theirs and disposing ours rather than leaking an engine.

Author: bradleygauthier

reference/python/src/qp_capsule/storage.py

@@ -105,24 +105,58 @@ def _get_session_factory(self) -> async_sessionmaker[AsyncSession]:
         return self._session_factory
 
     async def _ensure_db(self) -> None:
-        """Initialize database if needed."""
-        if self._engine is None:
-            self.db_path.parent.mkdir(parents=True, exist_ok=True)
-
-            self._engine = create_async_engine(
-                f"sqlite+aiosqlite:///{self.db_path}",
-                echo=False,
-                **self._engine_kwargs,
-            )
+        """Initialize the database if needed. Atomic: all-or-nothing.
+
+        Readiness is keyed off ``_session_factory`` (what callers actually need),
+        and the engine plus the factory are published together, only after
+        ``create_all`` has succeeded.
+
+        The previous version assigned ``self._engine`` BEFORE creating the schema
+        and gated on ``if self._engine is None``. So any failure inside
+        ``create_all`` (a locked db file, a full disk, an engine whose pool is
+        bound to an event loop that has since closed) left the instance with an
+        engine but no session factory. Every later call then short-circuited,
+        returned without doing anything, and ``_get_session_factory()`` raised
+        "Database not initialized — call _ensure_db() first" for the entire life
+        of the process, even though ``_ensure_db()`` was being called every time.
+        One transient error permanently disabled capsule persistence, and the
+        audit trail went silently dead. It is now self-healing: on failure both
+        attributes stay unset and the next call retries from scratch.
 
-            async with self._engine.begin() as conn:
-                await conn.run_sync(Base.metadata.create_all)
+        Raises:
+            Exception: Whatever the underlying engine/schema creation raised. The
+                partially built engine is disposed first, so nothing is retained.
+        """
+        if self._session_factory is not None:
+            return
 
-            self._session_factory = async_sessionmaker(
-                self._engine,
-                class_=AsyncSession,
-                expire_on_commit=False,
-            )
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+
+        engine = create_async_engine(
+            f"sqlite+aiosqlite:///{self.db_path}",
+            echo=False,
+            **self._engine_kwargs,
+        )
+
+        try:
+            async with engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+        except Exception:
+            await engine.dispose()
+            raise
+
+        # A concurrent caller may have finished initializing while this one was
+        # building. Keep theirs and drop ours rather than leaking an engine.
+        if self._session_factory is not None:
+            await engine.dispose()
+            return
+
+        self._engine = engine
+        self._session_factory = async_sessionmaker(
+            engine,
+            class_=AsyncSession,
+            expire_on_commit=False,
+        )
 
     async def store(self, capsule: Capsule, tenant_id: str | None = None) -> Capsule:
         """