feat: add schema v1 ddl for presence schedules and overrides

custom_components/nestquest/__init__.py

@@ -10,6 +10,9 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 
 from .const import DOMAIN, LOGGER
+from .db import NestQuestDatabase
+from .schema import SCHEMA_V1_STATEMENTS, async_apply_ddl
+from .store import async_get_db_path
 
 _LOGGER = LOGGER
 
@@ -20,6 +23,7 @@ class NestQuestRuntimeData:
 
     entry_id: str
     options: dict[str, Any]
+    database: NestQuestDatabase
     remove_update_listener: Callable[[], Any]
 
 
@@ -31,6 +35,14 @@ async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up NestQuest from a config entry."""
     hass.data.setdefault(DOMAIN, {})
+    existing = hass.data[DOMAIN].get(entry.entry_id)
+    if existing is not None:
+        db = getattr(existing, "database", None)
+        if db is None:
+            db = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
+            existing.database = db
+        entry.runtime_data = existing
+        return True
 
     async def _async_update_listener(
         listener_hass: HomeAssistant, listener_entry: ConfigEntry
@@ -38,10 +50,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         """Reload the entry when its options change."""
         await listener_hass.config_entries.async_reload(listener_entry.entry_id)
 
+    database = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
+    try:
+        await async_apply_ddl(database, SCHEMA_V1_STATEMENTS)
+        remove_update_listener = entry.add_update_listener(_async_update_listener)
+    except BaseException:
+        await database.close()
+        raise
     runtime_data = NestQuestRuntimeData(
         entry_id=entry.entry_id,
         options=dict(entry.options),
-        remove_update_listener=entry.add_update_listener(_async_update_listener),
+        database=database,
+        remove_update_listener=remove_update_listener,
     )
     entry.runtime_data = runtime_data
     hass.data[DOMAIN][entry.entry_id] = runtime_data
@@ -58,10 +78,22 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         runtime_data = getattr(entry, "runtime_data", None)
     if runtime_data is not None:
         remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
-        if remove_update_listener is not None:
-            result = remove_update_listener()
-            if inspect.isawaitable(result):
-                await result
+        unload_error: BaseException | None = None
+        try:
+            if remove_update_listener is not None:
+                result = remove_update_listener()
+                if inspect.isawaitable(result):
+                    await result
+        except BaseException as err:
+            unload_error = err
+        finally:
+            database = getattr(runtime_data, "database", None)
+            if database is not None:
+                await database.close()
+        if unload_error is not None:
+            if getattr(entry, "runtime_data", None) is not None:
+                entry.runtime_data = None
+            raise unload_error
     if getattr(entry, "runtime_data", None) is not None:
         entry.runtime_data = None
     return True

custom_components/nestquest/db.py

@@ -0,0 +1,422 @@
+"""Async SQLite connection wrapper for the NestQuest integration."""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+from collections.abc import AsyncIterator, Awaitable, Sequence
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from homeassistant.core import HomeAssistant
+
+
+@dataclass(frozen=True)
+class ExecutionResult:
+    """Plain-data outcome of one statement.
+
+    ``rowcount`` and ``lastrowid`` mirror the corresponding
+    :class:`sqlite3.Cursor` attributes; both are captured on the executor
+    thread so no sqlite3 object ever crosses back to the event loop.
+    """
+
+    rowcount: int
+    lastrowid: int | None
+
+
+class NestQuestDatabase:
+    """Async wrapper around a single SQLite connection.
+
+    Every sqlite3 call is delegated to the Home Assistant executor through
+    ``hass.async_add_executor_job`` so the event loop is never blocked, and
+    only plain Python data (row tuples and :class:`ExecutionResult`) is
+    returned to the event loop.
+
+    Every operation is serialized through one asyncio.Lock per wrapper.  A
+    ``transaction()`` holds that lock for its whole BEGIN..COMMIT span, so
+    interleaved coroutines cannot corrupt the transaction and concurrent
+    BEGINs are impossible; statements issued while a transaction is open
+    queue until it completes.  Concurrent transactions on one wrapper are
+    not supported: a second ``transaction()`` entered while one is active
+    raises :class:`RuntimeError`.  ``close()`` serializes on the same lock
+    and is idempotent.
+
+    Every connection-affecting executor job (connect, pragmas, statements,
+    transaction lifecycle and close) is awaited through the settlement
+    helper :meth:`_await_settled`: if the awaiting task is cancelled while
+    a job is in flight, the job still settles on the worker thread before
+    the cancellation propagates, so the wrapper lock is never released
+    while the executor thread is mid-job.
+    """
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the wrapper around a Home Assistant instance."""
+        self._hass = hass
+        self._conn: sqlite3.Connection | None = None
+        self._rowcount: int = -1
+        self._lastrowid: int | None = None
+        self._lock: asyncio.Lock | None = None
+        self._lock_loop: asyncio.AbstractEventLoop | None = None
+        self._in_transaction = False
+        self._tx_task: asyncio.Task | None = None
+
+    def _get_lock(self) -> asyncio.Lock:
+        """Return the per-wrapper lock, (re)bound to the running loop.
+
+        HA drives every wrapper call from one event loop, so the lock is
+        created once and reused; re-binding only when the running loop
+        differs keeps the wrapper usable from plain synchronous drivers
+        that run each call on a fresh loop.
+        """
+        loop = asyncio.get_running_loop()
+        if self._lock is None or self._lock_loop is not loop:
+            self._lock = asyncio.Lock()
+            self._lock_loop = loop
+        return self._lock
+
+    @property
+    def connected(self) -> bool:
+        """Return True while the underlying connection is open."""
+        return self._conn is not None
+
+    @property
+    def rowcount(self) -> int:
+        """Row count of the most recent statement, or -1 when unset."""
+        return self._rowcount
+
+    @property
+    def lastrowid(self) -> int | None:
+        """Row id of the most recent insert, or None when unset."""
+        return self._lastrowid
+
+    @asynccontextmanager
+    async def _serialized(self) -> AsyncIterator[None]:
+        """Hold the wrapper lock unless the caller owns the active transaction.
+
+        A transaction holds the lock for its whole BEGIN..COMMIT span;
+        statements issued by the transaction's own task are already
+        serialized by it and must not re-acquire the non-reentrant lock.
+        Any other task queues on the lock until the transaction ends.
+        """
+        if (
+            self._in_transaction
+            and self._tx_task is not None
+            and asyncio.current_task() is self._tx_task
+        ):
+            yield
+            return
+        async with self._get_lock():
+            yield
+
+    async def open(self, path: Path | str) -> NestQuestDatabase:
+        """Open the SQLite file with WAL journal mode and foreign keys on.
+
+        Cancellation-safe: if this call is cancelled while the connect job
+        is in flight, the job still settles (see :meth:`_await_settled`).
+        A connection the settled job created is never adopted by the
+        cancelled call; it is closed through a settled executor job instead
+        of being abandoned to garbage collection, and the cancellation
+        propagates afterwards, so a follow-up ``open()`` starts from a
+        clean state.
+        """
+        if self._conn is not None:
+            return self
+        async with self._get_lock():
+            if self._conn is not None:
+                return self
+
+            def _open() -> sqlite3.Connection:
+                conn = sqlite3.connect(
+                    path, check_same_thread=False, isolation_level=None
+                )
+                try:
+                    conn.execute("PRAGMA journal_mode=WAL")
+                    conn.execute("PRAGMA foreign_keys=ON")
+                except BaseException:
+                    conn.close()
+                    raise
+                return conn
+
+            settled, cancelled = await self._await_settled_state(
+                self._hass.async_add_executor_job(_open)
+            )
+            if cancelled:
+                if isinstance(settled, sqlite3.Connection):
+
+                    def _close_abandoned() -> None:
+                        settled.close()
+
+                    await self._await_settled(
+                        self._hass.async_add_executor_job(_close_abandoned)
+                    )
+                raise asyncio.CancelledError()
+            self._conn = settled
+        return self
+
+    async def execute(
+        self, sql: str, parameters: Sequence[Any] = ()
+    ) -> ExecutionResult:
+        """Run one statement in the executor and return its plain result."""
+        async with self._serialized():
+            conn = self._require_conn()
+
+            def _execute() -> ExecutionResult:
+                cursor = conn.execute(sql, parameters)
+                return ExecutionResult(cursor.rowcount, cursor.lastrowid)
+
+            result = await self._await_settled(
+                self._hass.async_add_executor_job(_execute)
+            )
+            self._rowcount = result.rowcount
+            self._lastrowid = result.lastrowid
+            return result
+
+    async def execute_many(
+        self, sql: str, parameters: Sequence[Sequence[Any]]
+    ) -> ExecutionResult:
+        """Run executemany in the executor and return its plain result."""
+        async with self._serialized():
+            conn = self._require_conn()
+
+            def _execute_many() -> ExecutionResult:
+                cursor = conn.executemany(sql, parameters)
+                return ExecutionResult(cursor.rowcount, cursor.lastrowid)
+
+            result = await self._await_settled(
+                self._hass.async_add_executor_job(_execute_many)
+            )
+            self._rowcount = result.rowcount
+            self._lastrowid = result.lastrowid
+            return result
+
+    async def fetch_one(
+        self, sql: str, parameters: Sequence[Any] = ()
+    ) -> tuple | None:
+        """Run a query in the executor and return its first row, or None."""
+        async with self._serialized():
+            conn = self._require_conn()
+
+            def _fetch_one() -> tuple[tuple | None, int, int | None]:
+                cursor = conn.execute(sql, parameters)
+                return cursor.fetchone(), cursor.rowcount, cursor.lastrowid
+
+            row, rowcount, lastrowid = await self._await_settled(
+                self._hass.async_add_executor_job(_fetch_one)
+            )
+            self._rowcount = rowcount
+            self._lastrowid = lastrowid
+            return row
+
+    async def fetch_all(
+        self, sql: str, parameters: Sequence[Any] = ()
+    ) -> list[tuple]:
+        """Run a query in the executor and return all of its rows."""
+        async with self._serialized():
+            conn = self._require_conn()
+
+            def _fetch_all() -> tuple[list[tuple], int, int | None]:
+                cursor = conn.execute(sql, parameters)
+                return cursor.fetchall(), cursor.rowcount, cursor.lastrowid
+
+            rows, rowcount, lastrowid = await self._await_settled(
+                self._hass.async_add_executor_job(_fetch_all)
+            )
+            self._rowcount = rowcount
+            self._lastrowid = lastrowid
+            return rows
+
+    @asynccontextmanager
+    async def transaction(self) -> AsyncIterator[None]:
+        """Run a block inside BEGIN/COMMIT, rolling back on any exception.
+
+        The wrapper lock is held for the entire BEGIN..COMMIT span, so no
+        other statement can interleave inside the transaction.  Entering a
+        second transaction while one is active raises RuntimeError.  A
+        failing COMMIT (e.g. a deferred foreign-key violation) also triggers
+        a ROLLBACK before the commit error propagates, so the real
+        connection is never left inside a stray transaction.
+
+        The lifecycle is cancellation-safe.  Transaction ownership and the
+        wrapper lock are only released after the executor work has fully
+        settled, even when the owning task is cancelled mid-flight: if a
+        cancelled BEGIN or COMMIT still completes on the worker thread,
+        the corresponding ROLLBACK is shielded and awaited before any state
+        is cleared.  A lifecycle statement that fails plus a failing
+        follow-up cleanup surface both errors in an ExceptionGroup.
+        """
+        if self._in_transaction:
+            raise RuntimeError(
+                "A NestQuest transaction is already active on this "
+                "connection; concurrent transactions are not supported"
+            )
+        async with self._get_lock():
+            conn = self._require_conn()
+            self._in_transaction = True
+            self._tx_task = asyncio.current_task()
+            try:
+
+                def _begin() -> None:
+                    conn.execute("BEGIN")
+
+                await self._settle_lifecycle(
+                    conn,
+                    self._hass.async_add_executor_job(_begin),
+                    "BEGIN did not settle cleanly and the follow-up "
+                    "cleanup also raised; the transaction may need manual "
+                    "cleanup",
+                )
+                try:
+                    yield
+                except BaseException:
+
+                    def _rollback() -> None:
+                        conn.execute("ROLLBACK")
+
+                    await self._settle_lifecycle(
+                        conn,
+                        self._hass.async_add_executor_job(_rollback),
+                        "ROLLBACK failed and the follow-up cleanup also "
+                        "raised; the transaction may need manual cleanup",
+                    )
+                    raise
+
+                def _commit() -> None:
+                    conn.execute("COMMIT")
+
+                await self._settle_lifecycle(
+                    conn,
+                    self._hass.async_add_executor_job(_commit),
+                    "COMMIT failed and the follow-up cleanup also raised; "
+                    "the transaction may need manual cleanup",
+                )
+            finally:
+                self._in_transaction = False
+                self._tx_task = None
+
+    async def _settle_lifecycle(
+        self, conn: sqlite3.Connection, job: Awaitable[Any], message: str
+    ) -> None:
+        """Run one BEGIN/ROLLBACK/COMMIT job, then force the connection clean.
+
+        The job always settles first (see :meth:`_await_settled`), so a
+        cancelled statement that the worker thread completed anyway is
+        never abandoned mid-flight.  If the job raised, a follow-up
+        ROLLBACK — itself settled — guarantees the connection ends outside
+        any SQLite transaction before the error propagates.  When both the
+        job and the cleanup raise, both errors are preserved in an
+        ExceptionGroup instead of the original being lost.
+        """
+        try:
+            await self._await_settled(job)
+        except BaseException as settle_error:
+            errors: list[BaseException] = [settle_error]
+            try:
+                await self._ensure_no_stray_transaction(conn)
+            except BaseException as cleanup_error:
+                errors.append(cleanup_error)
+            if len(errors) == 2:
+                raise BaseExceptionGroup(message, errors) from settle_error
+            raise
+
+    async def _ensure_no_stray_transaction(
+        self, conn: sqlite3.Connection
+    ) -> None:
+        """Force the connection out of any transaction via ROLLBACK.
+
+        Runs after a failed or cancelled lifecycle statement.  The worker
+        thread may still be finishing (or may already have completed) the
+        very statement whose await was cancelled, so both the check and
+        the ROLLBACK settle in the executor before the caller releases
+        transaction ownership.
+        """
+
+        def _cleanup() -> bool:
+            if conn.in_transaction:
+                conn.execute("ROLLBACK")
+            return conn.in_transaction
+
+        stray = await self._await_settled(
+            self._hass.async_add_executor_job(_cleanup)
+        )
+        if stray:
+            raise RuntimeError(
+                "The SQLite connection is still inside a transaction after "
+                "cleanup; the transaction may need manual cleanup"
+            )
+
+    async def _await_settled(self, job: Awaitable[Any]) -> Any:
+        """Await ``job``, letting it settle even if this task is cancelled.
+
+        Cancellation is observed, not swallowed: it is re-raised only once
+        the job has fully settled, so the caller can finalize against a
+        known connection state instead of abandoning in-flight work.  The
+        job is wrapped in a task because a bare coroutine dies un-awaited
+        once its await is cancelled.  A second cancellation during
+        settlement is absorbed the same way; if the job itself fails after
+        a cancellation, both errors surface in an ExceptionGroup.
+        """
+        result, cancelled = await self._await_settled_state(job)
+        if cancelled:
+            raise asyncio.CancelledError()
+        return result
+
+    async def _await_settled_state(self, job: Awaitable[Any]) -> tuple[Any, bool]:
+        """Await ``job`` to settlement and report ``(result, cancelled)``.
+
+        Same settlement guarantees as :meth:`_await_settled`, but returns
+        the settled outcome together with whether the awaiting task was
+        cancelled, so callers that must act on a settled result before
+        re-raising cancellation (e.g. closing an abandoned connection)
+        can do so without losing the result.
+        """
+        cancelled = False
+        inner = asyncio.ensure_future(job)
+        while not inner.done():
+            try:
+                await asyncio.shield(inner)
+            except asyncio.CancelledError:
+                cancelled = True
+        try:
+            result = inner.result()
+        except BaseException as job_error:
+            if cancelled:
+                raise BaseExceptionGroup(
+                    "executor job failed while settling after cancellation",
+                    [asyncio.CancelledError(), job_error],
+                ) from job_error
+            raise
+        return result, cancelled
+
+    async def close(self) -> None:
+        """Close the connection in the executor; safe to call repeatedly.
+
+        Serialized on the wrapper lock, so an in-flight operation or open
+        transaction always completes before the connection is closed.
+        """
+        if (
+            self._in_transaction
+            and self._tx_task is not None
+            and asyncio.current_task() is self._tx_task
+        ):
+            raise RuntimeError(
+                "Cannot close the NestQuest database from inside its own "
+                "transaction; exit the transaction first"
+            )
+        async with self._get_lock():
+            conn, self._conn = self._conn, None
+            self._rowcount, self._lastrowid = -1, None
+            if conn is None:
+                return
+
+            def _close() -> None:
+                conn.close()
+
+            await self._await_settled(self._hass.async_add_executor_job(_close))
+
+    def _require_conn(self) -> sqlite3.Connection:
+        if self._conn is None:
+            raise RuntimeError(
+                "The NestQuest database connection is closed; call open() first"
+            )
+        return self._conn

