test: seed pre-existing instance/history and add concurrency test (NQ-70-001, NQ-70-002)

Review round 2 (Codex gpt-5.6-terra, 2 LOW): 2015-01-06 is a Tuesday,
not a Monday — the anchor is now the computed first Monday of January
(asserted weekday 0); the docstring names the ISO-week rollover
precisely (the Monday on or after January 4th — January 4th in both
walked years, which end ISO week 53) instead of 'January 1st'. Suite:
1005 passed.

custom_components/nestquest/dao_presence.py

@@ -186,6 +186,15 @@ class PresenceOverridesDao:
         The child must exist (validated atomically under the
         connection lock).  Date shape is validated strictly, and the
         schema CHECK rejects end < start.
+
+        Overlap policy (Feature 05): an override for a child may not
+        overlap another override of the SAME child — two different
+        present/absent answers for one date are unresolvable.  The
+        overlap check (inclusive boundaries) runs inside the same
+        transaction as the INSERT, so a racing create cannot slip a
+        conflicting override between the check and the write; a
+        conflict raises ValueError naming the conflicting override's
+        id and range.
         """
         _validate_date(start_date, "start_date")
         _validate_date(end_date, "end_date")
@@ -196,6 +205,29 @@ class PresenceOverridesDao:
                 )
                 if child is None:
                     raise ValueError(f"child {child_id} does not exist")
+                # Only a well-ordered range has overlap semantics:
+                # inverted ranges fall through to the INSERT and hit
+                # the schema's end>=start CHECK, preserving the
+                # IntegrityError contract the matrix tests rely on.
+                if end_date >= start_date:
+                    conflicting = await self._database.fetch_one(
+                        f"SELECT {_OVERRIDE_COLUMNS} "
+                        "FROM presence_overrides "
+                        "WHERE child_id = ? AND end_date >= ? "
+                        "AND start_date <= ? ORDER BY id LIMIT 1",
+                        (child_id, start_date, end_date),
+                    )
+                else:
+                    conflicting = None
+                if conflicting is not None:
+                    conflict = _override_from_row(conflicting)
+                    raise ValueError(
+                        f"override for child {child_id} "
+                        f"[{start_date}, {end_date}] overlaps existing "
+                        f"override {conflict.id} "
+                        f"[{conflict.start_date}, {conflict.end_date}] "
+                        f"(is_present={conflict.is_present})"
+                    )
                 result = await self._database.execute(
                     "INSERT INTO presence_overrides (child_id, "
                     "start_date, end_date, is_present, note) "

custom_components/nestquest/dao_rules.py

@@ -40,6 +40,7 @@ from .const import QUEST_WINDOWS
 from .dao_children import ChildRecord, _child_from_row, _connection_lock
 from .dao_children import _CHILD_COLUMNS
 from .db import NestQuestDatabase
+from .recurrence import RuleType, RuleValidationError, ScheduleRule
 
 #: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
 #: in update methods: passing ``None`` must mean clearing a nullable
@@ -119,6 +120,26 @@ class ScheduleRuleRecord:
     end_date: str | None
 
 
+@dataclass(frozen=True)
+class ScheduleRuleStorage:
+    """The eight value columns a ScheduleRule maps to (no id column).
+
+    ``id`` is assigned by the database on insert, so the model<->storage
+    mapping concerns only the value columns.  ``nth_weekday_weekday`` is
+    deliberately absent here: for MONTHLY_WEEKDAY its value is folded
+    into ``weekday_set`` as that column's single CSV element.
+    """
+
+    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 QuestDefinitionRecord:
     """One row of ``quest_definitions`` (without its assignees)."""
@@ -142,6 +163,24 @@ class QuestDefinitionWindowRecord:
     due_time: str | None
 
 
+@dataclass(frozen=True)
+class QuestDefinitionSnapshot:
+    """A definition plus its rule, assignees and windows, read atomically.
+
+    Returned by :meth:`QuestDefinitionsDao.create_with_rule_and_windows`
+    and :meth:`QuestDefinitionsDao.edit_definition` with the rule,
+    assignees and windows read inside the SAME transaction that wrote
+    them, so the snapshot is a consistent creation/edit-time view — a
+    concurrent rule, assignee or window mutation cannot change it
+    between the write and the read.
+    """
+
+    definition: QuestDefinitionRecord
+    rule: ScheduleRuleRecord
+    assignees: list[ChildRecord]
+    windows: list[QuestDefinitionWindowRecord]
+
+
 _RULE_COLUMNS = (
     "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
     "month, start_date, end_date"
@@ -188,6 +227,250 @@ def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
     )
 
 
+#: Storage columns that are forbidden for each rule type.  A populated
+#: forbidden column means the row's shape is invalid and must raise on
+#: read, never be silently dropped.  ``weekday_set`` is absent for
+#: DAILY/MONTHLY_DAY/YEARLY; MONTHLY_WEEKDAY is the only shape allowed a
+#: non-NULL ``nth_weekday``; YEARLY is the only shape allowed ``month``.
+_FORBIDDEN_STORAGE_FIELDS: dict[str, tuple[RuleType, ...]] = {
+    "weekday_set": (RuleType.DAILY, RuleType.MONTHLY_DAY, RuleType.YEARLY),
+    "day_of_month": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_WEEKDAY,
+    ),
+    "nth_weekday": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_DAY,
+        RuleType.YEARLY,
+    ),
+    "month": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_DAY,
+        RuleType.MONTHLY_WEEKDAY,
+    ),
+}
+
+
+def schedule_rule_to_storage(rule: ScheduleRule) -> ScheduleRuleStorage:
+    """Map a validated ScheduleRule onto its storage columns.
+
+    ``rule_type`` stores as ``RuleType.storage_value`` (MONTHLY_DAY and
+    MONTHLY_WEEKDAY both store ``'monthly'``); ``weekday_set`` stores as
+    a sorted CSV list.  MONTHLY_WEEKDAY folds its model-only
+    ``nth_weekday_weekday`` into ``weekday_set`` as that list's single
+    element; every other type leaves the column as its own set
+    (WEEKLY/CUSTOM_DAYS) or NULL.
+    """
+    shape = rule.rule_type
+    if shape is RuleType.MONTHLY_WEEKDAY:
+        weekday_set = str(rule.nth_weekday_weekday)
+    elif shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
+        weekday_set = ",".join(str(w) for w in sorted(rule.weekday_set))
+    else:
+        weekday_set = None
+    return ScheduleRuleStorage(
+        rule_type=shape.storage_value,
+        interval=rule.interval,
+        weekday_set=weekday_set,
+        day_of_month=rule.day_of_month,
+        nth_weekday=rule.nth_weekday,
+        month=rule.month,
+        start_date=rule.start_date,
+        end_date=rule.end_date,
+    )
+
+
+def schedule_rule_from_storage(storage: ScheduleRuleStorage) -> ScheduleRule:
+    """Reconstruct a ScheduleRule from storage columns.
+
+    Disambiguates MONTHLY_DAY from MONTHLY_WEEKDAY by which field is
+    populated, and raises RuleValidationError rather than silently
+    mis-decoding an ambiguous or invalid shape.
+    """
+    rule_type = storage.rule_type
+    if rule_type == "daily":
+        shape = RuleType.DAILY
+    elif rule_type == "weekly":
+        shape = RuleType.WEEKLY
+    elif rule_type == "monthly":
+        shape = _disambiguate_monthly(storage)
+    elif rule_type == "yearly":
+        shape = RuleType.YEARLY
+    elif rule_type == "custom":
+        shape = RuleType.CUSTOM_DAYS
+    else:
+        raise RuleValidationError(
+            f"unknown storage rule_type {rule_type!r}"
+        )
+
+    _reject_forbidden_storage_fields(storage, shape)
+
+    weekday_set = None
+    nth_weekday_weekday = None
+    if shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
+        weekday_set = _weekdays_from_csv(storage.weekday_set)
+    elif shape is RuleType.MONTHLY_WEEKDAY:
+        nth_weekday_weekday = _single_weekday_from_csv(storage.weekday_set)
+
+    return ScheduleRule(
+        rule_type=shape,
+        interval=storage.interval,
+        weekday_set=weekday_set,
+        day_of_month=storage.day_of_month,
+        nth_weekday=storage.nth_weekday,
+        nth_weekday_weekday=nth_weekday_weekday,
+        month=storage.month,
+        start_date=storage.start_date,
+        end_date=storage.end_date,
+    )
+
+
+def schedule_rule_storage_from_record(
+    record: ScheduleRuleRecord,
+) -> ScheduleRuleStorage:
+    """Copy a rule row's value columns into a storage mapping.
+
+    ``ScheduleRuleRecord`` is the read shape (it carries ``id``);
+    :func:`schedule_rule_from_storage` consumes the value-only
+    :class:`ScheduleRuleStorage` shape, so callers re-reading a stored
+    rule convert through this helper.
+    """
+    return ScheduleRuleStorage(
+        rule_type=record.rule_type,
+        interval=record.interval,
+        weekday_set=record.weekday_set,
+        day_of_month=record.day_of_month,
+        nth_weekday=record.nth_weekday,
+        month=record.month,
+        start_date=record.start_date,
+        end_date=record.end_date,
+    )
+
+
+def _reject_forbidden_storage_fields(
+    storage: ScheduleRuleStorage, shape: RuleType
+) -> None:
+    """Raise unless every populated column is permitted for ``shape``.
+
+    The model constructor rejects a forbidden ``day_of_month``,
+    ``nth_weekday``, or ``month``, but ``weekday_set`` never reaches it
+    (it is folded for MONTHLY_WEEKDAY and dropped otherwise), so a daily
+    row carrying ``weekday_set="0"`` would otherwise decode as a valid
+    daily rule and silently lose the stored data.  This validates every
+    shape column explicitly so a populated-forbidden column raises
+    instead of being discarded.
+    """
+    for field, forbidden_shapes in _FORBIDDEN_STORAGE_FIELDS.items():
+        value = getattr(storage, field)
+        if value is not None and shape in forbidden_shapes:
+            raise RuleValidationError(
+                f"{shape.value} storage rows must not set {field}, "
+                f"got {value!r}"
+            )
+
+
+def _disambiguate_monthly(storage: ScheduleRuleStorage) -> RuleType:
+    """Resolve the shared 'monthly' storage value to a model type.
+
+    MONTHLY_DAY names ``day_of_month`` only; MONTHLY_WEEKDAY names
+    ``nth_weekday`` only.  Neither, or both, is ambiguous and raises.
+    """
+    day_of_month = storage.day_of_month
+    nth_weekday = storage.nth_weekday
+    if day_of_month is not None and nth_weekday is None:
+        return RuleType.MONTHLY_DAY
+    if nth_weekday is not None and day_of_month is None:
+        return RuleType.MONTHLY_WEEKDAY
+    raise RuleValidationError(
+        "storage rule_type 'monthly' is ambiguous: must name exactly one "
+        "of day_of_month or nth_weekday, got "
+        f"day_of_month={day_of_month!r}, nth_weekday={nth_weekday!r}"
+    )
+
+
+def _weekdays_from_csv(value: str | None) -> frozenset[int]:
+    """Parse a weekday CSV column into a frozenset of ints.
+
+    ``None`` and ``""`` both yield an empty frozenset (the caller decides
+    whether empty is legal); a non-integer token raises
+    RuleValidationError rather than leaking a raw ValueError.
+    """
+    if value is None or value == "":
+        return frozenset()
+    try:
+        entries = [int(token) for token in value.split(",")]
+    except ValueError:
+        raise RuleValidationError(
+            f"weekday_set must be a CSV of integers, got {value!r}"
+        ) from None
+    return frozenset(entries)
+
+
+def _single_weekday_from_csv(value: str | None) -> int:
+    """The one weekday a MONTHLY_WEEKDAY stores in weekday_set.
+
+    Parses the CSV into tokens FIRST and requires exactly one token,
+    so a duplicate pair like ``"1,1"`` is rejected rather than silently
+    canonicalized to a single weekday on re-storage.  An empty or
+    multi-element column raises RuleValidationError.
+    """
+    if value is None or value == "":
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
+            f"weekday, got {value!r}"
+        )
+    tokens = value.split(",")
+    if len(tokens) != 1:
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
+            f"weekday, got {value!r}"
+        )
+    try:
+        return int(tokens[0])
+    except ValueError:
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must be an integer, "
+            f"got {value!r}"
+        ) from None
+
+
+async def _insert_rule_row(
+    database: NestQuestDatabase, storage: ScheduleRuleStorage
+) -> int:
+    """Insert one ``schedule_rules`` row and return its new id.
+
+    Runs inside the caller's transaction (no lock or transaction of its
+    own) so a multi-table write can persist the rule together with the
+    rows that reference it.  The caller is responsible for validating
+    the storage fields — :meth:`ScheduleRulesDao.create` validates the
+    date shape, and :meth:`QuestDefinitionsDao.create_with_rule_and_windows`
+    receives storage already mapped from a validated
+    :class:`~.recurrence.ScheduleRule`.
+    """
+    result = await database.execute(
+        "INSERT INTO schedule_rules (rule_type, interval, weekday_set, "
+        "day_of_month, nth_weekday, month, start_date, end_date) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            storage.rule_type,
+            storage.interval,
+            storage.weekday_set,
+            storage.day_of_month,
+            storage.nth_weekday,
+            storage.month,
+            storage.start_date,
+            storage.end_date,
+        ),
+    )
+    return result.lastrowid
+
+
 class ScheduleRulesDao:
     """Typed async access to the ``schedule_rules`` table."""
 
@@ -218,22 +501,20 @@ class ScheduleRulesDao:
         _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_id = await _insert_rule_row(
+            self._database,
+            ScheduleRuleStorage(
+                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,
             ),
         )
-        rule = await self.get(result.lastrowid)
+        rule = await self.get(rule_id)
         assert rule is not None
         return rule
 
@@ -393,6 +674,191 @@ class QuestDefinitionsDao:
         assert definition is not None
         return definition
 
+    async def create_with_rule_and_windows(
+        self,
+        title: str,
+        schedule_rule: ScheduleRuleStorage,
+        created_at: str,
+        assignee_child_ids: list[int],
+        windows: list[tuple[str, str | None]],
+        *,
+        description: str | None = None,
+        icon: str | None = None,
+    ) -> QuestDefinitionSnapshot:
+        """Insert a rule, definition, assignees and windows atomically.
+
+        The whole write — the schedule rule row, the definition row,
+        every assignee link and every window declaration — runs inside
+        ONE transaction under the connection lock, so a validation or
+        write failure rolls the lot back together (all-or-nothing): the
+        caller can never observe a definition with a missing rule,
+        assignee or window, and a rejected create leaves no partial
+        rows.
+
+        Returns a :class:`QuestDefinitionSnapshot` whose assignees and
+        windows are read back inside the same transaction, so the
+        returned view cannot be raced by a concurrent assignee/window
+        mutation after the write.
+
+        The rule arrives as pre-validated storage fields (mapped from a
+        :class:`~.recurrence.ScheduleRule` via
+        :func:`schedule_rule_to_storage`); its dates are re-validated
+        here so a hand-built storage can never persist a malformed date.
+        The definition-level ``due_time`` column is deliberately left
+        NULL — per-window due times supersede it (D-008).  Assignees are
+        validated as existing-and-active at insert time, and windows are
+        validated for name and strict HH:MM due time, matching
+        :meth:`upsert_window`.
+        """
+        _validate_date(schedule_rule.start_date, "start_date")
+        if schedule_rule.end_date is not None:
+            _validate_date(schedule_rule.end_date, "end_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                for child_id in assignee_child_ids:
+                    await self._validate_assignable_child(child_id)
+                rule_id = await _insert_rule_row(self._database, schedule_rule)
+                result = await self._database.execute(
+                    "INSERT INTO quest_definitions (title, description, "
+                    "icon, schedule_rule_id, due_time, is_active, "
+                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (title, description, icon, rule_id, None, 1, created_at),
+                )
+                definition_id = result.lastrowid
+                for child_id in assignee_child_ids:
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_assignees "
+                        "(definition_id, child_id) VALUES (?, ?)",
+                        (definition_id, child_id),
+                    )
+                for window, due_time in windows:
+                    _validate_window(window)
+                    if due_time is not None:
+                        _validate_time(due_time, "due_time")
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_windows "
+                        "(definition_id, window, due_time) VALUES (?, ?, ?)",
+                        (definition_id, window, due_time),
+                    )
+                definition = await self.get(definition_id)
+                assignees = await self.list_assignees(definition_id)
+                window_records = await self.list_windows(definition_id)
+                rule_record = await ScheduleRulesDao(self._database).get(
+                    rule_id
+                )
+        assert definition is not None
+        assert rule_record is not None
+        return QuestDefinitionSnapshot(
+            definition=definition,
+            rule=rule_record,
+            assignees=assignees,
+            windows=window_records,
+        )
+
+    async def edit_definition(
+        self,
+        definition_id: int,
+        *,
+        title: str | None | object = _UNSET,
+        description: str | None | object = _UNSET,
+        icon: str | None | object = _UNSET,
+        rule: ScheduleRuleStorage | object = _UNSET,
+        windows: list[tuple[str, str | None]] | object = _UNSET,
+    ) -> QuestDefinitionSnapshot:
+        """Edit a definition's metadata, rule and windows atomically.
+
+        Mirrors :meth:`create_with_rule_and_windows`: everything runs
+        inside ONE transaction under the connection lock, and the
+        returned snapshot's rule, assignees and windows are read back
+        inside that transaction.  Arguments default to the module sentinel
+        ``_UNSET`` meaning "leave this field alone"; ``None`` writes SQL
+        NULL so optional metadata (description, icon) can be cleared.
+
+        - ``title``/``description``/``icon`` update the definition row
+          in place; the definition-level ``due_time`` column is never
+          written (per-window due times supersede it, D-008).
+        - ``rule`` (a pre-validated :class:`ScheduleRuleStorage`)
+          overwrites the referenced ``schedule_rules`` row in place: the
+          definition keeps its ``schedule_rule_id``, so no new or
+          orphaned rule row appears.
+        - ``windows`` REPLACES the whole window set: the current rows
+          are deleted and the given set re-inserted, so missing windows
+          are added, changed due times updated and removed windows
+          dropped.
+
+        Assignment is deliberately not settable here (multi-assignee is
+        managed by :meth:`add_assignee`/:meth:`remove_assignee`), and
+        activation is untouched.  Raises ValueError when the definition
+        does not exist.  Edits change future instances only: no
+        ``quest_instances`` or ``completion_events`` row is touched.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definition = await self.get(definition_id)
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                assignments: list[str] = []
+                parameters: list[object] = []
+                for column, value in (
+                    ("title", title),
+                    ("description", description),
+                    ("icon", icon),
+                ):
+                    if value is not _UNSET:
+                        assignments.append(f"{column} = ?")
+                        parameters.append(value)
+                if assignments:
+                    parameters.append(definition_id)
+                    await self._database.execute(
+                        f"UPDATE quest_definitions SET "
+                        f"{', '.join(assignments)} WHERE id = ?",
+                        tuple(parameters),
+                    )
+                if rule is not _UNSET:
+                    await ScheduleRulesDao(self._database).update(
+                        definition.schedule_rule_id,
+                        rule_type=rule.rule_type,
+                        interval=rule.interval,
+                        weekday_set=rule.weekday_set,
+                        day_of_month=rule.day_of_month,
+                        nth_weekday=rule.nth_weekday,
+                        month=rule.month,
+                        start_date=rule.start_date,
+                        end_date=rule.end_date,
+                    )
+                if windows is not _UNSET:
+                    await self._database.execute(
+                        "DELETE FROM quest_definition_windows "
+                        "WHERE definition_id = ?",
+                        (definition_id,),
+                    )
+                    for window, due_time in windows:
+                        _validate_window(window)
+                        if due_time is not None:
+                            _validate_time(due_time, "due_time")
+                        await self._database.execute(
+                            "INSERT INTO quest_definition_windows "
+                            "(definition_id, window, due_time) "
+                            "VALUES (?, ?, ?)",
+                            (definition_id, window, due_time),
+                        )
+                updated = await self.get(definition_id)
+                assignees = await self.list_assignees(definition_id)
+                window_records = await self.list_windows(definition_id)
+                rule_record = await ScheduleRulesDao(self._database).get(
+                    updated.schedule_rule_id
+                )
+        assert updated is not None
+        assert rule_record is not None
+        return QuestDefinitionSnapshot(
+            definition=updated,
+            rule=rule_record,
+            assignees=assignees,
+            windows=window_records,
+        )
+
     async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
         """Raise ValueError unless the rule exists.
 
@@ -496,14 +962,18 @@ class QuestDefinitionsDao:
 
     async def add_assignee(
         self, definition_id: int, child_id: int
-    ) -> None:
+    ) -> list[ChildRecord]:
         """Assign ``child_id`` to the definition (idempotent).
 
         Validates the assignee inside the same serialized transaction
         as the INSERT (existing and active AT ASSIGNMENT TIME), so a
         child deactivated concurrently cannot become an assignee.
         Assigning an already-assigned child is a no-op, mirroring the
