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

@@ -15,11 +15,13 @@ event.  A package-wide guard test (bottom of the test module) keeps
 mutation SQL for this table out of every other module.
 
 Upsert semantics: ``QuestInstancesDao.upsert`` is idempotent on
-(definition_id, due_date) — re-running materialization for a date
-whose instance already exists updates nothing that matters
-(child_id/due_time are refreshed only in the degenerate re-generation
-case where they changed) and never duplicates a row.  The schema's
-UNIQUE(definition_id, due_date) is the last line of defense.
+(definition_id, child_id, due_date, window) — re-running
+materialization for a tuple whose instance already exists refreshes
+only the generation-time snapshot columns and never duplicates a row.
+The schema's UNIQUE(definition_id, child_id, due_date, window) is the
+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
@@ -88,6 +90,7 @@ class QuestInstanceRecord:
     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, "
@@ -126,9 +129,10 @@ def _instance_from_row(row: tuple) -> QuestInstanceRecord:
         id=row[0],
         definition_id=row[1],
         child_id=row[2],
-        due_date=row[3],
-        due_time=row[4],
-        generated_at=row[5],
+        window=row[3],
+        due_date=row[4],
+        due_time=row[5],
+        generated_at=row[6],
     )
 
 
@@ -158,31 +162,40 @@ class QuestInstancesDao:
         due_date: str,
         generated_at: str,
         *,
+        window: str,
         due_time: str | None = None,
     ) -> 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
-        yields ONE row: the ON CONFLICT path refreshes the
-        generation-time snapshot columns (child_id, due_time) and
-        regenerates the stamp, but can never create a duplicate — the
-        schema UNIQUE(definition_id, due_date) backs this up.
+        The instance key is (definition_id, child_id, due_date, window)
+        (D-008): running materialization twice for the same tuple
+        yields ONE row — the ON CONFLICT path refreshes the
+        generation-time snapshot columns (due_time) and regenerates the
+        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,64 +220,74 @@ class QuestInstancesDao:
                         f"{today!r}"
                     )
                 definition = await self._database.fetch_one(
-                    "SELECT child_id FROM quest_definitions WHERE id = ?",
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
                     (definition_id,),
                 )
                 if definition is None:
                     raise ValueError(
                         f"quest definition {definition_id} does not exist"
                     )
-                if definition[0] != child_id:
-                    raise ValueError(
-                        f"quest definition {definition_id} is assigned to "
-                        f"child {definition[0]}, not {child_id}; instance "
-                        "must carry the definition's assignee"
-                    )
-                child = await self._database.fetch_one(
-                    "SELECT 1 FROM children WHERE id = ?", (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 child is None:
-                    raise ValueError(f"child {child_id} does not exist")
+                if assignee is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} is not "
+                        f"assigned to child {child_id}; instance must "
+                        "carry an assigned child"
+                    )
                 existing = await self._database.fetch_one(
                     "SELECT 1 FROM completion_events "
                     "WHERE instance_id = (SELECT id FROM quest_instances "
-                    "WHERE definition_id = ? AND due_date = ?)",
-                    (definition_id, due_date),
+                    "WHERE definition_id = ? AND child_id = ? "
+                    "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 quest_instances (definition_id, child_id, "
-                    "due_date, due_time, generated_at) "
-                    "VALUES (?, ?, ?, ?, ?) "
-                    "ON CONFLICT (definition_id, due_date) DO UPDATE SET "
-                    "child_id = excluded.child_id, "
+                    "window, due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
+                    "window) DO UPDATE SET "
                     "due_time = excluded.due_time, "
                     "generated_at = excluded.generated_at",
                     (
                         definition_id,
                         child_id,
+                        window,
                         due_date,
                         due_time,
                         generated_at,
                     ),
                 )
-                instance = await self.get(definition_id, 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,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        window: str,
     ) -> QuestInstanceRecord | None:
-        """Return the instance for (definition_id, due_date), or None."""
+        """Return the instance for the widened key tuple, or None."""
         row = await self._database.fetch_one(
             f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
-            "WHERE definition_id = ? AND due_date = ?",
-            (definition_id, due_date),
+            "WHERE definition_id = ? AND child_id = ? AND due_date = ? "
+            "AND window = ?",
+            (definition_id, child_id, due_date, window),
         )
         return _instance_from_row(row) if row is not None else None
 

