DAO unit test suite covering all eight tables (#23)

custom_components/nestquest/__init__.py

@@ -1,13 +1,17 @@
 """The NestQuest integration."""
 from __future__ import annotations
 
+import asyncio
 import inspect
+import sqlite3
 from collections.abc import Callable
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
 
 from .const import DOMAIN, LOGGER
 from .db import NestQuestDatabase
@@ -16,6 +20,191 @@ from .store import async_get_db_path
 
 _LOGGER = LOGGER
 
+#: SQLite reports "file is not a database" for corruption in every
+#: sqlite3 version this integration targets; the version-specific
+#: "encrypted or malformed" spelling (3.41+) maps to the same failure.
+_CORRUPTION_MARKERS = (
+    "file is not a database",
+    "database disk image is malformed",
+    "encrypted or malformed",
+    "is not a database",
+)
+
+#: Primary sqlite error codes that mean "the file is corrupt" (SQLite
+#: result codes, https://www.sqlite.org/c3ref/c_abort.html).  Extended
+#: codes carry the primary code in their low byte, so the match is on
+#: errcode & 0xff.
+_CORRUPT_PRIMARY_CODES = frozenset({11, 26})  # SQLITE_CORRUPT, SQLITE_NOTADB
+
+
+def _is_corruption_error(error: BaseException) -> bool:
+    """Return True when ``error`` indicates a corrupt database file.
+
+    Classification order: sqlite_errorcode (Python 3.11+) is primary
+    and locale-independent — SQLITE_CORRUPT and SQLITE_NOTADB, matched
+    on the primary byte so extended codes (SQLITE_CORRUPT_VTAB etc.)
+    classify correctly too.  Message matching is the fallback for
+    older Pythons: the corruption spellings are the documented English
+    texts; a non-English locale could evade them, but the code check
+    already covers every locale on supported Pythons.
+    """
+    if not isinstance(error, sqlite3.DatabaseError):
+        return False
+    errorcode = getattr(error, "sqlite_errorcode", None)
+    if errorcode is not None and (errorcode & 0xFF) in _CORRUPT_PRIMARY_CODES:
+        return True
+    message = str(error).lower()
+    return any(marker in message for marker in _CORRUPTION_MARKERS)
+
+
+async def _preflight_existing_file(
+    hass: HomeAssistant, db_path: Path
+) -> None:
+    """Detect corruption in an EXISTING file before it is opened read-write.
+
+    Opening read-write persists WAL mode — for a rollback-journal file
+    that rewrites the header byte 18/19 and can create -wal/-shm
+    sidecars — which would modify a corrupt file before detection and
+    break the leave-untouched guarantee.  This preflight therefore
+    validates the WHOLE file with a READ-ONLY connection (mode=ro +
+    immutable=1 URI, PRAGMA integrity_check, run on the executor like
+    every other DB call): corruption anywhere fails there, before
+    anything is written.  ``immutable=1`` is SQLite's documented
+    guarantee that the connection performs NO writes and creates NO
+    sidecars — even a WAL-header file gets its -wal/-shm reads served
+    without creating the shared-memory file.  Missing/empty files skip
+    the preflight (they are created fresh by the real open).  All
+    filesystem stat/URI work runs inside the executor job with the
+    probe so the event loop never touches disk.
+    """
+    def _probe() -> None:
+        # The file-state check and URI construction live inside the
+        # executor job: exists/stat/resolve are disk I/O.
+        if not db_path.exists() or db_path.stat().st_size == 0:
+            return
+        # Properly escape the path: '#'/'?'/'%' in a directory or file
+        # name would otherwise be parsed as URI fragments/parameters/
+        # escapes.  immutable=1: SQLite guarantees no writes and no
+        # sidecar creation from this connection.
+        uri = db_path.resolve().as_uri() + "?mode=ro&immutable=1"
+        conn = sqlite3.connect(uri, uri=True)
+        try:
+            row = conn.execute("PRAGMA integrity_check").fetchone()
+        finally:
+            conn.close()
+        verdict = row[0] if row else "missing"
+        if verdict != "ok":
+            # A non-ok integrity verdict IS corruption: raise with the
+            # verdict text (which matches the corruption markers) and
+            # the NOTADB code so classification is code-based.
+            raise sqlite3.DatabaseError(
+                f"integrity_check verdict {verdict!r} on read-only "
+                f"probe: file is not a database"
+            )
+
+    try:
+        # Every DB-touching call goes through the executor (the mock
+        # harness's async_add_executor_job is awaitable too).
+        await hass.async_add_executor_job(_probe)
+    except sqlite3.DatabaseError as err:
+        # Non-ok integrity verdicts are corruption, full stop: there is
+        # no benign reading of a failed whole-file integrity check, so
+        # ANY DatabaseError out of the probe that carries a corruption
+        # marker or code propagates as corruption.  Only exotic probe
+        # failures that classify as non-corruption (e.g. a transient
+        # 'database is locked' on the read) are downgraded to a debug
+        # log, letting the real open path surface them.
+        if _is_corruption_error(err):
+            raise
+        raise sqlite3.DatabaseError(
+            f"read-only corruption probe of {db_path} failed: {err}"
+        ) from err
+
+async def _async_open_database(
+    hass: HomeAssistant, db_path: Path
+) -> NestQuestDatabase:
+    """Open the NestQuest database, mapping failure modes for setup.
+
+    Three startup cases (Feature 02 done-condition):
+
+    - file missing: SQLite creates it fresh on open — no special path,
+      migrations stamp version 1 on the next step.
+    - file present and valid: opens normally.
+    - file present but corrupt: log an ERROR with the path, close
+      anything opened, and raise ConfigEntryNotReady so HA retries —
+      the file is left byte-for-byte untouched so the owner can
+      restore a backup.  No code path here deletes or overwrites an
+      existing database file.
+
+    Corruption can surface at EITHER stage: a truncated header fails
+    in the read-only preflight (before the file is ever opened
+    read-write), while damaged later pages fail during migrations —
+    the preflight guarantees no WAL sidecar or header write happens
+    on a corrupt file before it is detected.  Both are classified the
+    same way.
+    """
+    try:
+        await _preflight_existing_file(hass, db_path)
+        database = await NestQuestDatabase(hass).open(db_path)
+    except asyncio.CancelledError:
+        raise
+    except sqlite3.DatabaseError as err:
+        if _is_corruption_error(err):
+            _LOGGER.error(
+                "NestQuest database at %s is corrupt: %s. Refusing to "
+                "start with a corrupt database; the file was left "
+                "untouched so a backup can be restored. Home Assistant "
+                "will retry setup.",
+                db_path,
+                err,
+            )
+            raise ConfigEntryNotReady(
+                f"NestQuest database corrupt: {db_path}"
+            ) from err
+        # open() failed for a non-corruption reason (locked file,
+        # permission problem): retryable, but nothing was opened.
+        _LOGGER.error(
+            "NestQuest database at %s failed to open: %s. Home "
+            "Assistant will retry setup.",
+            db_path,
+            err,
+        )
+        raise ConfigEntryNotReady(
+            f"NestQuest database unavailable: {db_path}"
+        ) from err
+    try:
+        await apply_migrations(database)
+    except (sqlite3.DatabaseError, RuntimeError) as err:
+        await database.close()
+        if _is_corruption_error(err):
+            _LOGGER.error(
+                "NestQuest database at %s is corrupt: %s. Refusing to "
+                "start with a corrupt database; the file was left "
+                "untouched so a backup can be restored. Home Assistant "
+                "will retry setup.",
+                db_path,
+                err,
+            )
+            raise ConfigEntryNotReady(
+                f"NestQuest database corrupt: {db_path}"
+            ) from err
+        # A non-corruption failure (e.g. a locked file, a future schema
+        # version) is also a retryable setup problem, but without the
+        # corruption framing.
+        _LOGGER.error(
+            "NestQuest database at %s failed to open: %s. Home "
+            "Assistant will retry setup.",
+            db_path,
+            err,
+        )
+        raise ConfigEntryNotReady(
+            f"NestQuest database unavailable: {db_path}"
+        ) from err
+    except asyncio.CancelledError:
+        await database.close()
+        raise
+    return database
+
 
 @dataclass
 class NestQuestRuntimeData:
@@ -38,8 +227,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     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))
+        if db is None or not db.connected:
+            # Reopen through the SAME safety path as first setup: a
+            # record without a live connection (a previous setup's
+            # close-on-failure, or a hand-replaced file) must get the
+            # missing/valid/corrupt classification too, never a raw
+            # sqlite3 error and never skipped migrations.
+            db = await _async_open_database(
+                hass, await async_get_db_path(hass)
+            )
             existing.database = db
         entry.runtime_data = existing
         return True
@@ -50,9 +246,8 @@ 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))
+    database = await _async_open_database(hass, await async_get_db_path(hass))
     try:
-        await apply_migrations(database)
         remove_update_listener = entry.add_update_listener(_async_update_listener)
     except BaseException:
         await database.close()

custom_components/nestquest/dao_instances.py

