fix: regenerate_for_child is child-scoped (no cross-child refresh)

custom_components/nestquest/dao_instances.py

@@ -47,6 +47,18 @@ from .db import NestQuestDatabase
 _UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
 
 
+def _today() -> datetime.date:
+    """Return the current local calendar date (the single date source).
+
+    Every "today" read in this module goes through this one function so
+    the delete clamp and the materialize walk's no-past guard read the
+    same clock, and a caller or test can pin/roll that clock
+    deterministically (e.g. a queued-across-midnight regression).  Pure
+    clock read — no business rule lives here.
+    """
+    return datetime.date.today()
+
+
 def _validate_utc_timestamp(value: str, field: str) -> None:
     """Raise ValueError unless ``value`` is a strict UTC ISO-8601 stamp.
 
@@ -203,7 +215,7 @@ class QuestInstancesDao:
         # check runs after the lock is acquired, because a caller can
         # wait on the lock across midnight — a date that was "today"
         # at call time may be "yesterday" by the time the INSERT runs.
-        today = datetime.date.today().isoformat()
+        today = _today().isoformat()
         if due_date < today:
             raise ValueError(
                 f"instances are never generated in the past: due_date "
@@ -214,7 +226,7 @@ class QuestInstancesDao:
                 # Re-read "today" under the lock: this is the date the
                 # insert actually executes on, so the no-past rule
                 # holds across midnight rollovers.
-                today = datetime.date.today().isoformat()
+                today = _today().isoformat()
                 if due_date < today:
                     raise ValueError(
                         f"instances are never generated in the past: "
@@ -277,6 +289,121 @@ class QuestInstancesDao:
         assert instance is not None
         return instance
 
+    async def upsert_if_valid(
+        self,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        generated_at: str,
+        *,
+        window: str,
+        due_time: str | None = None,
+    ) -> QuestInstanceRecord | None:
+        """Insert one dated instance iff its tuple is still materializable.
+
+        This is the materialization walk's single write path.  Unlike the
+        generic :meth:`upsert`, which validates only existence and
+        assignment, this method re-validates EVERY materialization
+        precondition in the SAME connection-lock-held transaction as the
+        INSERT — closing the check-then-insert gap a concurrent config
+        write could otherwise slip through:
+
+        - the definition exists AND is active (a definition deactivated
+          after the input snapshot is skipped, never written);
+        - the child exists AND is active (a deactivated child is skipped);
+        - the (definition, child) assignment link still exists (a removed
+          assignment is skipped);
+        - the window is still declared on the definition (a removed window
+          is skipped).
+
+        The live checks above ONLY decide whether the tuple is still
+        valid; they never read the window's ``due_time``.  The instance is
+        written with the caller-supplied ``due_time`` — the value snapshotted
+        from the coherent input read — so a window-time edit landing
+        mid-walk cannot leak a newer time into a batch that was already
+        snapshotted against older rule/presence/assignment decisions.
+
+        Returns the stored instance, or ``None`` when a precondition
+        failed — the tuple no longer materializable, which the walk SKIPS
+        rather than raising, so the rest of the batch is unaffected.  The
+        no-past rule (instances are never generated in the past) is
+        preserved from :meth:`upsert` and still raises ValueError.  An
+        instance that ALREADY has a completion event is also skipped
+        (returns ``None``) rather than raised: it is immutable, so the
+        walk leaves it untouched instead of failing the batch — the
+        generic :meth:`upsert` keeps its hard refusal for the
+        delete/regenerate path.
+        """
+        _validate_date(due_date, "due_date")
+        _validate_window(window)
+        # Fail fast on obviously-past dates before queuing on the lock;
+        # the authoritative check re-runs under the lock (see upsert).
+        today = _today().isoformat()
+        if due_date < today:
+            raise ValueError(
+                f"instances are never generated in the past: due_date "
+                f"{due_date!r} is before today {today!r}"
+            )
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                today = _today().isoformat()
+                if due_date < today:
+                    raise ValueError(
+                        f"instances are never generated in the past: "
+                        f"due_date {due_date!r} is before today {today!r}"
+                    )
+                still_valid = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definitions d "
+                    "JOIN quest_definition_assignees a "
+                    "ON a.definition_id = d.id "
+                    "JOIN children c ON c.id = a.child_id "
+                    "JOIN quest_definition_windows w "
+                    "ON w.definition_id = d.id "
+                    "WHERE d.id = ? AND a.child_id = ? AND w.window = ? "
+                    "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1",
+                    (definition_id, child_id, window),
+                )
+                if still_valid is None:
+                    # A config change invalidated the tuple since the
+                    # snapshot: skip it rather than abort the batch.
+                    return None
+                existing = await self._database.fetch_one(
+                    "SELECT 1 FROM completion_events "
+                    "WHERE instance_id = (SELECT id FROM quest_instances "
+                    "WHERE definition_id = ? AND child_id = ? "
+                    "AND due_date = ? AND window = ?)",
+                    (definition_id, child_id, due_date, window),
+                )
+                if existing is not None:
+                    # The instance is already completed and therefore
+                    # immutable: the walk SKIPS it (no-op) rather than
+                    # rewriting its snapshot columns or failing the
+                    # batch.  This is the materialization walk's skip,
+                    # distinct from the generic :meth:`upsert` refusal.
+                    return None
+                await self._database.execute(
+                    "INSERT INTO quest_instances (definition_id, child_id, "
+                    "window, due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
+                    "window) DO UPDATE SET "
+                    "due_time = excluded.due_time, "
+                    "generated_at = excluded.generated_at",
+                    (
+                        definition_id,
+                        child_id,
+                        window,
+                        due_date,
+                        due_time,
+                        generated_at,
+                    ),
+                )
+                instance = await self.get(
+                    definition_id, child_id, due_date, window
+                )
+        assert instance is not None
+        return instance
+
     async def get(
         self,
         definition_id: int,
@@ -360,7 +487,7 @@ class QuestInstancesDao:
                 # midnight cannot delete instances that became past
                 # while it waited (backdating is clamped to the
                 # execution date, not the call date).
-                today = datetime.date.today().isoformat()
+                today = _today().isoformat()
                 effective_cutoff = max(cutoff_date, today)
                 result = await self._database.execute(
                     "DELETE FROM quest_instances WHERE definition_id = ? "
@@ -372,6 +499,49 @@ class QuestInstancesDao:
                 )
         return result.rowcount
 
+    async def delete_future_uncompleted_for_child(
+        self, child_id: int, cutoff_date: str
+    ) -> tuple[int, str]:
+        """Delete the child's open instances at/after cutoff, across every
+        definition, returning ``(deleted, effective_cutoff)``.
+
+        Child-scoped counterpart to :meth:`delete_future_uncompleted`,
+        used when a child's presence changes (Feature 09): only instances
+        at or after the cutoff with NO completion event are removed, inside
+        one transaction — an instance that gains a completion event
+        mid-flight cannot be deleted.  The cutoff cannot be backdated below
+        today, for the same reason as the definition-scoped method (see its
+        docstring).  Any instance with a completion event is never touched,
+        regardless of its date.
+
+        Returns the number deleted and the effective cutoff that was
+        actually applied (``max(cutoff_date, today)``, read under the
+        lock).  The caller (:func:`~.materialize.regenerate_for_child`)
+        passes that effective cutoff back as its re-materialization anchor,
+        so the delete and the rebuild share ONE execution-day even when the
+        call is queued across midnight — instead of the caller re-reading a
+        stale "today" and materializing a window the delete already
+        superseded.
+        """
+        _validate_date(cutoff_date, "cutoff_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                # "today" is read under the lock, matching the
+                # definition-scoped method: the DELETE executes on this
+                # date, so a caller queued across midnight cannot delete
+                # instances that became past while it waited.
+                today = _today().isoformat()
+                effective_cutoff = max(cutoff_date, today)
+                result = await self._database.execute(
+                    "DELETE FROM quest_instances WHERE child_id = ? "
+                    "AND due_date >= ? AND NOT EXISTS ("
+                    "  SELECT 1 FROM completion_events "
+                    "  WHERE instance_id = quest_instances.id"
+                    ")",
+                    (child_id, effective_cutoff),
+                )
+        return result.rowcount, effective_cutoff
+
 
 class CompletionEventsDao:
     """Typed append-only access to the ``completion_events`` table.

