Review fix: strict non-bool int child_id validation on all three calls

Review round 1 (Codex gpt-5.6-terra): SQLite binds bools as integers,
so edit_child(True, ...) would silently edit child 1 and
set_child_active had the same hole (reorder already validated its
list). A shared _validate_child_id guard now runs before lookup in
edit_child, set_child_active and reorder_children's list scan, with a
coercible-id test matrix for both call shapes (True/False/1.0/"1"/
None) asserting the refused call leaves the profile untouched. Suite:
875 passed.

custom_components/nestquest/children.py

@@ -0,0 +1,280 @@
+"""Business layer for the children registry (Feature 03).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_children.ChildrenDao`: callers get validation, name
+policies and the duplicate-name warning here, and never talk to the
+DAO or SQL directly.  Children are never hard-deleted — deactivation
+is a separate explicit operation (:func:`set_child_active`), so
+completion history keeps its references (feature guardrail).
+
+Name policy: the display name is required and trimmed; an empty or
+whitespace-only name is rejected with ValueError.  A duplicate display
+name is ALLOWED (two children may legitimately share a nickname) but
+logged as a warning so an accidental double-add is visible in the HA
+log.  The duplicate check is case-insensitive and whitespace-normalised
+on both sides, because "ada" and "Ada " are the same name to a parent.
+
+Colour and avatar are optional free-form strings, trimmed when present;
+``sort_order`` is an int with the DAO/schema default of 0.  Clearing an
+optional field back to NULL is deliberately rejected: the DAO's update
+writes only provided fields (it cannot express NULL), so the business
+layer refuses an explicit ``None`` instead of silently ignoring it.
+Timestamps follow the module timestamp policy: the business layer
+stamps UTC ISO-8601 at second precision so every row it creates is
+uniform.
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+
+from .dao_children import ChildRecord, ChildrenDao, _connection_lock
+from .db import NestQuestDatabase
+
+LOGGER = logging.getLogger(__name__)
+
+#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
+#: in :func:`edit_child`.
+_UNSET = object()
+
+
+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 (whitespace-only names are empty to a parent).
+    """
+    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_sort_order(value: object) -> int:
+    """Reject non-int sort orders (bools included) before the DAO."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"sort_order 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 _warn_on_duplicate_name(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    exclude_child_id: int | None = None,
+) -> None:
+    """Log a warning when ``display_name`` duplicates an existing child.
+
+    Case-insensitive and whitespace-normalised on both sides.  The
+    duplicate is still allowed — this only makes it visible; the log
+    line names both rows so an accidental double-add can be found.
+    ``exclude_child_id`` removes the child being edited from the scan,
+    so an edit only warns when the result collides with a DIFFERENT
+    child's name — including case-only edits, where the self-comparison
+    would otherwise hide a real duplicate.
+    """
+    lowered = display_name.casefold()
+    for child in await ChildrenDao(database).list_all():
+        if exclude_child_id is not None and child.id == exclude_child_id:
+            continue
+        if child.display_name.strip().casefold() == lowered:
+            LOGGER.warning(
+                "Child display name %r duplicates existing child %d "
+                "(%r); allowed, but check this was intentional",
+                display_name,
+                child.id,
+                child.display_name,
+            )
+            return
+
+
+async def create_child(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    colour: str | None = None,
+    avatar_ref: str | None = None,
+    sort_order: int = 0,
+) -> ChildRecord:
+    """Create a child profile and return the stored record.
+
+    The display name is required, trimmed, and rejected when empty.
+    A duplicate display name (case-insensitive) is allowed but logged
+    as a warning.  The scan and the INSERT run inside the
+    connection-scoped lock shared with the DAOs, so a concurrent
+    create of the same normalised name queues behind the first insert
+    and sees it — the duplicate still warns; it can never slip through
+    unlogged.
+    """
+    name = _validate_text(display_name, "display_name", required=True)
+    assert name is not None
+    colour_value = _validate_text(colour, "colour")
+    avatar_value = _validate_text(avatar_ref, "avatar_ref")
+    order = _validate_sort_order(sort_order)
+    async with _connection_lock(database):
+        await _warn_on_duplicate_name(database, name)
+        created = await ChildrenDao(database).create(
+            name,
+            _now_stamp(),
+            colour=colour_value,
+            avatar_ref=avatar_value,
+            sort_order=order,
+        )
+    return created
+
+
+async def edit_child(
+    database: NestQuestDatabase,
+    child_id: int,
+    *,
+    display_name: str | object = _UNSET,
+    colour: str | None | object = _UNSET,
+    avatar_ref: str | None | object = _UNSET,
+    sort_order: int | object = _UNSET,
+) -> ChildRecord:
+    """Edit a child profile and return the updated record.
+
+    Arguments default to the module sentinel ``_UNSET`` meaning "leave
+    this field alone".  An explicitly passed ``None`` for an optional
+    field raises: the DAO's update writes only provided fields and
+    cannot express NULL, so a silent no-op would mask the caller's
+    intent (see the module docstring).  A provided display name is
+    validated like create's; the duplicate scan excludes only the child
+    being edited, so a case-only edit colliding with another child's
+    normalised name still warns.  Raises ValueError when the child does
+    not exist.
+
+    Atomicity: the existence check, the duplicate scan, the UPDATE and
+    the readback all run inside the connection-scoped lock shared with
+    the DAOs, so another edit cannot slip between this call's UPDATE
+    and its readback — the returned record is exactly the one this
+    edit produced, not a later writer's.
+    """
+    _validate_child_id(child_id)
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        existing = await dao.get(child_id)
+        if existing is None:
+            raise ValueError(f"child {child_id} does not exist")
+        updates: dict[str, object] = {}
+        if display_name is not _UNSET:
+            name = _validate_text(
+                display_name, "display_name", required=True
+            )
+            assert name is not None
+            await _warn_on_duplicate_name(
+                database, name, exclude_child_id=child_id
+            )
+            updates["display_name"] = name
+        if colour is not _UNSET:
+            value = _validate_text(colour, "colour")
+            if value is None:
+                raise ValueError(
+                    "clearing colour is not supported; pass a replacement "
+                    "value or omit the field"
+                )
+            updates["colour"] = value
+        if avatar_ref is not _UNSET:
+            value = _validate_text(avatar_ref, "avatar_ref")
+            if value is None:
+                raise ValueError(
+                    "clearing avatar_ref is not supported; pass a "
+                    "replacement value or omit the field"
+                )
+            updates["avatar_ref"] = value
+        if sort_order is not _UNSET:
+            updates["sort_order"] = _validate_sort_order(sort_order)
+        if updates:
+            await dao.update(child_id, **updates)
+        edited = await dao.get(child_id)
+    assert edited is not None
+    return edited
+
+
+async def set_child_active(
+    database: NestQuestDatabase, child_id: int, is_active: bool
+) -> ChildRecord:
+    """Deactivate or reactivate a child; returns the updated record.
+
+    Deactivation hides the child from the panel without deleting their
+    history (feature guardrail); this is the only removal path.
+    ``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 when the
+    child does not exist.
+
+    Atomicity: existence check, UPDATE and readback run inside the
+    connection-scoped lock, so a concurrent opposite transition cannot
+    make this call report the other caller's value.
+    """
+    _validate_child_id(child_id)
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        if await dao.get(child_id) is None:
+            raise ValueError(f"child {child_id} does not exist")
+        if not isinstance(is_active, bool):
+            raise ValueError(
+                f"is_active must be a real bool, got {is_active!r}"
+            )
+        await dao.set_active(child_id, is_active)
+        updated = await dao.get(child_id)
+    assert updated is not None
+    return updated
+
+
+async def reorder_children(
+    database: NestQuestDatabase, ordered_ids: list[int]
+) -> None:
+    """Rewrite the children's panel display order in one transaction.
+
+    ``ordered_ids`` must be a complete permutation of the children
+    table — every child id exactly once.  A partial list (missing an
+    existing child), an unknown id, or a duplicate is rejected with
+    ValueError before any write, so the current order survives a
+    rejected call.  The DAO enforces this inside its own transaction
+    (check and write are     atomic under the connection lock); this
+    wrapper additionally rejects non-integer ids up front, since
+    ``bool`` and floats would silently compare against stored integer
+    ids at the SQL layer.
+    """
+    for child_id in ordered_ids:
+        _validate_child_id(child_id)
+    await ChildrenDao(database).reorder(ordered_ids)
+
+
+async def list_children(
+    database: NestQuestDatabase, *, active_only: bool = False
+) -> list[ChildRecord]:
+    """Return children in display order (sort_order, then id)."""
+    dao = ChildrenDao(database)
+    if active_only:
+        return await dao.list_active()
+    return await dao.list_all()

custom_components/nestquest/const.py

@@ -20,3 +20,28 @@ DEFAULT_DAY_ROLLOVER_TIME = "00:00"
 
 CONF_PANEL_IDLE_TIMEOUT = "panel_idle_timeout"
 DEFAULT_PANEL_IDLE_TIMEOUT = 300
+
+# Quest windows (D-008): a definition may declare several day windows;
+# each produces its own instance per day and groups the panel's Quest
+# Log into three columns.  The clock ranges are stored here so they stay
+# configurable later.  Boundaries: a window's end equals the next
+# window's start; classification (Feature 07 materialisation) treats a
+# due time exactly on the boundary as belonging to the LATER window, so
+# 12:00 is afternoon (not morning) and 17:00 is evening (not
+# afternoon).  Times after 21:00 belong to no window.
+WINDOW_MORNING = "morning"
+WINDOW_AFTERNOON = "afternoon"
+WINDOW_EVENING = "evening"
+
+QUEST_WINDOWS: tuple[str, ...] = (
+    WINDOW_MORNING,
+    WINDOW_AFTERNOON,
+    WINDOW_EVENING,
+)
+
+#: Inclusive "HH:MM" bounds per window, in 24-hour local time.
+WINDOW_CLOCK_RANGES: dict[str, tuple[str, str]] = {
+    WINDOW_MORNING: ("00:00", "11:59"),
+    WINDOW_AFTERNOON: ("12:00", "17:00"),
+    WINDOW_EVENING: ("17:00", "21:00"),
+}

custom_components/nestquest/dao_children.py

@@ -191,13 +191,20 @@ class ChildrenDao:
     async def reorder(self, ordered_ids: list[int]) -> None:
         """Persist a new sort order: ``ordered_ids[i]`` gets sort_order i.
 
+        The list must be a COMPLETE permutation of the children table:
+        every existing child id appears exactly once.  Partial lists
+        (missing an existing child) and unknown ids (naming a child
+        that does not exist) are rejected with ValueError BEFORE any
+        write, leaving the current order untouched — a silent partial
+        reorder would leave unlisted children with stale positions and
+        hide a caller bug.  Duplicate ids are rejected up front: they
+        cannot all hold distinct positions, and silently assigning the
+        last occurrence's index would hide a caller bug.
+
         Runs inside one transaction; positions are staged at offset
         -len (all negative, so any permutation of the same ids never
         collides mid-reorder under the wrapper's single-connection
-        serialization), then finalized 0..n-1.  Duplicate ids are
+        serialization), then finalized 0..n-1.
-        rejected up front: they cannot all hold distinct positions, and
-        silently assigning the last occurrence's index would hide a
-        caller bug.
 
         Concurrent reorder() calls on one connection are serialized by
         a connection-scoped asyncio lock (shared across every ChildrenDao
@@ -205,10 +212,10 @@ class ChildrenDao:
         transaction() rejects a second concurrent transaction with
         RuntimeError, so racing callers would crash instead of
         queueing.  The lock makes the second call wait, then run
-        against the first call's final state.
+        against the first call's final state — which also keeps the
+        completeness check honest: the re-read of the table happens
+        inside the same lock and transaction as the writes.
         """
-        if not ordered_ids:
-            return
         if len(set(ordered_ids)) != len(ordered_ids):
             raise ValueError(
                 "reorder() received duplicate child ids; every id must "
@@ -216,6 +223,29 @@ class ChildrenDao:
             )
         async with _connection_lock(self._database):
             async with self._database.transaction():
+                rows = await self._database.fetch_all(
+                    f"SELECT id FROM children"
+                )
+                existing = {row[0] for row in rows}
+                listed = set(ordered_ids)
+                missing = sorted(existing - listed)
+                unknown = sorted(listed - existing)
+                if missing or unknown:
+                    details = []
+                    if missing:
+                        details.append(
+                            f"missing child ids {missing} (a partial "
+                            "reorder would leave them with stale "
+                            "positions)"
+                        )
+                    if unknown:
+                        details.append(
+                            f"unknown child ids {unknown}"
+                        )
+                    raise ValueError(
+                        "reorder() must list every child exactly once; "
+                        + "; ".join(details)
+                    )
                 offset = len(ordered_ids)
                 await self._database.execute_many(
                     "UPDATE children SET sort_order = ? WHERE id = ?",

custom_components/nestquest/dao_instances.py

@@ -1,4 +1,4 @@
-"""Typed DAO layer for task_instances and the append-only completion_events.
+"""Typed DAO layer for quest_instances and the append-only completion_events.
 
 All SQL for these tables lives in this module per the feature
 guardrails: callers get typed dataclasses back and never see raw rows
@@ -14,12 +14,14 @@ separate reversal event, and instance state derives from the latest
 event.  A package-wide guard test (bottom of the test module) keeps
 mutation SQL for this table out of every other module.
 
-Upsert semantics: ``TaskInstancesDao.upsert`` is idempotent on
+Upsert semantics: ``QuestInstancesDao.upsert`` is idempotent on
-(definition_id, due_date) — re-running materialization for a date
+(definition_id, child_id, due_date, window) — re-running
-whose instance already exists updates nothing that matters
+materialization for a tuple whose instance already exists refreshes
-(child_id/due_time are refreshed only in the degenerate re-generation
+only the generation-time snapshot columns and never duplicates a row.
-case where they changed) and never duplicates a row.  The schema's
+The schema's UNIQUE(definition_id, child_id, due_date, window) is the
-UNIQUE(definition_id, due_date) is the last line of defense.
+last line of defense.  The upsert refuses a child that is not an
+assignee of the definition and a window outside ``const.QUEST_WINDOWS``
+(D-008 multi-assignee, multi-window model).
 
 ``delete_future_uncompleted`` implements the Feature 06/07 rule that
 reassignment and schedule edits regenerate future instances: it
@@ -35,7 +37,7 @@ import datetime
 from dataclasses import dataclass
 
 from .dao_children import _connection_lock
-from .dao_rules import _validate_date
+from .dao_rules import _validate_date, _validate_window
 from .db import NestQuestDatabase
 
 #: Timestamp policy for completion_events.occurred_at: strict UTC
@@ -82,12 +84,13 @@ def _next_day_start(date_str: str) -> str:
 
 
 @dataclass(frozen=True)
-class TaskInstanceRecord:
+class QuestInstanceRecord:
-    """One row of ``task_instances``."""
+    """One row of ``quest_instances``."""
 
     id: int
     definition_id: int
     child_id: int
+    window: str
     due_date: str
     due_time: str | None
     generated_at: str
@@ -103,16 +106,17 @@ class CompletionEventRecord:
     event_type: str
     actor_source: str
     actor_user_id: str | None
+    actor_child_id: int | None
     occurred_at: str
     was_on_time: bool | None
 
 
 _INSTANCE_COLUMNS = (
-    "id, definition_id, child_id, due_date, due_time, generated_at"
+    "id, definition_id, child_id, window, due_date, due_time, generated_at"
 )
 _EVENT_COLUMNS = (
     "id, instance_id, child_id, event_type, actor_source, actor_user_id, "
-    "occurred_at, was_on_time"
+    "actor_child_id, occurred_at, was_on_time"
 )
 
 #: The only event types the append path accepts; the schema CHECK is
@@ -121,14 +125,15 @@ EVENT_COMPLETED = "completed"
 EVENT_UNCOMPLETED = "uncompleted"
 
 
-def _instance_from_row(row: tuple) -> TaskInstanceRecord:
+def _instance_from_row(row: tuple) -> QuestInstanceRecord:
-    return TaskInstanceRecord(
+    return QuestInstanceRecord(
         id=row[0],
         definition_id=row[1],
         child_id=row[2],
-        due_date=row[3],
+        window=row[3],
-        due_time=row[4],
+        due_date=row[4],
-        generated_at=row[5],
+        due_time=row[5],
+        generated_at=row[6],
     )
 
 
@@ -140,13 +145,14 @@ def _event_from_row(row: tuple) -> CompletionEventRecord:
         event_type=row[3],
         actor_source=row[4],
         actor_user_id=row[5],
-        occurred_at=row[6],
+        actor_child_id=row[6],
-        was_on_time=None if row[7] is None else bool(row[7]),
+        occurred_at=row[7],
+        was_on_time=None if row[8] is None else bool(row[8]),
     )
 
 
-class TaskInstancesDao:
+class QuestInstancesDao:
-    """Typed async access to the ``task_instances`` table."""
+    """Typed async access to the ``quest_instances`` table."""
 
     def __init__(self, database: NestQuestDatabase) -> None:
         self._database = database
@@ -158,31 +164,40 @@ class TaskInstancesDao:
         due_date: str,
         generated_at: str,
         *,
+        window: str,
         due_time: str | None = None,
-    ) -> TaskInstanceRecord:
+    ) -> QuestInstanceRecord:
-        """Create the dated instance for (definition_id, due_date) idempotently.
+        """Create the dated instance for the widened key idempotently.
 
-        Running materialization twice for the same definition/date
+        The instance key is (definition_id, child_id, due_date, window)
-        yields ONE row: the ON CONFLICT path refreshes the
+        (D-008): running materialization twice for the same tuple
-        generation-time snapshot columns (child_id, due_time) and
+        yields ONE row — the ON CONFLICT path refreshes the
-        regenerates the stamp, but can never create a duplicate — the
+        generation-time snapshot columns (due_time) and regenerates the
-        schema UNIQUE(definition_id, due_date) backs this up.
+        stamp, but can never create a duplicate.  The schema's
+        UNIQUE(definition_id, child_id, due_date, window) backs this
+        up, so the twice-daily case (two windows, one child, one date)
+        yields two rows and the shared case (three assignees, one
+        window) yields three.
 
         Guardrails enforced here:
 
         - ``due_date`` must be TODAY or later: instances are never
           generated in the past (Feature 07).  The caller's
           materialization horizon is future-dated by definition.
+        - ``window`` must be one of ``const.QUEST_WINDOWS``.
         - An existing instance WITH a completion event is IMMUTABLE:
           the conflict path must not rewrite its snapshot columns
           (Feature 07/08).  The upsert refuses with ValueError in that
           case — silently no-oping would mask a materialization bug.
+        - The child must be an assignee of the definition (D-008
+          multi-assignee model).
 
         The definition must exist (validated atomically under the
         connection lock so a concurrent definition deactivation or
         child deletion cannot slip between check and insert).
         """
         _validate_date(due_date, "due_date")
+        _validate_window(window)
         # Fail fast on obviously-past dates BEFORE queuing on the
         # connection lock (a nice early error); the AUTHORITATIVE
         # check runs after the lock is acquired, because a caller can
@@ -207,82 +222,92 @@ class TaskInstancesDao:
                         f"{today!r}"
                     )
                 definition = await self._database.fetch_one(
-                    "SELECT child_id FROM task_definitions WHERE id = ?",
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
                     (definition_id,),
                 )
                 if definition is None:
                     raise ValueError(
-                        f"task definition {definition_id} does not exist"
+                        f"quest definition {definition_id} does not exist"
                     )
-                if definition[0] != child_id:
+                assignee = await self._database.fetch_one(
-                    raise ValueError(
+                    "SELECT 1 FROM quest_definition_assignees "
-                        f"task definition {definition_id} is assigned to "
+                    "WHERE definition_id = ? AND child_id = ?",
-                        f"child {definition[0]}, not {child_id}; instance "
+                    (definition_id, child_id),
-                        "must carry the definition's assignee"
-                    )
-                child = await self._database.fetch_one(
-                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
                 )
-                if child is None:
+                if assignee is None:
-                    raise ValueError(f"child {child_id} does not exist")
+                    raise ValueError(
+                        f"quest definition {definition_id} is not "
+                        f"assigned to child {child_id}; instance must "
+                        "carry an assigned child"
+                    )
                 existing = await self._database.fetch_one(
                     "SELECT 1 FROM completion_events "
-                    "WHERE instance_id = (SELECT id FROM task_instances "
+                    "WHERE instance_id = (SELECT id FROM quest_instances "
-                    "WHERE definition_id = ? AND due_date = ?)",
+                    "WHERE definition_id = ? AND child_id = ? "
-                    (definition_id, due_date),
+                    "AND due_date = ? AND window = ?)",
+                    (definition_id, child_id, due_date, window),
                 )
                 if existing is not None:
                     raise ValueError(
                         f"instance for (definition {definition_id}, "
-                        f"due_date {due_date!r}) already has a completion "
+                        f"child {child_id}, due_date {due_date!r}, "
+                        f"window {window!r}) already has a completion "
                         "event and is immutable; it cannot be regenerated"
                     )
                 await self._database.execute(
-                    "INSERT INTO task_instances (definition_id, child_id, "
+                    "INSERT INTO quest_instances (definition_id, child_id, "
-                    "due_date, due_time, generated_at) "
+                    "window, due_date, due_time, generated_at) "
-                    "VALUES (?, ?, ?, ?, ?) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
-                    "ON CONFLICT (definition_id, due_date) DO UPDATE SET "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
-                    "child_id = excluded.child_id, "
+                    "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, due_date)
+                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, due_date: str
+        self,
-    ) -> TaskInstanceRecord | None:
+        definition_id: int,
-        """Return the instance for (definition_id, due_date), or None."""
+        child_id: int,
+        due_date: str,
+        window: str,
+    ) -> QuestInstanceRecord | None:
+        """Return the instance for the widened key tuple, or None."""
         row = await self._database.fetch_one(
-            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
-            "WHERE definition_id = ? AND due_date = ?",
+            "WHERE definition_id = ? AND child_id = ? AND due_date = ? "
-            (definition_id, due_date),
+            "AND window = ?",
+            (definition_id, child_id, due_date, window),
         )
         return _instance_from_row(row) if row is not None else None
 
-    async def get_by_id(self, instance_id: int) -> TaskInstanceRecord | None:
+    async def get_by_id(self, instance_id: int) -> QuestInstanceRecord | None:
         """Return the instance with ``instance_id``, or None."""
         row = await self._database.fetch_one(
-            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances WHERE id = ?",
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances WHERE id = ?",
             (instance_id,),
         )
         return _instance_from_row(row) if row is not None else None
 
     async def list_by_child_and_date(
         self, child_id: int, due_date: str
-    ) -> list[TaskInstanceRecord]:
+    ) -> list[QuestInstanceRecord]:
         """Return the child's instances due exactly on ``due_date``."""
         _validate_date(due_date, "due_date")
         rows = await self._database.fetch_all(
-            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
             "WHERE child_id = ? AND due_date = ? ORDER BY id",
             (child_id, due_date),
         )
@@ -293,7 +318,7 @@ class TaskInstancesDao:
         child_id: int,
         range_start: str,
         range_end: str,
-    ) -> list[TaskInstanceRecord]:
+    ) -> list[QuestInstanceRecord]:
         """Return the child's instances with due_date in the closed range."""
         _validate_date(range_start, "range_start")
         _validate_date(range_end, "range_end")
@@ -303,7 +328,7 @@ class TaskInstancesDao:
                 f"{range_end!r} < {range_start!r}"
             )
         rows = await self._database.fetch_all(
-            f"SELECT {_INSTANCE_COLUMNS} FROM task_instances "
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
             "WHERE child_id = ? AND due_date >= ? AND due_date <= ? "
             "ORDER BY due_date, id",
             (child_id, range_start, range_end),
@@ -338,10 +363,10 @@ class TaskInstancesDao:
                 today = datetime.date.today().isoformat()
                 effective_cutoff = max(cutoff_date, today)
                 result = await self._database.execute(
-                    "DELETE FROM task_instances WHERE definition_id = ? "
+                    "DELETE FROM quest_instances WHERE definition_id = ? "
                     "AND due_date >= ? AND NOT EXISTS ("
                     "  SELECT 1 FROM completion_events "
-                    "  WHERE instance_id = task_instances.id"
+                    "  WHERE instance_id = quest_instances.id"
                     ")",
                     (definition_id, effective_cutoff),
                 )
@@ -373,19 +398,21 @@ class CompletionEventsDao:
         was_on_time: bool | None,
         *,
         actor_user_id: str | None = None,
+        actor_child_id: int | None = None,
     ) -> CompletionEventRecord:
         """Append one completion or un-completion event.
 
-        ``actor_source`` is 'user' (with ``actor_user_id`` set) or
+        ``actor_source`` is 'user' (an admin action: ``actor_user_id``
-        'panel' (no user id — the tapped child is ``child_id``); the
+        set, ``actor_child_id`` NULL) or 'panel' (a panel tap: no user
-        actor-pair CHECK enforces this shape at the schema level.
+        id, and ``actor_child_id`` carrying the tapped child profile,
-        ``occurred_at`` must be the module's one strict UTC shape
+        D-008); the actor-pair CHECKs enforce this shape at the schema
-        (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range queries
+        level.  ``occurred_at`` must be the module's one strict UTC
-        stay exact.  ``was_on_time`` is required (True/False) for
+        shape (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range
-        completions; for un-completions it may be None or the reversed
+        queries stay exact.  ``was_on_time`` is required (True/False)
-        completion's flag (Feature 08 still deciding).  The instance
+        for completions; for un-completions it may be None or the
-        and child must exist — validated atomically under the
+        reversed completion's flag (Feature 08 still deciding).  The
-        connection lock.
+        instance, child and — for panel events — the tapped child must
+        exist — validated atomically under the connection lock.
         """
         if event_type not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
             raise ValueError(
@@ -405,6 +432,15 @@ class CompletionEventsDao:
             raise ValueError(
                 "actor_source 'panel' must not carry actor_user_id"
             )
+        if actor_source == "panel" and actor_child_id is None:
+            raise ValueError(
+                "actor_source 'panel' requires actor_child_id "
+                "(the tapped child profile)"
+            )
+        if actor_source == "user" and actor_child_id is not None:
+            raise ValueError(
+                "actor_source 'user' must not carry actor_child_id"
+            )
         if was_on_time is not None and type(was_on_time) is not bool:
             raise ValueError(
                 f"was_on_time must be True, False or None, got "
@@ -418,16 +454,16 @@ class CompletionEventsDao:
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 instance = await self._database.fetch_one(
-                    "SELECT child_id FROM task_instances WHERE id = ?",
+                    "SELECT child_id FROM quest_instances WHERE id = ?",
                     (instance_id,),
                 )
                 if instance is None:
                     raise ValueError(
-                        f"task instance {instance_id} does not exist"
+                        f"quest instance {instance_id} does not exist"
                     )
                 if instance[0] != child_id:
                     raise ValueError(
-                        f"task instance {instance_id} belongs to child "
+                        f"quest instance {instance_id} belongs to child "
                         f"{instance[0]}, not {child_id}"
                     )
                 child = await self._database.fetch_one(
@@ -435,16 +471,28 @@ class CompletionEventsDao:
                 )
                 if child is None:
                     raise ValueError(f"child {child_id} does not exist")
+                if actor_child_id is not None:
+                    tapped = await self._database.fetch_one(
+                        "SELECT 1 FROM children WHERE id = ?",
+                        (actor_child_id,),
+                    )
+                    if tapped is None:
+                        raise ValueError(
+                            f"actor_child_id {actor_child_id} does not "
+                            "reference an existing child"
+                        )
                 result = await self._database.execute(
                     "INSERT INTO completion_events (instance_id, child_id, "
-                    "event_type, actor_source, actor_user_id, occurred_at, "
+                    "event_type, actor_source, actor_user_id, "
-                    "was_on_time) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    "actor_child_id, occurred_at, was_on_time) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         instance_id,
                         child_id,
                         event_type,
                         actor_source,
                         actor_user_id,
+                        actor_child_id,
                         occurred_at,
                         None if was_on_time is None else int(was_on_time),
                     ),
@@ -493,7 +541,7 @@ class CompletionEventsDao:
             )
         if due_date_scope:
             predicate = (
-                "SELECT id FROM task_instances WHERE child_id = ? "
+                "SELECT id FROM quest_instances WHERE child_id = ? "
                 "AND due_date >= ? AND due_date <= ?"
             )
             parameters: list[object] = [child_id, range_start, range_end]

