fix: name assignee_child_ids in child validation errors (NQ67-ERR-001)

custom_components/nestquest/dao_rules.py

@@ -40,6 +40,7 @@ from .const import QUEST_WINDOWS
 from .dao_children import ChildRecord, _child_from_row, _connection_lock
 from .dao_children import _CHILD_COLUMNS
 from .db import NestQuestDatabase
+from .recurrence import RuleType, RuleValidationError, ScheduleRule
 
 #: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
 #: in update methods: passing ``None`` must mean clearing a nullable
@@ -119,6 +120,26 @@ class ScheduleRuleRecord:
     end_date: str | None
 
 
+@dataclass(frozen=True)
+class ScheduleRuleStorage:
+    """The eight value columns a ScheduleRule maps to (no id column).
+
+    ``id`` is assigned by the database on insert, so the model<->storage
+    mapping concerns only the value columns.  ``nth_weekday_weekday`` is
+    deliberately absent here: for MONTHLY_WEEKDAY its value is folded
+    into ``weekday_set`` as that column's single CSV element.
+    """
+
+    rule_type: str
+    interval: int
+    weekday_set: str | None
+    day_of_month: int | None
+    nth_weekday: int | None
+    month: int | None
+    start_date: str
+    end_date: str | None
+
+
 @dataclass(frozen=True)
 class QuestDefinitionRecord:
     """One row of ``quest_definitions`` (without its assignees)."""
@@ -142,6 +163,22 @@ class QuestDefinitionWindowRecord:
     due_time: str | None
 
 
+@dataclass(frozen=True)
+class QuestDefinitionSnapshot:
+    """A definition plus its assignees and windows, read atomically.
+
+    Returned by :meth:`QuestDefinitionsDao.create_with_rule_and_windows`
+    with the assignees and windows read inside the SAME transaction that
+    inserted them, so the snapshot is a consistent creation-time view —
+    a concurrent assignee or window mutation cannot change it between
+    the write and the read.
+    """
+
+    definition: QuestDefinitionRecord
+    assignees: list[ChildRecord]
+    windows: list[QuestDefinitionWindowRecord]
+
+
 _RULE_COLUMNS = (
     "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
     "month, start_date, end_date"
@@ -188,6 +225,228 @@ def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
     )
 
 