custom_components/nestquest/dao_presence.py

@@ -276,4 +276,59 @@ class PresenceOverridesDao:
             "DELETE FROM presence_overrides WHERE id = ?",
             (override_id,),
         )
-        return result.rowcount > 0
+        return result.rowcount > 0
+
+
+async def read_snapshot_unlocked(
+    database: NestQuestDatabase,
+    child_ids: list[int],
+    range_start: str,
+    range_end: str,
+) -> tuple[
+    dict[int, PresenceScheduleRecord],
+    dict[int, list[PresenceOverrideRecord]],
+]:
+    """Read schedules and overrides for ``child_ids`` WITHOUT locking.
+
+    MUST be called inside the connection lock and an open transaction, so
+    the schedules and overrides returned here stay coherent with every
+    other table the caller reads inside that same transaction (the
+    materialization walk's single input snapshot does exactly that).  The
+    date range comes already validated by the caller.
+
+    Returns ``(schedules, overrides)``:
+
+    - ``schedules`` maps ``child_id`` to its :class:`PresenceScheduleRecord`
+      for every child that HAS a schedule row.  A child absent from the
+      mapping has no schedule, i.e. is present every day (Feature 05).
+    - ``overrides`` maps ``child_id`` to that child's overrides overlapping
+      ``[range_start, range_end]`` (inclusive boundaries), ordered by start
+      date; a child with no matching override is absent from the mapping.
+
+    ``child_ids`` must name existing children (the caller derives them
+    from a trusted snapshot, within the same transaction).
+    """
+    children = sorted(set(child_ids))
+    if not children:
+        return {}, {}
+    placeholders = ",".join("?" for _ in children)
+    schedule_rows = await database.fetch_all(
+        f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules "
+        f"WHERE child_id IN ({placeholders})",
+        tuple(children),
+    )
+    override_rows = await database.fetch_all(
+        f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides "
+        f"WHERE child_id IN ({placeholders}) AND end_date >= ? "
+        "AND start_date <= ? ORDER BY start_date",
+        tuple(children) + (range_start, range_end),
+    )
+    schedules = {
+        record.child_id: record
+        for record in map(_schedule_from_row, schedule_rows)
+    }
+    overrides: dict[int, list[PresenceOverrideRecord]] = {}
+    for row in override_rows:
+        record = _override_from_row(row)
+        overrides.setdefault(record.child_id, []).append(record)
+    return schedules, overrides

custom_components/nestquest/dao_rules.py

@@ -907,6 +907,81 @@ class QuestDefinitionsDao:
         )
         return [_definition_from_row(row) for row in rows]
 