custom_components/nestquest/dao_rules.py

@@ -1,37 +1,44 @@
-"""Typed DAO layer for the schedule_rules and task_definitions tables.
+"""Typed DAO layer for the schedule_rules, quest_definitions and
+quest_definition_assignees tables.
 
-All SQL for ``schedule_rules`` and ``task_definitions`` lives in this
+All SQL for ``schedule_rules``, ``quest_definitions`` and
-module per the feature guardrails: callers get typed dataclasses back
+``quest_definition_assignees`` lives in this module per the feature
-and never see raw rows or SQL.  Every method is async and runs through
+guardrails: callers get typed dataclasses back and never see raw rows
+or SQL.  Every method is async and runs through
 :class:`~.db.NestQuestDatabase`, so each statement executes on the HA
 executor and the event loop never blocks.
 
 Guardrail mapping:
 
 - ``schedule_rules`` deletion is delete-if-unreferenced: a rule still
-  referenced by any task definition is rejected (the done-condition),
+  referenced by any quest definition is rejected (the done-condition),
   not cascaded.
-- ``task_definitions`` has no delete path at all: definitions are
+- ``quest_definitions`` has no delete path at all: definitions are
   deactivated, never hard-deleted, and history must survive.
-- ``set_assignee`` changes future instances only — the DAO writes the
+- Assignment is multi-assignee (D-008): :meth:`add_assignee` and
-  definition row and nothing else; reassignment never rewrites
+  :meth:`remove_assignee` change future instances only — the DAO writes
+  the assignee table and nothing else; assignment edits never rewrite
   existing instances or completion history (Feature 06 guardrail).
 - Validation of rule well-formedness and child activeness lives partly
   in the schema CHECKs (rule-type coherence) and partly in
-  :meth:`TaskDefinitionsDao.create` (active child, rule existence);
+  :meth:`QuestDefinitionsDao.create` (rule existence) and
-  no permission checks here, that is the Feature 09 gate's job.
+  :meth:`QuestDefinitionsDao.add_assignee` (active child); no
+  permission checks here, that is the Feature 09 gate's job.
 
-Concurrency: ``delete_rule_if_unreferenced`` performs its reference
+Concurrency: ``delete_rule_if_unreferenced`` and the assignee
-check and DELETE inside one transaction under the connection-scoped
+mutations perform their checks and writes inside one transaction under
-lock shared with :mod:`.dao_children`, so a definition referencing the
+the connection-scoped lock shared with :mod:`.dao_children`, so a
-rule cannot appear between check and delete on this connection.
+definition referencing the rule (or an assignee appearing) cannot slip
+in between check and write on this connection.
 """
 from __future__ import annotations
 
 import datetime
 from dataclasses import dataclass
 
-from .dao_children import _connection_lock
+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
 
 #: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
@@ -65,6 +72,38 @@ def _validate_date(value: str, field: str) -> None:
         )
 
 
+_TIME_FORMAT = "%H:%M"
+
+
+def _validate_time(value: str, field: str) -> None:
+    """Raise ValueError unless ``value`` is a strict 24-hour HH:MM.
+
+    Same strictness policy as _validate_date: parse with the exact
+    format, then round-trip so '9:30' (non-padded) is rejected.  Time
+    shapes are caller-side policy (the schema stores TEXT): windows'
+    optional ``due_time`` and definition ``due_time`` both use this.
+    """
+    try:
+        parsed = datetime.datetime.strptime(value, _TIME_FORMAT).time()
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"{field} must be a 24-hour HH:MM time, got {value!r}"
+        ) from None
+    if parsed.strftime(_TIME_FORMAT) != value:
+        raise ValueError(
+            f"{field} must be a strict HH:MM time, got {value!r}"
+        )
+
+
+def _validate_window(window: str) -> None:
+    """Raise ValueError unless ``window`` is a ``const`` window name."""
+    if window not in QUEST_WINDOWS:
+        raise ValueError(
+            f"window must be one of {', '.join(QUEST_WINDOWS)}, "
+            f"got {window!r}"
+        )
+
+
 @dataclass(frozen=True)
 class ScheduleRuleRecord:
     """One row of ``schedule_rules``."""
@@ -81,28 +120,37 @@ class ScheduleRuleRecord:
 
 
 @dataclass(frozen=True)
-class TaskDefinitionRecord:
+class QuestDefinitionRecord:
-    """One row of ``task_definitions``."""
+    """One row of ``quest_definitions`` (without its assignees)."""
 
     id: int
     title: str
     description: str | None
     icon: str | None
-    child_id: int
     schedule_rule_id: int
     due_time: str | None
     is_active: bool
     created_at: str
 
 
+@dataclass(frozen=True)
+class QuestDefinitionWindowRecord:
+    """One row of ``quest_definition_windows``."""
+
+    definition_id: int
+    window: str
+    due_time: str | None
+
+
 _RULE_COLUMNS = (
     "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
     "month, start_date, end_date"
 )
 _DEFINITION_COLUMNS = (
-    "id, title, description, icon, child_id, schedule_rule_id, due_time, "
+    "id, title, description, icon, schedule_rule_id, due_time, "
     "is_active, created_at"
 )
+_WINDOW_COLUMNS = "definition_id, window, due_time"
 
 
 def _rule_from_row(row: tuple) -> ScheduleRuleRecord:
@@ -119,17 +167,24 @@ def _rule_from_row(row: tuple) -> ScheduleRuleRecord:
     )
 
 
-def _definition_from_row(row: tuple) -> TaskDefinitionRecord:
+def _definition_from_row(row: tuple) -> QuestDefinitionRecord:
-    return TaskDefinitionRecord(
+    return QuestDefinitionRecord(
         id=row[0],
         title=row[1],
         description=row[2],
         icon=row[3],
-        child_id=row[4],
+        schedule_rule_id=row[4],
-        schedule_rule_id=row[5],
+        due_time=row[5],
-        due_time=row[6],
+        is_active=bool(row[6]),
-        is_active=bool(row[7]),
+        created_at=row[7],
-        created_at=row[8],
+    )
+
+
+def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
+    return QuestDefinitionWindowRecord(
+        definition_id=row[0],
+        window=row[1],
+        due_time=row[2],
     )
 
 
@@ -258,7 +313,7 @@ class ScheduleRulesDao:
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 row = await self._database.fetch_one(
-                    "SELECT 1 FROM task_definitions "
+                    "SELECT 1 FROM quest_definitions "
                     "WHERE schedule_rule_id = ? LIMIT 1",
                     (rule_id,),
                 )
@@ -271,12 +326,15 @@ class ScheduleRulesDao:
                 return result.rowcount > 0
 
 
-class TaskDefinitionsDao:
+class QuestDefinitionsDao:
-    """Typed async access to the ``task_definitions`` table.
+    """Typed async access to ``quest_definitions`` and its assignees.
 
     No delete method exists: definitions are deactivated via
     :meth:`set_active`, never hard-deleted, so completion history keeps
-    its references (feature guardrail).
+    its references (feature guardrail).  Assignment is multi-assignee
+    (D-008): the definition row carries no child; assignees live in
+    ``quest_definition_assignees`` and are managed explicitly so the
+    operations stay separately loggable and permission-gateable.
     """
 
     def __init__(self, database: NestQuestDatabase) -> None:
@@ -285,7 +343,6 @@ class TaskDefinitionsDao:
     async def create(
         self,
         title: str,
-        child_id: int,
         schedule_rule_id: int,
         created_at: str,
         *,
@@ -293,53 +350,55 @@ class TaskDefinitionsDao:
         icon: str | None = None,
         due_time: str | None = None,
         is_active: bool = True,
-    ) -> TaskDefinitionRecord:
+        assignee_child_ids: list[int] | None = None,
+    ) -> QuestDefinitionRecord:
         """Insert one definition and return the record as stored.
 
-        Validates the assignment target as part of the same serialized
+        Validates the schedule rule as part of the same serialized
-        transaction as the insert: the child must exist and be active,
+        transaction as the insert: the rule must exist AT INSERT TIME —
-        and the rule must exist, AT INSERT TIME — a child deactivated
+        a rule deleted concurrently cannot slip between the check and
-        or a rule deleted concurrently cannot slip between the checks
+        the INSERT.  ``assignee_child_ids`` (each child must exist and
-        and the INSERT, which would otherwise create an invalid
+        be active) is applied in the same transaction, so a definition
-        assignment or surface as a raw FK IntegrityError instead of
+        can never be observed with a stale or missing assignee set; an
-        the documented ValueError.  An INACTIVE child would pass the
+        empty list or None creates an unassigned definition, which
-        schema foreign keys, which is why the activeness check lives
+        callers may fill via :meth:`add_assignee`.
-        here.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
-                await self._validate_assignable(
+                await self._validate_rule_exists(schedule_rule_id)
-                    child_id, schedule_rule_id
+                for child_id in assignee_child_ids or []:
-                )
+                    await self._validate_assignable_child(child_id)
                 result = await self._database.execute(
-                    "INSERT INTO task_definitions (title, description, "
+                    "INSERT INTO quest_definitions (title, description, "
-                    "icon, child_id, schedule_rule_id, due_time, "
+                    "icon, schedule_rule_id, due_time, is_active, "
-                    "is_active, created_at) "
+                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
-                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         title,
                         description,
                         icon,
-                        child_id,
                         schedule_rule_id,
                         due_time,
                         int(is_active),
                         created_at,
                     ),
                 )
-                definition = await self.get(result.lastrowid)
+                definition_id = result.lastrowid
+                for child_id in assignee_child_ids or []:
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_assignees "
+                        "(definition_id, child_id) VALUES (?, ?)",
+                        (definition_id, child_id),
+                    )
+                definition = await self.get(definition_id)
         assert definition is not None
         return definition
 
-    async def _validate_assignable(
+    async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
-        self, child_id: int, schedule_rule_id: int
+        """Raise ValueError unless the rule exists.
-    ) -> None:
-        """Raise ValueError unless child (active) and rule both exist.
 
         MUST be called inside the connection lock so the verdict cannot
         go stale before the caller's INSERT commits.
         """
-        await self._validate_assignable_child(child_id)
         rule = await self._database.fetch_one(
             "SELECT 1 FROM schedule_rules WHERE id = ?",
             (schedule_rule_id,),
@@ -349,28 +408,35 @@ class TaskDefinitionsDao:
                 f"schedule rule {schedule_rule_id} does not exist"
             )
 
-    async def get(self, definition_id: int) -> TaskDefinitionRecord | None:
+    async def get(self, definition_id: int) -> QuestDefinitionRecord | None:
         """Return the definition with ``definition_id``, or None."""
         row = await self._database.fetch_one(
-            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            f"SELECT {_DEFINITION_COLUMNS} FROM quest_definitions "
             "WHERE id = ?",
             (definition_id,),
         )
         return _definition_from_row(row) if row is not None else None
 
-    async def list_by_child(self, child_id: int) -> list[TaskDefinitionRecord]:
+    async def list_by_child(self, child_id: int) -> list[QuestDefinitionRecord]:
-        """Return all definitions assigned to ``child_id``, newest first."""
+        """Return all definitions assigned to ``child_id``, newest first.
+
+        Assignment is read through ``quest_definition_assignees``
+        (D-008): a definition appears once per child regardless of how
+        many assignees share it.
+        """
         rows = await self._database.fetch_all(
-            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            f"SELECT {_DEFINITION_COLUMNS} FROM quest_definitions "
-            "WHERE child_id = ? ORDER BY id DESC",
+            "WHERE id IN (SELECT definition_id "
+            "FROM quest_definition_assignees WHERE child_id = ?) "
+            "ORDER BY id DESC",
             (child_id,),
         )
         return [_definition_from_row(row) for row in rows]
 
-    async def list_active(self) -> list[TaskDefinitionRecord]:
+    async def list_active(self) -> list[QuestDefinitionRecord]:
         """Return all active definitions, oldest first (stable order)."""
         rows = await self._database.fetch_all(
-            f"SELECT {_DEFINITION_COLUMNS} FROM task_definitions "
+            f"SELECT {_DEFINITION_COLUMNS} FROM quest_definitions "
             "WHERE is_active = 1 ORDER BY id"
         )
         return [_definition_from_row(row) for row in rows]
@@ -391,8 +457,9 @@ class TaskDefinitionsDao:
         SQL NULL, so callers can remove optional metadata
         (description, icon, due_time).  Assignment and activation are
         deliberately NOT settable here: they have their own explicit
-        operations (:meth:`set_assignee`, :meth:`set_active`) so they
+        operations (:meth:`add_assignee`, :meth:`remove_assignee`,
-        stay separately loggable and permission-gateable.
+        :meth:`set_active`) so they stay separately loggable and
+        permission-gateable.
         """
         assignments: list[str] = []
         parameters: list[object] = []
@@ -409,7 +476,7 @@ class TaskDefinitionsDao:
             return 0
         parameters.append(definition_id)
         result = await self._database.execute(
-            f"UPDATE task_definitions SET {', '.join(assignments)} "
+            f"UPDATE quest_definitions SET {', '.join(assignments)} "
             "WHERE id = ?",
             tuple(parameters),
         )
@@ -422,37 +489,166 @@ class TaskDefinitionsDao:
         existing instances and completion history untouched.
         """
         result = await self._database.execute(
-            "UPDATE task_definitions SET is_active = ? WHERE id = ?",
+            "UPDATE quest_definitions SET is_active = ? WHERE id = ?",
             (int(is_active), definition_id),
         )
         return result.rowcount
 
-    async def set_assignee(
+    async def add_assignee(
         self, definition_id: int, child_id: int
-    ) -> int:
+    ) -> None:
-        """Reassign the definition to ``child_id``; rows updated (0 if absent).
+        """Assign ``child_id`` to the definition (idempotent).
 
-        Validates the new assignee inside the same serialized
+        Validates the assignee inside the same serialized transaction
-        transaction as the UPDATE (existing and active AT ASSIGNMENT
+        as the INSERT (existing and active AT ASSIGNMENT TIME), so a
-        TIME), so a child deactivated concurrently cannot become the
+        child deactivated concurrently cannot become an assignee.
-        assignee.  Changes future instances only: already-generated
+        Assigning an already-assigned child is a no-op, mirroring the
-        instances and completion history are never rewritten here.
+        composite primary key's storage-level guarantee.  Changes
+        future instances only: already-generated instances and
+        completion history are never rewritten here.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 await self._validate_assignable_child(child_id)
-                result = await self._database.execute(
+                definition = await self._database.fetch_one(
-                    "UPDATE task_definitions SET child_id = ? "
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
-                    "WHERE id = ?",
+                    (definition_id,),
-                    (child_id, definition_id),
                 )
-        return result.rowcount
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                await self._database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, ?) "
+                    "ON CONFLICT (definition_id, child_id) DO NOTHING",
+                    (definition_id, child_id),
+                )
+
+    async def remove_assignee(
+        self, definition_id: int, child_id: int
+    ) -> bool:
+        """Remove the assignee link; True when a row was removed.
+
+        Future materialization stops for this (definition, child) pair;
+        existing instances and completion history are never touched
+        here (Feature 06 guardrail).
+        """
+        async with _connection_lock(self._database):
+            result = await self._database.execute(
+                "DELETE FROM quest_definition_assignees "
+                "WHERE definition_id = ? AND child_id = ?",
+                (definition_id, child_id),
+            )
+        return result.rowcount > 0
+
+    async def list_assignees(self, definition_id: int) -> list[ChildRecord]:
+        """Return the definition's assignees as child records.
+
+        Ordered by the children table's stable display order
+        (``sort_order``, then id) so callers render a deterministic
+        roster.  A definition with no assignees returns [].
+        """
+        rows = await self._database.fetch_all(
+            f"SELECT {_CHILD_COLUMNS} FROM children "
+            "WHERE id IN (SELECT child_id "
+            "FROM quest_definition_assignees WHERE definition_id = ?) "
+            "ORDER BY sort_order, id",
+            (definition_id,),
+        )
+        return [_child_from_row(row) for row in rows]
+
+    async def upsert_window(
+        self,
+        definition_id: int,
+        window: str,
+        *,
+        due_time: str | None = None,
+    ) -> QuestDefinitionWindowRecord:
+        """Declare (or re-declare) ``window`` on the definition.
+
+        The composite primary key makes the write idempotent per
+        (definition, window): re-declaring updates ``due_time`` in
+        place instead of duplicating.  The window name must be one of
+        ``const.QUEST_WINDOWS`` (the schema CHECK backs this up) and a
+        provided ``due_time`` must be strict 24-hour HH:MM.  The
+        definition must exist — validated under the connection lock so
+        the verdict cannot go stale before the write.
+        """
+        _validate_window(window)
+        if due_time is not None:
+            _validate_time(due_time, "due_time")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definition = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
+                    (definition_id,),
+                )
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                await self._database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, ?) "
+                    "ON CONFLICT (definition_id, window) DO UPDATE SET "
+                    "due_time = excluded.due_time",
+                    (definition_id, window, due_time),
+                )
+                row = await self._database.fetch_one(
+                    f"SELECT {_WINDOW_COLUMNS} "
+                    "FROM quest_definition_windows "
+                    "WHERE definition_id = ? AND window = ?",
+                    (definition_id, window),
+                )
+        assert row is not None
+        return _window_from_row(row)
+
+    async def list_windows(
+        self, definition_id: int
+    ) -> list[QuestDefinitionWindowRecord]:
+        """Return the definition's windows in ``const`` order.
+
+        Ordered by the canonical window sequence (morning, afternoon,
+        evening) rather than insertion order, so callers always render
+        the Quest Log's three columns consistently.  A definition with
+        no windows returns [].
+        """
+        rows = await self._database.fetch_all(
+            f"SELECT {_WINDOW_COLUMNS} FROM quest_definition_windows "
+            "WHERE definition_id = ?",
+            (definition_id,),
+        )
+        by_name = {row[1]: _window_from_row(row) for row in rows}
+        return [
+            by_name[name]
+            for name in QUEST_WINDOWS
+            if name in by_name
+        ]
+
+    async def remove_window(
+        self, definition_id: int, window: str
+    ) -> bool:
+        """Remove the window declaration; True when a row was removed.
+
+        Future materialization stops producing instances for this
+        (definition, window); existing instances and completion history
+        are never touched here (Feature 06 guardrail).
+        """
+        _validate_window(window)
+        async with _connection_lock(self._database):
+            result = await self._database.execute(
+                "DELETE FROM quest_definition_windows "
+                "WHERE definition_id = ? AND window = ?",
+                (definition_id, window),
+            )
+        return result.rowcount > 0
 
     async def _validate_assignable_child(self, child_id: int) -> None:
         """Raise ValueError unless the child exists and is active.
 
         MUST be called inside the connection lock so the verdict cannot
-        go stale before the caller's UPDATE commits.
+        go stale before the caller's INSERT commits.
         """
         child = await self._database.fetch_one(
             "SELECT is_active FROM children WHERE id = ?", (child_id,)

custom_components/nestquest/migrations.py

@@ -14,6 +14,32 @@ Version 0 means "the file exists but carries no NestQuest schema yet"
 the runner against version 0 applies migration 1, which applies the
 full v1 DDL from :mod:`.schema` and stamps version 1.
 
+Migration 1 was rewritten pre-release (D-007): nothing was ever
+shipped, so the v1 DDL now creates the ``quest_*`` tables directly
+instead of the legacy ``task_*`` spellings, and no compatibility shim
+exists.  Dev machines that already hold a version-1 database with
+``task_*`` tables are handled by migration 2, which detects the legacy
+table names and ``ALTER TABLE ... RENAME TO``s them to their current
+names before anything else assumes the new model.  Migration 3 brings
+pre-D-008 databases to the multi-assignee model: it creates
+``quest_definition_assignees`` and, when ``quest_definitions`` still
+carries a single ``child_id`` column, rebuilds the table without it,
+copying each definition's assignee across.  Migration 4 adds the
+windows table, migration 5 rebuilds ``quest_instances`` onto the
+widened (definition_id, child_id, due_date, window) key, and migration
+6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
+column.  Each callable migration runs inside its own transaction
+together with its version stamp, so a crash mid-step rolls the
+statements and the stamp back together.
+
+Migration entries are either an ordered sequence of SQL statements or
+an async callable taking ``(database, target_version)``.  Callables
+exist for steps whose SQL depends on the database's current state (a
+legacy rename only fires when a legacy table is present; a rebuild only
+when the old column shape is present) and own their transaction so they
+can bracket it with connection pragmas the runner's uniform transaction
+cannot express.
+
 Each migration's statements and its version stamp run inside ONE
 transaction, and the stamp is written as DELETE + INSERT rather than
 UPDATE because a version-0 database has no version row to update yet.
@@ -43,13 +69,29 @@ from __future__ import annotations
 
 import asyncio
 import logging
-from collections.abc import Sequence
+from collections.abc import Awaitable, Callable, Sequence
+from typing import Union
 
 from .db import NestQuestDatabase
-from .schema import SCHEMA_V1_STATEMENTS
+from .schema import (
+    COMPLETION_EVENTS_TABLE_SQL,
+    QUEST_DEFINITIONS_TABLE_SQL,
+    QUEST_INSTANCES_TABLE_SQL,
+    SCHEMA_V1_STATEMENTS,
+    SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+)
 
 LOGGER = logging.getLogger(__name__)
 
+#: One migration entry: either an ordered sequence of SQL statements or
+#: an async callable taking ``(database, target_version)`` that manages
+#: its own transaction and stamps via :func:`stamp_schema_version`.
+MigrationStep = Union[
+    Sequence[str],
+    Callable[[NestQuestDatabase, int], Awaitable[None]],
+]
+
 #: Metadata table holding the applied schema version.  Prefixed with the
 #: integration domain so a config directory shared with other tools can
 #: never collide with a foreign ``schema_version`` table.