@@ -0,0 +1,542 @@
+"""Typed DAO layer for task_instances and the append-only completion_events.
+
+All SQL for these tables lives in this module per the feature
+guardrails: callers get typed dataclasses back and never see raw rows
+or SQL.  Every method is async and runs through
+:class:`~.db.NestQuestDatabase`, so each statement executes on the HA
+executor and the event loop never blocks.
+
+Append-only policy (Feature 08, D-005): ``completion_events`` is
+strictly append-only.  This module is where that property is enforced
+mechanically — the completion-events DAO below exposes NO update or
+delete method at all; completions append, un-completions append a
+separate reversal event, and instance state derives from the latest
+event.  A package-wide guard test (bottom of the test module) keeps
+mutation SQL for this table out of every other module.
+
+Upsert semantics: ``TaskInstancesDao.upsert`` is idempotent on
+(definition_id, due_date) — re-running materialization for a date
+whose instance already exists updates nothing that matters
+(child_id/due_time are refreshed only in the degenerate re-generation
+case where they changed) and never duplicates a row.  The schema's
+UNIQUE(definition_id, due_date) is the last line of defense.
+
+``delete_future_uncompleted`` implements the Feature 06/07 rule that
+reassignment and schedule edits regenerate future instances: it
+removes only instances at or after the cutoff that have NO completion
+event, inside one transaction, so an instance completed mid-call can
+never be deleted.  Instances in the past are never touched (Feature 07
+guardrail: never delete history), and no method here deletes a single
+instance that has a completion event.
+"""
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+
+from .dao_children import _connection_lock
+from .dao_rules import _validate_date
+from .db import NestQuestDatabase
+
+#: Timestamp policy for completion_events.occurred_at: strict UTC
+#: ISO-8601 with explicit '+00:00' offset (module timestamp policy).
+#: One accepted shape bounds lexicographic range queries correctly, so
+#: the occurred-at scope does not guess at timestamp spellings.
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+def _validate_utc_timestamp(value: str, field: str) -> None:
+    """Raise ValueError unless ``value`` is a strict UTC ISO-8601 stamp.
+
+    Same strictness policy as _validate_date: parse with the exact
+    format, then round-trip the parsed value against the input so
+    non-padded or re-ordered shapes are rejected.  The round-trip
+    strips the UTC offset (isoformat() omits it when the offset is
+    zero), so the suffix is appended before comparing.
+    """
+    try:
+        parsed = datetime.datetime.strptime(
+            value, _UTC_TIMESTAMP_FORMAT
+        )
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"{field} must be a UTC ISO-8601 timestamp "
+            f"(YYYY-MM-DDTHH:MM:SS+00:00), got {value!r}"
+        ) from None
+    if parsed.isoformat() + "+00:00" != value:
+        raise ValueError(
+            f"{field} must be a strict UTC timestamp "
+            f"(YYYY-MM-DDTHH:MM:SS+00:00), got {value!r}"
+        )
+
+
+def _next_day_start(date_str: str) -> str:
+    """Return the UTC timestamp one second past ``date_str``'s last second.
+
+    Gives range queries an exact half-open upper bound without
+    fractional-second guessing.
+    """
+    day = datetime.date.fromisoformat(date_str)
+    next_day = day + datetime.timedelta(days=1)
+    return f"{next_day.isoformat()}T00:00:00+00:00"
+
+
+@dataclass(frozen=True)
+class TaskInstanceRecord:
+    """One row of ``task_instances``."""
+
+    id: int
+    definition_id: int
+    child_id: int
+    due_date: str
+    due_time: str | None
+    generated_at: str
+
+
+@dataclass(frozen=True)
+class CompletionEventRecord:
+    """One row of ``completion_events`` (append-only table)."""
+
+    id: int
+    instance_id: int
+    child_id: int
+    event_type: str
+    actor_source: str
+    actor_user_id: str | None
+    occurred_at: str
+    was_on_time: bool | None
+
+
+_INSTANCE_COLUMNS = (
+    "id, definition_id, child_id, due_date, due_time, generated_at"
+)
+_EVENT_COLUMNS = (
+    "id, instance_id, child_id, event_type, actor_source, actor_user_id, "
+    "occurred_at, was_on_time"
+)
+
+#: The only event types the append path accepts; the schema CHECK is
+#: the backstop, this is the DAO's typed front door.
+EVENT_COMPLETED = "completed"
+EVENT_UNCOMPLETED = "uncompleted"
+
+
+def _instance_from_row(row: tuple) -> TaskInstanceRecord:
+    return TaskInstanceRecord(
+        id=row[0],
+        definition_id=row[1],
+        child_id=row[2],
+        due_date=row[3],
+        due_time=row[4],
+        generated_at=row[5],
+    )
+
+
+def _event_from_row(row: tuple) -> CompletionEventRecord:
+    return CompletionEventRecord(
+        id=row[0],
+        instance_id=row[1],
+        child_id=row[2],
+        event_type=row[3],
+        actor_source=row[4],
+        actor_user_id=row[5],
+        occurred_at=row[6],
+        was_on_time=None if row[7] is None else bool(row[7]),
+    )
+
+
+class TaskInstancesDao:
+    """Typed async access to the ``task_instances`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def upsert(
+        self,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        generated_at: str,
+        *,
+        due_time: str | None = None,
+    ) -> TaskInstanceRecord:
+        """Create the dated instance for (definition_id, due_date) idempotently.
+
+        Running materialization twice for the same definition/date
+        yields ONE row: the ON CONFLICT path refreshes the
+        generation-time snapshot columns (child_id, due_time) and
+        regenerates the stamp, but can never create a duplicate — the
+        schema UNIQUE(definition_id, due_date) backs this up.
+
+        Guardrails enforced here:
+
+        - ``due_date`` must be TODAY or later: instances are never
+          generated in the past (Feature 07).  The caller's
+          materialization horizon is future-dated by definition.
+        - An existing instance WITH a completion event is IMMUTABLE:
+          the conflict path must not rewrite its snapshot columns
+          (Feature 07/08).  The upsert refuses with ValueError in that
+          case — silently no-oping would mask a materialization bug.
+
+        The definition must exist (validated atomically under the
+        connection lock so a concurrent definition deactivation or
+        child deletion cannot slip between check and insert).
+        """
+        _validate_date(due_date, "due_date")
+        # Fail fast on obviously-past dates BEFORE queuing on the
+        # connection lock (a nice early error); the AUTHORITATIVE
+        # check runs after the lock is acquired, because a caller can
+        # wait on the lock across midnight — a date that was "today"
+        # at call time may be "yesterday" by the time the INSERT runs.
+        today = datetime.date.today().isoformat()
+        if due_date < today:
+            raise ValueError(
+                f"instances are never generated in the past: due_date "
+                f"{due_date!r} is before today {today!r}"
+            )
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                # Re-read "today" under the lock: this is the date the
+                # insert actually executes on, so the no-past rule
+                # holds across midnight rollovers.
+                today = datetime.date.today().isoformat()
+                if due_date < today:
+                    raise ValueError(
+                        f"instances are never generated in the past: "
+                        f"due_date {due_date!r} is before today "
+                        f"{today!r}"
+                    )
+                definition = await self._database.fetch_one(
+                    "SELECT child_id FROM task_definitions WHERE id = ?",
+                    (definition_id,),
+                )
+                if definition is None:
+                    raise ValueError(
+                        f"task definition {definition_id} does not exist"
+                    )
+                if definition[0] != child_id:
+                    raise ValueError(
+                        f"task definition {definition_id} is assigned to "
+                        f"child {definition[0]}, not {child_id}; instance "
+                        "must carry the definition's assignee"
+                    )
+                child = await self._database.fetch_one(
+                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
+                )
+                if child is None:
+                    raise ValueError(f"child {child_id} does not exist")
+                existing = await self._database.fetch_one(
+                    "SELECT 1 FROM completion_events "
+                    "WHERE instance_id = (SELECT id FROM task_instances "
+                    "WHERE definition_id = ? AND due_date = ?)",
+                    (definition_id, due_date),
+                )
+                if existing is not None:
+                    raise ValueError(
+                        f"instance for (definition {definition_id}, "
+                        f"due_date {due_date!r}) already has a completion "
+                        "event and is immutable; it cannot be regenerated"
+                    )
+                await self._database.execute(
+                    "INSERT INTO task_instances (definition_id, child_id, "
+                    "due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, due_date) DO UPDATE SET "
+                    "child_id = excluded.child_id, "
+                    "due_time = excluded.due_time, "
+                    "generated_at = excluded.generated_at",
+                    (
+                        definition_id,
+                        child_id,
+                        due_date,
+                        due_time,
+                        generated_at,
+                    ),
+                )
+                instance = await self.get(definition_id, due_date)
+        assert instance is not None
+        return instance
+
+    async def get(
+        self, definition_id: int, due_date: str
+    ) -> TaskInstanceRecord | None:
+        """Return the instance for (definition_id, due_date), or None."""
+        row = await self._database.fetch_one(
+            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            "WHERE definition_id = ? AND due_date = ?",
+            (definition_id, due_date),
+        )
+        return _instance_from_row(row) if row is not None else None
+
+    async def get_by_id(self, instance_id: int) -> TaskInstanceRecord | None:
+        """Return the instance with ``instance_id``, or None."""
+        row = await self._database.fetch_one(
+            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances WHERE id = ?",
+            (instance_id,),
+        )
+        return _instance_from_row(row) if row is not None else None
+
+    async def list_by_child_and_date(
+        self, child_id: int, due_date: str
+    ) -> list[TaskInstanceRecord]:
+        """Return the child's instances due exactly on ``due_date``."""
+        _validate_date(due_date, "due_date")
+        rows = await self._database.fetch_all(
+            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            "WHERE child_id = ? AND due_date = ? ORDER BY id",
+            (child_id, due_date),
+        )
+        return [_instance_from_row(row) for row in rows]
+
+    async def list_by_date_range(
+        self,
+        child_id: int,
+        range_start: str,
+        range_end: str,
+    ) -> list[TaskInstanceRecord]:
+        """Return the child's instances with due_date in the closed range."""
+        _validate_date(range_start, "range_start")
+        _validate_date(range_end, "range_end")
+        if range_end < range_start:
+            raise ValueError(
+                "range_end must be on or after range_start, got "
+                f"{range_end!r} < {range_start!r}"
+            )
+        rows = await self._database.fetch_all(
+            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            "WHERE child_id = ? AND due_date >= ? AND due_date <= ? "
+            "ORDER BY due_date, id",
+            (child_id, range_start, range_end),
+        )
+        return [_instance_from_row(row) for row in rows]
+
+    async def delete_future_uncompleted(
+        self, definition_id: int, cutoff_date: str
+    ) -> int:
+        """Delete the definition's open instances at/after cutoff.
+
+        Implements regeneration after a definition edit or reassignment
+        (Features 06/07): only instances at or after the cutoff with NO
+        completion event are removed, inside one transaction — an
+        instance that gains a completion event mid-flight cannot be
+        deleted.  The cutoff cannot be backdated below today: "future"
+        is measured against TODAY (materialization regenerates the
+        rolling horizon from now), so a caller cannot use this to
+        rewrite past open instances either — the day-rollover missed
+        sweep (Feature 11) handles past-due instances instead.  Any
+        instance with a completion event is never touched, regardless
+        of its date.  Returns the number deleted.
+        """
+        _validate_date(cutoff_date, "cutoff_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                # "today" is read under the lock: the DELETE actually
+                # executes on this date, so a caller queued across
+                # midnight cannot delete instances that became past
+                # while it waited (backdating is clamped to the
+                # execution date, not the call date).
+                today = datetime.date.today().isoformat()
+                effective_cutoff = max(cutoff_date, today)
+                result = await self._database.execute(
+                    "DELETE FROM task_instances WHERE definition_id = ? "
+                    "AND due_date >= ? AND NOT EXISTS ("
+                    "  SELECT 1 FROM completion_events "
+                    "  WHERE instance_id = task_instances.id"
+                    ")",
+                    (definition_id, effective_cutoff),
+                )
+        return result.rowcount
+
+
+class CompletionEventsDao:
+    """Typed append-only access to the ``completion_events`` table.
+
+    APPEND-ONLY BY CONSTRUCTION: this class exposes ``append`` and the
+    three read methods below, and NOTHING else.  There is no update,
+    no delete, no upsert — completions and un-completions both arrive
+    through :meth:`append` as new rows, and history is immutable
+    (Feature 08 done-condition: no code path updates or deletes an
+    event).  The package-wide leak-guard test additionally fails if
+    mutation SQL for this table ever appears in any module.
+    """
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def append(
+        self,
+        instance_id: int,
+        child_id: int,
+        event_type: str,
+        actor_source: str,
+        occurred_at: str,
+        was_on_time: bool | None,
+        *,
+        actor_user_id: str | None = None,
+    ) -> CompletionEventRecord:
+        """Append one completion or un-completion event.
+
+        ``actor_source`` is 'user' (with ``actor_user_id`` set) or
+        'panel' (no user id — the tapped child is ``child_id``); the
+        actor-pair CHECK enforces this shape at the schema level.
+        ``occurred_at`` must be the module's one strict UTC shape
+        (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range queries
+        stay exact.  ``was_on_time`` is required (True/False) for
+        completions; for un-completions it may be None or the reversed
+        completion's flag (Feature 08 still deciding).  The instance
+        and child must exist — validated atomically under the
+        connection lock.
+        """
+        if event_type not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
+            raise ValueError(
+                f"event_type must be '{EVENT_COMPLETED}' or "
+                f"'{EVENT_UNCOMPLETED}', got {event_type!r}"
+            )
+        if actor_source not in ("user", "panel"):
+            raise ValueError(
+                f"actor_source must be 'user' or 'panel', got "
+                f"{actor_source!r}"
+            )
+        if actor_source == "user" and not actor_user_id:
+            raise ValueError(
+                "actor_source 'user' requires actor_user_id"
+            )
+        if actor_source == "panel" and actor_user_id is not None:
+            raise ValueError(
+                "actor_source 'panel' must not carry actor_user_id"
+            )
+        if was_on_time is not None and type(was_on_time) is not bool:
+            raise ValueError(
+                f"was_on_time must be True, False or None, got "
+                f"{was_on_time!r}"
+            )
+        if event_type == EVENT_COMPLETED and was_on_time is None:
+            raise ValueError(
+                "a completed event requires was_on_time (True/False)"
+            )
+        _validate_utc_timestamp(occurred_at, "occurred_at")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                instance = await self._database.fetch_one(
+                    "SELECT child_id FROM task_instances WHERE id = ?",
+                    (instance_id,),
+                )
+                if instance is None:
+                    raise ValueError(
+                        f"task instance {instance_id} does not exist"
+                    )
+                if instance[0] != child_id:
+                    raise ValueError(
+                        f"task instance {instance_id} belongs to child "
+                        f"{instance[0]}, not {child_id}"
+                    )
+                child = await self._database.fetch_one(
+                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
+                )
+                if child is None:
+                    raise ValueError(f"child {child_id} does not exist")
+                result = await self._database.execute(
+                    "INSERT INTO completion_events (instance_id, child_id, "
+                    "event_type, actor_source, actor_user_id, occurred_at, "
+                    "was_on_time) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        instance_id,
+                        child_id,
+                        event_type,
+                        actor_source,
+                        actor_user_id,
+                        occurred_at,
+                        None if was_on_time is None else int(was_on_time),
+                    ),
+                )
+                event = await self._database.fetch_one(
+                    f"SELECT {_EVENT_COLUMNS} FROM completion_events "
+                    "WHERE id = ?",
+                    (result.lastrowid,),
+                )
+        assert event is not None
+        return _event_from_row(event)
+
+    async def list_by_instance(
+        self, instance_id: int
+    ) -> list[CompletionEventRecord]:
+        """Return every event for the instance, in append order."""
+        rows = await self._database.fetch_all(
+            f"SELECT {_EVENT_COLUMNS} FROM completion_events "
+            "WHERE instance_id = ? ORDER BY id",
+            (instance_id,),
+        )
+        return [_event_from_row(row) for row in rows]
+
+    async def list_by_child_and_date_range(
+        self,
+        child_id: int,
+        range_start: str,
+        range_end: str,
+        *,
+        due_date_scope: bool = True,
+    ) -> list[CompletionEventRecord]:
+        """Return the child's events for instances due in the closed range.
+
+        Scopes by the INSTANCE's due date (the day the task was owed),
+        not the event's wall-clock timestamp: history reporting asks
+        "what happened to the tasks due that day" (Feature 13).  Set
+        ``due_date_scope=False`` to filter by ``occurred_at`` date
+        instead.  Ordered by instance, then event id.
+        """
+        _validate_date(range_start, "range_start")
+        _validate_date(range_end, "range_end")
+        if range_end < range_start:
+            raise ValueError(
+                "range_end must be on or after range_start, got "
+                f"{range_end!r} < {range_start!r}"
+            )
+        if due_date_scope:
+            predicate = (
+                "SELECT id FROM task_instances WHERE child_id = ? "
+                "AND due_date >= ? AND due_date <= ?"
+            )
+            parameters: list[object] = [child_id, range_start, range_end]
+        else:
+            # Occurred-at scope: occurred_at is validated to the one
+            # accepted UTC shape at append time, so half-open day
+            # bounds in that same shape are exact (no spelling drift).
+            occurred_start = f"{range_start}T00:00:00+00:00"
+            occurred_end = f"{range_end}T23:59:59+00:00"
+            query = (
+                f"SELECT {_EVENT_COLUMNS} FROM completion_events "
+                "WHERE child_id = ? AND occurred_at >= ? "
+                "AND occurred_at < ? ORDER BY instance_id, id"
+            )
+            rows = await self._database.fetch_all(
+                query,
+                [
+                    child_id,
+                    occurred_start,
+                    _next_day_start(range_end),
+                ],
+            )
+            return [_event_from_row(row) for row in rows]
+        rows = await self._database.fetch_all(
+            f"SELECT {_EVENT_COLUMNS} FROM completion_events "
+            f"WHERE instance_id IN ({predicate}) "
+            "ORDER BY instance_id, id",
+            tuple(parameters),
+        )
+        return [_event_from_row(row) for row in rows]
+
+    async def get_latest_for_instance(
+        self, instance_id: int
+    ) -> CompletionEventRecord | None:
+        """Return the instance's most recent event, or None if untouched.
+
+        Instance current state derives from this event (Feature 08):
+        'completed' means done, 'uncompleted' as the latest means open
+        again, None means never completed.
+        """
+        row = await self._database.fetch_one(
+            f"SELECT {_EVENT_COLUMNS} FROM completion_events "
+            "WHERE instance_id = ? ORDER BY id DESC LIMIT 1",
+            (instance_id,),
+        )
+        return _event_from_row(row) if row is not None else None

