Multi-assignee definitions via quest_definition_assignees (D-008)

Per IMPLEMENTATION-ORDER.md 1.3: a quest_definition_assignees table
(definition_id, child_id) with a composite primary key and FKs to
quest_definitions and children replaces the single child_id column on
quest_definitions.

- schema.py: definitions DDL loses child_id; new assignees table joins
  SCHEMA_V1_STATEMENTS; the definitions table body is extracted to
  QUEST_DEFINITIONS_TABLE_SQL so the rebuild migration cannot drift
  from the canonical shape.
- migrations.py: callable migrations now own their transaction and
  stamp via stamp_schema_version(database, version), because the
  rebuild must bracket itself with PRAGMA foreign_keys OFF/ON. New
  migration 3 creates the assignees table and, when a pre-D-008
  quest_definitions still carries child_id, rebuilds the table via
  SQLite's canonical procedure (create, copy, move assignees, drop,
  rename), verifies PRAGMA foreign_key_check inside the transaction,
  and stamps. Migration 2 converted to the new callable contract.
- dao_rules.py: QuestDefinitionsDao.create drops the child_id argument
  and applies assignee_child_ids atomically; set_assignee is replaced
  by add_assignee (idempotent, active-child validated under the
  connection lock), remove_assignee and list_assignees; list_by_child
  reads through the assignees table.
- dao_instances.py: upsert validates the child against
  quest_definition_assignees instead of the definitions column.
- tests: schema shape tests for the assignees table (columns, PK, FKs,
  duplicate/unknown constraints); DAO tests rewritten for the new
  assignment API; migration tests for the v2 -> v3 rebuild (rows and
  instances survive, child_id gone, foreign_key_check clean).

custom_components/nestquest/dao_instances.py

@@ -19,7 +19,9 @@ Upsert semantics: ``QuestInstancesDao.upsert`` is idempotent on
 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.
+UNIQUE(definition_id, due_date) is the last line of defense.  The
+upsert refuses a child that is not an assignee of the definition
+(D-008 multi-assignee model).
 
 ``delete_future_uncompleted`` implements the Feature 06/07 rule that
 reassignment and schedule edits regenerate future instances: it
