fix: consistent edit snapshot and rollback coverage (CQ-01, CQ-02)

custom_components/nestquest/dao_rules.py

@@ -165,16 +165,18 @@ class QuestDefinitionWindowRecord:
 
 @dataclass(frozen=True)
 class QuestDefinitionSnapshot:
-    """A definition plus its assignees and windows, read atomically.
+    """A definition plus its rule, 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.
+    and :meth:`QuestDefinitionsDao.edit_definition` with the rule,
+    assignees and windows read inside the SAME transaction that wrote
+    them, so the snapshot is a consistent creation/edit-time view — a
+    concurrent rule, assignee or window mutation cannot change it
+    between the write and the read.
     """
 
     definition: QuestDefinitionRecord
+    rule: ScheduleRuleRecord
     assignees: list[ChildRecord]
     windows: list[QuestDefinitionWindowRecord]
 
@@ -329,6 +331,28 @@ def schedule_rule_from_storage(storage: ScheduleRuleStorage) -> ScheduleRule:
     )
 
 
+def schedule_rule_storage_from_record(
+    record: ScheduleRuleRecord,
+) -> ScheduleRuleStorage:
+    """Copy a rule row's value columns into a storage mapping.
+
+    ``ScheduleRuleRecord`` is the read shape (it carries ``id``);
+    :func:`schedule_rule_from_storage` consumes the value-only
+    :class:`ScheduleRuleStorage` shape, so callers re-reading a stored
+    rule convert through this helper.
+    """
+    return ScheduleRuleStorage(
+        rule_type=record.rule_type,
+        interval=record.interval,
+        weekday_set=record.weekday_set,
+        day_of_month=record.day_of_month,
+        nth_weekday=record.nth_weekday,
+        month=record.month,
+        start_date=record.start_date,
+        end_date=record.end_date,
+    )
+
+
 def _reject_forbidden_storage_fields(
     storage: ScheduleRuleStorage, shape: RuleType
 ) -> None:
@@ -719,9 +743,118 @@ class QuestDefinitionsDao:
                 definition = await self.get(definition_id)
                 assignees = await self.list_assignees(definition_id)
                 window_records = await self.list_windows(definition_id)
+                rule_record = await ScheduleRulesDao(self._database).get(
+                    rule_id
+                )
         assert definition is not None
+        assert rule_record is not None
         return QuestDefinitionSnapshot(
             definition=definition,
+            rule=rule_record,
+            assignees=assignees,
+            windows=window_records,
+        )
+
+    async def edit_definition(
+        self,
+        definition_id: int,
+        *,
+        title: str | None | object = _UNSET,
+        description: str | None | object = _UNSET,
+        icon: str | None | object = _UNSET,
+        rule: ScheduleRuleStorage | object = _UNSET,
+        windows: list[tuple[str, str | None]] | object = _UNSET,
+    ) -> QuestDefinitionSnapshot:
+        """Edit a definition's metadata, rule and windows atomically.
+
+        Mirrors :meth:`create_with_rule_and_windows`: everything runs
+        inside ONE transaction under the connection lock, and the
+        returned snapshot's rule, assignees and windows are read back
+        inside that transaction.  Arguments default to the module sentinel
+        ``_UNSET`` meaning "leave this field alone"; ``None`` writes SQL
+        NULL so optional metadata (description, icon) can be cleared.
+
+        - ``title``/``description``/``icon`` update the definition row
+          in place; the definition-level ``due_time`` column is never
+          written (per-window due times supersede it, D-008).
+        - ``rule`` (a pre-validated :class:`ScheduleRuleStorage`)
+          overwrites the referenced ``schedule_rules`` row in place: the
+          definition keeps its ``schedule_rule_id``, so no new or
+          orphaned rule row appears.
+        - ``windows`` REPLACES the whole window set: the current rows
+          are deleted and the given set re-inserted, so missing windows
+          are added, changed due times updated and removed windows
+          dropped.
+
+        Assignment is deliberately not settable here (multi-assignee is
+        managed by :meth:`add_assignee`/:meth:`remove_assignee`), and
+        activation is untouched.  Raises ValueError when the definition
+        does not exist.  Edits change future instances only: no
+        ``quest_instances`` or ``completion_events`` row is touched.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definition = await self.get(definition_id)
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                assignments: list[str] = []
+                parameters: list[object] = []
+                for column, value in (
+                    ("title", title),
+                    ("description", description),
+                    ("icon", icon),
+                ):
+                    if value is not _UNSET:
+                        assignments.append(f"{column} = ?")
+                        parameters.append(value)
+                if assignments:
+                    parameters.append(definition_id)
+                    await self._database.execute(
+                        f"UPDATE quest_definitions SET "
+                        f"{', '.join(assignments)} WHERE id = ?",
+                        tuple(parameters),
+                    )
+                if rule is not _UNSET:
+                    await ScheduleRulesDao(self._database).update(
+                        definition.schedule_rule_id,
+                        rule_type=rule.rule_type,
+                        interval=rule.interval,
+                        weekday_set=rule.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,
+                    )
+                if windows is not _UNSET:
+                    await self._database.execute(
+                        "DELETE FROM quest_definition_windows "
+                        "WHERE definition_id = ?",
+                        (definition_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),
+                        )
+                updated = await self.get(definition_id)
+                assignees = await self.list_assignees(definition_id)
+                window_records = await self.list_windows(definition_id)
+                rule_record = await ScheduleRulesDao(self._database).get(
+                    updated.schedule_rule_id
+                )
+        assert updated is not None
+        assert rule_record is not None
+        return QuestDefinitionSnapshot(
+            definition=updated,
+            rule=rule_record,
             assignees=assignees,
             windows=window_records,
         )