tests/conftest.py

@@ -41,6 +41,7 @@ _HA_MODULES = (
     "homeassistant.core",
     "homeassistant.config_entries",
     "homeassistant.data_entry_flow",
+    "homeassistant.exceptions",
 )
 
 def _find_spec(name: str):
@@ -77,6 +78,7 @@ for _parent, _child in (
     ("homeassistant", "config_entries"),
     ("homeassistant", "data_entry_flow"),
     ("homeassistant", "core"),
+    ("homeassistant", "exceptions"),
 ):
     setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
 
@@ -158,6 +160,7 @@ def _callback_decorator(fn):
 _config_entries_mock = _ha_mock("homeassistant.config_entries")
 _core_mock = _ha_mock("homeassistant.core")
 _data_entry_flow_mock = _ha_mock("homeassistant.data_entry_flow")
+_exceptions_mock = _ha_mock("homeassistant.exceptions")
 
 _config_entries_mock.ConfigFlow = _ConfigFlowBase
 _config_entries_mock.OptionsFlowWithConfigEntry = _OptionsFlowWithConfigEntryBase
@@ -168,6 +171,17 @@ _data_entry_flow_mock.RESULT_TYPE_ABORT = "abort"
 _data_entry_flow_mock.FlowResult = dict
 
 
+class ConfigEntryNotReady(Exception):
+    """Stand-in mirroring homeassistant.exceptions.ConfigEntryNotReady.
+
+    HA retries setup when this is raised; the tests assert the corrupt
+    database path raises it instead of crashing with an unrelated error.
+    """
+
+
+_exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
+
+
 # ---------------------------------------------------------------------------
 # Shared helpers for the behavior tests.
 # ---------------------------------------------------------------------------