custom_components/nestquest/schema.py

@@ -0,0 +1,232 @@
+"""Declarative schema DDL for the NestQuest SQLite database.
+
+Every v1 table is declared here as plain SQL statements so later schema
+tasks can append their own DDL to :data:`SCHEMA_V1_STATEMENTS` without
+touching this module's tables.  All statements use
+``CREATE TABLE IF NOT EXISTS`` so application is idempotent; the versioned
+migration runner is a separate later concern.
+
+Timestamp policy: all timestamps are UTC ISO-8601 strings, set by the
+caller/DAO rather than a DB default, per the feature guardrails.  No
+``length(created_at) = 25`` CHECK is added: the exact string shape
+produced by callers is not fixed yet (``datetime.isoformat()`` may
+include microseconds, changing the length), so enforcing a fixed length
+here would reject valid values.  Only non-null is enforced for now.
+
+Date policy: ``start_date``/``end_date`` are ISO date strings (YYYY-MM-DD,
+no time component) compared lexically by the ``end_date >= start_date``
+CHECK; due_date/due_time string shapes are caller-validated by the rule
+engine and DAO layers, not by DDL.
+
+``weekday_set`` format: comma-separated weekday integers (CSV), 0 =
+Monday .. 6 = Sunday, e.g. ``'0,2,4'`` for Mon/Wed/Fri.  Chosen over JSON
+because it sorts, compares and parses trivially in SQL.  The format is
+enforced at the DB level: a CHECK matches ``weekday_set`` against GLOB
+patterns for element counts 1..7 of single digits 0-6 (GLOB has no
+repetition quantifier, so the element count is enumerated; more than 7
+elements is impossible for a weekday set).  ``''``, ``'x'``, ``'7'`` and
+``'0,x'`` are all rejected; NULL stays allowed for non-weekly rules.
+This keeps the schema layer storage-only — the DAO layer (tasks
+8c72300c+) does not need its own validator, it can rely on the CHECK.
+
+Type affinity: SQLite applies column affinity on insert, but CHECKs
+still observe text and fractional values ('abc' stays text, '1.5'
+becomes REAL), so every INTEGER column carrying rule semantics carries
+a ``typeof(...) = 'integer'`` guard in its CHECK.
+
+Rule-type coherence: two CHECKs enforce that ``weekly`` carries a
+non-null ``weekday_set``, ``monthly`` carries a ``day_of_month`` or an
+``nth_weekday``, ``yearly`` carries a ``month``, and ``daily``/``custom``
+carry none of the month-specific fields.  Relaxed deliberately: the DDL
+does NOT forbid ``month`` on weekly/monthly rules (an "every 2 weeks"
+rule or a monthly rule could later be pinned to a month) nor extra
+fields on weekly/monthly/yearly rules generally, so the CHECKs only
+reject what the rule engine definitively cannot express now.
+
+Presence pattern format: ``presence_schedules.pattern`` encodes the
+whole N-week repeating cycle in one TEXT column as pipe-separated
+segments, one per week of the cycle: segment ``i`` is a ``weekday_set``
+CSV (same single-digit 0-6 convention as ``schedule_rules.weekday_set``)
+for week ``i``, e.g. ``'0,2,4|1,3'`` = Mon/Wed/Fri in week 0, Tue/Thu in
+week 1.  An empty segment means the child is absent every day of that
+week; a wholly empty pattern with ``cycle_length_weeks = 1`` means
+absent every day of every week, which is distinct from having no
+schedule row at all (a child with no schedule is present every day, per
+the feature guardrails).  Chosen over JSON or per-week rows to stay
+consistent with the CSV ``weekday_set`` convention and fully checkable
+by plain CHECKs.  DB-level validation is pure CHECK, no triggers or
+custom functions: char-class GLOBs reject every character outside
+digits 0-6, comma and pipe; adjacency GLOBs reject empty elements
+(double comma, comma next to a pipe, leading/trailing comma) and
+two-digit runs; a further GLOB rejects contiguous runs of 8+ comma
+separated digits, which caps a segment at 7 elements — the GLOB cannot
+match across a ``|``, so cross-segment boundaries stay safe; and the
+segment count is pinned to ``cycle_length_weeks`` by
+``LENGTH(pattern) - LENGTH(REPLACE(pattern, '|', '')) + 1``.  Per
+weekday_set this is not full CSV-set normalisation (duplicates like
+``'0,0'`` or unsorted sets stay storable, as with ``schedule_rules``);
+the engine treats the segment as a set.  A
+one-row-per-``week_index`` normalization was considered and rejected:
+the flat single-column encoding keeps the done-condition's table shape
+and proved fully checkable with plain CHECKs.
+
+``cycle_length_weeks`` is capped at 4: 2 (alternating weeks) is the
+common case, and a small cap keeps anchor arithmetic and the encoded
+pattern human-readable.  Revisit the cap — and re-run the pattern
+validation matrix in tests — if a longer cycle is ever needed.
+``anchor_date`` is an ISO date string pinning the cycle: the week index
+of a date is (days since anchor) // 7 mod ``cycle_length_weeks``,
+computed by the presence engine (Feature 05) from anchor-date
+arithmetic, never ISO week parity.  Like other date columns it carries
+no shape CHECK, matching the established date policy.
+
+Presence overrides: ``presence_overrides.end_date`` is NOT NULL,
+unlike ``schedule_rules.end_date`` — an override is always a concrete
+date range; a single-day override stores the same date in both columns.
+``is_present`` 0/1 marks the child absent/present for the whole range;
+``note`` is optional free text.
+"""
+from __future__ import annotations
+
+from .db import NestQuestDatabase
+
+#: GLOB patterns matching a valid ``weekday_set`` CSV: one GLOB per
+#: possible element count 1..7 (a weekday set cannot exceed 7 elements).
+#: Each element is a single digit 0-6, separated by a literal comma.
+#: GLOB has no repetition quantifier, so the element count is enumerated.
+#: ``''`` and NULL are deliberately NOT matched: NULL is allowed for
+#: non-weekly rules via a separate ``IS NULL`` arm, and '' is invalid
+#: because weekly rules need at least one weekday.
+WEEKDAY_SET_GLOB_PATTERNS: list[str] = [
+    "[0-6]",
+    "[0-6],[0-6]",
+    "[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]",
+]
+
+SCHEMA_V1_CHILDREN_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS children (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        display_name TEXT NOT NULL,
+        colour TEXT,
+        avatar_ref TEXT,
+        sort_order INTEGER NOT NULL DEFAULT 0 CHECK (typeof(sort_order) = 'integer'),
+        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+        created_at TEXT NOT NULL
+    )
+    """,
+]
+
+SCHEMA_V1_ADMIN_USERS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS admin_users (
+        ha_user_id TEXT PRIMARY KEY NOT NULL,
+        added_at TEXT NOT NULL
+    )
+    """,
+]
+
+_WEEKDAY_SET_CHECK = " OR ".join(
+    f"weekday_set GLOB '{pattern}'" for pattern in WEEKDAY_SET_GLOB_PATTERNS
+)
+
+SCHEMA_V1_SCHEDULE_RULES_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS schedule_rules (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        rule_type TEXT NOT NULL CHECK (rule_type IN ('daily', 'weekly', 'monthly', 'yearly', 'custom')),
+        interval INTEGER NOT NULL DEFAULT 1 CHECK (typeof(interval) = 'integer' AND interval >= 1),
+        weekday_set TEXT CHECK (weekday_set IS NULL OR ({_WEEKDAY_SET_CHECK})),
+        day_of_month INTEGER CHECK (day_of_month IS NULL OR (typeof(day_of_month) = 'integer' AND day_of_month >= 1 AND day_of_month <= 31)),
+        nth_weekday INTEGER CHECK (nth_weekday IS NULL OR (typeof(nth_weekday) = 'integer' AND nth_weekday IN (-1, 1, 2, 3, 4, 5))),
+        month INTEGER CHECK (month IS NULL OR (typeof(month) = 'integer' AND month >= 1 AND month <= 12)),
+        start_date TEXT NOT NULL,
+        end_date TEXT,
+        CHECK (end_date IS NULL OR end_date >= start_date),
+        CHECK (
+            (rule_type = 'weekly' AND weekday_set IS NOT NULL)
+            OR (rule_type = 'monthly' AND (day_of_month IS NOT NULL OR nth_weekday IS NOT NULL))
+            OR (rule_type = 'yearly' AND month IS NOT NULL)
+            OR (rule_type IN ('daily', 'custom') AND day_of_month IS NULL AND nth_weekday IS NULL AND month IS NULL)
+        )
+    )
+    """,
+]
+
+SCHEMA_V1_TASK_DEFINITIONS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS task_definitions (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        title TEXT NOT NULL,
+        description TEXT,
+        icon TEXT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        due_time TEXT,
+        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+        created_at TEXT NOT NULL
+    )
+    """,
+]
+
+SCHEMA_V1_PRESENCE_SCHEDULES_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS presence_schedules (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        child_id INTEGER NOT NULL UNIQUE REFERENCES children(id),
+        cycle_length_weeks INTEGER NOT NULL CHECK (typeof(cycle_length_weeks) = 'integer' AND cycle_length_weeks >= 1 AND cycle_length_weeks <= 4),
+        anchor_date TEXT NOT NULL,
+        pattern TEXT NOT NULL CHECK (
+            NOT pattern GLOB '*[^0-6,|]*'
+            AND NOT pattern GLOB '*[0-6][0-6]*'
+            AND NOT pattern GLOB '*,,*'
+            AND NOT pattern GLOB '*,|*'
+            AND NOT pattern GLOB '*|,*'
+            AND NOT pattern GLOB ',*'
+            AND NOT pattern GLOB '*,'
+            AND NOT pattern GLOB '*[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]*'
+            AND (LENGTH(pattern) - LENGTH(REPLACE(pattern, '|', '')) + 1) = cycle_length_weeks
+        )
+    )
+    """,
+]
+
+SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS presence_overrides (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        start_date TEXT NOT NULL,
+        end_date TEXT NOT NULL,
+        is_present INTEGER NOT NULL CHECK (typeof(is_present) = 'integer' AND is_present IN (0, 1)),
+        note TEXT,
+        CHECK (end_date >= start_date)
+    )
+    """,
+]
+
+SCHEMA_V1_STATEMENTS: list[str] = [
+    *SCHEMA_V1_CHILDREN_DDL,
+    *SCHEMA_V1_ADMIN_USERS_DDL,
+    *SCHEMA_V1_SCHEDULE_RULES_DDL,
+    *SCHEMA_V1_TASK_DEFINITIONS_DDL,
+    *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
+    *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
+]
+
+
+async def async_apply_ddl(
+    database: NestQuestDatabase, statements: list[str]
+) -> None:
+    """Apply each DDL statement inside one transaction on ``database``.
+
+    Idempotent: every statement is ``CREATE TABLE IF NOT EXISTS``, so
+    applying the same list twice leaves the schema and its rows intact.
+    """
+    async with database.transaction():
+        for sql in statements:
+            await database.execute(sql)