+    async def _snapshots_for(
+        self, definitions: list[QuestDefinitionRecord]
+    ) -> list[QuestDefinitionSnapshot]:
+        """Read each definition's rule, assignees and windows.
+
+        MUST be called inside the connection lock (and ideally a
+        transaction) so the definitions handed in and the rows this
+        reads back cannot be raced by a concurrent edit between list
+        and bundle.  Each rule is read through
+        :class:`ScheduleRulesDao` on the same connection, so the
+        snapshot's rule record is the one the definition references AT
+        the moment of the locked read.
+        """
+        rules_dao = ScheduleRulesDao(self._database)
+        snapshots: list[QuestDefinitionSnapshot] = []
+        for definition in definitions:
+            rule_record = await rules_dao.get(definition.schedule_rule_id)
+            assert rule_record is not None
+            snapshots.append(
+                QuestDefinitionSnapshot(
+                    definition=definition,
+                    rule=rule_record,
+                    assignees=await self.list_assignees(definition.id),
+                    windows=await self.list_windows(definition.id),
+                )
+            )
+        return snapshots
+
+    async def list_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
+        """Return every active definition with its rule, assignees and
+        windows, all read inside one locked transaction.
+
+        The definitive list and each definition's linked rows are read
+        in a single BEGIN..COMMIT span under the connection lock, so a
+        concurrent atomic edit cannot interleave between the definition
+        list and the bundle read: the caller receives a coherent
+        snapshot (each definition's rule, assignees and windows as they
+        were at one instant), never a mixed old-rule/new-windows state.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definitions = await self.list_active()
+                return await self._snapshots_for(definitions)
+
+    async def read_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
+        """Return every active definition's snapshot WITHOUT locking.
+
+        Mirrors :meth:`list_snapshots_active` minus the lock and
+        transaction: the definitions and their rules, assignees and
+        windows are read on the caller's connection under whatever
+        transaction the caller is already holding, so they stay coherent
+        with every OTHER table the caller reads inside that same
+        transaction.  MUST be called inside the connection lock and an
+        open transaction (the single-transaction input snapshot the
+        materialization walk uses is that caller).
+        """
+        definitions = await self.list_active()
+        return await self._snapshots_for(definitions)
+
+    async def list_snapshots_by_child(
+        self, child_id: int
+    ) -> list[QuestDefinitionSnapshot]:
+        """Return each definition assigned to ``child_id`` with its rule,
+        assignees and windows, all read inside one locked transaction.
+
+        Mirrors :meth:`list_snapshots_active` but selects by assignment
+        (``list_by_child`` order: newest first).  Locks read-and-bundle
+        into one transaction so the bundle never mixes one edit's
+        definition with another's windows.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definitions = await self.list_by_child(child_id)
+                return await self._snapshots_for(definitions)
+
     async def update(
         self,
         definition_id: int,
@@ -1129,4 +1204,47 @@ class QuestDefinitionsDao:
         if not child[0]:
             raise ValueError(
                 f"child {child_id} is inactive and cannot be assigned"
-            )
+            )
+
+
+async def load_materialization_input(
+    database: NestQuestDatabase,
+    range_start: str,
+    range_end: str,
+):
+    """Read every materialization input in ONE locked transaction.
+
+    Returns ``(snapshots, schedules, overrides)``:
+
+    - ``snapshots``: every active definition with its rule, assignees and
+      windows (the :class:`QuestDefinitionSnapshot` shape).
+    - ``schedules`` / ``overrides``: the presence schedules and
+      range-scoped overrides of every assignee child, in
+      :mod:`.dao_presence`'s record shapes.
+
+    The whole read — definitions, rules, assignees, windows, and the
+    assignees' presence schedules and overrides — runs inside a single
+    BEGIN..COMMIT span under the connection lock, so a concurrent edit
+    (definition, assignment, window, or presence) cannot interleave
+    between the parts and the walk never sees a mixed-time view.  Each
+    table's SQL stays in its own DAO module: the presence portion is
+    delegated to :mod:`.dao_presence` (a lazy import here, since that
+    module imports this one).
+    """
+    definitions_dao = QuestDefinitionsDao(database)
+    async with _connection_lock(database):
+        async with database.transaction():
+            snapshots = await definitions_dao.read_snapshots_active()
+            child_ids = sorted(
+                {
+                    child.id
+                    for snapshot in snapshots
+                    for child in snapshot.assignees
+                }
+            )
+            from . import dao_presence  # lazy: dao_presence imports us
+
+            schedules, overrides = await dao_presence.read_snapshot_unlocked(
+                database, child_ids, range_start, range_end
+            )
+    return snapshots, schedules, overrides

custom_components/nestquest/materialize.py