tests/test_dao_children.py

@@ -484,6 +484,8 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
         "custom_components/nestquest/dao_presence.py",  # validates the
         # schedule/override child exists before writing (same FK shape
         # as dao_rules: one existence SELECT per write)
+        "custom_components/nestquest/dao_instances.py",  # validates the
+        # instance/event child exists before writing (same FK shape)
         "custom_components/nestquest/schema.py",  # declares the DDL
         "custom_components/nestquest/migrations.py",  # applies the DDL
         "tests/test_schema.py",  # tests the DDL
@@ -491,6 +493,10 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
         "tests/test_dao_children.py",  # this file, scanned separately
         "tests/test_dao_rules.py",  # rules/definitions guard, own scope
         "tests/test_dao_presence.py",  # presence guard, own scope
+        "tests/test_dao_instances.py",  # instances guard, own scope
+        "tests/test_startup_db.py",  # row-count probe of the children
+        # table only (proving a fresh setup loads); guard below covers
+        # everything else
     }
     sql_pattern = re.compile(
         r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+"
@@ -498,18 +504,329 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
         re.IGNORECASE,
     )
     offenders: list[str] = []
+    # test_dao_matrix.py is exempted for exactly its two sanctioned
+    # raw constraint-violation probes (NOT NULL on children, PRIMARY
+    # KEY on admin_users) — the Feature-02 closing suite must prove
+    # the schema surfaces violations to the DAO layer; its guard below
+    # covers everything else in that file.
+    probe_names = {
+        "test_matrix_children_crud_and_constraints",
+        "test_matrix_admin_users_crud_and_constraints",
+    }
     for root in scan_roots:
         for py in sorted(root.rglob("*.py")):
             relative = py.relative_to(repo_root).as_posix()
             if relative in allowed:
                 continue
-            if sql_pattern.search(py.read_text()):
+            text = py.read_text()
+            if relative == "tests/test_dao_matrix.py":
+                text = _strip_function_spans(text, probe_names)
+            if sql_pattern.search(text):
                 offenders.append(relative)
     assert offenders == [], (
         f"SQL touching children/admin_users leaked into: {offenders}"
     )
 
 