tests/conftest.py

@@ -25,6 +25,9 @@ import asyncio
 import importlib.util
 import inspect
 import sys
+import tempfile
+import weakref
+from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
@@ -218,13 +221,35 @@ def wire_entry_to_registry(entry, registry: ListenerRegistry):
     return entry
 
 
+class _HassNamespace(SimpleNamespace):
+    """SimpleNamespace subclass so the hass stub can hold weak references."""
+
+
 def make_hass() -> tuple:
     """Create a hass stub plus its listener registry, mirroring HA's shape."""
-    hass = SimpleNamespace(
+    config_dir = Path(tempfile.mkdtemp(prefix="nestquest-hass-"))
+    hass = _HassNamespace(
         data={},
         config=MagicMock(),
         config_entries=SimpleNamespace(async_reload=None),
     )
+
+    def _cleanup_config_dir():
+        import shutil
+
+        shutil.rmtree(config_dir, ignore_errors=True)
+
+    weakref.finalize(hass, _cleanup_config_dir)
+    hass.config.path = MagicMock(
+        side_effect=lambda name: str(config_dir / (name or "config"))
+    )
+
+    def _run_executor_job(fn, *args, **kwargs):
+        return asyncio.get_running_loop().run_in_executor(
+            None, lambda: fn(*args, **kwargs)
+        )
+
+    hass.async_add_executor_job = _run_executor_job
     registry = ListenerRegistry(hass)
 
     async def _async_reload(entry_id: str) -> None:

tests/test_db.py

@@ -0,0 +1,998 @@
+"""Tests for db.py, the async SQLite connection wrapper of NestQuest."""
+from __future__ import annotations
+
+import ast
+import asyncio
+import sqlite3
+import threading
+from pathlib import Path
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from conftest import make_hass
+
+from custom_components.nestquest import db as db_module
+from custom_components.nestquest.db import NestQuestDatabase
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _open_db(path) -> tuple[NestQuestDatabase, MagicMock]:
+    hass = _make_hass_mock()
+    database = NestQuestDatabase(hass)
+    _run(database.open(path))
+    return database, hass
+
+
+# ---------------------------------------------------------------------------
+# open(): pragmas, executor delegation, idempotence
+# ---------------------------------------------------------------------------
+
+
+def test_open_enables_wal_journal_mode(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "wal.db")
+    row = _run(database.fetch_one("PRAGMA journal_mode"))
+    assert row is not None and str(row[0]).lower() == "wal"
+    _run(database.close())
+
+
+def test_open_enables_foreign_key_enforcement(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "fk.db")
+    row = _run(database.fetch_one("PRAGMA foreign_keys"))
+    assert row is not None and int(row[0]) == 1
+    _run(database.close())
+
+
+def test_open_delegates_connection_to_executor(tmp_path) -> None:
+    hass = _make_hass_mock()
+    database = NestQuestDatabase(hass)
+    _run(database.open(tmp_path / "exec.db"))
+    first_call = hass.async_add_executor_job.call_args_list[0]
+    assert callable(first_call.args[0])
+    _run(database.close())
+
+
+def test_open_twice_returns_same_wrapper_without_second_connect(tmp_path) -> None:
+    path = tmp_path / "once.db"
+    hass = _make_hass_mock()
+    database = NestQuestDatabase(hass)
+    first = _run(database.open(path))
+    second = _run(database.open(path))
+    assert first is database
+    assert second is database
+    assert database.connected is True
+    opens = [
+        c
+        for c in hass.async_add_executor_job.call_args_list
+        if getattr(c.args[0], "__name__", "") == "_open"
+    ]
+    assert len(opens) == 1
+    _run(database.close())
+
+
+def test_open_accepts_str_path(tmp_path) -> None:
+    database, _hass = _open_db(str(tmp_path / "str.db"))
+    assert database.connected is True
+    _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Executor delegation: every sqlite3 call happens off the event loop
+# ---------------------------------------------------------------------------
+
+
+def test_every_method_routes_sqlite3_through_executor(tmp_path) -> None:
+    """Each wrapper call must go via hass.async_add_executor_job."""
+    database, hass = _open_db(tmp_path / "route.db")
+    calls_before = hass.async_add_executor_job.await_count
+
+    _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"))
+    _run(database.execute("INSERT INTO t (name) VALUES (?)", ("a",)))
+    _run(database.execute_many("INSERT INTO t (name) VALUES (?)", [("b",), ("c",)]))
+    row = _run(database.fetch_one("SELECT name FROM t WHERE id = 1"))
+    rows = _run(database.fetch_all("SELECT name FROM t ORDER BY id"))
+
+    assert row == ("a",)
+    assert rows == [("a",), ("b",), ("c",)]
+    assert hass.async_add_executor_job.await_count == calls_before + 5
+    _run(database.close())
+
+
+async def test_every_wrapper_call_awaits_executor_job(tmp_path) -> None:
+    """The executor surface is awaited per wrapper call on a live loop."""
+    hass = _make_hass_mock()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "loop.db")
+
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)")
+    await database.fetch_one("SELECT 1")
+    await database.fetch_all("SELECT 1")
+    await database.execute_many("INSERT INTO t (id) VALUES (?)", [(1,)])
+    async with database.transaction():
+        pass
+
+    assert hass.async_add_executor_job.await_count == 7
+    assert all(callable(c.args[0]) for c in hass.async_add_executor_job.call_args_list)
+    await database.close()
+
+
+async def test_public_methods_return_only_plain_data(tmp_path) -> None:
+    """execute/execute_many return plain dataclasses, never sqlite3 objects."""
+    hass = _make_hass_mock()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "plain.db")
+    result = await database.execute(
+        "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"
+    )
+    assert not isinstance(result, (sqlite3.Connection, sqlite3.Cursor))
+    insert = await database.execute("INSERT INTO t (name) VALUES (?)", ("a",))
+    assert not isinstance(insert, (sqlite3.Connection, sqlite3.Cursor))
+    assert insert.rowcount == 1
+    assert insert.lastrowid == 1
+    many = await database.execute_many(
+        "INSERT INTO t (name) VALUES (?)", [("b",), ("c",)]
+    )
+    assert not isinstance(many, (sqlite3.Connection, sqlite3.Cursor))
+    assert many.rowcount == 2
+    row = await database.fetch_one("SELECT name FROM t WHERE id = 1")
+    assert row == ("a",)
+    rows = await database.fetch_all("SELECT name FROM t ORDER BY id")
+    assert rows == [("a",), ("b",), ("c",)]
+    for value in (*rows, row):
+        assert not isinstance(value, sqlite3.Cursor)
+    await database.close()
+
+
+async def test_no_sqlite3_call_on_event_loop_with_real_executor(tmp_path) -> None:
+    """With the conftest hass (real run_in_executor), queries still succeed."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "real.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)")
+    assert await database.fetch_one("SELECT 1") == (1,)
+    await database.close()
+
+
+async def test_real_executor_never_touches_loop_thread(tmp_path, monkeypatch) -> None:
+    """Every real sqlite3 call must run on a non-event-loop thread.
+
+    sqlite3.connect is instrumented so a proxy wraps the connection and
+    every cursor it hands out; each proxied call records the thread it ran
+    on.  A recorded thread id equal to the event-loop thread would mean a
+    sqlite3 call crossed the executor boundary.
+    """
+    loop_thread = threading.get_ident()
+    recorded: list[int] = []
+    guard = threading.Lock()
+
+    def _note() -> None:
+        with guard:
+            recorded.append(threading.get_ident())
+
+    class _Proxy:
+        """Generic proxy recording the thread of every attribute call."""
+
+        def __init__(self, target: Any) -> None:
+            object.__setattr__(self, "_target", target)
+
+        def __getattr__(self, name: str):
+            attr = getattr(object.__getattribute__(self, "_target"), name)
+            if not callable(attr):
+                return attr
+
+            def _wrapped(*args, **kwargs):
+                _note()
+                result = attr(*args, **kwargs)
+                if isinstance(result, sqlite3.Cursor):
+                    return _Proxy(result)
+                return result
+
+            return _wrapped
+
+    real_connect = sqlite3.connect
+    connect_threads: list[int] = []
+
+    def _traced_connect(*args, **kwargs):
+        _note()
+        connect_threads.append(threading.get_ident())
+        return _Proxy(real_connect(*args, **kwargs))
+
+    monkeypatch.setattr(db_module.sqlite3, "connect", _traced_connect)
+
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "threads.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)")
+    await database.execute("INSERT INTO t (id) VALUES (?)", (1,))
+    await database.execute_many("INSERT INTO t (id) VALUES (?)", [(2,), (3,)])
+    assert await database.fetch_one("SELECT id FROM t WHERE id = 1") == (1,)
+    assert await database.fetch_all("SELECT id FROM t") == [(1,), (2,), (3,)]
+    async with database.transaction():
+        await database.execute("INSERT INTO t (id) VALUES (?)", (4,))
+    await database.close()
+
+    assert recorded, "expected at least one recorded sqlite3 call"
+    assert len(connect_threads) == 1, (
+        f"expected exactly one traced sqlite3.connect, got {len(connect_threads)}"
+    )
+    assert loop_thread not in recorded, (
+        f"sqlite3 call ran on the event-loop thread {loop_thread}; "
+        f"recorded threads: {sorted(set(recorded))}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Source scan: sqlite3.* only inside executor-delegated functions
+# ---------------------------------------------------------------------------
+
+
+def _iter_with_scope(node, chain):
+    """Yield (node, enclosing-function chain), depth-first."""
+    for child in ast.iter_child_nodes(node):
+        yield child, chain
+        next_chain = chain + (child,) if isinstance(
+            child, (ast.FunctionDef, ast.AsyncFunctionDef)
+        ) else chain
+        yield from _iter_with_scope(child, next_chain)
+
+
+def test_db_sqlite3_calls_only_inside_executor_delegated_functions() -> None:
+    """Every sqlite3.* call in db.py sits inside an executor-delegating function.
+
+    A function is executor-delegating when it (or, for nested closures, any of
+    its enclosing functions) passes work to ``hass.async_add_executor_job``.
+    """
+    source = Path(db_module.__file__).read_text(encoding="utf-8")
+    tree = ast.parse(source, filename=str(db_module.__file__))
+
+    delegated = {
+        node.name
+        for node in ast.walk(tree)
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and any(
+            isinstance(sub, ast.Call)
+            and isinstance(sub.func, ast.Attribute)
+            and sub.func.attr == "async_add_executor_job"
+            for sub in ast.walk(node)
+        )
+    }
+    assert delegated, "expected executor-delegated functions in db.py"
+
+    violations = []
+    for node, chain in _iter_with_scope(tree, ()):
+        if (
+            isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and isinstance(node.func.value, ast.Name)
+            and node.func.value.id == "sqlite3"
+            and not any(f.name in delegated for f in chain)
+        ):
+            violations.append((node.lineno, node.func.attr))
+    assert violations == [], (
+        f"Direct sqlite3 calls outside executor-delegated functions: {violations}"
+    )
+
+
+def test_package_wide_sqlite3_scan_excludes_only_db_module() -> None:
+    """The package-wide store scan exempts exactly the db module."""
+    pkg_dir = Path(db_module.__file__).parent
+    scanned = [p.name for p in pkg_dir.glob("*.py") if p.name != "db.py"]
+    assert scanned, "expected other package modules to be scanned"
+    assert "db.py" not in scanned
+
+
+# ---------------------------------------------------------------------------
+# close(): idempotence and post-close errors
+# ---------------------------------------------------------------------------
+
+
+def test_close_is_idempotent(tmp_path) -> None:
+    database, hass = _open_db(tmp_path / "idem.db")
+    _run(database.close())
+    assert database.connected is False
+    _run(database.close())
+    assert database.connected is False
+    assert hass.async_add_executor_job.await_count == 2
+    assert database.rowcount == -1
+    assert database.lastrowid is None
+
+
+def test_operations_after_close_raise_clear_error(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "closed.db")
+    _run(database.close())
+
+    async def _txn():
+        async with database.transaction():
+            pass
+
+    with pytest.raises(RuntimeError, match="closed"):
+        _run(database.execute("SELECT 1"))
+    with pytest.raises(RuntimeError, match="closed"):
+        _run(database.fetch_one("SELECT 1"))
+    with pytest.raises(RuntimeError, match="closed"):
+        _run(database.fetch_all("SELECT 1"))
+    with pytest.raises(RuntimeError, match="closed"):
+        _run(database.execute_many("SELECT 1", []))
+    with pytest.raises(RuntimeError, match="closed"):
+        _run(_txn())
+
+
+# ---------------------------------------------------------------------------
+# Transactions: commit on success, rollback on exception
+# ---------------------------------------------------------------------------
+
+
+def test_transaction_commits_on_success(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "commit.db")
+    _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"))
+
+    async def _block():
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+
+    _run(_block())
+    assert _run(database.fetch_all("SELECT name FROM t")) == [("kept",)]
+    _run(database.close())
+
+
+def test_transaction_rolls_back_on_exception(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "rollback.db")
+    _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"))
+    _run(database.execute("INSERT INTO t (name) VALUES (?)", ("before",)))
+
+    async def _block():
+        with pytest.raises(ValueError, match="boom"):
+            async with database.transaction():
+                await database.execute(
+                    "INSERT INTO t (name) VALUES (?)", ("dropped",)
+                )
+                raise ValueError("boom")
+
+    _run(_block())
+    assert _run(database.fetch_all("SELECT name FROM t")) == [("before",)]
+    _run(database.close())
+
+
+async def test_commit_failure_rolls_back_and_next_transaction_works(
+    tmp_path,
+) -> None:
+    """A COMMIT failure rolls back and leaves the wrapper reusable.
+
+    With PRAGMA defer_foreign_keys=ON a violating insert succeeds inside
+    the transaction and the foreign-key violation only surfaces at COMMIT.
+    The commit error must propagate, the real connection must end up
+    outside any transaction, and the next transaction() plus statement
+    must run cleanly.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "deferred.db")
+    await database.execute("CREATE TABLE parent (id INTEGER PRIMARY KEY)")
+    await database.execute(
+        "CREATE TABLE child ("
+        "id INTEGER PRIMARY KEY, "
+        "parent_id INTEGER NOT NULL REFERENCES parent(id))"
+    )
+
+    with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+        async with database.transaction():
+            await database.execute("PRAGMA defer_foreign_keys=ON")
+            await database.execute(
+                "INSERT INTO child (parent_id) VALUES (?)", (999,)
+            )
+
+    assert database._in_transaction is False
+    assert database._tx_task is None
+    assert database._conn is not None
+    assert database._conn.in_transaction is False
+    assert await database.fetch_all("SELECT COUNT(*) FROM child") == [(0,)]
+
+    async with database.transaction():
+        await database.execute("INSERT INTO parent (id) VALUES (?)", (1,))
+        await database.execute(
+            "INSERT INTO child (parent_id) VALUES (?)", (1,)
+        )
+    assert await database.fetch_all("SELECT COUNT(*) FROM child") == [(1,)]
+    await database.close()
+
+
+async def test_cancel_during_open_closes_abandoned_connection(tmp_path) -> None:
+    """NQ-PR12-010: cancel during open() must close the settled connection.
+
+    The _open executor job is gated: the open task is cancelled while the
+    connect is still in flight, then the worker is allowed to complete it
+    anyway.  The settled connection must never be adopted, must be closed
+    through a settled executor-backed close before the cancellation
+    propagates, and a follow-up open() must succeed cleanly.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+
+    open_started = asyncio.Event()
+    release_open = asyncio.Event()
+    jobs: list[str] = []
+    captured: dict[str, Any] = {}
+    real_add = hass.async_add_executor_job
+
+    async def _gated_add_executor_job(fn, *args, **kwargs):
+        name = getattr(fn, "__name__", "")
+        jobs.append(name)
+        if name == "_open":
+            open_started.set()
+            await release_open.wait()
+        result = await real_add(fn, *args, **kwargs)
+        if name == "_open":
+            captured["conn"] = result
+        return result
+
+    hass.async_add_executor_job = _gated_add_executor_job
+
+    open_task = asyncio.create_task(database.open(tmp_path / "cancel-open.db"))
+    await open_started.wait()
+    open_task.cancel()
+    await asyncio.sleep(0.05)
+    assert database._conn is None, (
+        "the cancelled open adopted a connection that was still settling"
+    )
+    release_open.set()
+    with pytest.raises(asyncio.CancelledError):
+        await open_task
+
+    assert database.connected is False
+    assert "_close_abandoned" in jobs, (
+        "the abandoned connection was not closed via an executor job"
+    )
+    conn = captured.get("conn")
+    assert conn is not None
+
+    def _probe() -> str:
+        try:
+            conn.execute("SELECT 1")
+        except sqlite3.ProgrammingError:
+            return "closed"
+        return "open"
+
+    assert await real_add(_probe) == "closed"
+
+    hass.async_add_executor_job = real_add
+    await database.open(tmp_path / "cancel-open.db")
+    assert database.connected is True
+    assert await database.fetch_one("SELECT 1") == (1,)
+    await database.close()
+
+
+async def test_cancel_during_begin_settles_then_rolls_back(tmp_path) -> None:
+    """NQ-PR12-007: cancel during BEGIN must not leak a stray transaction.
+
+    The BEGIN executor job is gated: the transaction task is cancelled
+    while BEGIN is still in flight, then the worker is allowed to
+    complete it anyway.  The wrapper must keep transaction ownership
+    until the submitted work settles, roll the completed BEGIN back,
+    clear its state, and stay usable for the next transaction.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "cancel-begin.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+    await database.execute("INSERT INTO t (name) VALUES (?)", ("before",))
+
+    begin_started = asyncio.Event()
+    release_begin = asyncio.Event()
+    real_add = hass.async_add_executor_job
+
+    async def _gated_add_executor_job(fn, *args, **kwargs):
+        if getattr(fn, "__name__", "") == "_begin":
+            begin_started.set()
+            await release_begin.wait()
+        return await real_add(fn, *args, **kwargs)
+
+    hass.async_add_executor_job = _gated_add_executor_job
+
+    async def _txn() -> None:
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+
+    txn_task = asyncio.create_task(_txn())
+    await begin_started.wait()
+    txn_task.cancel()
+    await asyncio.sleep(0.05)
+    assert database._in_transaction is True, (
+        "ownership was released while the cancelled BEGIN was still in flight"
+    )
+    release_begin.set()
+    with pytest.raises(asyncio.CancelledError):
+        await txn_task
+
+    assert database._in_transaction is False
+    assert database._tx_task is None
+    assert database._conn is not None
+    assert database._conn.in_transaction is False
+
+    hass.async_add_executor_job = real_add
+    assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
+        ("before",)
+    ]
+    async with database.transaction():
+        await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+    assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
+        ("before",),
+        ("kept",),
+    ]
+    await database.close()
+
+
+async def test_cancel_during_commit_settles_then_propagates_cancellation(
+    tmp_path,
+) -> None:
+    """NQ-PR12-007: cancel during COMMIT must not mask the cancellation.
+
+    COMMIT is gated; the task is cancelled while it is in flight and the
+    worker then completes it anyway.  The cancellation must propagate as
+    CancelledError (not be converted by a doomed ROLLBACK of an already
+    committed transaction), the connection must end outside any
+    transaction, and the wrapper must stay usable.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "cancel-commit.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+
+    commit_started = asyncio.Event()
+    release_commit = asyncio.Event()
+    real_add = hass.async_add_executor_job
+
+    async def _gated_add_executor_job(fn, *args, **kwargs):
+        if getattr(fn, "__name__", "") == "_commit":
+            commit_started.set()
+            await release_commit.wait()
+        return await real_add(fn, *args, **kwargs)
+
+    hass.async_add_executor_job = _gated_add_executor_job
+
+    async def _txn() -> None:
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+
+    txn_task = asyncio.create_task(_txn())
+    await commit_started.wait()
+    txn_task.cancel()
+    await asyncio.sleep(0.05)
+    assert database._in_transaction is True, (
+        "ownership was released while the cancelled COMMIT was still in flight"
+    )
+    release_commit.set()
+    with pytest.raises(asyncio.CancelledError):
+        await txn_task
+
+    assert database._in_transaction is False
+    assert database._tx_task is None
+    assert database._conn is not None
+    assert database._conn.in_transaction is False
+
+    hass.async_add_executor_job = real_add
+    assert await database.fetch_all("SELECT name FROM t") == [("kept",)]
+    async with database.transaction():
+        await database.execute("INSERT INTO t (name) VALUES (?)", ("again",))
+    assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
+        ("kept",),
+        ("again",),
+    ]
+    await database.close()
+
+
+async def test_cancel_during_in_transaction_execute_settles_then_rolls_back(
+    tmp_path,
+) -> None:
+    """NQ-PR12-009: cancel during an in-transaction statement must not race ROLLBACK.
+
+    The in-transaction INSERT executor job is gated: the transaction task
+    is cancelled while the statement is still in flight, then the worker
+    is allowed to complete it anyway.  The ROLLBACK must only run after
+    the statement has settled — while the statement is in flight the
+    transaction is still open — and the cancelled statement's write must
+    not persist.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "cancel-execute.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+    await database.execute("INSERT INTO t (name) VALUES (?)", ("before",))
+
+    insert_started = asyncio.Event()
+    release_insert = asyncio.Event()
+    real_add = hass.async_add_executor_job
+
+    async def _gated_add_executor_job(fn, *args, **kwargs):
+        if getattr(fn, "__name__", "") == "_execute":
+            insert_started.set()
+            await release_insert.wait()
+        return await real_add(fn, *args, **kwargs)
+
+    hass.async_add_executor_job = _gated_add_executor_job
+
+    async def _txn() -> None:
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("lost",))
+
+    txn_task = asyncio.create_task(_txn())
+    await insert_started.wait()
+    txn_task.cancel()
+    await asyncio.sleep(0.05)
+    assert database._in_transaction is True, (
+        "ownership was released while the cancelled statement was in flight"
+    )
+    assert database._conn is not None
+    assert database._conn.in_transaction is True, (
+        "ROLLBACK ran before the cancelled statement settled"
+    )
+    release_insert.set()
+    with pytest.raises(asyncio.CancelledError):
+        await txn_task
+
+    assert database._in_transaction is False
+    assert database._tx_task is None
+    assert database._conn is not None
+    assert database._conn.in_transaction is False
+
+    hass.async_add_executor_job = real_add
+    assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
+        ("before",)
+    ]
+    async with database.transaction():
+        await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+    assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [
+        ("before",),
+        ("kept",),
+    ]
+    await database.close()
+
+
+async def test_commit_and_rollback_failures_are_both_preserved(tmp_path) -> None:
+    """NQ-PR12-008: both errors survive when COMMIT and ROLLBACK both fail.
+
+    The COMMIT executor call is forced to fail, and the follow-up
+    cleanup ROLLBACK is forced to fail as well.  The raised error must
+    keep both original exceptions inspectable instead of discarding the
+    commit failure behind the rollback failure.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "both-fail.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)")
+
+    commit_failure = sqlite3.OperationalError("commit exploded")
+    rollback_failure = sqlite3.OperationalError("rollback exploded")
+    real_add = hass.async_add_executor_job
+
+    async def _failing_add_executor_job(fn, *args, **kwargs):
+        name = getattr(fn, "__name__", "")
+        if name == "_commit":
+            raise commit_failure
+        if name == "_cleanup":
+            raise rollback_failure
+        return await real_add(fn, *args, **kwargs)
+
+    hass.async_add_executor_job = _failing_add_executor_job
+
+    with pytest.raises(BaseExceptionGroup) as exc_info:
+        async with database.transaction():
+            pass
+
+    assert commit_failure in exc_info.value.exceptions
+    assert rollback_failure in exc_info.value.exceptions
+    assert database._in_transaction is False
+    assert database._tx_task is None
+    await database.close()
+
+
+async def test_transaction_persists_across_connections(tmp_path) -> None:
+    """A committed transaction is visible to a brand-new connection."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "persist.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+    async with database.transaction():
+        await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+    await database.close()
+
+    other = NestQuestDatabase(hass)
+    await other.open(tmp_path / "persist.db")
+    assert await other.fetch_all("SELECT name FROM t") == [("kept",)]
+    await other.close()
+
+
+# ---------------------------------------------------------------------------
+# Serialization: one lock per wrapper, no cross-commit interleaving
+# ---------------------------------------------------------------------------
+
+
+async def test_concurrent_writers_do_not_interleave_commits(tmp_path) -> None:
+    """Real-executor proof: nothing commits inside another transaction's span.
+
+    A transaction runner repeatedly counts rows, inserts, and counts again
+    inside BEGIN..COMMIT; the count must grow by exactly one, proving no
+    other writer committed in between even though several writer tasks fire
+    statements concurrently on real executor threads.  Statements issued
+    while a transaction is open queue behind it on the wrapper lock.
+    """
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "serial.db")
+    await database.execute("CREATE TABLE t (kind TEXT, seq INTEGER)")
+    tx_rounds, writers, writer_rounds = 6, 5, 8
+    total = tx_rounds + writers * writer_rounds
+
+    async def _tx_runner() -> None:
+        for rnd in range(tx_rounds):
+            async with database.transaction():
+                rows = await database.fetch_all("SELECT COUNT(*) FROM t")
+                seen = rows[0][0] if rows else -1
+                await database.execute(
+                    "INSERT INTO t (kind, seq) VALUES (?, ?)", ("tx", rnd)
+                )
+                rows = await database.fetch_all("SELECT COUNT(*) FROM t")
+                grown = rows[0][0] if rows else -1
+                if grown != seen + 1:
+                    raise AssertionError(
+                        f"interleaved commit: count {seen} -> {grown} "
+                        f"in tx round {rnd}"
+                    )
+
+    async def _writer(worker: int) -> None:
+        for rnd in range(writer_rounds):
+            await database.execute(
+                "INSERT INTO t (kind, seq) VALUES (?, ?)", ("w", worker)
+            )
+
+    await asyncio.gather(
+        _tx_runner(), *(_writer(w) for w in range(writers))
+    )
+
+    rows = await database.fetch_all("SELECT COUNT(*) FROM t")
+    assert rows[0][0] == total, f"expected {total} rows, got {rows[0][0]}"
+    await database.close()
+
+
+async def test_transaction_span_excludes_concurrent_transaction(tmp_path) -> None:
+    """A second transaction caller is rejected with a clear error."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "reject.db")
+    await database.execute("CREATE TABLE t (id INTEGER)")
+
+    started = asyncio.Event()
+
+    async def _first() -> None:
+        async with database.transaction():
+            started.set()
+            await asyncio.sleep(0.05)
+
+    first = asyncio.create_task(_first())
+    await started.wait()
+    with pytest.raises(RuntimeError, match="transaction"):
+        async with database.transaction():
+            pass
+    await first
+    await database.execute("INSERT INTO t (id) VALUES (?)", (1,))
+    assert await database.fetch_one("SELECT id FROM t") == (1,)
+    await database.close()
+
+
+async def test_transaction_blocks_other_statements_until_commit(tmp_path) -> None:
+    """A statement issued during an open transaction queues behind it."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "block.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+
+    order: list[str] = []
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def _txn() -> None:
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("in_tx",))
+            order.append("tx-start")
+            started.set()
+            await release.wait()
+            order.append("tx-end")
+
+    async def _outsider() -> None:
+        await started.wait()
+        await database.execute("INSERT INTO t (name) VALUES (?)", ("after",))
+        order.append("outside")
+
+    txn_task = asyncio.create_task(_txn())
+    await started.wait()
+    outsider_task = asyncio.create_task(_outsider())
+    await asyncio.sleep(0.05)
+    assert order == ["tx-start"], f"outsider ran inside transaction: {order}"
+    release.set()
+    await asyncio.gather(txn_task, outsider_task)
+    assert order == ["tx-start", "tx-end", "outside"]
+    rows = await database.fetch_all("SELECT name FROM t ORDER BY id")
+    assert rows == [("in_tx",), ("after",)]
+    await database.close()
+
+
+async def test_concurrent_transaction_is_rejected_with_clear_error(
+    tmp_path,
+) -> None:
+    """A second concurrent transaction() gets a clear RuntimeError."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "concurrent-tx.db")
+    await database.execute("CREATE TABLE t (id INTEGER)")
+
+    started = asyncio.Event()
+
+    async def _first() -> None:
+        async with database.transaction():
+            started.set()
+            await asyncio.sleep(0.05)
+
+    first = asyncio.create_task(_first())
+    await started.wait()
+    with pytest.raises(RuntimeError, match="transaction"):
+        async with database.transaction():
+            pass
+    await first
+    await database.close()
+
+
+async def test_close_waits_for_in_flight_transaction(tmp_path) -> None:
+    """close() serializes: an open transaction finishes before shutdown."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "closerace.db")
+    await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")
+
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def _txn() -> None:
+        async with database.transaction():
+            await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",))
+            started.set()
+            await release.wait()
+
+    txn_task = asyncio.create_task(_txn())
+    await started.wait()
+    close_task = asyncio.create_task(database.close())
+    await asyncio.sleep(0.05)
+    assert database.connected is True, "close() raced an open transaction"
+    release.set()
+    await asyncio.gather(txn_task, close_task)
+    assert database.connected is False
+
+    witness = NestQuestDatabase(hass)
+    await witness.open(tmp_path / "closerace.db")
+    assert await witness.fetch_all("SELECT name FROM t") == [("kept",)]
+    await witness.close()
+
+
+async def test_close_during_transaction_from_same_task_is_rejected(
+    tmp_path,
+) -> None:
+    """close() inside its own transaction body raises instead of deadlocking."""
+    hass, _registry = make_hass()
+    database = NestQuestDatabase(hass)
+    await database.open(tmp_path / "selfclose.db")
+    await database.execute("CREATE TABLE t (id INTEGER)")
+    with pytest.raises(RuntimeError, match="transaction"):
+        async with database.transaction():
+            await database.close()
+    assert database.connected is True
+    await database.close()
+
+
+# ---------------------------------------------------------------------------
+# rowcount / lastrowid surface
+# ---------------------------------------------------------------------------
+
+
+def test_rowcount_and_lastrowid_before_any_operation(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "unset.db")
+    assert database.rowcount == -1
+    assert database.lastrowid is None
+    _run(database.close())
+
+
+def test_execute_exposes_rowcount_and_lastrowid(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "counts.db")
+    _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"))
+    _run(database.execute("INSERT INTO t (name) VALUES (?)", ("x",)))
+    assert database.rowcount == 1
+    assert database.lastrowid == 1
+    _run(database.execute("INSERT INTO t (name) VALUES (?)", ("y",)))
+    assert database.rowcount == 1
+    assert database.lastrowid == 2
+    _run(database.close())
+
+
+def test_execute_many_exposes_rowcount(tmp_path) -> None:
+    database, _hass = _open_db(tmp_path / "many.db")
+    _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)"))
+    _run(database.execute_many("INSERT INTO t (name) VALUES (?)", [("a",), ("b",)]))
+    assert database.rowcount == 2
+    _run(database.execute("INSERT INTO t (name) VALUES (?)", ("c",)))
+    assert database.lastrowid == 3
+    _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Lifecycle wiring: opened once per entry, closed on unload
+# ---------------------------------------------------------------------------
+
+
+async def test_setup_entry_opens_database_in_runtime_data(hass, make_entry) -> None:
+    from conftest import wire_entry_to_registry
+
+    from custom_components.nestquest import async_setup_entry, async_unload_entry
+
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    runtime = entry.runtime_data
+    assert runtime.database is not None
+    assert runtime.database.connected is True
+    await async_unload_entry(hass, entry)
+
+
+async def test_unload_entry_closes_database(hass, make_entry) -> None:
+    from conftest import wire_entry_to_registry
+
+    from custom_components.nestquest import async_setup_entry, async_unload_entry
+
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    await async_setup_entry(hass, entry)
+    database = entry.runtime_data.database
+    assert database.connected is True
+    assert await async_unload_entry(hass, entry) is True
+    assert database.connected is False
+    with pytest.raises(RuntimeError, match="closed"):
+        await database.fetch_one("SELECT 1")
+
+
+async def test_second_setup_does_not_open_second_connection(hass, make_entry) -> None:
+    from conftest import wire_entry_to_registry
+
+    from custom_components.nestquest import async_setup_entry, async_unload_entry
+
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    await async_setup_entry(hass, entry)
+    first = entry.runtime_data.database
+
+    await async_setup_entry(hass, entry)
+    second = entry.runtime_data.database
+
+    assert first is second
+    assert first.connected is True
+    await async_unload_entry(hass, entry)
+
+
+async def test_setup_unload_cycles_reopen_fresh_connection(hass, make_entry) -> None:
+    from conftest import wire_entry_to_registry
+
+    from custom_components.nestquest import async_setup_entry, async_unload_entry
+
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    seen: list[NestQuestDatabase] = []
+    for _ in range(3):
+        await async_setup_entry(hass, entry)
+        runtime = entry.runtime_data
+        assert runtime.database.connected is True
+        seen.append(runtime.database)
+        await async_unload_entry(hass, entry)
+        assert runtime.database.connected is False
+    assert len(set(seen)) == 3
+    assert hass.data == {}

