Widen instance key to (definition_id, child_id, due_date, window) (D-008)

Per IMPLEMENTATION-ORDER.md 1.2: quest_instances gains a NOT NULL
window column (CHECK-constrained to the const window spellings) and
the table-level UNIQUE constraint widens to the full fan-out tuple.

- schema.py: instance table body extracted to QUEST_INSTANCES_TABLE_SQL
  shared with the rebuild migration; uniqueness docstring rewritten for
  the widened key.
- migrations.py: migration 5 rebuilds a pre-window quest_instances via
  the canonical procedure; migrated rows are pinned to the documented
  default 'morning' window (pre-window rows were whole-day obligations;
  future dates are regenerated with real windows by materialization),
  preserving ids so completion_events keep pointing at the same
  instances; PRAGMA foreign_key_check verified inside the transaction.
- dao_instances.py: upsert takes required keyword window and is
  idempotent on the widened tuple; the immutable-once-completed check
  and the read-back get() are keyed on the full tuple; window spelling
  validated via dao_rules._validate_window.
- tests: schema shape (window column, widened autoindex columns),
  twice-daily and shared-assignee fan-out cases at both the DDL and DAO
  layers (incl. idempotent re-runs), cross-child same-key allowance,
  window CHECK/NOT NULL constraints, v2 seed frozen to the true
  version-2 shapes and its migration assertions widened.

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_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
@@ -108,7 +111,7 @@ class CompletionEventRecord:
 
 
 _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, "
@@ -121,14 +124,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],
     )
 
 
@@ -145,8 +149,8 @@ def _event_from_row(row: tuple) -> CompletionEventRecord:
     )
 
 
-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 +162,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 +220,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(
+                    "SELECT 1 FROM quest_definition_assignees "
+                    "WHERE definition_id = ? AND child_id = ?",
+                    (definition_id, child_id),
+                )
+                if assignee is None:
                     raise ValueError(
-                        f"task definition {definition_id} is assigned to "
+                        f"quest definition {definition_id} is not "
-                        f"child {definition[0]}, not {child_id}; instance "
+                        f"assigned to child {child_id}; instance must "
-                        "must carry the definition's assignee"
+                        "carry an assigned child"
                     )
-                child = await self._database.fetch_one(
-                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
-                )
-                if child is None:
-                    raise ValueError(f"child {child_id} does not exist")
                 existing = await self._database.fetch_one(
                     "SELECT 1 FROM completion_events "
-                    "WHERE instance_id = (SELECT id FROM task_instances "
+                    "WHERE 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 +316,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 +326,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 +361,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),
                 )
@@ -418,16 +441,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(
@@ -493,7 +516,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,31 @@ 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 and migration 5 rebuilds ``quest_instances`` onto the
+widened (definition_id, child_id, due_date, window) key.  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 +68,28 @@ 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 (
+    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 +113,233 @@ 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")
+
+
 #: 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,
 ]
 
 
@@ -140,7 +403,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 +452,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 +472,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,7 +152,7 @@ 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.
 
@@ -219,18 +243,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
-    CREATE TABLE IF NOT EXISTS task_definitions (
+#: 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,
-        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 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 quest_definition_assignees (
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(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,
+        PRIMARY KEY (definition_id, window)
     )
     """,
 ]
@@ -271,17 +322,25 @@ 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
+#: (definition_id, child_id, due_date, window) — one instance per
+#: assigned child per declared window per firing date).
+QUEST_INSTANCES_TABLE_SQL = """(
     id INTEGER PRIMARY KEY AUTOINCREMENT,
-        definition_id INTEGER NOT NULL REFERENCES task_definitions(id),
+    definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
     child_id INTEGER NOT NULL REFERENCES children(id),
+    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, due_date)
+    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}
     """,
 ]
 
@@ -289,7 +348,7 @@ SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
     """
     CREATE TABLE IF NOT EXISTS completion_events (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
-        instance_id INTEGER NOT NULL REFERENCES task_instances(id),
+        instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
         child_id INTEGER NOT NULL REFERENCES children(id),
         event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
         actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
@@ -315,10 +374,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_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_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,8 +661,7 @@ 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_user_id=None,
@@ -599,8 +678,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 +699,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,
@@ -669,8 +746,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
@@ -694,8 +770,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 +791,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,8 +813,7 @@ 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
         )
@@ -770,8 +843,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,8 +853,7 @@ 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
         )
@@ -801,14 +872,11 @@ 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
         )