+def _strip_function_spans(text: str, names: set[str]) -> str:
+    """Return ``text`` minus the source spans of the named functions."""
+    import ast
+
+    lines = text.splitlines(keepends=True)
+    excluded: set[int] = set()
+    tree = ast.parse(text)
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and node.name in names
+        ):
+            excluded.update(
+                range(node.lineno, (node.end_lineno or node.lineno) + 1)
+            )
+    return "".join(
+        line
+        for number, line in enumerate(lines, start=1)
+        if number not in excluded
+    )
+
+
+def test_startup_db_test_file_uses_dao_elsewhere() -> None:
+    """Compensating self-scan for the startup-test exemption: outside
+    the single sanctioned COUNT(*) probe, test_startup_db.py must go
+    through the DAO for children/admin_users data access.
+    """
+    import re
+    from pathlib import Path
+
+    startup_test = Path(__file__).parent / "test_startup_db.py"
+    stripped = _strip_function_spans(
+        startup_test.read_text(),
+        {"test_setup_with_valid_file_preserves_rows"},
+    )
+    sql_pattern = re.compile(
+        r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(children|admin_users)\b",
+        re.IGNORECASE,
+    )
+    assert sql_pattern.search(stripped) is None, (
+        "test_startup_db.py must go through the DAO for these tables "
+        "outside its sanctioned row-count probe"
+    )
+
+
+def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None:
+    """Compensating self-scan for the matrix-file exemption, structurally
+    exact: parse the two sanctioned functions' ASTs and require that
+    (a) they contain NO other raw SQL naming children/admin_users, and
+    (b) every raw table insert among them is an expression nested inside
+    a ``pytest.raises`` context whose matcher mentions IntegrityError.
+    Any additional raw write elsewhere in the file fails this guard.
+    """
+    import ast
+    import re
+    from pathlib import Path
+
+    matrix = Path(__file__).parent / "test_dao_matrix.py"
+    text = matrix.read_text()
+    stripped = _strip_function_spans(
+        text,
+        {
+            "test_matrix_children_crud_and_constraints",
+            "test_matrix_admin_users_crud_and_constraints",
+        },
+    )
+    sql_pattern = re.compile(
+        r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(children|admin_users)\b",
+        re.IGNORECASE,
+    )
+    assert sql_pattern.search(stripped) is None, (
+        "test_dao_matrix.py raw children/admin_users SQL is limited to "
+        "the two sanctioned constraint probes"
+    )
+
+    tree = ast.parse(text)
+    probe_tables = {
+        "test_matrix_children_crud_and_constraints": ("children", "INSERT"),
+        "test_matrix_admin_users_crud_and_constraints": (
+            "admin_users",
+            "INSERT",
+        ),
+    }
+    any_table = re.compile(
+        r"\b(children|admin_users)\b", re.IGNORECASE
+    )
+    allowed_stmt = re.compile(
+        r"^(INSERT|SELECT)", re.IGNORECASE
+    )
+    #: The sanctioned probe per function: exactly one raw statement
+    #: INSERTing that table, inside a pytest.raises whose matcher is
+    #: IntegrityError (checked as an AST Name/Attribute chain).
+    probes_per_function: dict[str, int] = {name: 0 for name in probe_tables}
+    for node in ast.walk(tree):
+        if (
+            not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            or node.name not in probe_tables
+        ):
+            continue
+        table, expected_verb = probe_tables[node.name]
+
+        def _unquote(rendered: str) -> str:
+            rendered = rendered.strip()
+            if len(rendered) >= 2 and rendered[0] == rendered[-1] and (
+                rendered[0] in "'\""
+            ):
+                return rendered[1:-1]
+            return rendered
+
+        # Collect EVERY database SQL entry point (execute, fetch_one,
+        # fetch_all, execute_many) wherever it appears in the awaited
+        # expression tree: a bypass through fetch_one/execute_many OR a
+        # call wrapped in another awaitable (asyncio.gather, a helper
+        # coroutine, a Task) must be caught the same as a bare await.
+        raw_statements: list[tuple[ast.AST, str]] = []
+        for sub in ast.walk(node):
+            if (
+                isinstance(sub, ast.Call)
+                and isinstance(sub.value if False else sub.func, ast.Attribute)
+                and sub.func.attr
+                in {"execute", "fetch_one", "fetch_all", "execute_many"}
+            ):
+                sql_arg = sub.args[0] if sub.args else None
+                rendered = ast.unparse(sql_arg) if sql_arg else ""
+                raw_statements.append((sub, _unquote(rendered)))
+
+        def _raises_blocks():
+            """Yield (block, is_pytest_integrity) for pytest.raises blocks.
+
+            Structurally exact: the call func must resolve to
+            pytest.raises — Attribute(pytest, raises) or a bare Name
+            whose id is bound to pytest via the file's own import
+            section (validated below).  The matcher argument must be an
+            AST chain whose final segment is exactly IntegrityError (so
+            NotIntegrityError and fake.IntegrityError fail).
+            """
+            # Names bound to pytest's raises: parse the FILE's actual
+            # Import nodes in TWO passes so order cannot matter — pass
+            # one collects every binding pytest's raises gets (whatever
+            # alias), pass two rejects ANY competing binding of those
+            # names (or of 'pytest' itself) from a non-pytest import,
+            # wherever it appears.  A spoof imported BEFORE the genuine
+            # binding therefore still fails.
+            pytest_raises_aliases: set[str] = set()
+            for file_node in ast.walk(tree):
+                if (
+                    isinstance(file_node, ast.ImportFrom)
+                    and file_node.module == "pytest"
+                ):
+                    for alias in file_node.names:
+                        if alias.name == "raises":
+                            pytest_raises_aliases.add(
+                                alias.asname or alias.name
+                            )
+            raises_bindings: set[str] = set(pytest_raises_aliases)
+            for file_node in ast.walk(tree):
+                if isinstance(file_node, ast.ImportFrom):
+                    if file_node.module == "pytest":
+                        continue
+                    for alias in file_node.names:
+                        bound = alias.asname or alias.name
+                        if bound in raises_bindings | {"pytest"}:
+                            raise AssertionError(
+                                f"shadowed import of {bound!r} from "
+                                f"{file_node.module!r} defeats the "
+                                "pytest.raises probe validation"
+                            )
+                elif isinstance(file_node, ast.Import):
+                    # import fake_module as pytest (or as any raises
+                    # alias) fails closed.  A plain `import pytest`
+                    # (alias.name == 'pytest', no asname) is the genuine
+                    # module binding and is allowed.
+                    for alias in file_node.names:
+                        bound = alias.asname or alias.name
+                        genuine_pytest = (
+                            alias.name == "pytest" and alias.asname is None
+                        )
+                        if bound in raises_bindings | {"pytest"} and (
+                            not genuine_pytest
+                        ):
+                            raise AssertionError(
+                                f"import shadowing {bound!r} defeats the "
+                                "pytest.raises probe validation"
+                            )
+            for sub in ast.walk(node):
+                if not (
+                    isinstance(sub, ast.With)
+                    and sub.items
+                    and isinstance(sub.items[0].context_expr, ast.Call)
+                ):
+                    continue
+                call = sub.items[0].context_expr
+                func = call.func
+                if isinstance(func, ast.Attribute):
+                    if (
+                        func.attr != "raises"
+                        or not isinstance(func.value, ast.Name)
+                        or func.value.id != "pytest"
+                    ):
+                        continue
+                elif isinstance(func, ast.Name):
+                    if func.id not in raises_bindings:
+                        continue
+                else:
+                    continue
+                matcher_ok = False
+                if call.args:
+                    matcher = call.args[0]
+                    matcher_ok = _is_integrity_error_node(matcher)
+                yield sub, matcher_ok
+
+        def _is_integrity_error_node(matcher) -> bool:
+            """True for an AST chain of Name/Attribute ending exactly at
+            IntegrityError: every Attribute base must itself be a Name
+            or Attribute (a Call base like __import__('x') fails)."""
+            node = matcher
+            while isinstance(node, ast.Attribute):
+                node = node.value
+            if not isinstance(node, ast.Name):
+                return False
+            final = matcher
+            while isinstance(final, ast.Attribute):
+                final = final.attr
+            return (
+                isinstance(final, str)
+                and ast.unparse(matcher).split(".")[-1] == "IntegrityError"
+            )
+
+        # INSERT OR variant / quoted / qualified / schema-qualified
+        # names: normalize the SQL so ANY spelling of an insert into
+        # the table counts as a table insert and requires the raises
+        # block.  Strip quotes/brackets/backticks from identifiers and
+        # drop any schema qualifier, whatever it is.
+        def _is_insert_into_table(rendered: str, table: str) -> bool:
+            insert_match = re.search(
+                r"\bINSERT(\s+OR\s+\w+)?\s+INTO\s+(.+)",
+                rendered,
+                flags=re.IGNORECASE | re.DOTALL,
+            )
+            if not insert_match:
+                return False
+            target = insert_match.group(2)
+            # Tokenize the target's leading identifier: strip quotes
+            # and brackets, take the LAST dot-separated segment,
+            # casefolded so 'CHILDREN' cannot bypass the lowercase
+            # table name the scan looks for.
+            identifier = re.split(r"[\s(]", target, maxsplit=1)[0]
+            segments = re.split(r"\.", identifier)
+            last = segments[-1].strip("`'\"[]")
+            return last.casefold() == table.casefold()
+
+        # In a sanctioned function, EVERY raw statement naming either
+        # table must be exactly that function's one expected-table
+        # constraint insert — an admin_users insert inside the children
+        # function (or vice versa) is a bypass attempt and fails here,
+        # as does any second insert of the expected table.
+        for stmt_node, rendered in raw_statements:
+            if not any_table.search(rendered):
+                continue
+            assert _is_insert_into_table(rendered, table), (
+                f"{node.name}: raw statement naming the other table is "
+                f"forbidden; only the {table} probe is sanctioned: "
+                f"{rendered[:80]!r}"
+            )
+            # This is the sanctioned probe: require pytest.raises
+            # with an exact IntegrityError matcher around it.
+            inside = any(
+                block.lineno <= stmt_node.lineno
+                and stmt_node.end_lineno <= block.end_lineno
+                and matcher_ok
+                for block, matcher_ok in _raises_blocks()
+            )
+            assert inside, (
+                f"the raw {table} insert in {node.name} must be nested "
+                "in a pytest.raises(...IntegrityError...) block "
+                "(expected-to-fail constraint probe, not a write)"
+            )
+            probes_per_function[node.name] += 1
+        # UPDATE/DELETE naming the tables anywhere in a sanctioned
+        # function is forbidden outright.
+        for _, rendered in raw_statements:
+            if any_table.search(rendered) and re.match(
+                r"(UPDATE|DELETE)", rendered.strip(), re.IGNORECASE
+            ):
+                raise AssertionError(
+                    f"{node.name} must not hold UPDATE/DELETE SQL for "
+                    f"children/admin_users: {rendered[:80]!r}"
+                )
+    assert all(
+        count == 1 for count in probes_per_function.values()
+    ) and len(probes_per_function) == 2, (
+        "each sanctioned function must hold exactly one raw constraint "
+        f"probe; found {probes_per_function}"
+    )
+
+
+
 def test_dao_test_file_uses_dao_not_raw_table_sql() -> None:
     """This test module must exercise the DAO, not raw SQL, for these
     tables.  The two guard functions in this file necessarily mention
@@ -532,6 +849,7 @@ def test_dao_test_file_uses_dao_not_raw_table_sql() -> None:
     guard_names = {
         "test_children_and_admin_sql_lives_only_in_dao_module",
         "test_dao_test_file_uses_dao_not_raw_table_sql",
+        "test_dao_matrix_test_file_raw_probes_are_only_constraint_probes",
     }
     excluded: set[int] = set()
     for node in ast.walk(tree):

tests/test_dao_matrix.py

@@ -0,0 +1,423 @@
+"""Cross-table DAO unit suite: CRUD and constraint coverage for all tables.
+
+The per-DAO test modules (test_dao_children, test_dao_rules,
+test_dao_presence, test_dao_instances) cover each table's operations in
+depth; this module is the FEATURE-02 closing sweep: a single systematic
+matrix proving that for EVERY table, the DAO layer supports insert,
+read, update and surfaces constraint violations — plus the instance
+upsert idempotency and the completion_events no-mutation guarantee —
+all against a temporary database file from a clean checkout.
+
+No raw SQL touches the DAO-owned tables here except the sanctioned
+COUNT(*) probes (each guarded by the per-DAO leak guards); everything
+else goes through the typed DAO methods.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import sqlite3
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from custom_components.nestquest.dao_children import (
+    AdminUsersDao,
+    ChildrenDao,
+)
+from custom_components.nestquest.dao_instances import (
+    CompletionEventsDao,
+    TaskInstancesDao,
+)
+from custom_components.nestquest.dao_presence import (
+    PresenceOverridesDao,
+    PresenceSchedulesDao,
+)
+from custom_components.nestquest.dao_rules import (
+    ScheduleRulesDao,
+    TaskDefinitionsDao,
+)
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+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 _future_day(offset: int) -> str:
+    return (
+        datetime.date.today() + datetime.timedelta(days=offset)
+    ).isoformat()
+
+
+def _stamp() -> str:
+    return f"{datetime.date.today().isoformat()}T08:00:00+00:00"
+
+
+D1 = _future_day(1)
+D2 = _future_day(2)
+D20 = _future_day(20)
+
+#: Every table the v1 schema owns, with the DAO that governs it.
+ALL_TABLES = (
+    "children",
+    "admin_users",
+    "schedule_rules",
+    "task_definitions",
+    "presence_schedules",
+    "presence_overrides",
+    "task_instances",
+    "completion_events",
+)
+
+
+class World:
+    """All DAOs over one migrated temporary database."""
+
+    def __init__(self, database: NestQuestDatabase):
+        self.database = database
+        self.children = ChildrenDao(database)
+        self.admins = AdminUsersDao(database)
+        self.rules = ScheduleRulesDao(database)
+        self.definitions = TaskDefinitionsDao(database)
+        self.schedules = PresenceSchedulesDao(database)
+        self.overrides = PresenceOverridesDao(database)
+        self.instances = TaskInstancesDao(database)
+        self.events = CompletionEventsDao(database)
+
+    async def seed(self) -> None:
+        """Populate one valid fixture chain through the DAO layer."""
+        self.child = await self.children.create("Ada", _stamp())
+        self.child2 = await self.children.create("Bo", _stamp())
+        assert await self.admins.add("user-1", _stamp()) is True
+        self.rule = await self.rules.create("daily", D1)
+        self.definition = await self.definitions.create(
+            "Brush teeth", self.child.id, self.rule.id, _stamp()
+        )
+        self.schedule = await self.schedules.upsert_by_child(
+            self.child.id, 2, D1, "0,2,4|1,3"
+        )
+        self.override = await self.overrides.create(
+            self.child.id, D1, D2, True, note="test"
+        )
+        self.instance = await self.instances.upsert(
+            self.definition.id, self.child.id, D1, _stamp()
+        )
+
+
+async def _world(path) -> World:
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    world = World(database)
+    await world.seed()
+    return world
+
+
+def _with_world(name):
+    def _run_test(body):
+        async def _main():
+            world = await _world(name)
+            try:
+                return await body(world)
+            finally:
+                await world.database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# The matrix: every table gets insert / read / update / constraint check
+# ---------------------------------------------------------------------------
+
+
+def test_matrix_children_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        fetched = await w.children.get(w.child.id)
+        assert fetched is not None and fetched.display_name == "Ada"
+        # update
+        await w.children.update(w.child.id, display_name="Ada L.")
+        assert (await w.children.get(w.child.id)).display_name == "Ada L."
+        # constraint: NOT NULL display_name
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.database.execute(
+                "INSERT INTO children (display_name, created_at) "
+                "VALUES (NULL, ?)",
+                (_stamp(),),
+            )
+        return True
+
+    assert _with_world(tmp_path / "m-children.db")(_body) is True
+
+
+def test_matrix_admin_users_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert + read
+        assert await w.admins.exists("user-1") is True
+        records = await w.admins.list()
+        assert len(records) == 1
+        # update path: idempotent add is the allowlist's update semantics
+        assert await w.admins.add("user-1", _stamp()) is False
+        # constraint: ha_user_id PRIMARY KEY
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.database.execute(
+                "INSERT INTO admin_users (ha_user_id, added_at) "
+                "VALUES (?, ?)",
+                ("user-1", _stamp()),
+            )
+        # remove
+        assert await w.admins.remove("user-1") is True
+        assert await w.admins.exists("user-1") is False
+        return True
+
+    assert _with_world(tmp_path / "m-admins.db")(_body) is True
+
+
+def test_matrix_schedule_rules_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        fetched = await w.rules.get(w.rule.id)
+        assert fetched.rule_type == "daily"
+        # update
+        await w.rules.update(w.rule.id, interval=3)
+        assert (await w.rules.get(w.rule.id)).interval == 3
+        # constraint: rule_type CHECK
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.rules.create("sometimes", D1)
+        # delete-if-unreferenced: referenced -> rejected
+        assert await w.rules.delete_if_unreferenced(w.rule.id) is False
+        return True
+
+    assert _with_world(tmp_path / "m-rules.db")(_body) is True
+
+
+def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        fetched = await w.definitions.get(w.definition.id)
+        assert fetched.title == "Brush teeth"
+        # update
+        await w.definitions.update(w.definition.id, title="Brush TEETH")
+        assert (await w.definitions.get(w.definition.id)).title == (
+            "Brush TEETH"
+        )
+        # constraint: FK to a nonexistent child rejected with ValueError
+        with pytest.raises(ValueError, match="does not exist"):
+            await w.definitions.create("X", 999, w.rule.id, _stamp())
+        # constraint: inactive child rejected
+        await w.children.set_active(w.child2.id, False)
+        with pytest.raises(ValueError, match="inactive"):
+            await w.definitions.set_assignee(w.definition.id, w.child2.id)
+        # set_active: deactivate, never delete
+        assert await w.definitions.set_active(w.definition.id, False) == 1
+        assert (await w.definitions.get(w.definition.id)).is_active is False
+        return True
+
+    assert _with_world(tmp_path / "m-definitions.db")(_body) is True
+
+
+def test_matrix_presence_schedules_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        fetched = await w.schedules.get_by_child(w.child.id)
+        assert fetched.pattern == "0,2,4|1,3"
+        # update path: upsert replaces in place (same row id)
+        replaced = await w.schedules.upsert_by_child(
+            w.child.id, 2, D2, "1,3|0,2,4"
+        )
+        assert replaced.id == fetched.id
+        assert (await w.schedules.get_by_child(w.child.id)).pattern == (
+            "1,3|0,2,4"
+        )
+        # constraint: malformed pattern CHECK
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.schedules.upsert_by_child(w.child.id, 2, D1, "0,2")
+        # delete: row gone, child present every day
+        assert await w.schedules.delete(w.child.id) is True
+        assert await w.schedules.get_by_child(w.child.id) is None
+        return True
+
+    assert _with_world(tmp_path / "m-schedules.db")(_body) is True
+
+
+def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        found = await w.overrides.list_by_child_and_range(
+            w.child.id, D1, D2
+        )
+        assert [r.id for r in found] == [w.override.id]
+        # update path: delete + recreate (overrides are append-oriented)
+        assert await w.overrides.delete(w.override.id) is True
+        replacement = await w.overrides.create(
+            w.child.id, D1, D2, False, note="updated"
+        )
+        assert replacement.is_present is False
+        # constraint: end before start CHECK
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.overrides.create(w.child.id, D2, D1, True)
+        return True
+
+    assert _with_world(tmp_path / "m-overrides.db")(_body) is True
+
+
+def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # insert (seeded) + read
+        fetched = await w.instances.get(w.definition.id, D1)
+        assert fetched == w.instance
+        # update path: conflict refresh on the OPEN instance — same row
+        # id, snapshot columns (due_time) actually rewritten.
+        refreshed = await w.instances.upsert(
+            w.definition.id, w.child.id, D1, _stamp(), due_time="17:00"
+        )
+        assert refreshed.id == fetched.id, (
+            "the conflict path must refresh the existing row in place"
+        )
+        assert refreshed.due_time == "17:00"
+        assert fetched.due_time is None
+        assert await w.instances.get_by_id(fetched.id) is not None
+        return True
+
+    assert _with_world(tmp_path / "m-instances.db")(_body) is True
+
+
+def test_matrix_task_instances_constraints(tmp_path) -> None:
+    async def _body(w: World):
+        # D20 has no instance yet: a raw insert succeeds, then an
+        # identical second insert must fail on the UNIQUE constraint —
+        # proving the schema backs the DAO's idempotency.
+        await w.database.execute(
+            "INSERT INTO task_instances (definition_id, child_id, "
+            "due_date, due_time, generated_at) "
+            "VALUES (?, ?, ?, NULL, ?)",
+            (w.definition.id, w.child.id, D20, _stamp()),
+        )
+        with pytest.raises(sqlite3.IntegrityError):
+            await w.database.execute(
+                "INSERT INTO task_instances (definition_id, child_id, "
+                "due_date, due_time, generated_at) "
+                "VALUES (?, ?, ?, NULL, ?)",
+                (w.definition.id, w.child.id, D20, _stamp()),
+            )
+        # delete future uncompleted removes D20's open instance
+        deleted = await w.instances.delete_future_uncompleted(
+            w.definition.id, D20
+        )
+        assert deleted == 1
+        return True
+
+    assert _with_world(tmp_path / "m-instances-constraints.db")(_body) is True
+
+
+def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None:
+    async def _body(w: World):
+        first = await w.instances.upsert(
+            w.definition.id, w.child.id, D2, _stamp()
+        )
+        second = await w.instances.upsert(
+            w.definition.id, w.child.id, D2, _stamp(), due_time="17:00"
+        )
+        assert second.id == first.id
+        count = await w.database.fetch_one(
+            "SELECT COUNT(*) FROM task_instances"
+        )
+        # Seed created one (D1) plus this one (D2) — no duplicate.
+        assert count == (2,)
+        return True
+
+    assert _with_world(tmp_path / "m-upsert-idempotent.db")(_body) is True
+
+
+# ---------------------------------------------------------------------------
+# completion_events: append-only by construction
+# ---------------------------------------------------------------------------
+
+
+def test_matrix_completion_events_append_read_and_no_mutation(
+    tmp_path,
+) -> None:
+    async def _body(w: World):
+        # insert: append a completion then a reversal, then re-complete
+        await w.events.append(
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), True
+        )
+        await w.events.append(
+            w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None,
+            actor_user_id="user-1",
+        )
+        await w.events.append(
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), False
+        )
+        # read: full ordered history + latest
+        history = await w.events.list_by_instance(w.instance.id)
+        assert [e.event_type for e in history] == [
+            "completed",
+            "uncompleted",
+            "completed",
+        ]
+        latest = await w.events.get_latest_for_instance(w.instance.id)
+        assert latest.event_type == "completed"
+        assert latest.was_on_time is False
+        # constraint: unknown instance FK + bad actor pair surface as
+        # ValueError from the DAO's atomic validation
+        with pytest.raises(ValueError, match="does not exist"):
+            await w.events.append(
+                999, w.child.id, "completed", "user", _stamp(), True,
+                actor_user_id="user-1",
+            )
+        return history
+
+    assert _with_world(tmp_path / "m-events.db")(_body) is not None
+
+
+def test_completion_events_absolutely_no_mutation_path() -> None:
+    """The structural half of the done-condition: the DAO class exposes
+    no update/delete, AND no module in the package holds mutation SQL
+    for the table (already asserted package-wide in test_dao_instances;
+    re-asserted here so THIS suite stands alone in a clean checkout).
+    """
+    import inspect
+    import re
+    from pathlib import Path
+
+    methods = {
+        name
+        for name, _ in inspect.getmembers(
+            CompletionEventsDao, inspect.isfunction
+        )
+        if not name.startswith("__")
+    }
+    assert methods == {
+        "append",
+        "list_by_instance",
+        "list_by_child_and_date_range",
+        "get_latest_for_instance",
+    }
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+    mutation = re.compile(
+        r"(UPDATE(\s+OR\s+\w+)?|DELETE\s+FROM)\s+"
+        r"((main|temp)\s*\.\s*)?[`'\"]*(\[)?completion_events\b",
+        re.IGNORECASE,
+    )
+    offenders = [
+        py.name
+        for py in sorted(package.glob("*.py"))
+        if py.name != "__pycache__" and mutation.search(py.read_text())
+    ]
+    assert offenders == []