tests/test_lifecycle.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 
 import asyncio
 
+import pytest
+
 from conftest import make_config_entry, make_hass, wire_entry_to_registry
 
 from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
@@ -196,4 +198,61 @@ async def test_setup_awaitable_listener_removal_supported() -> None:
     assert result is True
     assert removed == [True]
     assert hass.data == {}
-    assert entry.runtime_data is None
+    assert entry.runtime_data is None
+
+
+async def test_setup_closes_database_when_listener_registration_fails() -> None:
+    """A listener-registration failure must not leak the opened database."""
+    from custom_components.nestquest.db import NestQuestDatabase
+
+    class _Boom(Exception):
+        pass
+
+    entry = make_config_entry(entry_id="listener-boom")
+    entry.add_update_listener = lambda listener: (_ for _ in ()).throw(_Boom())
+    hass, _registry = make_hass()
+    with pytest.raises(_Boom):
+        await async_setup_entry(hass, entry)
+    assert hass.data.get(DOMAIN, {}) == {}
+    assert not hasattr(entry, "runtime_data")
+
+
+async def test_unload_closes_database_even_when_listener_removal_raises() -> None:
+    """Unload removes the listener, closes the DB in a finally, then raises."""
+    from custom_components.nestquest.db import NestQuestDatabase
+
+    class _RemoveBoom(Exception):
+        pass
+
+    def _remover():
+        raise _RemoveBoom()
+
+    entry = make_config_entry(entry_id="remove-boom")
+    entry.add_update_listener = lambda listener: _remover
+    hass, _registry = make_hass()
+    await async_setup_entry(hass, entry)
+    database = entry.runtime_data.database
+    with pytest.raises(_RemoveBoom):
+        await async_unload_entry(hass, entry)
+    assert database.connected is False, "database leaked despite remover error"
+    assert entry.runtime_data is None
+    assert DOMAIN not in hass.data
+
+
+async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
+    """Setup applies the full v1 DDL: all four v1 tables exist afterwards."""
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    rows = await database.fetch_all(
+        "SELECT name FROM sqlite_master WHERE type = 'table' "
+        "AND name IN ('children', 'admin_users', 'schedule_rules', "
+        "'task_definitions') ORDER BY name"
+    )
+    assert [row[0] for row in rows] == [
+        "admin_users",
+        "children",
+        "schedule_rules",
+        "task_definitions",
+    ]
+    await async_unload_entry(hass, entry)

tests/test_store.py

@@ -100,11 +100,14 @@ def test_executor_helper_creates_directory(tmp_path) -> None:
     assert nested.is_dir()
 
 
-def test_no_sqlite3_import_in_package() -> None:
+def test_no_sqlite3_import_outside_db_module() -> None:
+    """sqlite3 is banned package-wide except db.py, the connection wrapper."""
     pkg_dir = Path(store.__file__).parent
     py_files = list(pkg_dir.glob("*.py"))
     assert py_files
     for py_file in py_files:
+        if py_file.name == "db.py":
+            continue
         tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
         for node in ast.walk(tree):
             if isinstance(node, ast.Import):