custom_components/nestquest/dao_rules.py

@@ -1,8 +1,10 @@
-"""Typed DAO layer for the schedule_rules and quest_definitions tables.
+"""Typed DAO layer for the schedule_rules, quest_definitions and
+quest_definition_assignees tables.
 
-All SQL for ``schedule_rules`` and ``quest_definitions`` lives in this
-module per the feature guardrails: callers get typed dataclasses back
-and never see raw rows or SQL.  Every method is async and runs through
+All SQL for ``schedule_rules``, ``quest_definitions`` and
+``quest_definition_assignees`` lives in this module per the feature
+guardrails: callers get typed dataclasses back and never see raw rows
+or SQL.  Every method is async and runs through
 :class:`~.db.NestQuestDatabase`, so each statement executes on the HA
 executor and the event loop never blocks.
 
@@ -13,25 +15,30 @@ Guardrail mapping:
   not cascaded.
 - ``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
-  definition row and nothing else; reassignment never rewrites
+- Assignment is multi-assignee (D-008): :meth:`add_assignee` and
+  :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:`QuestDefinitionsDao.create` (active child, rule existence);
-  no permission checks here, that is the Feature 09 gate's job.
+  :meth:`QuestDefinitionsDao.create` (rule existence) and
+  :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
-check and DELETE inside one transaction under the connection-scoped
-lock shared with :mod:`.dao_children`, so a definition referencing the
-rule cannot appear between check and delete on this connection.
+Concurrency: ``delete_rule_if_unreferenced`` and the assignee
+mutations perform their checks and writes inside one transaction under
+the connection-scoped lock shared with :mod:`.dao_children`, so a
+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``."""
@@ -82,27 +121,36 @@ class ScheduleRuleRecord:
 
 @dataclass(frozen=True)
 class QuestDefinitionRecord:
-    """One row of ``quest_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:
@@ -125,11 +173,18 @@ def _definition_from_row(row: tuple) -> QuestDefinitionRecord:
         title=row[1],
         description=row[2],
         icon=row[3],
-        child_id=row[4],
-        schedule_rule_id=row[5],
-        due_time=row[6],
-        is_active=bool(row[7]),
-        created_at=row[8],
+        schedule_rule_id=row[4],
+        due_time=row[5],
+        is_active=bool(row[6]),
+        created_at=row[7],
+    )
+
+
+def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
+    return QuestDefinitionWindowRecord(
+        definition_id=row[0],
+        window=row[1],
+        due_time=row[2],
     )
 
 
@@ -272,11 +327,14 @@ class ScheduleRulesDao:
 
 
 class QuestDefinitionsDao:
-    """Typed async access to the ``quest_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 QuestDefinitionsDao:
     async def create(
         self,
         title: str,
-        child_id: int,
         schedule_rule_id: int,
         created_at: str,
         *,
@@ -293,53 +350,55 @@ class QuestDefinitionsDao:
         icon: str | None = None,
         due_time: str | None = None,
         is_active: bool = True,
+        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
-        transaction as the insert: the child must exist and be active,
-        and the rule must exist, AT INSERT TIME — a child deactivated
-        or a rule deleted concurrently cannot slip between the checks
-        and the INSERT, which would otherwise create an invalid
-        assignment or surface as a raw FK IntegrityError instead of
-        the documented ValueError.  An INACTIVE child would pass the
-        schema foreign keys, which is why the activeness check lives
-        here.
+        Validates the schedule rule as part of the same serialized
+        transaction as the insert: the rule must exist AT INSERT TIME —
+        a rule deleted concurrently cannot slip between the check and
+        the INSERT.  ``assignee_child_ids`` (each child must exist and
+        be active) is applied in the same transaction, so a definition
+        can never be observed with a stale or missing assignee set; an
+        empty list or None creates an unassigned definition, which
+        callers may fill via :meth:`add_assignee`.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
-                await self._validate_assignable(
-                    child_id, schedule_rule_id
-                )
+                await self._validate_rule_exists(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 quest_definitions (title, description, "
-                    "icon, child_id, schedule_rule_id, due_time, "
-                    "is_active, created_at) "
-                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                    "icon, schedule_rule_id, due_time, is_active, "
+                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                     (
                         title,
                         description,
                         icon,
-                        child_id,
                         schedule_rule_id,
                         due_time,
                         int(is_active),
                         created_at,
                     ),
                 )
-                definition = await self.get(result.lastrowid)
+                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(
-        self, child_id: int, schedule_rule_id: int
-    ) -> None:
-        """Raise ValueError unless child (active) and rule both exist.
+    async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
+        """Raise ValueError unless the rule exists.
 
         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,),
@@ -359,10 +418,17 @@ class QuestDefinitionsDao:
         return _definition_from_row(row) if row is not None else None
 
     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 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]
@@ -391,8 +457,9 @@ class QuestDefinitionsDao:
         SQL NULL, so callers can remove optional metadata
         (description, icon, due_time).  Assignment and activation are
         deliberately NOT settable here: they have their own explicit
-        operations (:meth:`set_assignee`, :meth:`set_active`) so they
-        stay separately loggable and permission-gateable.
+        operations (:meth:`add_assignee`, :meth:`remove_assignee`,
+        :meth:`set_active`) so they stay separately loggable and
+        permission-gateable.
         """
         assignments: list[str] = []
         parameters: list[object] = []
