docs: correct gate comment and get_by_child cross-reference

transaction() submits and awaits BEGIN before the DAO's first fetch
runs; gating that fetch proves BEGIN has executed - SQLite is actually
inside the open transaction - when task B is scheduled and queues.

custom_components/nestquest/dao_presence.py

@@ -0,0 +1,247 @@
+"""Typed DAO layer for the presence_schedules and presence_overrides tables.
+
+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.
+
+Semantics per the done-condition and Feature 05:
+
+- A child has AT MOST ONE presence schedule (schema UNIQUE); upserting
+  by child updates the existing row rather than raising.
+- A child with NO schedule row is present every day — the absence of a
+  row is meaningful state, expressed by :meth:`PresenceSchedulesDao.get_by_child`
+  returning None, never by an empty pattern row.
+- Overrides beat the pattern for their date range (that resolution is
+  the presence engine's job, Feature 05); the DAO only stores and
+  lists them.  A single-day override stores the same date as both
+  start and end (schema CHECK requires end >= start).
+
+Date arguments are validated as strict YYYY-MM-DD per the same policy
+the rules DAO applies: the schema's lexical CHECKs cannot prove
+calendar validity.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .dao_children import _connection_lock
+from .dao_rules import _UNSET, _validate_date
+from .db import NestQuestDatabase
+
+
+@dataclass(frozen=True)
+class PresenceScheduleRecord:
+    """One row of ``presence_schedules``."""
+
+    id: int
+    child_id: int
+    cycle_length_weeks: int
+    anchor_date: str
+    pattern: str
+
+
+@dataclass(frozen=True)
+class PresenceOverrideRecord:
+    """One row of ``presence_overrides``."""
+
+    id: int
+    child_id: int
+    start_date: str
+    end_date: str
+    is_present: bool
+    note: str | None
+
+
+_SCHEDULE_COLUMNS = (
+    "id, child_id, cycle_length_weeks, anchor_date, pattern"
+)
+_OVERRIDE_COLUMNS = (
+    "id, child_id, start_date, end_date, is_present, note"
+)
+
+
+def _schedule_from_row(row: tuple) -> PresenceScheduleRecord:
+    return PresenceScheduleRecord(
+        id=row[0],
+        child_id=row[1],
+        cycle_length_weeks=row[2],
+        anchor_date=row[3],
+        pattern=row[4],
+    )
+
+
+def _override_from_row(row: tuple) -> PresenceOverrideRecord:
+    return PresenceOverrideRecord(
+        id=row[0],
+        child_id=row[1],
+        start_date=row[2],
+        end_date=row[3],
+        is_present=bool(row[4]),
+        note=row[5],
+    )
+
+
+class PresenceSchedulesDao:
+    """Typed async access to the ``presence_schedules`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def upsert_by_child(
+        self,
+        child_id: int,
+        cycle_length_weeks: int,
+        anchor_date: str,
+        pattern: str,
+    ) -> PresenceScheduleRecord:
+        """Set the child's presence schedule, replacing any existing one.
+
+        Upsert semantics are the done-condition: a second schedule for
+        the same child updates the existing row rather than raising.
+        Runs under the connection lock so a concurrent upsert/delete
+        for the same child serializes instead of colliding on the
+        schema's UNIQUE(child_id).  Pattern validity (segment shape,
+        segment count = cycle_length_weeks) is enforced by the schema
+        CHECKs; the child must exist (FK) — validated atomically here
+        so a deleted child cannot pass a stale check.
+        """
+        _validate_date(anchor_date, "anchor_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                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")
+                await self._database.execute(
+                    "INSERT INTO presence_schedules (child_id, "
+                    "cycle_length_weeks, anchor_date, pattern) "
+                    "VALUES (?, ?, ?, ?) "
+                    "ON CONFLICT (child_id) DO UPDATE SET "
+                    "cycle_length_weeks = excluded.cycle_length_weeks, "
+                    "anchor_date = excluded.anchor_date, "
+                    "pattern = excluded.pattern",
+                    (child_id, cycle_length_weeks, anchor_date, pattern),
+                )
+                schedule = await self.get_by_child(child_id)
+        assert schedule is not None
+        return schedule
+
+    async def get_by_child(
+        self, child_id: int
+    ) -> PresenceScheduleRecord | None:
+        """Return the child's schedule, or None when the child has none.
+
+        None is meaningful and UNAMBIGUOUS here: the child exists and
+        has no schedule row, i.e. the child is present every day
+        (Feature 05 guardrail).  An unknown child id raises ValueError
+        instead of returning None, because "nonexistent child" must
+        never be silently read as "present every day".
+        """
+        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")
+        row = await self._database.fetch_one(
+            f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules "
+            "WHERE child_id = ?",
+            (child_id,),
+        )
+        return _schedule_from_row(row) if row is not None else None
+
+    async def delete(self, child_id: int) -> bool:
+        """Remove the child's schedule; True if a row was removed.
+
+        After deletion the child is present every day (no row means
+        no restriction).  Deleting a schedule the child never had
+        returns False.
+        """
+        result = await self._database.execute(
+            "DELETE FROM presence_schedules WHERE child_id = ?",
+            (child_id,),
+        )
+        return result.rowcount > 0
+
+
+class PresenceOverridesDao:
+    """Typed async access to the ``presence_overrides`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def create(
+        self,
+        child_id: int,
+        start_date: str,
+        end_date: str,
+        is_present: bool,
+        *,
+        note: str | None = None,
+    ) -> PresenceOverrideRecord:
+        """Insert one override and return the record as stored.
+
+        The child must exist (validated atomically under the
+        connection lock).  Date shape is validated strictly, and the
+        schema CHECK rejects end < start.
+        """
+        _validate_date(start_date, "start_date")
+        _validate_date(end_date, "end_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                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 presence_overrides (child_id, "
+                    "start_date, end_date, is_present, note) "
+                    "VALUES (?, ?, ?, ?, ?)",
+                    (child_id, start_date, end_date, int(is_present), note),
+                )
+                override = await self._database.fetch_one(
+                    f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides "
+                    "WHERE id = ?",
+                    (result.lastrowid,),
+                )
+        assert override is not None
+        return _override_from_row(override)
+
+    async def list_by_child_and_range(
+        self,
+        child_id: int,
+        range_start: str,
+        range_end: str,
+    ) -> list[PresenceOverrideRecord]:
+        """Return the child's overrides overlapping [range_start, range_end].
+
+        An override overlaps the range when it ends on or after the
+        range starts and starts on or before the range ends (inclusive
+        boundary dates, matching the storage model where overrides are
+        closed date ranges).  Ordered by start date.
+        """
+        _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 {_OVERRIDE_COLUMNS} FROM presence_overrides "
+            "WHERE child_id = ? AND end_date >= ? AND start_date <= ? "
+            "ORDER BY start_date",
+            (child_id, range_start, range_end),
+        )
+        return [_override_from_row(row) for row in rows]
+
+    async def delete(self, override_id: int) -> bool:
+        """Remove one override by id; True if a row was removed."""
+        result = await self._database.execute(
+            "DELETE FROM presence_overrides WHERE id = ?",
+            (override_id,),
+        )
+        return result.rowcount > 0

custom_components/nestquest/dao_rules.py