@@ -0,0 +1,302 @@
+"""The materialization walk (Feature 07 core).
+
+:func:`materialize` turns the schedule rules + custody presence of the
+active quest definitions into concrete ``quest_instances`` rows over a
+date range.  It is the pure walk over definitions x dates x assignees x
+windows; HA scheduling (day-rollover triggers) is a separate later
+task and lives elsewhere.
+
+The walk is deterministic and idempotent:
+
+- All inputs — the active definitions with their rules, assignees and
+  windows, AND the assignee children's presence schedules and scoped
+  overrides — are read inside ONE locked transaction (see
+  :func:`~.dao_rules.load_materialization_input`), so the walk sees a
+  single coherent snapshot rather than a definition state from one
+  instant and a presence state from another.
+- Presence is snapshotted into an immutable
+  :class:`~.presence.PresenceEngine` from that same coherent read.
+- For each date in the range, each definition whose decoded rule
+  :func:`~.recurrence.occurs_on` on that date, each assignee who is
+  ``is_present`` on that date and each declared window, one instance is
+  written through :class:`~.dao_instances.QuestInstancesDao.upsert_if_valid`
+  with a ``generated_at`` stamp shared by the whole batch (one coherent
+  time per run).
+
+A config change that lands AFTER the snapshot is read but BEFORE a
+tuple's write (an assignment removed, a definition or child deactivated,
+a window removed) makes that tuple SKIP rather than abort the batch:
+:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid` re-validates
+every materialization precondition (active definition, active child,
+live assignment, declared window, and its CURRENT ``due_time``) inside
+the SAME transaction as the INSERT, so the check and the write cannot be
+split by a racing config edit.
+
+The upsert is idempotent on (definition_id, child_id, due_date,
+window), so re-running the materialization over the same range never
+duplicates a row; a tuple whose instance is already completed is SKIPPED
+(never rewritten, never raised) by
+:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`.
+
+Regeneration: a config or presence change must rebuild future instances
+so they track the new state.  :func:`regenerate_for_definition` covers
+definition edits / reassignment (Features 06/07); :func:`regenerate_for_child`
+covers a single child's presence change (Feature 09).  Both delete the
+affected open instances at or after today — never a completed instance,
+never the past — and re-run the walk over the rolling horizon
+``[today, today + DEFAULT_HORIZON_DAYS]``.
+
+HOOK (Feature 09 presence services): the presence business layer that
+lands in Feature 09 MUST call :func:`regenerate_for_child` after EVERY
+presence-state write so a child's future instances track its presence:
+``set_presence_pattern`` (including clearing/setting an empty pattern),
+deleting/clearing the presence schedule
+(:meth:`~.dao_presence.PresenceSchedulesDao.delete`),
+``create_presence_override``, and ``delete_presence_override``.  A
+schedule delete changes presence state too (the child becomes
+always-present), so it must regenerate just like the others.  This hook
+is documented, not yet wired: no presence business layer exists yet, and
+the presence DAO (``dao_presence.py``) must stay free of business rules
+— never call :func:`regenerate_for_child` from inside the DAO layer.
+"""
+from __future__ import annotations
+
+import datetime
+
+from .const import DEFAULT_HORIZON_DAYS
+from .dao_instances import QuestInstancesDao, _today
+from .dao_rules import (
+    _validate_date,
+    load_materialization_input,
+    schedule_rule_from_storage,
+    schedule_rule_storage_from_record,
+)
+from .db import NestQuestDatabase
+from .presence import PresenceEngine, PresenceOverride, PresenceSchedule
+from .recurrence import ScheduleRule, occurs_on
+
+
+def _now_stamp() -> str:
+    """Return the batch's shared UTC second-precision generation stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _decode_rule(snapshot) -> ScheduleRule:
+    """Decode a snapshot's rule record back into a ScheduleRule.
+
+    The record carried by the snapshot is the one the definition
+    referenced at read time, so decoding it here (rather than re-reading
+    through ``ScheduleRulesDao``) preserves the coherence the snapshot
+    guarantees.
+    """
+    return schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+
+
+def _build_engine(
+    schedules_records,
+    overrides_records,
+) -> PresenceEngine:
+    """Convert raw presence records into an immutable PresenceEngine."""
+    schedules: dict[int, PresenceSchedule] = {}
+    for child_id, record in schedules_records.items():
+        schedules[child_id] = PresenceSchedule.decode(
+            child_id, record.anchor_date, record.pattern
+        )
+    overrides: dict[int, list[PresenceOverride]] = {}
+    for child_id, records in overrides_records.items():
+        overrides[child_id] = [
+            PresenceOverride(
+                record.child_id,
+                record.start_date,
+                record.end_date,
+                record.is_present,
+                record.note,
+            )
+            for record in records
+        ]
+    return PresenceEngine(schedules, overrides)
+
+
+async def materialize(
+    database: NestQuestDatabase,
+    start_date: str,
+    end_date: str,
+    *,
+    child_ids: list[int] | None = None,
+) -> int:
+    """Generate quest instances for the closed range [start_date, end_date].
+
+    ``start_date`` and ``end_date`` must be strict ISO calendar dates
+    (YYYY-MM-DD) with ``end_date`` on or after ``start_date``; anything
+    else raises ValueError naming the offending field.  Returns the number
+    of instances upserted for the range (idempotent — a re-run refreshes
+    the same rows, never duplicates them).
+
+    ``child_ids`` optionally scopes the walk to exactly those children:
+    assignees outside the list are never written, so a caller that only
+    wants one child's instances rebuilt (e.g. :func:`regenerate_for_child`
+    after a presence change) cannot touch any unrelated child's rows.
+    When omitted (or None) the walk covers every assignee as before.
+
+    All inputs are read in ONE locked transaction; each tuple's write then
+    goes through the atomic :meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`,
+    which re-validates the tuple and its current ``due_time`` in the same
+    transaction as the INSERT, so a config change that arrives mid-batch
+    skips the affected tuples rather than aborting the batch or leaving a
+    partial result.
+    """
+    _validate_date(start_date, "start_date")
+    _validate_date(end_date, "end_date")
+    if end_date < start_date:
+        raise ValueError(
+            f"end_date must be on or after start_date, got "
+            f"{end_date!r} < {start_date!r}"
+        )
+
+    snapshots, schedules_records, overrides_records = await (
+        load_materialization_input(database, start_date, end_date)
+    )
+    engine = _build_engine(schedules_records, overrides_records)
+    child_filter = None if child_ids is None else set(child_ids)
+
+    definitions = [
+        (snapshot.definition.id, _decode_rule(snapshot),
+         snapshot.assignees, snapshot.windows)
+        for snapshot in snapshots
+    ]
+
+    generated_at = _now_stamp()
+    instances = QuestInstancesDao(database)
+
+    start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date()
+    end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date()
+
+    count = 0
+    cursor = start
+    while cursor <= end:
+        iso = cursor.isoformat()
+        for definition_id, rule, assignees, windows in definitions:
+            if not occurs_on(rule, cursor):
+                continue
+            for child in assignees:
+                if child_filter is not None and child.id not in child_filter:
+                    continue
+                if not engine.is_present(child.id, cursor):
+                    continue
+                for window in windows:
+                    written = await instances.upsert_if_valid(
+                        definition_id,
+                        child.id,
+                        iso,
+                        generated_at,
+                        window=window.window,
+                        due_time=window.due_time,
+                    )
+                    if written is not None:
+                        count += 1
+        cursor += datetime.timedelta(days=1)
+    return count
+
+
+async def regenerate_for_definition(
+    database: NestQuestDatabase,
+    definition_id: int,
+) -> int:
+    """Regenerate a definition's future instances after a config change.
+
+    The Feature 06 edit/assignment/activation paths call this after a
+    successful write so future instances track the new state: the
+    definition's open instances at or after today are deleted (never a
+    completed instance, never the past) via
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted`,
+    then the materialization walk re-runs over the rolling horizon
+    ``[today, today + DEFAULT_HORIZON_DAYS]`` so the (changed) rule,
+    assignee set and windows re-materialize against today's state.
+
+    "today" is computed the same way the walk computes it —
+    :func:`~.dao_instances._today` — so the delete cutoff and the
+    re-materialization range share one anchor, and the walk's
+    ``upsert_if_valid`` no-past guard never rejects the regenerated
+    range.  Returns the number of instances the re-materialization
+    upserted across the whole snapshot (idempotent — other active
+    definitions' tuples refresh in place, never duplicate).
+    """
+    today = _today()
+    start_date = today.isoformat()
+    end_date = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+
+    instances = QuestInstancesDao(database)
+    await instances.delete_future_uncompleted(definition_id, start_date)
+    return await materialize(database, start_date, end_date)
+
+
+# HOOK (Feature 09 presence services): ``regenerate_for_child`` is the
+# child-presence counterpart to ``regenerate_for_definition``.  The
+# presence business layer that lands in Feature 09 MUST call
+# ``regenerate_for_child`` after EVERY presence-state write, so future
+# instances track a child's changed presence: set_presence_pattern
+# (including clearing/setting an empty pattern), deleting/clearing the
+# presence schedule (PresenceSchedulesDao.delete),
+# create_presence_override, and delete_presence_override.  A schedule
+# delete changes presence state too (the child becomes always-present),
+# so it must regenerate just like the others.  It is deliberately NOT
+# wired into ``dao_presence.py``: the DAO layer stays pure storage (no
+# business rules), and no presence business layer exists yet to host the
+# call — see the module docstring note above.
+async def regenerate_for_child(
+    database: NestQuestDatabase,
+    child_id: int,
+) -> int:
+    """Regenerate a child's future instances after a presence change.
+
+    The Feature 09 presence services call this after every
+    presence-state write — set_presence_pattern (including clearing or
+    setting an empty pattern), deleting/clearing the presence schedule
+    (:meth:`~.dao_presence.PresenceSchedulesDao.delete`),
+    create_presence_override, and delete_presence_override — so a
+    child's future instances track its new presence: the child's open
+    instances at or after today are deleted across ALL its definitions
+    (never a completed instance, never the past) via
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted_for_child`,
+    then the materialization walk re-runs over the rolling horizon
+    ``[today, today + DEFAULT_HORIZON_DAYS]`` scoped to that child only,
+    so the child's now-absent/present dates re-materialize against
+    today's presence WITHOUT touching any other child's rows.
+
+    ``child_id`` must be a plain int — bool and float are rejected
+    (SQLite binds a bool as 0/1 and a float would round, so a malformed
+    id must never mutate the wrong profile).
+
+    "today" is computed the same way the walk computes it —
+    :func:`~.dao_instances._today` — and the re-materialization range uses the
+    delete's returned effective cutoff (see
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted_for_child`),
+    so the delete cutoff and the re-materialization window share ONE
+    execution-day anchor even when the call is queued across midnight.
+    Returns the number of instances the re-materialization upserted for
+    the child across the whole snapshot.
+    """
+    if isinstance(child_id, bool) or not isinstance(child_id, int):
+        raise ValueError(f"child_id must be an integer, got {child_id!r}")
+
+    today = _today()
+
+    instances = QuestInstancesDao(database)
+    # Use the delete's returned effective cutoff as the materialize anchor,
+    # NOT the independently-computed "today" above.  If the call is queued
+    # across midnight the delete clamps its cutoff to the execution-day;
+    # materializing from the stale "today" (now yesterday) would trip the
+    # walk's no-past guard and leave the child's future instances deleted
+    # but not rebuilt.  One anchor for both keeps them consistent.
+    _, effective_cutoff = await instances.delete_future_uncompleted_for_child(
+        child_id, today.isoformat()
+    )
+    start = datetime.date.fromisoformat(effective_cutoff)
+    end_date = (start + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+    return await materialize(
+        database, start.isoformat(), end_date, child_ids=[child_id]
+    )

custom_components/nestquest/quest_definitions.py

@@ -44,6 +44,7 @@ from dataclasses import dataclass
 from .dao_children import ChildRecord, _connection_lock
 from .dao_rules import (
     QuestDefinitionRecord,
+    QuestDefinitionSnapshot,
     QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
     ScheduleRuleStorage,
@@ -57,6 +58,7 @@ from .dao_rules import (
     schedule_rule_to_storage,
 )
 from .db import NestQuestDatabase
+from .materialize import regenerate_for_definition
 from .recurrence import ScheduleRule, occurs_on
 
 
@@ -318,19 +320,30 @@ async def edit_quest_definition(
         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,
-    )
+    try:
+        snapshot = await dao.edit_definition(
+            definition_id,
+            title=title_value,
+            description=description_value,
+            icon=icon_value,
+            rule=rule_storage,
+            windows=window_specs,
+        )
+    except ValueError as error:
+        # The DAO's missing-definition error names only the id; re-raise
+        # so the public contract names the field the caller passed (matching
+        # assign/unassign/deactivate).
+        message = str(error)
+        if message.startswith("quest definition "):
+            raise ValueError(f"definition_id: {message}") from error
+        raise
 
     decoded_rule = schedule_rule_from_storage(
         schedule_rule_storage_from_record(snapshot.rule)
     )
 
+    await regenerate_for_definition(database, definition_id)
+
     return CreatedQuestDefinition(
         definition=snapshot.definition,
         rule=decoded_rule,
@@ -361,7 +374,7 @@ async def assign_child(
     _validate_child_id(child_id)
     dao = QuestDefinitionsDao(database)
     try:
-        return await dao.add_assignee(definition_id, child_id)
+        assignees = 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.
@@ -371,6 +384,8 @@ async def assign_child(
         if message.startswith("quest definition "):
             raise ValueError(f"definition_id: {message}") from error
         raise
+    await regenerate_for_definition(database, definition_id)
+    return assignees
 
 
 async def unassign_child(
@@ -395,7 +410,9 @@ async def unassign_child(
         raise ValueError(
             f"definition_id: quest definition {definition_id} does not exist"
         )
-    return await dao.remove_assignee(definition_id, child_id)
+    removed = await dao.remove_assignee(definition_id, child_id)
+    await regenerate_for_definition(database, definition_id)
+    return removed
 
 
 async def set_quest_definition_active(
@@ -447,6 +464,7 @@ async def set_quest_definition_active(
     rule = schedule_rule_from_storage(
         schedule_rule_storage_from_record(rule_record)
     )
+    await regenerate_for_definition(database, definition_id)
     return CreatedQuestDefinition(
         definition=updated,
         rule=rule,
@@ -455,33 +473,25 @@ async def set_quest_definition_active(
     )
 
 
-async def _bundle_definition(
-    database: NestQuestDatabase,
-    dao: QuestDefinitionsDao,
-    definition: QuestDefinitionRecord,
+def _bundle_snapshot(
+    snapshot: QuestDefinitionSnapshot,
 ) -> CreatedQuestDefinition:
-    """Assemble a definition's rich view: decoded rule + assignees + windows.
+    """Decode a DAO snapshot into the layer's rich view.
 
-    Reads the schedule rule through :class:`~.dao_rules.ScheduleRulesDao`
-    and the assignees and windows through the DAO, then wraps them in the
-    same :class:`CreatedQuestDefinition` shape the create/edit/activate
-    paths return, so every query helper here reuses one snapshot type
-    rather than inventing a second one.
+    The rule record carried by the snapshot is the one the definition
+    referenced at read time, so decoding it here (rather than re-reading
+    through :class:`~.dao_rules.ScheduleRulesDao`) preserves the
+    coherence the DAO's locked snapshot guarantees: the rule, assignees
+    and windows all come from one atomic read, never three separate ones.
     """