@@ -207,18 +209,23 @@ 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:
+                assignee = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definition_assignees "
+                    "WHERE definition_id = ? AND child_id = ?",
+                    (definition_id, child_id),
+                )
+                if assignee is None:
                     raise ValueError(
-                        f"quest definition {definition_id} is assigned to "
+                        f"quest definition {definition_id} is not "
-                        f"child {definition[0]}, not {child_id}; instance "
+                        f"assigned to child {child_id}; instance must "
-                        "must carry the definition's assignee"
+                        "carry an assigned child"
                     )
                 child = await self._database.fetch_one(
                     "SELECT 1 FROM children WHERE id = ?", (child_id,)

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
+All SQL for ``schedule_rules``, ``quest_definitions`` and
-module per the feature guardrails: callers get typed dataclasses back
+``quest_definition_assignees`` lives in this module per the feature
-and never see raw rows or SQL.  Every method is async and runs through
+guardrails: callers get typed dataclasses back and never see raw rows
+or SQL.  Every method is async and runs through
 :class:`~.db.NestQuestDatabase`, so each statement executes on the HA
 executor and the event loop never blocks.
 
@@ -13,25 +15,29 @@ 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
+- Assignment is multi-assignee (D-008): :meth:`add_assignee` and
-  definition row and nothing else; reassignment never rewrites
+  :meth:`remove_assignee` change future instances only — the DAO writes
+  the assignee table and nothing else; assignment edits never rewrite
   existing instances or completion history (Feature 06 guardrail).
 - Validation of rule well-formedness and child activeness lives partly
   in the schema CHECKs (rule-type coherence) and partly in
-  :meth:`QuestDefinitionsDao.create` (active child, rule existence);
+  :meth:`QuestDefinitionsDao.create` (rule existence) and
-  no permission checks here, that is the Feature 09 gate's job.
+  :meth:`QuestDefinitionsDao.add_assignee` (active child); no
+  permission checks here, that is the Feature 09 gate's job.
 
-Concurrency: ``delete_rule_if_unreferenced`` performs its reference
+Concurrency: ``delete_rule_if_unreferenced`` and the assignee
-check and DELETE inside one transaction under the connection-scoped
+mutations perform their checks and writes inside one transaction under
-lock shared with :mod:`.dao_children`, so a definition referencing the
+the connection-scoped lock shared with :mod:`.dao_children`, so a
-rule cannot appear between check and delete on this connection.
+definition referencing the rule (or an assignee appearing) cannot slip
+in between check and write on this connection.
 """
 from __future__ import annotations
 
 import datetime
 from dataclasses import dataclass
 
-from .dao_children import _connection_lock
+from .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"
@@ -82,13 +88,12 @@ 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
@@ -100,7 +105,7 @@ _RULE_COLUMNS = (
     "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"
 )
 
@@ -125,11 +130,10 @@ def _definition_from_row(row: tuple) -> QuestDefinitionRecord:
         title=row[1],
         description=row[2],
         icon=row[3],
-        child_id=row[4],
+        schedule_rule_id=row[4],
-        schedule_rule_id=row[5],
+        due_time=row[5],
-        due_time=row[6],
+        is_active=bool(row[6]),
-        is_active=bool(row[7]),
+        created_at=row[7],
-        created_at=row[8],
     )
 
 
@@ -272,11 +276,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 +292,6 @@ class QuestDefinitionsDao:
     async def create(
         self,
         title: str,
-        child_id: int,
         schedule_rule_id: int,
         created_at: str,
         *,
@@ -293,53 +299,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
+        Validates the schedule rule as part of the same serialized
-        transaction as the insert: the child must exist and be active,
+        transaction as the insert: the rule must exist AT INSERT TIME —
-        and the rule must exist, AT INSERT TIME — a child deactivated
+        a rule deleted concurrently cannot slip between the check and
-        or a rule deleted concurrently cannot slip between the checks
+        the INSERT.  ``assignee_child_ids`` (each child must exist and
-        and the INSERT, which would otherwise create an invalid
+        be active) is applied in the same transaction, so a definition
-        assignment or surface as a raw FK IntegrityError instead of
+        can never be observed with a stale or missing assignee set; an
-        the documented ValueError.  An INACTIVE child would pass the
+        empty list or None creates an unassigned definition, which
-        schema foreign keys, which is why the activeness check lives
+        callers may fill via :meth:`add_assignee`.
-        here.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
-                await self._validate_assignable(
+                await self._validate_rule_exists(schedule_rule_id)
-                    child_id, schedule_rule_id
+                for child_id in assignee_child_ids or []:
-                )
+                    await self._validate_assignable_child(child_id)
                 result = await self._database.execute(
                     "INSERT INTO quest_definitions (title, description, "
-                    "icon, child_id, schedule_rule_id, due_time, "
+                    "icon, schedule_rule_id, due_time, is_active, "
-                    "is_active, created_at) "
+                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
-                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         title,
                         description,
                         icon,
-                        child_id,
                         schedule_rule_id,
                         due_time,
                         int(is_active),
                         created_at,
                     ),
                 )
-                definition = await self.get(result.lastrowid)
+                definition_id = result.lastrowid
+                for child_id in assignee_child_ids or []:
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_assignees "
+                        "(definition_id, child_id) VALUES (?, ?)",
+                        (definition_id, child_id),
+                    )
+                definition = await self.get(definition_id)
         assert definition is not None
         return definition
 
-    async def _validate_assignable(
+    async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
-        self, child_id: int, schedule_rule_id: int
+        """Raise ValueError unless the rule exists.
-    ) -> None:
-        """Raise ValueError unless child (active) and rule both exist.
 
         MUST be called inside the connection lock so the verdict cannot
         go stale before the caller's INSERT commits.
         """