tests/test_dao_rules.py

@@ -789,15 +789,15 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
 
     allowed = {
         "custom_components/nestquest/dao_rules.py",  # this DAO
+        "custom_components/nestquest/dao_instances.py",  # validates the
+        # definition exists + fetches its assignee before generating an
+        # instance (the instance must snapshot the definition's child)
         "custom_components/nestquest/schema.py",  # declares the DDL
         "custom_components/nestquest/migrations.py",  # applies the DDL
         "tests/test_schema.py",  # tests the DDL
         "tests/test_migrations.py",  # tests migration application
         "tests/test_dao_rules.py",  # this file, scanned separately
-        # dao_children.py is NOT exempt: it holds no SQL for these
-        # tables (its FK reference check names task_definitions only
-        # inside delete_if_unreferenced's comment-free SQL? No — that
-        # lives HERE in dao_rules).  Any future leak there must fail.
+        "tests/test_dao_instances.py",  # instances guard, own scope
     }
     sql_pattern = re.compile(
         r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
@@ -818,6 +818,50 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
     )
 
 
+def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
+    """Compensating self-scan for the test_dao_instances.py exemption:
+    that file may query task_instances/completion_events raw (its own
+    guard's scope) but must go through the DAO for schedule_rules and
+    task_definitions, except its single sanctioned definition-create
+    helper usage and guard spans.
+    """
+    import ast
+    import re
+    from pathlib import Path
+
+    instances_test = Path(__file__).parent / "test_dao_instances.py"
+    text = instances_test.read_text()
+    lines = text.splitlines(keepends=True)
+    tree = ast.parse(text)
+    guard_names = {
+        "test_rules_and_definitions_sql_lives_only_in_dao_module",
+        "test_no_mutation_sql_for_completion_events_anywhere",
+    }
+    excluded: set[int] = set()
+    for node in ast.walk(tree):
+        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        if node.name in guard_names:
+            excluded.update(
+                range(node.lineno, (node.end_lineno or node.lineno) + 1)
+            )
+
+    remaining = "".join(
+        line
+        for number, line in enumerate(lines, start=1)
+        if number not in excluded
+    )
+    sql_pattern = re.compile(
+        r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|task_definitions)\b",
+        re.IGNORECASE,
+    )
+    assert sql_pattern.search(remaining) is None, (
+        "test_dao_instances.py must go through the DAO, not raw SQL, "
+        "for schedule_rules/task_definitions"
+    )
+
+
 def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
     """This test module must exercise the DAO, not raw SQL, for these
     tables.  Exceptions: the guard functions' own spans (their regexes

tests/test_migrations.py

@@ -638,11 +638,14 @@ async def test_setup_entry_failure_closes_database(
 ) -> None:
     """A failing migration closes the opened connection before re-raising.
 