-    rule_record = await ScheduleRulesDao(database).get(
-        definition.schedule_rule_id
-    )
-    assert rule_record is not None
     rule = schedule_rule_from_storage(
-        schedule_rule_storage_from_record(rule_record)
+        schedule_rule_storage_from_record(snapshot.rule)
     )
-    assignees = await dao.list_assignees(definition.id)
-    windows = await dao.list_windows(definition.id)
     return CreatedQuestDefinition(
-        definition=definition,
+        definition=snapshot.definition,
         rule=rule,
-        assignees=assignees,
-        windows=windows,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
     )
 
 
@@ -494,14 +504,14 @@ async def list_active_definitions(
     order, so callers get a stable, repeatable sequence.  Each entry is
     the same rich :class:`CreatedQuestDefinition` snapshot as the rest of
     the layer: the stored row plus its decoded
-    :class:`~.recurrence.ScheduleRule`, assignees and windows.
+    :class:`~.recurrence.ScheduleRule`, assignees and windows.  The
+    definitions and their links are read by the DAO inside one locked
+    transaction, so each bundle is a coherent snapshot that a concurrent
+    edit cannot interleave.
     """
     dao = QuestDefinitionsDao(database)
-    definitions = await dao.list_active()
-    return [
-        await _bundle_definition(database, dao, definition)
-        for definition in definitions
-    ]
+    snapshots = await dao.list_snapshots_active()
+    return [_bundle_snapshot(snapshot) for snapshot in snapshots]
 
 
 async def list_definitions_for_child(
@@ -515,15 +525,12 @@ async def list_definitions_for_child(
     rising definition id, matching ``list_active_definitions`` and
     ``list_definitions_firing_on``.  Each entry is a
     :class:`CreatedQuestDefinition` snapshot bundling the definition with
-    its decoded rule, assignees and windows.
+    its decoded rule, assignees and windows, read atomically by the DAO.
     """
     _validate_child_id(child_id)
     dao = QuestDefinitionsDao(database)
-    definitions = await dao.list_by_child(child_id)
-    bundles = [
-        await _bundle_definition(database, dao, definition)
-        for definition in definitions
-    ]
+    snapshots = await dao.list_snapshots_by_child(child_id)
+    bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
     return sorted(bundles, key=lambda bundle: bundle.definition.id)
 
 
@@ -540,13 +547,13 @@ async def list_definitions_firing_on(
     definitions whose decoded rule fires on the date are returned, in
     rising definition-id order, each as the same rich
     :class:`CreatedQuestDefinition` snapshot as the rest of the layer.
+    The definitions and their rules are read inside one locked
+    transaction, so a bundle can never filter on one rule while
+    presenting another state's windows.
     """
     _validate_date(date, "date")
     target = datetime.datetime.strptime(date, "%Y-%m-%d").date()
     dao = QuestDefinitionsDao(database)
-    definitions = await dao.list_active()
-    bundles = [
-        await _bundle_definition(database, dao, definition)
-        for definition in definitions
-    ]
+    snapshots = await dao.list_snapshots_active()
+    bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
     return [bundle for bundle in bundles if occurs_on(bundle.rule, target)]

tests/test_quest_definitions.py

@@ -786,6 +786,24 @@ def test_edit_rejects_nonexistent_definition(tmp_path) -> None:
     _with_db(tmp_path, "edit-missing-definition.db")(_body)
 
 
+def test_edit_nonexistent_definition_names_field(tmp_path) -> None:
+    """A missing definition's error names the ``definition_id`` field.
+
+    The DAO raises a bare "quest definition <id> does not exist"; the
+    business layer must re-raise it prefixed with the field the caller
+    passed, matching assign/unassign/deactivate.
+    """
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError,
+            match="definition_id: quest definition 999 does not exist",
+        ):
+            await edit_quest_definition(database, 999, title="X")
+        return None
+
+    _with_db(tmp_path, "edit-missing-definition-field.db")(_body)
+
+
 @pytest.mark.parametrize("bad", [True, 1.5, "1", None])
 def test_edit_rejects_non_int_definition_id(tmp_path, bad) -> None:
     async def _body(database, children, rules, definitions, child):
@@ -1261,11 +1279,19 @@ def test_set_quest_definition_active_nonexistent_definition_names_field(
     _with_db(tmp_path, "set-active-missing-definition.db")(_body)
 
 
-def test_set_quest_definition_active_never_deletes_or_touches_instances(
+def test_set_quest_definition_active_preserves_completed_and_history(
     tmp_path,
 ) -> None:
-    """Deactivation hides the definition but deletes nothing, and
-    pre-existing instances and completion history survive unchanged."""
+    """Deactivation stops future generation; reactivation resumes it.
+
+    The Feature 07 wiring makes both transitions regenerate the
+    definition's rolling horizon: future OPEN instances are removed on
+    deactivation and re-materialized on reactivation.  A pre-existing
+    COMPLETED instance and its completion events always survive both
+    transitions unchanged (regeneration never touches completed rows or
+    the past), and the append-only completion-events table is never
+    rewritten.
+    """
     async def _body(database, children, rules, definitions, child):
         created = await create_quest_definition(
             database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
@@ -1292,9 +1318,6 @@ def test_set_quest_definition_active_never_deletes_or_touches_instances(
             actor_user_id="admin-1",
         )
 
-        instance_count = await database.fetch_one(
-            "SELECT COUNT(*) FROM quest_instances"
-        )
         event_count = await database.fetch_one(
             "SELECT COUNT(*) FROM completion_events"
         )
@@ -1314,29 +1337,25 @@ def test_set_quest_definition_active_never_deletes_or_touches_instances(
             definition_id
         )] == ["morning"]
         # The pre-existing instance and completion event survive with
-        # unchanged ids and contents, and no new rows appeared.
+        # unchanged ids and contents, and no new event rows appeared.
         assert await instances.get_by_id(instance.id) == instance
         assert await events.list_by_instance(instance.id) == [event]
-        assert await database.fetch_one(
-            "SELECT COUNT(*) FROM quest_instances"
-        ) == instance_count
         assert await database.fetch_one(
             "SELECT COUNT(*) FROM completion_events"
         ) == event_count
 
         await set_quest_definition_active(database, definition_id, True)
         assert (await definitions.get(definition_id)).is_active is True
+        # Reactivation regenerates future OPEN instances, but the
+        # completed instance and its history survive the sweep unchanged.
         assert await instances.get_by_id(instance.id) == instance
         assert await events.list_by_instance(instance.id) == [event]
-        assert await database.fetch_one(
-            "SELECT COUNT(*) FROM quest_instances"
-        ) == instance_count
         assert await database.fetch_one(
             "SELECT COUNT(*) FROM completion_events"
         ) == event_count
         return None
 
-    _with_db(tmp_path, "set-active-never-deletes.db")(_body)
+    _with_db(tmp_path, "set-active-preserves-completed.db")(_body)
 
 
 def _gated_hass(
@@ -1621,3 +1640,301 @@ def test_list_definitions_firing_on_rejects_malformed_date(tmp_path, bad) -> Non
         return None
 
     _with_db(tmp_path, "list-firing-on-bad-date.db")(_body)
+
+
+def test_list_query_snapshot_coherent_under_racing_edit(tmp_path) -> None:
+    """A racing edit cannot mix states into a query helper's snapshot.
+
+    The query helper pauses mid-snapshot (inside its locked transaction)
+    while an edit that changes BOTH the rule and the windows is queued
+    behind the connection lock.  Because the helper reads definitions,
+    rules and windows in one locked transaction, its returned bundle is
+    either the whole old state or the whole new state — never the old
+    rule paired with the new windows (the exact interleaving the unlocked
+    reads used to allow).  The pause hook holds
+    ``list_windows`` mid-flight so the edit is deterministically queued
+    against the helper's lock.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        weekly = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2},
+            start_date="2026-09-14",
+        )
+
+        import custom_components.nestquest.dao_rules as dao_rules
+
+        original_list = dao_rules.QuestDefinitionsDao.list_windows
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def _pausing_list(self, definition_id):
+            if not started.is_set():
+                started.set()
+                await release.wait()
+            return await original_list(self, definition_id)
+
+        dao_rules.QuestDefinitionsDao.list_windows = _pausing_list
+        try:
+            read_task = asyncio.ensure_future(
+                list_definitions_firing_on(database, "2026-09-16")
+            )
+            await started.wait()
+            edit_task = asyncio.ensure_future(
+                edit_quest_definition(
+                    database,
+                    created.definition.id,
+                    rule=weekly,
+                    windows=["evening"],
+                )
+            )
+            await asyncio.sleep(0)
+            release.set()
+            result, edited = await asyncio.gather(read_task, edit_task)
+        finally:
+            dao_rules.QuestDefinitionsDao.list_windows = original_list
+
+        # The helper finished before the queued edit, so its snapshot is
+        # the whole OLD state: the daily rule with its morning window.
+        assert [b.definition.id for b in result] == [created.definition.id]
+        assert result[0].rule == _daily_rule()
+        assert [w.window for w in result[0].windows] == ["morning"]
+        # The edit then reported the whole NEW state, not a mix either.
+        assert edited.rule == weekly
+        assert [w.window for w in edited.windows] == ["evening"]
+        return result
+
+    _with_db(tmp_path, "list-snapshot-concurrent.db")(_body)
+
+
+def test_list_definitions_for_child_snapshot_coherent_under_racing_unassign(
+    tmp_path,
+) -> None:
+    """A racing unassign cannot strip the child from the child-scoped
+    snapshot that ``list_definitions_for_child`` returns.
+
+    The helper pauses mid-snapshot (inside its locked transaction) while
+    an unassign of the very child it is listing is queued behind the
+    connection lock.  Because the definition set and its assignees are
+    read in one locked transaction, the returned bundle still shows the
+    child assigned — the coherent pre-unassign snapshot — rather than the
+    mixed state (definition returned "for child" but with the child
+    already gone from its roster) that the unlocked reads used to allow.
+    The pause hook holds ``list_assignees`` mid-flight so the unassign is
+    deterministically queued against the helper's lock.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+
+        import custom_components.nestquest.dao_rules as dao_rules
+
+        original_list = dao_rules.QuestDefinitionsDao.list_assignees
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def _pausing_list(self, definition_id):
+            if not started.is_set():
+                started.set()
+                await release.wait()
+            return await original_list(self, definition_id)
+
+        dao_rules.QuestDefinitionsDao.list_assignees = _pausing_list
+        try:
+            read_task = asyncio.ensure_future(
+                list_definitions_for_child(database, child.id)
+            )
+            await started.wait()
+            unassign_task = asyncio.ensure_future(
+                unassign_child(database, created.definition.id, child.id)
+            )
+            for _ in range(5):
+                await asyncio.sleep(0)
+            release.set()
+            result, removed = await asyncio.gather(read_task, unassign_task)
+        finally:
+            dao_rules.QuestDefinitionsDao.list_assignees = original_list
+
+        # The helper finished before the queued unassign, so its snapshot
+        # is the whole OLD state: the child still assigned to the
+        # definition it returned for that child.
+        assert [b.definition.id for b in result] == [created.definition.id]
+        assert [c.id for c in result[0].assignees] == [child.id]
+        assert removed is True
+        return result
+
+    _with_db(tmp_path, "list-for-child-snapshot-concurrent.db")(_body)
+
+
+def test_list_active_definitions_snapshot_coherent_under_racing_edit(
+    tmp_path,
+) -> None:
+    """A racing edit cannot mix states into ``list_active_definitions``.
+
+    The helper pauses mid-snapshot (inside its locked transaction) while
+    an edit that changes BOTH the rule and the windows is queued behind
+    the connection lock.  Because definitions, rules and windows are read
+    in one locked transaction, the returned bundle is the whole old state
+    — the daily rule with its morning window — never the old rule paired
+    with the new windows.  The pause hook holds ``list_windows``
+    mid-flight so the edit is deterministically queued against the
+    helper's lock.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        weekly = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2},
+            start_date="2026-09-14",
+        )
+
+        import custom_components.nestquest.dao_rules as dao_rules
+
+        original_list = dao_rules.QuestDefinitionsDao.list_windows
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def _pausing_list(self, definition_id):
+            if not started.is_set():
+                started.set()
+                await release.wait()
+            return await original_list(self, definition_id)
+
+        dao_rules.QuestDefinitionsDao.list_windows = _pausing_list
+        try:
+            read_task = asyncio.ensure_future(list_active_definitions(database))
+            await started.wait()
+            edit_task = asyncio.ensure_future(
+                edit_quest_definition(
+                    database,
+                    created.definition.id,
+                    rule=weekly,
+                    windows=["evening"],
+                )
+            )
+            await asyncio.sleep(0)
+            release.set()
+            result, edited = await asyncio.gather(read_task, edit_task)
+        finally:
+            dao_rules.QuestDefinitionsDao.list_windows = original_list
+
+        # The helper finished before the queued edit, so its snapshot is
+        # the whole OLD state: the daily rule with its morning window.
+        assert [b.definition.id for b in result] == [created.definition.id]
+        assert result[0].rule == _daily_rule()
+        assert [w.window for w in result[0].windows] == ["morning"]
+        # The edit then reported the whole NEW state, not a mix either.
+        assert edited.rule == weekly
+        assert [w.window for w in edited.windows] == ["evening"]
+        return result
+
+    _with_db(tmp_path, "list-active-snapshot-concurrent.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# full-lifecycle integration: one definition through every operation
+# ---------------------------------------------------------------------------
+
+
+def test_full_lifecycle_preserves_completed_and_history(tmp_path) -> None:
+    """End-to-end lifecycle preserves completed rows and append-only history.
+
+    Each stage (create -> edit rule/windows -> assign a second child ->
+    unassign the original child -> deactivate -> reactivate) now
+    regenerates the definition's future OPEN instances (Feature 07
+    wiring), but a pre-existing COMPLETED instance and its completion
+    events survive every stage unchanged — the completed row keeps its
+    id and snapshot columns, the completion-events table stays append-only.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            [("morning", "07:00")],
+        )
+        definition_id = created.definition.id
+
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+        instance = await instances.upsert(
+            definition_id, child.id, _today_iso(), NOW, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "user",
+            NOW,
+            True,
+            actor_user_id="admin-1",
+        )
+        event_count = await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        )
+
+        async def _assert_history_preserved():
+            assert await instances.get_by_id(instance.id) == instance
+            assert await events.list_by_instance(instance.id) == [event]
+            assert await database.fetch_one(
+                "SELECT COUNT(*) FROM completion_events"
+            ) == event_count
+
+        await _assert_history_preserved()
+
+        # Edit rule and windows together; the definition's rule row is
+        # rewritten in place and its window set replaced.
+        edited = await edit_quest_definition(
+            database,
+            definition_id,
+            rule=ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={1, 3},
+                start_date="2026-09-14",
+            ),
+            windows=[("morning", "08:00"), ("evening", "19:30")],
+        )
+        assert edited.rule == ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={1, 3},
+            start_date="2026-09-14",
+        )
+        by_name = {w.window: w.due_time for w in edited.windows}
+        assert by_name == {"morning": "08:00", "evening": "19:30"}
+        await _assert_history_preserved()
+
+        # Assign a second child: multi-assignee, both present.
+        bo = await children.create("Bo", NOW)
+        roster = await assign_child(database, definition_id, bo.id)
+        assert sorted(c.id for c in roster) == sorted([child.id, bo.id])
+        await _assert_history_preserved()
+
+        # Unassign the original child: only the link drops.
+        assert await unassign_child(database, definition_id, child.id) is True
+        assert [
+            c.id for c in await definitions.list_assignees(definition_id)
+        ] == [bo.id]
+        await _assert_history_preserved()
+
+        # Deactivate, then reactivate: the definition survives both.
+        deactivated = await set_quest_definition_active(
+            database, definition_id, False
+        )
+        assert deactivated.definition.is_active is False
+        await _assert_history_preserved()
+        reactivated = await set_quest_definition_active(
+            database, definition_id, True
+        )
+        assert reactivated.definition.is_active is True
+        await _assert_history_preserved()
+
+        return reactivated
+
+    _with_db(tmp_path, "full-lifecycle.db")(_body)