@@ -427,32 +494,161 @@ class QuestDefinitionsDao:
         )
         return result.rowcount
 
-    async def set_assignee(
+    async def add_assignee(
         self, definition_id: int, child_id: int
-    ) -> int:
-        """Reassign the definition to ``child_id``; rows updated (0 if absent).
+    ) -> None:
+        """Assign ``child_id`` to the definition (idempotent).
 
-        Validates the new assignee inside the same serialized
-        transaction as the UPDATE (existing and active AT ASSIGNMENT
-        TIME), so a child deactivated concurrently cannot become the
-        assignee.  Changes future instances only: already-generated
-        instances and completion history are never rewritten here.
+        Validates the assignee inside the same serialized transaction
+        as the INSERT (existing and active AT ASSIGNMENT TIME), so a
+        child deactivated concurrently cannot become an assignee.
+        Assigning an already-assigned child is a no-op, mirroring the
+        composite primary key's storage-level guarantee.  Changes
+        future instances only: already-generated instances and
+        completion history are never rewritten here.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 await self._validate_assignable_child(child_id)
-                result = await self._database.execute(
-                    "UPDATE quest_definitions SET child_id = ? "
-                    "WHERE id = ?",
-                    (child_id, definition_id),
+                definition = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definitions WHERE 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

@@ -20,15 +20,24 @@ 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.  The rename step runs
-inside the same transaction as its version stamp, so a crash mid-rename
-rolls both back together.
+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 the open database.  Callables exist for steps
-whose SQL depends on the database's current state (the legacy rename
-only fires when a legacy table is actually present) and run under the
-same per-migration transaction as statement lists.
+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
@@ -63,13 +72,23 @@ 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 run under the entry's migration transaction.
-MigrationStep = Union[Sequence[str], Callable[[NestQuestDatabase], Awaitable[None]]]
+#: 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
@@ -94,6 +113,10 @@ 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.
@@ -103,7 +126,42 @@ LEGACY_TASK_TABLE_RENAMES: tuple[tuple[str, str], ...] = (
 )
 
 
-async def _rename_legacy_task_tables(database: NestQuestDatabase) -> None:
+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
@@ -112,22 +170,160 @@ async def _rename_legacy_task_tables(database: NestQuestDatabase) -> None:
     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 inside the migration transaction,
-    so the renames and the version stamp commit or roll back together.
+    internal autoindex names.  Runs in its own transaction together
+    with the version stamp, so a crash rolls both back.
     """
-    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
+    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
@@ -136,6 +332,14 @@ async def _rename_legacy_task_tables(database: NestQuestDatabase) -> None:
 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,
 ]
 
 
@@ -248,17 +452,18 @@ async def apply_migrations(
 
         for target in range(current + 1, latest + 1):
             applied = False
-            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.
+            step = migrations[target - 1]
+            if callable(step):
+                # Callable migrations own their transaction: some steps
+                # (table rebuilds) must run connection pragmas such as
+                # ``foreign_keys = OFF`` BEFORE opening one, which the
+                # runner's uniform transaction cannot express, and they
+                # stamp inside their own transaction so a failure rolls
+                # the statements and the version stamp back together.
+                # The version is re-read here under the migration lock:
+                # runners on one connection are serialized by it, so
+                # 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(
@@ -267,41 +472,39 @@ async def apply_migrations(
                         f"{latest}); downgrades are not supported"
                     )
                 if in_tx_version < target:
-                    step = migrations[target - 1]
-                    if callable(step):
-                        await step(database)
-                        applied_statements = 0
-                    else:
+                    await step(database, 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)
-                    # 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
             if applied:
-                if applied_statements == 0:
-                    LOGGER.info(
-                        "NestQuest schema migrated to version %d",
-                        target,
-                    )
-                else:
-                    LOGGER.info(
-                        "NestQuest schema migrated to version %d "
-                        "(%d statement%s)",
-                        target,
-                        applied_statements,
-                        "s" if applied_statements != 1 else "",
-                    )
+                LOGGER.info(
+                    "NestQuest schema migrated to version %d",
+                    target,
+                )
 
         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.
 
+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
-("why this instance exists"): reassignment and rule edits change future
-instances only and never rewrite these rows.  ``due_time`` snapshots
-the definition's optional due time at generation for the same reason —
-the instance keeps its 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).
+``child_id``, ``due_date`` and ``window`` snapshot the
+materialization-time facts ("why this instance exists"): assignment
+edits and rule edits change future instances only and never rewrite
+these rows.  ``window`` records which declared day window produced the
+instance (D-008), CHECK-constrained to the same three spellings
+``const`` owns.  ``due_time`` snapshots the window's optional due time
+at generation for the same reason — the instance keeps its
+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
-table-level ``UNIQUE`` constraint rather than a separate ``CREATE
-UNIQUE INDEX`` statement.  SQLite materializes it as an implicit unique
-index (``sqlite_autoindex_quest_instances_1``) — the required "unique
-index prevents duplicate instances" — and the constraint form cannot be
-silently dropped the way a standalone index can, so the duplicate door
-stays shut.  Idempotent materialization keys on exactly this pair.  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.
+Uniqueness: the (definition_id, child_id, due_date, window) tuple is
+declared as a table-level ``UNIQUE`` constraint rather than a separate
+``CREATE UNIQUE INDEX`` statement.  SQLite materializes it as an
+implicit unique index (``sqlite_autoindex_quest_instances_1``) — the
+required "unique index prevents duplicate instances" — and the
+constraint form cannot be silently dropped the way a standalone index
+can, so the duplicate door stays shut.  Idempotent materialization keys
+on exactly this tuple: the twice-daily case (one definition, two
+windows, one child, one date) yields two rows and the shared case (one
+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
@@ -219,18 +243,45 @@ SCHEMA_V1_SCHEDULE_RULES_DDL: list[str] = [
     """,
 ]
 