@@ -0,0 +1,465 @@
+"""Typed DAO layer for the schedule_rules and task_definitions tables.
+
+All SQL for ``schedule_rules`` and ``task_definitions`` 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.
+
+Guardrail mapping:
+
+- ``schedule_rules`` deletion is delete-if-unreferenced: a rule still
+  referenced by any task definition is rejected (the done-condition),
+  not cascaded.
+- ``task_definitions`` has no delete path at all: definitions are
+  deactivated, never hard-deleted, and history must survive.
+- ``set_assignee`` changes future instances only — the DAO writes the
+  definition row and nothing else; reassignment never rewrites
+  existing instances or completion history (Feature 06 guardrail).
+- Validation of rule well-formedness and child activeness lives partly
+  in the schema CHECKs (rule-type coherence) and partly in
+  :meth:`TaskDefinitionsDao.create` (active child, rule existence);
+  no permission checks here, that is the Feature 09 gate's job.
+
+Concurrency: ``delete_rule_if_unreferenced`` performs its reference
+check and DELETE inside one transaction under the connection-scoped
+lock shared with :mod:`.dao_children`, so a definition referencing the
+rule cannot appear between check and delete on this connection.
+"""
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+
+from .dao_children import _connection_lock
+from .db import NestQuestDatabase
+
+#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
+#: in update methods: passing ``None`` must mean clearing a nullable
+#: column, not skipping it.
+_UNSET = object()
+
+#: Date-shape policy: schedule-rule dates are ISO YYYY-MM-DD.  The
+#: schema's lexical end>=start CHECK cannot prove a value is a real
+#: calendar date, so the DAO validates the shape it accepts.
+_DATE_FORMAT = "%Y-%m-%d"
+
+
+def _validate_date(value: str, field: str) -> None:
+    """Raise ValueError unless ``value`` is a YYYY-MM-DD calendar date.
+
+    ``datetime.date.fromisoformat`` accepts ISO dates but is lenient on
+    some builds (e.g. '2026-9-4'); strptime with an exact format is
+    strict, and strptime round-trips the parsed value to reject zero
+    padding ('2026-9-4' parses but its isoformat differs).
+    """
+    try:
+        parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date()
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"{field} must be an ISO date (YYYY-MM-DD), got {value!r}"
+        ) from None
+    if parsed.isoformat() != value:
+        raise ValueError(
+            f"{field} must be a strict YYYY-MM-DD date, got {value!r}"
+        )
+
+
+@dataclass(frozen=True)
+class ScheduleRuleRecord:
+    """One row of ``schedule_rules``."""
+
+    id: int
+    rule_type: str
+    interval: int
+    weekday_set: str | None
+    day_of_month: int | None
+    nth_weekday: int | None
+    month: int | None
+    start_date: str
+    end_date: str | None
+
+
+@dataclass(frozen=True)
+class TaskDefinitionRecord:
+    """One row of ``task_definitions``."""
+
+    id: int
+    title: str
+    description: str | None
+    icon: str | None
+    child_id: int
+    schedule_rule_id: int
+    due_time: str | None
+    is_active: bool
+    created_at: str
+
+
+_RULE_COLUMNS = (
+    "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
+    "month, start_date, end_date"
+)
+_DEFINITION_COLUMNS = (
+    "id, title, description, icon, child_id, schedule_rule_id, due_time, "
+    "is_active, created_at"
+)
+
+
+def _rule_from_row(row: tuple) -> ScheduleRuleRecord:
+    return ScheduleRuleRecord(
+        id=row[0],
+        rule_type=row[1],
+        interval=row[2],
+        weekday_set=row[3],
+        day_of_month=row[4],
+        nth_weekday=row[5],
+        month=row[6],
+        start_date=row[7],
+        end_date=row[8],
+    )
+
+
+def _definition_from_row(row: tuple) -> TaskDefinitionRecord:
+    return TaskDefinitionRecord(
+        id=row[0],
+        title=row[1],
+        description=row[2],
+        icon=row[3],
+        child_id=row[4],
+        schedule_rule_id=row[5],
+        due_time=row[6],
+        is_active=bool(row[7]),
+        created_at=row[8],
+    )
+
+
+class ScheduleRulesDao:
+    """Typed async access to the ``schedule_rules`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def create(
+        self,
+        rule_type: str,
+        start_date: str,
+        *,
+        interval: int = 1,
+        weekday_set: str | None = None,
+        day_of_month: int | None = None,
+        nth_weekday: int | None = None,
+        month: int | None = None,
+        end_date: str | None = None,
+    ) -> ScheduleRuleRecord:
+        """Insert one rule and return the record as stored.
+
+        Rule-type coherence (a weekly rule needs ``weekday_set``, a
+        yearly rule needs ``month``, etc.) is enforced by the schema
+        CHECKs and surfaces as :class:`sqlite3.IntegrityError`.  Date
+        shape (strict YYYY-MM-DD) is validated here because the
+        schema's lexical end>=start CHECK cannot prove a real calendar
+        date.
+        """
+        _validate_date(start_date, "start_date")
+        if end_date is not None:
+            _validate_date(end_date, "end_date")
+        result = await self._database.execute(
+            "INSERT INTO schedule_rules (rule_type, interval, weekday_set, "
+            "day_of_month, nth_weekday, month, start_date, end_date) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                rule_type,
+                interval,
+                weekday_set,
+                day_of_month,
+                nth_weekday,
+                month,
+                start_date,
+                end_date,
+            ),
+        )
+        rule = await self.get(result.lastrowid)
+        assert rule is not None
+        return rule
+
+    async def get(self, rule_id: int) -> ScheduleRuleRecord | None:
+        """Return the rule with ``rule_id``, or None."""
+        row = await self._database.fetch_one(
+            f"SELECT {_RULE_COLUMNS} FROM schedule_rules WHERE id = ?",
+            (rule_id,),
+        )
+        return _rule_from_row(row) if row is not None else None
+
+    async def update(
+        self,
+        rule_id: int,
+        *,
+        rule_type: str | None | object = _UNSET,
+        interval: int | None | object = _UNSET,
+        weekday_set: str | None | object = _UNSET,
+        day_of_month: int | None | object = _UNSET,
+        nth_weekday: int | None | object = _UNSET,
+        month: int | None | object = _UNSET,
+        start_date: str | None | object = _UNSET,
+        end_date: str | None | object = _UNSET,
+    ) -> int:
+        """Update the given fields; returns rows updated (0 if absent).
+
+        Arguments default to the module sentinel ``_UNSET`` meaning
+        "leave this column alone"; passing ``None`` explicitly writes
+        SQL NULL, so callers can reopen an ended rule (``end_date``
+        back to NULL) or clear nullable shape fields.  Passing
+        ``None``/a new value for a date field validates its shape.
+        Editing a rule that a definition references changes future
+        materializations only — this method touches no instances
+        (Feature 06/07 separation).
+        """
+        date_args = (
+            ("start_date", start_date),
+            ("end_date", end_date),
+        )
+        for field, value in date_args:
+            if value is not _UNSET and value is not None:
+                _validate_date(value, field)
+        assignments: list[str] = []
+        parameters: list[object] = []
+        for column, value in (
+            ("rule_type", rule_type),
+            ("interval", interval),
+            ("weekday_set", weekday_set),
+            ("day_of_month", day_of_month),
+            ("nth_weekday", nth_weekday),
+            ("month", month),
+            ("start_date", start_date),
+            ("end_date", end_date),
+        ):
+            if value is not _UNSET:
+                assignments.append(f"{column} = ?")
+                parameters.append(value)
+        if not assignments:
+            return 0
+        parameters.append(rule_id)
+        result = await self._database.execute(
+            f"UPDATE schedule_rules SET {', '.join(assignments)} "
+            "WHERE id = ?",
+            tuple(parameters),
+        )
+        return result.rowcount
+
+    async def delete_if_unreferenced(self, rule_id: int) -> bool:
+        """Delete the rule only if no definition references it.
+
+        Returns True if the rule was deleted, False if it is still
+        referenced (or does not exist — distinguish via :meth:`get`).
+        The reference check and the DELETE run inside one transaction
+        under the connection lock, so a definition cannot start
+        referencing the rule between the check and the delete.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                row = await self._database.fetch_one(
+                    "SELECT 1 FROM task_definitions "
+                    "WHERE schedule_rule_id = ? LIMIT 1",
+                    (rule_id,),
+                )
+                if row is not None:
+                    return False
+                result = await self._database.execute(
+                    "DELETE FROM schedule_rules WHERE id = ?",
+                    (rule_id,),
+                )
+                return result.rowcount > 0
+
+
+class TaskDefinitionsDao:
+    """Typed async access to the ``task_definitions`` table.
+
+    No delete method exists: definitions are deactivated via
+    :meth:`set_active`, never hard-deleted, so completion history keeps
+    its references (feature guardrail).
+    """
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def create(
+        self,
+        title: str,
+        child_id: int,
+        schedule_rule_id: int,
+        created_at: str,
+        *,
+        description: str | None = None,
+        icon: str | None = None,
+        due_time: str | None = None,
+        is_active: bool = True,
+    ) -> TaskDefinitionRecord:
+        """Insert one definition and return the record as stored.
+
+        Validates the assignment target as part of the same serialized
+        transaction as the insert: the child must exist and be active,
+        and the rule must exist, AT INSERT TIME — a child deactivated
+        or a rule deleted concurrently cannot slip between the checks
+        and the INSERT, which would otherwise create an invalid
+        assignment or surface as a raw FK IntegrityError instead of
+        the documented ValueError.  An INACTIVE child would pass the
+        schema foreign keys, which is why the activeness check lives
+        here.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                await self._validate_assignable(
+                    child_id, schedule_rule_id
+                )
+                result = await self._database.execute(
+                    "INSERT INTO task_definitions (title, description, "
+                    "icon, child_id, schedule_rule_id, due_time, "
+                    "is_active, created_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        title,
+                        description,
+                        icon,
+                        child_id,
+                        schedule_rule_id,
+                        due_time,
+                        int(is_active),
+                        created_at,
+                    ),
+                )
+                definition = await self.get(result.lastrowid)
+        assert definition is not None
+        return definition
+
+    async def _validate_assignable(
+        self, child_id: int, schedule_rule_id: int
+    ) -> None:
+        """Raise ValueError unless child (active) and rule both exist.
+
+        MUST be called inside the connection lock so the verdict cannot
+        go stale before the caller's INSERT commits.
+        """
+        await self._validate_assignable_child(child_id)
+        rule = await self._database.fetch_one(
+            "SELECT 1 FROM schedule_rules WHERE id = ?",
+            (schedule_rule_id,),
+        )
+        if rule is None:
+            raise ValueError(
+                f"schedule rule {schedule_rule_id} does not exist"
+            )
+
+    async def get(self, definition_id: int) -> TaskDefinitionRecord | None:
+        """Return the definition with ``definition_id``, or None."""
+        row = await self._database.fetch_one(
+            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            "WHERE id = ?",
+            (definition_id,),
+        )
+        return _definition_from_row(row) if row is not None else None
+
+    async def list_by_child(self, child_id: int) -> list[TaskDefinitionRecord]:
+        """Return all definitions assigned to ``child_id``, newest first."""
+        rows = await self._database.fetch_all(
+            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            "WHERE child_id = ? ORDER BY id DESC",
+            (child_id,),
+        )
+        return [_definition_from_row(row) for row in rows]
+
+    async def list_active(self) -> list[TaskDefinitionRecord]:
+        """Return all active definitions, oldest first (stable order)."""
+        rows = await self._database.fetch_all(
+            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            "WHERE is_active = 1 ORDER BY id"
+        )
+        return [_definition_from_row(row) for row in rows]
+
+    async def update(
+        self,
+        definition_id: int,
+        *,
+        title: str | None | object = _UNSET,
+        description: str | None | object = _UNSET,
+        icon: str | None | object = _UNSET,
+        due_time: str | None | object = _UNSET,
+    ) -> int:
+        """Update the given fields; returns rows updated (0 if absent).
+
+        Arguments default to the module sentinel ``_UNSET`` meaning
+        "leave this column alone"; passing ``None`` explicitly writes
+        SQL NULL, so callers can remove optional metadata
+        (description, icon, due_time).  Assignment and activation are
+        deliberately NOT settable here: they have their own explicit
+        operations (:meth:`set_assignee`, :meth:`set_active`) so they
+        stay separately loggable and permission-gateable.
+        """
+        assignments: list[str] = []
+        parameters: list[object] = []
+        for column, value in (
+            ("title", title),
+            ("description", description),
+            ("icon", icon),
+            ("due_time", due_time),
+        ):
+            if value is not _UNSET:
+                assignments.append(f"{column} = ?")
+                parameters.append(value)
+        if not assignments:
+            return 0
+        parameters.append(definition_id)
+        result = await self._database.execute(
+            f"UPDATE task_definitions SET {', '.join(assignments)} "
+            "WHERE id = ?",
+            tuple(parameters),
+        )
+        return result.rowcount
+
+    async def set_active(self, definition_id: int, is_active: bool) -> int:
+        """Deactivate/reactivate; returns rows updated (0 if absent).
+
+        Deactivating stops future instance generation but leaves
+        existing instances and completion history untouched.
+        """
+        result = await self._database.execute(
+            "UPDATE task_definitions SET is_active = ? WHERE id = ?",
+            (int(is_active), definition_id),
+        )
+        return result.rowcount
+
+    async def set_assignee(
+        self, definition_id: int, child_id: int
+    ) -> int:
+        """Reassign the definition to ``child_id``; rows updated (0 if absent).
+
+        Validates the new assignee inside the same serialized
+        transaction as the UPDATE (existing and active AT ASSIGNMENT
+        TIME), so a child deactivated concurrently cannot become the
+        assignee.  Changes future instances only: already-generated
+        instances and completion history are never rewritten here.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                await self._validate_assignable_child(child_id)
+                result = await self._database.execute(
+                    "UPDATE task_definitions SET child_id = ? "
+                    "WHERE id = ?",
+                    (child_id, definition_id),
+                )
+        return result.rowcount
+
+    async def _validate_assignable_child(self, child_id: int) -> None:
+        """Raise ValueError unless the child exists and is active.
+
+        MUST be called inside the connection lock so the verdict cannot
+        go stale before the caller's UPDATE commits.
+        """
+        child = await self._database.fetch_one(
+            "SELECT is_active FROM children WHERE id = ?", (child_id,)
+        )
+        if child is None:
+            raise ValueError(f"child {child_id} does not exist")
+        if not child[0]:
+            raise ValueError(
+                f"child {child_id} is inactive and cannot be assigned"
+            )