-        await self._validate_assignable_child(child_id)
         rule = await self._database.fetch_one(
             "SELECT 1 FROM schedule_rules WHERE id = ?",
             (schedule_rule_id,),
@@ -359,10 +367,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 +406,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
+        operations (:meth:`add_assignee`, :meth:`remove_assignee`,
-        stay separately loggable and permission-gateable.
+        :meth:`set_active`) so they stay separately loggable and
+        permission-gateable.
         """
         assignments: list[str] = []
         parameters: list[object] = []
@@ -427,32 +443,75 @@ class QuestDefinitionsDao:
         )
         return result.rowcount
 
-    async def set_assignee(
+    async def add_assignee(
         self, definition_id: int, child_id: int
-    ) -> int:
+    ) -> None:
-        """Reassign the definition to ``child_id``; rows updated (0 if absent).
+        """Assign ``child_id`` to the definition (idempotent).
 
-        Validates the new assignee inside the same serialized
+        Validates the assignee inside the same serialized transaction
-        transaction as the UPDATE (existing and active AT ASSIGNMENT
+        as the INSERT (existing and active AT ASSIGNMENT TIME), so a
-        TIME), so a child deactivated concurrently cannot become the
+        child deactivated concurrently cannot become an assignee.
-        assignee.  Changes future instances only: already-generated
+        Assigning an already-assigned child is a no-op, mirroring the
-        instances and completion history are never rewritten here.
+        composite primary key's storage-level guarantee.  Changes
+        future instances only: already-generated instances and
+        completion history are never rewritten here.
         """
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 await self._validate_assignable_child(child_id)
-                result = await self._database.execute(
+                definition = await self._database.fetch_one(
-                    "UPDATE quest_definitions SET child_id = ? "
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
-                    "WHERE id = ?",
+                    (definition_id,),
-                    (child_id, definition_id),
                 )
-        return result.rowcount
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                await self._database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, ?) "
+                    "ON CONFLICT (definition_id, child_id) DO NOTHING",
+                    (definition_id, child_id),
+                )
+
+    async def remove_assignee(
+        self, definition_id: int, child_id: int
+    ) -> bool:
+        """Remove the assignee link; True when a row was removed.
+
+        Future materialization stops for this (definition, child) pair;
+        existing instances and completion history are never touched
+        here (Feature 06 guardrail).
+        """
+        async with _connection_lock(self._database):
+            result = await self._database.execute(
+                "DELETE FROM quest_definition_assignees "
+                "WHERE definition_id = ? AND child_id = ?",
+                (definition_id, child_id),
+            )
+        return result.rowcount > 0
+
+    async def list_assignees(self, definition_id: int) -> list[ChildRecord]:
+        """Return the definition's assignees as child records.
+
+        Ordered by the children table's stable display order
+        (``sort_order``, then id) so callers render a deterministic
+        roster.  A definition with no assignees returns [].
+        """
+        rows = await self._database.fetch_all(
+            f"SELECT {_CHILD_COLUMNS} FROM children "
+            "WHERE id IN (SELECT child_id "
+            "FROM quest_definition_assignees WHERE definition_id = ?) "
+            "ORDER BY sort_order, id",
+            (definition_id,),
+        )
+        return [_child_from_row(row) for row in rows]
 
     async def _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,21 @@ 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