custom_components/nestquest/quest_definitions.py

@@ -1,10 +1,11 @@
-"""Business layer for creating quest definitions (Feature 06).
+"""Business layer for creating and editing 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).
+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).
 
 Validation policy mirrors :mod:`.children`:
 
@@ -43,9 +44,12 @@ from .dao_rules import (
     QuestDefinitionRecord,
     QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
+    ScheduleRuleStorage,
+    _UNSET,
     _validate_time,
     _validate_window,
     schedule_rule_from_storage,
+    schedule_rule_storage_from_record,
     schedule_rule_to_storage,
 )
 from .db import NestQuestDatabase
@@ -54,7 +58,7 @@ from .recurrence import ScheduleRule
 
 @dataclass(frozen=True)
 class CreatedQuestDefinition:
-    """The persisted result of :func:`create_quest_definition`.
+    """The persisted result of a create or edit.
 
     ``definition`` is the stored row; ``rule`` is the schedule rule
     decoded back through the storage mapping; ``assignees`` and
@@ -169,6 +173,18 @@ def _normalize_windows(
     return normalized
 
 
+def _validate_definition_id(value: object) -> int:
+    """Reject non-int definition ids (bools included) before any lookup.
+
+    SQLite binds Python bools as integers, so ``True`` would silently
+    address definition 1 and a float would round — malformed service
+    input must never mutate the wrong definition.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"definition_id must be an integer, got {value!r}")
+    return value
+
+
 async def create_quest_definition(
     database: NestQuestDatabase,
     title: str,
@@ -226,3 +242,82 @@ async def create_quest_definition(
         assignees=snapshot.assignees,
         windows=snapshot.windows,
     )
+
+
+async def edit_quest_definition(
+    database: NestQuestDatabase,
+    definition_id: int,
+    *,
+    title: str | object = _UNSET,
+    description: str | None | object = _UNSET,
+    icon: str | None | object = _UNSET,
+    rule: ScheduleRule | object = _UNSET,
+    windows: list[str | tuple[str, str | None]] | object = _UNSET,
+) -> CreatedQuestDefinition:
+    """Edit a quest definition's metadata, rule and windows atomically.
+
+    Arguments default to the sentinel ``_UNSET`` meaning "leave this
+    field alone".  A provided ``title`` is validated like create's
+    (required and trimmed); ``description`` and ``icon`` are optional
+    and may be cleared by passing ``None`` explicitly.  A provided
+    ``rule`` must be an already-validated :class:`~.recurrence.ScheduleRule`
+    and REPLACES the definition's schedule rule IN PLACE — the existing
+    ``schedule_rules`` row is updated, never orphaned or duplicated.  A
+    provided ``windows`` list REPLACES the whole window set (each entry
+    a ``const.QUEST_WINDOWS`` name, optionally with a strict HH:MM due
+    time), matching create's non-empty contract.  Assignment is NOT
+    settable here.  Raises ValueError when the definition does not
+    exist or any argument is rejected.
+
+    Returns the updated definition together with its decoded rule,
+    assignees and windows — all read back inside the DAO's transaction,
+    so the returned bundle is a consistent edit snapshot.  Edits change
+    future instances only; no ``quest_instances`` or
+    ``completion_events`` row is touched.
+    """
+    _validate_definition_id(definition_id)
+
+    title_value: str | None | object = _UNSET
+    if title is not _UNSET:
+        name = _validate_text(title, "title", required=True)
+        assert name is not None
+        title_value = name
+
+    description_value: str | None | object = _UNSET
+    if description is not _UNSET:
+        description_value = _validate_text(description, "description")
+
+    icon_value: str | None | object = _UNSET
+    if icon is not _UNSET:
+        icon_value = _validate_text(icon, "icon")
+
+    rule_storage: ScheduleRuleStorage | object = _UNSET
+    if rule is not _UNSET:
+        if not isinstance(rule, ScheduleRule):
+            raise ValueError(f"rule must be a ScheduleRule, got {rule!r}")
+        rule_storage = schedule_rule_to_storage(rule)
+
+    window_specs: list[tuple[str, str | None]] | object = _UNSET
+    if windows is not _UNSET:
+        window_specs = _normalize_windows(windows)
+
+    dao = QuestDefinitionsDao(database)
+    snapshot = await dao.edit_definition(
+        definition_id,
+        title=title_value,
+        description=description_value,
+        icon=icon_value,
+        rule=rule_storage,
+        windows=window_specs,
+    )
+
+    decoded_rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=decoded_rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
+    )

tests/test_quest_definitions.py

@@ -17,6 +17,7 @@ from custom_components.nestquest.migrations import apply_migrations
 from custom_components.nestquest.quest_definitions import (
     CreatedQuestDefinition,
     create_quest_definition,
+    edit_quest_definition,
 )
 from custom_components.nestquest.recurrence import RuleType, ScheduleRule
 
@@ -461,3 +462,421 @@ def test_post_write_failure_rolls_back_everything(tmp_path) -> None:
         return None
 
     _with_db(tmp_path, "post-write-rollback.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_quest_definition: successful edits
+# ---------------------------------------------------------------------------
+
+
+def test_edit_title_description_icon(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            ["morning"],
+            description="Old description",
+            icon="mdi:old",
+        )
+        result = await edit_quest_definition(
+            database,
+            created.definition.id,
+            title="  Floss  ",
+            description="  New description  ",
+            icon="  mdi:new  ",
+        )
+        assert result.definition.title == "Floss"
+        assert result.definition.description == "New description"
+        assert result.definition.icon == "mdi:new"
+        # Untouched fields survive the edit unchanged.
+        assert result.definition.id == created.definition.id
+        assert result.definition.schedule_rule_id == created.definition.schedule_rule_id
+        assert result.definition.due_time is None
+        assert result.definition.is_active is True
+        assert result.definition.created_at == created.definition.created_at
+        # Rule, assignees and windows are read back through the DAO.
+        assert result.rule == _daily_rule()
+        assert [c.id for c in result.assignees] == [child.id]
+        assert [w.window for w in result.windows] == ["morning"]
+        stored = await definitions.get(created.definition.id)
+        assert stored is not None
+        assert stored.title == "Floss"
+        assert stored.description == "New description"
+        assert stored.icon == "mdi:new"
+        return None
+
+    _with_db(tmp_path, "edit-metadata.db")(_body)
+
+
+def test_edit_omitted_fields_left_unchanged(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            ["morning"],
+            description="Keep me",
+        )
+        result = await edit_quest_definition(
+            database, created.definition.id, title="Renamed"
+        )
+        assert result.definition.title == "Renamed"
+        assert result.definition.description == "Keep me"
+        assert result.definition.icon is None
+        return None
+
+    _with_db(tmp_path, "edit-omitted.db")(_body)
+
+
+def test_edit_clears_optional_metadata(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            ["morning"],
+            description="Old",
+            icon="mdi:old",
+        )
+        result = await edit_quest_definition(
+            database,
+            created.definition.id,
+            description=None,
+            icon=None,
+        )
+        assert result.definition.description is None
+        assert result.definition.icon is None
+        return None
+
+    _with_db(tmp_path, "edit-clear-metadata.db")(_body)
+
+
+def test_edit_rule_updates_existing_row_in_place(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        rule_id = created.definition.schedule_rule_id
+        assert rule_id == 1
+        new_rule = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={1, 3},
+            start_date="2026-09-14",
+        )
+        result = await edit_quest_definition(
+            database, created.definition.id, rule=new_rule
+        )
+        assert result.rule == new_rule
+        # The definition keeps the SAME rule row: no new or orphaned row.
+        assert result.definition.schedule_rule_id == rule_id
+        rule_row = await rules.get(rule_id)
+        assert rule_row is not None
+        assert rule_row.rule_type == "weekly"
+        assert rule_row.weekday_set == "1,3"
+        assert await rules.get(rule_id + 1) is None
+        return None
+
+    _with_db(tmp_path, "edit-rule-in-place.db")(_body)
+
+
+def test_edit_rule_shape_change_clears_stale_columns(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        weekly = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2},
+            start_date="2026-09-14",
+        )
+        created = await create_quest_definition(
+            database, "Chore", weekly, [child.id], ["morning"]
+        )
+        rule_id = created.definition.schedule_rule_id
+        result = await edit_quest_definition(
+            database, created.definition.id, rule=_daily_rule()
+        )
+        assert result.rule == _daily_rule()
+        rule_row = await rules.get(rule_id)
+        assert rule_row is not None
+        # The old weekly shape's weekday_set must be cleared, not left stale.
+        assert rule_row.rule_type == "daily"
+        assert rule_row.weekday_set is None
+        return None
+
+    _with_db(tmp_path, "edit-rule-shape.db")(_body)
+
+
+def test_edit_replaces_windows(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            [("morning", "07:00"), ("evening", "19:00")],
+        )
+        result = await edit_quest_definition(
+            database,
+            created.definition.id,
+            windows=[("morning", "08:00"), ("afternoon", "12:00")],
+        )
+        by_name = {w.window: w.due_time for w in result.windows}
+        # morning's due time changed, afternoon added, evening removed.
+        assert by_name == {"morning": "08:00", "afternoon": "12:00"}
+        persisted = {
+            w.window: w.due_time
+            for w in await definitions.list_windows(created.definition.id)
+        }
+        assert persisted == {"morning": "08:00", "afternoon": "12:00"}
+        return None
+
+    _with_db(tmp_path, "edit-replace-windows.db")(_body)
+
+
+def test_edit_preserves_assignees(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"]
+        )
+        result = await edit_quest_definition(
+            database,
+            created.definition.id,
+            title="Renamed",
+            rule=ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={0},
+                start_date="2026-09-14",
+            ),
+            windows=[("evening", "20:00")],
+        )
+        assert sorted(c.id for c in result.assignees) == sorted([child.id, bo.id])
+        persisted = [
+            c.id for c in await definitions.list_assignees(created.definition.id)
+        ]
+        assert sorted(persisted) == sorted([child.id, bo.id])
+        return None
+
+    _with_db(tmp_path, "edit-preserves-assignees.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_quest_definition: argument rejections
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("bad", ["", "   ", None])
+def test_edit_rejects_empty_title(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="title is required"):
+            await edit_quest_definition(
+                database, created.definition.id, title=bad
+            )
+        return None
+
+    _with_db(tmp_path, "edit-bad-title.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [None, "weekly", 42, {"rule_type": "daily"}])
+def test_edit_rejects_non_schedule_rule(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="rule must be a ScheduleRule"):
+            await edit_quest_definition(
+                database, created.definition.id, rule=bad
+            )
+        return None
+
+    _with_db(tmp_path, "edit-bad-rule.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [["noon"], [("MORNING", None)], [("", None)]])
+def test_edit_rejects_invalid_window(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="window must be one of"):
+            await edit_quest_definition(
+                database, created.definition.id, windows=bad
+            )
+        return None
+
+    _with_db(tmp_path, "edit-bad-window.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["9:30", "25:00", "09:60", "0900"])
+def test_edit_rejects_malformed_due_time(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="HH:MM"):
+            await edit_quest_definition(
+                database, created.definition.id, windows=[("morning", bad)]
+            )
+        return None
+
+    _with_db(tmp_path, "edit-bad-due-time.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [[]])
+def test_edit_rejects_empty_windows(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="windows must not be empty"):
+            await edit_quest_definition(
+                database, created.definition.id, windows=bad
+            )
+        return None
+
+    _with_db(tmp_path, "edit-empty-windows.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [("morning",), "morning", {"morning"}])
+def test_edit_rejects_non_list_windows(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="windows must be a list"):
+            await edit_quest_definition(
+                database, created.definition.id, windows=bad
+            )
+        return None
+
+    _with_db(tmp_path, "edit-non-list-windows.db")(_body)
+
+
+def test_edit_rejects_nonexistent_definition(tmp_path) -> None:
+    async def _body(database, children, rules, definitions, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await edit_quest_definition(database, 999, title="X")
+        return None
+
+    _with_db(tmp_path, "edit-missing-definition.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, 1.5, "1", None])
+def test_edit_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 edit_quest_definition(database, bad, title="X")
+        return None
+
+    _with_db(tmp_path, "edit-bad-definition-id.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_quest_definition: consistent snapshot and rollback (CQ-01, CQ-02)
+# ---------------------------------------------------------------------------
+
+
+def test_concurrent_edit_omitted_rule_snapshot_consistent(tmp_path) -> None:
+    """Two concurrent edits never return a mixed-time snapshot.
+
+    One edit changes windows only (rule omitted); the other changes the
+    rule only (windows omitted).  Because each edit reads its rule,
+    assignees and windows back inside its own locked transaction, the
+    windows-only edit reports the rule as it was AT THAT EDIT's
+    transaction: the ORIGINAL rule when it ran first, the NEW rule when
+    it ran second.  That yields a strict invariant — the windows-only
+    result's rule is the original exactly when the rule-only result
+    still saw the original windows — which a post-transaction rule read
+    would violate.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database, "Brush teeth", _daily_rule(), [child.id], ["morning"]
+        )
+        weekly = ScheduleRule(
+            rule_type=RuleType.WEEKLY,
+            weekday_set={2},
+            start_date="2026-09-14",
+        )
+        windows_result, rule_result = await asyncio.gather(
+            edit_quest_definition(
+                database,
+                created.definition.id,
+                windows=[("evening", "19:00")],
+            ),
+            edit_quest_definition(database, created.definition.id, rule=weekly),
+        )
+        # Each edit reports its own change...
+        assert [w.window for w in windows_result.windows] == ["evening"]
+        assert rule_result.rule == weekly
+        # ...and the pair is self-consistent: the windows-only edit saw
+        # the original rule iff the rule-only edit saw the original
+        # windows (they ran in that order).
+        assert (windows_result.rule == _daily_rule()) == (
+            [w.window for w in rule_result.windows] == ["evening"]
+        )
+        return None
+
+    _with_db(tmp_path, "edit-concurrent-snapshot.db")(_body)
+
+
+def test_edit_post_write_failure_rolls_back_everything(tmp_path) -> None:
+    """A post-write failure during an edit rolls every table back together.
+
+    Deterministic trigger: call the DAO's edit directly with a window
+    name outside const.QUEST_WINDOWS.  The DAO applies the metadata and
+    rule writes and the window delete, THEN validates the window name
+    and raises ValueError.  That failure lands after several writes
+    inside the transaction, so the metadata, rule and window changes
+    must all roll back — the baseline definition, rule and windows
+    survive unchanged.
+    """
+    async def _body(database, children, rules, definitions, child):
+        created = await create_quest_definition(
+            database,
+            "Brush teeth",
+            _daily_rule(),
+            [child.id],
+            [("morning", "07:00")],
+        )
+        rule_id = created.definition.schedule_rule_id
+        weekly_storage = schedule_rule_to_storage(
+            ScheduleRule(
+                rule_type=RuleType.WEEKLY,
+                weekday_set={2},
+                start_date="2026-09-14",
+            )
+        )
+        dao = QuestDefinitionsDao(database)
+        with pytest.raises(ValueError, match="window must be one of"):
+            await dao.edit_definition(
+                created.definition.id,
+                title="Changed",
+                rule=weekly_storage,
+                windows=[("noon", None)],
+            )
+        stored = await definitions.get(created.definition.id)
+        assert stored is not None
+        assert stored.title == "Brush teeth"
+        rule_row = await rules.get(rule_id)
+        assert rule_row is not None
+        assert rule_row.rule_type == "daily"
+        assert rule_row.weekday_set is None
+        by_name = {
+            w.window: w.due_time
+            for w in await definitions.list_windows(created.definition.id)
+        }
+        assert by_name == {"morning": "07:00"}
+        assert [
+            c.id for c in await definitions.list_assignees(created.definition.id)
+        ] == [child.id]
+        return None
+
+    _with_db(tmp_path, "edit-post-write-rollback.db")(_body)