tests/test_dao_children.py

@@ -478,11 +478,19 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
 
     allowed = {
         "custom_components/nestquest/dao_children.py",  # the DAO itself
+        "custom_components/nestquest/dao_rules.py",  # validates child
+        # activeness on assignment (FK to children, done-condition
+        # 'validates that the assigned child is active')
+        "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/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_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
     }
     sql_pattern = re.compile(
         r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+"

tests/test_dao_presence.py

@@ -0,0 +1,622 @@
+"""Tests for dao_presence.py: presence schedules and overrides DAO."""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+
+import pytest
+
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_presence import (
+    PresenceOverrideRecord,
+    PresenceOverridesDao,
+    PresenceScheduleRecord,
+    PresenceSchedulesDao,
+)
+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():
+    from unittest.mock import AsyncMock, MagicMock
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+NOW = "2026-09-14T12:00:00+00:00"
+
+
+async def _prepare(path) -> tuple:
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    schedules = PresenceSchedulesDao(database)
+    overrides = PresenceOverridesDao(database)
+    children = ChildrenDao(database)
+    child = await children.create("Ada", NOW)
+    return database, schedules, overrides, children, child
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            database, schedules, overrides, children, child = await _prepare(
+                tmp_path / name
+            )
+            try:
+                return await body(
+                    database, schedules, overrides, children, child
+                )
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# presence_schedules: upsert by child
+# ---------------------------------------------------------------------------
+
+
+def test_upsert_creates_then_updates_same_row(tmp_path) -> None:
+    """Second upsert for the same child UPDATES, not raises."""
+    async def _body(database, schedules, overrides, children, child):
+        first = await schedules.upsert_by_child(
+            child.id, 2, "2026-09-07", "0,2,4|1,3"
+        )
+        assert isinstance(first, PresenceScheduleRecord)
+        assert first.child_id == child.id
+        second = await schedules.upsert_by_child(
+            child.id, 2, "2026-09-14", "1,3|0,2,4"
+        )
+        assert second.id == first.id, "upsert must reuse the existing row"
+        fetched = await schedules.get_by_child(child.id)
+        assert fetched.anchor_date == "2026-09-14"
+        assert fetched.pattern == "1,3|0,2,4"
+        rows = await database.fetch_one(
+            "SELECT COUNT(*) FROM presence_schedules"
+        )
+        assert rows == (1,)
+        return second
+
+    _with_db(tmp_path, "upsert.db")(_body)
+
+
+def test_upsert_typed_record_fields(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        record = await schedules.upsert_by_child(
+            child.id, 4, "2026-09-07", "0|1|2|3"
+        )
+        assert record.id == 1
+        assert record.child_id == child.id
+        assert record.cycle_length_weeks == 4
+        assert record.anchor_date == "2026-09-07"
+        assert record.pattern == "0|1|2|3"
+        return record
+
+    _with_db(tmp_path, "upsert-fields.db")(_body)
+
+
+def test_upsert_unknown_child_raises_value_error(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await schedules.upsert_by_child(
+                999, 2, "2026-09-07", "0,2|1,3"
+            )
+        return None
+
+    _with_db(tmp_path, "upsert-bad-child.db")(_body)
+
+
+def test_upsert_malformed_anchor_date_raises(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        for bad in ("not-a-date", "2026-9-7", "2026-13-01", ""):
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await schedules.upsert_by_child(child.id, 2, bad, "0|1")
+        return None
+
+    _with_db(tmp_path, "upsert-bad-date.db")(_body)
+
+
+def test_upsert_malformed_pattern_rejected_by_schema(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            await schedules.upsert_by_child(
+                child.id, 2, "2026-09-07", "0,2"
+            )
+        return None
+
+    _with_db(tmp_path, "upsert-bad-pattern.db")(_body)
+
+
+def test_upsert_two_children_have_distinct_rows(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        other = await children.create("Bo", NOW)
+        await schedules.upsert_by_child(child.id, 2, "2026-09-07", "0,2|1,3")
+        await schedules.upsert_by_child(other.id, 2, "2026-09-07", "1,3|0,2")
+        mine = await schedules.get_by_child(child.id)
+        theirs = await schedules.get_by_child(other.id)
+        assert mine.id != theirs.id
+        assert mine.pattern == "0,2|1,3"
+        assert theirs.pattern == "1,3|0,2"
+        return mine
+
+    _with_db(tmp_path, "upsert-two-children.db")(_body)
+
+
+def test_upsert_concurrent_same_child_serializes(tmp_path) -> None:
+    """Racing upserts for one child both succeed; exactly one row.
+
+    Deterministic interleaving: a controllable executor gates task A's
+    job AFTER it has acquired the connection lock and begun its
+    transaction, so task B is created while A provably owns the
+    transaction and must queue on the lock.  Both upserts land, and
+    the table holds exactly one row (the ON CONFLICT upsert, not two
+    plain INSERTs, is what avoids a UNIQUE violation).
+    """
+    async def _main():
+        from unittest.mock import MagicMock
+
+        hass = MagicMock()
+        # Gate exactly ONE job: the first _fetch_one issued after
+        # arming, which is task A's child-existence SELECT inside its
+        # transaction.  The transaction's BEGIN runs first and is NOT
+        # gated, so when the gated fetch signals, BEGIN has executed
+        # and SQLite is inside the open transaction (A holds the
+        # connection lock from BEGIN until commit).  Everything before
+        # arming (open, migrations, child create)
+        # and everything after (B's jobs, A's remaining jobs, reads)
+        # pass straight through.
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+
+        async def _gated_executor(fn, *args):
+            # Gate the SECOND gated-window job: the first is the
+            # transaction's BEGIN (submitted and awaited by
+            # transaction() before upsert_by_child's child-SELECT
+            # runs).  Pausing the child-existence SELECT proves the
+            # BEGIN has EXECUTED — SQLite is actually inside the open
+            # transaction, with A holding the connection lock — when
+            # task B gets scheduled and must queue.
+            if armed["active"]:
+                name = getattr(fn, "__name__", "")
+                if name == "_fetch_one" and not armed["gated"]:
+                    armed["gated"] = True
+                    a_started.set()
+                    await gate_open.wait()
+            return fn(*args)
+
+        hass.async_add_executor_job = _gated_executor
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "upsert-race.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            schedules = PresenceSchedulesDao(database)
+            child = await children.create("Ada", NOW)
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                schedules.upsert_by_child(
+                    child.id, 2, "2026-09-07", "0,2|1,3"
+                )
+            )
+            # Wait until task A's child-existence SELECT (inside its
+            # open transaction, BEGIN already executed) is paused, then
+            # schedule task B so it must queue behind the open
+            # transaction.
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                schedules.upsert_by_child(
+                    child.id, 2, "2026-09-14", "1,3|0,2"
+                )
+            )
+            await asyncio.sleep(0)
+            # Release A; B's upsert must wait for A's commit and then
+            # land as the ON CONFLICT update path.
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+            assert record_a.id == record_b.id
+            fetched = await schedules.get_by_child(child.id)
+            assert fetched is not None
+            rows = await database.fetch_one(
+                "SELECT COUNT(*) FROM presence_schedules"
+            )
+            assert rows == (1,)
+            return fetched
+        finally:
+            await database.close()
+
+    fetched = _run(_main())
+    assert fetched.pattern in {"0,2|1,3", "1,3|0,2"}
+
+
+# ---------------------------------------------------------------------------
+# presence_schedules: get / delete
+# ---------------------------------------------------------------------------
+
+
+def test_get_by_child_missing_schedule_returns_none(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        return await schedules.get_by_child(child.id)
+
+    assert _with_db(tmp_path, "get-none.db")(_body) is None
+
+
+def test_get_by_child_unknown_child_raises(tmp_path) -> None:
+    """None must mean ONLY 'no schedule row': an unknown child id is a
+    caller bug, and silently reading a nonexistent child as 'present
+    every day' would be wrong — so get raises instead.
+    """
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await schedules.get_by_child(999)
+        return None
+
+    _with_db(tmp_path, "get-unknown.db")(_body)
+
+
+def test_delete_removes_and_returns_true_then_false(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        await schedules.upsert_by_child(child.id, 2, "2026-09-07", "0,2|1,3")
+        assert await schedules.delete(child.id) is True
+        assert await schedules.get_by_child(child.id) is None
+        assert await schedules.delete(child.id) is False
+        return None
+
+    _with_db(tmp_path, "delete.db")(_body)
+
+
+def test_delete_child_without_schedule_returns_false(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        return await schedules.delete(child.id)
+
+    assert _with_db(tmp_path, "delete-none.db")(_body) is False
+
+
+def test_delete_then_upsert_creates_fresh_row(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        first = await schedules.upsert_by_child(
+            child.id, 2, "2026-09-07", "0,2|1,3"
+        )
+        await schedules.delete(child.id)
+        second = await schedules.upsert_by_child(
+            child.id, 2, "2026-09-07", "0,2|1,3"
+        )
+        assert second.id != first.id
+        return second
+
+    _with_db(tmp_path, "delete-recreate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# presence_overrides: create / list / delete
+# ---------------------------------------------------------------------------
+
+
+def test_override_create_returns_typed_record(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        record = await overrides.create(
+            child.id, "2026-12-21", "2027-01-04", False,
+            note="Holiday abroad",
+        )
+        assert isinstance(record, PresenceOverrideRecord)
+        assert record.child_id == child.id
+        assert record.start_date == "2026-12-21"
+        assert record.end_date == "2027-01-04"
+        assert record.is_present is False
+        assert record.note == "Holiday abroad"
+        return record
+
+    _with_db(tmp_path, "override-create.db")(_body)
+
+
+def test_override_create_single_day_same_start_and_end(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        record = await overrides.create(
+            child.id, "2026-09-14", "2026-09-14", True
+        )
+        assert record.start_date == record.end_date
+        assert record.is_present is True
+        assert record.note is None
+        return record
+
+    _with_db(tmp_path, "override-single-day.db")(_body)
+
+
+def test_override_create_unknown_child_raises(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await overrides.create(
+                999, "2026-09-14", "2026-09-14", True
+            )
+        return None
+
+    _with_db(tmp_path, "override-bad-child.db")(_body)
+
+
+def test_override_create_malformed_dates_raise(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        for bad in ("not-a-date", "2026-9-14", ""):
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await overrides.create(child.id, bad, "2026-09-14", True)
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await overrides.create(child.id, "2026-09-14", bad, True)
+        return None
+
+    _with_db(tmp_path, "override-bad-dates.db")(_body)
+
+
+def test_override_end_before_start_rejected_by_schema(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            await overrides.create(
+                child.id, "2026-09-14", "2026-09-13", True
+            )
+        return None
+
+    _with_db(tmp_path, "override-end-before-start.db")(_body)
+
+
+def test_override_list_by_child_and_range_filters_correctly(
+    tmp_path,
+) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        await overrides.create(
+            child.id, "2026-09-01", "2026-09-05", False, note="before"
+        )
+        inside = await overrides.create(
+            child.id, "2026-09-10", "2026-09-12", False, note="inside"
+        )
+        overlapping = await overrides.create(
+            child.id, "2026-09-20", "2026-09-25", True, note="overlap"
+        )
+        after = await overrides.create(
+            child.id, "2026-10-01", "2026-10-02", False, note="after"
+        )
+        found = await overrides.list_by_child_and_range(
+            child.id, "2026-09-09", "2026-09-22"
+        )
+        assert [r.id for r in found] == [inside.id, overlapping.id]
+        # Inclusive boundaries: a range touching the override's edges
+        # still matches.
+        edge = await overrides.list_by_child_and_range(
+            child.id, "2026-09-20", "2026-09-20"
+        )
+        assert [r.id for r in edge] == [overlapping.id]
+        return found
+
+    _with_db(tmp_path, "override-list-range.db")(_body)
+
+
+def test_override_list_orders_by_start_date(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        late = await overrides.create(
+            child.id, "2026-09-20", "2026-09-21", True
+        )
+        early = await overrides.create(
+            child.id, "2026-09-10", "2026-09-11", False
+        )
+        found = await overrides.list_by_child_and_range(
+            child.id, "2026-09-01", "2026-09-30"
+        )
+        assert [r.id for r in found] == [early.id, late.id]
+        return found
+
+    _with_db(tmp_path, "override-list-order.db")(_body)
+
+
+def test_override_list_per_child_isolation(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        other = await children.create("Bo", NOW)
+        await overrides.create(child.id, "2026-09-10", "2026-09-11", True)
+        await overrides.create(other.id, "2026-09-10", "2026-09-11", False)
+        mine = await overrides.list_by_child_and_range(
+            child.id, "2026-09-01", "2026-09-30"
+        )
+        assert len(mine) == 1
+        assert mine[0].is_present is True
+        return mine
+
+    _with_db(tmp_path, "override-isolation.db")(_body)
+
+
+def test_override_list_malformed_range_raises(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        with pytest.raises(ValueError, match="YYYY-MM-DD"):
+            await overrides.list_by_child_and_range(
+                child.id, "oops", "2026-09-30"
+            )
+        with pytest.raises(ValueError, match="on or after"):
+            await overrides.list_by_child_and_range(
+                child.id, "2026-09-30", "2026-09-01"
+            )
+        return None
+
+    _with_db(tmp_path, "override-list-bad-range.db")(_body)
+
+
+def test_override_list_unknown_child_returns_empty(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        return await overrides.list_by_child_and_range(
+            999, "2026-09-01", "2026-09-30"
+        )
+
+    assert _with_db(tmp_path, "override-list-unknown.db")(_body) == []
+
+
+def test_override_delete_round_trip(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        record = await overrides.create(
+            child.id, "2026-09-10", "2026-09-11", True
+        )
+        assert await overrides.delete(record.id) is True
+        found = await overrides.list_by_child_and_range(
+            child.id, "2026-09-01", "2026-09-30"
+        )
+        assert found == []
+        assert await overrides.delete(record.id) is False
+        return None
+
+    _with_db(tmp_path, "override-delete.db")(_body)
+
+
+def test_override_delete_missing_returns_false(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        return await overrides.delete(999)
+
+    assert _with_db(tmp_path, "override-delete-missing.db")(_body) is False
+
+
+def test_override_delete_leaves_other_overrides(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        first = await overrides.create(
+            child.id, "2026-09-10", "2026-09-11", True
+        )
+        second = await overrides.create(
+            child.id, "2026-09-20", "2026-09-21", False
+        )
+        assert await overrides.delete(first.id) is True
+        found = await overrides.list_by_child_and_range(
+            child.id, "2026-09-01", "2026-09-30"
+        )
+        assert [r.id for r in found] == [second.id]
+        return found
+
+    _with_db(tmp_path, "override-delete-others.db")(_body)
+
+
+def test_schedule_and_override_independent_for_same_child(tmp_path) -> None:
+    """Deleting the schedule does not touch overrides and vice versa."""
+    async def _body(database, schedules, overrides, children, child):
+        await schedules.upsert_by_child(child.id, 2, "2026-09-07", "0,2|1,3")
+        override = await overrides.create(
+            child.id, "2026-09-10", "2026-09-11", True
+        )
+        assert await schedules.delete(child.id) is True
+        found = await overrides.list_by_child_and_range(
+            child.id, "2026-09-01", "2026-09-30"
+        )
+        assert [r.id for r in found] == [override.id]
+        return found
+
+    _with_db(tmp_path, "independent.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# No SQL for these tables outside the DAO module
+# ---------------------------------------------------------------------------
+
+
+def test_presence_sql_lives_only_in_dao_module() -> None:
+    """Guardrail: presence_schedules/presence_overrides SQL may appear
+    only in the DAO module (and the schema DDL declarations).
+    RECURSIVE scan of package and tests, FROM/INTO/UPDATE/DELETE FROM/
+    JOIN pattern, with exact repo-relative exclusions justified by
+    role: schema.py declares the DDL, migrations.py applies it, and
+    the test files for exactly those layers verify DDL, not data
+    paths.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+    repo_root = package.parent.parent
+    scan_roots = [package, repo_root / "tests"]
+
+    allowed = {
+        "custom_components/nestquest/dao_presence.py",  # this DAO
+        "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_presence.py",  # this file, scanned separately
+    }
+    sql_pattern = re.compile(
+        r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
+        r"(presence_schedules|presence_overrides)\b",
+        re.IGNORECASE,
+    )
+    offenders: list[str] = []
+    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()):
+                offenders.append(relative)
+    assert offenders == [], (
+        f"SQL touching presence_schedules/presence_overrides leaked "
+        f"into: {offenders}"
+    )
+
+
+def test_dao_presence_test_file_uses_dao_not_raw_table_sql() -> None:
+    """This test module must exercise the DAO, not raw SQL, for these
+    tables.  Exemptions are EXACT lines, not whole-function spans: the
+    two single-statement COUNT(*) row probes (the only honest way to
+    prove 'exactly one row exists' without a listing method) and the
+    guard functions' own regex spans.  Any other raw SQL added anywhere
+    in this file fails the guard.
+    """
+    import ast
+    import re
+    from pathlib import Path
+
+    path = Path(__file__)
+    text = path.read_text()
+    lines = text.splitlines(keepends=True)
+    tree = ast.parse(text)
+    guard_names = {
+        "test_presence_sql_lives_only_in_dao_module",
+        "test_dao_presence_test_file_uses_dao_not_raw_table_sql",
+    }
+    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)
+            )
+            continue
+        # Narrow probe exemption: only the string-literal line holding
+        # the sanctioned COUNT(*) query, not the enclosing function.
+        for sub in ast.walk(node):
+            if (
+                isinstance(sub, ast.Constant)
+                and isinstance(sub.value, str)
+                and sub.value.strip()
+                == "SELECT COUNT(*) FROM presence_schedules"
+                and sub.lineno is not None
+            ):
+                excluded.add(sub.lineno)
+
+    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+[`'\"]*(\[)?(presence_schedules|presence_overrides)"
+        r"\b",
+        re.IGNORECASE,
+    )
+    assert sql_pattern.search(remaining) is None, (
+        "tests must go through the DAO, not raw SQL, for these tables"
+    )

tests/test_dao_rules.py

@@ -0,0 +1,862 @@
+"""Tests for dao_rules.py: typed DAO for schedule_rules and task_definitions."""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+
+import pytest
+
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_rules import (
+    ScheduleRuleRecord,
+    ScheduleRulesDao,
+    TaskDefinitionRecord,
+    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():
+    from unittest.mock import AsyncMock, MagicMock
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+NOW = "2026-09-14T12:00:00+00:00"
+
+
+async def _prepare(path) -> tuple:
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    rules = ScheduleRulesDao(database)
+    definitions = TaskDefinitionsDao(database)
+    children = ChildrenDao(database)
+    child = await children.create("Ada", NOW)
+    return database, rules, definitions, children, child
+
+
+def _with_db(tmp_path, name):
+    """Run an async body against a prepared database, closing it after."""
+
+    def _run_test(body):
+        async def _main():
+            database, rules, definitions, children, child = await _prepare(
+                tmp_path / name
+            )
+            try:
+                return await body(database, rules, definitions, children, child)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# schedule_rules: create / get
+# ---------------------------------------------------------------------------
+
+
+def test_rule_create_returns_typed_record(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create(
+            "weekly", "2026-09-14", weekday_set="0,2,4", interval=2
+        )
+        assert isinstance(rule, ScheduleRuleRecord)
+        assert rule.id == 1
+        assert rule.rule_type == "weekly"
+        assert rule.interval == 2
+        assert rule.weekday_set == "0,2,4"
+        assert rule.day_of_month is None
+        assert rule.nth_weekday is None
+        assert rule.month is None
+        assert rule.start_date == "2026-09-14"
+        assert rule.end_date is None
+        return rule
+
+    _with_db(tmp_path, "rule-create.db")(_body)
+
+
+def test_rule_create_monthly_with_day_of_month(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create(
+            "monthly", "2026-09-14", day_of_month=15, weekday_set=None
+        )
+        assert rule.day_of_month == 15
+        assert rule.weekday_set is None
+        return rule
+
+    _with_db(tmp_path, "rule-monthly.db")(_body)
+
+
+def test_rule_get_missing_returns_none(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await rules.get(999)
+
+    assert _with_db(tmp_path, "rule-get-missing.db")(_body) is None
+
+
+def test_rule_create_coherence_enforced_by_schema(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            await rules.create("weekly", "2026-09-14", weekday_set=None)
+        return None
+
+    _with_db(tmp_path, "rule-coherence.db")(_body)
+
+
+def test_rule_create_rejects_malformed_dates(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        for bad in (
+            "not-a-date",
+            "2026-9-4",
+            "2026-13-01",
+            "2026-02-30",
+            "09-14",
+            "",
+        ):
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await rules.create("daily", bad)
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await rules.create("daily", "2026-09-14", end_date=bad)
+        return None
+
+    _with_db(tmp_path, "rule-create-dates.db")(_body)
+
+
+def test_rule_create_end_before_start_rejected(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            await rules.create(
+                "daily", "2026-09-14", end_date="2026-09-13"
+            )
+        return None
+
+    _with_db(tmp_path, "rule-dates.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# schedule_rules: update
+# ---------------------------------------------------------------------------
+
+
+def test_rule_update_fields_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("weekly", "2026-09-14", weekday_set="0,2")
+        updated = await rules.update(
+            rule.id, weekday_set="1,3", interval=3, end_date="2026-12-31"
+        )
+        assert updated == 1
+        fetched = await rules.get(rule.id)
+        assert fetched.weekday_set == "1,3"
+        assert fetched.interval == 3
+        assert fetched.end_date == "2026-12-31"
+        assert fetched.rule_type == "weekly"
+        return fetched
+
+    _with_db(tmp_path, "rule-update.db")(_body)
+
+
+def test_rule_update_omitted_fields_untouched(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("weekly", "2026-09-14", weekday_set="0,2")
+        await rules.update(rule.id, interval=2)
+        fetched = await rules.get(rule.id)
+        assert fetched.weekday_set == "0,2"
+        assert fetched.start_date == "2026-09-14"
+        return fetched
+
+    _with_db(tmp_path, "rule-update-partial.db")(_body)
+
+
+def test_rule_update_no_fields_returns_zero(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        return await rules.update(rule.id)
+
+    assert _with_db(tmp_path, "rule-update-empty.db")(_body) == 0
+
+
+def test_rule_update_missing_rule_returns_zero(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await rules.update(999, interval=2)
+
+    assert _with_db(tmp_path, "rule-update-missing.db")(_body) == 0
+
+
+def test_rule_update_bad_type_rejected_by_check(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            await rules.update(rule.id, rule_type="sometimes")
+        return None
+
+    _with_db(tmp_path, "rule-update-bad-type.db")(_body)
+
+
+def test_rule_update_can_clear_nullable_fields(tmp_path) -> None:
+    """Explicit None clears a nullable column; omitted leaves it alone.
+
+    A caller must be able to reopen an ended rule (end_date back to
+    NULL) and clear rule-shape fields; the _UNSET sentinel keeps
+    'omitted' distinct from 'write NULL'.
+    """
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create(
+            "weekly",
+            "2026-09-14",
+            weekday_set="0,2",
+            end_date="2026-12-31",
+            month=3,
+        )
+        # Omitted: nothing changes.
+        await rules.update(rule.id, interval=2)
+        kept = await rules.get(rule.id)
+        assert kept.end_date == "2026-12-31"
+        assert kept.month == 3
+        # Explicit None: clears both.
+        await rules.update(rule.id, end_date=None, month=None)
+        cleared = await rules.get(rule.id)
+        assert cleared.end_date is None
+        assert cleared.month is None
+        assert cleared.interval == 2
+        return cleared
+
+    _with_db(tmp_path, "rule-update-clear.db")(_body)
+
+
+def test_rule_update_validates_date_shape(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        for bad in ("not-a-date", "2026-9-4", "2026-13-01", "09-14"):
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await rules.update(rule.id, start_date=bad)
+            with pytest.raises(ValueError, match="YYYY-MM-DD"):
+                await rules.update(rule.id, end_date=bad)
+        # None is still allowed (clears the column) without validation.
+        await rules.update(rule.id, end_date=None)
+        return None
+
+    _with_db(tmp_path, "rule-update-dates.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# schedule_rules: delete-if-unreferenced
+# ---------------------------------------------------------------------------
+
+
+def test_rule_delete_unreferenced_succeeds(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        assert await rules.delete_if_unreferenced(rule.id) is True
+        return await rules.get(rule.id)
+
+    assert _with_db(tmp_path, "rule-delete-ok.db")(_body) is None
+
+
+def test_rule_delete_referenced_is_rejected(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        await definitions.create("Brush teeth", child.id, rule.id, NOW)
+        assert await rules.delete_if_unreferenced(rule.id) is False
+        fetched = await rules.get(rule.id)
+        assert fetched is not None, (
+            "referenced rule must survive the delete attempt"
+        )
+        return fetched
+
+    _with_db(tmp_path, "rule-delete-referenced.db")(_body)
+
+
+def test_rule_delete_missing_returns_false(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await rules.delete_if_unreferenced(999)
+
+    assert _with_db(tmp_path, "rule-delete-missing.db")(_body) is False
+
+
+def test_rule_delete_after_last_reference_removed_succeeds(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create(
+            "Brush teeth", child.id, rule.id, NOW
+        )
+        assert await rules.delete_if_unreferenced(rule.id) is False
+        await definitions.set_active(definition.id, False)
+        # Deactivation is NOT deletion: the definition still references
+        # the rule, so the delete must still be rejected.
+        assert await rules.delete_if_unreferenced(rule.id) is False
+        return None
+
+    _with_db(tmp_path, "rule-delete-inactive-ref.db")(_body)
+
+
+def test_rule_delete_concurrent_with_definition_create_is_safe(
+    tmp_path,
+) -> None:
+    """A racing definition creation cannot produce a dangling reference.
+
+    Both operations queue on the connection lock, so exactly one of two
+    safe outcomes happens: the create wins and the delete is rejected
+    (definition exists, rule exists), or the delete wins and the create
+    fails cleanly with ValueError (rule gone).  A definition row
+    pointing at a deleted rule is impossible in either case.
+
+    BOTH orderings are forced: delete-first by starting the delete and
+    yielding; create-first by hooking ScheduleRulesDao.get so the
+    create's validation pauses mid-flight before the delete starts.
+    """
+    async def _main(order: str):
+        database = NestQuestDatabase(_make_hass_mock())
+        await database.open(tmp_path / f"rule-delete-race-{order}.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            rules = ScheduleRulesDao(database)
+            definitions = TaskDefinitionsDao(database)
+            child = await children.create("Ada", NOW)
+            rule_a = await rules.create("daily", "2026-09-14")
+
+            if order == "create-first":
+                # Pause the create inside its rule validation: the hook
+                # blocks after the rule-lookup statement until the
+                # delete task has been scheduled, so the create holds
+                # the connection lock while the delete queues behind
+                # it — then the create proceeds and wins.
+                import custom_components.nestquest.dao_rules as dao_rules
+
+                original_validate = (
+                    dao_rules.TaskDefinitionsDao._validate_assignable
+                )
+                release = asyncio.Event()
+                started = asyncio.Event()
+
+                async def _pausing_validate(self, child_id, rule_id):
+                    # Run the real validation FIRST (this is the
+                    # time-of-check), THEN pause: the delete task gets
+                    # scheduled while the create sits between its
+                    # passed validation and its INSERT.  The serialized
+                    # transaction must keep the delete queued until the
+                    # insert commits, so the create must win the race
+                    # and the delete must be rejected as referenced.
+                    await original_validate(self, child_id, rule_id)
+                    if rule_id == rule_a.id and not started.is_set():
+                        started.set()
+                        await release.wait()
+
+                dao_rules.TaskDefinitionsDao._validate_assignable = (
+                    _pausing_validate
+                )
+                try:
+                    create_task = asyncio.ensure_future(
+                        definitions.create(
+                            "Brush teeth", child.id, rule_a.id, NOW
+                        )
+                    )
+                    await started.wait()
+                    delete_task = asyncio.ensure_future(
+                        rules.delete_if_unreferenced(rule_a.id)
+                    )
+                    await asyncio.sleep(0)
+                    started.clear()
+                    release.set()
+                    deleted, creation = await asyncio.gather(
+                        delete_task, create_task, return_exceptions=True
+                    )
+                finally:
+                    dao_rules.TaskDefinitionsDao._validate_assignable = (
+                        original_validate
+                    )
+            else:
+                async def _delete():
+                    return await rules.delete_if_unreferenced(rule_a.id)
+
+                async def _create():
+                    return await definitions.create(
+                        "Brush teeth", child.id, rule_a.id, NOW
+                    )
+
+                delete_task = asyncio.ensure_future(_delete())
+                await asyncio.sleep(0)
+                create_task = asyncio.ensure_future(_create())
+                deleted, creation = await asyncio.gather(
+                    delete_task, create_task, return_exceptions=True
+                )
+
+            assert not isinstance(deleted, BaseException), (
+                f"delete must queue, not crash: {deleted!r}"
+            )
+            rule_row = await rules.get(rule_a.id)
+            if deleted:
+                # Delete won: the create must have failed cleanly.
+                assert isinstance(creation, ValueError), (
+                    f"creating against a deleted rule must raise "
+                    f"ValueError, got {creation!r}"
+                )
+                assert rule_row is None
+            else:
+                # Create won: the delete was rejected, both rows exist.
+                assert isinstance(creation, TaskDefinitionRecord), (
+                    f"create must succeed, got {creation!r}"
+                )
+                assert rule_row is not None
+            return deleted
+        finally:
+            await database.close()
+
+    delete_won = _run(_main("delete-first"))
+    create_won = _run(_main("create-first"))
+    # delete-first: the delete acquires the lock before the create can
+    # validate, so the rule is gone and the create fails cleanly.
+    assert delete_won is True
+    # create-first: validation passed, the create holds the lock across
+    # its INSERT, so the queued delete must find the definition and be
+    # rejected — the create must NOT lose its rule after validating it.
+    assert create_won is False
+
+
+# ---------------------------------------------------------------------------
+# task_definitions: create / get
+# ---------------------------------------------------------------------------
+
+
+def test_definition_create_returns_typed_record(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("weekly", "2026-09-14", weekday_set="0,2")
+        definition = await definitions.create(
+            "Brush teeth",
+            child.id,
+            rule.id,
+            NOW,
+            description="Morning and night",
+            icon="mdi:tooth",
+            due_time="08:00",
+        )
+        assert isinstance(definition, TaskDefinitionRecord)
+        assert definition.title == "Brush teeth"
+        assert definition.child_id == child.id
+        assert definition.schedule_rule_id == rule.id
+        assert definition.due_time == "08:00"
+        assert definition.is_active is True
+        assert definition.created_at == NOW
+        return definition
+
+    _with_db(tmp_path, "definition-create.db")(_body)
+
+
+def test_definition_get_missing_returns_none(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await definitions.get(999)
+
+    assert _with_db(tmp_path, "definition-get-missing.db")(_body) is None
+
+
+def test_definition_create_unknown_child_raises_value_error(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.create("X", 999, rule.id, NOW)
+        return None
+
+    _with_db(tmp_path, "definition-bad-child.db")(_body)
+
+
+def test_definition_create_inactive_child_raises_value_error(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        await children.set_active(child.id, False)
+        with pytest.raises(ValueError, match="inactive"):
+            await definitions.create("X", child.id, rule.id, NOW)
+        return None
+
+    _with_db(tmp_path, "definition-inactive-child.db")(_body)
+
+
+def test_definition_create_unknown_rule_raises_value_error(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.create("X", child.id, 999, NOW)
+        return None
+
+    _with_db(tmp_path, "definition-bad-rule.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# task_definitions: list by child / list active
+# ---------------------------------------------------------------------------
+
+
+def test_definition_list_by_child_filters_and_orders(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other_child = await children.create("Bo", NOW)
+        await definitions.create("A", child.id, rule.id, NOW)
+        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.create("C", other_child.id, rule.id, NOW)
+        mine = await definitions.list_by_child(child.id)
+        assert [d.title for d in mine] == ["B", "A"]
+        return mine
+
+    _with_db(tmp_path, "definition-list-by-child.db")(_body)
+
+
+def test_definition_list_by_child_missing_child_returns_empty(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await definitions.list_by_child(999)
+
+    assert _with_db(tmp_path, "definition-list-missing.db")(_body) == []
+
+
+def test_definition_list_active_excludes_inactive(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        first = await definitions.create("A", child.id, rule.id, NOW)
+        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.set_active(first.id, False)
+        active = await definitions.list_active()
+        assert [d.title for d in active] == ["B"]
+        assert all(d.is_active for d in active)
+        return active
+
+    _with_db(tmp_path, "definition-list-active.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# task_definitions: update / set_active / set_assignee
+# ---------------------------------------------------------------------------
+
+
+def test_definition_update_fields_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create(
+            "A", child.id, rule.id, NOW, due_time="08:00"
+        )
+        assert await definitions.update(
+            definition.id,
+            title="A2",
+            description="New",
+            icon="mdi:star",
+            due_time="09:00",
+        ) == 1
+        fetched = await definitions.get(definition.id)
+        assert fetched.title == "A2"
+        assert fetched.description == "New"
+        assert fetched.icon == "mdi:star"
+        assert fetched.due_time == "09:00"
+        return fetched
+
+    _with_db(tmp_path, "definition-update.db")(_body)
+
+
+def test_definition_update_cannot_change_assignee_or_active(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        # title= / child_id= / is_active= are NOT update() parameters;
+        # this asserts the API shape: reassignment and activation have
+        # their own explicit methods.
+        import inspect
+
+        parameters = inspect.signature(definitions.update).parameters
+        assert set(parameters) == {
+            "definition_id",
+            "title",
+            "description",
+            "icon",
+            "due_time",
+        }
+        assert await definitions.update(
+            definition.id, title="A2"
+        ) == 1
+        fetched = await definitions.get(definition.id)
+        assert fetched.child_id == child.id
+        assert fetched.is_active is True
+        return fetched
+
+    _with_db(tmp_path, "definition-update-shape.db")(_body)
+
+
+def test_definition_update_no_fields_returns_zero(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        return await definitions.update(definition.id)
+
+    assert _with_db(tmp_path, "definition-update-empty.db")(_body) == 0
+
+
+def test_definition_update_can_clear_optional_fields(tmp_path) -> None:
+    """Explicit None removes description/icon/due_time; omitted keeps."""
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create(
+            "A",
+            child.id,
+            rule.id,
+            NOW,
+            description="d",
+            icon="mdi:x",
+            due_time="08:00",
+        )
+        await definitions.update(definition.id, title="A2")
+        kept = await definitions.get(definition.id)
+        assert kept.description == "d"
+        assert kept.icon == "mdi:x"
+        await definitions.update(
+            definition.id, description=None, icon=None, due_time=None
+        )
+        cleared = await definitions.get(definition.id)
+        assert cleared.description is None
+        assert cleared.icon is None
+        assert cleared.due_time is None
+        assert cleared.title == "A2"
+        return cleared
+
+    _with_db(tmp_path, "definition-update-clear.db")(_body)
+
+
+def test_definition_update_missing_returns_zero(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await definitions.update(999, title="X")
+
+    assert _with_db(tmp_path, "definition-update-missing.db")(_body) == 0
+
+
+def test_definition_set_active_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        assert await definitions.set_active(definition.id, False) == 1
+        fetched = await definitions.get(definition.id)
+        assert fetched.is_active is False
+        assert await definitions.set_active(definition.id, True) == 1
+        fetched = await definitions.get(definition.id)
+        assert fetched.is_active is True
+        return fetched
+
+    _with_db(tmp_path, "definition-set-active.db")(_body)
+
+
+def test_definition_set_active_missing_returns_zero(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        return await definitions.set_active(999, False)
+
+    assert _with_db(tmp_path, "definition-set-active-missing.db")(_body) == 0
+
+
+def test_definition_no_delete_method_exists(tmp_path) -> None:
+    """Guardrail: definitions are deactivated, never hard-deleted."""
+    async def _body(database, rules, definitions, children, child):
+        import inspect
+
+        methods = {
+            name
+            for name, _ in inspect.getmembers(
+                TaskDefinitionsDao, inspect.isfunction
+            )
+        }
+        assert "delete" not in methods
+        assert not any(name.startswith("delete") for name in methods)
+        return methods
+
+    _with_db(tmp_path, "definition-no-delete.db")(_body)
+
+
+def test_definition_set_assignee_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        assert await definitions.set_assignee(definition.id, other.id) == 1
+        fetched = await definitions.get(definition.id)
+        assert fetched.child_id == other.id
+        return fetched
+
+    _with_db(tmp_path, "definition-set-assignee.db")(_body)
+
+
+def test_definition_set_assignee_inactive_child_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        await children.set_active(other.id, False)
+        with pytest.raises(ValueError, match="inactive"):
+            await definitions.set_assignee(definition.id, other.id)
+        fetched = await definitions.get(definition.id)
+        assert fetched.child_id == child.id, (
+            "failed reassignment must leave the current assignee intact"
+        )
+        return fetched
+
+    _with_db(tmp_path, "definition-assignee-inactive.db")(_body)
+
+
+def test_definition_set_assignee_unknown_child_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.set_assignee(definition.id, 999)
+        return None
+
+    _with_db(tmp_path, "definition-assignee-unknown.db")(_body)
+
+
+def test_definition_set_assignee_missing_definition_returns_zero(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        other = await children.create("Bo", NOW)
+        return await definitions.set_assignee(999, other.id)
+
+    assert _with_db(tmp_path, "definition-assignee-missing.db")(_body) == 0
+
+
+def test_definition_set_assignee_leaves_history_rows_alone(
+    tmp_path,
+) -> None:
+    """Reassignment writes the definition row only: any instances that
+    already exist keep their generation-time child_id (Feature 06
+    guardrail 'changes future instances only').
+    """
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create("A", child.id, rule.id, NOW)
+        # Simulate an already-generated instance (materialization owns
+        # this table, but its row shape proves reassignment isolation).
+        await database.execute(
+            "INSERT INTO task_instances (definition_id, child_id, "
+            "due_date, generated_at) VALUES (?, ?, ?, ?)",
+            (definition.id, child.id, "2026-09-15", NOW),
+        )
+        await definitions.set_assignee(definition.id, other.id)
+        row = await database.fetch_one(
+            "SELECT child_id FROM task_instances WHERE definition_id = ?",
+            (definition.id,),
+        )
+        assert row == (child.id,)
+        return None
+
+    _with_db(tmp_path, "definition-assignee-history.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# No SQL for these tables outside the DAO module
+# ---------------------------------------------------------------------------
+
+
+def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
+    """Guardrail: schedule_rules/task_definitions SQL may appear only
+    in the DAO modules (and the schema DDL declarations).  RECURSIVE
+    scan of package and tests, FROM/INTO/UPDATE/DELETE FROM/JOIN
+    pattern, with exact repo-relative exclusions justified by role:
+    schema.py declares the DDL, migrations.py applies it, and the
+    test files for exactly those layers verify DDL, not data paths.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+    repo_root = package.parent.parent
+    scan_roots = [package, repo_root / "tests"]
+
+    allowed = {
+        "custom_components/nestquest/dao_rules.py",  # this DAO
+        "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.
+    }
+    sql_pattern = re.compile(
+        r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
+        r"(schedule_rules|task_definitions)\b",
+        re.IGNORECASE,
+    )
+    offenders: list[str] = []
+    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()):
+                offenders.append(relative)
+    assert offenders == [], (
+        f"SQL touching schedule_rules/task_definitions leaked into: "
+        f"{offenders}"
+    )
+
+
+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
+    mention the names), and the reassignment-isolation test, which
+    legitimately touches task_instances — a table this DAO does not
+    own — to prove the guardrail that reassignment leaves it alone.
+    """
+    import ast
+    import re
+    from pathlib import Path
+
+    path = Path(__file__)
+    text = path.read_text()
+    lines = text.splitlines(keepends=True)
+    tree = ast.parse(text)
+    guard_names = {
+        "test_rules_and_definitions_sql_lives_only_in_dao_module",
+        "test_dao_rules_test_file_uses_dao_not_raw_table_sql",
+    }
+    excluded: set[int] = set()
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and 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, (
+        "tests must go through the DAO, not raw SQL, for these tables"
+    )