@@ -846,8 +914,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
@@ -1027,8 +1094,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 +1124,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 +1149,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,)

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,442 @@ 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, and migration 5 rebuilds
+    quest_instances onto the widened (definition, child, date, window)
+    key.
+    """
     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 len(MIGRATIONS) == 5
+
+
+# ---------------------------------------------------------------------------
+# 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)
+    )
+    """,
+]
+
+
+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 ",
+    )
+    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:
+        _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')"
+        )
+    )
+
+
+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
+        # 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 +602,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 +668,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 +681,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 +729,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 +759,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 +1032,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 +1083,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_schema.py

@@ -15,8 +15,10 @@ from custom_components.nestquest.schema import (
     SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     SCHEMA_V1_SCHEDULE_RULES_DDL,
     SCHEMA_V1_STATEMENTS,
-    SCHEMA_V1_TASK_DEFINITIONS_DDL,
+    SCHEMA_V1_QUEST_DEFINITIONS_DDL,
-    SCHEMA_V1_TASK_INSTANCES_DDL,
+    SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+    SCHEMA_V1_QUEST_INSTANCES_DDL,
     async_apply_ddl,
 )
 
@@ -73,15 +75,17 @@ def _insert_rule(database, **overrides):
     return result.lastrowid
 
 
-def _insert_definition(database, child_id=1, rule_id=1, **overrides):
+def _insert_definition(database, rule_id=1, assignees=(1,), **overrides):
-    """Insert one valid task_definition, returning the definition id.
+    """Insert one valid quest_definition, returning the definition id.
 