@@ -73,10 +115,305 @@ VERSION_TABLE_DDL = f"""
 #: version ``n-1`` to version ``n``.
 MIGRATION_1_V1_DDL: list[str] = list(SCHEMA_V1_STATEMENTS)
 
+#: The assignees table DDL, shared with migration 1 via the schema
+#: module so a rebuilt and a freshly created table cannot drift.
+_ASSIGNEES_TABLE_DDL: str = SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL[0]
+
+#: Legacy (pre-D-007) table name -> current table name.  Order matters:
+#: the parent table renames first so the child's foreign key reference
+#: is rewritten while its parent's new name is already in place.
+LEGACY_TASK_TABLE_RENAMES: tuple[tuple[str, str], ...] = (
+    ("task_definitions", "quest_definitions"),
+    ("task_instances", "quest_instances"),
+)
+
+
+async def stamp_schema_version(
+    database: NestQuestDatabase, version: int
+) -> None:
+    """Write the schema version stamp (inside the caller's transaction).
+
+    INSERT keyed on the singleton row (id = 1) with an upsert: a fresh
+    file has no row yet, so UPDATE alone would silently match zero rows
+    there and leave a migrated-but-unstamped database.
+    """
+    await database.execute(
+        f"INSERT INTO {VERSION_TABLE} (id, version) "
+        "VALUES (1, ?) "
+        "ON CONFLICT (id) DO UPDATE SET version = excluded.version",
+        (version,),
+    )
+
+
+async def _table_has_column(
+    database: NestQuestDatabase, table: str, column: str
+) -> bool:
+    """Return True when ``table`` currently carries ``column``."""
+    rows = await database.fetch_all(f"PRAGMA table_info({table})")
+    return any(row[1] == column for row in rows)
+
+
+#: Window assigned to instances migrated from the pre-window instance
+#: shape (D-008).  Pre-window rows were generated for whole-day
+#: obligations; the morning window is the documented default landing
+#: spot, and future dates are regenerated with real window declarations
+#: by materialization (delete_future_uncompleted + re-run).
+LEGACY_MIGRATED_WINDOW = "morning"
+
+
+async def _rename_legacy_task_tables(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Rename any surviving ``task_*`` tables to their ``quest_*`` names.
+
+    Dev-machines-only path (D-007): a database stamped version 1 by the
+    pre-rename runner holds ``task_definitions``/``task_instances``.
+    Each legacy name still present is renamed; fresh databases find
+    nothing and this step is a no-op.  Modern SQLite rewrites foreign
+    key clauses in other tables to follow the rename, so the renamed
+    schema is indistinguishable from a freshly created one apart from
+    internal autoindex names.  Runs in its own transaction together
+    with the version stamp, so a crash rolls both back.
+    """
+    async with database.transaction():
+        for legacy_name, current_name in LEGACY_TASK_TABLE_RENAMES:
+            row = await database.fetch_one(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' "
+                "AND name = ?",
+                (legacy_name,),
+            )
+            if row is not None:
+                await database.execute(
+                    f'ALTER TABLE "{legacy_name}" RENAME TO "{current_name}"'
+                )
+                LOGGER.info(
+                    "Renamed legacy table %s to %s",
+                    legacy_name,
+                    current_name,
+                )
+        await stamp_schema_version(database, target_version)
+
+
+async def _add_definition_assignees(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring definitions to the multi-assignee model (D-008).
+
+    Two shapes arrive here:
+
+    - Fresh or renamed databases whose ``quest_definitions`` already
+      matches the current canonical DDL (no ``child_id`` column): only
+      the assignees table needs creating (``IF NOT EXISTS`` is a no-op
+      when migration 1 already made it).
+    - Databases stamped at version 2 whose ``quest_definitions`` still
+      carries the single ``child_id`` column: the table is rebuilt via
+      SQLite's canonical procedure — create the new table, copy rows,
+      move each definition's assignee into
+      ``quest_definition_assignees``, drop the old table, rename — so
+      ``DROP COLUMN``'s foreign-key restriction never applies and no
+      definition or instance row is lost.
+
+    The copy runs with ``foreign_keys = OFF`` because dropping the old
+    table under enforcement would fire the implicit DELETE and violate
+    the ``quest_instances`` foreign key; ``PRAGMA foreign_key_check``
+    runs inside the same transaction before the stamp, so a corrupt
+    copy aborts the whole migration instead of persisting.  The pragma
+    is a connection-level no-op inside a transaction, hence the
+    off/transaction/on bracket.
+    """
+    legacy_child_id = await _table_has_column(
+        database, "quest_definitions", "child_id"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            await database.execute(_ASSIGNEES_TABLE_DDL)
+            if legacy_child_id:
+                await database.execute(
+                    "CREATE TABLE quest_definitions_rebuilt "
+                    f"{QUEST_DEFINITIONS_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO quest_definitions_rebuilt "
+                    "(id, title, description, icon, schedule_rule_id, "
+                    "due_time, is_active, created_at) "
+                    "SELECT id, title, description, icon, "
+                    "schedule_rule_id, due_time, is_active, created_at "
+                    "FROM quest_definitions ORDER BY id"
+                )
+                await database.execute(
+                    "INSERT OR IGNORE INTO quest_definition_assignees "
+                    "(definition_id, child_id) "
+                    "SELECT id, child_id FROM quest_definitions"
+                )
+                await database.execute("DROP TABLE quest_definitions")
+                await database.execute(
+                    "ALTER TABLE quest_definitions_rebuilt "
+                    "RENAME TO quest_definitions"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "definition rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
+async def _widen_instance_key(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``quest_instances`` to the D-008 instance key.
+
+    Databases whose instance table still carries the pre-window shape
+    (no ``window`` column, UNIQUE on (definition_id, due_date)) are
+    rebuilt via SQLite's canonical procedure into the widened shape —
+    UNIQUE (definition_id, child_id, due_date, window) — with every
+    migrated row landing in :data:`LEGACY_MIGRATED_WINDOW` so existing
+    ids (and the completion events referencing them) survive intact.
+    The old rows were unique per (definition, due_date), so pinning one
+    window cannot collide with the widened key.  Fresh databases whose
+    table already matches the canonical shape only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` (dropping the old
+    table under enforcement would fire the implicit DELETE and violate
+    the ``completion_events`` foreign key) and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp, so a corrupt copy aborts the whole migration.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "quest_instances", "window"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE quest_instances_rebuilt "
+                    f"{QUEST_INSTANCES_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO quest_instances_rebuilt "
+                    "(id, definition_id, child_id, window, due_date, "
+                    "due_time, generated_at) "
+                    "SELECT id, definition_id, child_id, ?, due_date, "
+                    "due_time, generated_at "
+                    "FROM quest_instances ORDER BY id",
+                    (LEGACY_MIGRATED_WINDOW,),
+                )
+                await database.execute("DROP TABLE quest_instances")
+                await database.execute(
+                    "ALTER TABLE quest_instances_rebuilt "
+                    "RENAME TO quest_instances"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "instance rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Widened quest_instances key; migrated rows pinned "
+                    "to the %r window",
+                    LEGACY_MIGRATED_WINDOW,
+                )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
+async def _add_actor_child_id(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``completion_events`` to the D-008 actor-child shape.
+
+    The ``actor_child_id`` column records which panel profile was
+    tapped; it cannot be added with ``ALTER TABLE ADD COLUMN`` alone
+    because the canonical shape also carries an actor-pair coherence
+    CHECK (panel events require it, user events forbid it), and CHECK
+    constraints cannot be added to an existing table.  Databases with
+    the legacy shape are therefore rebuilt via SQLite's canonical
+    procedure, backfilling each legacy panel row's ``actor_child_id``
+    to ``child_id`` — pre-D-008 panel completions were always recorded
+    against the tapped child themselves — while admin rows stay NULL.
+    Row ids are preserved so ``quest_instances`` references and the
+    append-only history stay intact.  Fresh databases already match
+    the canonical shape and only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp.  The copy is an INSERT..SELECT, not an UPDATE: the
+    completion_events table keeps its append-only guarantee.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "completion_events", "actor_child_id"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE completion_events_rebuilt "
+                    f"{COMPLETION_EVENTS_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO completion_events_rebuilt "
+                    "(id, instance_id, child_id, event_type, actor_source, "
+                    "actor_user_id, actor_child_id, occurred_at, "
+                    "was_on_time) "
+                    "SELECT id, instance_id, child_id, event_type, "
+                    "actor_source, actor_user_id, "
+                    "CASE WHEN actor_source = 'panel' THEN child_id "
+                    "ELSE NULL END, "
+                    "occurred_at, was_on_time "
+                    "FROM completion_events ORDER BY id"
+                )
+                await database.execute("DROP TABLE completion_events")
+                await database.execute(
+                    "ALTER TABLE completion_events_rebuilt "
+                    "RENAME TO completion_events"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "completion_events rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Added actor_child_id to completion_events; legacy "
+                    "panel rows backfilled to their own child_id"
+                )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
 #: The ordered migration list.  Append-only: never edit an applied
-#: entry, add the next one instead.
+#: entry, add the next one instead.  (Migration 1's content was
-MIGRATIONS: Sequence[Sequence[str]] = [
+#: rewritten pre-release per D-007 — no version 1 database shipped.)
+MIGRATIONS: Sequence[MigrationStep] = [
     MIGRATION_1_V1_DDL,
+    _rename_legacy_task_tables,
+    _add_definition_assignees,
+    # Migration 4 (D-008): the windows table is a plain new table — a
+    # CREATE IF NOT EXISTS suffices for both fresh databases (which
+    # already made it in migration 1) and pre-window databases.
+    list(SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL),
+    # Migration 5 (D-008): rebuild quest_instances onto the widened
+    # (definition_id, child_id, due_date, window) key.
+    _widen_instance_key,
+    # Migration 6 (D-008): rebuild completion_events with the
+    # actor_child_id column and its actor-pair coherence CHECK,
+    # backfilling legacy panel rows to their own child_id.
+    _add_actor_child_id,
 ]
 
 
@@ -140,7 +477,7 @@ def _migration_lock(database: NestQuestDatabase) -> asyncio.Lock:
 
 async def apply_migrations(
     database: NestQuestDatabase,
-    migrations: Sequence[Sequence[str]] = MIGRATIONS,
+    migrations: Sequence[MigrationStep] = MIGRATIONS,
 ) -> int:
     """Bring the database schema up to the latest version.
 
@@ -189,17 +526,18 @@ async def apply_migrations(
 
         for target in range(current + 1, latest + 1):
             applied = False
-            async with database.transaction():
+            step = migrations[target - 1]
-                # Re-read the version INSIDE the write transaction: the
+            if callable(step):
-                # outer read was advisory, and the wrapper serializes
+                # Callable migrations own their transaction: some steps
-                # individual statements, not the read-decide-apply
+                # (table rebuilds) must run connection pragmas such as
-                # decision.  Under the transaction's lock the version
+                # ``foreign_keys = OFF`` BEFORE opening one, which the
-                # cannot change between the read and the stamp, and the
+                # runner's uniform transaction cannot express, and they
-                # loser of an interleaved race (a queued second caller
+                # stamp inside their own transaction so a failure rolls
-                # re-running apply_migrations after the winner already
+                # the statements and the version stamp back together.
-                # migrated) re-reads here, sees the winner's stamp, and
+                # The version is re-read here under the migration lock:
-                # skips instead of re-applying a non-idempotent
+                # runners on one connection are serialized by it, so
-                # migration with a stale version.
+                # the version cannot change between this read and the
+                # callable's commit.
                 in_tx_version = await read_schema_version(database)
                 if in_tx_version > latest:
                     raise RuntimeError(
@@ -208,29 +546,39 @@ async def apply_migrations(
                         f"{latest}); downgrades are not supported"
                     )
                 if in_tx_version < target:
-                    for sql in migrations[target - 1]:
+                    await step(database, target)
-                        await database.execute(sql)
-                    # INSERT keyed on the singleton row (id = 1): a
-                    # fresh file has no row yet, and migration 1 is the
-                    # only migration a version-0 database can run, so
-                    # the row is seeded by the same statement batch
-                    # when needed.  UPDATE alone would silently match
-                    # zero rows there.
-                    await database.execute(
-                        f"INSERT INTO {VERSION_TABLE} (id, version) "
-                        "VALUES (1, ?) "
-                        "ON CONFLICT (id) DO UPDATE SET version = "
-                        "excluded.version",
-                        (target,),
-                    )
                     applied = True
+            else:
+                async with database.transaction():
+                    # Re-read the version INSIDE the write transaction:
+                    # the outer read was advisory, and the wrapper
+                    # serializes individual statements, not the
+                    # read-decide-apply decision.  Under the
+                    # transaction's lock the version cannot change
+                    # between the read and the stamp, and the loser of
+                    # an interleaved race (a queued second caller
+                    # re-running apply_migrations after the winner
+                    # already migrated) re-reads here, sees the
+                    # winner's stamp, and skips instead of re-applying
+                    # a non-idempotent migration with a stale version.
+                    in_tx_version = await read_schema_version(database)
+                    if in_tx_version > latest:
+                        raise RuntimeError(
+                            f"Database schema version {in_tx_version} is "
+                            f"newer than this integration understands "
+                            f"(latest known: {latest}); downgrades are "
+                            f"not supported"
+                        )
+                    if in_tx_version < target:
+                        for sql in step:
+                            await database.execute(sql)
+                        await stamp_schema_version(database, target)
+                        applied = True
+                        applied_statements = len(step)
             if applied:
                 LOGGER.info(
-                    "NestQuest schema migrated to version %d "
+                    "NestQuest schema migrated to version %d",
-                    "(%d statement%s)",
                     target,
-                    len(migrations[target - 1]),
-                    "s" if len(migrations[target - 1]) != 1 else "",
                 )
 
         return latest

custom_components/nestquest/schema.py

@@ -86,30 +86,54 @@ date range; a single-day override stores the same date in both columns.
 ``is_present`` 0/1 marks the child absent/present for the whole range;
 ``note`` is optional free text.
 
-Task instances: ``task_instances`` deliberately carries NO completion
+Multi-assignee definitions (D-008): ``quest_definitions`` carries no
+``child_id`` column.  Assignment lives in
+``quest_definition_assignees``, one row per (definition, child) with a
+composite primary key and foreign keys to the definition and the child,
+so "brush teeth" is one definition covering three children.  The
+migration runner rebuilds a pre-D-008 table into this shape and copies
+its single-assignee column across.  The composite primary key makes
+duplicate assignment impossible at the storage layer.
+
+Quest windows (D-008): ``quest_definition_windows`` declares which day
+windows a definition spans — one row per (definition, window), the
+composite primary key making a window idempotent per definition.
+``window`` is CHECK-constrained to the three spellings ``const`` owns;
+``due_time`` optionally pins a due time inside the window and is
+caller-validated (strict HH:MM) by the DAO layer, matching the date
+policy above.  The window clock ranges live in ``const.py``, not here:
+this layer stores what the caller declares and never interprets clocks.
+
+Quest instances: ``quest_instances`` deliberately carries NO completion
 status column.  An instance's current state derives from the latest row
 in ``completion_events`` (Feature 08), so materialization (Feature 07)
 can insert instances without knowing anything about completion and the
 append-only log stays the single source of truth.  ``definition_id``,
-``due_date`` and ``child_id`` snapshot the materialization-time facts
+``child_id``, ``due_date`` and ``window`` snapshot the
-("why this instance exists"): reassignment and rule edits change future
+materialization-time facts ("why this instance exists"): assignment
-instances only and never rewrite these rows.  ``due_time`` snapshots
+edits and rule edits change future instances only and never rewrite
-the definition's optional due time at generation for the same reason —
+these rows.  ``window`` records which declared day window produced the
-the instance keeps its generation-time value even if the definition is
+instance (D-008), CHECK-constrained to the same three spellings
-edited later.  ``generated_at`` is a caller-set UTC ISO-8601 timestamp
+``const`` owns.  ``due_time`` snapshots the window's optional due time
-per the module timestamp policy (no DB default, so a generation batch
+at generation for the same reason — the instance keeps its
-stamps one coherent time).
+generation-time value even if the definition is edited later.
+``generated_at`` is a caller-set UTC ISO-8601 timestamp per the module
+timestamp policy (no DB default, so a generation batch stamps one
+coherent time).
 
-Uniqueness: the (definition_id, due_date) pair is declared as a
+Uniqueness: the (definition_id, child_id, due_date, window) tuple is
-table-level ``UNIQUE`` constraint rather than a separate ``CREATE
+declared as a table-level ``UNIQUE`` constraint rather than a separate
-UNIQUE INDEX`` statement.  SQLite materializes it as an implicit unique
+``CREATE UNIQUE INDEX`` statement.  SQLite materializes it as an
-index (``sqlite_autoindex_task_instances_1``) — the required "unique
+implicit unique index (``sqlite_autoindex_quest_instances_1``) — the
-index prevents duplicate instances" — and the constraint form cannot be
+required "unique index prevents duplicate instances" — and the
-silently dropped the way a standalone index can, so the duplicate door
+constraint form cannot be silently dropped the way a standalone index
-stays shut.  Idempotent materialization keys on exactly this pair.  No
+can, so the duplicate door stays shut.  Idempotent materialization keys
-secondary indexes beyond it are declared yet: history and board queries
+on exactly this tuple: the twice-daily case (one definition, two
-(Features 08/10/13) should justify their indexes as a later migration
+windows, one child, one date) yields two rows and the shared case (one
-once their query shapes exist.
+definition, three assignees, one window) yields three.  No secondary
+indexes beyond it are declared yet: history and board queries (Features
+08/10/13) should justify their indexes as a later migration once their
+query shapes exist.
 
 No ``ON DELETE`` clauses are declared on any foreign key in this
 module: children and definitions are deactivated, never deleted, so a
@@ -128,18 +152,19 @@ representable — missed is a derived state announced on the HA event bus
 (Feature 10), never an appended event.
 
 ``child_id`` is deliberately denormalized (it is also derivable via
-``task_instances``): history reporting (Feature 13) filters and rates
+``quest_instances``): history reporting (Feature 13) filters and rates
 per child directly against the log, and the log should stay
 self-describing without joins.
 
 Actor policy: ``actor_source`` distinguishes an authenticated HA user
 (``'user'``, with ``actor_user_id`` holding the HA user id) from a
 panel tap with no user context (``'panel'``, with ``actor_user_id``
-NULL — the tapped child profile is already ``child_id``).  A coherence
+NULL — the tapped child profile is ``actor_child_id``, D-008).  Two
-CHECK makes the pair shape non-negotiable: ``'user'`` requires
+coherence CHECKs make both shapes non-negotiable: ``'user'`` requires
-``actor_user_id``, ``'panel'`` forbids it.  Other actor sources are
+``actor_user_id`` and forbids ``actor_child_id``; ``'panel'`` requires
-rejected; if an unattended system actor is ever needed, that is a
+``actor_child_id`` (the tapped profile) and forbids ``actor_user_id``.
-schema migration, not a silent widening.
+Other actor sources are rejected; if an unattended system actor is
+ever needed, that is a schema migration, not a silent widening.
 
 On-time policy: ``was_on_time`` is required (integer 0 or 1) for
 ``completed`` events — at completion time the due date is known and the
@@ -219,18 +244,45 @@ SCHEMA_V1_SCHEDULE_RULES_DDL: list[str] = [
     """,
 ]
 
-SCHEMA_V1_TASK_DEFINITIONS_DDL: list[str] = [
+#: Column body of ``quest_definitions``, shared between the v1 DDL
+#: below and the multi-assignee rebuild migration, which must recreate
+#: the table in exactly this shape (D-008: one definition, many
+#: assignees via ``quest_definition_assignees`` — no ``child_id``
+#: column on the definition itself).
+QUEST_DEFINITIONS_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    title TEXT NOT NULL,
+    description TEXT,
+    icon TEXT,
+    schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+    due_time TEXT,
+    is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+    created_at TEXT NOT NULL
+)"""
+
+SCHEMA_V1_QUEST_DEFINITIONS_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_definitions {QUEST_DEFINITIONS_TABLE_SQL}
+    """,
+]
+
+SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL: list[str] = [
     """
-    CREATE TABLE IF NOT EXISTS task_definitions (
+    CREATE TABLE IF NOT EXISTS quest_definition_assignees (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
-        title TEXT NOT NULL,
-        description TEXT,
-        icon TEXT,
         child_id INTEGER NOT NULL REFERENCES children(id),
-        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        PRIMARY KEY (definition_id, child_id)
+    )
+    """,
+]
+
+SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_definition_windows (
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
         due_time TEXT,
-        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+        PRIMARY KEY (definition_id, window)
-        created_at TEXT NOT NULL
     )
     """,
 ]
@@ -271,43 +323,64 @@ SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
     """,
 ]
 
-SCHEMA_V1_TASK_INSTANCES_DDL: list[str] = [
+#: Column body of ``quest_instances``, shared between the v1 DDL
-    """
+#: below and the instance-key rebuild migration, which must recreate
-    CREATE TABLE IF NOT EXISTS task_instances (
+#: the table in exactly this shape (D-008: the instance key is
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
+#: (definition_id, child_id, due_date, window) — one instance per
-        definition_id INTEGER NOT NULL REFERENCES task_definitions(id),
+#: assigned child per declared window per firing date).
-        child_id INTEGER NOT NULL REFERENCES children(id),
+QUEST_INSTANCES_TABLE_SQL = """(
-        due_date TEXT NOT NULL,
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
-        due_time TEXT,
+    definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
-        generated_at TEXT NOT NULL,
+    child_id INTEGER NOT NULL REFERENCES children(id),
-        UNIQUE (definition_id, due_date)
+    window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
-    )
+    due_date TEXT NOT NULL,
+    due_time TEXT,
+    generated_at TEXT NOT NULL,
+    UNIQUE (definition_id, child_id, due_date, window)
+)"""
+
+SCHEMA_V1_QUEST_INSTANCES_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_instances {QUEST_INSTANCES_TABLE_SQL}
     """,
 ]
 
-SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
+#: Column body of ``completion_events``, shared between the v1 DDL
-    """
+#: below and the actor-child rebuild migration, which must recreate the
-    CREATE TABLE IF NOT EXISTS completion_events (
+#: table in exactly this shape (D-008: ``actor_child_id`` records which
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
+#: panel profile was tapped — set for panel events, NULL for admin
-        instance_id INTEGER NOT NULL REFERENCES task_instances(id),
+#: ones; legacy panel rows backfill to ``child_id``, the child they
-        child_id INTEGER NOT NULL REFERENCES children(id),
+#: always were).
-        event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
+COMPLETION_EVENTS_TABLE_SQL = """(
-        actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
-        actor_user_id TEXT,
+    instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
-        occurred_at TEXT NOT NULL,
+    child_id INTEGER NOT NULL REFERENCES children(id),
-        was_on_time INTEGER CHECK (
+    event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
-            was_on_time IS NULL
+    actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
-            OR (typeof(was_on_time) = 'integer' AND was_on_time IN (0, 1))
+    actor_user_id TEXT,
-        ),
+    actor_child_id INTEGER REFERENCES children(id),
-        CHECK (
+    occurred_at TEXT NOT NULL,
-            (actor_source = 'user' AND actor_user_id IS NOT NULL)
+    was_on_time INTEGER CHECK (
-            OR (actor_source = 'panel' AND actor_user_id IS NULL)
+        was_on_time IS NULL
-        ),
+        OR (typeof(was_on_time) = 'integer' AND was_on_time IN (0, 1))
-        CHECK (
+    ),
-            (event_type = 'completed' AND was_on_time IS NOT NULL)
+    CHECK (
-            OR (event_type = 'uncompleted')
+        (actor_source = 'user' AND actor_user_id IS NOT NULL)
-        )
+        OR (actor_source = 'panel' AND actor_user_id IS NULL)
+    ),
+    CHECK (
+        (actor_source = 'user' AND actor_child_id IS NULL)
+        OR (actor_source = 'panel' AND actor_child_id IS NOT NULL)
+    ),
+    CHECK (
+        (event_type = 'completed' AND was_on_time IS NOT NULL)
+        OR (event_type = 'uncompleted')
     )
+)"""
+
+SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS completion_events {COMPLETION_EVENTS_TABLE_SQL}
     """,
 ]
 
@@ -315,10 +388,12 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_CHILDREN_DDL,
     *SCHEMA_V1_ADMIN_USERS_DDL,
     *SCHEMA_V1_SCHEDULE_RULES_DDL,
-    *SCHEMA_V1_TASK_DEFINITIONS_DDL,
+    *SCHEMA_V1_QUEST_DEFINITIONS_DDL,
+    *SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    *SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
     *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
-    *SCHEMA_V1_TASK_INSTANCES_DDL,
+    *SCHEMA_V1_QUEST_INSTANCES_DDL,
     *SCHEMA_V1_COMPLETION_EVENTS_DDL,
 ]
 

design/ADMIN-SPEC.md

@@ -0,0 +1,136 @@
+# Admin spec — parent phone card
+
+Screens `1d`–`1i`. One card, `nestquest-admin-card`, four tabs. Reference viewport
+390×844. Plain CubeCraft styling and plain domain words — no parchment, no Cinzel, no
+quest vocabulary in this surface.
+
+## 1. Shell
+
+- Page `--nq-a-page`; cards `--nq-a-surface` with 1px `--nq-a-border`.
+- **Header** white, 1px bottom border: screen title 24px/800, tracking −.02em, plus a
+  12px/600 `--nq-a-ink-secondary` sub-line. Optional trailing action (icon square 36px,
+  or a gradient `New` button).
+- **Tab bar** absolute bottom, `--nq-a-tabbar-height` 70px + 12px bottom padding, white,
+  1px top border, 4 items: `house` Today · `list-checks` Tasks · `calendar-days` Schedule ·
+  `history` History. Active icon and label `--nq-brand-blue` at 700; inactive icon
+  `--nq-a-ink-tertiary`, label `--nq-a-ink-secondary` 600.
+- **Every scroll column ends with 92px bottom padding** so content clears the tab bar.
+  This was a real defect in review — do not omit it.
+- Two densities exist. Ship **compact**; `--nq-a-density-*` carries the roomy values.
+- Everything this card mutates goes through a NestQuest service, which enforces the admin
+  allowlist server-side. The card must assume it can be rendered by a non-admin and still
+  be safe — hiding controls is not the boundary (D-003).
+
+## 2. Today (1d compact, 1h roomy)
+
+### 2.1 Header
+Title `Today`; sub `<weekday>, <Month D> · cycle day N of 14`; trailing `settings` square.
+
+### 2.2 Stat row
+Three equal cards, 8px gap, padding 12px: numeral 22px/800 (done in `--nq-brand-blue`,
+remaining in `--nq-a-ink`, overdue in `--nq-a-danger`), label 10px/600 uppercase tracking
+.08em `--nq-a-ink-secondary` — `Done`, `Remaining`, `Overdue`.
+
+### 2.3 Child rows
+One card, rows divided by `--nq-a-divider`, row padding 11px 13px:
+30px gradient avatar square (radius 7px, initial 13px/800 white) · name 14px/700 ·
+meta 11px/600 (`3 of 7 · 1 overdue`; all-done in `--nq-a-success`; away reads
+`Away · returns <Mon D>`) · 82×6px progress track with gradient fill. An away child's
+avatar is flat `#9898A8` and the row carries a lowercase `override` pill when the absence
+comes from a presence override rather than the pattern.
+
+### 2.4 Roomy variant (1h)
+Header becomes a hero: kicker 11px/700 uppercase, `8 of 12 done` 32px/800, 10px gradient
+progress bar, 13px/600 sub-line. Each child gets its own 12px-radius card with
+`0 4px 12px rgba(0,0,0,.06)`, 48px avatar, 19px/800 name, right-aligned `3/7` +
+status line, and its overdue item inlined as a `--nq-a-danger-bg` row with a
+`Mark done` action.
+
+### 2.5 Needs attention
+Section label 11px/700 uppercase, then one card of event rows: Lucide glyph 18px ·
+title `<Quest> · <Child>` 13px/700 · meta 11px/600 (`Overdue since 9:00 AM` in
+`--nq-a-danger`; `Completed 7:04 AM from panel` in `--nq-a-ink-secondary`) · trailing
+text action `Mark done` (`--nq-brand-blue`) or `Undo` (`--nq-brand-purple`).
+
+### 2.6 Permission note
+An info strip (`--nq-a-info-bg`, `info` glyph, 11.5px/600 `--nq-a-info-ink`):
+"Undo is admin-only. A panel tap can never reverse a completion." Keep this copy — it is
+the one place the security model is explained to the user.
+
+## 3. Definitions (1e)
+
+### 3.1 List
+Header `Tasks` / `7 definitions · 3 children` + gradient `New` button. Filter chips row
+(All / one per child): active `#18181D` fill white text, inactive white with border.
+Rows: glyph 17px · title 13.5px/700 · meta 11px/600 `<assignees> · <recurrence> · <windows>`
+· `chevron-right`. An inactive definition renders the whole row at
+`--nq-a-ink-tertiary` with `· inactive` in the meta — never hard-deleted (glossary).
+The row being edited takes `--nq-a-selected` with brand-blue glyph, `Editing` meta and
+`chevron-down`.
+
+### 3.2 Edit sheet
+Bottom sheet, `--nq-a-radius-sheet`, `--nq-a-sheet-shadow`, 38×4px grabber, padding
+14px 18px 24px. Title 19px/800. Fields, 11px gap, each with a 10px/700 uppercase label:
+
+| Field | Control |
+|---|---|
+| Assigned children | multi-select row → sheet; shows `Declan` or `All three` |
+| Repeats | select (Daily / Weekly / Monthly / Yearly / Custom + interval) |
+| Due time | time field per window |
+| Window | 3-up segmented, **multi-select** — a definition may own several |
+| Skip on away days | toggle, gradient when on, sub "No instance when the child is absent" |
+
+Footer: `Cancel` (flex 1, bordered) + `Save changes` (flex 2, gradient), both 13px vertical
+padding. Sheet enters over the dimmed list in `--nq-dur-modal`.
+
+### 3.3 Icon picker
+Each definition carries a Lucide glyph, shown on both surfaces. **The allowed subset is not
+yet defined** — pick 24–40 household-relevant glyphs and ship them as a grid picker.
+
+## 4. Schedule (1f)
+
+- Header `Schedule` / `2-week cycle · anchor <weekday>, <Mon D>`.
+- **Month grid** card: month label 11px/700 uppercase + chevron pair; weekday initials
+  10px/700 `--nq-a-ink-tertiary`; day cells `aspect-ratio: 1`, radius 6px —
+  rotation-home days `--nq-a-selected` with `#201F7A` numerals, away days
+  `--nq-a-page` with `--nq-a-ink-secondary`, override days `--nq-a-warning-bg` with
+  `--nq-a-warning-ink` and a 1px warning border, today filled with the brand gradient in
+  white 800. Legend row beneath, 10.5px/600, one swatch per state.
+- **Presence pattern** card: one row per child, name 13px/700 + rule 12px/600
+  (`Weeks 1 & 3 of cycle`, `Every day`).
+- **Overrides** card: `Add` text action; rows with `calendar-x` / `calendar-check`,
+  title `<Child> away · <Mon D – D>`, meta reason, trailing `trash-2`.
+- Cycle position is always computed from the anchor date. **Never** from ISO week numbers
+  or week parity (D-004).
+
+## 5. History (1g)
+
+- Header `History` / `Append-only event log · last 14 days` + bordered `CSV` button
+  (`download` glyph) that exports the filtered range.
+- Stat row: `91%` On time · `128` Completed · `7` Missed, numerals 20px/800.
+- Filter chips: `All events` / `Reversals` / `Missed`.
+- Day-grouped sections (label 11px/700 uppercase) of event rows: 7px status dot
+  (green completed, `--nq-a-reversal` un-completed, `--nq-a-danger` missed) ·
+  title `<Quest> · <Child>` 12.5px/700 · meta 10.5px/600 giving event type, actor and
+  on-time flag (`completed · panel (Declan) · on time`, `uncompleted · joshua (admin)`,
+  `missed · nightly sweep`, `completed · panel (Declan) · late 22 min`) · time 11px/600.
+- Footer strip (`--nq-a-page`, `lock` glyph): "Events are never edited or deleted. A
+  reversal is its own row." Rows are never editable in the UI (D-005).
+
+## 6. Presence override editor (1i)
+
+Pushed screen with a back chevron; title `Presence override` / sub "Beats the custody
+pattern for these dates".
+
+- Form card (roomy density, 18px padding, 16px gaps):
+  **Child** 3-up segmented, selected = gradient;
+  **Status for these dates** 2-up segmented (`Away` / `Home`), selected = `#18181D`;
+  **From** / **Through** date fields side by side, 13px padding, 14px/700 values;
+  **Reason (optional)** text field.
+- **Consequence warning** — `--nq-a-warning-bg` card with `alert-circle`:
+  headline `This removes N upcoming tasks` 13.5px/700 `--nq-a-warning-ink`, body 12px/600
+  `--nq-a-warning-ink-2` naming the child, the date range, and stating "Completed days are
+  untouched." The count must be computed live from the materialiser before saving.
+- Footer: `Cancel` (flex 1, white, bordered) + `Save override` (flex 2, gradient), 15px
+  vertical padding.
+- Saving must never delete an instance that already has a completion event.

design/DESIGN-HANDOFF.md

@@ -0,0 +1,82 @@
+# NestQuest — design handoff
+
+Approved UI design for the NestQuest Home Assistant integration, ready to implement.
+Source of truth for visuals: the mockup file `NestQuest Mockups.dc.html` (9 screens,
+option ids `1a`–`1j`). This folder is the written form of those screens.
+
+## What is in here
+
+| File | Contents |
+|---|---|
+| `PANEL-SPEC.md` | Kids' DAKOS wall panel: party board, quest log, confirm, quest complete |
+| `ADMIN-SPEC.md` | Parent phone admin: today, definitions, schedule, history, overrides |
+| `ENTITIES-AND-SERVICES.md` | Every screen mapped to the HA entities, services and events it needs |
+| `IMPLEMENTATION-ORDER.md` | Build order, with the decisions that change existing code first |
+| `tokens/nestquest-panel-tokens.css` | Panel card custom properties |
+| `tokens/nestquest-admin-tokens.css` | Admin card custom properties |
+
+## Screen index
+
+| id | Screen | Surface | Spec |
+|---|---|---|---|
+| 1a | The Party (attract) | panel 1920×1080 | PANEL §2 |
+| 1b | Quest Log | panel 1920×1080 | PANEL §3 |
+| 1j | Complete confirm | panel 1920×1080 | PANEL §4 |
+| 1c | Quest complete | panel 1920×1080 | PANEL §5 |
+| 1d | Today (compact) | phone 390×844 | ADMIN §2 |
+| 1e | Definitions + edit sheet | phone 390×844 | ADMIN §3 |
+| 1f | Schedule + custody | phone 390×844 | ADMIN §4 |
+| 1g | History + CSV | phone 390×844 | ADMIN §5 |
+| 1h | Today (roomy) | phone 390×844 | ADMIN §2.4 |
+| 1i | Presence override | phone 390×844 | ADMIN §6 |
+
+## Design decisions taken during this handoff
+
+These were open in CTXD and are now settled. Several contradict the current repo and
+docs — read `IMPLEMENTATION-ORDER.md` §1 before writing code.
+
+1. **`quest_*` naming throughout.** The domain model is renamed from `task_*` to
+   `quest_*` in DB, services, entities and UI. Migration note in IMPLEMENTATION-ORDER §1.1.
+2. **Instance key becomes `(definition_id, child_id, due_date, window)`.** Required by
+   multi-assignee definitions and by twice-daily quests. Supersedes the documented
+   `(definition_id, due_date)`.
+3. **A definition may have multiple assignees**, and materialises one instance per
+   assigned child per date per window. Each child completes their own.
+4. **A definition may declare more than one window** (morning / afternoon / evening),
+   producing one instance per window per day.
+5. **Missed quests vanish from the panel at rollover.** Kids never see a wall of failure;
+   the missed event is in the parent's history only.
+6. **Three Lovelace cards**: `nestquest-party-board-card`, `nestquest-quest-log-card`,
+   `nestquest-admin-card`. Board and log live in separate Lovelace views; a crest tap
+   navigates.
+7. **Completion needs a confirm step** (screen 1j) because children cannot undo.
+8. **The quest log owns idle return**: 40 s of no touch navigates back to the board.
+   TouchHub Auto-Return must be configured longer than 40 s so the two do not fight.
+9. **Actor on a panel completion** is `panel` plus the tapped child profile, per glossary.
+10. **Weather is a configured entity** on the party board card (`weather_entity`), not
+    NestQuest data.
+11. **Fonts are bundled** as local woff2 in the card's `www/` folder — the panel may be
+    offline. Nunito (400/600/700/800) and Cinzel + Cinzel Decorative.
+
+## Vocabulary
+
+The panel speaks tabletop; the admin speaks plainly. Both sit on the same `quest_*` model.
+
+| Model / admin | Panel display string |
+|---|---|
+| quest instance | quest |
+| child | adventurer |
+| child's task list | Quest Log |
+| household | The Party |
+| all complete | Quest complete |
+| absent | On travels |
+
+## Open questions
+
+- Empty and error states are unspecified by request (child with no quests, everyone away,
+  integration unconfigured, DB unreachable). Panel must not render a blank parchment —
+  decide before shipping the card.
+- No gamification is in scope. XP, levels, streaks and rarity were explicitly deferred,
+  but the append-only event log already supports computing them later.
+- Icon per quest definition is chosen by the parent from a Lucide subset; the subset is
+  not yet defined (`ADMIN-SPEC.md` §3.3).

design/ENTITIES-AND-SERVICES.md

@@ -0,0 +1,117 @@
+# Screens → HA entities, services and events
+
+Planned surface from CTXD CONTEXT.MD, adjusted for the design decisions in
+`DESIGN-HANDOFF.md`. Names use the `quest_*` model.
+
+## 1. Entities the cards read
+
+| Entity | Type | Used by |
+|---|---|---|
+| `sensor.nestquest_<child>_quests_due_today` | int | 1a plate progress, 1b counts, 1d stats |
+| `sensor.nestquest_<child>_quests_completed_today` | int | 1a, 1b, 1c, 1d, 1h |
+| `sensor.nestquest_<child>_quests_remaining_today` | int | 1b remaining block, 1d |
+| `sensor.nestquest_<child>_completion_pct_today` | float | progress bars |
+| `sensor.nestquest_<child>_next_quest` | str + attrs | optional panel hint |
+| `binary_sensor.nestquest_<child>_all_done` | bool | 1c trigger, 1d "all done" |
+| `binary_sensor.nestquest_<child>_present_today` | bool | 1a away plate, 1d away row |
+| `sensor.nestquest_household_quests_due_today` | int | 1b party line, 1c party stat |
+| `sensor.nestquest_household_quests_completed_today` | int | same |
+| `sensor.nestquest_cycle_day` | int | 1d sub-line, 1f header |
+| *configured* `weather.<entity>` | HA core | panel dock |
+
+Per-instance data is too granular for entity state. The cards read the day's instance list
+from an attribute payload on the per-child sensor (or a websocket command —
+implementer's call), shaped:
+
+```json
+{
+  "instances": [
+    { "id": 4821, "definition_id": 12, "child_id": 3, "title": "Fill bird feeder",
+      "icon": "bird", "window": "afternoon", "due_time": "17:00",
+      "state": "open", "overdue": true, "completed_at": null, "on_time": null }
+  ]
+}
+```
+
+`state` ∈ `open` | `completed`. Missed instances are **omitted** from panel payloads
+(decision 5) and included in admin payloads.
+
+## 2. Services the cards call
+
+| Service | Admin only | Called from |
+|---|---|---|
+| `nestquest.complete_quest` | no (panel allowed) | 1j confirm |
+| `nestquest.uncomplete_quest` | **yes** | 1d `Undo` |
+| `nestquest.create_quest_definition` | **yes** | 1e `New` |
+| `nestquest.update_quest_definition` | **yes** | 1e sheet `Save changes` |
+| `nestquest.set_quest_definition_active` | **yes** | 1e row action |
+| `nestquest.set_presence_pattern` | **yes** | 1f presence card |
+| `nestquest.create_presence_override` | **yes** | 1i `Save override` |
+| `nestquest.delete_presence_override` | **yes** | 1f override row `trash-2` |
+| `nestquest.export_history_csv` | **yes** | 1g `CSV` |
+| `nestquest.manage_child` | **yes** | settings |
+
+`complete_quest` payload from the panel:
+
+```yaml
+service: nestquest.complete_quest
+data:
+  instance_id: 4821
+  actor: panel            # kiosk user context
+  actor_child_id: 3       # the tapped profile (decision 9)
+```
+
+Every service resolves `call.context.user_id` against the admin allowlist and **fails
+closed** with no user context (D-003). The panel's kiosk user is never allowlisted, so
+`uncomplete_quest` is unreachable from the wall even though the card never renders the
+control.
+
+## 3. Events the cards listen for
+
+| Event | Effect |
+|---|---|
+| `nestquest_quest_completed` | stamp the seal, re-sort the column, refresh counts |
+| `nestquest_quest_uncompleted` | un-seal, restore the card (admin action, panel must reflect it) |
+| `nestquest_quest_missed` | remove the instance from the panel; add a history row |
+| `nestquest_child_day_complete` | show screen 1c |
+
+All four come off the `DataUpdateCoordinator`; cards should also refresh on
+`state_changed` for their sensors so they recover if an event is dropped.
+
+## 4. Card configuration
+
+```yaml
+# kids' panel, view 1
+type: custom:nestquest-party-board-card
+weather_entity: weather.home          # optional; dock falls back to date/time
+quest_log_path: /nestquest/log        # view the crest tap navigates to
+child_order: [declan, jordyn, chloe]
+
+# kids' panel, view 2
+type: custom:nestquest-quest-log-card
+board_path: /nestquest/board
+idle_return_seconds: 40               # must be shorter than TouchHub Auto-Return
+confirm_timeout_seconds: 15
+complete_screen_seconds: 12
+
+# parent phone
+type: custom:nestquest-admin-card
+density: compact                      # compact | roomy
+default_tab: today
+```
+
+The quest log reads the selected child from the view's URL/`hass` path parameter so the
+board → log hop is a plain Lovelace navigation (decision 6).
+
+## 5. Data-shape consequences of the design
+
+1. Instance key is `(definition_id, child_id, due_date, window)` — the schema's
+   `task_instances` unique index must widen.
+2. `quest_definitions` needs an **assignees** join table, replacing the single
+   `child_id` column.
+3. `quest_definitions` needs a **windows** set (morning/afternoon/evening) plus an
+   optional due time per window.
+4. The materialiser emits one instance per (assignee × window × firing date), still
+   idempotent on the widened key, still only when the child is present.
+5. `completion_events` gains `actor_child_id` so a panel completion records which
+   profile was tapped, while `actor` stays `panel`.

design/IMPLEMENTATION-ORDER.md

@@ -0,0 +1,80 @@
+# Implementation order
+
+Written for an agent working in `code.cubecraftlabs.com/Maniacs_Mansion/NestQuest`.
+Steps 1.x change existing code and must land first; the frontend is step 4 onward.
+
+## 1. Model changes (breaking, pre-release)
+
+### 1.1 Rename `task_*` → `quest_*`
+Nothing is released (main/dev are at "first commit", version 0.1.0), so rename in place
+rather than aliasing.
+
+- `schema.py`: `task_definitions` → `quest_definitions`, `task_instances` →
+  `quest_instances`. Keep `completion_events`, `children`, `admin_users`,
+  `schedule_rules`, `presence_schedules`, `presence_overrides`.
+- `db.py` / `store.py`: rename DAO methods and any SQL literals.
+- `const.py`: no table names live there today; keep it that way.
+- `tests/test_schema.py`, `tests/test_store.py`, `tests/test_db.py`: update expectations.
+- **Migration note:** the versioned migration runner has no released version to upgrade
+  from. Bump the schema version, and make migration 1 create the `quest_*` tables
+  directly. If any dev machine already holds a `nestquest.db` with `task_*` tables, the
+  runner should detect the legacy names and `ALTER TABLE ... RENAME TO` before applying
+  the widened index — a dev-only path, worth ~20 lines and a test.
+
+### 1.2 Widen the instance key
+Unique index becomes `(definition_id, child_id, due_date, window)`. Materialisation stays
+idempotent on that tuple. Add a test for the twice-daily case (one definition, two windows,
+one child, one date → two rows) and the shared case (one definition, three assignees, one
+window → three rows).
+
+### 1.3 Multi-assignee definitions
+Add `quest_definition_assignees(definition_id, child_id)`. Drop the single `child_id`
+column on the definition. The "exactly one assigned child" rule in GLOSSARY.MD no longer
+holds — CTXD has been updated.
+
+### 1.4 Windows
+Add `quest_definition_windows(definition_id, window, due_time)` with
+`window ∈ ('morning','afternoon','evening')`. Window clock ranges are fixed for now:
+morning ≤ 11:59, afternoon 12:00–17:00, evening 17:00–21:00. Store the ranges in
+`const.py` so they are configurable later.
+
+### 1.5 `actor_child_id` on completion events
+Nullable int. Set for panel completions, null for admin ones. Still append-only.
+
+## 2. Engines and materialisation
+Unchanged in intent: pure recurrence engine, pure presence engine with anchor-date
+arithmetic, daily materialiser over `horizon_days` at `day_rollover_time`. Only the
+instance fan-out changes (assignees × windows).
+
+## 3. Services, permissions, entities
+Implement the service table in `ENTITIES-AND-SERVICES.md` §2 behind the permission
+decorator, then the sensors and binary sensors in §1, then the four events in §3.
+`complete_quest` is the only service a non-admin context may call.
+
+## 4. Frontend groundwork
+1. Create `custom_components/nestquest/www/`.
+2. Bundle fonts as woff2: Nunito 400/600/700/800; Cinzel 600/700/900; Cinzel Decorative
+   700/900. Declare `@font-face` with `font-display: swap` in a shared
+   `nestquest-fonts.css`. No CDN — the panel may be offline.
+3. Copy `design/tokens/nestquest-panel-tokens.css` and
+   `design/tokens/nestquest-admin-tokens.css` in as the cards' `:host` blocks.
+4. Register the three cards with `customCards` and ship editors later; hand-written YAML
+   is fine for v1.
+
+## 5. Cards, in this order
+1. `nestquest-party-board-card` (screen 1a) — read-only, proves the entity payload.
+2. `nestquest-quest-log-card` (1b) — read-only first, no completion.
+3. Confirm dialog + `complete_quest` (1j) with optimistic seal and revert-on-failure.
+4. Quest complete screen (1c) driven by `nestquest_child_day_complete`.
+5. Idle return (40 s) and the documented TouchHub Auto-Return constraint.
+6. `nestquest-admin-card`: Today (1d) → Definitions + sheet (1e) → Schedule (1f) →
+   History + CSV (1g) → Presence override (1i).
+
+## 6. Before shipping
+- `use_x_frame_options: false` is required for TouchHub embedding — document the
+  clickjacking trade-off next to the setting, per D-006.
+- Verify no NestQuest control in the dock strip is tappable.
+- Verify the kiosk HA user is absent from the admin allowlist, and that
+  `uncomplete_quest` denies it with a logged denial.
+- Empty and error states are still unspecified (see DESIGN-HANDOFF "Open questions").
+  Do not ship a card that renders a blank parchment.

design/PANEL-SPEC.md

@@ -0,0 +1,170 @@
+# Panel spec — kids' DAKOS wall panel
+
+Screens `1a`, `1b`, `1j`, `1c` of the mockups. Two cards:
+`nestquest-party-board-card` (§2) and `nestquest-quest-log-card` (§3–5).
+
+## 1. Frame and constraints
+
+- **Canvas** 1920×1080, landscape, fixed. No scrolling — content must fit. Every screen in
+  the mockups measures exactly 1920×1080 with no overflow; keep it that way.
+- **Viewing distance** ~3 ft. **No text below 22px.** Meta lines are 22px/600 Nunito;
+  quest titles 30px/700 Cinzel; section heads 32px.
+- **Contrast** every ink token is ≥4.5:1 on parchment. Do not introduce alpha-muted text.
+- **Ground** `--nq-p-parchment` + `--nq-p-vignette`, CSS only, no image assets.
+- **Double frame** inset 26px (`--nq-p-frame-outer`, radius 14px) and inset 36px
+  (`--nq-p-frame-inner`, radius 8px), both `pointer-events: none`.
+- **Dock** bottom strip, 84px, inset 46px, radius 14px, `--nq-p-dock-bg`. Weather only.
+  **Nothing in the dock is tappable** (D-006).
+- **No login.** Every action comes from the kiosk HA user; the kiosk user is never in the
+  admin allowlist.
+
+## 2. The Party — attract state (1a)
+
+Default view on the wall. Layout top to bottom:
+
+1. **Wordmark row** at y=64. Rule · d20 hexagon · `NestQuest` (66px/900 display, brand
+   gradient via `background-clip: text`) · d20 · rule. d20 marks are 34px hexagons
+   (`clip-path: polygon(50% 0, 93% 25%, 93% 75%, 50% 100%, 7% 75%, 7% 25%)`) filled with
+   the brand gradient, numeral 15px/700 Cinzel.
+2. **Kicker** `The Party · <weekday>, <Month D>` — 21px/600 Cinzel, tracking .30em, uppercase.
+3. **Adventurer plates**, 3-up grid, 56px gap, page inset 110px, top y=262. One plate per
+   child returned by the party sensor, in configured order.
+4. **Hint line** `Tap your crest to open your Quest Log`, 27px/600 Nunito, y≈930.
+5. **Dock.**
+
+### 2.1 Adventurer plate (present)
+
+Radius 20px, `--nq-p-card-panel`, 2px `--nq-p-panel-border`, `--nq-p-panel-shadow`,
+padding 44px 28px 40px, contents centred, 26px gap. **The whole plate is the tap target**
+(≥188px wide, ≥600px tall — far above minimum).
+
+- **Crest** 188×214, shield via `clip-path: polygon(0 0, 100% 0, 100% 62%, 50% 100%, 0 62%)`,
+  `--nq-p-crest`, inset ring `inset 0 0 0 4px rgba(255,255,255,.22)`, child's initial
+  82px/700 display in white.
+- **Name** 48px/700 Cinzel, `--nq-p-ink`, tracking .02em.
+- **Presence pill** radius 9999px, `#e8f4ea` on `rgba(22,120,60,.35)`, `check-circle` 22px
+  `#157a3c`, label `Home today` 19px/600 Cinzel uppercase `#146b36`.
+- **Progress line** `N of M quests claimed` 26px/700 Cinzel `--nq-p-ink-secondary`.
+- **Progress bar** full width, 16px, radius 9999px, track `rgba(92,62,26,.18)` with 1px
+  border, fill `--nq-brand-gradient-h`.
+
+### 2.2 Adventurer plate (away)
+
+Same geometry, muted: `linear-gradient(#f2ecdd, #e6dcc6)`, **2px dashed** border, no lift
+shadow. Crest uses `--nq-p-crest-away`; initial stays **full-opacity white** (never alpha).
+Name in `--nq-p-ink-away`. Pill: `tent` icon, label `On travels`, neutral fill. Progress
+line reads `Returns <weekday>, <Mon D>`; empty progress track, no fill. **Not tappable** —
+the plate must not navigate, since the child has no instances.
+
+### 2.3 Dock
+
+`cloud-sun` 40px · temperature 38px/800 Nunito `--nq-p-dock-ink` · 1px divider ·
+condition + high/low 26px/600 `--nq-p-dock-ink-secondary` · divider · one short forecast
+phrase. All read from the configured `weather_entity`; if unset, render the dock with date
+and time only rather than removing it (the strip is part of the frame).
+
+## 3. Quest Log (1b)
+
+Reached by tapping a present crest. Header at y=56, page inset 62px:
+
+- Small crest 96×110 (initial 44px/700 display).
+- Title `<Name>'s Quest Log` 56px/700 display.
+- Sub `<weekday>, <Month D> · N of M claimed` 20px/600 Cinzel, tracking .24em, uppercase.
+- **Remaining block** right-aligned: radius 14px panel, d20 mark 44px with the remaining
+  count, `N quests left` 28px/700 Cinzel, and a party line
+  `Party progress · X of Y today` 22px/600 Nunito.
+
+### 3.1 Window columns
+
+Three equal columns, 34px gap, top y=236, one per window — **Morning** (until 11:59 AM),
+**Afternoon** (12:00–5:00 PM), **Evening** (5:00–9:00 PM). Column head: icon
+(`sunrise` / `sun` / `moon`) 34px, name 32px/700 Cinzel, sub-label 21px/600 Cinzel
+uppercase with the window's clock range, right-aligned `done/total` 26px/700 Cinzel,
+14px bottom padding, `--nq-p-rule` underline.
+
+Cards stack at `--nq-p-quest-gap` (24px). Order within a column: **incomplete first,
+sealed below**. A window with no quests renders its head and nothing else — do not
+collapse the column, the three-column rhythm is the layout.
+
+### 3.2 Quest card — open
+
+`--nq-p-quest-min-height` 116px, radius 12px, `--nq-p-card-open`, 1px
+`--nq-p-card-border`, `--nq-p-card-shadow`, padding 22px, 22px gap, row layout:
+
+1. **Icon tile** 64×64, radius 10px, `--nq-p-icon-tile`, Lucide glyph 34px white.
+2. **Text** title 30px/700 Cinzel `--nq-p-ink`; meta 22px/600 Nunito
+   `--nq-p-ink-secondary` reading `Due by <h:mm A>`. If past due:
+   `Overdue · due <h:mm A>` in 22px/700 Cinzel uppercase, tracking .10em,
+   `--nq-p-ink-late`.
+3. **Complete button** `--nq-p-button-height` 72px × min 150px, radius 10px,
+   `--nq-brand-gradient`, `check` 30px + `Complete` 24px/700 Cinzel white,
+   `box-shadow: 0 3px 0 rgba(40,20,60,.35)`.
+
+Both the card body and the button open the confirm dialog (§4). Press feedback:
+`scale(.98)`, `--nq-dur-micro`.
+
+### 3.3 Quest card — sealed
+
+`--nq-p-card-done`, `opacity: .82`, no shadow, **not tappable**. Icon tile
+`--nq-p-icon-tile-done` with the glyph in `--nq-p-ink-secondary`. Title 30px/700 Cinzel
+`--nq-p-ink-muted` with `text-decoration: line-through`. Meta reads
+`Claimed <h:mm A>` 22px/600 Nunito `--nq-p-ink-secondary`.
+
+**Wax seal** `--nq-p-seal-size` 78px circle, `--nq-p-seal`, `--nq-p-seal-shadow`,
+`check` 38px in `rgba(255,235,235,.95)`, absolutely positioned `right: 16px; top: -12px`,
+rotated a few degrees — **vary the rotation per card** (−9°, +6°, −4° in the mockup) so the
+board looks stamped by hand, not templated. Entrance: fade + `scale(1.15 → 1)`,
+`--nq-dur-base`, `--nq-ease-out`.
+
+### 3.4 Party roll-up
+
+An optional dashed panel in the afternoon column: label `Party roll-up` 18px/600 Cinzel
+uppercase, body 23px/600 Nunito `--nq-p-ink-secondary` — one or two sentences on the other
+children ("Jordyn has finished her log. Chloe is on travels until Sunday."). Read-only.
+
+### 3.5 Idle return
+
+40 s with no touch → navigate back to the party board view. Reset on any pointer or touch
+event. The card owns this; TouchHub Auto-Return must be set longer than 40 s.
+
+## 4. Complete confirm (1j)
+
+Modal over the dimmed log — scrim `rgba(30,20,10,.62)` over the live board.
+
+Dialog 1000px wide at `top: 196px; left: 460px`, radius 20px, parchment panel with 3px
+`rgba(92,62,26,.5)`, `box-shadow: 0 26px 60px rgba(20,10,0,.5)`, padding 64px 68px 56px.
+
+- Header row: 104px icon tile (radius 12px, brand gradient, glyph 56px) · window kicker
+  21px/600 Cinzel uppercase tracking .24em · quest title 58px/700 display.
+- Body 32px/600 Nunito `#3f2f1c`: `Mark this quest complete? Once the seal is set, only a
+  parent can undo it.`
+- Buttons, 24px gap, both `--nq-p-confirm-button` 108px tall:
+  **Not yet** flex 1, `rgba(92,62,26,.1)` with 2px border, `x` 34px + 30px/700 Cinzel;
+  **Complete** flex 2, brand gradient, `check` 38px + 32px/700 Cinzel white.
+- Dialog enters with fade + `scale(.96 → 1)`, `--nq-dur-modal`, `--nq-ease-out`.
+- Dismiss: **Not yet**, scrim tap, or 15 s of no input.
+
+On confirm, call the completion service optimistically, stamp the seal, and re-sort the
+column. On service failure, revert the card and surface a single-line parchment toast.
+
+## 5. Quest complete (1c)
+
+Shown when the last instance for a child's day is completed; auto-returns to the board
+after 12 s (countdown line `Returning to The Party in N seconds`, 25px/600 Nunito).
+
+- Large seal 168px at y=96 (`check` 86px), rotated −6°, `0 10px 24px rgba(60,10,20,.42)`.
+- `Quest complete` 86px/900 display, brand-gradient text fill.
+- Sub 34px/600 Nunito `--nq-p-ink-secondary`, max-width 1000px, centred, `text-wrap: pretty`:
+  names the child, the date, and how the day went.
+- **Stat blocks** 4-up at y=540, inset 200px, 26px gap: radius 16px parchment panels,
+  numeral 58px/900 Cinzel, label 17px/600 Cinzel uppercase tracking .14em —
+  `Quests claimed`, `On time`, `Late`, `The Party` (household X/Y).
+- Dock switches to the evening variant (`cloud-moon`).
+
+## 6. Accessibility and input
+
+- Minimum tap target on this surface is the quest card itself (116px) and the 72px button;
+  nothing interactive is smaller than 72px.
+- 24px between adjacent tap targets so a stray finger cannot claim the wrong quest.
+- No hover states — this is a touch-only surface. Provide `:active` feedback only.
+- No animation loops. Entrances only, per the CubeCraft motion rules.

design/tokens/nestquest-admin-tokens.css

@@ -0,0 +1,76 @@
+/* ============================================================
+   NestQuest — Parent admin card tokens
+   Surface: phone, 390x844 reference. Plain CubeCraft styling,
+   no tabletop treatment, plain domain words in the UI.
+   Scope on the card host: nestquest-admin-card.
+   Derived from the approved mockups (admin screens 1d-1i).
+   Two densities were built; COMPACT is the default. Flip
+   --nq-a-density-* to the roomy values for the alternative.
+   ============================================================ */
+
+:host {
+  /* ── Brand + neutrals (CubeCraft tokens, verbatim) ─────── */
+  --nq-brand-blue:        #3B3AB8;
+  --nq-brand-purple:      #A035CC;
+  --nq-brand-gradient:    linear-gradient(135deg, var(--nq-brand-blue), var(--nq-brand-purple));
+  --nq-brand-gradient-h:  linear-gradient(90deg,  var(--nq-brand-blue), var(--nq-brand-purple));
+
+  --nq-a-page:            #F4F4F6;
+  --nq-a-surface:         #FFFFFF;
+  --nq-a-surface-subtle:  #FAFAFA;
+  --nq-a-selected:        #EBEBF8;   /* row being edited */
+  --nq-a-border:          #E4E4EA;
+  --nq-a-divider:         #EDEDF1;
+
+  --nq-a-ink:             #18181D;
+  --nq-a-ink-secondary:   #52525E;
+  --nq-a-ink-tertiary:    #70707E;   /* inactive rows, unselected nav icons */
+
+  --nq-a-success:         #15803D;
+  --nq-a-danger:          #DC2626;
+  --nq-a-danger-strong:   #B91C1C;
+  --nq-a-danger-bg:       #FEE2E2;
+  --nq-a-warning-ink:     #92400E;
+  --nq-a-warning-ink-2:   #78350F;
+  --nq-a-warning-bg:      #FEF3C7;
+  --nq-a-warning-border:  rgba(217,119,6,.35);
+  --nq-a-info-bg:         #DBEAFE;
+  --nq-a-info-ink:        #1E3A8A;
+  --nq-a-info-icon:       #1D4ED8;
+  --nq-a-reversal:        #7D28A0;   /* un-complete events in history */
+
+  /* ── Type ─────────────────────────────────────────────── */
+  --nq-a-font:            Nunito, system-ui, sans-serif;
+  --nq-a-size-screen:     24px;  /* 800, screen title */
+  --nq-a-size-hero:       32px;  /* 800, roomy "8 of 12 done" */
+  --nq-a-size-card-title: 19px;  /* 800, roomy child card */
+  --nq-a-size-stat:       22px;  /* 800, compact stat number */
+  --nq-a-size-row:        13.5px;/* 700, list row title */
+  --nq-a-size-meta:       11px;  /* 600, row meta */
+  --nq-a-size-label:      10px;  /* 700, uppercase field label */
+  --nq-a-track-label:     .10em;
+
+  /* ── Density: COMPACT (default) ───────────────────────── */
+  --nq-a-density-page-pad:   14px 16px 92px;  /* bottom clears the 70px tab bar */
+  --nq-a-density-gap:        10px;
+  --nq-a-density-card-pad:   13px;
+  --nq-a-density-row-pad:    11px 13px;
+  --nq-a-density-radius:     8px;
+  --nq-a-density-shadow:     none;
+  /* Density: ROOMY — page-pad 18px 20px 92px; gap 16px;
+     card-pad 18px; radius 12px; shadow 0 4px 12px rgba(0,0,0,.06) */
+
+  /* ── Metrics ──────────────────────────────────────────── */
+  --nq-a-tabbar-height:   70px;   /* + 12px safe-area padding */
+  --nq-a-tap-min:         44px;   /* every row/control clears this */
+  --nq-a-radius-pill:     9999px;
+  --nq-a-radius-sheet:    20px 20px 0 0;
+  --nq-a-sheet-shadow:    0 -8px 28px rgba(9,9,11,.18);
+  --nq-a-progress-height: 6px;    /* 8px roomy, 10px hero */
+
+  /* ── Motion ───────────────────────────────────────────── */
+  --nq-ease-out:          cubic-bezier(0, 0, .2, 1);
+  --nq-dur-micro:         120ms;
+  --nq-dur-base:          200ms;
+  --nq-dur-modal:         350ms;
+}

design/tokens/nestquest-panel-tokens.css

@@ -0,0 +1,94 @@
+/* ============================================================
+   NestQuest — Kids' panel card tokens
+   Surface: DAKboard DAKOS wall panel, 1920x1080 landscape,
+   viewed from ~3 ft, no login, read/complete only.
+   Scope these on the card host: nestquest-quest-log-card,
+   nestquest-party-board-card.
+   Derived from the approved mockups (panel screens 1a, 1b, 1c, 1j).
+   ============================================================ */
+
+:host {
+  /* ── Parchment ground ─────────────────────────────────── */
+  --nq-p-parchment-top:      #f0e4c7;
+  --nq-p-parchment-bottom:   #ddcca3;
+  --nq-p-parchment: radial-gradient(ellipse at 25% 10%, rgba(255,255,255,.5), transparent 55%),
+                    radial-gradient(ellipse at 85% 90%, rgba(120,86,44,.28), transparent 60%),
+                    repeating-linear-gradient(93deg, rgba(150,115,70,.05) 0 2px, transparent 2px 6px),
+                    repeating-linear-gradient(2deg, rgba(150,115,70,.04) 0 3px, transparent 3px 7px),
+                    linear-gradient(var(--nq-p-parchment-top), var(--nq-p-parchment-bottom));
+  --nq-p-vignette:           inset 0 0 200px rgba(80,52,20,.35);
+
+  /* Card surfaces sitting on parchment */
+  --nq-p-card-open:          linear-gradient(#fdf7e6, #f2e7c9);   /* incomplete quest */
+  --nq-p-card-done:          linear-gradient(#f0e8d3, #e6dcc2);   /* sealed quest */
+  --nq-p-card-panel:         linear-gradient(#fcf6e6, #f1e6ca);   /* adventurer plate, stat block */
+  --nq-p-card-border:        rgba(120,88,48,.38);
+  --nq-p-panel-border:       rgba(92,62,26,.40);
+  --nq-p-card-shadow:        0 3px 0 rgba(120,88,48,.20), inset 0 1px 0 rgba(255,255,255,.70);
+  --nq-p-panel-shadow:       0 6px 0 rgba(92,62,26,.18), inset 0 2px 0 rgba(255,255,255,.70);
+  --nq-p-frame-outer:        2px solid rgba(92,62,26,.45);
+  --nq-p-frame-inner:        1px solid rgba(92,62,26,.28);
+  --nq-p-rule:               2px solid rgba(92,62,26,.35);
+
+  /* ── Ink (all values >= 4.5:1 on parchment) ───────────── */
+  --nq-p-ink:                #2b1f14;   /* titles, quest names */
+  --nq-p-ink-secondary:      #5c452a;   /* meta, window labels, counts */
+  --nq-p-ink-muted:          #6f6455;   /* sealed quest titles only */
+  --nq-p-ink-away:           #4d4433;   /* away adventurer name */
+  --nq-p-ink-late:           #8f1526;   /* overdue line */
+
+  /* ── Brand (CubeCraft, toned down: accents only) ──────── */
+  --nq-brand-blue:           #3B3AB8;
+  --nq-brand-purple:         #A035CC;
+  --nq-brand-gradient:       linear-gradient(135deg, var(--nq-brand-blue), var(--nq-brand-purple));
+  --nq-p-icon-tile:          var(--nq-brand-gradient);            /* open quest icon tile */
+  --nq-p-icon-tile-done:     rgba(92,62,26,.16);
+  --nq-p-crest:              var(--nq-brand-gradient);
+  --nq-p-crest-away:         linear-gradient(135deg, #6b6b7a, #7a7286);
+
+  /* ── Wax seal (completed marker) ──────────────────────── */
+  --nq-p-seal:               radial-gradient(circle at 35% 30%, #a8283a, #6d1322);
+  --nq-p-seal-shadow:        0 4px 10px rgba(60,10,20,.40), inset 0 0 0 4px rgba(255,255,255,.14);
+  --nq-p-seal-size:          78px;   /* 168px on the quest-complete screen */
+
+  /* ── Dock (bottom strip, non-interactive) ─────────────── */
+  --nq-p-dock-bg:            rgba(43,31,20,.90);
+  --nq-p-dock-ink:           #f7efdb;
+  --nq-p-dock-ink-secondary: #d8c9a6;
+  --nq-p-dock-divider:       rgba(233,220,189,.30);
+  --nq-p-dock-height:        84px;
+
+  /* ── Type ─────────────────────────────────────────────── */
+  --nq-p-font-display:       'Cinzel Decorative', Cinzel, serif;  /* NestQuest, names, Quest complete */
+  --nq-p-font-heading:       Cinzel, serif;                       /* section heads, quest titles, numerals */
+  --nq-p-font-body:          Nunito, system-ui, sans-serif;       /* meta lines, dock */
+
+  --nq-p-size-hero:          86px;   /* 900 weight, "Quest complete" */
+  --nq-p-size-wordmark:      66px;   /* 900 weight */
+  --nq-p-size-title:         56px;   /* 700, "Declan's Quest Log" */
+  --nq-p-size-name:          48px;   /* 700, adventurer name */
+  --nq-p-size-section:       32px;   /* 700, Morning / Afternoon / Evening */
+  --nq-p-size-quest:         30px;   /* 700, quest title */
+  --nq-p-size-body:          22px;   /* 600 Nunito, meta — PANEL MINIMUM */
+  --nq-p-size-label:         21px;   /* 600 Cinzel, uppercase, tracked */
+  --nq-p-track-label:        .14em;
+  --nq-p-track-kicker:       .30em;
+
+  /* ── Metrics: touch ───────────────────────────────────── */
+  --nq-p-quest-min-height:   116px;  /* whole card is the tap target */
+  --nq-p-quest-gap:          24px;   /* separation between tap targets */
+  --nq-p-button-height:      72px;   /* "Complete" button on a quest card */
+  --nq-p-button-min-width:   150px;
+  --nq-p-confirm-button:     108px;  /* confirm dialog buttons */
+  --nq-p-radius-card:        12px;
+  --nq-p-radius-panel:       20px;
+  --nq-p-radius-pill:        9999px;
+  --nq-p-page-inset:         62px;
+  --nq-p-frame-inset:        26px;
+
+  /* ── Motion (CubeCraft easing) ────────────────────────── */
+  --nq-ease-out:             cubic-bezier(0, 0, .2, 1);
+  --nq-dur-micro:            120ms;
+  --nq-dur-base:             200ms;
+  --nq-dur-modal:            350ms;
+}

tests/test_children_registry.py

@@ -0,0 +1,842 @@
+"""Tests for children.py: the Feature 03 child-profile business layer."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from custom_components.nestquest.children import (
+    create_child,
+    edit_child,
+    list_children,
+    reorder_children,
+    set_child_active,
+)
+from custom_components.nestquest.dao_children import ChildRecord, ChildrenDao
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            database = NestQuestDatabase(_make_hass_mock())
+            await database.open(tmp_path / name)
+            try:
+                await apply_migrations(database)
+                return await body(database)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# create_child: required, trimmed display name
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_returns_typed_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database,
+            "  Declan  ",
+            colour="#2E7D32",
+            avatar_ref="avatars/declan.png",
+            sort_order=2,
+        )
+        assert isinstance(child, ChildRecord)
+        # The name is trimmed before storage.
+        assert child.display_name == "Declan"
+        assert child.colour == "#2E7D32"
+        assert child.avatar_ref == "avatars/declan.png"
+        assert child.sort_order == 2
+        assert child.is_active is True
+        # The business layer stamps created_at in strict UTC shape.
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", child.created_at
+        )
+        return child
+
+    _with_db(tmp_path, "create-typed.db")(_body)
+
+
+def test_create_child_defaults(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Chloe")
+        assert child.display_name == "Chloe"
+        assert child.colour is None
+        assert child.avatar_ref is None
+        assert child.sort_order == 0
+        return child
+
+    _with_db(tmp_path, "create-defaults.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["", "   ", "\t\n", None])
+def test_create_child_missing_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="display_name is required"):
+            await create_child(database, bad)
+        assert await list_children(database) == []
+        return None
+
+    _with_db(tmp_path, "create-empty-name.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [42, ["Ada"], 3.5])
+def test_create_child_non_string_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="must be a string"):
+            await create_child(database, bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-name-type.db")(_body)
+
+
+def test_create_child_optional_text_fields_validated(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="colour must be a string"):
+            await create_child(database, "Ada", colour=7)
+        with pytest.raises(ValueError, match="avatar_ref must be a string"):
+            await create_child(database, "Ada", avatar_ref=1.5)
+        return None
+
+    _with_db(tmp_path, "create-bad-optional.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["2", 1.5, True, None])
+def test_create_child_sort_order_must_be_int(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await create_child(database, "Ada", sort_order=bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# create_child: duplicate display name allowed but warned
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_duplicate_name_allowed_and_warned(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        with caplog.at_level(
+            logging.WARNING, logger="custom_components.nestquest.children"
+        ):
+            second = await create_child(database, "  ADA ")
+        # Whitespace/case-normalised duplicate: allowed, both rows exist.
+        assert first.display_name == "Ada"
+        assert second.display_name == "ADA"
+        names = [c.display_name for c in await list_children(database)]
+        assert names == ["Ada", "ADA"]
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1, (
+            "exactly one duplicate warning for the second create"
+        )
+        assert "Ada" in warnings[0].getMessage()
+        return second
+
+    _with_db(tmp_path, "create-duplicate.db")(_body)
+
+
+def test_create_child_distinct_name_logs_no_warning(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        await create_child(database, "Ada")
+        caplog.clear()
+        await create_child(database, "Jordyn")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert warnings == []
+        return None
+
+    _with_db(tmp_path, "create-no-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_child: returns the updated record
+# ---------------------------------------------------------------------------
+
+
+def test_edit_child_updates_fields_and_returns_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(
+            database,
+            child.id,
+            display_name="  Ada L. ",
+            colour="#222222",
+            avatar_ref="avatars/ada.png",
+            sort_order=3,
+        )
+        assert isinstance(edited, ChildRecord)
+        assert edited.id == child.id
+        assert edited.display_name == "Ada L."
+        assert edited.colour == "#222222"
+        assert edited.avatar_ref == "avatars/ada.png"
+        assert edited.sort_order == 3
+        # The stored row reflects every edit.
+        stored = await list_children(database)
+        assert len(stored) == 1
+        assert stored[0] == edited
+        return edited
+
+    _with_db(tmp_path, "edit-round-trip.db")(_body)
+
+
+def test_edit_child_omitted_fields_unchanged(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database, "Ada", colour="#111111", sort_order=5
+        )
+        edited = await edit_child(database, child.id, sort_order=9)
+        # Only sort_order moved; colour and name untouched.
+        assert edited.sort_order == 9
+        assert edited.colour == "#111111"
+        assert edited.display_name == "Ada"
+        return edited
+
+    _with_db(tmp_path, "edit-partial.db")(_body)
+
+
+def test_edit_child_no_fields_returns_current_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(database, child.id)
+        assert edited == child
+        return edited
+
+    _with_db(tmp_path, "edit-noop.db")(_body)
+
+
+def test_edit_child_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await edit_child(database, 999, display_name="X")
+        return None
+
+    _with_db(tmp_path, "edit-missing.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.0, "1", None])
+def test_edit_child_non_int_id_rejected(tmp_path, bad) -> None:
+    """SQLite binds bools as integers: True must not address child 1,
+    and floats must not round onto a real profile."""
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await edit_child(database, bad, display_name="X")
+        # The refused edit left the profile untouched.
+        assert (await list_children(database))[0].display_name == "Ada"
+        return None
+
+    _with_db(tmp_path, "edit-bad-id.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.0, "1", None])
+def test_set_child_active_non_int_id_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await set_child_active(database, bad, False)
+        assert (await list_children(database))[0].is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-bad-id.db")(_body)
+
+
+def test_edit_child_name_validation(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        for bad in ("", "   ", None, 42):
+            with pytest.raises(ValueError, match="display_name"):
+                await edit_child(database, child.id, display_name=bad)
+        # The failed edits left the record untouched.
+        assert (await list_children(database))[0].display_name == "Ada"
+        return None
+
+    _with_db(tmp_path, "edit-name-validation.db")(_body)
+
+
+def test_edit_child_sort_order_must_be_int(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await edit_child(database, child.id, sort_order="3")
+        return None
+
+    _with_db(tmp_path, "edit-bad-order.db")(_body)
+
+
+def test_edit_child_explicit_none_clear_rejected(tmp_path) -> None:
+    """Clearing is refused loudly, not silently ignored: the DAO cannot
+    express NULL writes, so an explicit None must never masquerade as a
+    no-op."""
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        with pytest.raises(ValueError, match="clearing colour"):
+            await edit_child(database, child.id, colour=None)
+        with pytest.raises(ValueError, match="clearing avatar_ref"):
+            await edit_child(database, child.id, avatar_ref=None)
+        # The refused clears left the colour intact.
+        assert (await list_children(database))[0].colour == "#111111"
+        return None
+
+    _with_db(tmp_path, "edit-clear-unsupported.db")(_body)
+
+
+def test_edit_child_rename_to_duplicate_warns_but_different_name_not(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        bo = await create_child(database, "Bo")
+        caplog.clear()
+        # Renaming Bo onto Ada's name warns (different child's name).
+        edited = await edit_child(database, bo.id, display_name="ada")
+        assert edited.display_name == "ada"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert "Ada" in warnings[0].getMessage()
+        return edited
+
+    _with_db(tmp_path, "edit-rename-duplicate.db")(_body)
+
+
+def test_edit_child_rename_to_own_name_casing_alone_warns_not(
+    tmp_path, caplog
+) -> None:
+    """Renaming a child to their own name (casing aside) is not a
+    duplicate when NO OTHER child carries the normalised name."""
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        caplog.clear()
+        edited = await edit_child(database, ada.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        assert not any(
+            "duplicates existing child" in r.getMessage()
+            for r in caplog.records
+        )
+        return edited
+
+    _with_db(tmp_path, "edit-rename-self.db")(_body)
+
+
+def test_create_child_concurrent_same_name_still_warns(tmp_path) -> None:
+    """Racing creates of the same normalised name cannot slip through
+    unlogged: the scan and INSERT are serialized on the connection
+    lock, so the second create sees the first's committed row and
+    warns.  Deterministic gate on task A's INSERT while task B queues
+    behind the lock.  Without the lock, B's scan would run before A's
+    INSERT commits and no warning would fire."""
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "create-race.db")
+        try:
+            await apply_migrations(database)
+            logger = logging.getLogger(
+                "custom_components.nestquest.children"
+            )
+            seen: list[logging.LogRecord] = []
+
+            class _Capture(logging.Handler):
+                def emit(self, record: logging.LogRecord) -> None:
+                    seen.append(record)
+
+            handler = _Capture()
+            logger.addHandler(handler)
+            try:
+                armed["active"] = True
+                task_a = asyncio.ensure_future(
+                    create_child(database, "Ada")
+                )
+                await a_started.wait()
+                armed["active"] = False
+                task_b = asyncio.ensure_future(
+                    create_child(database, " ada ")
+                )
+                await asyncio.sleep(0)
+                gate_open.set()
+                first, second = await asyncio.gather(task_a, task_b)
+            finally:
+                logger.removeHandler(handler)
+
+            # Both creates land; the duplicate is allowed.
+            assert first.display_name == "Ada"
+            assert second.display_name == "ada"
+            names = [c.display_name for c in await list_children(database)]
+            assert names == ["Ada", "ada"]
+            # Exactly one warning: B's scan saw A's committed row.
+            warnings = [
+                r for r in seen if "duplicates existing child" in r.getMessage()
+            ]
+            assert len(warnings) == 1, (
+                "the queued duplicate create must warn exactly once"
+            )
+            assert "Ada" in warnings[0].getMessage()
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_edit_child_case_only_edit_colliding_with_other_child_warns(
+    tmp_path, caplog
+) -> None:
+    """A case-only edit must not hide a real duplicate: with two Ada
+    profiles already present, editing either one's name to another
+    casing still warns (the scan excludes only the edited child)."""
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        await create_child(database, "ada")
+        caplog.clear()
+        edited = await edit_child(database, first.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        # Editing the name to a value that STAYS duplicated (no change,
+        # other child still carries it) warns too: the scan is on the
+        # resulting state, not on whether the word moved.
+        caplog.clear()
+        await edit_child(database, first.id, display_name=" ada ")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        return None
+
+    _with_db(tmp_path, "edit-case-only-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# set_child_active / list_children
+# ---------------------------------------------------------------------------
+
+
+def test_set_child_active_round_trip(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert [c.id for c in await list_children(database)] == [child.id]
+        assert await list_children(database, active_only=True) == []
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert len(await list_children(database, active_only=True)) == 1
+        return reactivated
+
+    _with_db(tmp_path, "set-active.db")(_body)
+
+
+def test_set_child_active_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await set_child_active(database, 999, False)
+        return None
+
+    _with_db(tmp_path, "set-active-missing.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# reorder_children: complete permutation or nothing changes
+# ---------------------------------------------------------------------------
+
+
+def test_reorder_children_persists_display_order(tmp_path) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        third = await create_child(database, "Cleo", sort_order=2)
+        await reorder_children(database, [third.id, first.id, second.id])
+        # The active list returns children in the new sort order.
+        listed = await list_children(database, active_only=True)
+        assert [c.id for c in listed] == [third.id, first.id, second.id]
+        assert [c.sort_order for c in listed] == [0, 1, 2]
+        return listed
+
+    _with_db(tmp_path, "reorder-round-trip.db")(_body)
+
+
+def test_reorder_children_partial_list_rejected_order_unchanged(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        third = await create_child(database, "Cleo", sort_order=2)
+        with pytest.raises(ValueError, match="missing child ids"):
+            await reorder_children(database, [third.id, first.id])
+        listed = await list_children(database)
+        # Nothing changed: rejection happens before any write.
+        assert [c.id for c in listed] == [first.id, second.id, third.id]
+        assert [c.sort_order for c in listed] == [0, 1, 2]
+        return listed
+
+    _with_db(tmp_path, "reorder-partial.db")(_body)
+
+
+def test_reorder_children_unknown_id_rejected_order_unchanged(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        with pytest.raises(ValueError, match="unknown child ids"):
+            await reorder_children(database, [second.id, first.id, 999])
+        listed = await list_children(database)
+        assert [c.id for c in listed] == [first.id, second.id]
+        assert [c.sort_order for c in listed] == [0, 1]
+        return listed
+
+    _with_db(tmp_path, "reorder-unknown.db")(_body)
+
+
+def test_reorder_children_empty_list_rejected_when_children_exist(
+    tmp_path,
+) -> None:
+    """An empty list is the maximally partial reorder: rejected while
+    any child exists (against an empty table it is a complete
+    permutation and stays a harmless no-op)."""
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        with pytest.raises(ValueError, match="missing child ids"):
+            await reorder_children(database, [])
+        assert [c.sort_order for c in await list_children(database)] == [0]
+        return None
+
+    _with_db(tmp_path, "reorder-empty.db")(_body)
+
+
+def test_reorder_children_duplicate_and_non_int_rejected(tmp_path) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        with pytest.raises(ValueError, match="duplicate child ids"):
+            await reorder_children(database, [first.id, first.id, second.id])
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await reorder_children(database, [first.id, second.id, True])
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await reorder_children(database, [first.id, second.id, "3"])
+        listed = await list_children(database)
+        assert [c.sort_order for c in listed] == [0, 1]
+        return listed
+
+    _with_db(tmp_path, "reorder-bad-inputs.db")(_body)
+
+
+def test_reorder_children_inactive_children_included(tmp_path) -> None:
+    """Inactive children keep their display position: the permutation
+    covers the whole table, not just the active list."""
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        await set_child_active(database, second.id, False)
+        # Both ids are required even though one is inactive.
+        await reorder_children(database, [second.id, first.id])
+        listed = await list_children(database)
+        assert [c.id for c in listed] == [second.id, first.id]
+        assert [c.sort_order for c in listed] == [0, 1]
+        # The active list skips the inactive child but keeps position.
+        assert [c.id for c in await list_children(database, active_only=True)] == [
+            first.id
+        ]
+        return listed
+
+    _with_db(tmp_path, "reorder-inactive.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["1", "0", 1, 0, 1.0, None, "true"])
+def test_set_child_active_rejects_non_bool(tmp_path, bad) -> None:
+    """The DAO's int() would silently coerce "1"/0/1.0 into a state the
+    caller never asked for; the business layer refuses non-bools."""
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="is_active must be a real bool"):
+            await set_child_active(database, child.id, bad)
+        # The refused transition left the child untouched.
+        assert (await list_children(database))[0].is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-non-bool.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Concurrency: edit/set_active are atomic under the connection lock
+# ---------------------------------------------------------------------------
+
+
+def _gated_hass(armed: dict, gate_open: asyncio.Event, a_started: asyncio.Event):
+    """A hass mock whose executor gates the FIRST `_execute` job after
+    arming — the mutate statement of whichever business call starts
+    first — proving the caller holds the connection lock at that
+    moment and the second caller must queue behind it."""
+    from unittest.mock import MagicMock
+
+    hass = MagicMock()
+
+    async def _gated_executor(fn, *args):
+        if armed["active"]:
+            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
+    return hass
+
+
+def test_edit_child_concurrent_opposite_edits_return_own_values(
+    tmp_path,
+) -> None:
+    """Two racing edits cannot cross-contaminate each other's readback.
+
+    Deterministic gate: task A is paused at its UPDATE (it provably
+    holds the connection lock), task B is then created and must queue;
+    A's readback runs before B's edit, so A returns A's value and B
+    returns B's.  Without the lock, A's readback would see B's write.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "edit-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(
+                database, "Ada", colour="#111111", sort_order=0
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                edit_child(database, child.id, colour="#AAAAAA", sort_order=1)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                edit_child(
+                    database, child.id, colour="#BBBBBB", sort_order=2
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            # Each caller gets back exactly what it wrote.
+            assert record_a.colour == "#AAAAAA"
+            assert record_a.sort_order == 1
+            assert record_b.colour == "#BBBBBB"
+            assert record_b.sort_order == 2
+            # B landed last: the stored row is B's (read via the DAO —
+            # the children leak guard forbids raw probes here).
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert (stored.colour, stored.sort_order) == ("#BBBBBB", 2)
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_set_child_active_concurrent_opposite_transitions_return_own(
+    tmp_path,
+) -> None:
+    """Racing activate/deactivate calls each report their own request.
+
+    Same deterministic gate on task A's UPDATE; task B queues behind
+    the connection lock and lands the opposite transition after A.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "active-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(database, "Ada")
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                set_child_active(database, child.id, False)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                set_child_active(database, child.id, True)
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            assert record_a.is_active is False
+            assert record_b.is_active is True
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert stored.is_active is True
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_list_children_orders_by_sort_order_then_id(tmp_path) -> None:
+    async def _body(database):
+        second = await create_child(database, "Bo", sort_order=1)
+        first = await create_child(database, "Ada", sort_order=0)
+        third = await create_child(database, "Cleo", sort_order=1)
+        listed = await list_children(database)
+        # sort_order 0 first, then the two sort_order-1 rows by id.
+        assert [c.id for c in listed] == [first.id, second.id, third.id]
+        return listed
+
+    _with_db(tmp_path, "list-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Deactivation keeps every referencing row intact (Feature 03)
+# ---------------------------------------------------------------------------
+
+
+def test_deactivating_child_preserves_definitions_instances_events(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        from datetime import date, timedelta
+
+        from custom_components.nestquest.dao_children import ChildrenDao
+        from custom_components.nestquest.dao_instances import (
+            CompletionEventsDao,
+            QuestInstancesDao,
+        )
+        from custom_components.nestquest.dao_rules import (
+            QuestDefinitionsDao,
+            ScheduleRulesDao,
+        )
+
+        children = ChildrenDao(database)
+        rules = ScheduleRulesDao(database)
+        definitions = QuestDefinitionsDao(database)
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+
+        child = await create_child(database, "Ada")
+        rule = await rules.create("daily", "2026-01-01")
+        definition = await definitions.create(
+            "Brush teeth",
+            rule.id,
+            "2026-09-15T12:00:00+00:00",
+            assignee_child_ids=[child.id],
+        )
+        due = (date.today() + timedelta(days=1)).isoformat()
+        stamp = f"{due}T08:00:00+00:00"
+        instance = await instances.upsert(
+            definition.id, child.id, due, stamp, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "panel",
+            f"{due}T09:00:00+00:00",
+            True,
+            actor_child_id=child.id,
+        )
+
+        # Deactivate: the referencing rows must not move.
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert await children.get(child.id) == deactivated
+        assert (await definitions.get(definition.id)) == definition
+        assert (await instances.get_by_id(instance.id)) == instance
+        assert await events.list_by_instance(instance.id) == [event]
+        # Excluded from the active list, still present in the full list.
+        assert await list_children(database, active_only=True) == []
+        assert [c.id for c in await list_children(database)] == [child.id]
+
+        # Reactivate: fully restored.
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert [c.id for c in await list_children(database, active_only=True)] == [
+            child.id
+        ]
+        assert (await instances.get_by_id(instance.id)) == instance
+        return reactivated
+
+    _with_db(tmp_path, "deactivate-data-intact.db")(_body)
+
+
+def test_no_code_path_deletes_a_child_row() -> None:
+    """Structural guardrail: children are deactivated, never deleted.
+
+    The DAO exposes no delete method at all, and no module in the
+    package (nor any test outside the sanctioned DDL/migration layers,
+    which create the tables) carries a statement that removes child
+    rows — the leak guard in test_dao_children enforces the file
+    allowlist, this asserts the no-delete property itself across every
+    file that may legitimately name the table.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+
+    delete_pattern = re.compile(
+        r"DELETE\s+(FROM\s+)?((main|temp)\s*\.\s*)?[`'\"]*(\[)?children\b",
+        re.IGNORECASE,
+    )
+    for py in sorted(package.rglob("*.py")):
+        assert delete_pattern.search(py.read_text()) is None, (
+            f"a code path in {py.name} deletes child rows"
+        )
+
+    import inspect
+
+    from custom_components.nestquest.dao_children import ChildrenDao
+
+    methods = {
+        name
+        for name, _ in inspect.getmembers(ChildrenDao, inspect.isfunction)
+    }
+    assert not any(
+        name.startswith("delete") or name == "remove" for name in methods
+    ), f"ChildrenDao exposed a delete-shaped method: {methods}"

tests/test_const.py

@@ -73,6 +73,36 @@ def test_panel_idle_timeout_config() -> None:
     assert const.DEFAULT_PANEL_IDLE_TIMEOUT > 0
 
 
+def test_quest_windows_canonical_order() -> None:
+    """QUEST_WINDOWS is exactly the three spellings in canonical order."""
+    assert const.QUEST_WINDOWS == ("morning", "afternoon", "evening")
+    assert const.WINDOW_MORNING == "morning"
+    assert const.WINDOW_AFTERNOON == "afternoon"
+    assert const.WINDOW_EVENING == "evening"
+    # No duplicates, every name covered by its individual constant.
+    assert len(set(const.QUEST_WINDOWS)) == 3
+
+
+def test_window_clock_ranges_cover_every_window() -> None:
+    """Each window has a strict HH:MM (start, end) range.
+
+    The ranges follow the documented boundaries (D-008): morning ends
+    11:59, afternoon spans 12:00-17:00, evening 17:00-21:00, and each
+    window's end equals the next window's start (the boundary belongs
+    to the later window at classification time).
+    """
+    assert const.WINDOW_CLOCK_RANGES == {
+        "morning": ("00:00", "11:59"),
+        "afternoon": ("12:00", "17:00"),
+        "evening": ("17:00", "21:00"),
+    }
+    for window in const.QUEST_WINDOWS:
+        start, end = const.WINDOW_CLOCK_RANGES[window]
+        for value in (start, end):
+            parsed = datetime.strptime(value, "%H:%M").time()
+            assert parsed.strftime("%H:%M") == value
+
+
 def test_no_hardcoded_domain_or_db_filename_in_modules() -> None:
     """Ensure no module other than const.py hardcodes DOMAIN string or SQLITE_DB_FILENAME."""
     pkg_dir = Path(const.__file__).parent

tests/test_dao_children.py

@@ -227,24 +227,34 @@ def test_reorder_is_transactional_and_idempotent(tmp_path) -> None:
         _run(database.close())
 
 
-def test_reorder_empty_list_is_noop(tmp_path) -> None:
+def test_reorder_empty_list_is_rejected(tmp_path) -> None:
+    """An empty list is the maximally partial reorder: rejected while
+    any child exists (Feature 03 reorder contract)."""
+    import pytest as pytest_module
+
     database = _open_db(tmp_path / "reorder-empty.db")
     try:
         dao, _ = _daos(database)
         _make_child(dao, "Ada")
-        _run(dao.reorder([]))
+        with pytest_module.raises(ValueError, match="missing child ids"):
+            _run(dao.reorder([]))
         children = _run(dao.list_all())
         assert children[0].sort_order == 0
     finally:
         _run(database.close())
 
 
-def test_reorder_unknown_id_leaves_it_untouched(tmp_path) -> None:
+def test_reorder_unknown_id_is_rejected(tmp_path) -> None:
+    """An unknown id is rejected before any write; the order is
+    untouched (Feature 03 reorder contract)."""
+    import pytest as pytest_module
+
     database = _open_db(tmp_path / "reorder-unknown.db")
     try:
         dao, _ = _daos(database)
         a = _make_child(dao, "Ada")
-        _run(dao.reorder([a.id, 999]))
+        with pytest_module.raises(ValueError, match="unknown child ids"):
+            _run(dao.reorder([a.id, 999]))
         children = _run(dao.list_all())
         assert children[0].sort_order == 0
     finally:

tests/test_dao_instances.py

@@ -1,4 +1,4 @@
-"""Tests for dao_instances.py: task_instances + append-only completion_events."""
+"""Tests for dao_instances.py: quest_instances + append-only completion_events."""
 from __future__ import annotations
 
 import asyncio
@@ -11,10 +11,10 @@ from custom_components.nestquest.dao_children import ChildrenDao
 from custom_components.nestquest.dao_instances import (
     CompletionEventRecord,
     CompletionEventsDao,
-    TaskInstanceRecord,
+    QuestInstanceRecord,
-    TaskInstancesDao,
+    QuestInstancesDao,
 )
-from custom_components.nestquest.dao_rules import ScheduleRulesDao, TaskDefinitionsDao
+from custom_components.nestquest.dao_rules import ScheduleRulesDao, QuestDefinitionsDao
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
 
@@ -76,13 +76,13 @@ async def _prepare(path) -> tuple:
     await apply_migrations(database)
     children = ChildrenDao(database)
     rules = ScheduleRulesDao(database)
-    definitions = TaskDefinitionsDao(database)
+    definitions = QuestDefinitionsDao(database)
-    instances = TaskInstancesDao(database)
+    instances = QuestInstancesDao(database)
     events = CompletionEventsDao(database)
     child = await children.create("Ada", _now_stamp())
     rule = await rules.create("daily", D1)
     definition = await definitions.create(
-        "Brush teeth", child.id, rule.id, _now_stamp()
+        "Brush teeth", rule.id, _now_stamp(), assignee_child_ids=[child.id]
     )
     return (
         database,
@@ -111,7 +111,7 @@ def _with_db(tmp_path, name):
 
 
 # ---------------------------------------------------------------------------
-# task_instances: upsert idempotency
+# quest_instances: upsert idempotency
 # ---------------------------------------------------------------------------
 
 
@@ -119,15 +119,13 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
+        assert isinstance(first, QuestInstanceRecord)
-        assert isinstance(first, TaskInstanceRecord)
         second = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         assert second.id == first.id
         rows = await database.fetch_one(
-            "SELECT COUNT(*) FROM task_instances"
+            "SELECT COUNT(*) FROM quest_instances"
         )
         assert rows == (1,)
         return second
@@ -138,27 +136,116 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
 def test_upsert_different_dates_create_distinct_rows(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        a = await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        a = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
-        b = await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        b = await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         assert a.id != b.id
         return (a, b)
 
     _with_db(tmp_path, "upsert-multi-date.db")(_body)
 
 
+def test_upsert_twice_daily_fanout_two_windows_two_rows(tmp_path) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows, each idempotent on its own tuple."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        morning = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        evening = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        assert morning.id != evening.id
+        rows = await database.fetch_all(
+            "SELECT window, child_id, due_date FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ?",
+            (definition.id, D1),
+        )
+        assert sorted(rows) == [
+            ("evening", child.id, D1),
+            ("morning", child.id, D1),
+        ]
+        # Re-running both is still idempotent.
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (2,)
+        return (morning, evening)
+
+    _with_db(tmp_path, "upsert-twice-daily.db")(_body)
+
+
+def test_upsert_shared_definition_three_assignees_three_rows(
+    tmp_path,
+) -> None:
+    """D-008 shared case: one definition, three assignees, one window
+    — three rows, one per child."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        second = await children.create("Bo", _now_stamp())
+        third = await children.create("Cleo", _now_stamp())
+        await definitions.add_assignee(definition.id, second.id)
+        await definitions.add_assignee(definition.id, third.id)
+        created = [
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+            for assignee in (child, second, third)
+        ]
+        assert len({row.id for row in created}) == 3
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ? AND window = ?",
+            (definition.id, D1, "morning"),
+        )
+        assert count == (3,)
+        # Re-running the fan-out changes nothing.
+        for assignee in (child, second, third):
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (3,)
+        return created
+
+    _with_db(tmp_path, "upsert-shared-fanout.db")(_body)
+
+
+def test_upsert_window_outside_const_list_raises(tmp_path) -> None:
+    """The DAO rejects a window spelling the schema CHECK would reject
+    anyway, with the typed ValueError instead of raw IntegrityError."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        with pytest.raises(ValueError, match="window must be one of"):
+            await instances.upsert(
+                definition.id, child.id, D1, _now_stamp(), window="noon"
+            )
+        return None
+
+    _with_db(tmp_path, "upsert-bad-window.db")(_body)
+
+
 def test_upsert_refreshes_snapshot_columns_on_conflict(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="08:00"
+            definition.id, child.id, D1, _now_stamp(), due_time="08:00", window="morning")
-        )
         second = await instances.upsert(
             definition.id,
             child.id,
             D1,
             f"{D1}T13:00:00+00:00",
-            due_time="08:30",
+            due_time="08:30", window="morning")
-        )
         assert second.id == first.id
         assert second.due_time == "08:30"
         assert second.generated_at == f"{D1}T13:00:00+00:00"
@@ -172,7 +259,7 @@ def test_upsert_rejects_wrong_child_for_definition(tmp_path) -> None:
                     events, child, definition):
         other = await children.create("Bo", _now_stamp())
         with pytest.raises(ValueError, match="assigned to"):
-            await instances.upsert(definition.id, other.id, D1, _now_stamp())
+            await instances.upsert(definition.id, other.id, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-wrong-child.db")(_body)
@@ -182,20 +269,18 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         with pytest.raises(ValueError, match="definition 999"):
-            await instances.upsert(999, child.id, D1, _now_stamp())
+            await instances.upsert(999, child.id, D1, _now_stamp(), window="morning")
         # Unknown child: the definition-assignee check fires first and
-        # is the one that matters (an instance must carry its
+        # is the one that matters (an instance must carry an assigned
-        # definition's assignee).  To reach the child-existence check
+        # child).  Under D-008 the assignee check is a membership test
-        # at all, the definition would have to claim child 999 — an
+        # against quest_definition_assignees, so child 999 simply is
-        # FK-impossible state — so the child check only guards direct
+        # not an assignee — the error names the pair.
-        # callers with a matching fake assignee; assert the guard
-        # order instead of forcing the FK-impossible state.
         rule = await rules.create("daily", D1)
         other_definition = await definitions.create(
-            "X", child.id, rule.id, _now_stamp()
+            "X", rule.id, _now_stamp(), assignee_child_ids=[child.id]
         )
-        with pytest.raises(ValueError, match="not 999"):
+        with pytest.raises(ValueError, match="not assigned to child 999"):
-            await instances.upsert(other_definition.id, 999, D1, _now_stamp())
+            await instances.upsert(other_definition.id, 999, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-unknown.db")(_body)
@@ -206,7 +291,7 @@ def test_upsert_malformed_due_date_raises(tmp_path) -> None:
                     events, child, definition):
         for bad in ("not-a-date", "2026-9-15", ""):
             with pytest.raises(ValueError, match="YYYY-MM-DD"):
-                await instances.upsert(definition.id, child.id, bad, _now_stamp())
+                await instances.upsert(definition.id, child.id, bad, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-bad-date.db")(_body)
@@ -218,11 +303,11 @@ def test_upsert_rejects_past_due_dates(tmp_path) -> None:
                     events, child, definition):
         yesterday = _future_day(-1)
         with pytest.raises(ValueError, match="never generated in the past"):
-            await instances.upsert(definition.id, child.id, yesterday, _now_stamp())
+            await instances.upsert(definition.id, child.id, yesterday, _now_stamp(), window="morning")
         # Today itself is allowed (the rollover job generates today's
         # instances before the day is over).
         today = datetime.date.today().isoformat()
-        record = await instances.upsert(definition.id, child.id, today, _now_stamp())
+        record = await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         assert record.due_date == today
         return record
 
@@ -298,7 +383,7 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
             await asyncio.wait_for(holder_inside.wait(), timeout=5)
 
             task = asyncio.ensure_future(
-                instances.upsert(definition.id, child.id, today, _now_stamp())
+                instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
             )
             # A is now queued on the connection lock (it cannot pass
             # its in-lock check while B holds the lock).  Shift the
@@ -323,13 +408,16 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
             await apply_migrations(database)
             children = ChildrenDao(database)
             rules = ScheduleRulesDao(database)
-            definitions = TaskDefinitionsDao(database)
+            definitions = QuestDefinitionsDao(database)
-            instances = TaskInstancesDao(database)
+            instances = QuestInstancesDao(database)
             events = CompletionEventsDao(database)
             child = await children.create("Ada", _now_stamp())
             rule = await rules.create("daily", D1)
             definition = await definitions.create(
-                "Brush teeth", child.id, rule.id, _now_stamp()
+                "Brush teeth",
+                rule.id,
+                _now_stamp(),
+                assignee_child_ids=[child.id],
             )
             return await _body(
                 database, children, rules, definitions, instances,
@@ -348,17 +436,15 @@ def test_upsert_refuses_immutable_completed_instance(tmp_path) -> None:
                     events, child, definition):
         generated_at = _now_stamp()
         instance = await instances.upsert(
-            definition.id, child.id, D1, generated_at, due_time="08:00"
+            definition.id, child.id, D1, generated_at, due_time="08:00", window="morning")
-        )
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         with pytest.raises(ValueError, match="immutable"):
             await instances.upsert(
-                definition.id, child.id, D1, _now_stamp(), due_time="09:00"
+                definition.id, child.id, D1, _now_stamp(), due_time="09:00", window="morning")
-            )
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
-        fetched = await instances.get(definition.id, D1)
         assert fetched.due_time == "08:00", (
             "failed regeneration must not touch the completed instance"
         )
@@ -369,7 +455,7 @@ def test_upsert_refuses_immutable_completed_instance(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# task_instances: get / list
+# quest_instances: get / list
 # ---------------------------------------------------------------------------
 
 
@@ -377,12 +463,11 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         created = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="17:30"
+            definition.id, child.id, D1, _now_stamp(), due_time="17:30", window="morning")
-        )
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
-        fetched = await instances.get(definition.id, D1)
         assert fetched == created
         assert fetched.due_time == "17:30"
-        assert await instances.get(definition.id, D2) is None
+        assert await instances.get(definition.id, child.id, D2, "morning") is None
         return fetched
 
     _with_db(tmp_path, "get-round-trip.db")(_body)
@@ -391,7 +476,7 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
 def test_get_by_id_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        created = await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        created = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         fetched = await instances.get_by_id(created.id)
         assert fetched == created
         assert await instances.get_by_id(999) is None
@@ -406,13 +491,12 @@ def test_list_by_child_and_date_filters(tmp_path) -> None:
         other = await children.create("Bo", _now_stamp())
         rule = await rules.create("daily", D1)
         other_definition = await definitions.create(
-            "Make bed", other.id, rule.id, _now_stamp()
+            "Make bed", rule.id, _now_stamp(), assignee_child_ids=[other.id]
         )
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
-        await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         await instances.upsert(
-            other_definition.id, other.id, D1, _now_stamp()
+            other_definition.id, other.id, D1, _now_stamp(), window="morning")
-        )
         mine = await instances.list_by_child_and_date(child.id, D1)
         assert len(mine) == 1
         assert mine[0].child_id == child.id
@@ -426,7 +510,7 @@ def test_list_by_date_range_filters_and_orders(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         for day in (D1, D3, D5, D20):
-            await instances.upsert(definition.id, child.id, day, _now_stamp())
+            await instances.upsert(definition.id, child.id, day, _now_stamp(), window="morning")
         found = await instances.list_by_date_range(
             child.id, D2, D6
         )
@@ -453,25 +537,24 @@ def test_list_by_date_range_rejects_inverted_range(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# task_instances: delete_future_uncompleted
+# quest_instances: delete_future_uncompleted
 # ---------------------------------------------------------------------------
 
 
 def test_delete_future_uncompleted_removes_only_eligible(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         future_open = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
+            definition.id, child.id, D5, _now_stamp(), window="morning")
-        )
+        await instances.upsert(definition.id, child.id, D6, _now_stamp(), window="morning")
-        await instances.upsert(definition.id, child.id, D6, _now_stamp())
         deleted = await instances.delete_future_uncompleted(
             definition.id, D5
         )
         assert deleted == 2
-        assert await instances.get(definition.id, D1) is not None
+        assert await instances.get(definition.id, child.id, D1, "morning") is not None
-        assert await instances.get(definition.id, D5) is None
+        assert await instances.get(definition.id, child.id, D5, "morning") is None
-        assert await instances.get(definition.id, D6) is None
+        assert await instances.get(definition.id, child.id, D6, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-future.db")(_body)
@@ -481,11 +564,9 @@ def test_delete_future_uncompleted_spares_completed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         completed = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
+            definition.id, child.id, D5, _now_stamp(), window="morning")
-        )
         open_one = await instances.upsert(
-            definition.id, child.id, D6, _now_stamp()
+            definition.id, child.id, D6, _now_stamp(), window="morning")
-        )
         await events.append(
             completed.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -520,12 +601,12 @@ def test_delete_future_uncompleted_cannot_backdate_cutoff(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         today = datetime.date.today().isoformat()
-        await instances.upsert(definition.id, child.id, today, _now_stamp())
+        await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, "2020-01-01"
         )
         assert deleted == 1
-        assert await instances.get(definition.id, today) is None
+        assert await instances.get(definition.id, child.id, today, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-no-backdate.db")(_body)
@@ -537,16 +618,16 @@ def test_delete_future_uncompleted_other_definitions_untouched(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         other = await definitions.create(
-            "Make bed", child.id, (await rules.create("daily", D1)).id,
+            "Make bed", (await rules.create("daily", D1)).id,
-            _now_stamp(),
+            _now_stamp(), assignee_child_ids=[child.id],
         )
-        await instances.upsert(definition.id, child.id, D20, _now_stamp())
+        await instances.upsert(definition.id, child.id, D20, _now_stamp(), window="morning")
-        await instances.upsert(other.id, child.id, D20, _now_stamp())
+        await instances.upsert(other.id, child.id, D20, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, D20
         )
         assert deleted == 1
-        assert await instances.get(other.id, D20) is not None
+        assert await instances.get(other.id, child.id, D20, "morning") is not None
         return deleted
 
     _with_db(tmp_path, "delete-scoped.db")(_body)
@@ -561,8 +642,7 @@ def test_append_completion_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         event = await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -581,10 +661,10 @@ def test_append_panel_event_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         event = await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id,
             actor_user_id=None,
         )
         assert event.actor_source == "panel"
@@ -599,8 +679,7 @@ def test_append_uncompleted_with_optional_on_time(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         reversal = await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), None,
             actor_user_id="user-1",
@@ -621,8 +700,7 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         with pytest.raises(ValueError, match="event_type"):
             await events.append(
                 instance.id, child.id, "missed", "user", _now_stamp(), True,
@@ -641,17 +719,31 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
         with pytest.raises(ValueError, match="must not carry"):
             await events.append(
                 instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id,
                 actor_user_id="user-1",
             )
+        with pytest.raises(ValueError, match="requires actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), True,
+            )
+        with pytest.raises(ValueError, match="must not carry actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "user", _now_stamp(), True,
+                actor_user_id="user-1",
+                actor_child_id=child.id,
+            )
         with pytest.raises(ValueError, match="requires was_on_time"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), None
+                instance.id, child.id, "completed", "panel", _now_stamp(), None,
+                actor_child_id=child.id
             )
         # was_on_time must be a real bool: int-convertible junk that
         # would silently coerce (0.5 -> False) is rejected outright.
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), 0.5
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), 0.5, actor_child_id=child.id,
             )
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
@@ -669,8 +761,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         for bad in (
             "2026-09-14 12:00:00+00:00",  # space separator
             "2026-09-14T12:00:00",  # no offset
@@ -682,7 +773,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
             with pytest.raises(ValueError, match="UTC"):
                 await events.append(
                     instance.id, child.id, "completed", "panel", bad,
-                    True,
+                    True, actor_child_id=child.id,
                 )
         return None
 
@@ -694,8 +785,7 @@ def test_append_requires_was_on_time_strict_bool(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         for bad in (0.5, 1.5, 2, "true", 1):
             with pytest.raises(ValueError, match="must be True, False"):
                 await events.append(
@@ -716,8 +806,7 @@ def test_append_rejects_unknown_instance_or_child(tmp_path) -> None:
                 actor_user_id="user-1",
             )
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         with pytest.raises(ValueError, match="belongs to child"):
             await events.append(
                 instance.id, 999, "completed", "user", _now_stamp(), True,
@@ -739,17 +828,18 @@ def test_completed_uncompleted_recompleted_three_ordered_events(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), False
+            instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id
         )
         history = await events.list_by_instance(instance.id)
         assert [e.event_type for e in history] == [
@@ -770,8 +860,7 @@ def test_get_latest_for_instance_none_when_untouched(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         return await events.get_latest_for_instance(instance.id)
 
     assert _with_db(tmp_path, "latest-none.db")(_body) is None
@@ -781,10 +870,10 @@ def test_get_latest_returns_uncompleted_when_reversed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
@@ -801,26 +890,26 @@ def test_list_by_child_and_date_range_scopes_by_due_date(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         early = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         inside = await instances.upsert(
-            definition.id, child.id, D2, _now_stamp()
+            definition.id, child.id, D2, _now_stamp(), window="morning")
-        )
         late = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
+            definition.id, child.id, D20, _now_stamp(), window="morning")
+        await events.append(
+            early.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
-            early.id, child.id, "completed", "panel", _now_stamp(), True
+            inside.id, child.id, "completed", "panel", _now_stamp(), False,
-        )
+                actor_child_id=child.id
-        await events.append(
-            inside.id, child.id, "completed", "panel", _now_stamp(), False
         )
         await events.append(
             inside.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            late.id, child.id, "completed", "panel", _now_stamp(), True
+            late.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         found = await events.list_by_child_and_date_range(
             child.id, D2, D2
@@ -846,8 +935,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         late_due = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
+            definition.id, child.id, D20, _now_stamp(), window="morning")
-        )
         # The event's occurred_at is today at 08:00 UTC (_now_stamp);
         # the instance is due D20.  The due-date scope queried on D20
         # still finds the event (scoping follows the instance's due
@@ -855,7 +943,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
         # queried on today's date finds it too.
         event = await events.append(
             late_due.id, child.id, "completed", "panel",
-            _now_stamp(), False,
+            _now_stamp(), False, actor_child_id=child.id,
         )
         by_due = await events.list_by_child_and_date_range(
             child.id, D20, D20
@@ -943,7 +1031,12 @@ def test_no_mutation_sql_for_completion_events_anywhere() -> None:
     # UPDATE/DELETE/non-DAO-INSERT paths hit the pattern.
     insert_exempt = {
         "custom_components/nestquest/dao_instances.py",
+        "custom_components/nestquest/migrations.py",  # the rebuild
+        # migrations re-create the table via INSERT..SELECT (append-
+        # order preserved); UPDATE/DELETE stays forbidden everywhere
         "tests/test_schema.py",
+        "tests/test_migrations.py",  # seeds frozen legacy shapes to
+        # exercise the rebuild migrations
         "tests/test_dao_instances.py",  # this guard file; scanned by
         # its own guard's regex spans below instead
     }
@@ -1027,8 +1120,7 @@ def test_completion_events_dao_instance_has_no_mutation_capability(
 
         database.execute = _recording
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
+            definition.id, child.id, D1, _now_stamp(), window="morning")
-        )
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -1058,8 +1150,7 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
                 prepared
             )
             instance = await instances.upsert(
-                definition.id, child.id, D1, _now_stamp()
+                definition.id, child.id, D1, _now_stamp(), window="morning")
-            )
             tasks = [
                 asyncio.ensure_future(
                     events.append(
@@ -1084,13 +1175,13 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
     async def _prepare_in(database):
         children = ChildrenDao(database)
         rules = ScheduleRulesDao(database)
-        definitions = TaskDefinitionsDao(database)
+        definitions = QuestDefinitionsDao(database)
-        instances = TaskInstancesDao(database)
+        instances = QuestInstancesDao(database)
         events = CompletionEventsDao(database)
         child = await children.create("Ada", _now_stamp())
         rule = await rules.create("daily", D1)
         definition = await definitions.create(
-            "Brush teeth", child.id, rule.id, _now_stamp()
+            "Brush teeth", rule.id, _now_stamp(), assignee_child_ids=[child.id]
         )
         return (database, children, rules, definitions, instances,
                 events, child, definition)

tests/test_dao_matrix.py

@@ -27,7 +27,7 @@ from custom_components.nestquest.dao_children import (
 )
 from custom_components.nestquest.dao_instances import (
     CompletionEventsDao,
-    TaskInstancesDao,
+    QuestInstancesDao,
 )
 from custom_components.nestquest.dao_presence import (
     PresenceOverridesDao,
@@ -35,7 +35,7 @@ from custom_components.nestquest.dao_presence import (
 )
 from custom_components.nestquest.dao_rules import (
     ScheduleRulesDao,
-    TaskDefinitionsDao,
+    QuestDefinitionsDao,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
@@ -70,10 +70,11 @@ ALL_TABLES = (
     "children",
     "admin_users",
     "schedule_rules",
-    "task_definitions",
+    "quest_definitions",
+    "quest_definition_assignees",
     "presence_schedules",
     "presence_overrides",
-    "task_instances",
+    "quest_instances",
     "completion_events",
 )
 
@@ -86,10 +87,10 @@ class World:
         self.children = ChildrenDao(database)
         self.admins = AdminUsersDao(database)
         self.rules = ScheduleRulesDao(database)
-        self.definitions = TaskDefinitionsDao(database)
+        self.definitions = QuestDefinitionsDao(database)
         self.schedules = PresenceSchedulesDao(database)
         self.overrides = PresenceOverridesDao(database)
-        self.instances = TaskInstancesDao(database)
+        self.instances = QuestInstancesDao(database)
         self.events = CompletionEventsDao(database)
 
     async def seed(self) -> None:
@@ -99,7 +100,10 @@ class World:
         assert await self.admins.add("user-1", _stamp()) is True
         self.rule = await self.rules.create("daily", D1)
         self.definition = await self.definitions.create(
-            "Brush teeth", self.child.id, self.rule.id, _stamp()
+            "Brush teeth",
+            self.rule.id,
+            _stamp(),
+            assignee_child_ids=[self.child.id],
         )
         self.schedule = await self.schedules.upsert_by_child(
             self.child.id, 2, D1, "0,2,4|1,3"
@@ -108,7 +112,8 @@ class World:
             self.child.id, D1, D2, True, note="test"
         )
         self.instance = await self.instances.upsert(
-            self.definition.id, self.child.id, D1, _stamp()
+            self.definition.id, self.child.id, D1, _stamp(),
+            window="morning",
         )
 
 
@@ -201,11 +206,14 @@ def test_matrix_schedule_rules_crud_and_constraints(tmp_path) -> None:
     assert _with_world(tmp_path / "m-rules.db")(_body) is True
 
 
-def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None:
+def test_matrix_quest_definitions_crud_and_constraints(tmp_path) -> None:
     async def _body(w: World):
         # insert (seeded) + read
         fetched = await w.definitions.get(w.definition.id)
         assert fetched.title == "Brush teeth"
+        assert [c.id for c in await w.definitions.list_assignees(
+            w.definition.id
+        )] == [w.child.id]
         # update
         await w.definitions.update(w.definition.id, title="Brush TEETH")
         assert (await w.definitions.get(w.definition.id)).title == (
@@ -213,11 +221,21 @@ def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None:
         )
         # constraint: FK to a nonexistent child rejected with ValueError
         with pytest.raises(ValueError, match="does not exist"):
-            await w.definitions.create("X", 999, w.rule.id, _stamp())
+            await w.definitions.create(
-        # constraint: inactive child rejected
+                "X", w.rule.id, _stamp(), assignee_child_ids=[999]
+            )
+        # constraint: inactive child rejected on assignment
         await w.children.set_active(w.child2.id, False)
         with pytest.raises(ValueError, match="inactive"):
-            await w.definitions.set_assignee(w.definition.id, w.child2.id)
+            await w.definitions.add_assignee(w.definition.id, w.child2.id)
+        # assignee removal + re-adding the active child
+        assert await w.definitions.remove_assignee(
+            w.definition.id, w.child.id
+        ) is True
+        assert await w.definitions.remove_assignee(
+            w.definition.id, w.child.id
+        ) is False
+        await w.definitions.add_assignee(w.definition.id, w.child.id)
         # set_active: deactivate, never delete
         assert await w.definitions.set_active(w.definition.id, False) == 1
         assert (await w.definitions.get(w.definition.id)).is_active is False
@@ -271,15 +289,18 @@ def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
     assert _with_world(tmp_path / "m-overrides.db")(_body) is True
 
 
-def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
+def test_matrix_quest_instances_crud_and_constraints(tmp_path) -> None:
     async def _body(w: World):
         # insert (seeded) + read
-        fetched = await w.instances.get(w.definition.id, D1)
+        fetched = await w.instances.get(
+            w.definition.id, w.child.id, D1, "morning"
+        )
         assert fetched == w.instance
         # update path: conflict refresh on the OPEN instance — same row
         # id, snapshot columns (due_time) actually rewritten.
         refreshed = await w.instances.upsert(
-            w.definition.id, w.child.id, D1, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D1, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert refreshed.id == fetched.id, (
             "the conflict path must refresh the existing row in place"
@@ -292,22 +313,22 @@ def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
     assert _with_world(tmp_path / "m-instances.db")(_body) is True
 
 
-def test_matrix_task_instances_constraints(tmp_path) -> None:
+def test_matrix_quest_instances_constraints(tmp_path) -> None:
     async def _body(w: World):
         # D20 has no instance yet: a raw insert succeeds, then an
         # identical second insert must fail on the UNIQUE constraint —
         # proving the schema backs the DAO's idempotency.
         await w.database.execute(
-            "INSERT INTO task_instances (definition_id, child_id, "
+            "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, due_time, generated_at) "
+            "window, due_date, due_time, generated_at) "
-            "VALUES (?, ?, ?, NULL, ?)",
+            "VALUES (?, ?, 'morning', ?, NULL, ?)",
             (w.definition.id, w.child.id, D20, _stamp()),
         )
         with pytest.raises(sqlite3.IntegrityError):
             await w.database.execute(
-                "INSERT INTO task_instances (definition_id, child_id, "
+                "INSERT INTO quest_instances (definition_id, child_id, "
-                "due_date, due_time, generated_at) "
+                "window, due_date, due_time, generated_at) "
-                "VALUES (?, ?, ?, NULL, ?)",
+                "VALUES (?, ?, 'morning', ?, NULL, ?)",
                 (w.definition.id, w.child.id, D20, _stamp()),
             )
         # delete future uncompleted removes D20's open instance
@@ -320,17 +341,18 @@ def test_matrix_task_instances_constraints(tmp_path) -> None:
     assert _with_world(tmp_path / "m-instances-constraints.db")(_body) is True
 
 
-def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None:
+def test_matrix_quest_instances_upsert_idempotency(tmp_path) -> None:
     async def _body(w: World):
         first = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp()
+            w.definition.id, w.child.id, D2, _stamp(), window="morning"
         )
         second = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D2, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert second.id == first.id
         count = await w.database.fetch_one(
-            "SELECT COUNT(*) FROM task_instances"
+            "SELECT COUNT(*) FROM quest_instances"
         )
         # Seed created one (D1) plus this one (D2) — no duplicate.
         assert count == (2,)
@@ -350,14 +372,16 @@ def test_matrix_completion_events_append_read_and_no_mutation(
     async def _body(w: World):
         # insert: append a completion then a reversal, then re-complete
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), True
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), True,
+            actor_child_id=w.child.id,
         )
         await w.events.append(
             w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None,
             actor_user_id="user-1",
         )
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), False
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), False,
+            actor_child_id=w.child.id,
         )
         # read: full ordered history + latest
         history = await w.events.list_by_instance(w.instance.id)

tests/test_dao_rules.py

@@ -1,4 +1,4 @@
-"""Tests for dao_rules.py: typed DAO for schedule_rules and task_definitions."""
+"""Tests for dao_rules.py: typed DAO for schedule_rules and quest_definitions."""
 from __future__ import annotations
 
 import asyncio
@@ -10,8 +10,9 @@ from custom_components.nestquest.dao_children import ChildrenDao
 from custom_components.nestquest.dao_rules import (
     ScheduleRuleRecord,
     ScheduleRulesDao,
-    TaskDefinitionRecord,
+    QuestDefinitionRecord,
-    TaskDefinitionsDao,
+    QuestDefinitionWindowRecord,
+    QuestDefinitionsDao,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
@@ -37,7 +38,7 @@ async def _prepare(path) -> tuple:
     await database.open(path)
     await apply_migrations(database)
     rules = ScheduleRulesDao(database)
-    definitions = TaskDefinitionsDao(database)
+    definitions = QuestDefinitionsDao(database)
     children = ChildrenDao(database)
     child = await children.create("Ada", NOW)
     return database, rules, definitions, children, child
@@ -266,7 +267,7 @@ def test_rule_delete_unreferenced_succeeds(tmp_path) -> None:
 def test_rule_delete_referenced_is_rejected(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        await definitions.create("Brush teeth", child.id, rule.id, NOW)
+        await definitions.create("Brush teeth", rule.id, NOW)
         assert await rules.delete_if_unreferenced(rule.id) is False
         fetched = await rules.get(rule.id)
         assert fetched is not None, (
@@ -288,7 +289,7 @@ def test_rule_delete_after_last_reference_removed_succeeds(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         definition = await definitions.create(
-            "Brush teeth", child.id, rule.id, NOW
+            "Brush teeth", rule.id, NOW
         )
         assert await rules.delete_if_unreferenced(rule.id) is False
         await definitions.set_active(definition.id, False)
@@ -322,7 +323,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
             await apply_migrations(database)
             children = ChildrenDao(database)
             rules = ScheduleRulesDao(database)
-            definitions = TaskDefinitionsDao(database)
+            definitions = QuestDefinitionsDao(database)
             child = await children.create("Ada", NOW)
             rule_a = await rules.create("daily", "2026-09-14")
 
@@ -335,12 +336,12 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
                 import custom_components.nestquest.dao_rules as dao_rules
 
                 original_validate = (
-                    dao_rules.TaskDefinitionsDao._validate_assignable
+                    dao_rules.QuestDefinitionsDao._validate_rule_exists
                 )
                 release = asyncio.Event()
                 started = asyncio.Event()
 
-                async def _pausing_validate(self, child_id, rule_id):
+                async def _pausing_validate(self, rule_id):
                     # Run the real validation FIRST (this is the
                     # time-of-check), THEN pause: the delete task gets
                     # scheduled while the create sits between its
@@ -348,19 +349,17 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
                     # transaction must keep the delete queued until the
                     # insert commits, so the create must win the race
                     # and the delete must be rejected as referenced.
-                    await original_validate(self, child_id, rule_id)
+                    await original_validate(self, rule_id)
                     if rule_id == rule_a.id and not started.is_set():
                         started.set()
                         await release.wait()
 
-                dao_rules.TaskDefinitionsDao._validate_assignable = (
+                dao_rules.QuestDefinitionsDao._validate_rule_exists = (
                     _pausing_validate
                 )
                 try:
                     create_task = asyncio.ensure_future(
-                        definitions.create(
+                        definitions.create("Brush teeth", rule_a.id, NOW)
-                            "Brush teeth", child.id, rule_a.id, NOW
-                        )
                     )
                     await started.wait()
                     delete_task = asyncio.ensure_future(
@@ -373,7 +372,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
                         delete_task, create_task, return_exceptions=True
                     )
                 finally:
-                    dao_rules.TaskDefinitionsDao._validate_assignable = (
+                    dao_rules.QuestDefinitionsDao._validate_assignable = (
                         original_validate
                     )
             else:
@@ -382,7 +381,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
 
                 async def _create():
                     return await definitions.create(
-                        "Brush teeth", child.id, rule_a.id, NOW
+                        "Brush teeth", rule_a.id, NOW
                     )
 
                 delete_task = asyncio.ensure_future(_delete())
@@ -405,7 +404,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
                 assert rule_row is None
             else:
                 # Create won: the delete was rejected, both rows exist.
-                assert isinstance(creation, TaskDefinitionRecord), (
+                assert isinstance(creation, QuestDefinitionRecord), (
                     f"create must succeed, got {creation!r}"
                 )
                 assert rule_row is not None
@@ -425,7 +424,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
 
 
 # ---------------------------------------------------------------------------
-# task_definitions: create / get
+# quest_definitions: create / get
 # ---------------------------------------------------------------------------
 
 
@@ -434,20 +433,24 @@ def test_definition_create_returns_typed_record(tmp_path) -> None:
         rule = await rules.create("weekly", "2026-09-14", weekday_set="0,2")
         definition = await definitions.create(
             "Brush teeth",
-            child.id,
             rule.id,
             NOW,
             description="Morning and night",
             icon="mdi:tooth",
             due_time="08:00",
+            assignee_child_ids=[child.id],
         )
-        assert isinstance(definition, TaskDefinitionRecord)
+        assert isinstance(definition, QuestDefinitionRecord)
         assert definition.title == "Brush teeth"
-        assert definition.child_id == child.id
         assert definition.schedule_rule_id == rule.id
         assert definition.due_time == "08:00"
         assert definition.is_active is True
         assert definition.created_at == NOW
+        assert not hasattr(definition, "child_id"), (
+            "D-008: assignment lives in quest_definition_assignees"
+        )
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [child.id]
         return definition
 
     _with_db(tmp_path, "definition-create.db")(_body)
@@ -460,26 +463,30 @@ def test_definition_get_missing_returns_none(tmp_path) -> None:
     assert _with_db(tmp_path, "definition-get-missing.db")(_body) is None
 
 
-def test_definition_create_unknown_child_raises_value_error(
+def test_definition_create_unknown_assignee_raises_value_error(
     tmp_path,
 ) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         with pytest.raises(ValueError, match="does not exist"):
-            await definitions.create("X", 999, rule.id, NOW)
+            await definitions.create(
+                "X", rule.id, NOW, assignee_child_ids=[999]
+            )
         return None
 
     _with_db(tmp_path, "definition-bad-child.db")(_body)
 
 
-def test_definition_create_inactive_child_raises_value_error(
+def test_definition_create_inactive_assignee_raises_value_error(
     tmp_path,
 ) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         await children.set_active(child.id, False)
         with pytest.raises(ValueError, match="inactive"):
-            await definitions.create("X", child.id, rule.id, NOW)
+            await definitions.create(
+                "X", rule.id, NOW, assignee_child_ids=[child.id]
+            )
         return None
 
     _with_db(tmp_path, "definition-inactive-child.db")(_body)
@@ -490,14 +497,14 @@ def test_definition_create_unknown_rule_raises_value_error(
 ) -> None:
     async def _body(database, rules, definitions, children, child):
         with pytest.raises(ValueError, match="does not exist"):
-            await definitions.create("X", child.id, 999, NOW)
+            await definitions.create("X", 999, NOW)
         return None
 
     _with_db(tmp_path, "definition-bad-rule.db")(_body)
 
 
 # ---------------------------------------------------------------------------
-# task_definitions: list by child / list active
+# quest_definitions: list by child / list active
 # ---------------------------------------------------------------------------
 
 
@@ -505,9 +512,11 @@ def test_definition_list_by_child_filters_and_orders(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other_child = await children.create("Bo", NOW)
-        await definitions.create("A", child.id, rule.id, NOW)
+        await definitions.create("A", rule.id, NOW, assignee_child_ids=[child.id])
-        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.create("B", rule.id, NOW, assignee_child_ids=[child.id])
-        await definitions.create("C", other_child.id, rule.id, NOW)
+        await definitions.create(
+            "C", rule.id, NOW, assignee_child_ids=[other_child.id]
+        )
         mine = await definitions.list_by_child(child.id)
         assert [d.title for d in mine] == ["B", "A"]
         return mine
@@ -527,8 +536,8 @@ def test_definition_list_by_child_missing_child_returns_empty(
 def test_definition_list_active_excludes_inactive(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        first = await definitions.create("A", child.id, rule.id, NOW)
+        first = await definitions.create("A", rule.id, NOW)
-        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.create("B", rule.id, NOW)
         await definitions.set_active(first.id, False)
         active = await definitions.list_active()
         assert [d.title for d in active] == ["B"]
@@ -539,7 +548,7 @@ def test_definition_list_active_excludes_inactive(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# task_definitions: update / set_active / set_assignee
+# quest_definitions: update / set_active / set_assignee
 # ---------------------------------------------------------------------------
 
 
@@ -547,7 +556,7 @@ def test_definition_update_fields_round_trip(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         definition = await definitions.create(
-            "A", child.id, rule.id, NOW, due_time="08:00"
+            "A", rule.id, NOW, due_time="08:00"
         )
         assert await definitions.update(
             definition.id,
@@ -572,9 +581,11 @@ def test_definition_update_cannot_change_assignee_or_active(
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         # title= / child_id= / is_active= are NOT update() parameters;
-        # this asserts the API shape: reassignment and activation have
+        # this asserts the API shape: assignment and activation have
         # their own explicit methods.
         import inspect
 
@@ -590,7 +601,10 @@ def test_definition_update_cannot_change_assignee_or_active(
             definition.id, title="A2"
         ) == 1
         fetched = await definitions.get(definition.id)
-        assert fetched.child_id == child.id
+        assert [c.id for c in await definitions.list_assignees(definition.id)] == [
+            child.id
+        ]
+        assert other.id not in [c.id for c in await definitions.list_assignees(definition.id)]
         assert fetched.is_active is True
         return fetched
 
@@ -600,7 +614,7 @@ def test_definition_update_cannot_change_assignee_or_active(
 def test_definition_update_no_fields_returns_zero(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create("A", rule.id, NOW)
         return await definitions.update(definition.id)
 
     assert _with_db(tmp_path, "definition-update-empty.db")(_body) == 0
@@ -612,7 +626,6 @@ def test_definition_update_can_clear_optional_fields(tmp_path) -> None:
         rule = await rules.create("daily", "2026-09-14")
         definition = await definitions.create(
             "A",
-            child.id,
             rule.id,
             NOW,
             description="d",
@@ -646,7 +659,7 @@ def test_definition_update_missing_returns_zero(tmp_path) -> None:
 def test_definition_set_active_round_trip(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create("A", rule.id, NOW)
         assert await definitions.set_active(definition.id, False) == 1
         fetched = await definitions.get(definition.id)
         assert fetched.is_active is False
@@ -673,7 +686,7 @@ def test_definition_no_delete_method_exists(tmp_path) -> None:
         methods = {
             name
             for name, _ in inspect.getmembers(
-                TaskDefinitionsDao, inspect.isfunction
+                QuestDefinitionsDao, inspect.isfunction
             )
         }
         assert "delete" not in methods
@@ -683,78 +696,244 @@ def test_definition_no_delete_method_exists(tmp_path) -> None:
     _with_db(tmp_path, "definition-no-delete.db")(_body)
 
 
-def test_definition_set_assignee_round_trip(tmp_path) -> None:
+def test_definition_add_assignee_round_trip(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
-        assert await definitions.set_assignee(definition.id, other.id) == 1
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
-        fetched = await definitions.get(definition.id)
+        )
-        assert fetched.child_id == other.id
+        await definitions.add_assignee(definition.id, other.id)
-        return fetched
+        assignees = await definitions.list_assignees(definition.id)
+        assert sorted(c.id for c in assignees) == sorted([child.id, other.id])
+        # Idempotent: re-adding an existing assignee stays a no-op.
+        await definitions.add_assignee(definition.id, other.id)
+        assignees = await definitions.list_assignees(definition.id)
+        assert sorted(c.id for c in assignees) == sorted([child.id, other.id])
+        return assignees
 
-    _with_db(tmp_path, "definition-set-assignee.db")(_body)
+    _with_db(tmp_path, "definition-add-assignee.db")(_body)
 
 
-def test_definition_set_assignee_inactive_child_raises(tmp_path) -> None:
+def test_definition_add_assignee_inactive_child_raises(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         await children.set_active(other.id, False)
         with pytest.raises(ValueError, match="inactive"):
-            await definitions.set_assignee(definition.id, other.id)
+            await definitions.add_assignee(definition.id, other.id)
-        fetched = await definitions.get(definition.id)
+        assignees = await definitions.list_assignees(definition.id)
-        assert fetched.child_id == child.id, (
+        assert [c.id for c in assignees] == [child.id], (
-            "failed reassignment must leave the current assignee intact"
+            "failed assignment must leave the current assignees intact"
         )
-        return fetched
+        return assignees
 
     _with_db(tmp_path, "definition-assignee-inactive.db")(_body)
 
 
-def test_definition_set_assignee_unknown_child_raises(tmp_path) -> None:
+def test_definition_add_assignee_unknown_child_raises(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         with pytest.raises(ValueError, match="does not exist"):
-            await definitions.set_assignee(definition.id, 999)
+            await definitions.add_assignee(definition.id, 999)
         return None
 
     _with_db(tmp_path, "definition-assignee-unknown.db")(_body)
 
 
-def test_definition_set_assignee_missing_definition_returns_zero(
+def test_definition_add_assignee_missing_definition_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.add_assignee(999, child.id)
+        return None
+
+    _with_db(tmp_path, "definition-assignee-missing.db")(_body)
+
+
+def test_definition_remove_assignee_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id, other.id]
+        )
+        assert await definitions.remove_assignee(definition.id, other.id) is True
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [child.id]
+        # Removing a non-assignee reports False.
+        assert await definitions.remove_assignee(definition.id, other.id) is False
+        return assignees
+
+    _with_db(tmp_path, "definition-remove-assignee.db")(_body)
+
+
+def test_definition_list_assignees_orders_by_child_sort_order(
     tmp_path,
 ) -> None:
     async def _body(database, rules, definitions, children, child):
-        other = await children.create("Bo", NOW)
+        rule = await rules.create("daily", "2026-09-14")
-        return await definitions.set_assignee(999, other.id)
+        second = await children.create("Bo", NOW)
+        third = await children.create("Cleo", NOW)
+        await children.reorder([third.id, child.id, second.id])
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[second.id, child.id, third.id]
+        )
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [third.id, child.id, second.id]
+        return assignees
 
-    assert _with_db(tmp_path, "definition-assignee-missing.db")(_body) == 0
+    _with_db(tmp_path, "definition-assignees-order.db")(_body)
 
 
-def test_definition_set_assignee_leaves_history_rows_alone(
+def test_definition_upsert_window_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        window = await definitions.upsert_window(
+            definition.id, "morning", due_time="09:00"
+        )
+        assert window == QuestDefinitionWindowRecord(
+            definition.id, "morning", "09:00"
+        )
+        # Upsert: re-declaring the same window updates in place, never
+        # duplicates (one row per definition+window).
+        updated = await definitions.upsert_window(
+            definition.id, "morning", due_time="09:30"
+        )
+        assert updated.due_time == "09:30"
+        rows = await definitions.list_windows(definition.id)
+        assert [w.window for w in rows] == ["morning"]
+        # All three windows list in const order regardless of insertion.
+        await definitions.upsert_window(definition.id, "evening")
+        await definitions.upsert_window(definition.id, "afternoon")
+        rows = await definitions.list_windows(definition.id)
+        assert [w.window for w in rows] == [
+            "morning",
+            "afternoon",
+            "evening",
+        ]
+        assert rows[1].due_time is None
+        return rows
+
+    _with_db(tmp_path, "definition-window-roundtrip.db")(_body)
+
+
+def test_definition_window_unknown_name_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        with pytest.raises(ValueError, match="window must be one of"):
+            await definitions.upsert_window(definition.id, "noon")
+        with pytest.raises(ValueError, match="window must be one of"):
+            await definitions.remove_window(definition.id, "MORNING")
+        return None
+
+    _with_db(tmp_path, "definition-window-bad-name.db")(_body)
+
+
+def test_definition_window_bad_due_time_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        with pytest.raises(ValueError, match="HH:MM"):
+            await definitions.upsert_window(
+                definition.id, "morning", due_time="9:30"
+            )
+        with pytest.raises(ValueError, match="HH:MM"):
+            await definitions.upsert_window(
+                definition.id, "morning", due_time="25:00"
+            )
+        return None
+
+    _with_db(tmp_path, "definition-window-bad-time.db")(_body)
+
+
+def test_definition_window_missing_definition_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.upsert_window(999, "morning")
+        return None
+
+    _with_db(tmp_path, "definition-window-missing.db")(_body)
+
+
+def test_definition_remove_window_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        await definitions.upsert_window(definition.id, "morning")
+        assert await definitions.remove_window(
+            definition.id, "morning"
+        ) is True
+        assert await definitions.list_windows(definition.id) == []
+        # Removing an absent window reports False.
+        assert await definitions.remove_window(
+            definition.id, "morning"
+        ) is False
+        return None
+
+    _with_db(tmp_path, "definition-window-remove.db")(_body)
+
+
+def test_definition_unassigned_has_no_assignees(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        assert await definitions.list_assignees(definition.id) == []
+        return None
+
+    _with_db(tmp_path, "definition-no-assignees.db")(_body)
+
+
+def test_definition_shared_by_multiple_children_via_list_by_child(
     tmp_path,
 ) -> None:
-    """Reassignment writes the definition row only: any instances that
+    """D-008: one definition covering several children appears for
-    already exist keep their generation-time child_id (Feature 06
+    every assignee through list_by_child, once each."""
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        second = await children.create("Bo", NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id, second.id]
+        )
+        for assignee in (child, second):
+            mine = await definitions.list_by_child(assignee.id)
+            assert [d.id for d in mine] == [definition.id]
+        return definition
+
+    _with_db(tmp_path, "definition-shared.db")(_body)
+
+
+def test_definition_assignment_leaves_history_rows_alone(
+    tmp_path,
+) -> None:
+    """Assignment edits write the assignees table only: any instances
+    that already exist keep their generation-time child_id (Feature 06
     guardrail 'changes future instances only').
     """
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         # Simulate an already-generated instance (materialization owns
-        # this table, but its row shape proves reassignment isolation).
+        # this table, but its row shape proves assignment isolation).
         await database.execute(
-            "INSERT INTO task_instances (definition_id, child_id, "
+            "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, generated_at) VALUES (?, ?, ?, ?)",
+            "window, due_date, generated_at) VALUES (?, ?, 'morning', ?, ?)",
             (definition.id, child.id, "2026-09-15", NOW),
         )
-        await definitions.set_assignee(definition.id, other.id)
+        await definitions.add_assignee(definition.id, other.id)
+        await definitions.remove_assignee(definition.id, child.id)
         row = await database.fetch_one(
-            "SELECT child_id FROM task_instances WHERE definition_id = ?",
+            "SELECT child_id FROM quest_instances WHERE definition_id = ?",
             (definition.id,),
         )
         assert row == (child.id,)
@@ -769,7 +948,7 @@ def test_definition_set_assignee_leaves_history_rows_alone(
 
 
 def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
-    """Guardrail: schedule_rules/task_definitions SQL may appear only
+    """Guardrail: schedule_rules/quest_definitions SQL may appear only
     in the DAO modules (and the schema DDL declarations).  RECURSIVE
     scan of package and tests, FROM/INTO/UPDATE/DELETE FROM/JOIN
     pattern, with exact repo-relative exclusions justified by role:
@@ -801,7 +980,7 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
     }
     sql_pattern = re.compile(
         r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
-        r"(schedule_rules|task_definitions)\b",
+        r"(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     offenders: list[str] = []
@@ -813,16 +992,16 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
             if sql_pattern.search(py.read_text()):
                 offenders.append(relative)
     assert offenders == [], (
-        f"SQL touching schedule_rules/task_definitions leaked into: "
+        f"SQL touching schedule_rules/quest_definitions/assignees leaked: "
         f"{offenders}"
     )
 
 
 def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
     """Compensating self-scan for the test_dao_instances.py exemption:
-    that file may query task_instances/completion_events raw (its own
+    that file may query quest_instances/completion_events raw (its own
     guard's scope) but must go through the DAO for schedule_rules and
-    task_definitions, except its single sanctioned definition-create
+    quest_definitions, except its single sanctioned definition-create
     helper usage and guard spans.
     """
     import ast
@@ -853,12 +1032,12 @@ def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
     )
     sql_pattern = re.compile(
         r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
-        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|task_definitions)\b",
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (
         "test_dao_instances.py must go through the DAO, not raw SQL, "
-        "for schedule_rules/task_definitions"
+        "for schedule_rules/quest_definitions"
     )
 
 
@@ -866,7 +1045,7 @@ def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
     """This test module must exercise the DAO, not raw SQL, for these
     tables.  Exceptions: the guard functions' own spans (their regexes
     mention the names), and the reassignment-isolation test, which
-    legitimately touches task_instances — a table this DAO does not
+    legitimately touches quest_instances — a table this DAO does not
     own — to prove the guardrail that reassignment leaves it alone.
     """
     import ast
@@ -898,7 +1077,7 @@ def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
     )
     sql_pattern = re.compile(
         r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
-        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|task_definitions)\b",
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (

tests/test_lifecycle.py

@@ -247,12 +247,12 @@ async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
     rows = await database.fetch_all(
         "SELECT name FROM sqlite_master WHERE type = 'table' "
         "AND name IN ('children', 'admin_users', 'schedule_rules', "
-        "'task_definitions') ORDER BY name"
+        "'quest_definitions') ORDER BY name"
     )
     assert [row[0] for row in rows] == [
         "admin_users",
         "children",
+        "quest_definitions",
         "schedule_rules",
-        "task_definitions",
     ]
     await async_unload_entry(hass, entry)

tests/test_migrations.py

@@ -17,7 +17,10 @@ from custom_components.nestquest.migrations import (
     apply_migrations,
     read_schema_version,
 )
-from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS
+from custom_components.nestquest.schema import (
+    SCHEMA_V1_STATEMENTS,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+)
 
 
 def _run(coro):
@@ -62,10 +65,12 @@ def _counts(database) -> dict[str, int]:
         "children",
         "admin_users",
         "schedule_rules",
-        "task_definitions",
+        "quest_definitions",
+        "quest_definition_assignees",
+        "quest_definition_windows",
         "presence_schedules",
         "presence_overrides",
-        "task_instances",
+        "quest_instances",
         "completion_events",
     )
     counts = {}
@@ -76,16 +81,16 @@ def _counts(database) -> dict[str, int]:
 
 
 # ---------------------------------------------------------------------------
-# Fresh file: version 0 -> version 1
+# Fresh file: version 0 -> latest version
 # ---------------------------------------------------------------------------
 
 
-def test_fresh_file_migrates_to_version_1(tmp_path) -> None:
+def test_fresh_file_migrates_to_latest_version(tmp_path) -> None:
     database = _open_db(tmp_path / "fresh.db")
     try:
         final = _migrate(database)
-        assert final == 1
+        assert final == len(MIGRATIONS)
-        assert _version(database) == 1
+        assert _version(database) == len(MIGRATIONS)
     finally:
         _run(database.close())
 
@@ -98,10 +103,12 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "children",
             "admin_users",
             "schedule_rules",
-            "task_definitions",
+            "quest_definitions",
+            "quest_definition_assignees",
+            "quest_definition_windows",
             "presence_schedules",
             "presence_overrides",
-            "task_instances",
+            "quest_instances",
             "completion_events",
             VERSION_TABLE,
         }
@@ -135,9 +142,496 @@ def test_empty_version_table_reads_as_version_0(tmp_path) -> None:
         _run(database.close())
 
 
-def test_migration_1_is_the_full_v1_ddl() -> None:
+def test_migration_list_shape() -> None:
+    """Migration 1 is the v1 DDL; 2 and 3 are callable steps.
+
+    Nothing ever shipped, so migration 1 was rewritten pre-release
+    (D-007) to create the quest_* tables directly; migration 2 exists
+    for dev databases stamped version 1 by the pre-rename runner,
+    migration 3 brings pre-D-008 definitions to the multi-assignee
+    model, migration 4 adds the windows table, migration 5 rebuilds
+    quest_instances onto the widened (definition, child, date, window)
+    key, and migration 6 adds actor_child_id to completion_events.
+    """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
-    assert len(MIGRATIONS) == 1
+    assert callable(MIGRATIONS[1])
+    assert callable(MIGRATIONS[2])
+    assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
+    assert callable(MIGRATIONS[4])
+    assert callable(MIGRATIONS[5])
+    assert len(MIGRATIONS) == 6
+
+
+# ---------------------------------------------------------------------------
+# Legacy dev database: task_* tables stamped version 1 -> quest_*
+# ---------------------------------------------------------------------------
+
+
+_LEGACY_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS children (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        display_name TEXT NOT NULL,
+        colour TEXT,
+        avatar_ref TEXT,
+        sort_order INTEGER NOT NULL DEFAULT 0,
+        is_active INTEGER NOT NULL DEFAULT 1,
+        created_at TEXT NOT NULL
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS admin_users (
+        ha_user_id TEXT PRIMARY KEY NOT NULL,
+        added_at TEXT NOT NULL
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS schedule_rules (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        rule_type TEXT NOT NULL,
+        interval INTEGER NOT NULL DEFAULT 1,
+        weekday_set TEXT,
+        day_of_month INTEGER,
+        nth_weekday INTEGER,
+        month INTEGER,
+        start_date TEXT NOT NULL,
+        end_date TEXT
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS task_definitions (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        title TEXT NOT NULL,
+        description TEXT,
+        icon TEXT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        due_time TEXT,
+        is_active INTEGER NOT NULL DEFAULT 1,
+        created_at TEXT NOT NULL
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS presence_schedules (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        child_id INTEGER NOT NULL UNIQUE REFERENCES children(id),
+        cycle_length_weeks INTEGER NOT NULL,
+        anchor_date TEXT NOT NULL,
+        pattern TEXT NOT NULL
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS presence_overrides (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        start_date TEXT NOT NULL,
+        end_date TEXT NOT NULL,
+        is_present INTEGER NOT NULL,
+        note TEXT
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS task_instances (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES task_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        due_date TEXT NOT NULL,
+        due_time TEXT,
+        generated_at TEXT NOT NULL,
+        UNIQUE (definition_id, due_date)
+    )
+    """,
+    """
+    CREATE TABLE IF NOT EXISTS completion_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        instance_id INTEGER NOT NULL REFERENCES task_instances(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        event_type TEXT NOT NULL,
+        actor_source TEXT NOT NULL,
+        actor_user_id TEXT,
+        occurred_at TEXT NOT NULL,
+        was_on_time INTEGER
+    )
+    """,
+]
+
+
+def _seed_legacy_database(database) -> None:
+    """Create a database indistinguishable from a pre-rename dev file.
+
+    Applies the legacy task_* DDL, seeds one row in every renamed
+    table, and stamps version 1 exactly the way the old runner did.
+    """
+    for sql in _LEGACY_DDL:
+        _run(database.execute(sql))
+    _run(database.execute(VERSION_TABLE_DDL))
+    _run(
+        database.execute(
+            f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, 1)"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO children (display_name, created_at) "
+            "VALUES ('Ada', '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO schedule_rules (rule_type, start_date) "
+            "VALUES ('daily', '2026-09-14')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO task_definitions (title, child_id, "
+            "schedule_rule_id, created_at) "
+            "VALUES ('Dishes', 1, 1, '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO task_instances (definition_id, child_id, "
+            "due_date, generated_at) "
+            "VALUES (1, 1, '2026-09-15', '2026-09-14T00:00:00+00:00')"
+        )
+    )
+
+
+def test_legacy_task_tables_are_renamed_and_keep_their_rows(tmp_path) -> None:
+    database = _open_db(tmp_path / "legacy.db")
+    try:
+        _seed_legacy_database(database)
+        final = _migrate(database)
+        assert final == len(MIGRATIONS)
+        assert _version(database) == len(MIGRATIONS)
+        tables = _tables(database)
+        assert "task_definitions" not in tables
+        assert "task_instances" not in tables
+        assert {"quest_definitions", "quest_instances"} <= tables
+        # Every seeded row survived the rename.
+        assert _run(database.fetch_one("SELECT COUNT(*) FROM children")) == (1,)
+        assert (
+            _run(
+                database.fetch_one(
+                    "SELECT title FROM quest_definitions WHERE id = 1"
+                )
+            )
+            == ("Dishes",)
+        )
+        assert (
+            _run(
+                database.fetch_one(
+                    "SELECT due_date FROM quest_instances WHERE id = 1"
+                )
+            )
+            == ("2026-09-15",)
+        )
+        # Migration 3 moved the legacy single-assignee column into the
+        # assignees table: the definition kept its child, the column is
+        # gone.
+        assignees = _run(
+            database.fetch_one(
+                "SELECT child_id FROM quest_definition_assignees "
+                "WHERE definition_id = 1"
+            )
+        )
+        assert assignees == (1,)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definitions)")
+        )
+        assert "child_id" not in {row[1] for row in columns}
+        # The untouched completion_events table still exists and is empty.
+        assert "completion_events" in tables
+        assert (
+            _run(database.fetch_one("SELECT COUNT(*) FROM completion_events"))
+            == (0,)
+        )
+    finally:
+        _run(database.close())
+
+
+def test_renamed_instances_foreign_key_follows_the_rename(tmp_path) -> None:
+    database = _open_db(tmp_path / "legacy-fk.db")
+    try:
+        _seed_legacy_database(database)
+        _migrate(database)
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_instances)"))
+        referenced = {row[2] for row in fks}
+        assert "quest_definitions" in referenced
+        assert "task_definitions" not in referenced
+        # The renamed table is live: a new row validates its FKs.
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) "
+                "VALUES ('Ben', '2026-09-14T00:00:00+00:00')"
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_instances (definition_id, child_id, "
+                    "due_date, generated_at) "
+                    "VALUES (999, 2, '2026-09-16', "
+                    "'2026-09-14T00:00:00+00:00')"
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_legacy_rename_is_idempotent_across_restarts(tmp_path) -> None:
+    database = _open_db(tmp_path / "legacy-restart.db")
+    try:
+        _seed_legacy_database(database)
+        _migrate(database)
+        final = _migrate(database)
+        assert final == len(MIGRATIONS)
+        tables = _tables(database)
+        assert "quest_definitions" in tables
+        assert "task_definitions" not in tables
+        assert (
+            _run(database.fetch_one("SELECT COUNT(*) FROM quest_definitions"))
+            == (1,)
+        )
+    finally:
+        _run(database.close())
+
+
+def test_fresh_database_never_holds_legacy_table_names(tmp_path) -> None:
+    database = _open_db(tmp_path / "fresh-legacy.db")
+    try:
+        _migrate(database)
+        tables = _tables(database)
+        assert not any(name.startswith("task_") for name in tables)
+        assert _version(database) == len(MIGRATIONS)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Pre-D-008 database: quest_definitions with child_id stamped version 2
+# ---------------------------------------------------------------------------
+
+_V2_DEFINITIONS_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_definitions (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        title TEXT NOT NULL,
+        description TEXT,
+        icon TEXT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        due_time TEXT,
+        is_active INTEGER NOT NULL DEFAULT 1,
+        created_at TEXT NOT NULL
+    )
+    """,
+]
+
+_V2_INSTANCES_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_instances (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        due_date TEXT NOT NULL,
+        due_time TEXT,
+        generated_at TEXT NOT NULL,
+        UNIQUE (definition_id, due_date)
+    )
+    """,
+]
+
+_V2_EVENTS_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS completion_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        event_type TEXT NOT NULL,
+        actor_source TEXT NOT NULL,
+        actor_user_id TEXT,
+        occurred_at TEXT NOT NULL,
+        was_on_time INTEGER
+    )
+    """,
+]
+
+
+def _seed_v2_multi_assignee_upgrade(database) -> None:
+    """Create a database shaped like a post-rename, pre-D-008 dev file.
+
+    Fresh quest_* tables — but quest_definitions still carries the
+    single child_id column and no assignees table exists — stamped at
+    version 2, with definition and instance rows that must survive.
+    """
+    # Freeze the version-2 shape: definitions carry child_id, the
+    # instance table has no window column, and the assignees/windows
+    # tables do not exist yet — everything else matches the current DDL.
+    _v2_replaced = (
+        "quest_definitions ",
+        "quest_definition_assignees ",
+        "quest_definition_windows ",
+        "quest_instances ",
+        "completion_events ",
+    )
+    for sql in SCHEMA_V1_STATEMENTS:
+        if any(name in sql for name in _v2_replaced):
+            continue
+        _run(database.execute(sql))
+    for sql in _V2_DEFINITIONS_DDL + _V2_INSTANCES_DDL + _V2_EVENTS_DDL:
+        _run(database.execute(sql))
+    _run(database.execute(VERSION_TABLE_DDL))
+    _run(
+        database.execute(
+            f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, 2)"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO children (display_name, created_at) "
+            "VALUES ('Ada', '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO children (display_name, created_at) "
+            "VALUES ('Bo', '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO schedule_rules (rule_type, start_date) "
+            "VALUES ('daily', '2026-09-14')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO quest_definitions (title, child_id, "
+            "schedule_rule_id, created_at) "
+            "VALUES ('Dishes', 2, 1, '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO quest_instances (definition_id, child_id, "
+            "due_date, generated_at) "
+            "VALUES (1, 2, '2026-09-15', '2026-09-14T00:00:00+00:00')"
+        )
+    )
+    # One legacy panel event and one legacy admin event: the panel one
+    # must backfill actor_child_id to its own child; the admin one must
+    # stay NULL.
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, occurred_at, was_on_time) "
+            "VALUES (1, 2, 'completed', 'panel', "
+            "'2026-09-15T10:00:00+00:00', 1)"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, actor_user_id, occurred_at, "
+            "was_on_time) "
+            "VALUES (1, 2, 'uncompleted', 'user', 'parent-1', "
+            "'2026-09-15T11:00:00+00:00', 1)"
+        )
+    )
+
+
+def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
+    database = _open_db(tmp_path / "v2-upgrade.db")
+    try:
+        _seed_v2_multi_assignee_upgrade(database)
+        final = _migrate(database)
+        assert final == len(MIGRATIONS)
+        assert _version(database) == len(MIGRATIONS)
+        # The definition row survived and lost its child_id column.
+        row = _run(
+            database.fetch_one(
+                "SELECT title, schedule_rule_id, is_active "
+                "FROM quest_definitions WHERE id = 1"
+            )
+        )
+        assert row == ("Dishes", 1, 1)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definitions)")
+        )
+        assert "child_id" not in {c[1] for c in columns}
+        # The single assignee moved into the assignees table.
+        assignees = _run(
+            database.fetch_all(
+                "SELECT definition_id, child_id "
+                "FROM quest_definition_assignees ORDER BY child_id"
+            )
+        )
+        assert assignees == [(1, 2)]
+        # The instance survived the rebuild, pinned to the migrated
+        # window, still referencing the definition.
+        instance = _run(
+            database.fetch_one(
+                "SELECT definition_id, child_id, window, due_date "
+                "FROM quest_instances WHERE id = 1"
+            )
+        )
+        assert instance == (1, 2, "morning", "2026-09-15")
+        # The rebuilt instance table carries the widened unique key.
+        index_rows = _run(
+            database.fetch_all(
+                "PRAGMA index_list(quest_instances)"
+            )
+        )
+        unique_columns = []
+        for index_row in index_rows:
+            if index_row[2]:  # unique flag
+                info = _run(
+                    database.fetch_all(
+                        f"PRAGMA index_info('{index_row[1]}')"
+                    )
+                )
+                unique_columns.append([col[2] for col in info])
+        assert [
+            "definition_id", "child_id", "due_date", "window"
+        ] in unique_columns
+        # The completion events survived the actor_child_id rebuild:
+        # the panel row backfilled to its own child, the admin row
+        # stayed NULL, and ids are preserved.
+        events = _run(
+            database.fetch_all(
+                "SELECT id, event_type, actor_source, actor_child_id "
+                "FROM completion_events ORDER BY id"
+            )
+        )
+        assert events == [
+            (1, "completed", "panel", 2),
+            (2, "uncompleted", "user", None),
+        ]
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(completion_events)")
+        )
+        assert "actor_child_id" in {c[1] for c in columns}
+        # No dangling references anywhere.
+        assert _run(database.fetch_all("PRAGMA foreign_key_check")) == []
+        # The rebuilt table is live for the new model: a definition
+        # insert carries no child column at all.
+        _run(
+            database.execute(
+                "INSERT INTO quest_definitions (title, schedule_rule_id, "
+                "created_at) VALUES ('Laundry', 1, "
+                "'2026-09-14T00:00:00+00:00')"
+            )
+        )
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_assignees "
+                "(definition_id, child_id) VALUES (2, 1)"
+            )
+        )
+    finally:
+        _run(database.close())
 
 
 # ---------------------------------------------------------------------------
@@ -162,8 +656,8 @@ def test_second_run_applies_nothing_and_keeps_rows(tmp_path) -> None:
             )
         )
         final = _migrate(database)
-        assert final == 1
+        assert final == len(MIGRATIONS)
-        assert _version(database) == 1
+        assert _version(database) == len(MIGRATIONS)
         assert _counts(database)["children"] == 1
         assert _counts(database)["admin_users"] == 1
     finally:
@@ -228,7 +722,7 @@ def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None:
         with caplog.at_level(logging.DEBUG, logger="custom_components.nestquest.migrations"):
             _migrate(database)
         assert any(
-            "already at version 1" in r.getMessage()
+            f"already at version {len(MIGRATIONS)}" in r.getMessage()
             for r in caplog.records
         )
     finally:
@@ -241,33 +735,47 @@ def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None:
 
 
 def _failing_migrations(prior_statements: list[str]) -> list[list[str]]:
-    """Two-migration list where migration 2 fails mid-application."""
+    """Migration list where the LAST entry fails mid-application.
-    return [
+
-        prior_statements,
+    Sized one entry longer than the real MIGRATIONS and padded with
+    harmless no-op fillers, so a fully-migrated database (already at
+    len(MIGRATIONS)) still has exactly one pending migration — the
+    failing one — to attempt.
+    """
+    steps: list[list[str]] = [prior_statements]
+    steps += [
+        [
+            f"CREATE TABLE IF NOT EXISTS filler_{index} "
+            "(id INTEGER PRIMARY KEY)"
+        ]
+        for index in range(len(MIGRATIONS) - 1)
+    ]
+    steps.append(
         [
             "CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)",
             "CREATE TABLE broken (",
-        ],
+        ]
-    ]
+    )
+    return steps
 
 
 def test_failing_migration_rolls_back_to_prior_version(tmp_path) -> None:
     database = _open_db(tmp_path / "rollback.db")
     try:
-        # Manually simulate a database at version 1: apply migration 1's
+        # Manually simulate a fully-migrated database: run the real
-        # statements directly and stamp version 1 through the runner.
+        # migration list, which stamps the latest version.
         _migrate(database)
         version_row = _run(
             database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
         )
-        assert version_row == (1,)
+        assert version_row == (len(MIGRATIONS),)
 
         with pytest.raises(sqlite3.OperationalError):
             _migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS))
 
         # Version stamp unchanged, statements from the failed migration
         # absent, and the v1 tables untouched.
-        assert _version(database) == 1
+        assert _version(database) == len(MIGRATIONS)
         assert "later_table" not in _tables(database)
         assert _counts(database)["children"] == 0
         assert _counts(database)["completion_events"] == 0
@@ -275,21 +783,24 @@ def test_failing_migration_rolls_back_to_prior_version(tmp_path) -> None:
         _run(database.close())
 
 
-def test_failing_migration_on_fresh_file_leaves_version_0(tmp_path) -> None:
+def test_failing_migration_on_fresh_file_leaves_last_good_version(
+    tmp_path,
+) -> None:
     database = _open_db(tmp_path / "fresh-rollback.db")
     try:
-        # The fresh file still reaches migration 2 (the failing one):
+        # The fresh file reaches the failing migration (the last one):
-        # migration 1 commits, stamping version 1, before migration 2
+        # earlier migrations commit and stamp before it fails and rolls
-        # fails and rolls back.  The run aborts at version 1 — the last
+        # back.  The run aborts at the last fully-applied migration —
-        # fully-applied migration — not at version 0.
+        # not at version 0.
+        broken = _failing_migrations(SCHEMA_V1_STATEMENTS)
         with pytest.raises(sqlite3.OperationalError):
-            _migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS))
+            _migrate(database, broken)
-        assert _version(database) == 1
+        assert _version(database) == len(broken) - 1
-        # Migration 2's tables must not survive the failed attempt.
+        # The failing migration's tables must not survive the attempt.
         assert "later_table" not in _tables(database)
         # Migration 1's tables do: it committed before the failure.
         assert "children" in _tables(database)
-        assert "task_instances" in _tables(database)
+        assert "quest_instances" in _tables(database)
     finally:
         _run(database.close())
 
@@ -302,12 +813,12 @@ def test_failed_migration_can_be_retried_after_fix(tmp_path) -> None:
         with pytest.raises(sqlite3.OperationalError):
             _migrate(database, broken)
         # "Fix" the migration and retry: it now applies cleanly.
-        broken[1] = [
+        broken[-1] = [
             "CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)"
         ]
         final = _migrate(database, broken)
-        assert final == 2
+        assert final == len(broken)
-        assert _version(database) == 2
+        assert _version(database) == len(broken)
         assert "later_table" in _tables(database)
     finally:
         _run(database.close())
@@ -575,7 +1086,7 @@ async def test_setup_entry_migrates_fresh_database(hass, make_entry) -> None:
 
     entry = await _setup_entry(hass, make_entry(), hass.registry)
     database = entry.runtime_data.database
-    assert await read_schema_version(database) == 1
+    assert await read_schema_version(database) == len(MIGRATIONS)
     # The v1 tables exist because the runner applied migration 1.
     names = {
         row[0]
@@ -626,7 +1137,7 @@ async def test_options_reload_reopens_and_rechecks_version(
     database_b = entry_reloaded.runtime_data.database
     assert database_b is not database_a
     try:
-        assert await read_schema_version(database_b) == 1
+        assert await read_schema_version(database_b) == len(MIGRATIONS)
         row = await database_b.fetch_one("SELECT COUNT(*) FROM children")
         assert row == (1,)
     finally:

tests/test_startup_db.py

@@ -18,6 +18,7 @@ from custom_components.nestquest import (
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import (
+    MIGRATIONS,
     VERSION_TABLE,
     apply_migrations,
 )
@@ -114,12 +115,13 @@ def test_missing_file_opens_fresh(tmp_path) -> None:
     assert not db_path.exists()
     database = _run(_async_open_database(_make_hass_mock(), db_path))
     try:
-        # The file was created fresh and migrations stamped version 1.
+        # The file was created fresh and migrations stamped the
+        # latest schema version.
         assert db_path.exists()
         row = _run(
             database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
         )
-        assert row == (1,)
+        assert row == (len(MIGRATIONS),)
     finally:
         _run(database.close())
 
@@ -151,7 +153,7 @@ def test_valid_file_opens_and_migrations_noop(tmp_path) -> None:
         row = _run(
             database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
         )
-        assert row == (1,)
+        assert row == (len(MIGRATIONS),)
     finally:
         _run(database.close())
     # Opening a valid file must not rewrite it (the no-op path).
@@ -401,7 +403,7 @@ async def test_setup_with_missing_file_creates_and_loads(
         row = await database.fetch_one(
             f"SELECT version FROM {VERSION_TABLE}"
         )
-        assert row == (1,)
+        assert row == (len(MIGRATIONS),)
     finally:
         await database.close()
     assert await async_unload_entry(hass, entry) is True