-        composite primary key's storage-level guarantee.  Changes
+        composite primary key's storage-level guarantee.  Returns the
+        definition's assignees read back INSIDE the same transaction
+        that performed the write, so the returned roster is a
+        consistent post-assignment snapshot — a concurrent assign or
+        unassign cannot race it after the write commits.  Changes
         future instances only: already-generated instances and
         completion history are never rewritten here.
         """
@@ -524,6 +994,7 @@ class QuestDefinitionsDao:
                     "ON CONFLICT (definition_id, child_id) DO NOTHING",
                     (definition_id, child_id),
                 )
+                return await self.list_assignees(definition_id)
 
     async def remove_assignee(
         self, definition_id: int, child_id: int

custom_components/nestquest/presence.py

@@ -317,3 +317,161 @@ def cycle_week_index(
     target = _parse_date(target_date, "target_date")
     days = (target - schedule.anchor_date).days
     return (days // 7) % schedule.cycle_length_weeks
+
+
+#: Search cap for :meth:`PresenceEngine.next_present_dates`: two full
+#: years of daily checks.  A child absent every day (an all-absent
+#: pattern with a blocking override) would otherwise loop forever; the
+#: cap bounds the scan and surfaces the situation as an error instead.
+MAX_PREVIEW_SCAN_DAYS = 366 * 2
+
+
+class PresenceEngine:
+    """Answers "is this child at this house on this date?".
+
+    Pure and side-effect-free: built from an immutable snapshot of the
+    children's schedules (``child_id -> PresenceSchedule``), it never
+    touches the database — callers build a snapshot from the DAO layer
+    and pass it in.  A child with NO schedule entry is present every
+    day (the third household child's case); a child WITH a schedule is
+    present exactly when the date's weekday is in that schedule's
+    pattern for the date's cycle week index (anchor-date arithmetic —
+    :func:`cycle_week_index`, never ISO week parity).
+    """
+
+    def __init__(
+        self,
+        schedules: dict[int, PresenceSchedule],
+        overrides: dict[int, list[PresenceOverride]] | None = None,
+    ) -> None:
+        if not isinstance(schedules, dict):
+            raise ValueError(
+                f"schedules must map child_id to PresenceSchedule, got "
+                f"{schedules!r}"
+            )
+        for child_id, schedule in schedules.items():
+            # Validate the KEY as a model child id too: True and 1.0
+            # compare equal to 1, so {True: schedule(1)} would be
+            # accepted and then behave differently on lookup.
+            _validate_child_id(child_id)
+            if not isinstance(schedule, PresenceSchedule):
+                raise ValueError(
+                    f"schedules[{child_id!r}] must be a PresenceSchedule, "
+                    f"got {schedule!r}"
+                )
+            if schedule.child_id != child_id:
+                raise ValueError(
+                    f"schedules is keyed {child_id!r} but holds the "
+                    f"schedule of child {schedule.child_id}"
+                )
+        if overrides is None:
+            overrides = {}
+        if not isinstance(overrides, dict):
+            raise ValueError(
+                "overrides must map child_id to a list of "
+                f"PresenceOverride, got {overrides!r}"
+            )
+        for child_id, entries in overrides.items():
+            _validate_child_id(child_id)
+            if not isinstance(entries, list):
+                raise ValueError(
+                    f"overrides[{child_id!r}] must be a list of "
+                    f"PresenceOverride, got {entries!r}"
+                )
+            for entry in entries:
+                if not isinstance(entry, PresenceOverride):
+                    raise ValueError(
+                        f"overrides[{child_id!r}] entries must be "
+                        f"PresenceOverride, got {entry!r}"
+                    )
+                if entry.child_id != child_id:
+                    raise ValueError(
+                        f"overrides is keyed {child_id!r} but holds an "
+                        f"override of child {entry.child_id}"
+                    )
+        # Read-only snapshot: reassignment is refused and no mutator
+        # exists, so a built engine answers consistently for its
+        # lifetime.  The overrides lists are copied to immutable
+        # tuples — a shallow mapping copy alone would leave caller-
+        # owned lists reachable through _overrides, and mutating one
+        # later would change is_present results.
+        object.__setattr__(self, "_schedules", MappingProxyType(dict(schedules)))
+        object.__setattr__(
+            self,
+            "_overrides",
+            MappingProxyType(
+                {child_id: tuple(entries) for child_id, entries in overrides.items()}
+            ),
+        )
+
+    def is_present(self, child_id: object, date: object) -> bool:
+        """Return whether ``child_id`` is present on ``date``.
+
+        Overrides take precedence over the repeating pattern: when an
+        override covers the date, its flag IS the answer — including
+        for a child with no schedule at all.  The DAO layer guarantees
+        a child's overrides never overlap; if a snapshot still carries
+        two covering the same date, the one with the LATEST start date
+        wins (deterministic, and matches "the most specific swap wins").
+        """
+        if isinstance(child_id, bool) or not isinstance(child_id, int):
+            raise ValueError(f"child_id must be an integer, got {child_id!r}")
+        target = _parse_date(date, "date")
+        entries = self._overrides.get(child_id) or []
+        covering = [
+            entry for entry in entries if entry.covers(target)
+        ]
+        if covering:
+            winner = max(covering, key=lambda entry: entry.start_date)
+            return winner.is_present
+        schedule = self._schedules.get(child_id)
+        if schedule is None:
+            # No schedule at all: present every day (household default).
+            return True
+        week = cycle_week_index(schedule, target)
+        return target.weekday() in schedule.pattern[week]
+
+    def next_present_dates(
+        self,
+        child_id: object,
+        start_date: object,
+        count: int,
+    ) -> list[datetime.date]:
+        """Return the next ``count`` dates the child is present.
+
+        Walks forward from ``start_date`` (INCLUSIVE — a parent
+        previewing "starting Monday" expects Monday itself when the
+        child is present) checking :meth:`is_present` for every day,
+        so overrides and the pattern are honoured identically.  The
+        admin Schedule tab renders this preview so a parent can confirm
+        a custody pattern is right before saving it.
+
+        A child absent every scanned day raises ValueError after the
+        two-year search cap — an always-absent child would otherwise
+        loop forever.  ``count`` must be a positive integer.
+        """
+        _validate_child_id(child_id)
+        start = _parse_date(start_date, "start_date")
+        if isinstance(count, bool) or not isinstance(count, int):
+            raise ValueError(f"count must be an integer, got {count!r}")
+        if count < 1:
+            raise ValueError(f"count must be at least 1, got {count}")
+        found: list[datetime.date] = []
+        day = start
+        for _ in range(MAX_PREVIEW_SCAN_DAYS):
+            if self.is_present(child_id, day):
+                found.append(day)
+                if len(found) == count:
+                    return found
+            day += datetime.timedelta(days=1)
+        raise ValueError(
+            f"child {child_id} is not present on any of the "
+            f"{MAX_PREVIEW_SCAN_DAYS} days from {start.isoformat()}; "
+            f"{count} present day(s) cannot be previewed"
+        )
+
+    def __setattr__(self, name: object, value: object) -> None:
+        raise AttributeError(
+            "PresenceEngine is an immutable snapshot; build a new engine "
+            "instead of mutating this one"
+        )

custom_components/nestquest/quest_definitions.py

@@ -0,0 +1,454 @@
+"""Business layer for creating and editing quest definitions (Feature 06).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_rules.QuestDefinitionsDao`: callers get validation and the
+all-or-nothing create/edit here, and never touch the DAO SQL directly.
+:func:`create_quest_definition` is the create path,
+:func:`edit_quest_definition` the edit path, :func:`assign_child` and
+:func:`unassign_child` the assignment paths, and
+:func:`set_quest_definition_active` the deactivate/reactivate path.
+There is no permission check here (that is Feature 09).
+
+Validation policy mirrors :mod:`.children`:
+
+- ``title`` is required and trimmed; empty or whitespace-only raises
+  ValueError.
+- ``rule`` must be an already-validated :class:`~.recurrence.ScheduleRule`
+  (its constructor is total — rule validation is NOT re-implemented
+  here).
+- ``assignee_child_ids`` must be a non-empty list (strictly a ``list``);
+  every id is a plain int (bools and floats rejected, since SQLite would
+  bind them onto a real child) and must reference an existing ACTIVE
+  child (the DAO enforces the active/exists check inside the same
+  transaction).  Duplicate ids are rejected here, before the composite
+  primary key could surface them as an :class:`sqlite3.IntegrityError`.
+- ``windows`` must be a non-empty list (strictly a ``list``); each entry
+  is a window name from :data:`~.const.QUEST_WINDOWS` — either a bare
+  name (no due time) or a ``(name, due_time)`` pair whose due time is a
+  strict 24-hour HH:MM string.  Duplicate window names are rejected
+  here, before the composite primary key could surface them as an
+  :class:`sqlite3.IntegrityError`.
+
+The rule, definition, assignees and windows are persisted in ONE
+transaction (the DAO's :meth:`~.dao_rules.QuestDefinitionsDao.create_with_rule_and_windows`),
+so a rejected or failed create persists nothing.  The definition-level
+``due_time`` column is superseded by per-window due times (D-008) and is
+never set here.  ``created_at`` is stamped UTC ISO-8601 at second
+precision, matching :mod:`.children`.
+"""
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+
+from .dao_children import ChildRecord, _connection_lock
+from .dao_rules import (
+    QuestDefinitionRecord,
+    QuestDefinitionWindowRecord,
+    QuestDefinitionsDao,
+    ScheduleRuleStorage,
+    ScheduleRulesDao,
+    _UNSET,
+    _validate_time,
+    _validate_window,
+    schedule_rule_from_storage,
+    schedule_rule_storage_from_record,
+    schedule_rule_to_storage,
+)
+from .db import NestQuestDatabase
+from .recurrence import ScheduleRule
+
+
+@dataclass(frozen=True)
+class CreatedQuestDefinition:
+    """The persisted result of a create, edit or activation change.
+
+    ``definition`` is the stored row; ``rule`` is the schedule rule
+    decoded back through the storage mapping; ``assignees`` and
+    ``windows`` are the definition's linked child records and window
+    declarations, read through the DAO.
+    """
+
+    definition: QuestDefinitionRecord
+    rule: ScheduleRule
+    assignees: list[ChildRecord]
+    windows: list[QuestDefinitionWindowRecord]
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _validate_text(
+    value: object, field: str, *, required: bool = False
+) -> str | None:
+    """Validate a free-text field and return it trimmed.
+
+    ``None`` raises when ``required`` and otherwise returns None.  A
+    non-string value is rejected; an empty result is rejected for
+    required fields.
+    """
+    if value is None:
+        if required:
+            raise ValueError(f"{field} is required")
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{field} must be a string, got {value!r}")
+    trimmed = value.strip()
+    if required and not trimmed:
+        raise ValueError(f"{field} is required")
+    return trimmed
+
+
+def _validate_assignee_ids(value: object) -> list[int]:
+    """Return the assignee ids as plain ints, or raise naming the field.
+
+    The argument must be a non-empty list — strictly a ``list``: tuples
+    and other sequences are rejected.  Each entry must be a real int
+    (bools and floats rejected) so SQLite cannot bind ``True`` onto
+    child 1 or a float onto a neighbouring profile.
+    """
+    if type(value) is not list:
+        raise ValueError(
+            f"assignee_child_ids must be a list, got {type(value).__name__}"
+        )
+    if not value:
+        raise ValueError("assignee_child_ids must not be empty")
+    ids: list[int] = []
+    for entry in value:
+        if isinstance(entry, bool) or not isinstance(entry, int):
+            raise ValueError(
+                "assignee_child_ids must contain integer child ids, "
+                f"got {entry!r}"
+            )
+        ids.append(entry)
+    if len(set(ids)) != len(ids):
+        raise ValueError(
+            "assignee_child_ids must not contain duplicate child ids"
+        )
+    return ids
+
+
+def _normalize_windows(
+    value: object,
+) -> list[tuple[str, str | None]]:
+    """Normalize the windows argument to ``(name, due_time)`` pairs.
+
+    The outer argument must be a non-empty list — strictly a ``list``:
+    tuples and other sequences are rejected, while each individual entry
+    may still be a ``(window, due_time)`` tuple.  A bare window name
+    (due time None) is also accepted.  The name must be a
+    :data:`~.const.QUEST_WINDOWS` spelling and a present due time must be
+    a strict 24-hour HH:MM, both validated up front so a malformed
+    window is rejected before any write.
+    """
+    if type(value) is not list:
+        raise ValueError(
+            f"windows must be a list, got {type(value).__name__}"
+        )
+    if not value:
+        raise ValueError("windows must not be empty")
+    normalized: list[tuple[str, str | None]] = []
+    seen: set[str] = set()
+    for index, entry in enumerate(value):
+        if isinstance(entry, str):
+            name: object = entry
+            due_time: object = None
+        elif isinstance(entry, (tuple, list)) and len(entry) == 2:
+            name, due_time = entry[0], entry[1]
+        else:
+            raise ValueError(
+                f"windows[{index}] must be a window name or a "
+                f"(name, due_time) pair, got {entry!r}"
+            )
+        _validate_window(name)
+        if due_time is not None:
+            _validate_time(due_time, "due_time")
+        if name in seen:
+            raise ValueError(
+                f"windows must not contain duplicate window {name!r}"
+            )
+        seen.add(name)
+        normalized.append((name, due_time))
+    return normalized
+
+
+def _validate_definition_id(value: object) -> int:
+    """Reject non-int definition ids (bools included) before any lookup.
+
+    SQLite binds Python bools as integers, so ``True`` would silently
+    address definition 1 and a float would round — malformed service
+    input must never mutate the wrong definition.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"definition_id must be an integer, got {value!r}")
+    return value
+
+
+def _validate_child_id(value: object) -> int:
+    """Reject non-int child ids (bools included) before any lookup.
+
+    SQLite binds Python bools as integers, so ``True`` would silently
+    address child 1 and a float would round — malformed service input
+    must never mutate the wrong profile.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"child_id must be an integer, got {value!r}")
+    return value
+
+
+async def create_quest_definition(
+    database: NestQuestDatabase,
+    title: str,
+    rule: ScheduleRule,
+    assignee_child_ids: list[int],
+    windows: list[str | tuple[str, str | None]],
+    *,
+    description: str | None = None,
+    icon: str | None = None,
+) -> CreatedQuestDefinition:
+    """Create a quest definition (rule + assignees + windows) atomically.
+
+    Validates the arguments up front, then persists the schedule rule,
+    the definition, its assignees and its windows in one transaction via
+    the DAO.  Returns the stored definition together with its decoded
+    rule, assignees and windows — all read back inside that same
+    transaction, so the returned bundle is a consistent creation
+    snapshot.  Raises ValueError on any rejected argument; a rejected or
+    failed create persists nothing.
+    """
+    name = _validate_text(title, "title", required=True)
+    assert name is not None
+    if not isinstance(rule, ScheduleRule):
+        raise ValueError(f"rule must be a ScheduleRule, got {rule!r}")
+    assignee_ids = _validate_assignee_ids(assignee_child_ids)
+    window_specs = _normalize_windows(windows)
+    description_value = _validate_text(description, "description")
+    icon_value = _validate_text(icon, "icon")
+
+    storage = schedule_rule_to_storage(rule)
+    dao = QuestDefinitionsDao(database)
+    try:
+        snapshot = await dao.create_with_rule_and_windows(
+            name,
+            storage,
+            _now_stamp(),
+            assignee_ids,
+            window_specs,
+            description=description_value,
+            icon=icon_value,
+        )
+    except ValueError as error:
+        # The DAO's assignee check names only the child; re-raise so the
+        # public contract names the field the caller actually passed.
+        message = str(error)
+        if message.startswith("child "):
+            raise ValueError(
+                f"assignee_child_ids: {message}"
+            ) from error
+        raise
+    decoded_rule = schedule_rule_from_storage(storage)
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=decoded_rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
+    )
+
+
+async def edit_quest_definition(
+    database: NestQuestDatabase,
+    definition_id: int,
+    *,
+    title: str | object = _UNSET,
+    description: str | None | object = _UNSET,
+    icon: str | None | object = _UNSET,
+    rule: ScheduleRule | object = _UNSET,
+    windows: list[str | tuple[str, str | None]] | object = _UNSET,
+) -> CreatedQuestDefinition:
+    """Edit a quest definition's metadata, rule and windows atomically.
+
+    Arguments default to the sentinel ``_UNSET`` meaning "leave this
+    field alone".  A provided ``title`` is validated like create's
+    (required and trimmed); ``description`` and ``icon`` are optional
+    and may be cleared by passing ``None`` explicitly.  A provided
+    ``rule`` must be an already-validated :class:`~.recurrence.ScheduleRule`
+    and REPLACES the definition's schedule rule IN PLACE — the existing
+    ``schedule_rules`` row is updated, never orphaned or duplicated.  A
+    provided ``windows`` list REPLACES the whole window set (each entry
+    a ``const.QUEST_WINDOWS`` name, optionally with a strict HH:MM due
+    time), matching create's non-empty contract.  Assignment is NOT
+    settable here.  Raises ValueError when the definition does not
+    exist or any argument is rejected.
+
+    Returns the updated definition together with its decoded rule,
+    assignees and windows — all read back inside the DAO's transaction,
+    so the returned bundle is a consistent edit snapshot.  Edits change
+    future instances only; no ``quest_instances`` or
+    ``completion_events`` row is touched.
+    """
+    _validate_definition_id(definition_id)
+
+    title_value: str | None | object = _UNSET
+    if title is not _UNSET:
+        name = _validate_text(title, "title", required=True)
+        assert name is not None
+        title_value = name
+
+    description_value: str | None | object = _UNSET
+    if description is not _UNSET:
+        description_value = _validate_text(description, "description")
+
+    icon_value: str | None | object = _UNSET
+    if icon is not _UNSET:
+        icon_value = _validate_text(icon, "icon")
+
+    rule_storage: ScheduleRuleStorage | object = _UNSET
+    if rule is not _UNSET:
+        if not isinstance(rule, ScheduleRule):
+            raise ValueError(f"rule must be a ScheduleRule, got {rule!r}")
+        rule_storage = schedule_rule_to_storage(rule)
+
+    window_specs: list[tuple[str, str | None]] | object = _UNSET
+    if windows is not _UNSET:
+        window_specs = _normalize_windows(windows)
+
+    dao = QuestDefinitionsDao(database)
+    snapshot = await dao.edit_definition(
+        definition_id,
+        title=title_value,
+        description=description_value,
+        icon=icon_value,
+        rule=rule_storage,
+        windows=window_specs,
+    )
+
+    decoded_rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=decoded_rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
+    )
+
+
+async def assign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+) -> list[ChildRecord]:
+    """Assign ``child_id`` to the definition (idempotent).
+
+    Validates that both ids are plain ints (bools and floats rejected)
+    and that the child exists and is active — the DAO enforces the
+    active/exists check inside its transaction, and also confirms the
+    definition exists.  Assigning an already-assigned child is a no-op
+    (no duplicate row).  Returns the definition's assignees read back
+    inside the DAO's same transaction, a consistent post-assignment
+    roster (a concurrent assign/unassign cannot race it after the
+    write).  Raises ValueError on any rejected argument.  Only the
+    ``quest_definition_assignees`` link is written; existing instances
+    and completion history are never touched.
+    """
+    _validate_definition_id(definition_id)
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    try:
+        return await dao.add_assignee(definition_id, child_id)
+    except ValueError as error:
+        # The DAO names only the child/definition; re-raise so the
+        # public contract names the field the caller actually passed.
+        message = str(error)
+        if message.startswith("child "):
+            raise ValueError(f"child_id: {message}") from error
+        if message.startswith("quest definition "):
+            raise ValueError(f"definition_id: {message}") from error
+        raise
+
+
+async def unassign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+) -> bool:
+    """Unassign ``child_id`` from the definition; True when removed.
+
+    Validates both ids are plain ints (bools and floats rejected) and
+    that the definition exists, then removes the (definition, child)
+    link.  Returns True when a row was removed and False when the child
+    was not currently assigned.  Raises ValueError on a rejected
+    argument.  Only the ``quest_definition_assignees`` link is written;
+    existing instances and completion history are never touched.
+    """
+    _validate_definition_id(definition_id)
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    definition = await dao.get(definition_id)
+    if definition is None:
+        raise ValueError(
+            f"definition_id: quest definition {definition_id} does not exist"
+        )
+    return await dao.remove_assignee(definition_id, child_id)
+
+
+async def set_quest_definition_active(
+    database: NestQuestDatabase,
+    definition_id: int,
+    is_active: bool,
+) -> CreatedQuestDefinition:
+    """Deactivate or reactivate a definition; returns the updated view.
+
+    Flipping the flag is the only removal path (Feature 06 guardrail):
+    a deactivated definition stops future instance generation but its
+    row, schedule rule, assignees and windows all survive, and existing
+    instances plus completion history are never touched.  Reactivation
+    re-enables future generation; generating or removing instances is
+    Feature 07 and does not happen here.
+
+    ``definition_id`` must be a plain int (bools and floats rejected,
+    since SQLite would bind ``True`` onto definition 1) and
+    ``is_active`` must be a real bool — the DAO's ``int()`` would
+    otherwise silently coerce strings and numerics ("1", 0) into a
+    state the caller never asked for.  Raises ValueError naming the
+    offending field when the id is malformed, ``is_active`` is not a
+    bool, or the definition does not exist.
+
+    Atomicity: the existence check, the UPDATE and the readback of the
+    definition, its rule, assignees and windows all run inside the
+    connection-scoped lock, so the returned bundle is a consistent
+    snapshot that a concurrent mutation cannot race.
+    """
+    _validate_definition_id(definition_id)
+    if not isinstance(is_active, bool):
+        raise ValueError(f"is_active must be a real bool, got {is_active!r}")
+    async with _connection_lock(database):
+        dao = QuestDefinitionsDao(database)
+        if await dao.get(definition_id) is None:
+            raise ValueError(
+                f"definition_id: quest definition {definition_id} "
+                "does not exist"
+            )
+        await dao.set_active(definition_id, is_active)
+        updated = await dao.get(definition_id)
+        assert updated is not None
+        rule_record = await ScheduleRulesDao(database).get(
+            updated.schedule_rule_id
+        )
+        assert rule_record is not None
+        assignees = await dao.list_assignees(definition_id)
+        windows = await dao.list_windows(definition_id)
+    rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(rule_record)
+    )
+    return CreatedQuestDefinition(
+        definition=updated,
+        rule=rule,
+        assignees=assignees,
+        windows=windows,
+    )

tests/test_dao_presence.py

@@ -619,4 +619,173 @@ def test_dao_presence_test_file_uses_dao_not_raw_table_sql() -> None:
     )
     assert sql_pattern.search(remaining) is None, (
         "tests must go through the DAO, not raw SQL, for these tables"
-    )
+    )
+
+# ---------------------------------------------------------------------------
+# Presence overrides: overlap rejection naming the conflict
+# ---------------------------------------------------------------------------
+
+
+def test_override_create_round_trip_single_and_range(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        single = await overrides.create(
+            child.id, "2026-07-20", "2026-07-20", False
+        )
+        assert (single.start_date, single.end_date) == ("2026-07-20",) * 2
+        span = await overrides.create(
+            child.id, "2026-08-01", "2026-08-05", True, note="traded week"
+        )
+        assert span.is_present is True
+        assert span.note == "traded week"
+        # Distinct, non-overlapping ranges coexist.
+        listed = await overrides.list_by_child_and_range(
+            child.id, "2026-01-01", "2026-12-31"
+        )
+        assert [row.id for row in listed] == [single.id, span.id]
+        return listed
+
+    _with_db(tmp_path, "override-round-trip.db")(_body)
+
+
+def test_override_overlap_rejected_and_names_the_conflict(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        existing = await overrides.create(
+            child.id, "2026-07-20", "2026-07-24", False, note="holiday"
+        )
+        for start, end in (
+            ("2026-07-24", "2026-07-26"),  # touches the existing end
+            ("2026-07-18", "2026-07-20"),  # touches the existing start
+            ("2026-07-21", "2026-07-22"),  # fully inside
+            ("2026-07-01", "2026-07-30"),  # fully covering
+        ):
+            with pytest.raises(ValueError, match="overlaps existing"):
+                await overrides.create(
+                    child.id, start, end, True, note="swap"
+                )
+        # The error NAMES the conflicting override (id + its range).
+        with pytest.raises(
+            ValueError,
+            match=(
+                rf"override {existing.id} "
+                rf"\[2026-07-20, 2026-07-24\].*is_present=False"
+            ),
+        ):
+            await overrides.create(child.id, "2026-07-22", "2026-07-22", True)
+        # Nothing was written by any rejected attempt.
+        listed = await overrides.list_by_child_and_range(
+            child.id, "2026-01-01", "2026-12-31"
+        )
+        assert [row.id for row in listed] == [existing.id]
+        return existing
+
+    _with_db(tmp_path, "override-overlap.db")(_body)
+
+
+def test_override_overlap_is_per_child(tmp_path) -> None:
+    """Different children's overrides may cover the same dates: the
+    conflict rule is per child."""
+    async def _body(database, schedules, overrides, children, child):
+        other = await children.create("Bo", NOW)
+        await overrides.create(child.id, "2026-07-20", "2026-07-24", False)
+        twin = await overrides.create(
+            other.id, "2026-07-20", "2026-07-24", True
+        )
+        assert twin.child_id == other.id
+        listed = await overrides.list_by_child_and_range(
+            other.id, "2026-01-01", "2026-12-31"
+        )
+        assert [row.id for row in listed] == [twin.id]
+        return twin
+
+    _with_db(tmp_path, "override-per-child.db")(_body)
+
+
+def test_override_remove_then_recreate_overlapping(tmp_path) -> None:
+    async def _body(database, schedules, overrides, children, child):
+        first = await overrides.create(
+            child.id, "2026-07-20", "2026-07-24", False
+        )
+        assert await overrides.delete(first.id) is True
+        assert await overrides.delete(first.id) is False
+        replacement = await overrides.create(
+            child.id, "2026-07-20", "2026-07-24", True, note="swap back"
+        )
+        assert replacement.is_present is True
+        return replacement
+
+    _with_db(tmp_path, "override-recreate.db")(_body)
+
+
+def test_override_create_racing_conflicts_serialize(tmp_path) -> None:
+    """Two racing creates for the same child and dates: exactly one
+    lands, the other is rejected with the overlap error.  DETERMINISTIC
+    gate: task A is paused at its INSERT — inside its open transaction,
+    after its overlap probe — so task B is created while A provably
+    holds the connection lock and must queue; B then re-runs the probe
+    against A's committed row and is rejected."""
+    import asyncio as asyncio_module
+    from unittest.mock import MagicMock
+
+    async def _main():
+        from custom_components.nestquest.dao_children import ChildrenDao
+        from custom_components.nestquest.dao_presence import (
+            PresenceOverridesDao,
+        )
+        from custom_components.nestquest.db import NestQuestDatabase
+        from custom_components.nestquest.migrations import apply_migrations
+
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio_module.Event()
+        a_started = asyncio_module.Event()
+
+        hass = MagicMock()
+
+        async def _gated_executor(fn, *args):
+            if armed["active"]:
+                # Gate the FIRST _execute after arming: task A's INSERT,
+                # which runs inside its open transaction (BEGIN already
+                # executed) after the overlap probe.
+                if getattr(fn, "__name__", "") == "_execute" 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 / "override-race.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            overrides = PresenceOverridesDao(database)
+            child = await children.create("Ada", NOW)
+
+            armed["active"] = True
+            task_a = asyncio_module.ensure_future(
+                overrides.create(child.id, "2026-07-20", "2026-07-24", True)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio_module.ensure_future(
+                overrides.create(child.id, "2026-07-22", "2026-07-26", False)
+            )
+            await asyncio_module.sleep(0)
+            gate_open.set()
+            record_a, error_b = await asyncio_module.gather(
+                task_a, task_b, return_exceptions=True
+            )
+            assert isinstance(record_a, object) and not isinstance(
+                record_a, BaseException
+            ), f"A must land: {record_a!r}"
+            assert isinstance(error_b, ValueError), (
+                f"B must be rejected with the overlap error: {error_b!r}"
+            )
+            assert "overlaps existing" in str(error_b)
+            listed = await overrides.list_by_child_and_range(
+                child.id, "2026-01-01", "2026-12-31"
+            )
+            assert [row.id for row in listed] == [record_a.id]
+        finally:
+            await database.close()
+
+    asyncio.new_event_loop().run_until_complete(_main())

tests/test_dao_rules.py

@@ -9,13 +9,21 @@ import pytest
 from custom_components.nestquest.dao_children import ChildrenDao
 from custom_components.nestquest.dao_rules import (
     ScheduleRuleRecord,
+    ScheduleRuleStorage,
     ScheduleRulesDao,
     QuestDefinitionRecord,
     QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
+    schedule_rule_from_storage,
+    schedule_rule_to_storage,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
+from custom_components.nestquest.recurrence import (
+    RuleType,
+    RuleValidationError,
+    ScheduleRule,
+)
 
 
 def _run(coro):
@@ -423,6 +431,249 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
     assert create_won is False
 
 
+# ---------------------------------------------------------------------------
+# ScheduleRule <-> storage mapping
+# ---------------------------------------------------------------------------
+
+
+_RULE_STORAGE_CASES = [
+    (
+        ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01"),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.DAILY,
+            interval=3,
+            start_date="2026-09-01",
+            end_date="2026-12-31",
+        ),
+        ScheduleRuleStorage(
+            "daily", 3, None, None, None, None, "2026-09-01", "2026-12-31"
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2, 0, 4},
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2,4", None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.CUSTOM_DAYS,
+            weekday_set={1, 3, 5},
+            interval=2,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "custom", 2, "1,3,5", None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_DAY,
+            day_of_month=15,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_DAY,
+            day_of_month=31,
+            interval=3,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 3, None, 31, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_WEEKDAY,
+            nth_weekday=2,
+            nth_weekday_weekday=1,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, "1", None, 2, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_WEEKDAY,
+            nth_weekday=-1,
+            nth_weekday_weekday=4,
+            interval=2,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 2, "4", None, -1, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01"
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, None, None, 6, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.YEARLY,
+            month=2,
+            day_of_month=29,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, 29, None, 2, "2026-09-01", None
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_to_storage_maps_each_type(rule, storage) -> None:
+    assert schedule_rule_to_storage(rule) == storage
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_from_storage_rebuilds_each_type(rule, storage) -> None:
+    assert schedule_rule_from_storage(storage) == rule
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_round_trips_losslessly_both_directions(
+    rule, storage
+) -> None:
+    assert schedule_rule_from_storage(schedule_rule_to_storage(rule)) == rule
+    assert schedule_rule_to_storage(schedule_rule_from_storage(storage)) == storage
+
+
+@pytest.mark.parametrize(
+    "storage",
+    [
+        # monthly with neither day_of_month nor nth_weekday
+        ScheduleRuleStorage(
+            "monthly", 1, None, None, None, None, "2026-09-01", None
+        ),
+        # monthly with both day_of_month and nth_weekday
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, 2, None, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY whose weekday_set is None
+        ScheduleRuleStorage(
+            "monthly", 1, None, None, 2, None, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY whose weekday_set is empty
+        ScheduleRuleStorage("monthly", 1, "", None, 2, None, "2026-09-01", None),
+        # MONTHLY_WEEKDAY whose weekday_set is multi-element
+        ScheduleRuleStorage(
+            "monthly", 1, "1,3", None, 2, None, "2026-09-01", None
+        ),
+        # unknown storage rule_type
+        ScheduleRuleStorage(
+            "sometimes", 1, None, None, None, None, "2026-09-01", None
+        ),
+    ],
+)
+def test_schedule_rule_from_storage_rejects_ambiguous_or_invalid(
+    storage,
+) -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(storage)
+
+
+def test_schedule_rule_from_storage_rejects_malformed_weekday_csv() -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "weekly", 1, "0,2,x", None, None, None, "2026-09-01", None
+            )
+        )
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "weekly", 1, "7", None, None, None, "2026-09-01", None
+            )
+        )
+
+
+@pytest.mark.parametrize(
+    "storage",
+    [
+        # DAILY carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "daily", 1, "0", None, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, 2, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, None, 6, "2026-09-01", None
+        ),
+        # WEEKLY carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2", None, None, 6, "2026-09-01", None
+        ),
+        # CUSTOM_DAYS carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "custom", 1, "0,2", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "custom", 1, "0,2", None, None, 6, "2026-09-01", None
+        ),
+        # MONTHLY_DAY carrying weekday_set (folded-only column) or month.
+        ScheduleRuleStorage(
+            "monthly", 1, "1", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, None, 6, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY carrying a forbidden month.
+        ScheduleRuleStorage(
+            "monthly", 1, "1", None, 2, 6, "2026-09-01", None
+        ),
+        # YEARLY carrying weekday_set or nth_weekday.
+        ScheduleRuleStorage(
+            "yearly", 1, "0", None, None, 6, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, None, 2, 6, "2026-09-01", None
+        ),
+    ],
+)
+def test_schedule_rule_from_storage_rejects_forbidden_fields(storage) -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(storage)
+
+
+def test_schedule_rule_from_storage_rejects_duplicate_weekday_token() -> None:
+    """A MONTHLY_WEEKDAY with a duplicate CSV token ("1,1") must raise,
+    never silently canonicalize to a single weekday on re-storage."""
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "monthly", 1, "1,1", None, 2, None, "2026-09-01", None
+            )
+        )
+
+
 # ---------------------------------------------------------------------------
 # quest_definitions: create / get
 # ---------------------------------------------------------------------------

tests/test_presence.py

@@ -7,6 +7,7 @@ import pytest
 
 from custom_components.nestquest.presence import (
     MAX_CYCLE_LENGTH_WEEKS,
+    PresenceEngine,
     PresenceOverride,
     PresenceSchedule,
     cycle_week_index,
@@ -371,3 +372,443 @@ def test_module_uses_no_iso_week_functions() -> None:
     ).read_text()
     assert "isocalendar" not in source
     assert "isoweek" not in source
+
+
+# ---------------------------------------------------------------------------
+# PresenceEngine: is_present with the always-present default
+# ---------------------------------------------------------------------------
+
+
+def _three_child_engine() -> PresenceEngine:
+    """Two children on opposite weeks of one two-week rotation, one
+    with no schedule at all (the always-present sibling)."""
+    anchor = datetime.date(2026, 1, 5)  # Monday, week 0
+    declan = PresenceSchedule(
+        1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
+    )
+    jordyn = PresenceSchedule(
+        2, 2, anchor, {0: set(), 1: {0, 1, 2, 3, 4, 5, 6}}
+    )
+    return PresenceEngine({1: declan, 2: jordyn})
+
+
+def test_engine_child_without_schedule_is_always_present() -> None:
+    """Every day of the four-week span, all three children get the
+    expected verdict: Chloe (no schedule) always present; the two
+    rotation children complementary per the anchor pattern."""
+    engine = _three_child_engine()
+    anchor = datetime.date(2026, 1, 5)
+    for offset in range(28):
+        day = anchor + datetime.timedelta(days=offset)
+        week = ((day - anchor).days // 7) % 2
+        assert engine.is_present(3, day) is True, f"offset {offset}: Chloe"
+        assert engine.is_present(1, day) is (week == 0), (
+            f"offset {offset}: Declan should follow week {week}"
+        )
+        assert engine.is_present(2, day) is (week == 1), (
+            f"offset {offset}: Jordyn should follow week {week}"
+        )
+
+
+def test_engine_alternating_children_are_complementary_for_four_weeks() -> None:
+    """Every day of a four-week span: the two rotation children swap
+    presence exactly per the anchor pattern — when one is home the
+    other is away."""
+    engine = _three_child_engine()
+    anchor = datetime.date(2026, 1, 5)
+    for offset in range(28):
+        day = anchor + datetime.timedelta(days=offset)
+        d = engine.is_present(1, day)
+        j = engine.is_present(2, day)
+        assert d == (not j), f"offset {offset}: {d}/{j} not complementary"
+        expected_week = ((day - anchor).days // 7) % 2
+        expected_present = day.weekday() in {0, 1, 2, 3, 4, 5, 6}
+        assert engine.is_present(1, day) == (expected_week == 0), (
+            f"offset {offset}: Declan week {expected_week}"
+        )
+    # Spot-check the swap days.
+    assert engine.is_present(1, anchor) is True
+    assert engine.is_present(2, anchor) is False
+    assert engine.is_present(1, anchor + datetime.timedelta(days=7)) is False
+    assert engine.is_present(2, anchor + datetime.timedelta(days=7)) is True
+
+
+def test_engine_weekend_absence_respects_the_pattern() -> None:
+    """A weekday-set pattern excludes days outside the set even in the
+    child's present week."""
+    engine = PresenceEngine(
+        {
+            1: PresenceSchedule(
+                1, 2, datetime.date(2026, 1, 5), {0: {1, 3}, 1: {2, 4}}
+            )
+        }
+    )
+    # 2026-01-06 (Tue) is week 0, weekday 1 -> present.
+    assert engine.is_present(1, datetime.date(2026, 1, 6)) is True
+    # 2026-01-07 (Wed) is week 0, weekday 2 -> absent.
+    assert engine.is_present(1, datetime.date(2026, 1, 7)) is False
+    # 2026-01-14 (Wed) is week 1, weekday 2 -> present.
+    assert engine.is_present(1, datetime.date(2026, 1, 14)) is True
+
+
+def test_engine_rejects_bad_construction_and_inputs() -> None:
+    schedule = PresenceSchedule(1, 1, ANCHOR, {0: {0}})
+    with pytest.raises(ValueError, match="must map child_id"):
+        PresenceEngine([schedule])
+    with pytest.raises(ValueError, match="must be a PresenceSchedule"):
+        PresenceEngine({1: "not-a-schedule"})
+    with pytest.raises(ValueError, match="holds the schedule"):
+        PresenceEngine({2: schedule})
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        PresenceEngine({True: schedule})
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        PresenceEngine({1.0: PresenceSchedule(1, 1, ANCHOR, {0: {0}})})
+    engine = PresenceEngine({1: schedule})
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        engine.is_present(True, ANCHOR)
+    with pytest.raises(ValueError, match="plain calendar date"):
+        engine.is_present(1, datetime.datetime(2026, 1, 5))
+
+
+def test_engine_is_immutable() -> None:
+    engine = _three_child_engine()
+    with pytest.raises(AttributeError, match="immutable snapshot"):
+        engine._schedules = {}
+    with pytest.raises(AttributeError, match="immutable snapshot"):
+        engine.something_new = 1
+
+
+# ---------------------------------------------------------------------------
+# Overrides take precedence over the pattern
+# ---------------------------------------------------------------------------
+
+
+def _engine_with_overrides(*overrides_for_declan) -> PresenceEngine:
+    anchor = datetime.date(2026, 1, 5)  # Monday
+    return PresenceEngine(
+        {
+            1: PresenceSchedule(
+                1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
+            )
+        },
+        overrides={1: list(overrides_for_declan)} if overrides_for_declan else {},
+    )
+
+
+def test_absent_override_on_normally_present_day_wins() -> None:
+    # 2026-01-06 is Tuesday of Declan's week 0: normally present.
+    engine = _engine_with_overrides(
+        PresenceOverride(1, "2026-01-06", "2026-01-06", False, note="away")
+    )
+    assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
+    # The pattern resumes outside the override.
+    assert engine.is_present(1, datetime.date(2026, 1, 7)) is True
+
+
+def test_present_override_on_normally_absent_day_wins() -> None:
+    # 2026-01-13 is in Declan's week 1 (away all week); a present
+    # override on the Tuesday of week 1 brings him home for that day.
+    engine = _engine_with_overrides(
+        PresenceOverride(1, "2026-01-13", "2026-01-13", True, note="trade")
+    )
+    assert engine.is_present(1, datetime.date(2026, 1, 13)) is True
+    assert engine.is_present(1, datetime.date(2026, 1, 14)) is False
+
+
+def test_override_applies_without_a_schedule() -> None:
+    """An override on a schedule-free child still applies: holidays and
+    swaps must work for the always-present sibling too."""
+    engine = PresenceEngine(
+        {},
+        overrides={
+            3: [
+                PresenceOverride(
+                    3, "2026-07-20", "2026-07-24", False, note="holiday"
+                )
+            ]
+        },
+    )
+    assert engine.is_present(3, datetime.date(2026, 7, 22)) is False
+    assert engine.is_present(3, datetime.date(2026, 7, 19)) is True
+    assert engine.is_present(3, datetime.date(2026, 7, 25)) is True
+
+
+def test_override_range_beats_pattern_for_its_span() -> None:
+    engine = _engine_with_overrides(
+        PresenceOverride(1, "2026-01-13", "2026-01-17", True, note="traded week")
+    )
+    for offset in range(12, 17):
+        day = datetime.date(2026, 1, 1) + datetime.timedelta(days=offset)
+        assert engine.is_present(1, day) is True, f"{day}"
+    # Days outside the span stay on the pattern (week 1: absent).
+    assert engine.is_present(1, datetime.date(2026, 1, 12)) is False
+    assert engine.is_present(1, datetime.date(2026, 1, 18)) is False
+
+
+def test_engine_override_snapshot_validation() -> None:
+    schedule = PresenceSchedule(1, 1, ANCHOR, {0: {0}})
+    entry = PresenceOverride(1, "2026-07-20", "2026-07-20", True)
+    with pytest.raises(ValueError, match="must map child_id"):
+        PresenceEngine({1: schedule}, overrides=[entry])
+    with pytest.raises(ValueError, match="must be a list"):
+        PresenceEngine({1: schedule}, overrides={1: entry})
+    with pytest.raises(ValueError, match="must be"):
+        PresenceEngine({1: schedule}, overrides={1: ["nope"]})
+    with pytest.raises(ValueError, match="holds an override"):
+        PresenceEngine({1: schedule}, overrides={2: [entry]})
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        PresenceEngine({1: schedule}, overrides={True: [entry]})
+
+
+def test_engine_overlapping_snapshot_uses_latest_start() -> None:
+    """The DAO layer refuses overlapping overrides per child; if a
+    snapshot carries them anyway (hand-built), the latest start date
+    wins deterministically."""
+    engine = _engine_with_overrides(
+        PresenceOverride(1, "2026-01-06", "2026-01-10", False),
+        PresenceOverride(1, "2026-01-08", "2026-01-12", True),
+    )
+    # Jan 6-7: only the earlier override covers -> absent.
+    assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
+    # Jan 8-10: both cover; the LATER start (present) wins.
+    assert engine.is_present(1, datetime.date(2026, 1, 9)) is True
+    # Jan 11-12: only the later override covers -> present.
+    assert engine.is_present(1, datetime.date(2026, 1, 11)) is True
+
+
+def test_engine_override_lists_are_frozen() -> None:
+    """The overrides snapshot must be truly read-only: the engine is
+    built directly on the CALLER's list, so mutating that list after
+    construction would change answers under a shallow copy — under the
+    frozen snapshot it must not."""
+    entry = PresenceOverride(1, "2026-01-06", "2026-01-06", False)
+    entries = [entry]
+    schedule = PresenceSchedule(
+        1, 2, ANCHOR, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
+    )
+    engine = PresenceEngine(
+        {1: schedule}, overrides={1: entries}  # caller-owned list
+    )
+    assert engine.is_present(1, datetime.date(2026, 1, 6)) is False
+    entries.append(PresenceOverride(1, "2026-01-07", "2026-01-07", False))
+    # The late append must NOT change the engine's answers...
+    assert engine.is_present(1, datetime.date(2026, 1, 7)) is True
+    # Tuples have no append at all (AttributeError), and the mapping
+    # itself refuses reassignment.
+    with pytest.raises(AttributeError):
+        engine._overrides[1].append(entry)
+    with pytest.raises(AttributeError, match="immutable snapshot"):
+        engine._overrides = {}
+
+
+# ---------------------------------------------------------------------------
+# next_present_dates: the parent's pattern preview
+# ---------------------------------------------------------------------------
+
+
+def _preview_engine() -> PresenceEngine:
+    """Declan: home all of week 0, away all of week 1 (anchor
+    2026-01-05, a Monday); an away override Tue Jan 13 (his away week)
+    would matter only for the override test."""
+    return PresenceEngine(
+        {
+            1: PresenceSchedule(
+                1, 2, datetime.date(2026, 1, 5),
+                {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()},
+            )
+        }
+    )
+
+
+def test_preview_returns_the_next_present_days_of_a_known_pattern() -> None:
+    engine = _preview_engine()
+    # Starting Monday Jan 5 (present week 0): five straight days.
+    preview = engine.next_present_dates(1, "2026-01-05", 5)
+    assert preview == [
+        datetime.date(2026, 1, 5),
+        datetime.date(2026, 1, 6),
+        datetime.date(2026, 1, 7),
+        datetime.date(2026, 1, 8),
+        datetime.date(2026, 1, 9),
+    ]
+    # Starting mid-week: the rest of week 0 first, then the absent
+    # week 1 is skipped and the next present week picks up.
+    preview = engine.next_present_dates(1, "2026-01-07", 3)
+    assert preview == [
+        datetime.date(2026, 1, 7),
+        datetime.date(2026, 1, 8),
+        datetime.date(2026, 1, 9),
+    ]
+    preview = engine.next_present_dates(1, "2026-01-07", 6)
+    # Week 0 runs Mon Jan 5 .. Sun Jan 11 (all present); the away week
+    # (Jan 12-18) is skipped; week 2 resumes Jan 19.
+    assert preview == [
+        datetime.date(2026, 1, 7),
+        datetime.date(2026, 1, 8),
+        datetime.date(2026, 1, 9),
+        datetime.date(2026, 1, 10),
+        datetime.date(2026, 1, 11),
+        datetime.date(2026, 1, 19),
+    ]
+
+
+def test_preview_start_date_is_inclusive() -> None:
+    engine = _preview_engine()
+    # Jan 12 (Monday of the away week) is absent: the first present day
+    # is Jan 19 — the start date itself when present.
+    preview = engine.next_present_dates(1, "2026-01-19", 1)
+    assert preview == [datetime.date(2026, 1, 19)]
+
+
+def test_preview_honours_overrides() -> None:
+    schedule = PresenceSchedule(
+        1, 2, datetime.date(2026, 1, 5),
+        {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()},
+    )
+    engine = PresenceEngine(
+        {1: schedule},
+        overrides={
+            1: [
+                PresenceOverride(
+                    1, "2026-01-05", "2026-01-06", False, note="holiday"
+                )
+            ]
+        },
+    )
+    # Jan 5-6 blocked by the override: the preview starts Jan 7.
+    preview = engine.next_present_dates(1, "2026-01-04", 3)
+    assert preview == [
+        datetime.date(2026, 1, 7),
+        datetime.date(2026, 1, 8),
+        datetime.date(2026, 1, 9),
+    ]
+
+
+def test_preview_schedule_free_child_returns_consecutive_days() -> None:
+    engine = PresenceEngine({})
+    preview = engine.next_present_dates(3, "2026-03-01", 3)
+    assert preview == [
+        datetime.date(2026, 3, 1),
+        datetime.date(2026, 3, 2),
+        datetime.date(2026, 3, 3),
+    ]
+
+
+def test_preview_rejects_bad_inputs() -> None:
+    engine = _preview_engine()
+    with pytest.raises(ValueError, match="count must be an integer"):
+        engine.next_present_dates(1, "2026-01-05", "3")
+    with pytest.raises(ValueError, match="count must be at least 1"):
+        engine.next_present_dates(1, "2026-01-05", 0)
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        engine.next_present_dates(True, "2026-01-05", 1)
+    with pytest.raises(ValueError, match="plain calendar date"):
+        engine.next_present_dates(1, datetime.datetime(2026, 1, 5), 1)
+
+
+def test_preview_always_absent_child_raises_not_loops() -> None:
+    """A child absent every scanned day must surface as an error, not
+    hang the admin preview."""
+    engine = PresenceEngine(
+        {
+            1: PresenceSchedule(
+                1, 1, datetime.date(2026, 1, 5), {0: set()}
+            )
+        }
+    )
+    with pytest.raises(ValueError, match="not present on any"):
+        engine.next_present_dates(1, "2026-01-05", 2)
+
+
+# ---------------------------------------------------------------------------
+# The week-parity regression guard (Feature 05, D-004)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("year", "length", "reason"),
+    [
+        (2020, 366, "leap year, 53 ISO weeks"),
+        (2015, 365, "365 days, 53 ISO weeks"),
+    ],
+)
+def test_custody_rotation_walks_full_years_without_inversion(
+    year, length, reason
+) -> None:
+    """REGRESSION GUARD for the week-parity bug (D-004).
+
+    Walks a two-week alternating rotation day by day across a FULL
+    {length}-day year ({year} — {reason}) plus the following January,
+    asserting the present/absent sequence never breaks or inverts at
+    any point, including January 1st.  ISO week parity silently flips
+    at the ISO-week rollover (the Monday on or after January 4th —
+    January 4th itself in both of these years, which end ISO week 53)
+    and would invert every custody schedule there; anchor-date
+    arithmetic must not.
+    """.format(length=length, year=year, reason=reason)
+    if year == 2020:
+        assert datetime.date(2020, 12, 31).isocalendar()[1] == 53
+        assert (datetime.date(2021, 1, 1) - datetime.date(2020, 1, 1)).days == 366
+    if year == 2015:
+        assert datetime.date(2015, 12, 31).isocalendar()[1] == 53
+        assert (datetime.date(2016, 1, 1) - datetime.date(2015, 1, 1)).days == 365
+
+    # The first Monday of January (2020: Jan 6; 2015: Jan 5) — an
+    # on-week anchor, whatever the year.
+    anchor = datetime.date(year, 1, 1) + datetime.timedelta(
+        days=(7 - datetime.date(year, 1, 1).weekday()) % 7
+    )
+    assert anchor.weekday() == 0
+    rotation = PresenceSchedule(
+        1, 2, anchor, {0: {0, 1, 2, 3, 4, 5, 6}, 1: set()}
+    )
+    engine = PresenceEngine({1: rotation})
+
+    # Walk EVERY day of the year.  The expectation is derived from the
+    # anchor's 7-day BLOCKS, not from the engine: each block of seven
+    # days starting at the anchor is uniformly present (even block
+    # index) or uniformly absent (odd block index), and consecutive
+    # blocks alternate.  A parity-based implementation resets on
+    # January 1st and breaks either the uniformity of a block or the
+    # alternation between them — most visibly in the block spanning
+    # December 31st -> January 1st of a 53-ISO-week year.
+    # Walk the FULL year AND the first two weeks of the following
+    # January: the block spanning December 31st -> January 1st must
+    # stay uniform ACROSS the boundary, so an implementation that
+    # resets when the target year differs from the anchor's fails
+    # here (the post-boundary days of its final block disagree).
+    first_day = datetime.date(year, 1, 1)
+    last_day = datetime.date(year + 1, 1, 14)
+    days_seen = 0
+    day = first_day
+    while day <= last_day:
+        days_since_anchor = (day - anchor).days
+        block = days_since_anchor // 7  # which 7-day block of the cycle
+        block_start = anchor + datetime.timedelta(days=block * 7)
+        # Uniformity within the block: every day of this 7-day block
+        # must answer exactly like the block's first day.
+        expected = ((block % 2) == 0)
+        assert engine.is_present(1, day) is expected, (
+            f"{day} (block {block}, day offset "
+            f"{(day - block_start).days}): present="
+            f"{engine.is_present(1, day)}, expected {expected}"
+        )
+        # The block's OTHER days must agree — including the days on
+        # the far side of January 1st.
+        for offset in range(7):
+            block_day = block_start + datetime.timedelta(days=offset)
+            if first_day <= block_day <= last_day:
+                assert engine.is_present(1, block_day) is expected, (
+                    f"{block_day} disagrees with its block {block}"
+                )
+        days_seen += 1
+        day += datetime.timedelta(days=1)
+    # The walked year itself contributed exactly its full length.
+    year_days = sum(
+        1
+        for offset in range((last_day - first_day).days + 1)
+        if (first_day + datetime.timedelta(days=offset)).year == year
+    )
+    assert year_days == length, (
+        f"walked {year_days} days of {year}, expected {length}"
+    )