Add full-lifecycle test proving quest-definition ops leave history untouched

custom_components/nestquest/dao_rules.py

@@ -962,14 +962,18 @@ class QuestDefinitionsDao:
 
     async def add_assignee(
         self, definition_id: int, child_id: int
-    ) -> None:
+    ) -> list[ChildRecord]:
         """Assign ``child_id`` to the definition (idempotent).
 
         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
+        composite primary key's storage-level guarantee.  Returns the
+        definition's assignees read back INSIDE the same transaction
+        that performed the write, so the returned roster is a
+        consistent post-assignment snapshot — a concurrent assign or
+        unassign cannot race it after the write commits.  Changes
         future instances only: already-generated instances and
         completion history are never rewritten here.
         """
@@ -990,6 +994,7 @@ class QuestDefinitionsDao:
                     "ON CONFLICT (definition_id, child_id) DO NOTHING",
                     (definition_id, child_id),
                 )
+                return await self.list_assignees(definition_id)
 
     async def remove_assignee(
         self, definition_id: int, child_id: int

custom_components/nestquest/quest_definitions.py

@@ -3,9 +3,11 @@
 Sits between the Feature 09 service gate and the typed
 :class:`~.dao_rules.QuestDefinitionsDao`: callers get validation and the
 all-or-nothing create/edit here, and never touch the DAO SQL directly.
-:func:`create_quest_definition` is the create path and
-:func:`edit_quest_definition` the edit path; there is no permission
-check here (that is Feature 09).
+:func:`create_quest_definition` is the create path,
+:func:`edit_quest_definition` the edit path, :func:`assign_child` and
+:func:`unassign_child` the assignment paths, and
+:func:`set_quest_definition_active` the deactivate/reactivate path.
+There is no permission check here (that is Feature 09).
 
 Validation policy mirrors :mod:`.children`:
 
@@ -39,13 +41,15 @@ from __future__ import annotations
 import datetime
 from dataclasses import dataclass
 
-from .dao_children import ChildRecord
+from .dao_children import ChildRecord, _connection_lock
 from .dao_rules import (
     QuestDefinitionRecord,
     QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
     ScheduleRuleStorage,
+    ScheduleRulesDao,
     _UNSET,
+    _validate_date,
     _validate_time,
     _validate_window,
     schedule_rule_from_storage,
@@ -53,12 +57,12 @@ from .dao_rules import (
     schedule_rule_to_storage,
 )
 from .db import NestQuestDatabase
-from .recurrence import ScheduleRule
+from .recurrence import ScheduleRule, occurs_on
 
 
 @dataclass(frozen=True)
 class CreatedQuestDefinition:
-    """The persisted result of a create or edit.
+    """The persisted result of a create, edit or activation change.
 
     ``definition`` is the stored row; ``rule`` is the schedule rule
     decoded back through the storage mapping; ``assignees`` and
@@ -185,6 +189,18 @@ def _validate_definition_id(value: object) -> int:
     return value
 
 
+def _validate_child_id(value: object) -> int:
+    """Reject non-int child ids (bools included) before any lookup.
+
+    SQLite binds Python bools as integers, so ``True`` would silently
+    address child 1 and a float would round — malformed service input
+    must never mutate the wrong profile.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"child_id must be an integer, got {value!r}")
+    return value
+
+
 async def create_quest_definition(
     database: NestQuestDatabase,
     title: str,
@@ -321,3 +337,216 @@ async def edit_quest_definition(
         assignees=snapshot.assignees,
         windows=snapshot.windows,
     )
+
+
+async def assign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+) -> list[ChildRecord]:
+    """Assign ``child_id`` to the definition (idempotent).
+
+    Validates that both ids are plain ints (bools and floats rejected)
+    and that the child exists and is active — the DAO enforces the
+    active/exists check inside its transaction, and also confirms the
+    definition exists.  Assigning an already-assigned child is a no-op
+    (no duplicate row).  Returns the definition's assignees read back
+    inside the DAO's same transaction, a consistent post-assignment
+    roster (a concurrent assign/unassign cannot race it after the
+    write).  Raises ValueError on any rejected argument.  Only the
+    ``quest_definition_assignees`` link is written; existing instances
+    and completion history are never touched.
+    """
+    _validate_definition_id(definition_id)
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    try:
+        return await dao.add_assignee(definition_id, child_id)
+    except ValueError as error:
+        # The DAO names only the child/definition; re-raise so the
+        # public contract names the field the caller actually passed.
+        message = str(error)
+        if message.startswith("child "):
+            raise ValueError(f"child_id: {message}") from error
+        if message.startswith("quest definition "):
+            raise ValueError(f"definition_id: {message}") from error
+        raise
+
+
+async def unassign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+) -> bool:
+    """Unassign ``child_id`` from the definition; True when removed.
+
+    Validates both ids are plain ints (bools and floats rejected) and
+    that the definition exists, then removes the (definition, child)
+    link.  Returns True when a row was removed and False when the child
+    was not currently assigned.  Raises ValueError on a rejected
+    argument.  Only the ``quest_definition_assignees`` link is written;
+    existing instances and completion history are never touched.
+    """
+    _validate_definition_id(definition_id)
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    definition = await dao.get(definition_id)
+    if definition is None:
+        raise ValueError(
+            f"definition_id: quest definition {definition_id} does not exist"
+        )
+    return await dao.remove_assignee(definition_id, child_id)
+
+
+async def set_quest_definition_active(
+    database: NestQuestDatabase,
+    definition_id: int,
+    is_active: bool,
+) -> CreatedQuestDefinition:
+    """Deactivate or reactivate a definition; returns the updated view.
+
+    Flipping the flag is the only removal path (Feature 06 guardrail):
+    a deactivated definition stops future instance generation but its
+    row, schedule rule, assignees and windows all survive, and existing
+    instances plus completion history are never touched.  Reactivation
+    re-enables future generation; generating or removing instances is
+    Feature 07 and does not happen here.
+
+    ``definition_id`` must be a plain int (bools and floats rejected,
+    since SQLite would bind ``True`` onto definition 1) and
+    ``is_active`` must be a real bool — the DAO's ``int()`` would
+    otherwise silently coerce strings and numerics ("1", 0) into a
+    state the caller never asked for.  Raises ValueError naming the
+    offending field when the id is malformed, ``is_active`` is not a
+    bool, or the definition does not exist.
+
+    Atomicity: the existence check, the UPDATE and the readback of the
+    definition, its rule, assignees and windows all run inside the
+    connection-scoped lock, so the returned bundle is a consistent
+    snapshot that a concurrent mutation cannot race.
+    """
+    _validate_definition_id(definition_id)
+    if not isinstance(is_active, bool):
+        raise ValueError(f"is_active must be a real bool, got {is_active!r}")
+    async with _connection_lock(database):
+        dao = QuestDefinitionsDao(database)
+        if await dao.get(definition_id) is None:
+            raise ValueError(
+                f"definition_id: quest definition {definition_id} "
+                "does not exist"
+            )
+        await dao.set_active(definition_id, is_active)
+        updated = await dao.get(definition_id)
+        assert updated is not None
+        rule_record = await ScheduleRulesDao(database).get(
+            updated.schedule_rule_id
+        )
+        assert rule_record is not None
+        assignees = await dao.list_assignees(definition_id)
+        windows = await dao.list_windows(definition_id)
+    rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(rule_record)
+    )
+    return CreatedQuestDefinition(
+        definition=updated,
+        rule=rule,
+        assignees=assignees,
+        windows=windows,
+    )
+
+
+async def _bundle_definition(
+    database: NestQuestDatabase,
+    dao: QuestDefinitionsDao,
+    definition: QuestDefinitionRecord,
+) -> CreatedQuestDefinition:
+    """Assemble a definition's rich view: decoded rule + assignees + windows.
+
+    Reads the schedule rule through :class:`~.dao_rules.ScheduleRulesDao`
+    and the assignees and windows through the DAO, then wraps them in the
+    same :class:`CreatedQuestDefinition` shape the create/edit/activate
+    paths return, so every query helper here reuses one snapshot type
+    rather than inventing a second one.
+    """
+    rule_record = await ScheduleRulesDao(database).get(
+        definition.schedule_rule_id
+    )
+    assert rule_record is not None
+    rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(rule_record)
+    )
+    assignees = await dao.list_assignees(definition.id)
+    windows = await dao.list_windows(definition.id)
+    return CreatedQuestDefinition(
+        definition=definition,
+        rule=rule,
+        assignees=assignees,
+        windows=windows,
+    )
+
+
+async def list_active_definitions(
+    database: NestQuestDatabase,
+) -> list[CreatedQuestDefinition]:
+    """Return every ACTIVE definition with its decoded rule and links.
+
+    Ordered by rising definition id, matching the DAO's ``list_active``
+    order, so callers get a stable, repeatable sequence.  Each entry is
+    the same rich :class:`CreatedQuestDefinition` snapshot as the rest of
+    the layer: the stored row plus its decoded
+    :class:`~.recurrence.ScheduleRule`, assignees and windows.
+    """
+    dao = QuestDefinitionsDao(database)
+    definitions = await dao.list_active()
+    return [
+        await _bundle_definition(database, dao, definition)
+        for definition in definitions
+    ]
+
+
+async def list_definitions_for_child(
+    database: NestQuestDatabase,
+    child_id: int,
+) -> list[CreatedQuestDefinition]:
+    """Return the definitions assigned to ``child_id`` (rich view).
+
+    ``child_id`` must be a plain int (bools and floats rejected, since
+    SQLite would bind ``True`` onto child 1).  Results are ordered by
+    rising definition id, matching ``list_active_definitions`` and
+    ``list_definitions_firing_on``.  Each entry is a
+    :class:`CreatedQuestDefinition` snapshot bundling the definition with
+    its decoded rule, assignees and windows.
+    """
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    definitions = await dao.list_by_child(child_id)
+    bundles = [
+        await _bundle_definition(database, dao, definition)
+        for definition in definitions
+    ]
+    return sorted(bundles, key=lambda bundle: bundle.definition.id)
+
+
+async def list_definitions_firing_on(
+    database: NestQuestDatabase,
+    date: str,
+) -> list[CreatedQuestDefinition]:
+    """Return the ACTIVE definitions whose rule fires on ``date``.
+
+    ``date`` must be a strict ISO calendar date (YYYY-MM-DD); a malformed
+    value raises ValueError naming the field.  Each active definition's
+    stored rule is decoded through the task-1 storage mapping and handed
+    to the recurrence engine's :func:`~.recurrence.occurs_on`; only the
+    definitions whose decoded rule fires on the date are returned, in
+    rising definition-id order, each as the same rich
+    :class:`CreatedQuestDefinition` snapshot as the rest of the layer.
+    """
+    _validate_date(date, "date")
+    target = datetime.datetime.strptime(date, "%Y-%m-%d").date()
+    dao = QuestDefinitionsDao(database)
+    definitions = await dao.list_active()
+    bundles = [
+        await _bundle_definition(database, dao, definition)
+        for definition in definitions
+    ]
+    return [bundle for bundle in bundles if occurs_on(bundle.rule, target)]

tests/test_quest_definitions.py

@@ -2,11 +2,16 @@
 from __future__ import annotations
 
 import asyncio
+import datetime
 import re
 
 import pytest
 
 from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_instances import (
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
 from custom_components.nestquest.dao_rules import (
     QuestDefinitionsDao,
     ScheduleRulesDao,
@@ -16,8 +21,14 @@ from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
 from custom_components.nestquest.quest_definitions import (
     CreatedQuestDefinition,
+    assign_child,
     create_quest_definition,
     edit_quest_definition,
+    list_active_definitions,
+    list_definitions_firing_on,
+    list_definitions_for_child,
+    set_quest_definition_active,
+    unassign_child,
 )
 from custom_components.nestquest.recurrence import RuleType, ScheduleRule
 
@@ -70,6 +81,15 @@ def _daily_rule() -> ScheduleRule:
     return ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-14")
 
 
+def _today_iso() -> str:
+    """Today's date in ISO form, computed at call time (never cached).
+
+    ``QuestInstancesDao.upsert`` refuses past due dates, so seeded
+    instances use a date that is always today-or-later at test time.
+    """
+    return datetime.date.today().isoformat()
+
+
 # ---------------------------------------------------------------------------
 # create_quest_definition: successful creates
 # ---------------------------------------------------------------------------
@@ -880,3 +900,834 @@ def test_edit_post_write_failure_rolls_back_everything(tmp_path) -> None:
         return None
 
     _with_db(tmp_path, "edit-post-write-rollback.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# assign_child / unassign_child: the assignment business-layer paths
+# ---------------------------------------------------------------------------
+
+
+def test_assign_child_multi_assignee_success(tmp_path) -> None:
+    """Assigning a second child leaves both as assignees (D-008)."""
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        bo = await children.create("Bo", NOW)
+        assignees = await assign_child(database, created.definition.id, bo.id)
+        assert sorted(c.id for c in assignees) == sorted([child.id, bo.id])
+        return assignees
+
+    _with_db(tmp_path, "assign-multi.db")(_body)
+
+
+def test_assign_child_duplicate_is_noop(tmp_path) -> None:
+    """Assigning an already-assigned child adds no duplicate row."""
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        assignees = await assign_child(
+            database, created.definition.id, child.id
+        )
+        assert [c.id for c in assignees] == [child.id]
+        again = await assign_child(
+            database, created.definition.id, child.id
+        )
+        assert [c.id for c in again] == [child.id]
+        return assignees
+
+    _with_db(tmp_path, "assign-duplicate.db")(_body)
+
+
+def test_assign_child_unknown_child_names_field(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(
+            ValueError, match="child_id.*does not exist"
+        ):
+            await assign_child(database, created.definition.id, 999)
+        return None
+
+    _with_db(tmp_path, "assign-unknown-child.db")(_body)
+
+
+def test_assign_child_inactive_child_names_field(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        bo = await children.create("Bo", NOW)
+        await children.set_active(bo.id, False)
+        with pytest.raises(ValueError, match="child_id.*inactive"):
+            await assign_child(database, created.definition.id, bo.id)
+        # The failed assignment leaves the roster unchanged.
+        assert [c.id for c in await definitions.list_assignees(
+            created.definition.id
+        )] == [child.id]
+        return None
+
+    _with_db(tmp_path, "assign-inactive-child.db")(_body)
+
+
+def test_assign_child_nonexistent_definition_names_field(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="definition_id.*does not exist"
+        ):
+            await assign_child(database, 999, child.id)
+        return None
+
+    _with_db(tmp_path, "assign-missing-definition.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
+def test_assign_child_rejects_non_int_definition_id(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="definition_id must be an integer"
+        ):
+            await assign_child(database, bad, child.id)
+        return None
+
+    _with_db(tmp_path, "assign-bad-definition-id.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
+def test_assign_child_rejects_non_int_child_id(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await assign_child(database, created.definition.id, bad)
+        return None
+
+    _with_db(tmp_path, "assign-bad-child-id.db")(_body)
+
+
+def test_unassign_child_success_returns_true(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        bo = await children.create("Bo", NOW)
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id, bo.id],
+            ["morning"],
+        )
+        assert await unassign_child(
+            database, created.definition.id, bo.id
+        ) is True
+        assert [c.id for c in await definitions.list_assignees(
+            created.definition.id
+        )] == [child.id]
+        return None
+
+    _with_db(tmp_path, "unassign-success.db")(_body)
+
+
+def test_unassign_child_not_assigned_returns_false(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        assert await unassign_child(
+            database, created.definition.id, 999
+        ) is False
+        return None
+
+    _with_db(tmp_path, "unassign-not-assigned.db")(_body)
+
+
+def test_unassign_child_nonexistent_definition_names_field(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="definition_id.*does not exist"
+        ):
+            await unassign_child(database, 999, child.id)
+        return None
+
+    _with_db(tmp_path, "unassign-missing-definition.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
+def test_unassign_child_rejects_non_int_definition_id(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="definition_id must be an integer"
+        ):
+            await unassign_child(database, bad, child.id)
+        return None
+
+    _with_db(tmp_path, "unassign-bad-definition-id.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
+def test_unassign_child_rejects_non_int_child_id(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await unassign_child(database, created.definition.id, bad)
+        return None
+
+    _with_db(tmp_path, "unassign-bad-child-id.db")(_body)
+
+
+def test_assign_and_unassign_leave_instances_and_history_alone(
+    tmp_path,
+) -> None:
+    """Assign/unassign write the assignees table only.
+
+    An already-generated instance keeps its generation-time child_id, and
+    no completion event appears — the assignment paths touch neither
+    ``quest_instances`` nor ``completion_events``.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        bo = await children.create("Bo", NOW)
+        await database.execute(
+            "INSERT INTO quest_instances (definition_id, child_id, "
+            "window, due_date, generated_at) VALUES (?, ?, 'morning', ?, ?)",
+            (created.definition.id, child.id, "2026-09-15", NOW),
+        )
+        await assign_child(database, created.definition.id, bo.id)
+        await unassign_child(database, created.definition.id, child.id)
+        row = await database.fetch_one(
+            "SELECT child_id FROM quest_instances WHERE definition_id = ?",
+            (created.definition.id,),
+        )
+        assert row == (child.id,)
+        events = await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        )
+        assert events == (0,)
+        return None
+
+    _with_db(tmp_path, "assign-unassign-history.db")(_body)
+
+
+def test_concurrent_assign_unassign_returns_consistent_roster(
+    tmp_path,
+) -> None:
+    """A concurrent unassign cannot strip the just-assigned child from
+    the roster that ``assign_child`` returns.
+
+    ``assign_child`` returns the roster read inside the same locked
+    transaction as the INSERT, so the child it assigned is always present
+    in its result — even when an unassign of that same child races.  The
+    pause hook holds ``list_assignees`` mid-flight so the unassign is
+    deterministically queued against the assignment's lock.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        bo = await children.create("Bo", NOW)
+
+        import custom_components.nestquest.dao_rules as dao_rules
+
+        original_list = dao_rules.QuestDefinitionsDao.list_assignees
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def _pausing_list(self, definition_id):
+            if not started.is_set():
+                started.set()
+                await release.wait()
+            return await original_list(self, definition_id)
+
+        dao_rules.QuestDefinitionsDao.list_assignees = _pausing_list
+        try:
+            assign_task = asyncio.ensure_future(
+                assign_child(database, created.definition.id, bo.id)
+            )
+            await started.wait()
+            unassign_task = asyncio.ensure_future(
+                unassign_child(database, created.definition.id, bo.id)
+            )
+            await asyncio.sleep(0)
+            release.set()
+            assignees, removed = await asyncio.gather(
+                assign_task, unassign_task
+            )
+        finally:
+            dao_rules.QuestDefinitionsDao.list_assignees = original_list
+
+        assert bo.id in [c.id for c in assignees], (
+            "assign_child must return a roster containing the child it "
+            "just assigned, even under a concurrent unassign"
+        )
+        assert removed is True
+        return assignees
+
+    _with_db(tmp_path, "assign-unassign-concurrent.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# set_quest_definition_active: deactivate / reactivate
+# ---------------------------------------------------------------------------
+
+
+def test_set_quest_definition_active_round_trip(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        definition_id = created.definition.id
+
+        deactivated = await set_quest_definition_active(
+            database, definition_id, False
+        )
+        assert isinstance(deactivated, CreatedQuestDefinition)
+        assert deactivated.definition.id == definition_id
+        assert deactivated.definition.is_active is False
+        # The decoded rule, assignees and windows form a consistent
+        # snapshot of the (unchanged) definition.
+        assert deactivated.rule == _daily_rule()
+        assert [c.id for c in deactivated.assignees] == [child.id]
+        assert [w.window for w in deactivated.windows] == ["morning"]
+        assert await definitions.list_active() == []
+
+        reactivated = await set_quest_definition_active(
+            database, definition_id, True
+        )
+        assert reactivated.definition.is_active is True
+        assert [d.id for d in await definitions.list_active()] == [
+            definition_id
+        ]
+        return reactivated
+
+    _with_db(tmp_path, "set-active-round-trip.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["1", "0", 1, 0, 1.0, None, "true"])
+def test_set_quest_definition_active_rejects_non_bool(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(
+            ValueError, match="is_active must be a real bool"
+        ):
+            await set_quest_definition_active(
+                database, created.definition.id, bad
+            )
+        # The refused transition left the definition active.
+        stored = await definitions.get(created.definition.id)
+        assert stored is not None
+        assert stored.is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-non-bool.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.5, "1", None])
+def test_set_quest_definition_active_rejects_non_int_definition_id(
+    tmp_path, bad
+) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(
+            ValueError, match="definition_id must be an integer"
+        ):
+            await set_quest_definition_active(database, bad, False)
+        stored = await definitions.get(created.definition.id)
+        assert stored is not None
+        assert stored.is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-non-int-id.db")(_body)
+
+
+def test_set_quest_definition_active_nonexistent_definition_names_field(
+    tmp_path,
+) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="definition_id.*does not exist"
+        ):
+            await set_quest_definition_active(database, 999, False)
+        return None
+
+    _with_db(tmp_path, "set-active-missing-definition.db")(_body)
+
+
+def test_set_quest_definition_active_never_deletes_or_touches_instances(
+    tmp_path,
+) -> None:
+    """Deactivation hides the definition but deletes nothing, and
+    pre-existing instances and completion history survive unchanged."""
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        definition_id = created.definition.id
+        rule_id = created.definition.schedule_rule_id
+
+        # Seed a real instance and a real completion event against the
+        # definition BEFORE deactivation, so the test can detect a
+        # delete or mutation of pre-existing data (not just an empty
+        # table staying empty).
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+        instance = await instances.upsert(
+            definition_id, child.id, _today_iso(), NOW, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "user",
+            NOW,
+            True,
+            actor_user_id="admin-1",
+        )
+
+        instance_count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        event_count = await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        )
+
+        await set_quest_definition_active(database, definition_id, False)
+
+        # The definition row survives (deactivated, not deleted).
+        stored = await definitions.get(definition_id)
+        assert stored is not None
+        assert stored.is_active is False
+        # Its rule, assignee and window rows all survive too.
+        assert await rules.get(rule_id) is not None
+        assert [c.id for c in await definitions.list_assignees(
+            definition_id
+        )] == [child.id]
+        assert [w.window for w in await definitions.list_windows(
+            definition_id
+        )] == ["morning"]
+        # The pre-existing instance and completion event survive with
+        # unchanged ids and contents, and no new rows appeared.
+        assert await instances.get_by_id(instance.id) == instance
+        assert await events.list_by_instance(instance.id) == [event]
+        assert await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        ) == instance_count
+        assert await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        ) == event_count
+
+        await set_quest_definition_active(database, definition_id, True)
+        assert (await definitions.get(definition_id)).is_active is True
+        assert await instances.get_by_id(instance.id) == instance
+        assert await events.list_by_instance(instance.id) == [event]
+        assert await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        ) == instance_count
+        assert await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        ) == event_count
+        return None
+
+    _with_db(tmp_path, "set-active-never-deletes.db")(_body)
+
+
+def _gated_hass(
+    armed: dict, gate_open: asyncio.Event, a_started: asyncio.Event
+):
+    """A hass mock whose executor gates the FIRST ``_execute`` job after
+    arming — the UPDATE of whichever ``set_quest_definition_active``
+    call starts first — proving the caller holds the connection lock at
+    that moment and the second caller must queue behind it."""
+    from unittest.mock import MagicMock
+
+    hass = MagicMock()
+
+    async def _gated_executor(fn, *args):
+        if armed["active"]:
+            if getattr(fn, "__name__", "") == "_execute" and not armed["gated"]:
+                armed["gated"] = True
+                a_started.set()
+                await gate_open.wait()
+        return fn(*args)
+
+    hass.async_add_executor_job = _gated_executor
+    return hass
+
+
+def test_set_quest_definition_active_concurrent_opposite_transitions_return_own(
+    tmp_path,
+) -> None:
+    """Racing deactivate/reactivate calls each report their own request.
+
+    Deterministic gate on task A's UPDATE: task B queues behind the
+    connection lock and lands the opposite transition after A.  Without
+    the lock, A's readback could see B's write and report B's value.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "active-race.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            child = await children.create("Ada", NOW)
+            created = await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                set_quest_definition_active(
+                    database, created.definition.id, False
+                )
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                set_quest_definition_active(
+                    database, created.definition.id, True
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            # Each caller gets back exactly the state it requested.
+            assert record_a.definition.is_active is False
+            assert record_b.definition.is_active is True
+            # Both snapshots stay coherent: the decoded rule and the
+            # assignees are those of the definition they were read from.
+            assert record_a.rule == _daily_rule()
+            assert record_b.rule == _daily_rule()
+            assert [c.id for c in record_a.assignees] == [child.id]
+            assert [c.id for c in record_b.assignees] == [child.id]
+            # B landed last: the stored row is B's (read via the DAO —
+            # the four-table leak guard forbids raw probes here).
+            stored = await QuestDefinitionsDao(database).get(
+                created.definition.id
+            )
+            assert stored is not None
+            assert stored.is_active is True
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+# ---------------------------------------------------------------------------
+# list query helpers: list_active_definitions / list_definitions_for_child /
+# list_definitions_firing_on
+# ---------------------------------------------------------------------------
+
+
+def test_list_active_definitions_returns_only_active_in_id_order(tmp_path) -> None:
+    """Only active definitions come back, in rising id order, each rich.
+
+    Two definitions are created (ids 1 and 2), then definition 1 is
+    deactivated.  ``list_active_definitions`` must return only definition
+    2, and every bundle must carry the decoded rule, assignees and
+    windows of its own definition.
+    """
+    async def _body(database, children, rules, definitions, child):
+        bo = await children.create("Bo", NOW)
+        first = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        weekly = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={0, 2},
+            start_date="2026-09-14",
+        )
+        second = await create_quest_definition(
+            database,
+            "Tidy room",
+            weekly,
+            [child.id, bo.id],
+            [("morning", "08:00"), ("evening", "19:00")],
+        )
+        assert [d.id for d in await definitions.list_active()] == [
+            first.definition.id,
+            second.definition.id,
+        ]
+
+        await set_quest_definition_active(database, first.definition.id, False)
+
+        result = await list_active_definitions(database)
+        assert [bundle.definition.id for bundle in result] == [second.definition.id]
+        assert isinstance(result[0], CreatedQuestDefinition)
+        assert result[0].rule == weekly
+        assert sorted(c.id for c in result[0].assignees) == sorted(
+            [child.id, bo.id]
+        )
+        by_name = {w.window: w.due_time for w in result[0].windows}
+        assert by_name == {"morning": "08:00", "evening": "19:00"}
+        return result
+
+    _with_db(tmp_path, "list-active.db")(_body)
+
+
+def test_list_definitions_for_child_returns_only_that_child(tmp_path) -> None:
+    """Only the given child's definitions come back, deterministically.
+
+    Definitions are created assigned to Ada (child), Bo, or both; the
+    helper must return exactly the definitions assigned to Ada, excluding
+    a Bo-only definition.  Inactive definitions are still assigned, so
+    they too come back (the helper filters by child, not by activity).
+    """
+    async def _body(database, children, rules, definitions, child):
+        bo = await children.create("Bo", NOW)
+        ada_only = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        both = await create_quest_definition(
+            database,
+            "Tidy room",
+            ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={1},
+                start_date="2026-09-14",
+            ),
+            [child.id, bo.id],
+            ["evening"],
+        )
+        bo_only = await create_quest_definition(
+            database, "Walk dog", _daily_rule(), [bo.id], ["afternoon"]
+        )
+
+        # The helper returns the given child's definitions in ascending
+        # id order, consistent with list_active_definitions.
+        result = await list_definitions_for_child(database, child.id)
+        assert [bundle.definition.id for bundle in result] == [
+            ada_only.definition.id,
+            both.definition.id,
+        ]
+        assert result[0].rule == _daily_rule()
+        assert [c.id for c in result[0].assignees] == [child.id]
+        assert [w.window for w in result[0].windows] == ["morning"]
+
+        bo_result = await list_definitions_for_child(database, bo.id)
+        assert [bundle.definition.id for bundle in bo_result] == [
+            both.definition.id,
+            bo_only.definition.id,
+        ]
+        return result
+
+    _with_db(tmp_path, "list-for-child.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.5, "1", None])
+def test_list_definitions_for_child_rejects_non_int_child_id(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await list_definitions_for_child(database, bad)
+        return None
+
+    _with_db(tmp_path, "list-for-child-bad-id.db")(_body)
+
+
+def test_list_definitions_firing_on_mixed_rules(tmp_path) -> None:
+    """A mixed daily + weekly + monthly set fires on the right dates.
+
+    Daily fires every day from its start; weekly fires on its weekdays;
+    monthly fires on its day-of-month.  Deactivated definitions never
+    fire, and the returned bundles carry the decoded rule.
+    """
+    async def _body(database, children, rules, definitions, child):
+        daily = await create_quest_definition(
+            database,
+            "Daily chore",
+            ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-14"),
+            [child.id],
+            ["morning"],
+        )
+        weekly = await create_quest_definition(
+            database,
+            "Weekly chore",
+            ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={0, 2},  # Monday, Wednesday
+                start_date="2026-09-14",
+            ),
+            [child.id],
+            ["evening"],
+        )
+        monthly = await create_quest_definition(
+            database,
+            "Monthly chore",
+            ScheduleRule(
+                rule_type=RuleType.MONTHLY_DAY,
+                day_of_month=15,
+                start_date="2026-09-14",
+            ),
+            [child.id],
+            ["afternoon"],
+        )
+
+        # 2026-09-14 is a Monday (weekday 0): daily + weekly fire;
+        # monthly (the 15th) does not.
+        hits = await list_definitions_firing_on(database, "2026-09-14")
+        assert [b.definition.id for b in hits] == [
+            daily.definition.id,
+            weekly.definition.id,
+        ]
+        # 2026-09-15 is a Tuesday (weekday 1, day 15): daily + monthly.
+        hits = await list_definitions_firing_on(database, "2026-09-15")
+        assert [b.definition.id for b in hits] == [
+            daily.definition.id,
+            monthly.definition.id,
+        ]
+        # 2026-09-16 is a Wednesday (weekday 2, day 16): daily + weekly.
+        hits = await list_definitions_firing_on(database, "2026-09-16")
+        assert [b.definition.id for b in hits] == [
+            daily.definition.id,
+            weekly.definition.id,
+        ]
+
+        # Deactivating a definition removes it from every future result.
+        await set_quest_definition_active(database, weekly.definition.id, False)
+        hits = await list_definitions_firing_on(database, "2026-09-14")
+        assert [b.definition.id for b in hits] == [daily.definition.id]
+        # The returned bundle carries the decoded rule, not the raw row.
+        assert hits[0].rule == ScheduleRule(
+            rule_type=RuleType.DAILY, start_date="2026-09-14"
+        )
+        return hits
+
+    _with_db(tmp_path, "list-firing-on.db")(_body)
+
+
+@pytest.mark.parametrize(
+    "bad",
+    ["2026-9-4", "2026-13-01", "2026-02-30", "not-a-date", None, 20260914],
+)
+def test_list_definitions_firing_on_rejects_malformed_date(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        with pytest.raises(ValueError, match="date must be"):
+            await list_definitions_firing_on(database, bad)
+        return None
+
+    _with_db(tmp_path, "list-firing-on-bad-date.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# full-lifecycle integration: one definition through every operation
+# ---------------------------------------------------------------------------
+
+
+def test_full_lifecycle_never_touches_instances_or_history(tmp_path) -> None:
+    """End-to-end lifecycle on one definition writes only the four
+    definition tables; ``quest_instances`` and ``completion_events`` are
+    left exactly as they were at every stage.
+
+    A real instance and completion event are seeded before the lifecycle
+    begins, so each stage is checked against pre-existing data being
+    mutated (not an empty table staying empty): create -> edit
+    rule/windows -> assign a second child -> unassign the original child
+    -> deactivate -> reactivate.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            [("morning", "07:00")],
+        )
+        definition_id = created.definition.id
+
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+        instance = await instances.upsert(
+            definition_id, child.id, _today_iso(), NOW, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "user",
+            NOW,
+            True,
+            actor_user_id="admin-1",
+        )
+        instance_count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        event_count = await database.fetch_one(
+            "SELECT COUNT(*) FROM completion_events"
+        )
+
+        async def _assert_history_untouched():
+            assert await instances.get_by_id(instance.id) == instance
+            assert await events.list_by_instance(instance.id) == [event]
+            assert await database.fetch_one(
+                "SELECT COUNT(*) FROM quest_instances"
+            ) == instance_count
+            assert await database.fetch_one(
+                "SELECT COUNT(*) FROM completion_events"
+            ) == event_count
+
+        await _assert_history_untouched()
+
+        # Edit rule and windows together; the definition's rule row is
+        # rewritten in place and its window set replaced.
+        edited = await edit_quest_definition(
+            database,
+            definition_id,
+            rule=ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={1, 3},
+                start_date="2026-09-14",
+            ),
+            windows=[("morning", "08:00"), ("evening", "19:30")],
+        )
+        assert edited.rule == ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={1, 3},
+            start_date="2026-09-14",
+        )
+        by_name = {w.window: w.due_time for w in edited.windows}
+        assert by_name == {"morning": "08:00", "evening": "19:30"}
+        await _assert_history_untouched()
+
+        # Assign a second child: multi-assignee, both present.
+        bo = await children.create("Bo", NOW)
+        roster = await assign_child(database, definition_id, bo.id)
+        assert sorted(c.id for c in roster) == sorted([child.id, bo.id])
+        await _assert_history_untouched()
+
+        # Unassign the original child: only the link drops.
+        assert await unassign_child(database, definition_id, child.id) is True
+        assert [
+            c.id for c in await definitions.list_assignees(definition_id)
+        ] == [bo.id]
+        await _assert_history_untouched()
+
+        # Deactivate, then reactivate: the definition survives both.
+        deactivated = await set_quest_definition_active(
+            database, definition_id, False
+        )
+        assert deactivated.definition.is_active is False
+        await _assert_history_untouched()
+        reactivated = await set_quest_definition_active(
+            database, definition_id, True
+        )
+        assert reactivated.definition.is_active is True
+        await _assert_history_untouched()
+
+        return reactivated
+
+    _with_db(tmp_path, "full-lifecycle.db")(_body)