+    ``assignees`` rows are written to ``quest_definition_assignees``
+    after the definition insert (the definition itself carries no
+    child).  Pass ``assignees=None`` for an unassigned definition.
     Pass ``_OMIT`` as a value to leave that column out of the INSERT so
     its DDL DEFAULT applies.
     """
     values: dict[str, object] = {
         "title": "Brush teeth",
-        "child_id": child_id,
         "schedule_rule_id": rule_id,
         "created_at": "2026-09-13T00:00:00+00:00",
     }
@@ -91,11 +95,19 @@ def _insert_definition(database, child_id=1, rule_id=1, **overrides):
     placeholders = ", ".join("?" for _ in values)
     result = _run(
         database.execute(
-            f"INSERT INTO task_definitions ({columns}) VALUES ({placeholders})",
+            f"INSERT INTO quest_definitions ({columns}) VALUES ({placeholders})",
             tuple(values.values()),
         )
     )
     assert result.lastrowid is not None
+    for child_id in assignees or ():
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_assignees "
+                "(definition_id, child_id) VALUES (?, ?)",
+                (result.lastrowid, child_id),
+            )
+        )
     return result.lastrowid
 
 
@@ -156,15 +168,17 @@ def _insert_override(database, child_id, **overrides):
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_six_tables() -> None:
+def test_schema_v1_statements_compose_the_eight_tables() -> None:
     assert SCHEMA_V1_STATEMENTS == [
         *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,
     ]
 
@@ -256,17 +270,17 @@ def test_schedule_rules_table_exists_after_applying_ddl(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_definitions_table_exists_after_applying_ddl(tmp_path) -> None:
+def test_quest_definitions_table_exists_after_applying_ddl(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions.db")
     try:
         _apply(database)
         row = _run(
             database.fetch_one(
                 "SELECT name FROM sqlite_master WHERE type = 'table' "
-                "AND name = 'task_definitions'"
+                "AND name = 'quest_definitions'"
             )
         )
-        assert row is not None and row[0] == "task_definitions"
+        assert row is not None and row[0] == "quest_definitions"
     finally:
         _run(database.close())
 
@@ -292,17 +306,16 @@ def test_schedule_rules_columns_types_and_constraints(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_definitions_columns_types_and_constraints(tmp_path) -> None:
+def test_quest_definitions_columns_types_and_constraints(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-columns.db")
     try:
         _apply(database)
-        columns = _run(database.fetch_all("PRAGMA table_info(task_definitions)"))
+        columns = _run(database.fetch_all("PRAGMA table_info(quest_definitions)"))
         assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
             ("id", "INTEGER", 0, None, 1),
             ("title", "TEXT", 1, None, 0),
             ("description", "TEXT", 0, None, 0),
             ("icon", "TEXT", 0, None, 0),
-            ("child_id", "INTEGER", 1, None, 0),
             ("schedule_rule_id", "INTEGER", 1, None, 0),
             ("due_time", "TEXT", 0, None, 0),
             ("is_active", "INTEGER", 1, "1", 0),
@@ -312,20 +325,73 @@ def test_task_definitions_columns_types_and_constraints(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_definitions_foreign_keys_declared(tmp_path) -> None:
+def test_quest_definitions_foreign_keys_declared(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-fks.db")
     try:
         _apply(database)
-        fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_definitions)"))
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_definitions)"))
         declared = {(row[2], row[3], row[4]) for row in fks}
         assert declared == {
-            ("children", "child_id", "id"),
             ("schedule_rules", "schedule_rule_id", "id"),
         }
     finally:
         _run(database.close())
 
 
+def test_quest_definition_assignees_table_exists_after_applying_ddl(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "assignees.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'quest_definition_assignees'"
+            )
+        )
+        assert row is not None and row[0] == "quest_definition_assignees"
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_assignees_columns_types_and_constraints(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "assignees-columns.db")
+    try:
+        _apply(database)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definition_assignees)")
+        )
+        # cid, name, type, notnull, dflt_value, pk — both columns are
+        # the two-part primary key, so pk is 1 and 2.
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("definition_id", "INTEGER", 1, None, 1),
+            ("child_id", "INTEGER", 1, None, 2),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_assignees_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "assignees-fks.db")
+    try:
+        _apply(database)
+        fks = _run(
+            database.fetch_all(
+                "PRAGMA foreign_key_list(quest_definition_assignees)"
+            )
+        )
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("quest_definitions", "definition_id", "id"),
+            ("children", "child_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
 # ---------------------------------------------------------------------------
 # Constraint enforcement
 # ---------------------------------------------------------------------------
@@ -519,7 +585,7 @@ def test_failed_statement_rolls_back_whole_application(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# schedule_rules and task_definitions constraint enforcement
+# schedule_rules and quest_definitions constraint enforcement
 # ---------------------------------------------------------------------------
 
 
@@ -813,7 +879,7 @@ def test_schedule_rules_start_date_not_null_enforced(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_definitions_insert_requires_existing_child_and_rule(
+def test_quest_definitions_insert_requires_existing_child_and_rule(
     tmp_path,
 ) -> None:
     database = _open_db(tmp_path / "definitions-fk-unknown.db")
@@ -826,33 +892,279 @@ def test_task_definitions_insert_requires_existing_child_and_rule(
             )
         )
         rule_id = _insert_rule(database)
-        definition_id = _insert_definition(
+        definition_id = _insert_definition(database, rule_id=rule_id)
-            database, child_id=1, rule_id=rule_id
-        )
         row = _run(
             database.fetch_one(
-                "SELECT title, child_id, schedule_rule_id FROM task_definitions "
+                "SELECT title, schedule_rule_id FROM quest_definitions "
                 "WHERE id = ?",
                 (definition_id,),
             )
         )
-        assert row == ("Brush teeth", 1, rule_id)
+        assert row == ("Brush teeth", rule_id)
+        assignee = _run(
+            database.fetch_one(
+                "SELECT child_id FROM quest_definition_assignees "
+                "WHERE definition_id = ?",
+                (definition_id,),
+            )
+        )
+        assert assignee == (1,)
     finally:
         _run(database.close())
 
 
-def test_task_definitions_unknown_child_id_fails(tmp_path) -> None:
+def test_quest_definition_assignees_unknown_child_fails(tmp_path) -> None:
-    database = _open_db(tmp_path / "definitions-bad-child.db")
+    database = _open_db(tmp_path / "assignees-bad-child.db")
     try:
         _apply(database)
         _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
         with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
-            _insert_definition(database, child_id=999)
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, 999)",
+                    (definition_id,),
+                )
+            )
     finally:
         _run(database.close())
 
 
-def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
+def test_quest_definition_assignees_unknown_definition_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "assignees-bad-definition.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (999, 1)"
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_assignees_duplicate_pair_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "assignees-duplicate.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        definition_id = _insert_definition(database)
+        with pytest.raises(sqlite3.IntegrityError):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, 1)",
+                    (definition_id,),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_table_exists_after_applying_ddl(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "windows.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'quest_definition_windows'"
+            )
+        )
+        assert row is not None and row[0] == "quest_definition_windows"
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_columns_types_and_constraints(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "windows-columns.db")
+    try:
+        _apply(database)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definition_windows)")
+        )
+        # cid, name, type, notnull, dflt_value, pk — the composite
+        # primary key spans definition_id and window.
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("definition_id", "INTEGER", 1, None, 1),
+            ("window", "TEXT", 1, None, 2),
+            ("due_time", "TEXT", 0, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-fks.db")
+    try:
+        _apply(database)
+        fks = _run(
+            database.fetch_all(
+                "PRAGMA foreign_key_list(quest_definition_windows)"
+            )
+        )
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("quest_definitions", "definition_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_valid_rows_round_trip(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-roundtrip.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        for window, due in (
+            ("morning", "09:00"),
+            ("afternoon", None),
+            ("evening", "19:30"),
+        ):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, ?)",
+                    (definition_id, window, due),
+                )
+            )
+        rows = _run(
+            database.fetch_all(
+                "SELECT window, due_time FROM quest_definition_windows "
+                "WHERE definition_id = ? ORDER BY window",
+                (definition_id,),
+            )
+        )
+        assert rows == [
+            ("afternoon", None),
+            ("evening", "19:30"),
+            ("morning", "09:00"),
+        ]
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    "window", ["Midnight", "MORNING", "noon", "", "evenings"]
+)
+def test_quest_definition_windows_window_check_enforced(
+    tmp_path, window
+) -> None:
+    database = _open_db(tmp_path / "windows-check.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, NULL)",
+                    (definition_id, window),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_duplicate_window_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-duplicate.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_windows "
+                "(definition_id, window, due_time) "
+                "VALUES (?, 'morning', NULL)",
+                (definition_id,),
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) "
+                    "VALUES (?, 'morning', '09:00')",
+                    (definition_id,),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_unknown_definition_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-bad-definition.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) "
+                    "VALUES (999, 'morning', NULL)"
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_assignees_multiple_children_allowed(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "assignees-multi.db")
+    try:
+        _apply(database)
+        for name in ("Ada", "Ben", "Cleo"):
+            _run(
+                database.execute(
+                    "INSERT INTO children (display_name, created_at) "
+                    "VALUES (?, ?)",
+                    (name, "2026-09-13T00:00:00+00:00"),
+                )
+            )
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=(1, 2, 3))
+        rows = _run(
+            database.fetch_all(
+                "SELECT child_id FROM quest_definition_assignees "
+                "WHERE definition_id = ? ORDER BY child_id",
+                (definition_id,),
+            )
+        )
+        assert [row[0] for row in rows] == [1, 2, 3]
+    finally:
+        _run(database.close())
+
+
+def test_quest_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-bad-rule.db")
     try:
         _apply(database)
@@ -863,24 +1175,24 @@ def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
             )
         )
         with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
-            _insert_definition(database, child_id=1, rule_id=999)
+            _insert_definition(database, rule_id=999)
     finally:
         _run(database.close())
 
 
-def test_task_definitions_is_active_check_enforced(tmp_path) -> None:
+def test_quest_definitions_is_active_check_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-is-active.db")
     try:
         _apply(database)
         _insert_rule(database)
         with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
-            _insert_definition(database, child_id=1, rule_id=1, is_active=2)
+            _insert_definition(database, rule_id=1, is_active=2)
     finally:
         _run(database.close())
 
 
 @pytest.mark.parametrize("bad_is_active", ["abc", 1.5])
-def test_task_definitions_is_active_non_integer_fails(
+def test_quest_definitions_is_active_non_integer_fails(
     tmp_path, bad_is_active
 ) -> None:
     database = _open_db(tmp_path / "definitions-is-active-typeof.db")
@@ -895,37 +1207,37 @@ def test_task_definitions_is_active_non_integer_fails(
         _insert_rule(database)
         with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
             _insert_definition(
-                database, child_id=1, rule_id=1, is_active=bad_is_active
+                database, rule_id=1, is_active=bad_is_active
             )
     finally:
         _run(database.close())
 
 
-def test_task_definitions_title_not_null_enforced(tmp_path) -> None:
+def test_quest_definitions_title_not_null_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-title.db")
     try:
         _apply(database)
         _insert_rule(database)
         with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
-            _insert_definition(database, child_id=1, rule_id=1, title=None)
+            _insert_definition(database, rule_id=1, title=None)
     finally:
         _run(database.close())
 
 
-def test_task_definitions_created_at_not_null_enforced(tmp_path) -> None:
+def test_quest_definitions_created_at_not_null_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-created-at.db")
     try:
         _apply(database)
         _insert_rule(database)
         with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
             _insert_definition(
-                database, child_id=1, rule_id=1, created_at=None
+                database, rule_id=1, created_at=None
             )
     finally:
         _run(database.close())
 
 
-def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None:
+def test_quest_definitions_defaults_apply_on_insert(tmp_path) -> None:
     database = _open_db(tmp_path / "definitions-defaults.db")
     try:
         _apply(database)
@@ -937,11 +1249,11 @@ def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None:
         )
         _insert_rule(database)
         definition_id = _insert_definition(
-            database, child_id=1, rule_id=1, is_active=_OMIT
+            database, rule_id=1, is_active=_OMIT
         )
         row = _run(
             database.fetch_one(
-                "SELECT is_active, due_time FROM task_definitions WHERE id = ?",
+                "SELECT is_active, due_time FROM quest_definitions WHERE id = ?",
                 (definition_id,),
             )
         )
@@ -966,10 +1278,10 @@ def test_applying_extended_ddl_twice_keeps_new_table_rows(tmp_path) -> None:
             )
         )
         _insert_rule(database)
-        _insert_definition(database, child_id=1, rule_id=1)
+        _insert_definition(database, rule_id=1)
         _apply(database)
         counts = []
-        for table in ("children", "schedule_rules", "task_definitions"):
+        for table in ("children", "schedule_rules", "quest_definitions"):
             row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
             counts.append(row[0])
         assert counts == [1, 1, 1]
@@ -1406,12 +1718,12 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# task_instances and completion_events: helpers
+# quest_instances and completion_events: helpers
 # ---------------------------------------------------------------------------
 
 
-def _insert_instance(database, definition_id=1, **overrides):
+def _insert_instance(database, definition_id=1, window="morning", **overrides):
-    """Insert one valid task_instance, returning the instance id.
+    """Insert one valid quest_instance, returning the instance id.
 
     Pass ``_OMIT`` as a value to leave that column out of the INSERT so
     its DDL DEFAULT applies.
@@ -1419,6 +1731,7 @@ def _insert_instance(database, definition_id=1, **overrides):
     values: dict[str, object] = {
         "definition_id": definition_id,
         "child_id": 1,
+        "window": window,
         "due_date": "2026-09-14",
         "generated_at": "2026-09-14T00:00:00+00:00",
     }
@@ -1428,7 +1741,7 @@ def _insert_instance(database, definition_id=1, **overrides):
     placeholders = ", ".join("?" for _ in values)
     result = _run(
         database.execute(
-            f"INSERT INTO task_instances ({columns}) VALUES ({placeholders})",
+            f"INSERT INTO quest_instances ({columns}) VALUES ({placeholders})",
             tuple(values.values()),
         )
     )
@@ -1466,28 +1779,28 @@ def _insert_event(database, instance_id=1, **overrides):
 
 
 def _setup_definition(database) -> None:
-    """Create one child, one schedule rule and one task definition."""
+    """Create one child, one schedule rule and one quest definition."""
     _child(database)
     _insert_rule(database)
-    _insert_definition(database, child_id=1, rule_id=1)
+    _insert_definition(database, rule_id=1)
 
 
 # ---------------------------------------------------------------------------
-# task_instances and completion_events: DDL shape
+# quest_instances and completion_events: DDL shape
 # ---------------------------------------------------------------------------
 
 
-def test_task_instances_table_exists_after_applying_ddl(tmp_path) -> None:
+def test_quest_instances_table_exists_after_applying_ddl(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances.db")
     try:
         _apply(database)
         row = _run(
             database.fetch_one(
                 "SELECT name FROM sqlite_master WHERE type = 'table' "
-                "AND name = 'task_instances'"
+                "AND name = 'quest_instances'"
             )
         )
-        assert row is not None and row[0] == "task_instances"
+        assert row is not None and row[0] == "quest_instances"
     finally:
         _run(database.close())
 
@@ -1507,16 +1820,17 @@ def test_completion_events_table_exists_after_applying_ddl(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_columns_types_and_constraints(tmp_path) -> None:
+def test_quest_instances_columns_types_and_constraints(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-columns.db")
     try:
         _apply(database)
-        columns = _run(database.fetch_all("PRAGMA table_info(task_instances)"))
+        columns = _run(database.fetch_all("PRAGMA table_info(quest_instances)"))
         # cid, name, type, notnull, dflt_value, pk
         assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
             ("id", "INTEGER", 0, None, 1),
             ("definition_id", "INTEGER", 1, None, 0),
             ("child_id", "INTEGER", 1, None, 0),
+            ("window", "TEXT", 1, None, 0),
             ("due_date", "TEXT", 1, None, 0),
             ("due_time", "TEXT", 0, None, 0),
             ("generated_at", "TEXT", 1, None, 0),
@@ -1544,14 +1858,14 @@ def test_completion_events_columns_types_and_constraints(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_foreign_keys_declared(tmp_path) -> None:
+def test_quest_instances_foreign_keys_declared(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-fks.db")
     try:
         _apply(database)
-        fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_instances)"))
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_instances)"))
         declared = {(row[2], row[3], row[4]) for row in fks}
         assert declared == {
-            ("task_definitions", "definition_id", "id"),
+            ("quest_definitions", "definition_id", "id"),
             ("children", "child_id", "id"),
         }
     finally:
@@ -1565,14 +1879,14 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
         fks = _run(database.fetch_all("PRAGMA foreign_key_list(completion_events)"))
         declared = {(row[2], row[3], row[4]) for row in fks}
         assert declared == {
-            ("task_instances", "instance_id", "id"),
+            ("quest_instances", "instance_id", "id"),
             ("children", "child_id", "id"),
         }
     finally:
         _run(database.close())
 
 
-def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
+def test_quest_instances_unique_index_on_widened_key(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-unique.db")
     try:
         _apply(database)
@@ -1580,26 +1894,31 @@ def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
         rows = _run(
             database.fetch_all(
                 "SELECT name, sql FROM sqlite_master WHERE type = 'index' "
-                "AND tbl_name = 'task_instances'"
+                "AND tbl_name = 'quest_instances'"
             )
         )
         implicit = [(name, sql) for name, sql in rows if sql is None]
         assert len(implicit) == 1, (
-            "the (definition_id, due_date) UNIQUE constraint must be backed "
+            "the (definition_id, child_id, due_date, window) UNIQUE "
-            "by exactly one implicit unique index"
+            "constraint must be backed by exactly one implicit unique index"
         )
         columns = _run(
             database.fetch_all(
                 f"PRAGMA index_info('{implicit[0][0]}')"
             )
         )
-        assert [row[2] for row in columns] == ["definition_id", "due_date"]
+        assert [row[2] for row in columns] == [
+            "definition_id",
+            "child_id",
+            "due_date",
+            "window",
+        ]
         # The implicit index is genuinely unique: UNIQUE(...) is the only
         # constraint SQLite backs with a sql-less autoindex on this table.
         row = _run(
             database.fetch_one(
                 "SELECT COUNT(DISTINCT name) FROM pragma_index_list"
-                "('task_instances') WHERE \"unique\" = 1"
+                "('quest_instances') WHERE \"unique\" = 1"
             )
         )
         assert row == (1,)
@@ -1608,11 +1927,11 @@ def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# task_instances: constraint enforcement
+# quest_instances: constraint enforcement
 # ---------------------------------------------------------------------------
 
 
-def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
+def test_quest_instances_valid_insert_round_trips(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-valid.db")
     try:
         _apply(database)
@@ -1620,17 +1939,21 @@ def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
         instance_id = _insert_instance(database, definition_id=1, child_id=1)
         row = _run(
             database.fetch_one(
-                "SELECT definition_id, child_id, due_date, due_time, generated_at "
+                "SELECT definition_id, child_id, window, due_date, "
-                "FROM task_instances WHERE id = ?",
+                "due_time, generated_at "
+                "FROM quest_instances WHERE id = ?",
                 (instance_id,),
             )
         )
-        assert row == (1, 1, "2026-09-14", None, "2026-09-14T00:00:00+00:00")
+        assert row == (
+            1, 1, "morning", "2026-09-14", None,
+            "2026-09-14T00:00:00+00:00",
+        )
     finally:
         _run(database.close())
 
 
-def test_task_instances_due_time_round_trips(tmp_path) -> None:
+def test_quest_instances_due_time_round_trips(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-due-time.db")
     try:
         _apply(database)
@@ -1640,7 +1963,7 @@ def test_task_instances_due_time_round_trips(tmp_path) -> None:
         )
         row = _run(
             database.fetch_one(
-                "SELECT due_time FROM task_instances WHERE id = ?",
+                "SELECT due_time FROM quest_instances WHERE id = ?",
                 (instance_id,),
             )
         )
@@ -1649,7 +1972,7 @@ def test_task_instances_due_time_round_trips(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_duplicate_definition_date_fails(tmp_path) -> None:
+def test_quest_instances_duplicate_tuple_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-duplicate.db")
     try:
         _apply(database)
@@ -1661,7 +1984,79 @@ def test_task_instances_duplicate_definition_date_fails(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_same_date_different_definitions_allowed(
+def test_quest_instances_duplicate_tuple_different_child_allowed(
+    tmp_path,
+) -> None:
+    """D-008: same definition, date and window for two different
+    assignees of the same definition are two distinct instances."""
+    database = _open_db(tmp_path / "task-instances-shared.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Bo", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_assignees "
+                "(definition_id, child_id) VALUES (1, 2)"
+            )
+        )
+        _insert_instance(database, definition_id=1, child_id=1)
+        _insert_instance(database, definition_id=1, child_id=2)
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
+        assert row == (2,)
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_duplicate_tuple_different_window_allowed(
+    tmp_path,
+) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows."""
+    database = _open_db(tmp_path / "task-instances-twice-daily.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _insert_instance(database, definition_id=1, window="morning")
+        _insert_instance(database, definition_id=1, window="evening")
+        rows = _run(
+            database.fetch_all(
+                "SELECT window FROM quest_instances ORDER BY window"
+            )
+        )
+        assert [row[0] for row in rows] == ["evening", "morning"]
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_unknown_window_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-bad-window.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_instance(database, definition_id=1, window="noon")
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_window_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-window-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_instance(database, definition_id=1, window=_OMIT)
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_same_date_different_definitions_allowed(
     tmp_path,
 ) -> None:
     database = _open_db(tmp_path / "task-instances-multi-def.db")
@@ -1675,29 +2070,29 @@ def test_task_instances_same_date_different_definitions_allowed(
             )
         )
         rule_id = _insert_rule(database)