-    Spies on the real wrapper instance: records whether close() ran and
-    whether the connection is left disconnected, rather than inferring
-    from hass.data (a failed setup never stores a runtime record).
+    A migration failure surfaces as ConfigEntryNotReady (HA retries
+    setup) with the original error preserved as __cause__.  Spies on
+    the real wrapper instance: records whether close() ran, rather
+    than inferring from hass.data (a failed setup never stores a
+    runtime record).
     """
     import pytest as pytest_module
+    from homeassistant.exceptions import ConfigEntryNotReady
 
     import custom_components.nestquest as nestquest_module
     from custom_components.nestquest.db import NestQuestDatabase
@@ -667,8 +670,10 @@ async def test_setup_entry_failure_closes_database(
     monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
 
     entry = make_entry()
-    with pytest_module.raises(RuntimeError, match="boom"):
+    with pytest_module.raises(ConfigEntryNotReady) as excinfo:
         await _setup_entry(hass, entry, hass.registry)
+    assert isinstance(excinfo.value.__cause__, RuntimeError)
+    assert "boom" in str(excinfo.value.__cause__)
 
     assert close_calls == [True], (
         "failed setup must close the connection it opened"

tests/test_startup_db.py

@@ -0,0 +1,447 @@
+"""Tests for startup handling of missing, valid, and corrupt databases."""
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import sqlite3
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from conftest import make_config_entry
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from custom_components.nestquest import (
+    _async_open_database,
+    _is_corruption_error,
+    async_setup_entry,
+    async_unload_entry,
+)
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import (
+    VERSION_TABLE,
+    apply_migrations,
+)
+from tests.test_lifecycle import _wire
+
+
+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) -> NestQuestDatabase:
+    database = NestQuestDatabase(_make_hass_mock())
+    _run(database.open(path))
+    return database
+
+
+async def _avalid_database(path) -> None:
+    """Create a valid, migrated NestQuest database file (async, on the
+    running loop, for use inside pytest-asyncio tests)."""
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    try:
+        await apply_migrations(database)
+    finally:
+        await database.close()
+
+
+def _valid_database(path) -> None:
+    """Create a valid, migrated NestQuest database file at ``path``."""
+    database = _open_db(path)
+    try:
+        _run(apply_migrations(database))
+    finally:
+        _run(database.close())
+
+
+def _truncated_database(path) -> None:
+    """Create a deliberately truncated (corrupt) database file.
+
+    A valid database is written, closed, then cut short mid-header:
+    opening it reports 'file is not a database' — exactly what an
+    interrupted write or failing disk leaves behind.
+    """
+    _valid_database(path)
+    data = path.read_bytes()
+    assert len(data) > 64, "valid database must exist before truncation"
+    path.write_bytes(data[: len(data) // 2])
+
+
+# ---------------------------------------------------------------------------
+# Corruption classification
+# ---------------------------------------------------------------------------
+
+
+def test_is_corruption_error_matches_sqlite_spellings() -> None:
+    assert _is_corruption_error(
+        sqlite3.DatabaseError("file is not a database")
+    )
+    assert _is_corruption_error(
+        sqlite3.DatabaseError("file is not a database".upper())
+    )
+    assert _is_corruption_error(
+        sqlite3.DatabaseError("file is encrypted or malformed")
+    )
+    assert _is_corruption_error(
+        sqlite3.OperationalError("file is not a database")
+    )
+
+
+def test_is_corruption_error_rejects_other_failures() -> None:
+    assert not _is_corruption_error(
+        sqlite3.OperationalError("database is locked")
+    )
+    assert not _is_corruption_error(
+        sqlite3.IntegrityError("NOT NULL constraint failed")
+    )
+    assert not _is_corruption_error(RuntimeError("boom"))
+    assert not _is_corruption_error(ValueError("nope"))
+
+
+# ---------------------------------------------------------------------------
+# Startup: three cases through _async_open_database
+# ---------------------------------------------------------------------------
+
+
+def test_missing_file_opens_fresh(tmp_path) -> None:
+    db_path = tmp_path / "nestquest.db"
+    assert not db_path.exists()
+    database = _run(_async_open_database(_make_hass_mock(), db_path))
+    try:
+        # The file was created fresh and migrations stamped version 1.
+        assert db_path.exists()
+        row = _run(
+            database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+def test_valid_file_opens_and_migrations_noop(tmp_path) -> None:
+    db_path = tmp_path / "nestquest.db"
+    _valid_database(db_path)
+    digest_before = hashlib.sha256(db_path.read_bytes()).hexdigest()
+    database = _run(_async_open_database(_make_hass_mock(), db_path))
+    try:
+        row = _run(
+            database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+    # Opening a valid file must not rewrite it (the no-op path).
+    assert (
+        hashlib.sha256(db_path.read_bytes()).hexdigest() == digest_before
+    )
+
+
+def test_corrupt_file_raises_not_ready_and_leaves_file_untouched(
+    tmp_path, caplog
+) -> None:
+    db_path = tmp_path / "nestquest.db"
+    _truncated_database(db_path)
+    corrupt_bytes = db_path.read_bytes()
+    digest = hashlib.sha256(corrupt_bytes).hexdigest()
+
+    with pytest.raises(ConfigEntryNotReady) as excinfo:
+        _run(_async_open_database(_make_hass_mock(), db_path))
+
+    # The file is byte-identical: the owner can restore/inspect it.
+    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
+    assert len(db_path.read_bytes()) == len(corrupt_bytes)
+    # The original error is preserved and the log names the path.
+    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
+    assert any(
+        "corrupt" in record.getMessage().lower()
+        and str(db_path) in record.getMessage()
+        for record in caplog.records
+        if record.levelname == "ERROR"
+    ), "the error log must name the corrupt path"
+
+
+def test_corrupt_file_raises_not_ready_never_raw_error(tmp_path) -> None:
+    """The corrupt case must surface as ConfigEntryNotReady, not crash
+    setup with a raw sqlite3.DatabaseError — and it must ALWAYS raise
+    (returning normally would mean corruption was silently accepted)."""
+    db_path = tmp_path / "nestquest.db"
+    _truncated_database(db_path)
+    with pytest.raises(ConfigEntryNotReady) as excinfo:
+        _run(_async_open_database(_make_hass_mock(), db_path))
+    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
+
+
+def test_migration_stage_corruption_leaves_db_and_sidecars_untouched(
+    tmp_path, monkeypatch
+) -> None:
+    """A valid header with a corrupt LATER page must be rejected by the
+    whole-file read-only preflight BEFORE the read-write open can
+    switch the journal mode to WAL.
+
+    The fixture is deliberately a ROLLBACK-JOURNAL file (created with
+    plain sqlite3, default journal_mode=delete, NOT via the wrapper):
+    that is the case where the WAL transition actually rewrites the
+    header — a WAL-mode fixture would mask the write under test.  The
+    open() spy proves the read-write open was never reached.
+    """
+    db_path = tmp_path / "nestquest.db"
+    conn = sqlite3.connect(db_path)
+    try:
+        conn.execute("PRAGMA journal_mode=DELETE")
+        conn.execute(
+            "CREATE TABLE filler (id INTEGER PRIMARY KEY, "
+            "payload TEXT NOT NULL)"
+        )
+        conn.executemany(
+            "INSERT INTO filler (payload) VALUES (?)",
+            [(f"row-{index}" * 8,) for index in range(200)],
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    data = bytearray(db_path.read_bytes())
+    assert conn.execute is not None
+    assert len(data) > 4200, "fixture must span more than one page"
+    # Corrupt a page well past the 100-byte header (offset 4096, the
+    # second page region): the header itself stays valid, so the
+    # read-write open succeeds and only the integrity probe/migrations
+    # discover damage.
+    for offset in range(4096, 4128):
+        data[offset] = 0xFF
+    db_path.write_bytes(bytes(data))
+    corrupt_bytes = db_path.read_bytes()
+    digest = hashlib.sha256(corrupt_bytes).hexdigest()
+
+    # Track whether the read-write open is reached at all: the whole
+    # preflight must reject this corruption BEFORE NestQuestDatabase
+    # .open() is ever called, which is the only way the file can stay
+    # byte-identical (the open persists WAL, rewriting the header).
+    open_calls: list[str] = []
+    original_open = NestQuestDatabase.open
+
+    async def _spy_open(self, path):
+        open_calls.append(str(path))
+        return await original_open(self, path)
+
+    monkeypatch.setattr(NestQuestDatabase, "open", _spy_open)
+
+    with pytest.raises(ConfigEntryNotReady) as excinfo:
+        _run(_async_open_database(_make_hass_mock(), db_path))
+    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
+
+    # The corruption was caught at the preflight: the read-write open
+    # (and with it the WAL header rewrite) was never attempted.
+    assert open_calls == [], (
+        "whole-file preflight must reject corruption before the "
+        "read-write open can transition the journal mode"
+    )
+
+    # The main file is byte-identical to the corrupt state we handed in.
+    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
+    # No WAL/SHM sidecars left behind (WAL mode was never persisted,
+    # or they were cleaned up on close).
+    assert not (tmp_path / "nestquest.db-wal").exists()
+    assert not (tmp_path / "nestquest.db-shm").exists()
+
+
+def test_wal_mode_corruption_leaves_file_and_sidecars_untouched(
+    tmp_path, monkeypatch
+) -> None:
+    """A corrupt WAL-HEADER file must be rejected without creating
+    -shm sidecars from the read-only preflight itself.
+
+    The immutable=1 probe URI is what guarantees a read-only
+    connection creates no sidecars even for a WAL-header database; the
+    sidecar assertions after the call prove it.
+    """
+    db_path = tmp_path / "nestquest.db"
+    # WAL fixture: created via the wrapper (journal_mode=WAL persists).
+    _valid_database(db_path)
+    data = bytearray(db_path.read_bytes())
+    # Corrupt the WAL header flag region: bytes 18/19 hold the
+    # read/write version numbers (2 = WAL).  Flip them to garbage so
+    # the file claims a broken journal mode.
+    assert len(data) > 64
+    data[18] = 0x99
+    data[19] = 0x99
+    db_path.write_bytes(bytes(data))
+    corrupt_bytes = db_path.read_bytes()
+    digest = hashlib.sha256(corrupt_bytes).hexdigest()
+
+    with pytest.raises(ConfigEntryNotReady):
+        _run(_async_open_database(_make_hass_mock(), db_path))
+
+    # Main file byte-identical; no -shm appeared from the probe.
+    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
+    assert not (tmp_path / "nestquest.db-shm").exists()
+
+
+def test_open_failure_closes_connection_when_opened(
+    tmp_path, monkeypatch
+) -> None:
+    """A failure DURING MIGRATIONS closes the opened connection; a
+    failure inside open() itself has nothing to close (the wrapper
+    never adopted a connection), so close must not be called there.
+    """
+    db_path = tmp_path / "nestquest.db"
+    _truncated_database(db_path)
+
+    close_calls: list[bool] = []
+    original_close = NestQuestDatabase.close
+
+    async def _spy_close(self):
+        close_calls.append(True)
+        return await original_close(self)
+
+    monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
+    with pytest.raises(ConfigEntryNotReady):
+        _run(_async_open_database(_make_hass_mock(), db_path))
+    # Corruption hit at PRAGMA time inside open(): no connection was
+    # adopted, so there is nothing to close and close() must NOT have
+    # been called (calling close on a never-opened wrapper would be
+    # harmless but the helper deliberately skips it).
+    assert close_calls == []
+
+    # Contrast: a failure DURING migrations (file opens fine) closes.
+    close_calls.clear()
+    valid_path = tmp_path / "migrations-fail.db"
+    _valid_database(valid_path)
+    # _valid_database's own close went through the spy; only the
+    # failure path's close counts from here.
+    close_calls.clear()
+    from custom_components.nestquest import migrations as migrations_mod
+    from custom_components.nestquest.db import NestQuestDatabase as _DB
+
+    async def _failing(database, migrations=None):
+        raise sqlite3.DatabaseError("database disk image is malformed")
+
+    monkeypatch.setattr(migrations_mod, "apply_migrations", _failing)
+    monkeypatch.setattr(
+        "custom_components.nestquest.apply_migrations", _failing
+    )
+    with pytest.raises(ConfigEntryNotReady) as excinfo:
+        _run(_async_open_database(_make_hass_mock(), valid_path))
+    # Exactly one close: the migrations-failure cleanup path.  (The
+    # spy also sees the wrapper's own idempotent close if it were
+    # double-called — it must not be.)
+    assert close_calls == [True], (
+        "an open connection must be closed before the error propagates"
+    )
+    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
+
+
+# ---------------------------------------------------------------------------
+# Startup through async_setup_entry (the real wiring)
+# ---------------------------------------------------------------------------
+
+
+def _make_setup(hass, registry):
+    def _setup(entry):
+        return async_setup_entry(hass, entry)
+
+    return _setup
+
+
+async def _setup_entry(hass, entry, registry) -> object:
+    entry = _wire(entry, registry)
+    await async_setup_entry(hass, entry)
+    return entry
+
+
+async def test_setup_with_corrupt_file_raises_not_ready(
+    hass, make_entry, monkeypatch, tmp_path, caplog
+) -> None:
+    from custom_components.nestquest.store import async_get_db_path
+
+    db_path = await async_get_db_path(hass)
+    await _avalid_database(db_path)
+    data = db_path.read_bytes()
+    assert len(data) > 64
+    db_path.write_bytes(data[: len(data) // 2])
+    corrupt_bytes = db_path.read_bytes()
+
+    entry = make_entry()
+    with pytest.raises(ConfigEntryNotReady):
+        await _setup_entry(hass, entry, hass.registry)
+    # File untouched, no runtime data stored.
+    assert db_path.read_bytes() == corrupt_bytes
+    assert hass.data["nestquest"].get(entry.entry_id) is None
+
+
+async def test_setup_with_missing_file_creates_and_loads(
+    hass, make_entry
+) -> None:
+    entry = await _setup_entry(hass, make_entry(), hass.registry)
+    database = entry.runtime_data.database
+    try:
+        row = await database.fetch_one(
+            f"SELECT version FROM {VERSION_TABLE}"
+        )
+        assert row == (1,)
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_with_valid_file_preserves_rows(
+    hass, make_entry
+) -> None:
+    entry_a = await _setup_entry(hass, make_entry(), hass.registry)
+    database_a = entry_a.runtime_data.database
+    await database_a.execute(
+        "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+        ("Ada", "2026-09-14T12:00:00+00:00"),
+    )
+    # Unload pops the runtime record AND closes the connection; a
+    # fresh setup for the same entry_id then reopens the file.
+    assert await async_unload_entry(hass, entry_a) is True
+
+    entry_b = await _setup_entry(hass, make_entry(), hass.registry)
+    database_b = entry_b.runtime_data.database
+    assert database_b is not database_a
+    try:
+        row = await database_b.fetch_one("SELECT COUNT(*) FROM children")
+        assert row == (1,)
+    finally:
+        await database_b.close()
+    assert await async_unload_entry(hass, entry_b) is True
+
+
+# ---------------------------------------------------------------------------
+# No code path deletes or overwrites an existing database file
+# ---------------------------------------------------------------------------
+
+
+def test_package_never_unlinks_or_truncates_db_files() -> None:
+    """Guardrail: no code path may delete or overwrite an existing
+    database file.  Scans the package for unlink/rmtree/write on the
+    DB filename constant or *.db patterns.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+    pattern = re.compile(
+        r"(unlink|rmtree|write_bytes|write_text|open\s*\(\s*[^)]*['\"]w)"
+        r"[^\n]*(nestquest\.db|\.db['\"])|"
+        r"(nestquest\.db|\.db['\"])[^\n]*"
+        r"(unlink|rmtree|write_bytes|write_text)",
+        re.IGNORECASE,
+    )
+    offenders: list[str] = []
+    for py in sorted(package.glob("*.py")):
+        if pattern.search(py.read_text()):
+            offenders.append(py.name)
+    assert offenders == [], (
+        f"code paths that could delete/overwrite the database: {offenders}"
+    )