+#: Storage columns that are forbidden for each rule type.  A populated
+#: forbidden column means the row's shape is invalid and must raise on
+#: read, never be silently dropped.  ``weekday_set`` is absent for
+#: DAILY/MONTHLY_DAY/YEARLY; MONTHLY_WEEKDAY is the only shape allowed a
+#: non-NULL ``nth_weekday``; YEARLY is the only shape allowed ``month``.
+_FORBIDDEN_STORAGE_FIELDS: dict[str, tuple[RuleType, ...]] = {
+    "weekday_set": (RuleType.DAILY, RuleType.MONTHLY_DAY, RuleType.YEARLY),
+    "day_of_month": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_WEEKDAY,
+    ),
+    "nth_weekday": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_DAY,
+        RuleType.YEARLY,
+    ),
+    "month": (
+        RuleType.DAILY,
+        RuleType.WEEKLY,
+        RuleType.CUSTOM_DAYS,
+        RuleType.MONTHLY_DAY,
+        RuleType.MONTHLY_WEEKDAY,
+    ),
+}
+
+
+def schedule_rule_to_storage(rule: ScheduleRule) -> ScheduleRuleStorage:
+    """Map a validated ScheduleRule onto its storage columns.
+
+    ``rule_type`` stores as ``RuleType.storage_value`` (MONTHLY_DAY and
+    MONTHLY_WEEKDAY both store ``'monthly'``); ``weekday_set`` stores as
+    a sorted CSV list.  MONTHLY_WEEKDAY folds its model-only
+    ``nth_weekday_weekday`` into ``weekday_set`` as that list's single
+    element; every other type leaves the column as its own set
+    (WEEKLY/CUSTOM_DAYS) or NULL.
+    """
+    shape = rule.rule_type
+    if shape is RuleType.MONTHLY_WEEKDAY:
+        weekday_set = str(rule.nth_weekday_weekday)
+    elif shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
+        weekday_set = ",".join(str(w) for w in sorted(rule.weekday_set))
+    else:
+        weekday_set = None
+    return ScheduleRuleStorage(
+        rule_type=shape.storage_value,
+        interval=rule.interval,
+        weekday_set=weekday_set,
+        day_of_month=rule.day_of_month,
+        nth_weekday=rule.nth_weekday,
+        month=rule.month,
+        start_date=rule.start_date,
+        end_date=rule.end_date,
+    )
+
+
+def schedule_rule_from_storage(storage: ScheduleRuleStorage) -> ScheduleRule:
+    """Reconstruct a ScheduleRule from storage columns.
+
+    Disambiguates MONTHLY_DAY from MONTHLY_WEEKDAY by which field is
+    populated, and raises RuleValidationError rather than silently
+    mis-decoding an ambiguous or invalid shape.
+    """
+    rule_type = storage.rule_type
+    if rule_type == "daily":
+        shape = RuleType.DAILY
+    elif rule_type == "weekly":
+        shape = RuleType.WEEKLY
+    elif rule_type == "monthly":
+        shape = _disambiguate_monthly(storage)
+    elif rule_type == "yearly":
+        shape = RuleType.YEARLY
+    elif rule_type == "custom":
+        shape = RuleType.CUSTOM_DAYS
+    else:
+        raise RuleValidationError(
+            f"unknown storage rule_type {rule_type!r}"
+        )
+
+    _reject_forbidden_storage_fields(storage, shape)
+
+    weekday_set = None
+    nth_weekday_weekday = None
+    if shape is RuleType.WEEKLY or shape is RuleType.CUSTOM_DAYS:
+        weekday_set = _weekdays_from_csv(storage.weekday_set)
+    elif shape is RuleType.MONTHLY_WEEKDAY:
+        nth_weekday_weekday = _single_weekday_from_csv(storage.weekday_set)
+
+    return ScheduleRule(
+        rule_type=shape,
+        interval=storage.interval,
+        weekday_set=weekday_set,
+        day_of_month=storage.day_of_month,
+        nth_weekday=storage.nth_weekday,
+        nth_weekday_weekday=nth_weekday_weekday,
+        month=storage.month,
+        start_date=storage.start_date,
+        end_date=storage.end_date,
+    )
+
+
+def _reject_forbidden_storage_fields(
+    storage: ScheduleRuleStorage, shape: RuleType
+) -> None:
+    """Raise unless every populated column is permitted for ``shape``.
+
+    The model constructor rejects a forbidden ``day_of_month``,
+    ``nth_weekday``, or ``month``, but ``weekday_set`` never reaches it
+    (it is folded for MONTHLY_WEEKDAY and dropped otherwise), so a daily
+    row carrying ``weekday_set="0"`` would otherwise decode as a valid
+    daily rule and silently lose the stored data.  This validates every
+    shape column explicitly so a populated-forbidden column raises
+    instead of being discarded.
+    """
+    for field, forbidden_shapes in _FORBIDDEN_STORAGE_FIELDS.items():
+        value = getattr(storage, field)
+        if value is not None and shape in forbidden_shapes:
+            raise RuleValidationError(
+                f"{shape.value} storage rows must not set {field}, "
+                f"got {value!r}"
+            )
+
+
+def _disambiguate_monthly(storage: ScheduleRuleStorage) -> RuleType:
+    """Resolve the shared 'monthly' storage value to a model type.
+
+    MONTHLY_DAY names ``day_of_month`` only; MONTHLY_WEEKDAY names
+    ``nth_weekday`` only.  Neither, or both, is ambiguous and raises.
+    """
+    day_of_month = storage.day_of_month
+    nth_weekday = storage.nth_weekday
+    if day_of_month is not None and nth_weekday is None:
+        return RuleType.MONTHLY_DAY
+    if nth_weekday is not None and day_of_month is None:
+        return RuleType.MONTHLY_WEEKDAY
+    raise RuleValidationError(
+        "storage rule_type 'monthly' is ambiguous: must name exactly one "
+        "of day_of_month or nth_weekday, got "
+        f"day_of_month={day_of_month!r}, nth_weekday={nth_weekday!r}"
+    )
+
+
+def _weekdays_from_csv(value: str | None) -> frozenset[int]:
+    """Parse a weekday CSV column into a frozenset of ints.
+
+    ``None`` and ``""`` both yield an empty frozenset (the caller decides
+    whether empty is legal); a non-integer token raises
+    RuleValidationError rather than leaking a raw ValueError.
+    """
+    if value is None or value == "":
+        return frozenset()
+    try:
+        entries = [int(token) for token in value.split(",")]
+    except ValueError:
+        raise RuleValidationError(
+            f"weekday_set must be a CSV of integers, got {value!r}"
+        ) from None
+    return frozenset(entries)
+
+
+def _single_weekday_from_csv(value: str | None) -> int:
+    """The one weekday a MONTHLY_WEEKDAY stores in weekday_set.
+
+    Parses the CSV into tokens FIRST and requires exactly one token,
+    so a duplicate pair like ``"1,1"`` is rejected rather than silently
+    canonicalized to a single weekday on re-storage.  An empty or
+    multi-element column raises RuleValidationError.
+    """
+    if value is None or value == "":
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
+            f"weekday, got {value!r}"
+        )
+    tokens = value.split(",")
+    if len(tokens) != 1:
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must hold exactly one "
+            f"weekday, got {value!r}"
+        )
+    try:
+        return int(tokens[0])
+    except ValueError:
+        raise RuleValidationError(
+            "MONTHLY_WEEKDAY storage weekday_set must be an integer, "
+            f"got {value!r}"
+        ) from None
+
+
+async def _insert_rule_row(
+    database: NestQuestDatabase, storage: ScheduleRuleStorage
+) -> int:
+    """Insert one ``schedule_rules`` row and return its new id.
+
+    Runs inside the caller's transaction (no lock or transaction of its
+    own) so a multi-table write can persist the rule together with the
+    rows that reference it.  The caller is responsible for validating
+    the storage fields — :meth:`ScheduleRulesDao.create` validates the
+    date shape, and :meth:`QuestDefinitionsDao.create_with_rule_and_windows`
+    receives storage already mapped from a validated
+    :class:`~.recurrence.ScheduleRule`.
+    """
+    result = await database.execute(
+        "INSERT INTO schedule_rules (rule_type, interval, weekday_set, "
+        "day_of_month, nth_weekday, month, start_date, end_date) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            storage.rule_type,
+            storage.interval,
+            storage.weekday_set,
+            storage.day_of_month,
+            storage.nth_weekday,
+            storage.month,
+            storage.start_date,
+            storage.end_date,
+        ),
+    )
+    return result.lastrowid
+
+
 class ScheduleRulesDao:
     """Typed async access to the ``schedule_rules`` table."""
 