-        _insert_definition(database, child_id=2, rule_id=rule_id)
+        _insert_definition(database, rule_id=rule_id, assignees=(2,))
         _insert_instance(database, definition_id=1, due_date="2026-09-14")
         _insert_instance(database, definition_id=2, due_date="2026-09-14")
-        row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
         assert row == (2,)
     finally:
         _run(database.close())
 
 
-def test_task_instances_same_definition_different_dates_allowed(tmp_path) -> None:
+def test_quest_instances_same_definition_different_dates_allowed(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-multi-date.db")
     try:
         _apply(database)
         _setup_definition(database)
         _insert_instance(database, definition_id=1, due_date="2026-09-14")
         _insert_instance(database, definition_id=1, due_date="2026-09-15")
-        row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
         assert row == (2,)
     finally:
         _run(database.close())
 
 
-def test_task_instances_unknown_definition_fails(tmp_path) -> None:
+def test_quest_instances_unknown_definition_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-bad-def.db")
     try:
         _apply(database)
@@ -1708,7 +2103,7 @@ def test_task_instances_unknown_definition_fails(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_unknown_child_fails(tmp_path) -> None:
+def test_quest_instances_unknown_child_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-bad-child.db")
     try:
         _apply(database)
@@ -1719,7 +2114,7 @@ def test_task_instances_unknown_child_fails(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_due_date_not_null_enforced(tmp_path) -> None:
+def test_quest_instances_due_date_not_null_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-due-null.db")
     try:
         _apply(database)
@@ -1730,7 +2125,7 @@ def test_task_instances_due_date_not_null_enforced(tmp_path) -> None:
         _run(database.close())
 
 
-def test_task_instances_generated_at_not_null_enforced(tmp_path) -> None:
+def test_quest_instances_generated_at_not_null_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-generated-null.db")
     try:
         _apply(database)
@@ -2026,8 +2421,8 @@ def test_applying_full_ddl_twice_keeps_rows(tmp_path) -> None:
         for table in (
             "children",
             "schedule_rules",
-            "task_definitions",
+            "quest_definitions",
-            "task_instances",
+            "quest_instances",
             "completion_events",
         ):
             row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))

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