+names before anything else assumes the new model.  Migration 3 brings
-inside the same transaction as its version stamp, so a crash mid-rename
+pre-D-008 databases to the multi-assignee model: it creates
-rolls both back together.
+``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.  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
+an async callable taking ``(database, target_version)``.  Callables
-whose SQL depends on the database's current state (the legacy rename
+exist for steps whose SQL depends on the database's current state (a
-only fires when a legacy table is actually present) and run under the
+legacy rename only fires when a legacy table is present; a rebuild only
-same per-migration transaction as statement lists.
+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 +69,21 @@ 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,
+    SCHEMA_V1_STATEMENTS,
+    SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_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.
+#: an async callable taking ``(database, target_version)`` that manages
-MigrationStep = Union[Sequence[str], Callable[[NestQuestDatabase], Awaitable[None]]]
+#: 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 +108,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 +121,34 @@ 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)
+
+
+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 +157,96 @@ 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,
+    internal autoindex names.  Runs in its own transaction together
-    so the renames and the version stamp commit or roll back together.
+    with the version stamp, so a crash rolls both back.
     """
-    for legacy_name, current_name in LEGACY_TASK_TABLE_RENAMES:
+    async with database.transaction():
-        row = await database.fetch_one(
+        for legacy_name, current_name in LEGACY_TASK_TABLE_RENAMES:
-            "SELECT 1 FROM sqlite_master WHERE type = 'table' "
+            row = await database.fetch_one(
-            "AND name = ?",
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' "
-            (legacy_name,),
+                "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
             )
+            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")
 
 
 #: The ordered migration list.  Append-only: never edit an applied
@@ -136,6 +255,7 @@ async def _rename_legacy_task_tables(database: NestQuestDatabase) -> None:
 MIGRATIONS: Sequence[MigrationStep] = [
     MIGRATION_1_V1_DDL,
     _rename_legacy_task_tables,
+    _add_definition_assignees,
 ]
 
 
@@ -248,17 +368,18 @@ async def apply_migrations(
 
         for target in range(current + 1, latest + 1):
             applied = False
-            async with database.transaction():
+            step = migrations[target - 1]
-                # Re-read the version INSIDE the write transaction: the
+            if callable(step):
-                # outer read was advisory, and the wrapper serializes
+                # Callable migrations own their transaction: some steps
-                # individual statements, not the read-decide-apply
+                # (table rebuilds) must run connection pragmas such as
-                # decision.  Under the transaction's lock the version
+                # ``foreign_keys = OFF`` BEFORE opening one, which the
-                # cannot change between the read and the stamp, and the
+                # runner's uniform transaction cannot express, and they
-                # loser of an interleaved race (a queued second caller
+                # stamp inside their own transaction so a failure rolls
-                # re-running apply_migrations after the winner already
+                # the statements and the version stamp back together.
-                # migrated) re-reads here, sees the winner's stamp, and
+                # The version is re-read here under the migration lock:
-                # skips instead of re-applying a non-idempotent
+                # runners on one connection are serialized by it, so
-                # migration with a stale version.
+                # the version cannot change between this read and the
+                # callable's commit.
                 in_tx_version = await read_schema_version(database)
                 if in_tx_version > latest:
                     raise RuntimeError(
@@ -267,41 +388,39 @@ async def apply_migrations(
                         f"{latest}); downgrades are not supported"
                     )
                 if in_tx_version < target:
-                    step = migrations[target - 1]
+                    await step(database, target)
-                    if callable(step):
+                    applied = True
-                        await step(database)
+            else:
-                        applied_statements = 0
+                async with database.transaction():
-                    else:
+                    # 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(
-                    LOGGER.info(
+                    "NestQuest schema migrated to version %d",
-                        "NestQuest schema migrated to version %d",
+                    target,
-                        target,
+                )
-                    )
-                else:
-                    LOGGER.info(
-                        "NestQuest schema migrated to version %d "
-                        "(%d statement%s)",
-                        target,
-                        applied_statements,
-                        "s" if applied_statements != 1 else "",
-                    )
 
         return latest

custom_components/nestquest/schema.py

@@ -86,6 +86,15 @@ 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 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)
@@ -219,18 +228,34 @@ 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 (
+    CREATE TABLE IF NOT EXISTS quest_definition_assignees (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
-        title TEXT NOT NULL,
-        description TEXT,
-        icon TEXT,
         child_id INTEGER NOT NULL REFERENCES children(id),
-        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        PRIMARY KEY (definition_id, child_id)
-        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
     )
     """,
 ]
@@ -316,6 +341,7 @@ 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_PRESENCE_SCHEDULES_DDL,
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     *SCHEMA_V1_QUEST_INSTANCES_DDL,

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,
@@ -184,17 +184,15 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
         with pytest.raises(ValueError, match="definition 999"):
             await instances.upsert(999, child.id, D1, _now_stamp())
         # Unknown child: the definition-assignee check fires first and
-        # is the one that matters (an instance must carry its
+        # is the one that matters (an instance must carry an assigned
-        # definition's assignee).  To reach the child-existence check
+        # child).  Under D-008 the assignee check is a membership test
-        # at all, the definition would have to claim child 999 — an
+        # against quest_definition_assignees, so child 999 simply is
-        # FK-impossible state — so the child check only guards direct
+        # not an assignee — the error names the pair.
-        # callers with a matching fake assignee; assert the guard
-        # order instead of forcing the FK-impossible state.
         rule = await rules.create("daily", D1)
         other_definition = await definitions.create(
-            "X", child.id, rule.id, _now_stamp()
+            "X", rule.id, _now_stamp(), assignee_child_ids=[child.id]
         )
-        with pytest.raises(ValueError, match="not 999"):
+        with pytest.raises(ValueError, match="not assigned to child 999"):
             await instances.upsert(other_definition.id, 999, D1, _now_stamp())
         return None
 