+#: Column body of ``quest_definitions``, shared between the v1 DDL
+#: below and the multi-assignee rebuild migration, which must recreate
+#: the table in exactly this shape (D-008: one definition, many
+#: assignees via ``quest_definition_assignees`` — no ``child_id``
+#: column on the definition itself).
+QUEST_DEFINITIONS_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    title TEXT NOT NULL,
+    description TEXT,
+    icon TEXT,
+    schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+    due_time TEXT,
+    is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+    created_at TEXT NOT NULL
+)"""
+
 SCHEMA_V1_QUEST_DEFINITIONS_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_definitions {QUEST_DEFINITIONS_TABLE_SQL}
+    """,
+]
+
+SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL: list[str] = [
     """
-    CREATE TABLE IF NOT EXISTS quest_definitions (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
-        title TEXT NOT NULL,
-        description TEXT,
-        icon TEXT,
+    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),
-        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        PRIMARY KEY (definition_id, child_id)
+    )
+    """,
+]
+
+SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_definition_windows (
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
         due_time TEXT,
-        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
-        created_at TEXT NOT NULL
+        PRIMARY KEY (definition_id, window)
     )
     """,
 ]
@@ -271,17 +322,25 @@ SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
     """,
 ]
 
+#: Column body of ``quest_instances``, shared between the v1 DDL
+#: below and the instance-key rebuild migration, which must recreate
+#: 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 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, child_id, due_date, window)
+)"""
+
 SCHEMA_V1_QUEST_INSTANCES_DDL: list[str] = [
-    """
-    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)
-    )
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_instances {QUEST_INSTANCES_TABLE_SQL}
     """,
 ]
 
@@ -316,6 +375,8 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_ADMIN_USERS_DDL,
     *SCHEMA_V1_SCHEDULE_RULES_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_QUEST_INSTANCES_DDL,

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

@@ -82,7 +82,7 @@ async def _prepare(path) -> tuple:
     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,
@@ -119,12 +119,10 @@ 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)
         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 quest_instances"
@@ -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())
-        b = await instances.upsert(definition.id, child.id, D2, _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(), 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
-        # definition's assignee).  To reach the child-existence check
-        # at all, the definition would have to claim child 999 — an
-        # FK-impossible state — so the child check only guards direct
-        # callers with a matching fake assignee; assert the guard
-        # order instead of forcing the FK-impossible state.
+        # is the one that matters (an instance must carry an assigned
+        # child).  Under D-008 the assignee check is a membership test
+        # against quest_definition_assignees, so child 999 simply is
+        # not an assignee — the error names the pair.
         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"):
-            await instances.upsert(other_definition.id, 999, D1, _now_stamp())
+        with pytest.raises(ValueError, match="not assigned to child 999"):
+            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
@@ -329,7 +414,10 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
             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"
-            )
-        fetched = await instances.get(definition.id, D1)
+                definition.id, child.id, D1, _now_stamp(), due_time="09:00", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         assert fetched.due_time == "08:00", (
             "failed regeneration must not touch the completed instance"
         )
@@ -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"
-        )
-        fetched = await instances.get(definition.id, D1)
+            definition.id, child.id, D1, _now_stamp(), due_time="17:30", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         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, D2, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
+        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
         )
@@ -460,18 +544,17 @@ def test_list_by_date_range_rejects_inverted_range(tmp_path) -> None:
 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()
-        )
-        await instances.upsert(definition.id, child.id, D6, _now_stamp())
+            definition.id, child.id, D5, _now_stamp(), window="morning")
+        await instances.upsert(definition.id, child.id, D6, _now_stamp(), window="morning")
         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, D5) is None
-        assert await instances.get(definition.id, D6) is None
+        assert await instances.get(definition.id, child.id, D1, "morning") is not None
+        assert await instances.get(definition.id, child.id, D5, "morning") 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,
-            _now_stamp(),
+            "Make bed", (await rules.create("daily", D1)).id,
+            _now_stamp(), assignee_child_ids=[child.id],
         )
-        await instances.upsert(definition.id, child.id, D20, _now_stamp())
-        await instances.upsert(other.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(), 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(
@@ -1090,7 +1155,7 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
         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

@@ -71,6 +71,7 @@ ALL_TABLES = (
     "admin_users",
     "schedule_rules",
     "quest_definitions",
+    "quest_definition_assignees",
     "presence_schedules",
     "presence_overrides",
     "quest_instances",
@@ -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",
         )
 
 
@@ -206,6 +211,9 @@ def test_matrix_quest_definitions_crud_and_constraints(tmp_path) -> None:
         # 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_quest_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())
-        # constraint: inactive child rejected
+            await w.definitions.create(
+                "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
@@ -274,12 +292,15 @@ def test_matrix_presence_overrides_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"
@@ -299,15 +320,15 @@ def test_matrix_quest_instances_constraints(tmp_path) -> None:
         # proving the schema backs the DAO's idempotency.
         await w.database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, due_time, generated_at) "
-            "VALUES (?, ?, ?, NULL, ?)",
+            "window, due_date, due_time, generated_at) "
+            "VALUES (?, ?, 'morning', ?, NULL, ?)",
             (w.definition.id, w.child.id, D20, _stamp()),
         )
         with pytest.raises(sqlite3.IntegrityError):
             await w.database.execute(
                 "INSERT INTO quest_instances (definition_id, child_id, "
-                "due_date, due_time, generated_at) "
-                "VALUES (?, ?, ?, NULL, ?)",
+                "window, due_date, due_time, generated_at) "
+                "VALUES (?, ?, 'morning', ?, NULL, ?)",
                 (w.definition.id, w.child.id, D20, _stamp()),
             )
         # delete future uncompleted removes D20's open instance
@@ -323,10 +344,11 @@ def test_matrix_quest_instances_constraints(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(

tests/test_dao_rules.py

@@ -11,6 +11,7 @@ from custom_components.nestquest.dao_rules import (
     ScheduleRuleRecord,
     ScheduleRulesDao,
     QuestDefinitionRecord,
+    QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
 )
 from custom_components.nestquest.db import NestQuestDatabase
@@ -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)
@@ -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.QuestDefinitionsDao._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.QuestDefinitionsDao._validate_assignable = (
+                dao_rules.QuestDefinitionsDao._validate_rule_exists = (
                     _pausing_validate
                 )
                 try:
                     create_task = asyncio.ensure_future(
-                        definitions.create(
-                            "Brush teeth", child.id, rule_a.id, NOW
-                        )
+                        definitions.create("Brush teeth", rule_a.id, NOW)
                     )
                     await started.wait()
                     delete_task = asyncio.ensure_future(
@@ -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())
@@ -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, 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,7 +497,7 @@ 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)
@@ -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("B", child.id, rule.id, NOW)
-        await definitions.create("C", other_child.id, rule.id, NOW)
+        await definitions.create("A", rule.id, NOW, assignee_child_ids=[child.id])
+        await definitions.create("B", rule.id, NOW, assignee_child_ids=[child.id])
+        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)
-        await definitions.create("B", child.id, rule.id, NOW)
+        first = await definitions.create("A", 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"]
@@ -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
@@ -683,76 +696,242 @@ 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)
-        assert await definitions.set_assignee(definition.id, other.id) == 1
-        fetched = await definitions.get(definition.id)
-        assert fetched.child_id == other.id
-        return fetched
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
+        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])
+        # 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)
-        fetched = await definitions.get(definition.id)
-        assert fetched.child_id == child.id, (
-            "failed reassignment must leave the current assignee intact"
+            await definitions.add_assignee(definition.id, other.id)
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [child.id], (
+            "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)
-        return await definitions.set_assignee(999, other.id)
+        rule = await rules.create("daily", "2026-09-14")
+        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
-    already exist keep their generation-time child_id (Feature 06
+    """D-008: one definition covering several children appears for
+    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 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 quest_instances WHERE definition_id = ?",
             (definition.id,),