tests/test_store.py

@@ -101,12 +101,20 @@ def test_executor_helper_creates_directory(tmp_path) -> None:
 
 
 def test_no_sqlite3_import_outside_db_module() -> None:
-    """sqlite3 is banned package-wide except db.py, the connection wrapper."""
+    """sqlite3 is banned package-wide except two justified modules.
+
+    db.py is the connection wrapper (all statement execution lives
+    there).  __init__.py imports sqlite3 ONLY for its exception types
+    to classify corruption on startup (done-condition: corrupt file ->
+    ConfigEntryNotReady); it never opens a connection or executes
+    SQL — that is asserted separately below.
+    """
     pkg_dir = Path(store.__file__).parent
     py_files = list(pkg_dir.glob("*.py"))
     assert py_files
+    allowed = {"db.py", "__init__.py"}
     for py_file in py_files:
-        if py_file.name == "db.py":
+        if py_file.name in allowed:
             continue
         tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
         for node in ast.walk(tree):
@@ -121,6 +129,36 @@ def test_no_sqlite3_import_outside_db_module() -> None:
                 )
 
 
+def test_init_module_never_opens_or_executes_sqlite() -> None:
+    """The __init__.py sqlite3 usage is limited to (a) exception typing
+    for corruption classification and (b) the ONE read-only preflight
+    connection (mode=ro URI, SELECT-only) that detects corruption in an
+    existing file before the real read-write open.  No read-write
+    connect, no cursor writes, no non-SELECT statements may appear.
+    """
+    import re
+    from pathlib import Path
+
+    init_path = (
+        Path(store.__file__).parent / "__init__.py"
+    ).read_text(encoding="utf-8")
+    # The preflight connect must be the URI read-only form, and only
+    # SELECT statements and the integrity_check PRAGMA may run there.
+    assert "sqlite3.connect(uri, uri=True)" in init_path
+    assert "?mode=ro" in init_path
+    assert 'PRAGMA integrity_check' in init_path
+    forbidden = re.compile(
+        r"sqlite3\s*\.\s*connect(?!\(\s*uri\s*,)"
+        r"|\bconn(ection)?\s*\.\s*execute\s*\(\s*[\"'](?!SELECT\s|PRAGMA\s+integrity_check)"
+        r"|\bexecutemany\b|\bexecutescript\b",
+        re.IGNORECASE,
+    )
+    assert forbidden.search(init_path) is None, (
+        "__init__.py may use sqlite3 only for exception typing and the "
+        "read-only corruption preflight (SELECT + integrity_check)"
+    )
+
+
 def test_no_file_creation_calls_in_store() -> None:
     source = Path(store.__file__).read_text(encoding="utf-8")
     tree = ast.parse(source, filename=str(store.__file__))