@@ -329,7 +327,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,
@@ -406,7 +407,7 @@ 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())
@@ -537,8 +538,8 @@ def test_delete_future_uncompleted_other_definitions_untouched(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         other = await definitions.create(
-            "Make bed", child.id, (await rules.create("daily", D1)).id,
+            "Make bed", (await rules.create("daily", D1)).id,
-            _now_stamp(),
+            _now_stamp(), assignee_child_ids=[child.id],
         )
         await instances.upsert(definition.id, child.id, D20, _now_stamp())
         await instances.upsert(other.id, child.id, D20, _now_stamp())
@@ -1090,7 +1091,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"
@@ -206,6 +210,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 +220,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())
+            await w.definitions.create(
-        # constraint: inactive child rejected
+                "X", w.rule.id, _stamp(), assignee_child_ids=[999]
+            )
+        # constraint: inactive child rejected on assignment
         await w.children.set_active(w.child2.id, False)
         with pytest.raises(ValueError, match="inactive"):
-            await w.definitions.set_assignee(w.definition.id, w.child2.id)
+            await w.definitions.add_assignee(w.definition.id, w.child2.id)
+        # assignee removal + re-adding the active child
+        assert await w.definitions.remove_assignee(
+            w.definition.id, w.child.id
+        ) is True
+        assert await w.definitions.remove_assignee(
+            w.definition.id, w.child.id
+        ) is False
+        await w.definitions.add_assignee(w.definition.id, w.child.id)
         # set_active: deactivate, never delete
         assert await w.definitions.set_active(w.definition.id, False) == 1
         assert (await w.definitions.get(w.definition.id)).is_active is False

tests/test_dao_rules.py