@@ -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|quest_definitions)\b",
+        r"(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     offenders: list[str] = []
@@ -813,7 +992,7 @@ 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/quest_definitions leaked into: "
+        f"SQL touching schedule_rules/quest_definitions/assignees leaked: "
         f"{offenders}"
     )
 
@@ -853,7 +1032,7 @@ 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|quest_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, (
@@ -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|quest_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_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):
@@ -63,6 +66,8 @@ def _counts(database) -> dict[str, int]:
         "admin_users",
         "schedule_rules",
         "quest_definitions",
+        "quest_definition_assignees",
+        "quest_definition_windows",
         "presence_schedules",
         "presence_overrides",
         "quest_instances",
@@ -99,6 +104,8 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "admin_users",
             "schedule_rules",
             "quest_definitions",
+            "quest_definition_assignees",
+            "quest_definition_windows",
             "presence_schedules",
             "presence_overrides",
             "quest_instances",
@@ -136,15 +143,22 @@ def test_empty_version_table_reads_as_version_0(tmp_path) -> None:
 
 
 def test_migration_list_shape() -> None:
-    """Migration 1 is the v1 DDL; migration 2 is the legacy rename step.
+    """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.
+    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 callable(MIGRATIONS[1])
-    assert len(MIGRATIONS) == 2
+    assert callable(MIGRATIONS[2])
+    assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
+    assert callable(MIGRATIONS[4])
+    assert len(MIGRATIONS) == 5
 
 
 # ---------------------------------------------------------------------------