@@ -218,22 +477,20 @@ class ScheduleRulesDao:
         _validate_date(start_date, "start_date")
         if end_date is not None:
             _validate_date(end_date, "end_date")
-        result = await self._database.execute(
-            "INSERT INTO schedule_rules (rule_type, interval, weekday_set, "
-            "day_of_month, nth_weekday, month, start_date, end_date) "
-            "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
-            (
-                rule_type,
-                interval,
-                weekday_set,
-                day_of_month,
-                nth_weekday,
-                month,
-                start_date,
-                end_date,
+        rule_id = await _insert_rule_row(
+            self._database,
+            ScheduleRuleStorage(
+                rule_type=rule_type,
+                interval=interval,
+                weekday_set=weekday_set,
+                day_of_month=day_of_month,
+                nth_weekday=nth_weekday,
+                month=month,
+                start_date=start_date,
+                end_date=end_date,
             ),
         )
-        rule = await self.get(result.lastrowid)
+        rule = await self.get(rule_id)
         assert rule is not None
         return rule
 
@@ -393,6 +650,82 @@ class QuestDefinitionsDao:
         assert definition is not None
         return definition
 
+    async def create_with_rule_and_windows(
+        self,
+        title: str,
+        schedule_rule: ScheduleRuleStorage,
+        created_at: str,
+        assignee_child_ids: list[int],
+        windows: list[tuple[str, str | None]],
+        *,
+        description: str | None = None,
+        icon: str | None = None,
+    ) -> QuestDefinitionSnapshot:
+        """Insert a rule, definition, assignees and windows atomically.
+
+        The whole write — the schedule rule row, the definition row,
+        every assignee link and every window declaration — runs inside
+        ONE transaction under the connection lock, so a validation or
+        write failure rolls the lot back together (all-or-nothing): the
+        caller can never observe a definition with a missing rule,
+        assignee or window, and a rejected create leaves no partial
+        rows.
+
+        Returns a :class:`QuestDefinitionSnapshot` whose assignees and
+        windows are read back inside the same transaction, so the
+        returned view cannot be raced by a concurrent assignee/window
+        mutation after the write.
+
+        The rule arrives as pre-validated storage fields (mapped from a
+        :class:`~.recurrence.ScheduleRule` via
+        :func:`schedule_rule_to_storage`); its dates are re-validated
+        here so a hand-built storage can never persist a malformed date.
+        The definition-level ``due_time`` column is deliberately left
+        NULL — per-window due times supersede it (D-008).  Assignees are
+        validated as existing-and-active at insert time, and windows are
+        validated for name and strict HH:MM due time, matching
+        :meth:`upsert_window`.
+        """
+        _validate_date(schedule_rule.start_date, "start_date")
+        if schedule_rule.end_date is not None:
+            _validate_date(schedule_rule.end_date, "end_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                for child_id in assignee_child_ids:
+                    await self._validate_assignable_child(child_id)
+                rule_id = await _insert_rule_row(self._database, schedule_rule)
+                result = await self._database.execute(
+                    "INSERT INTO quest_definitions (title, description, "
+                    "icon, schedule_rule_id, due_time, is_active, "
+                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (title, description, icon, rule_id, None, 1, created_at),
+                )
+                definition_id = result.lastrowid
+                for child_id in assignee_child_ids:
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_assignees "
+                        "(definition_id, child_id) VALUES (?, ?)",
+                        (definition_id, child_id),
+                    )
+                for window, due_time in windows:
+                    _validate_window(window)
+                    if due_time is not None:
+                        _validate_time(due_time, "due_time")
+                    await self._database.execute(
+                        "INSERT INTO quest_definition_windows "
+                        "(definition_id, window, due_time) VALUES (?, ?, ?)",
+                        (definition_id, window, due_time),
+                    )
+                definition = await self.get(definition_id)
+                assignees = await self.list_assignees(definition_id)
+                window_records = await self.list_windows(definition_id)
+        assert definition is not None
+        return QuestDefinitionSnapshot(
+            definition=definition,
+            assignees=assignees,
+            windows=window_records,
+        )
+
     async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
         """Raise ValueError unless the rule exists.
 

custom_components/nestquest/quest_definitions.py

@@ -0,0 +1,228 @@
+"""Business layer for creating quest definitions (Feature 06).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_rules.QuestDefinitionsDao`: callers get validation and the
+all-or-nothing create here, and never touch the DAO SQL directly.
+:func:`create_quest_definition` is the create path; there is no
+permission check here (that is Feature 09).
+
+Validation policy mirrors :mod:`.children`:
+
+- ``title`` is required and trimmed; empty or whitespace-only raises
+  ValueError.
+- ``rule`` must be an already-validated :class:`~.recurrence.ScheduleRule`
+  (its constructor is total — rule validation is NOT re-implemented
+  here).
+- ``assignee_child_ids`` must be a non-empty list (strictly a ``list``);
+  every id is a plain int (bools and floats rejected, since SQLite would
+  bind them onto a real child) and must reference an existing ACTIVE
+  child (the DAO enforces the active/exists check inside the same
+  transaction).  Duplicate ids are rejected here, before the composite
+  primary key could surface them as an :class:`sqlite3.IntegrityError`.
+- ``windows`` must be a non-empty list (strictly a ``list``); each entry
+  is a window name from :data:`~.const.QUEST_WINDOWS` — either a bare
+  name (no due time) or a ``(name, due_time)`` pair whose due time is a
+  strict 24-hour HH:MM string.  Duplicate window names are rejected
+  here, before the composite primary key could surface them as an
+  :class:`sqlite3.IntegrityError`.
+
+The rule, definition, assignees and windows are persisted in ONE
+transaction (the DAO's :meth:`~.dao_rules.QuestDefinitionsDao.create_with_rule_and_windows`),
+so a rejected or failed create persists nothing.  The definition-level
+``due_time`` column is superseded by per-window due times (D-008) and is
+never set here.  ``created_at`` is stamped UTC ISO-8601 at second
+precision, matching :mod:`.children`.
+"""
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+
+from .dao_children import ChildRecord
+from .dao_rules import (
+    QuestDefinitionRecord,
+    QuestDefinitionWindowRecord,
+    QuestDefinitionsDao,
+    _validate_time,
+    _validate_window,
+    schedule_rule_from_storage,
+    schedule_rule_to_storage,
+)
+from .db import NestQuestDatabase
+from .recurrence import ScheduleRule
+
+
+@dataclass(frozen=True)
+class CreatedQuestDefinition:
+    """The persisted result of :func:`create_quest_definition`.
+
+    ``definition`` is the stored row; ``rule`` is the schedule rule
+    decoded back through the storage mapping; ``assignees`` and
+    ``windows`` are the definition's linked child records and window
+    declarations, read through the DAO.
+    """
+
+    definition: QuestDefinitionRecord
+    rule: ScheduleRule
+    assignees: list[ChildRecord]
+    windows: list[QuestDefinitionWindowRecord]
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _validate_text(
+    value: object, field: str, *, required: bool = False
+) -> str | None:
+    """Validate a free-text field and return it trimmed.
+
+    ``None`` raises when ``required`` and otherwise returns None.  A
+    non-string value is rejected; an empty result is rejected for
+    required fields.
+    """
+    if value is None:
+        if required:
+            raise ValueError(f"{field} is required")
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{field} must be a string, got {value!r}")
+    trimmed = value.strip()
+    if required and not trimmed:
+        raise ValueError(f"{field} is required")
+    return trimmed
+
+
+def _validate_assignee_ids(value: object) -> list[int]:
+    """Return the assignee ids as plain ints, or raise naming the field.
+
+    The argument must be a non-empty list — strictly a ``list``: tuples
+    and other sequences are rejected.  Each entry must be a real int
+    (bools and floats rejected) so SQLite cannot bind ``True`` onto
+    child 1 or a float onto a neighbouring profile.
+    """
+    if type(value) is not list:
+        raise ValueError(
+            f"assignee_child_ids must be a list, got {type(value).__name__}"
+        )
+    if not value:
+        raise ValueError("assignee_child_ids must not be empty")
+    ids: list[int] = []
+    for entry in value:
+        if isinstance(entry, bool) or not isinstance(entry, int):
+            raise ValueError(
+                "assignee_child_ids must contain integer child ids, "
+                f"got {entry!r}"
+            )
+        ids.append(entry)
+    if len(set(ids)) != len(ids):
+        raise ValueError(
+            "assignee_child_ids must not contain duplicate child ids"
+        )
+    return ids
+
+
+def _normalize_windows(
+    value: object,
+) -> list[tuple[str, str | None]]:
+    """Normalize the windows argument to ``(name, due_time)`` pairs.
+
+    The outer argument must be a non-empty list — strictly a ``list``:
+    tuples and other sequences are rejected, while each individual entry
+    may still be a ``(window, due_time)`` tuple.  A bare window name
+    (due time None) is also accepted.  The name must be a
+    :data:`~.const.QUEST_WINDOWS` spelling and a present due time must be
+    a strict 24-hour HH:MM, both validated up front so a malformed
+    window is rejected before any write.
+    """
+    if type(value) is not list:
+        raise ValueError(
+            f"windows must be a list, got {type(value).__name__}"
+        )
+    if not value:
+        raise ValueError("windows must not be empty")
+    normalized: list[tuple[str, str | None]] = []
+    seen: set[str] = set()
+    for index, entry in enumerate(value):
+        if isinstance(entry, str):
+            name: object = entry
+            due_time: object = None
+        elif isinstance(entry, (tuple, list)) and len(entry) == 2:
+            name, due_time = entry[0], entry[1]
+        else:
+            raise ValueError(
+                f"windows[{index}] must be a window name or a "
+                f"(name, due_time) pair, got {entry!r}"
+            )
+        _validate_window(name)
+        if due_time is not None:
+            _validate_time(due_time, "due_time")
+        if name in seen:
+            raise ValueError(
+                f"windows must not contain duplicate window {name!r}"
+            )
+        seen.add(name)
+        normalized.append((name, due_time))
+    return normalized
+
+
+async def create_quest_definition(
+    database: NestQuestDatabase,
+    title: str,
+    rule: ScheduleRule,
+    assignee_child_ids: list[int],
+    windows: list[str | tuple[str, str | None]],
+    *,
+    description: str | None = None,
+    icon: str | None = None,
+) -> CreatedQuestDefinition:
+    """Create a quest definition (rule + assignees + windows) atomically.
+
+    Validates the arguments up front, then persists the schedule rule,
+    the definition, its assignees and its windows in one transaction via
+    the DAO.  Returns the stored definition together with its decoded
+    rule, assignees and windows — all read back inside that same
+    transaction, so the returned bundle is a consistent creation
+    snapshot.  Raises ValueError on any rejected argument; a rejected or
+    failed create persists nothing.
+    """
+    name = _validate_text(title, "title", required=True)
+    assert name is not None
+    if not isinstance(rule, ScheduleRule):
+        raise ValueError(f"rule must be a ScheduleRule, got {rule!r}")
+    assignee_ids = _validate_assignee_ids(assignee_child_ids)
+    window_specs = _normalize_windows(windows)
+    description_value = _validate_text(description, "description")
+    icon_value = _validate_text(icon, "icon")
+
+    storage = schedule_rule_to_storage(rule)
+    dao = QuestDefinitionsDao(database)
+    try:
+        snapshot = await dao.create_with_rule_and_windows(
+            name,
+            storage,
+            _now_stamp(),
+            assignee_ids,
+            window_specs,
+            description=description_value,
+            icon=icon_value,
+        )
+    except ValueError as error:
+        # The DAO's assignee check names only the child; re-raise so the
+        # public contract names the field the caller actually passed.
+        message = str(error)
+        if message.startswith("child "):
+            raise ValueError(
+                f"assignee_child_ids: {message}"
+            ) from error
+        raise
+    decoded_rule = schedule_rule_from_storage(storage)
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=decoded_rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
+    )

tests/test_dao_rules.py

@@ -9,13 +9,21 @@ import pytest
 from custom_components.nestquest.dao_children import ChildrenDao
 from custom_components.nestquest.dao_rules import (
     ScheduleRuleRecord,
+    ScheduleRuleStorage,
     ScheduleRulesDao,
     QuestDefinitionRecord,
     QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
+    schedule_rule_from_storage,
+    schedule_rule_to_storage,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
+from custom_components.nestquest.recurrence import (
+    RuleType,
+    RuleValidationError,
+    ScheduleRule,
+)
 
 
 def _run(coro):
@@ -423,6 +431,249 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
     assert create_won is False
 
 
+# ---------------------------------------------------------------------------
+# ScheduleRule <-> storage mapping
+# ---------------------------------------------------------------------------
+
+
+_RULE_STORAGE_CASES = [
+    (
+        ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-01"),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.DAILY,
+            interval=3,
+            start_date="2026-09-01",
+            end_date="2026-12-31",
+        ),
+        ScheduleRuleStorage(
+            "daily", 3, None, None, None, None, "2026-09-01", "2026-12-31"
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2, 0, 4},
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2,4", None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.CUSTOM_DAYS,
+            weekday_set={1, 3, 5},
+            interval=2,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "custom", 2, "1,3,5", None, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_DAY,
+            day_of_month=15,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_DAY,
+            day_of_month=31,
+            interval=3,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 3, None, 31, None, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_WEEKDAY,
+            nth_weekday=2,
+            nth_weekday_weekday=1,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, "1", None, 2, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.MONTHLY_WEEKDAY,
+            nth_weekday=-1,
+            nth_weekday_weekday=4,
+            interval=2,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "monthly", 2, "4", None, -1, None, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.YEARLY, month=6, start_date="2026-09-01"
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, None, None, 6, "2026-09-01", None
+        ),
+    ),
+    (
+        ScheduleRule(
+            rule_type=RuleType.YEARLY,
+            month=2,
+            day_of_month=29,
+            start_date="2026-09-01",
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, 29, None, 2, "2026-09-01", None
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_to_storage_maps_each_type(rule, storage) -> None:
+    assert schedule_rule_to_storage(rule) == storage
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_from_storage_rebuilds_each_type(rule, storage) -> None:
+    assert schedule_rule_from_storage(storage) == rule
+
+
+@pytest.mark.parametrize("rule, storage", _RULE_STORAGE_CASES)
+def test_schedule_rule_round_trips_losslessly_both_directions(
+    rule, storage
+) -> None:
+    assert schedule_rule_from_storage(schedule_rule_to_storage(rule)) == rule
+    assert schedule_rule_to_storage(schedule_rule_from_storage(storage)) == storage
+
+
+@pytest.mark.parametrize(
+    "storage",
+    [
+        # monthly with neither day_of_month nor nth_weekday
+        ScheduleRuleStorage(
+            "monthly", 1, None, None, None, None, "2026-09-01", None
+        ),
+        # monthly with both day_of_month and nth_weekday
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, 2, None, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY whose weekday_set is None
+        ScheduleRuleStorage(
+            "monthly", 1, None, None, 2, None, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY whose weekday_set is empty
+        ScheduleRuleStorage("monthly", 1, "", None, 2, None, "2026-09-01", None),
+        # MONTHLY_WEEKDAY whose weekday_set is multi-element
+        ScheduleRuleStorage(
+            "monthly", 1, "1,3", None, 2, None, "2026-09-01", None
+        ),
+        # unknown storage rule_type
+        ScheduleRuleStorage(
+            "sometimes", 1, None, None, None, None, "2026-09-01", None
+        ),
+    ],
+)
+def test_schedule_rule_from_storage_rejects_ambiguous_or_invalid(
+    storage,
+) -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(storage)
+
+
+def test_schedule_rule_from_storage_rejects_malformed_weekday_csv() -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "weekly", 1, "0,2,x", None, None, None, "2026-09-01", None
+            )
+        )
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "weekly", 1, "7", None, None, None, "2026-09-01", None
+            )
+        )
+
+
+@pytest.mark.parametrize(
+    "storage",
+    [
+        # DAILY carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "daily", 1, "0", None, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, 2, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "daily", 1, None, None, None, 6, "2026-09-01", None
+        ),
+        # WEEKLY carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "weekly", 1, "0,2", None, None, 6, "2026-09-01", None
+        ),
+        # CUSTOM_DAYS carrying shape columns it must not hold.
+        ScheduleRuleStorage(
+            "custom", 1, "0,2", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "custom", 1, "0,2", None, None, 6, "2026-09-01", None
+        ),
+        # MONTHLY_DAY carrying weekday_set (folded-only column) or month.
+        ScheduleRuleStorage(
+            "monthly", 1, "1", 15, None, None, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "monthly", 1, None, 15, None, 6, "2026-09-01", None
+        ),
+        # MONTHLY_WEEKDAY carrying a forbidden month.
+        ScheduleRuleStorage(
+            "monthly", 1, "1", None, 2, 6, "2026-09-01", None
+        ),
+        # YEARLY carrying weekday_set or nth_weekday.
+        ScheduleRuleStorage(
+            "yearly", 1, "0", None, None, 6, "2026-09-01", None
+        ),
+        ScheduleRuleStorage(
+            "yearly", 1, None, None, 2, 6, "2026-09-01", None
+        ),
+    ],
+)
+def test_schedule_rule_from_storage_rejects_forbidden_fields(storage) -> None:
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(storage)
+
+
+def test_schedule_rule_from_storage_rejects_duplicate_weekday_token() -> None:
+    """A MONTHLY_WEEKDAY with a duplicate CSV token ("1,1") must raise,
+    never silently canonicalize to a single weekday on re-storage."""
+    with pytest.raises(RuleValidationError):
+        schedule_rule_from_storage(
+            ScheduleRuleStorage(
+                "monthly", 1, "1,1", None, 2, None, "2026-09-01", None
+            )
+        )
+
+
 # ---------------------------------------------------------------------------
 # quest_definitions: create / get
 # ---------------------------------------------------------------------------

tests/test_quest_definitions.py

@@ -0,0 +1,463 @@
+"""Tests for quest_definitions.py: the Feature 06 create business layer."""
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_rules import (
+    QuestDefinitionsDao,
+    ScheduleRulesDao,
+    schedule_rule_to_storage,
+)
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+from custom_components.nestquest.quest_definitions import (
+    CreatedQuestDefinition,
+    create_quest_definition,
+)
+from custom_components.nestquest.recurrence import RuleType, ScheduleRule
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    from unittest.mock import AsyncMock, MagicMock
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+NOW = "2026-09-14T12:00:00+00:00"
+
+
+async def _prepare(path) -> tuple:
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    children = ChildrenDao(database)
+    rules = ScheduleRulesDao(database)
+    definitions = QuestDefinitionsDao(database)
+    child = await children.create("Ada", NOW)
+    return database, children, rules, definitions, child
+
+
+def _with_db(tmp_path, name):
+    """Run an async body against a prepared database, closing it after."""
+
+    def _run_test(body):
+        async def _main():
+            database, children, rules, definitions, child = await _prepare(
+                tmp_path / name
+            )
+            try:
+                return await body(database, children, rules, definitions, child)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+def _daily_rule() -> ScheduleRule:
+    return ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-09-14")
+
+
+# ---------------------------------------------------------------------------
+# create_quest_definition: successful creates
+# ---------------------------------------------------------------------------
+
+
+def test_create_one_child_one_window(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        rule = _daily_rule()
+        result = await create_quest_definition(
+            database,
+            "  Brush teeth  ",
+            rule,
+            [child.id],
+            ["morning"],
+            description="  Twice daily  ",
+            icon="  mdi:tooth  ",
+        )
+        assert isinstance(result, CreatedQuestDefinition)
+        definition = result.definition
+        assert definition.title == "Brush teeth"
+        assert definition.description == "Twice daily"
+        assert definition.icon == "mdi:tooth"
+        assert definition.is_active is True
+        # D-008: the definition-level due_time column is never set here.
+        assert definition.due_time is None
+        # Timestamp is stamped UTC at second precision by the business layer.
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00",
+            definition.created_at,
+        )
+        # The rule round-trips through the storage mapping.
+        assert result.rule == rule
+        assert definition.schedule_rule_id is not None
+        rule_row = await rules.get(definition.schedule_rule_id)
+        assert rule_row is not None
+        assert rule_row.rule_type == "daily"
+        # Assignees and windows are read back through the DAO.
+        assert [c.id for c in result.assignees] == [child.id]
+        assert [w.window for w in result.windows] == ["morning"]
+        assert result.windows[0].due_time is None
+        return result
+
+    _with_db(tmp_path, "create-one.db")(_body)
+
+
+def test_create_several_children_several_windows_with_due_times(
+    tmp_path,
+) -> None:
+    async def _body(database, children, rules, definitions, child):
+        bo = await children.create("Bo", NOW)
+        cleo = await children.create("Cleo", NOW)
+        rule = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={0, 2, 4},
+            start_date="2026-09-14",
+        )
+        result = await create_quest_definition(
+            database,
+            "Chore",
+            rule,
+            [child.id, bo.id, cleo.id],
+            [("morning", "09:00"), ("evening", "19:30")],
+        )
+        assert sorted(c.id for c in result.assignees) == sorted(
+            [child.id, bo.id, cleo.id]
+        )
+        by_name = {w.window: w.due_time for w in result.windows}
+        assert by_name == {"morning": "09:00", "evening": "19:30"}
+        assert result.rule == rule
+        stored = await definitions.get(result.definition.id)
+        assert stored is not None
+        assert stored.due_time is None
+        return result
+
+    _with_db(tmp_path, "create-many.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# create_quest_definition: argument rejections
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("bad", ["", "   ", "\t\n", None])
+def test_create_rejects_empty_title(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="title is required"):
+            await create_quest_definition(
+                database, bad, _daily_rule(), [child.id], ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-title.db")(_body)
+
+
+def test_create_rejects_non_string_title(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="title must be a string"):
+            await create_quest_definition(
+                database, 42, _daily_rule(), [child.id], ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-title-type.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [None, "daily", 42, {"rule_type": "daily"}])
+def test_create_rejects_non_schedule_rule(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="rule must be a ScheduleRule"):
+            await create_quest_definition(
+                database, "Brush teeth", bad, [child.id], ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-rule.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [[]])
+def test_create_rejects_empty_assignees(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="assignee_child_ids must not be empty"
+        ):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), bad, ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-empty-assignees.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [(1,), (1, 2), "1", {1}])
+def test_create_rejects_non_list_assignees(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="assignee_child_ids must be a list"
+        ):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), bad, ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-non-list-assignees.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.5, "1", None])
+def test_create_rejects_non_int_assignee(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="assignee_child_ids must contain integer"
+        ):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [bad], ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-assignee.db")(_body)
+
+
+def test_create_rejects_duplicate_assignees(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="assignee_child_ids must not contain duplicate"
+        ):
+            await create_quest_definition(
+                database,
+                "Brush teeth",
+                _daily_rule(),
+                [child.id, child.id],
+                ["morning"],
+            )
+        return None
+
+    _with_db(tmp_path, "create-duplicate-assignees.db")(_body)
+
+
+def test_create_rejects_inactive_child(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        inactive = await children.create("Bo", NOW)
+        await children.set_active(inactive.id, False)
+        with pytest.raises(
+            ValueError, match="assignee_child_ids.*inactive"
+        ):
+            await create_quest_definition(
+                database,
+                "Brush teeth",
+                _daily_rule(),
+                [inactive.id],
+                ["morning"],
+            )
+        return None
+
+    _with_db(tmp_path, "create-inactive-child.db")(_body)
+
+
+def test_create_rejects_unknown_child(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="assignee_child_ids.*does not exist"
+        ):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [999], ["morning"]
+            )
+        return None
+
+    _with_db(tmp_path, "create-unknown-child.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [[]])
+def test_create_rejects_empty_windows(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="windows must not be empty"):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [child.id], bad
+            )
+        return None
+
+    _with_db(tmp_path, "create-empty-windows.db")(_body)
+
+
+@pytest.mark.parametrize(
+    "bad", [("morning",), ("morning", "afternoon"), "morning", {"morning"}]
+)
+def test_create_rejects_non_list_windows(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="windows must be a list"):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [child.id], bad
+            )
+        return None
+
+    _with_db(tmp_path, "create-non-list-windows.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [["noon"], [("MORNING", None)], [("", None)]])
+def test_create_rejects_invalid_window(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="window must be one of"):
+            await create_quest_definition(
+                database, "Brush teeth", _daily_rule(), [child.id], bad
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-window.db")(_body)
+
+
+def test_create_rejects_duplicate_windows(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(
+            ValueError, match="windows must not contain duplicate"
+        ):
+            await create_quest_definition(
+                database,
+                "Brush teeth",
+                _daily_rule(),
+                [child.id],
+                ["morning", "morning"],
+            )
+        with pytest.raises(
+            ValueError, match="windows must not contain duplicate"
+        ):
+            await create_quest_definition(
+                database,
+                "Brush teeth",
+                _daily_rule(),
+                [child.id],
+                [("morning", "09:00"), ("morning", "10:00")],
+            )
+        return None
+
+    _with_db(tmp_path, "create-duplicate-windows.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["9:30", "25:00", "09:60", "0900"])
+def test_create_rejects_malformed_due_time(tmp_path, bad) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="HH:MM"):
+            await create_quest_definition(
+                database,
+                "Brush teeth",
+                _daily_rule(),
+                [child.id],
+                [("morning", bad)],
+            )
+        return None
+
+    _with_db(tmp_path, "create-bad-due-time.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# create_quest_definition: a rejected create persists nothing
+# ---------------------------------------------------------------------------
+
+
+async def _assert_only_baseline_rows(
+    rules, definitions, child, baseline_definition
+) -> None:
+    """Assert the baseline rule/definition/assignee/window rows survive
+    unchanged and no second row of any of them appeared.
+
+    A rejected create must leave each table's row count exactly as it
+    was: still one rule, one definition, one assignee, one window —
+    checked through the DAO (id 2 must be absent everywhere, id 1
+    intact), since the leak guard forbids raw SQL in this test file.
+    """
+    assert await rules.get(1) is not None
+    assert await rules.get(2) is None
+    assert await definitions.get(1) == baseline_definition
+    assert await definitions.get(2) is None
+    assert [c.id for c in await definitions.list_assignees(1)] == [child.id]
+    assert await definitions.list_assignees(2) == []
+    assert [w.window for w in await definitions.list_windows(1)] == [
+        "morning"
+    ]
+    assert await definitions.list_windows(2) == []
+
+
+def test_rejected_create_writes_nothing(tmp_path) -> None:
+    """A rejected create leaves every quest row count unchanged.
+
+    Two rejection paths are forced: an in-transaction rejection (an
+    inactive assignee fails validation inside the transaction and must
+    roll the rule/definition back together) and a pre-write rejection
+    (an invalid window is caught in the business layer before any write).
+    Both must leave the baseline single-rule/single-definition state
+    untouched, with no second row in any of the four tables.
+    """
+    async def _body(database, children, rules, definitions, child):
+        baseline_rule = await rules.create("daily", "2026-09-14")
+        baseline_definition = await definitions.create(
+            "Baseline", baseline_rule.id, NOW, assignee_child_ids=[child.id]
+        )
+        await definitions.upsert_window(baseline_definition.id, "morning")
+        inactive = await children.create("Bo", NOW)
+        await children.set_active(inactive.id, False)
+        rule = _daily_rule()
+
+        # In-transaction rejection: the inactive child fails validation
+        # inside the transaction, rolling the rule and definition back.
+        with pytest.raises(ValueError, match="inactive"):
+            await create_quest_definition(
+                database, "Brush teeth", rule, [inactive.id], ["morning"]
+            )
+        await _assert_only_baseline_rows(
+            rules, definitions, child, baseline_definition
+        )
+
+        # Pre-write rejection: an invalid window is caught up front, so
+        # no transaction ever opens.
+        with pytest.raises(ValueError, match="window must be one of"):
+            await create_quest_definition(
+                database, "Brush teeth", rule, [child.id], [("noon", None)]
+            )
+        await _assert_only_baseline_rows(
+            rules, definitions, child, baseline_definition
+        )
+        return None
+
+    _with_db(tmp_path, "create-rejected-nothing.db")(_body)
+
+
+def test_post_write_failure_rolls_back_everything(tmp_path) -> None:
+    """A failure after some writes rolls all four tables back together.
+
+    Deterministic post-write trigger: the DAO method inserts the rule,
+    the definition and the assignee rows, THEN validates the window name
+    and raises ValueError for a window outside const.QUEST_WINDOWS.
+    That failure lands after three writes inside the transaction, so the
+    rule, definition and assignee rows must all roll back — no partial
+    rows survive in any of the four tables.
+    """
+    async def _body(database, children, rules, definitions, child):
+        storage = schedule_rule_to_storage(_daily_rule())
+        dao = QuestDefinitionsDao(database)
+        with pytest.raises(ValueError, match="window must be one of"):
+            await dao.create_with_rule_and_windows(
+                "Brush teeth",
+                storage,
+                NOW,
+                [child.id],
+                [("noon", None)],
+            )
+        # All four tables are empty: the earlier rule, definition and
+        # assignee writes were rolled back, and no window row was added.
+        assert await rules.get(1) is None
+        assert await definitions.get(1) is None
+        assert await definitions.list_active() == []
+        assert await definitions.list_assignees(1) == []
+        assert await definitions.list_windows(1) == []
+        return None
+
+    _with_db(tmp_path, "post-write-rollback.db")(_body)