@@ -266,7 +266,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 +288,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 +335,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 +348,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(
+                        definitions.create("Brush teeth", rule_a.id, NOW)
-                            "Brush teeth", child.id, rule_a.id, NOW
-                        )
                     )
                     await started.wait()
                     delete_task = asyncio.ensure_future(
@@ -382,7 +380,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 +432,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 +462,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 +496,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 +511,11 @@ def test_definition_list_by_child_filters_and_orders(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other_child = await children.create("Bo", NOW)
-        await definitions.create("A", child.id, rule.id, NOW)
+        await definitions.create("A", rule.id, NOW, assignee_child_ids=[child.id])
-        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.create("B", rule.id, NOW, assignee_child_ids=[child.id])
-        await definitions.create("C", other_child.id, rule.id, NOW)
+        await definitions.create(
+            "C", rule.id, NOW, assignee_child_ids=[other_child.id]
+        )
         mine = await definitions.list_by_child(child.id)
         assert [d.title for d in mine] == ["B", "A"]
         return mine
@@ -527,8 +535,8 @@ def test_definition_list_by_child_missing_child_returns_empty(
 def test_definition_list_active_excludes_inactive(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        first = await definitions.create("A", child.id, rule.id, NOW)
+        first = await definitions.create("A", rule.id, NOW)
-        await definitions.create("B", child.id, rule.id, NOW)
+        await definitions.create("B", rule.id, NOW)
         await definitions.set_active(first.id, False)
         active = await definitions.list_active()
         assert [d.title for d in active] == ["B"]
@@ -547,7 +555,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 +580,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 +600,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 +613,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 +625,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 +658,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 +695,152 @@ def test_definition_no_delete_method_exists(tmp_path) -> None:
     _with_db(tmp_path, "definition-no-delete.db")(_body)
 
 
-def test_definition_set_assignee_round_trip(tmp_path) -> None:
+def test_definition_add_assignee_round_trip(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
-        assert await definitions.set_assignee(definition.id, other.id) == 1
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
-        fetched = await definitions.get(definition.id)
+        )
-        assert fetched.child_id == other.id
+        await definitions.add_assignee(definition.id, other.id)
-        return fetched
+        assignees = await definitions.list_assignees(definition.id)
+        assert sorted(c.id for c in assignees) == sorted([child.id, other.id])
+        # Idempotent: re-adding an existing assignee stays a no-op.
+        await definitions.add_assignee(definition.id, other.id)
+        assignees = await definitions.list_assignees(definition.id)
+        assert sorted(c.id for c in assignees) == sorted([child.id, other.id])
+        return assignees
 
-    _with_db(tmp_path, "definition-set-assignee.db")(_body)
+    _with_db(tmp_path, "definition-add-assignee.db")(_body)
 
 
-def test_definition_set_assignee_inactive_child_raises(tmp_path) -> None:
+def test_definition_add_assignee_inactive_child_raises(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         await children.set_active(other.id, False)
         with pytest.raises(ValueError, match="inactive"):
-            await definitions.set_assignee(definition.id, other.id)
+            await definitions.add_assignee(definition.id, other.id)
-        fetched = await definitions.get(definition.id)
+        assignees = await definitions.list_assignees(definition.id)
-        assert fetched.child_id == child.id, (
+        assert [c.id for c in assignees] == [child.id], (
-            "failed reassignment must leave the current assignee intact"
+            "failed assignment must leave the current assignees intact"
         )
-        return fetched
+        return assignees
 
     _with_db(tmp_path, "definition-assignee-inactive.db")(_body)
 
 
-def test_definition_set_assignee_unknown_child_raises(tmp_path) -> None:
+def test_definition_add_assignee_unknown_child_raises(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         with pytest.raises(ValueError, match="does not exist"):
-            await definitions.set_assignee(definition.id, 999)
+            await definitions.add_assignee(definition.id, 999)
         return None
 
     _with_db(tmp_path, "definition-assignee-unknown.db")(_body)
 
 
-def test_definition_set_assignee_missing_definition_returns_zero(
+def test_definition_add_assignee_missing_definition_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.add_assignee(999, child.id)
+        return None
+
+    _with_db(tmp_path, "definition-assignee-missing.db")(_body)
+
+
+def test_definition_remove_assignee_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        other = await children.create("Bo", NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id, other.id]
+        )
+        assert await definitions.remove_assignee(definition.id, other.id) is True
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [child.id]
+        # Removing a non-assignee reports False.
+        assert await definitions.remove_assignee(definition.id, other.id) is False
+        return assignees
+
+    _with_db(tmp_path, "definition-remove-assignee.db")(_body)
+
+
+def test_definition_list_assignees_orders_by_child_sort_order(
     tmp_path,
 ) -> None:
     async def _body(database, rules, definitions, children, child):
-        other = await children.create("Bo", NOW)
+        rule = await rules.create("daily", "2026-09-14")
-        return await definitions.set_assignee(999, other.id)
+        second = await children.create("Bo", NOW)
+        third = await children.create("Cleo", NOW)
+        await children.reorder([third.id, child.id, second.id])
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[second.id, child.id, third.id]
+        )
+        assignees = await definitions.list_assignees(definition.id)
+        assert [c.id for c in assignees] == [third.id, child.id, second.id]
+        return assignees
 
-    assert _with_db(tmp_path, "definition-assignee-missing.db")(_body) == 0
+    _with_db(tmp_path, "definition-assignees-order.db")(_body)
 
 
-def test_definition_set_assignee_leaves_history_rows_alone(
+def test_definition_unassigned_has_no_assignees(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        assert await definitions.list_assignees(definition.id) == []
+        return None
+
+    _with_db(tmp_path, "definition-no-assignees.db")(_body)
+
+
+def test_definition_shared_by_multiple_children_via_list_by_child(
     tmp_path,
 ) -> None:
-    """Reassignment writes the definition row only: any instances that
+    """D-008: one definition covering several children appears for
-    already exist keep their generation-time child_id (Feature 06
+    every assignee through list_by_child, once each."""
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        second = await children.create("Bo", NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id, second.id]
+        )
+        for assignee in (child, second):
+            mine = await definitions.list_by_child(assignee.id)
+            assert [d.id for d in mine] == [definition.id]
+        return definition
+
+    _with_db(tmp_path, "definition-shared.db")(_body)
+
+
+def test_definition_assignment_leaves_history_rows_alone(
+    tmp_path,
+) -> None:
+    """Assignment edits write the assignees table only: any instances
+    that already exist keep their generation-time child_id (Feature 06
     guardrail 'changes future instances only').
     """
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
         other = await children.create("Bo", NOW)
-        definition = await definitions.create("A", child.id, rule.id, NOW)
+        definition = await definitions.create(
+            "A", rule.id, NOW, assignee_child_ids=[child.id]
+        )
         # Simulate an already-generated instance (materialization owns
-        # this table, but its row shape proves reassignment isolation).
+        # this table, but its row shape proves assignment isolation).
         await database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
             "due_date, generated_at) VALUES (?, ?, ?, ?)",
             (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 +889,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)\b",
         re.IGNORECASE,
     )
     offenders: list[str] = []
@@ -813,7 +901,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 +941,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)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (
@@ -898,7 +986,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)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (

tests/test_migrations.py

@@ -63,6 +63,7 @@ def _counts(database) -> dict[str, int]:
         "admin_users",
         "schedule_rules",
         "quest_definitions",
+        "quest_definition_assignees",
         "presence_schedules",
         "presence_overrides",
         "quest_instances",
@@ -99,6 +100,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "admin_users",
             "schedule_rules",
             "quest_definitions",
+            "quest_definition_assignees",
             "presence_schedules",
             "presence_overrides",
             "quest_instances",
@@ -136,15 +138,18 @@ 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 the 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, and
+    migration 3 brings pre-D-008 definitions to the multi-assignee
+    model.
     """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
     assert callable(MIGRATIONS[1])
-    assert len(MIGRATIONS) == 2
+    assert callable(MIGRATIONS[2])
+    assert len(MIGRATIONS) == 3
 
 
 # ---------------------------------------------------------------------------
@@ -312,6 +317,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 +399,136 @@ 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
+    )
+    """,
+]
+
+
+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.
+    """
+    for sql in SCHEMA_V1_STATEMENTS:
+        if "quest_definitions " in sql:
+            continue  # replaced by the legacy-shape DDL below
+        _run(database.execute(sql))
+    for sql in _V2_DEFINITIONS_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, still referencing the definition.
+        instance = _run(
+            database.fetch_one(
+                "SELECT definition_id, child_id, due_date "
+                "FROM quest_instances WHERE id = 1"
+            )
+        )
+        assert instance == (1, 2, "2026-09-15")
+        # 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 +630,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
+    Sized one entry longer than the real MIGRATIONS and padded with
-    has already stamped version 2 by the time these tests re-run it —
+    harmless no-op fillers, so a fully-migrated database (already at
-    still has a pending migration to fail on.
+    len(MIGRATIONS)) still has exactly one pending migration — the
+    failing one — to attempt.
     """
-    return [
+    steps: list[list[str]] = [prior_statements]
-        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,7 @@ 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_INSTANCES_DDL,
     async_apply_ddl,
 )
@@ -73,15 +74,17 @@ def _insert_rule(database, **overrides):
     return result.lastrowid
 
 
-def _insert_definition(database, child_id=1, rule_id=1, **overrides):
+def _insert_definition(database, rule_id=1, assignees=(1,), **overrides):
-    """Insert one valid task_definition, returning the definition id.
+    """Insert one valid quest_definition, returning the definition id.
 
+    ``assignees`` rows are written to ``quest_definition_assignees``
+    after the definition insert (the definition itself carries no
+    child).  Pass ``assignees=None`` for an unassigned definition.
     Pass ``_OMIT`` as a value to leave that column out of the INSERT so
     its DDL DEFAULT applies.
     """
     values: dict[str, object] = {
         "title": "Brush teeth",
-        "child_id": child_id,
         "schedule_rule_id": rule_id,
         "created_at": "2026-09-13T00:00:00+00:00",
     }
@@ -96,6 +99,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 +167,13 @@ def _insert_override(database, child_id, **overrides):
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_six_tables() -> None:
+def test_schema_v1_statements_compose_the_seven_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_PRESENCE_SCHEDULES_DDL,
         *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
         *SCHEMA_V1_QUEST_INSTANCES_DDL,
@@ -302,7 +314,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 +330,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 +890,114 @@ def test_quest_definitions_insert_requires_existing_child_and_rule(
             )
         )
         rule_id = _insert_rule(database)
-        definition_id = _insert_definition(
+        definition_id = _insert_definition(database, rule_id=rule_id)
-            database, child_id=1, rule_id=rule_id
-        )
         row = _run(
             database.fetch_one(
-                "SELECT title, child_id, schedule_rule_id FROM 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:
+def test_quest_definition_assignees_unknown_child_fails(tmp_path) -> None:
-    database = _open_db(tmp_path / "definitions-bad-child.db")
+    database = _open_db(tmp_path / "assignees-bad-child.db")
     try:
         _apply(database)
         _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
         with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
-            _insert_definition(database, child_id=999)
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, 999)",
+                    (definition_id,),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_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_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 +1013,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 +1024,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 +1045,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 +1057,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 +1069,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 +1087,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 +1116,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"):
@@ -1469,7 +1619,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)
 
 
 # ---------------------------------------------------------------------------
@@ -1675,7 +1825,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"))