@@ -312,6 +326,20 @@ def test_legacy_task_tables_are_renamed_and_keep_their_rows(tmp_path) -> None:
             )
             == ("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 (
@@ -380,6 +408,178 @@ def test_fresh_database_never_holds_legacy_table_names(tmp_path) -> None:
         _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())
+
+
 # ---------------------------------------------------------------------------
 # Idempotence: second start is a no-op
 # ---------------------------------------------------------------------------
@@ -481,22 +681,28 @@ def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None:
 
 
 def _failing_migrations(prior_statements: list[str]) -> list[list[str]]:
-    """Three-migration list where migration 3 fails mid-application.
+    """Migration list where the LAST entry fails mid-application.
 
-    Migration 2 is a harmless no-op filler so the real runner — which
-    has already stamped version 2 by the time these tests re-run it —
-    still has a pending migration to fail on.
+    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.
     """
-    return [
-        prior_statements,
+    steps: list[list[str]] = [prior_statements]
+    steps += [
         [
-            "CREATE TABLE IF NOT EXISTS filler_table (id INTEGER PRIMARY KEY)"
-        ],
+            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:

tests/test_schema.py

@@ -16,6 +16,8 @@ from custom_components.nestquest.schema import (
     SCHEMA_V1_SCHEDULE_RULES_DDL,
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITIONS_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):
-    """Insert one valid task_definition, returning the definition id.
+def _insert_definition(database, rule_id=1, assignees=(1,), **overrides):
+    """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",
     }
@@ -96,6 +100,14 @@ def _insert_definition(database, child_id=1, rule_id=1, **overrides):
         )
     )
     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,12 +168,14 @@ 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_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_QUEST_INSTANCES_DDL,
@@ -302,7 +316,6 @@ def test_quest_definitions_columns_types_and_constraints(tmp_path) -> None:
             ("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),
@@ -319,13 +332,66 @@ def test_quest_definitions_foreign_keys_declared(tmp_path) -> None:
         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
 # ---------------------------------------------------------------------------
@@ -826,28 +892,274 @@ def test_quest_definitions_insert_requires_existing_child_and_rule(
             )
         )
         rule_id = _insert_rule(database)
-        definition_id = _insert_definition(
-            database, child_id=1, rule_id=rule_id
-        )
+        definition_id = _insert_definition(database, rule_id=rule_id)
         row = _run(
             database.fetch_one(
-                "SELECT title, child_id, schedule_rule_id FROM quest_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_quest_definitions_unknown_child_id_fails(tmp_path) -> None:
-    database = _open_db(tmp_path / "definitions-bad-child.db")
+def test_quest_definition_assignees_unknown_child_fails(tmp_path) -> None:
+    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_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())
 
@@ -863,7 +1175,7 @@ def test_quest_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())
 
@@ -874,7 +1186,7 @@ def test_quest_definitions_is_active_check_enforced(tmp_path) -> None:
         _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())
 
@@ -895,7 +1207,7 @@ def test_quest_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())
@@ -907,7 +1219,7 @@ def test_quest_definitions_title_not_null_enforced(tmp_path) -> None:
         _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())
 
@@ -919,7 +1231,7 @@ def test_quest_definitions_created_at_not_null_enforced(tmp_path) -> None:
         _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())
@@ -937,7 +1249,7 @@ def test_quest_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(
@@ -966,7 +1278,7 @@ 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", "quest_definitions"):
@@ -1410,8 +1722,8 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
 # ---------------------------------------------------------------------------
 
 
-def _insert_instance(database, definition_id=1, **overrides):
-    """Insert one valid task_instance, returning the instance id.
+def _insert_instance(database, definition_id=1, window="morning", **overrides):
+    """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",
     }
@@ -1469,7 +1782,7 @@ def _setup_definition(database) -> None:
     """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)
 
 
 # ---------------------------------------------------------------------------
@@ -1517,6 +1830,7 @@ def test_quest_instances_columns_types_and_constraints(tmp_path) -> None:
             ("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),
@@ -1572,7 +1886,7 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
         _run(database.close())
 
 
-def test_quest_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)
@@ -1585,15 +1899,20 @@ def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
         )
         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 "
-            "by exactly one implicit unique index"
+            "the (definition_id, child_id, due_date, window) UNIQUE "
+            "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(
@@ -1620,12 +1939,16 @@ def test_quest_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, "
+                "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())
 
@@ -1649,7 +1972,7 @@ def test_quest_instances_due_time_round_trips(tmp_path) -> None:
         _run(database.close())
 
 
-def test_quest_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,6 +1984,78 @@ def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
         _run(database.close())
 
 
+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:
@@ -1675,7 +2070,7 @@ def test_quest_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 quest_instances"))