fix: atomic complete_instance check+append

custom_components/nestquest/__init__.py

@@ -2,19 +2,32 @@
 from __future__ import annotations
 
 import asyncio
+import datetime
 import inspect
 import sqlite3
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.event import async_track_time_change
 
-from .const import CONF_ADMIN_USER_IDS, DOMAIN, LOGGER
+from .const import (
+    CONF_ADMIN_USER_IDS,
+    CONF_DAY_ROLLOVER_TIME,
+    CONF_HORIZON_DAYS,
+    DEFAULT_DAY_ROLLOVER_TIME,
+    DEFAULT_HORIZON_DAYS,
+    DOMAIN,
+    LOGGER,
+    SERVICE_REGENERATE,
+)
 from .db import NestQuestDatabase
+from .materialize import materialize as _materialize_run
 from .migrations import apply_migrations
 from .store import async_get_db_path
 from .admin_allowlist import seed_setup_admin
@@ -224,6 +237,7 @@ class NestQuestRuntimeData:
     options: dict[str, Any]
     database: NestQuestDatabase
     remove_update_listener: Callable[[], Any]
+    remove_time_change_listener: Callable[[], Any]
 
 
 async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
@@ -248,6 +262,138 @@ async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
     return owners
 
 
+def _day_rollover_hour_minute(options: dict[str, Any]) -> tuple[int, int]:
+    """Return the (hour, minute) the daily rollover fires at from ``options``.
+
+    The options flow validates ``day_rollover_time`` as a strict ``HH:MM``
+    string, so a stored value never needs re-validation here; a missing key
+    falls back to :data:`~.const.DEFAULT_DAY_ROLLOVER_TIME`.
+    """
+    rollover = options.get(CONF_DAY_ROLLOVER_TIME, DEFAULT_DAY_ROLLOVER_TIME)
+    hour, _, minute = rollover.partition(":")
+    return int(hour), int(minute)
+
+
+def _configured_horizon_days(options: dict[str, Any]) -> int:
+    """Return the configured generation horizon, validated, else the default.
+
+    The options flow already validates ``horizon_days`` as an integer >= 1,
+    but a stored ``entry.options`` dict could still carry a partial or
+    hand-edited value.  Re-checking here — and falling back to
+    :data:`~.const.DEFAULT_HORIZON_DAYS` on a missing, non-int or sub-1 value
+    — keeps a malformed option from shrinking or exploding the materialized
+    horizon.
+    """
+    value = options.get(CONF_HORIZON_DAYS, DEFAULT_HORIZON_DAYS)
+    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+        return DEFAULT_HORIZON_DAYS
+    return value
+
+
+async def _run_horizon_materialization(
+    hass: HomeAssistant,
+    database: NestQuestDatabase,
+    horizon_days: int = DEFAULT_HORIZON_DAYS,
+) -> None:
+    """Materialize the rolling horizon ``[today, today + horizon_days]``.
+
+    "today" is computed in HA local time (``hass.config.time_zone``) rather
+    than the system clock, so the horizon tracks the household's own day.
+    The walk itself runs through :func:`~.materialize.materialize`, whose DB
+    access is executor-only.
+
+    This is the ONE generation path: the startup backfill, the daily rollover
+    listener and the ``regenerate`` service all funnel through it, so there is
+    no duplicated materialization logic.  ``horizon_days`` sizes the window
+    and defaults to :data:`~.const.DEFAULT_HORIZON_DAYS`; callers thread the
+    entry's validated option value through.
+    """
+    time_zone = ZoneInfo(hass.config.time_zone)
+    today = datetime.datetime.now(time_zone).date()
+    start_date = today.isoformat()
+    end_date = (
+        today + datetime.timedelta(days=horizon_days)
+    ).isoformat()
+    await _materialize_run(database, start_date, end_date, today=today)
+
+
+def _register_day_rollover_listener(
+    hass: HomeAssistant,
+    database: NestQuestDatabase,
+    options: dict[str, Any],
+) -> Callable[[], Any]:
+    """Register the daily materialization listener and return its remover.
+
+    Fires once per day at the configured local ``day_rollover_time`` and
+    materializes the rolling horizon ``[today, today + horizon_days]``
+    through the shared :func:`_run_horizon_materialization` path, using the
+    entry's configured (validated) ``horizon_days``.
+    """
+    hour, minute = _day_rollover_hour_minute(options)
+    horizon_days = _configured_horizon_days(options)
+
+    async def _run_materialization(_now: datetime.datetime) -> None:
+        await _run_horizon_materialization(hass, database, horizon_days)
+
+    return async_track_time_change(
+        hass, _run_materialization, hour=hour, minute=minute
+    )
+
+
+def _find_live_runtime_data(hass: HomeAssistant) -> NestQuestRuntimeData | None:
+    """Return a live runtime record with a connected database, or None.
+
+    The domain-global services (e.g. ``regenerate``) have no database of
+    their own; they resolve whichever config entry's runtime data is
+    currently live at call time, so none of them can hold a stale handle
+    to a closed connection after another entry unloads.  The record also
+    carries that entry's options (e.g. the configured horizon).
+    """
+    for runtime_data in hass.data.get(DOMAIN, {}).values():
+        database = getattr(runtime_data, "database", None)
+        if database is not None and database.connected:
+            return runtime_data
+    return None
+
+
+def _register_regenerate_service(hass: HomeAssistant) -> None:
+    """Register ``nestquest.regenerate`` once at domain scope.
+
+    The service triggers a full on-demand regeneration over the rolling
+    horizon, running the SAME idempotent materialize path
+    (:func:`_run_horizon_materialization`) as the startup backfill and the
+    daily rollover listener.  It is DOMAIN-global — registered exactly once
+    (guarded by ``has_service``) and NOT bound to any config entry — so the
+    handler resolves the currently-live database at call time and a name
+    collision or duplicate setup is impossible.  Removal is handled by the
+    unload path only once the LAST entry is removed (see
+    :func:`_deregister_regenerate_service`).
+    """
+    if hass.services.has_service(DOMAIN, SERVICE_REGENERATE):
+        return
+
+    async def _regenerate(_call: Any) -> None:
+        runtime_data = _find_live_runtime_data(hass)
+        if runtime_data is None:
+            _LOGGER.warning(
+                "NestQuest regenerate requested but no config entry has a "
+                "live database; ignoring"
+            )
+            return
+        horizon_days = _configured_horizon_days(runtime_data.options)
+        await _run_horizon_materialization(
+            hass, runtime_data.database, horizon_days
+        )
+
+    hass.services.async_register(DOMAIN, SERVICE_REGENERATE, _regenerate)
+
+
+def _deregister_regenerate_service(hass: HomeAssistant) -> None:
+    """Remove ``nestquest.regenerate`` when the last entry unloads."""
+    if hass.services.has_service(DOMAIN, SERVICE_REGENERATE):
+        hass.services.async_remove(DOMAIN, SERVICE_REGENERATE)
+
+
 async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
     """Set up the NestQuest integration. YAML configuration is not used, returns True."""
     return True
@@ -305,6 +451,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         await listener_hass.config_entries.async_reload(listener_entry.entry_id)
 
     database = await _async_open_database(hass, await async_get_db_path(hass))
+    remove_update_listener: Callable[[], Any] | None = None
+    remove_time_change_listener: Callable[[], Any] | None = None
     try:
         # Seed the admin allowlist on first setup so the owner is never
         # locked out: an empty database takes the persisted admin copy
@@ -330,14 +478,46 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             owner_ids=owner_ids,
         )
         remove_update_listener = entry.add_update_listener(_async_update_listener)
+        remove_time_change_listener = _register_day_rollover_listener(
+            hass, database, dict(entry.options)
+        )
+        # Backfill any days the daily listener missed while the integration
+        # was off (or freshly installed): run the one idempotent walk over
+        # [today, today + horizon_days] once at setup, using the entry's
+        # configured horizon.  This runs through the SAME path as the daily
+        # listener and regenerate service.
+        await _run_horizon_materialization(
+            hass, database, _configured_horizon_days(dict(entry.options))
+        )
+        _register_regenerate_service(hass)
     except BaseException:
+        # A failure anywhere after the database is open must not leak the
+        # listeners already registered against that (now-closed) database:
+        # unwind EVERY remover we acquired before the failure, keeping the
+        # original error as the one raised.  Removal is best-effort — a
+        # failing remover must not mask the setup error that triggered it.
+        for remove_listener in (
+            remove_time_change_listener,
+            remove_update_listener,
+        ):
+            if remove_listener is None:
+                continue
+            try:
+                result = remove_listener()
+                if inspect.isawaitable(result):
+                    await result
+            except BaseException:
+                pass
         await database.close()
         raise
+    assert remove_update_listener is not None
+    assert remove_time_change_listener is not None
     runtime_data = NestQuestRuntimeData(
         entry_id=entry.entry_id,
         options=dict(entry.options),
         database=database,
         remove_update_listener=remove_update_listener,
+        remove_time_change_listener=remove_time_change_listener,
     )
     entry.runtime_data = runtime_data
     hass.data[DOMAIN][entry.entry_id] = runtime_data
@@ -348,28 +528,47 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload a config entry."""
     domain_data = hass.data.setdefault(DOMAIN, {})
     runtime_data = domain_data.pop(entry.entry_id, None)
-    if not domain_data:
+    is_last = not domain_data
+    if is_last:
         hass.data.pop(DOMAIN, None)
     if runtime_data is None:
         runtime_data = getattr(entry, "runtime_data", None)
+    unload_error: BaseException | None = None
     if runtime_data is not None:
         remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
-        unload_error: BaseException | None = None
+        remove_time_change_listener = getattr(
+            runtime_data, "remove_time_change_listener", None
+        )
         try:
-            if remove_update_listener is not None:
-                result = remove_update_listener()
-                if inspect.isawaitable(result):
-                    await result
-        except BaseException as err:
-            unload_error = err
+            for remove_listener in (
+                remove_update_listener,
+                remove_time_change_listener,
+            ):
+                if remove_listener is None:
+                    continue
+                try:
+                    result = remove_listener()
+                    if inspect.isawaitable(result):
+                        await result
+                except BaseException as err:
+                    # Cancel EVERY listener even when an earlier remover
+                    # raises: a skipped remover would leave a daily
+                    # callback firing against a closed database after
+                    # unload.  The first error is retained and re-raised
+                    # only after both removers (and the DB close) ran.
+                    if unload_error is None:
+                        unload_error = err
         finally:
             database = getattr(runtime_data, "database", None)
             if database is not None:
                 await database.close()
-        if unload_error is not None:
-            if getattr(entry, "runtime_data", None) is not None:
-                entry.runtime_data = None
-            raise unload_error
+    # The ``regenerate`` service is DOMAIN-global: it is torn down only
+    # once the FINAL entry unloads (``hass.data[DOMAIN]`` is now empty),
+    # never when a sibling entry is still loaded.
+    if is_last:
+        _deregister_regenerate_service(hass)
     if getattr(entry, "runtime_data", None) is not None:
         entry.runtime_data = None
+    if unload_error is not None:
+        raise unload_error
     return True

custom_components/nestquest/completion.py

@@ -0,0 +1,316 @@
+"""Business layer for the completion event log (Feature 08).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_instances.CompletionEventsDao`: callers get actor-shape
+validation and the UTC occurred-at stamp here, and never talk to the
+DAO or SQL directly.  This module is the only Feature-08 caller of
+:meth:`~.dao_instances.CompletionEventsDao.append`.  Completing an
+instance (no-op if already done) is :func:`complete_instance`;
+un-completing arrives in a later task.  Instance current state is
+derived from the latest event via :func:`instance_state`
+(``open`` / ``done``); there is no status column.
+
+Actor policy matches the schema CHECKs (D-008): ``actor_source`` is
+``'user'`` or ``'panel'`` only — never ``'service'``.  ``'user'``
+requires ``actor_user_id`` and forbids ``actor_child_id``; ``'panel'``
+requires ``actor_child_id`` (the tapped profile) and forbids
+``actor_user_id``.  Inverted shapes raise a field-naming ValueError
+before any write.
+
+``occurred_at`` is stamped UTC ISO-8601 at second precision with an
+explicit ``+00:00`` offset, matching the DAO's one accepted shape so
+lexicographic range queries stay exact.  ``was_on_time`` is required
+(True/False) for completed events; un-completions may pass None.  There
+is no permission check here (that is Feature 09).
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+
+from .dao_children import _CONNECTION_LOCKS, _connection_lock
+from .dao_instances import (
+    EVENT_COMPLETED,
+    EVENT_UNCOMPLETED,
+    CompletionEventRecord,
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from .db import NestQuestDatabase
+
+#: Timestamp policy for completion_events.occurred_at: strict UTC
+#: ISO-8601 with explicit '+00:00' offset (module timestamp policy).
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
+
+
+def _stamp_occurred_at(now: datetime.datetime | None) -> str:
+    """Return a DAO-accepted UTC stamp for ``now``, or the current time."""
+    if now is None:
+        return _now_stamp()
+    if not isinstance(now, datetime.datetime):
+        raise ValueError(f"now must be a datetime, got {now!r}")
+    if now.tzinfo is None:
+        raise ValueError("now must be timezone-aware")
+    return now.astimezone(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
+
+
+class _TaskReentrantLock:
+    """Shared connection lock the owning task may re-enter."""
+
+    def __init__(self, inner: asyncio.Lock) -> None:
+        self._inner = inner
+        self._task: asyncio.Task | None = None
+        self._depth = 0
+
+    def locked(self) -> bool:
+        return self._inner.locked()
+
+    async def acquire(self) -> bool:
+        task = asyncio.current_task()
+        if self._task is task:
+            self._depth += 1
+            return True
+        await self._inner.acquire()
+        self._task = task
+        self._depth = 1
+        return True
+
+    def release(self) -> None:
+        if self._task is not asyncio.current_task():
+            raise RuntimeError("lock released by a non-owner")
+        self._depth -= 1
+        if self._depth == 0:
+            self._task = None
+            self._inner.release()
+
+    async def __aenter__(self) -> _TaskReentrantLock:
+        await self.acquire()
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        self.release()
+
+
+def _reentrant_connection_lock(database) -> _TaskReentrantLock:
+    """Return the shared connection lock, made task-reentrant.
+
+    ``CompletionEventsDao.append`` also acquires this lock, so
+    check+append in :func:`complete_instance` must re-enter on the
+    same task.  The wrapper is stored in the shared dict so the DAO
+    sees the same object.
+    """
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _CONNECTION_LOCKS.get(key)
+    if isinstance(lock, _TaskReentrantLock):
+        return lock
+    wrapped = _TaskReentrantLock(
+        lock if lock is not None else asyncio.Lock()
+    )
+    _CONNECTION_LOCKS[key] = wrapped
+    return wrapped
+
+
+def _validate_int_id(value: object, field: str) -> int:
+    """Reject non-int ids (bools included) before any lookup."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"{field} must be an integer, got {value!r}")
+    return value
+
+
+def _validate_event_type(value: object) -> str:
+    """Return a schema-legal event type, or raise naming the field."""
+    if value not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
+        raise ValueError(
+            f"event_type must be '{EVENT_COMPLETED}' or "
+            f"'{EVENT_UNCOMPLETED}', got {value!r}"
+        )
+    return value
+
+
+def _validate_actor(
+    actor_source: object,
+    actor_user_id: object,
+    actor_child_id: object,
+) -> tuple[str, str | None, int | None]:
+    """Return a schema-legal actor triple, or raise naming the field.
+
+    ``'user'`` requires a non-empty ``actor_user_id`` string and forbids
+    ``actor_child_id``.  ``'panel'`` requires an integer
+    ``actor_child_id`` and forbids ``actor_user_id``.  Any other
+    ``actor_source`` (including ``'service'``) is rejected.
+    """
+    if actor_source not in ("user", "panel"):
+        raise ValueError(
+            f"actor_source must be 'user' or 'panel', got {actor_source!r}"
+        )
+    if actor_source == "user":
+        if not isinstance(actor_user_id, str) or not actor_user_id.strip():
+            raise ValueError("actor_source 'user' requires actor_user_id")
+        if actor_child_id is not None:
+            raise ValueError(
+                "actor_source 'user' must not carry actor_child_id"
+            )
+        return "user", actor_user_id.strip(), None
+    if actor_user_id is not None:
+        raise ValueError("actor_source 'panel' must not carry actor_user_id")
+    if actor_child_id is None:
+        raise ValueError(
+            "actor_source 'panel' requires actor_child_id"
+        )
+    return (
+        "panel",
+        None,
+        _validate_int_id(actor_child_id, "actor_child_id"),
+    )
+
+
+def _validate_was_on_time(
+    event_type: str, was_on_time: object
+) -> bool | None:
+    """Return a real bool or None; completed events require True/False."""
+    if was_on_time is not None and type(was_on_time) is not bool:
+        raise ValueError(
+            f"was_on_time must be True, False or None, got {was_on_time!r}"
+        )
+    if event_type == EVENT_COMPLETED and was_on_time is None:
+        raise ValueError(
+            "a completed event requires was_on_time (True/False)"
+        )
+    return was_on_time
+
+
+async def append_event(
+    database: NestQuestDatabase,
+    instance_id: int,
+    child_id: int,
+    event_type: str,
+    *,
+    actor_source: str,
+    actor_user_id: str | None = None,
+    actor_child_id: int | None = None,
+    was_on_time: bool | None = None,
+    now: datetime.datetime | None = None,
+) -> CompletionEventRecord:
+    """Append one completion or un-completion event and return the row.
+
+    Validates actor shapes and the completed-event on-time requirement
+    before calling :meth:`CompletionEventsDao.append`.  ``occurred_at``
+    is stamped here (UTC ``+00:00``); pass ``now`` to pin the clock.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    child_id = _validate_int_id(child_id, "child_id")
+    event_type = _validate_event_type(event_type)
+    actor_source, actor_user_id, actor_child_id = _validate_actor(
+        actor_source, actor_user_id, actor_child_id
+    )
+    was_on_time = _validate_was_on_time(event_type, was_on_time)
+    occurred_at = _stamp_occurred_at(now)
+    return await CompletionEventsDao(database).append(
+        instance_id,
+        child_id,
+        event_type,
+        actor_source,
+        occurred_at,
+        was_on_time,
+        actor_user_id=actor_user_id,
+        actor_child_id=actor_child_id,
+    )
+
+
+async def instance_state(
+    database: NestQuestDatabase,
+    instance_id: int,
+) -> str:
+    """Return ``open`` or ``done`` from the instance's latest event.
+
+    No events, or a latest event of uncompleted, is ``open``.  A latest
+    event of completed is ``done``.  Unknown ``instance_id`` raises
+    ValueError naming the field.  Existence and the latest event are
+    read inside the connection lock so the answer is one snapshot.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    async with _connection_lock(database):
+        instance = await QuestInstancesDao(database).get_by_id(instance_id)
+        if instance is None:
+            raise ValueError(
+                f"instance_id: quest instance {instance_id} does not exist"
+            )
+        latest = await CompletionEventsDao(database).get_latest_for_instance(
+            instance_id
+        )
+    if latest is None or latest.event_type == EVENT_UNCOMPLETED:
+        return "open"
+    if latest.event_type == EVENT_COMPLETED:
+        return "done"
+    raise ValueError(
+        f"event_type must be '{EVENT_COMPLETED}' or "
+        f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
+    )
+
+
+async def complete_instance(
+    database: NestQuestDatabase,
+    instance_id: int,
+    *,
+    actor_source: str,
+    actor_user_id: str | None = None,
+    actor_child_id: int | None = None,
+    now: datetime.datetime | None = None,
+    today: datetime.date | None = None,
+) -> str:
+    """Append a completed event unless the instance is already done.
+
+    Returns the derived state ``done``.  An already-done instance is a
+    no-op: nothing is appended.  Unknown ``instance_id`` raises
+    ValueError naming the field.  Existence, latest-event, and the
+    append run under one connection lock so a concurrent complete
+    cannot double-append.
+
+    ``was_on_time`` is recorded as True — a placeholder until the
+    on-time task computes it from ``now`` / ``today`` against the
+    instance due_date and optional due_time.  ``today`` is accepted so
+    callers can thread the HA-local date through without a signature
+    change; it is unused here.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    actor_source, actor_user_id, actor_child_id = _validate_actor(
+        actor_source, actor_user_id, actor_child_id
+    )
+    async with _reentrant_connection_lock(database):
+        instance = await QuestInstancesDao(database).get_by_id(instance_id)
+        if instance is None:
+            raise ValueError(
+                f"instance_id: quest instance {instance_id} does not exist"
+            )
+        latest = await CompletionEventsDao(database).get_latest_for_instance(
+            instance_id
+        )
+        if latest is not None and latest.event_type == EVENT_COMPLETED:
+            return "done"
+        if latest is not None and latest.event_type != EVENT_UNCOMPLETED:
+            raise ValueError(
+                f"event_type must be '{EVENT_COMPLETED}' or "
+                f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
+            )
+        await append_event(
+            database,
+            instance.id,
+            instance.child_id,
+            EVENT_COMPLETED,
+            actor_source=actor_source,
+            actor_user_id=actor_user_id,
+            actor_child_id=actor_child_id,
+            was_on_time=True,
+            now=now,
+        )
+    return "done"

custom_components/nestquest/const.py

@@ -9,6 +9,9 @@ LOGGER = logging.getLogger(LOGGER_NAME)
 
 PLATFORMS: list[str] = []
 
+#: Service name for the on-demand regeneration service (``nestquest.regenerate``).
+SERVICE_REGENERATE = "regenerate"
+
 SQLITE_DB_FILENAME = "nestquest.db"
 
 # Configuration keys and default values

custom_components/nestquest/dao_instances.py

@@ -47,6 +47,46 @@ from .db import NestQuestDatabase
 _UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
 
 
+def _today() -> datetime.date:
+    """Return the current local calendar date (the single date source).
+
+    Every "today" read in this module goes through this one function so
+    the delete clamp and the materialize walk's no-past guard read the
+    same clock, and a caller or test can pin/roll that clock
+    deterministically (e.g. a queued-across-midnight regression).  Pure
+    clock read — no business rule lives here.
+    """
+    return datetime.date.today()
+
+
+def _resolve_today(today: datetime.date | None) -> datetime.date:
+    """Return the caller-resolved "today", or the host date when absent.
+
+    The no-past guard and the delete cutoffs must compare due dates
+    against the SAME calendar date the materialization horizon was built
+    from.  Home Assistant's configured time zone can lag the host clock
+    around midnight, so when the caller has already resolved the local
+    date (``today``) that anchor is authoritative; ``None`` falls back to
+    the host clock (:func:`_today`), which is exactly the previous
+    behavior for callers with no HA-local date.
+    """
+    return _today() if today is None else today
+
+
+def _reject_past(due_date: str, today_iso: str) -> None:
+    """Raise unless ``due_date`` is on or after ``today_iso``.
+
+    The single no-past rule shared by :meth:`QuestInstancesDao.upsert`
+    and :meth:`QuestInstancesDao.upsert_if_valid`: instances are never
+    generated in the past, measured against the resolved "today".
+    """
+    if due_date < today_iso:
+        raise ValueError(
+            f"instances are never generated in the past: due_date "
+            f"{due_date!r} is before today {today_iso!r}"
+        )
+
+
 def _validate_utc_timestamp(value: str, field: str) -> None:
     """Raise ValueError unless ``value`` is a strict UTC ISO-8601 stamp.
 
@@ -166,6 +206,7 @@ class QuestInstancesDao:
         *,
         window: str,
         due_time: str | None = None,
+        today: datetime.date | None = None,
     ) -> QuestInstanceRecord:
         """Create the dated instance for the widened key idempotently.
 
@@ -184,6 +225,10 @@ class QuestInstancesDao:
         - ``due_date`` must be TODAY or later: instances are never
           generated in the past (Feature 07).  The caller's
           materialization horizon is future-dated by definition.
+          ``today`` optionally pins the resolved HA-local date the
+          guard compares against (falling back to the host clock when
+          omitted) so a household time zone behind the host around
+          midnight is not misread as "past".
         - ``window`` must be one of ``const.QUEST_WINDOWS``.
         - An existing instance WITH a completion event is IMMUTABLE:
           the conflict path must not rewrite its snapshot columns
@@ -203,24 +248,20 @@ class QuestInstancesDao:
         # check runs after the lock is acquired, because a caller can
         # wait on the lock across midnight — a date that was "today"
         # at call time may be "yesterday" by the time the INSERT runs.
-        today = datetime.date.today().isoformat()
-        if due_date < today:
-            raise ValueError(
-                f"instances are never generated in the past: due_date "
-                f"{due_date!r} is before today {today!r}"
-            )
+        # A caller-supplied ``today`` (HA-local) is used for BOTH
+        # checks and never re-read, since it is already the anchor the
+        # batch runs against.
+        today_date = _resolve_today(today)
+        _reject_past(due_date, today_date.isoformat())
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 # Re-read "today" under the lock: this is the date the
                 # insert actually executes on, so the no-past rule
-                # holds across midnight rollovers.
-                today = datetime.date.today().isoformat()
-                if due_date < today:
-                    raise ValueError(
-                        f"instances are never generated in the past: "
-                        f"due_date {due_date!r} is before today "
-                        f"{today!r}"
-                    )
+                # holds across midnight rollovers.  A pinned HA-local
+                # anchor is exempt from the re-read (see above).
+                if today is None:
+                    today_date = _today()
+                _reject_past(due_date, today_date.isoformat())
                 definition = await self._database.fetch_one(
                     "SELECT 1 FROM quest_definitions WHERE id = ?",
                     (definition_id,),
@@ -277,6 +318,123 @@ class QuestInstancesDao:
         assert instance is not None
         return instance
 
+    async def upsert_if_valid(
+        self,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        generated_at: str,
+        *,
+        window: str,
+        due_time: str | None = None,
+        today: datetime.date | None = None,
+    ) -> QuestInstanceRecord | None:
+        """Insert one dated instance iff its tuple is still materializable.
+
+        This is the materialization walk's single write path.  Unlike the
+        generic :meth:`upsert`, which validates only existence and
+        assignment, this method re-validates EVERY materialization
+        precondition in the SAME connection-lock-held transaction as the
+        INSERT — closing the check-then-insert gap a concurrent config
+        write could otherwise slip through:
+
+        - the definition exists AND is active (a definition deactivated
+          after the input snapshot is skipped, never written);
+        - the child exists AND is active (a deactivated child is skipped);
+        - the (definition, child) assignment link still exists (a removed
+          assignment is skipped);
+        - the window is still declared on the definition (a removed window
+          is skipped);
+        - the window's live ``due_time`` still EQUALS the caller-supplied
+          ``due_time`` (NULL-safe: NULL == NULL counts as equal; a window
+          with no due time matches a snapshot with no due time).  A
+          window-time edit landed after the input snapshot makes the
+          caller's ``due_time`` stale, and writing it would overwrite the
+          newer due_time a concurrent regeneration already upserted — so
+          the tuple is skipped instead of written.
+
+        The due_time check closes the one check-then-insert gap this
+        method previously left open: it re-validated every OTHER
+        precondition but then wrote the snapshot's ``due_time``
+        unconditionally, so a stale materialization could clobber a
+        newer, concurrently-regenerated due_time.
+
+        Returns the stored instance, or ``None`` when a precondition
+        failed — the tuple no longer materializable, which the walk SKIPS
+        rather than raising, so the rest of the batch is unaffected.  The
+        no-past rule (instances are never generated in the past) is
+        preserved from :meth:`upsert` and still raises ValueError; ``today``
+        optionally pins the HA-local anchor it compares against.  An
+        instance that ALREADY has a completion event is also skipped
+        (returns ``None``) rather than raised: it is immutable, so the
+        walk leaves it untouched instead of failing the batch — the
+        generic :meth:`upsert` keeps its hard refusal for the
+        delete/regenerate path.
+        """
+        _validate_date(due_date, "due_date")
+        _validate_window(window)
+        # Fail fast on obviously-past dates before queuing on the lock;
+        # the authoritative check re-runs under the lock (see upsert).
+        today_date = _resolve_today(today)
+        _reject_past(due_date, today_date.isoformat())
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                if today is None:
+                    today_date = _today()
+                _reject_past(due_date, today_date.isoformat())
+                still_valid = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definitions d "
+                    "JOIN quest_definition_assignees a "
+                    "ON a.definition_id = d.id "
+                    "JOIN children c ON c.id = a.child_id "
+                    "JOIN quest_definition_windows w "
+                    "ON w.definition_id = d.id "
+                    "WHERE d.id = ? AND a.child_id = ? AND w.window = ? "
+                    "AND d.is_active = 1 AND c.is_active = 1 "
+                    "AND w.due_time IS ? LIMIT 1",
+                    (definition_id, child_id, window, due_time),
+                )
+                if still_valid is None:
+                    # A config change invalidated the tuple since the
+                    # snapshot: skip it rather than abort the batch.
+                    return None
+                existing = await self._database.fetch_one(
+                    "SELECT 1 FROM completion_events "
+                    "WHERE instance_id = (SELECT id FROM quest_instances "
+                    "WHERE definition_id = ? AND child_id = ? "
+                    "AND due_date = ? AND window = ?)",
+                    (definition_id, child_id, due_date, window),
+                )
+                if existing is not None:
+                    # The instance is already completed and therefore
+                    # immutable: the walk SKIPS it (no-op) rather than
+                    # rewriting its snapshot columns or failing the
+                    # batch.  This is the materialization walk's skip,
+                    # distinct from the generic :meth:`upsert` refusal.
+                    return None
+                await self._database.execute(
+                    "INSERT INTO quest_instances (definition_id, child_id, "
+                    "window, due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
+                    "window) DO UPDATE SET "
+                    "due_time = excluded.due_time, "
+                    "generated_at = excluded.generated_at",
+                    (
+                        definition_id,
+                        child_id,
+                        window,
+                        due_date,
+                        due_time,
+                        generated_at,
+                    ),
+                )
+                instance = await self.get(
+                    definition_id, child_id, due_date, window
+                )
+        assert instance is not None
+        return instance
+
     async def get(
         self,
         definition_id: int,
@@ -336,7 +494,7 @@ class QuestInstancesDao:
         return [_instance_from_row(row) for row in rows]
 
     async def delete_future_uncompleted(
-        self, definition_id: int, cutoff_date: str
+        self, definition_id: int, cutoff_date: str, *, today: datetime.date | None = None
     ) -> int:
         """Delete the definition's open instances at/after cutoff.
 
@@ -348,9 +506,11 @@ class QuestInstancesDao:
         is measured against TODAY (materialization regenerates the
         rolling horizon from now), so a caller cannot use this to
         rewrite past open instances either — the day-rollover missed
-        sweep (Feature 11) handles past-due instances instead.  Any
-        instance with a completion event is never touched, regardless
-        of its date.  Returns the number deleted.
+        sweep (Feature 11) handles past-due instances instead.  ``today``
+        optionally pins the HA-local anchor the clamp compares against
+        (falling back to the host clock when omitted).  Any instance with
+        a completion event is never touched, regardless of its date.
+        Returns the number deleted.
         """
         _validate_date(cutoff_date, "cutoff_date")
         async with _connection_lock(self._database):
@@ -360,8 +520,8 @@ class QuestInstancesDao:
                 # midnight cannot delete instances that became past
                 # while it waited (backdating is clamped to the
                 # execution date, not the call date).
-                today = datetime.date.today().isoformat()
-                effective_cutoff = max(cutoff_date, today)
+                today_date = _resolve_today(today)
+                effective_cutoff = max(cutoff_date, today_date.isoformat())
                 result = await self._database.execute(
                     "DELETE FROM quest_instances WHERE definition_id = ? "
                     "AND due_date >= ? AND NOT EXISTS ("
@@ -372,6 +532,51 @@ class QuestInstancesDao:
                 )
         return result.rowcount
 
+    async def delete_future_uncompleted_for_child(
+        self, child_id: int, cutoff_date: str, *, today: datetime.date | None = None
+    ) -> tuple[int, str]:
+        """Delete the child's open instances at/after cutoff, across every
+        definition, returning ``(deleted, effective_cutoff)``.
+
+        Child-scoped counterpart to :meth:`delete_future_uncompleted`,
+        used when a child's presence changes (Feature 09): only instances
+        at or after the cutoff with NO completion event are removed, inside
+        one transaction — an instance that gains a completion event
+        mid-flight cannot be deleted.  The cutoff cannot be backdated below
+        today, for the same reason as the definition-scoped method (see its
+        docstring).  ``today`` optionally pins the HA-local anchor the clamp
+        compares against (falling back to the host clock when omitted).
+        Any instance with a completion event is never touched, regardless
+        of its date.
+
+        Returns the number deleted and the effective cutoff that was
+        actually applied (``max(cutoff_date, today)``, read under the
+        lock).  The caller (:func:`~.materialize.regenerate_for_child`)
+        passes that effective cutoff back as its re-materialization anchor,
+        so the delete and the rebuild share ONE execution-day even when the
+        call is queued across midnight — instead of the caller re-reading a
+        stale "today" and materializing a window the delete already
+        superseded.
+        """
+        _validate_date(cutoff_date, "cutoff_date")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                # "today" is read under the lock, matching the
+                # definition-scoped method: the DELETE executes on this
+                # date, so a caller queued across midnight cannot delete
+                # instances that became past while it waited.
+                today_date = _resolve_today(today)
+                effective_cutoff = max(cutoff_date, today_date.isoformat())
+                result = await self._database.execute(
+                    "DELETE FROM quest_instances WHERE child_id = ? "
+                    "AND due_date >= ? AND NOT EXISTS ("
+                    "  SELECT 1 FROM completion_events "
+                    "  WHERE instance_id = quest_instances.id"
+                    ")",
+                    (child_id, effective_cutoff),
+                )
+        return result.rowcount, effective_cutoff
+
 
 class CompletionEventsDao:
     """Typed append-only access to the ``completion_events`` table.

custom_components/nestquest/dao_presence.py

@@ -276,4 +276,59 @@ class PresenceOverridesDao:
             "DELETE FROM presence_overrides WHERE id = ?",
             (override_id,),
         )
-        return result.rowcount > 0
+        return result.rowcount > 0
+
+
+async def read_snapshot_unlocked(
+    database: NestQuestDatabase,
+    child_ids: list[int],
+    range_start: str,
+    range_end: str,
+) -> tuple[
+    dict[int, PresenceScheduleRecord],
+    dict[int, list[PresenceOverrideRecord]],
+]:
+    """Read schedules and overrides for ``child_ids`` WITHOUT locking.
+
+    MUST be called inside the connection lock and an open transaction, so
+    the schedules and overrides returned here stay coherent with every
+    other table the caller reads inside that same transaction (the
+    materialization walk's single input snapshot does exactly that).  The
+    date range comes already validated by the caller.
+
+    Returns ``(schedules, overrides)``:
+
+    - ``schedules`` maps ``child_id`` to its :class:`PresenceScheduleRecord`
+      for every child that HAS a schedule row.  A child absent from the
+      mapping has no schedule, i.e. is present every day (Feature 05).
+    - ``overrides`` maps ``child_id`` to that child's overrides overlapping
+      ``[range_start, range_end]`` (inclusive boundaries), ordered by start
+      date; a child with no matching override is absent from the mapping.
+
+    ``child_ids`` must name existing children (the caller derives them
+    from a trusted snapshot, within the same transaction).
+    """
+    children = sorted(set(child_ids))
+    if not children:
+        return {}, {}
+    placeholders = ",".join("?" for _ in children)
+    schedule_rows = await database.fetch_all(
+        f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules "
+        f"WHERE child_id IN ({placeholders})",
+        tuple(children),
+    )
+    override_rows = await database.fetch_all(
+        f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides "
+        f"WHERE child_id IN ({placeholders}) AND end_date >= ? "
+        "AND start_date <= ? ORDER BY start_date",
+        tuple(children) + (range_start, range_end),
+    )
+    schedules = {
+        record.child_id: record
+        for record in map(_schedule_from_row, schedule_rows)
+    }
+    overrides: dict[int, list[PresenceOverrideRecord]] = {}
+    for row in override_rows:
+        record = _override_from_row(row)
+        overrides.setdefault(record.child_id, []).append(record)
+    return schedules, overrides

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,24 @@ class QuestDefinitionWindowRecord:
     due_time: str | None
 
 
+@dataclass(frozen=True)
+class QuestDefinitionSnapshot:
+    """A definition plus its rule, assignees and windows, read atomically.
+
+    Returned by :meth:`QuestDefinitionsDao.create_with_rule_and_windows`
+    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]
+
+
 _RULE_COLUMNS = (
     "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
     "month, start_date, end_date"
@@ -188,6 +227,250 @@ 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 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:
+    """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 +501,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 +674,191 @@ 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)
+                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,
+        )
+
     async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
         """Raise ValueError unless the rule exists.
 
@@ -441,6 +907,81 @@ class QuestDefinitionsDao:
         )
         return [_definition_from_row(row) for row in rows]
 
+    async def _snapshots_for(
+        self, definitions: list[QuestDefinitionRecord]
+    ) -> list[QuestDefinitionSnapshot]:
+        """Read each definition's rule, assignees and windows.
+
+        MUST be called inside the connection lock (and ideally a
+        transaction) so the definitions handed in and the rows this
+        reads back cannot be raced by a concurrent edit between list
+        and bundle.  Each rule is read through
+        :class:`ScheduleRulesDao` on the same connection, so the
+        snapshot's rule record is the one the definition references AT
+        the moment of the locked read.
+        """
+        rules_dao = ScheduleRulesDao(self._database)
+        snapshots: list[QuestDefinitionSnapshot] = []
+        for definition in definitions:
+            rule_record = await rules_dao.get(definition.schedule_rule_id)
+            assert rule_record is not None
+            snapshots.append(
+                QuestDefinitionSnapshot(
+                    definition=definition,
+                    rule=rule_record,
+                    assignees=await self.list_assignees(definition.id),
+                    windows=await self.list_windows(definition.id),
+                )
+            )
+        return snapshots
+
+    async def list_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
+        """Return every active definition with its rule, assignees and
+        windows, all read inside one locked transaction.
+
+        The definitive list and each definition's linked rows are read
+        in a single BEGIN..COMMIT span under the connection lock, so a
+        concurrent atomic edit cannot interleave between the definition
+        list and the bundle read: the caller receives a coherent
+        snapshot (each definition's rule, assignees and windows as they
+        were at one instant), never a mixed old-rule/new-windows state.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definitions = await self.list_active()
+                return await self._snapshots_for(definitions)
+
+    async def read_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
+        """Return every active definition's snapshot WITHOUT locking.
+
+        Mirrors :meth:`list_snapshots_active` minus the lock and
+        transaction: the definitions and their rules, assignees and
+        windows are read on the caller's connection under whatever
+        transaction the caller is already holding, so they stay coherent
+        with every OTHER table the caller reads inside that same
+        transaction.  MUST be called inside the connection lock and an
+        open transaction (the single-transaction input snapshot the
+        materialization walk uses is that caller).
+        """
+        definitions = await self.list_active()
+        return await self._snapshots_for(definitions)
+
+    async def list_snapshots_by_child(
+        self, child_id: int
+    ) -> list[QuestDefinitionSnapshot]:
+        """Return each definition assigned to ``child_id`` with its rule,
+        assignees and windows, all read inside one locked transaction.
+
+        Mirrors :meth:`list_snapshots_active` but selects by assignment
+        (``list_by_child`` order: newest first).  Locks read-and-bundle
+        into one transaction so the bundle never mixes one edit's
+        definition with another's windows.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definitions = await self.list_by_child(child_id)
+                return await self._snapshots_for(definitions)
+
     async def update(
         self,
         definition_id: int,
@@ -496,14 +1037,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.
         """
@@ -524,6 +1069,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
@@ -658,4 +1204,47 @@ class QuestDefinitionsDao:
         if not child[0]:
             raise ValueError(
                 f"child {child_id} is inactive and cannot be assigned"
-            )
+            )
+
+
+async def load_materialization_input(
+    database: NestQuestDatabase,
+    range_start: str,
+    range_end: str,
+):
+    """Read every materialization input in ONE locked transaction.
+
+    Returns ``(snapshots, schedules, overrides)``:
+
+    - ``snapshots``: every active definition with its rule, assignees and
+      windows (the :class:`QuestDefinitionSnapshot` shape).
+    - ``schedules`` / ``overrides``: the presence schedules and
+      range-scoped overrides of every assignee child, in
+      :mod:`.dao_presence`'s record shapes.
+
+    The whole read — definitions, rules, assignees, windows, and the
+    assignees' presence schedules and overrides — runs inside a single
+    BEGIN..COMMIT span under the connection lock, so a concurrent edit
+    (definition, assignment, window, or presence) cannot interleave
+    between the parts and the walk never sees a mixed-time view.  Each
+    table's SQL stays in its own DAO module: the presence portion is
+    delegated to :mod:`.dao_presence` (a lazy import here, since that
+    module imports this one).
+    """
+    definitions_dao = QuestDefinitionsDao(database)
+    async with _connection_lock(database):
+        async with database.transaction():
+            snapshots = await definitions_dao.read_snapshots_active()
+            child_ids = sorted(
+                {
+                    child.id
+                    for snapshot in snapshots
+                    for child in snapshot.assignees
+                }
+            )
+            from . import dao_presence  # lazy: dao_presence imports us
+
+            schedules, overrides = await dao_presence.read_snapshot_unlocked(
+                database, child_ids, range_start, range_end
+            )
+    return snapshots, schedules, overrides

custom_components/nestquest/materialize.py

@@ -0,0 +1,377 @@
+"""The materialization walk (Feature 07 core).
+
+:func:`materialize` turns the schedule rules + custody presence of the
+active quest definitions into concrete ``quest_instances`` rows over a
+date range.  It is the pure walk over definitions x dates x assignees x
+windows; HA scheduling (day-rollover triggers) is a separate later
+task and lives elsewhere.
+
+The walk is deterministic and idempotent:
+
+- All inputs — the active definitions with their rules, assignees and
+  windows, AND the assignee children's presence schedules and scoped
+  overrides — are read inside ONE locked transaction (see
+  :func:`~.dao_rules.load_materialization_input`), so the walk sees a
+  single coherent snapshot rather than a definition state from one
+  instant and a presence state from another.
+- Presence is snapshotted into an immutable
+  :class:`~.presence.PresenceEngine` from that same coherent read.
+- For each date in the range, each definition whose decoded rule
+  :func:`~.recurrence.occurs_on` on that date, each assignee who is
+  ``is_present`` on that date and each declared window, one instance is
+  written through :class:`~.dao_instances.QuestInstancesDao.upsert_if_valid`
+  with a ``generated_at`` stamp shared by the whole batch (one coherent
+  time per run).
+
+A config change that lands AFTER the snapshot is read but BEFORE a
+tuple's write (an assignment removed, a definition or child deactivated,
+a window removed) makes that tuple SKIP rather than abort the batch:
+:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid` re-validates
+every materialization precondition (active definition, active child,
+live assignment, declared window, and its CURRENT ``due_time``) inside
+the SAME transaction as the INSERT, so the check and the write cannot be
+split by a racing config edit.
+
+The upsert is idempotent on (definition_id, child_id, due_date,
+window), so re-running the materialization over the same range never
+duplicates a row; a tuple whose instance is already completed is SKIPPED
+(never rewritten, never raised) by
+:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`.
+
+Regeneration: a config or presence change must rebuild future instances
+so they track the new state.  :func:`regenerate_for_definition` covers
+definition edits / reassignment (Features 06/07); :func:`regenerate_for_child`
+covers a single child's presence change (Feature 09).  Both delete the
+affected open instances at or after today — never a completed instance,
+never the past — and re-run the walk over the rolling horizon
+``[today, today + horizon_days]`` (``horizon_days`` defaults to
+``DEFAULT_HORIZON_DAYS``).
+
+HOOK (Feature 09 presence services): the presence business layer that
+lands in Feature 09 MUST call :func:`regenerate_for_child` after EVERY
+presence-state write so a child's future instances track its presence:
+``set_presence_pattern`` (including clearing/setting an empty pattern),
+deleting/clearing the presence schedule
+(:meth:`~.dao_presence.PresenceSchedulesDao.delete`),
+``create_presence_override``, and ``delete_presence_override``.  A
+schedule delete changes presence state too (the child becomes
+always-present), so it must regenerate just like the others.  This hook
+is documented, not yet wired: no presence business layer exists yet, and
+the presence DAO (``dao_presence.py``) must stay free of business rules
+— never call :func:`regenerate_for_child` from inside the DAO layer.
+"""
+from __future__ import annotations
+
+import datetime
+
+from .const import DEFAULT_HORIZON_DAYS
+from .dao_instances import QuestInstancesDao, _resolve_today
+from .dao_rules import (
+    _validate_date,
+    load_materialization_input,
+    schedule_rule_from_storage,
+    schedule_rule_storage_from_record,
+)
+from .db import NestQuestDatabase
+from .presence import PresenceEngine, PresenceOverride, PresenceSchedule
+from .recurrence import ScheduleRule, occurs_on
+
+
+def _now_stamp() -> str:
+    """Return the batch's shared UTC second-precision generation stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _resolve_horizon_days(horizon_days: int | None) -> int:
+    """Validate a regeneration horizon and return the resolved day count.
+
+    ``None`` resolves to :data:`~.const.DEFAULT_HORIZON_DAYS`.  A bool
+    (SQLite would happily bind ``True``/``False`` onto an int), a non-int,
+    or a value below 1 raises ValueError naming the field — the validation
+    runs BEFORE any future-instance delete so a malformed horizon can never
+    delete rows it then fails to rebuild.  A negative horizon would
+    otherwise delete first and only then be rejected by
+    :func:`materialize`'s inverted-range check; ``0``/``False`` would
+    silently shrink the window to a single day.
+    """
+    if horizon_days is None:
+        return DEFAULT_HORIZON_DAYS
+    if (
+        isinstance(horizon_days, bool)
+        or not isinstance(horizon_days, int)
+        or horizon_days < 1
+    ):
+        raise ValueError(
+            f"horizon_days must be a positive integer, got {horizon_days!r}"
+        )
+    return horizon_days
+
+
+def _decode_rule(snapshot) -> ScheduleRule:
+    """Decode a snapshot's rule record back into a ScheduleRule.
+
+    The record carried by the snapshot is the one the definition
+    referenced at read time, so decoding it here (rather than re-reading
+    through ``ScheduleRulesDao``) preserves the coherence the snapshot
+    guarantees.
+    """
+    return schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+
+
+def _build_engine(
+    schedules_records,
+    overrides_records,
+) -> PresenceEngine:
+    """Convert raw presence records into an immutable PresenceEngine."""
+    schedules: dict[int, PresenceSchedule] = {}
+    for child_id, record in schedules_records.items():
+        schedules[child_id] = PresenceSchedule.decode(
+            child_id, record.anchor_date, record.pattern
+        )
+    overrides: dict[int, list[PresenceOverride]] = {}
+    for child_id, records in overrides_records.items():
+        overrides[child_id] = [
+            PresenceOverride(
+                record.child_id,
+                record.start_date,
+                record.end_date,
+                record.is_present,
+                record.note,
+            )
+            for record in records
+        ]
+    return PresenceEngine(schedules, overrides)
+
+
+async def materialize(
+    database: NestQuestDatabase,
+    start_date: str,
+    end_date: str,
+    *,
+    child_ids: list[int] | None = None,
+    today: datetime.date | None = None,
+) -> int:
+    """Generate quest instances for the closed range [start_date, end_date].
+
+    ``start_date`` and ``end_date`` must be strict ISO calendar dates
+    (YYYY-MM-DD) with ``end_date`` on or after ``start_date``; anything
+    else raises ValueError naming the offending field.  Returns the number
+    of instances upserted for the range (idempotent — a re-run refreshes
+    the same rows, never duplicates them).
+
+    ``child_ids`` optionally scopes the walk to exactly those children:
+    assignees outside the list are never written, so a caller that only
+    wants one child's instances rebuilt (e.g. :func:`regenerate_for_child`
+    after a presence change) cannot touch any unrelated child's rows.
+    When omitted (or None) the walk covers every assignee as before.
+
+    ``today`` optionally pins the resolved HA-local date the walk's
+    no-past guard compares against; it is threaded into every
+    :meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid` write.  When
+    omitted the host clock (``datetime.date.today``) is used, which is the
+    historical default.  A caller that computed the horizon in Home
+    Assistant's configured time zone MUST pass that same date here so a
+    time zone behind the host around midnight is not rejected as "past".
+
+    ``start_date`` is clamped up to the resolved "today" (the same
+    ``today`` anchor threaded into the no-past guard), so a caller that
+    passes a ``start_date`` already in the past never yields a
+    past-dated instance: the walk simply begins on today and skips every
+    earlier date.  A range entirely in the past therefore produces zero
+    instances rather than an error, and the DAO's no-past guard remains
+    the authoritative backstop under the lock.
+
+    All inputs are read in ONE locked transaction; each tuple's write then
+    goes through the atomic :meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`,
+    which re-validates the tuple and its current ``due_time`` in the same
+    transaction as the INSERT, so a config change that arrives mid-batch
+    skips the affected tuples rather than aborting the batch or leaving a
+    partial result.
+    """
+    _validate_date(start_date, "start_date")
+    _validate_date(end_date, "end_date")
+    if end_date < start_date:
+        raise ValueError(
+            f"end_date must be on or after start_date, got "
+            f"{end_date!r} < {start_date!r}"
+        )
+
+    snapshots, schedules_records, overrides_records = await (
+        load_materialization_input(database, start_date, end_date)
+    )
+    engine = _build_engine(schedules_records, overrides_records)
+    child_filter = None if child_ids is None else set(child_ids)
+
+    definitions = [
+        (snapshot.definition.id, _decode_rule(snapshot),
+         snapshot.assignees, snapshot.windows)
+        for snapshot in snapshots
+    ]
+
+    generated_at = _now_stamp()
+    instances = QuestInstancesDao(database)
+
+    start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date()
+    end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date()
+
+    # Clamp a past start up to today (the HA-local anchor when the caller
+    # pinned one, else the host clock) so the walk can never emit a
+    # past-dated instance; the DAO's no-past guard remains the backstop.
+    # A range entirely in the past therefore yields zero instances.
+    today_date = _resolve_today(today)
+    start = max(start, today_date)
+
+    count = 0
+    cursor = start
+    while cursor <= end:
+        iso = cursor.isoformat()
+        for definition_id, rule, assignees, windows in definitions:
+            if not occurs_on(rule, cursor):
+                continue
+            for child in assignees:
+                if child_filter is not None and child.id not in child_filter:
+                    continue
+                if not engine.is_present(child.id, cursor):
+                    continue
+                for window in windows:
+                    written = await instances.upsert_if_valid(
+                        definition_id,
+                        child.id,
+                        iso,
+                        generated_at,
+                        window=window.window,
+                        due_time=window.due_time,
+                        today=today,
+                    )
+                    if written is not None:
+                        count += 1
+        cursor += datetime.timedelta(days=1)
+    return count
+
+
+async def regenerate_for_definition(
+    database: NestQuestDatabase,
+    definition_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> int:
+    """Regenerate a definition's future instances after a config change.
+
+    The Feature 06 edit/assignment/activation paths call this after a
+    successful write so future instances track the new state: the
+    definition's open instances at or after today are deleted (never a
+    completed instance, never the past) via
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted`,
+    then the materialization walk re-runs over the rolling horizon
+    ``[today, today + horizon_days]`` so the (changed) rule,
+    assignee set and windows re-materialize against today's state.
+
+    "today" is computed the same way the walk computes it —
+    :func:`~.dao_instances._today` — so the delete cutoff and the
+    re-materialization range share one anchor, and the walk's
+    ``upsert_if_valid`` no-past guard never rejects the regenerated
+    range.  ``today`` optionally pins a caller-resolved HA-local date
+    (threaded into both the delete cutoff and the re-materialization)
+    so a household time zone behind the host around midnight stays
+    consistent.  ``horizon_days`` sizes the re-materialization window;
+    when omitted it falls back to :data:`~.const.DEFAULT_HORIZON_DAYS`.
+    A bool, non-int, or sub-1 value raises ValueError BEFORE any delete.
+    Returns the number of instances the re-materialization
+    upserted across the whole snapshot (idempotent — other active
+    definitions' tuples refresh in place, never duplicate).
+    """
+    horizon = _resolve_horizon_days(horizon_days)
+    today_date = _resolve_today(today)
+    start_date = today_date.isoformat()
+    end_date = (
+        today_date + datetime.timedelta(days=horizon)
+    ).isoformat()
+
+    instances = QuestInstancesDao(database)
+    await instances.delete_future_uncompleted(
+        definition_id, start_date, today=today
+    )
+    return await materialize(database, start_date, end_date, today=today)
+
+
+# HOOK (Feature 09 presence services): ``regenerate_for_child`` is the
+# child-presence counterpart to ``regenerate_for_definition``.  The
+# presence business layer that lands in Feature 09 MUST call
+# ``regenerate_for_child`` after EVERY presence-state write, so future
+# instances track a child's changed presence: set_presence_pattern
+# (including clearing/setting an empty pattern), deleting/clearing the
+# presence schedule (PresenceSchedulesDao.delete),
+# create_presence_override, and delete_presence_override.  A schedule
+# delete changes presence state too (the child becomes always-present),
+# so it must regenerate just like the others.  It is deliberately NOT
+# wired into ``dao_presence.py``: the DAO layer stays pure storage (no
+# business rules), and no presence business layer exists yet to host the
+# call — see the module docstring note above.
+async def regenerate_for_child(
+    database: NestQuestDatabase,
+    child_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> int:
+    """Regenerate a child's future instances after a presence change.
+
+    The Feature 09 presence services call this after every
+    presence-state write — set_presence_pattern (including clearing or
+    setting an empty pattern), deleting/clearing the presence schedule
+    (:meth:`~.dao_presence.PresenceSchedulesDao.delete`),
+    create_presence_override, and delete_presence_override — so a
+    child's future instances track its new presence: the child's open
+    instances at or after today are deleted across ALL its definitions
+    (never a completed instance, never the past) via
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted_for_child`,
+    then the materialization walk re-runs over the rolling horizon
+    ``[today, today + horizon_days]`` scoped to that child only,
+    so the child's now-absent/present dates re-materialize against
+    today's presence WITHOUT touching any other child's rows.
+
+    ``child_id`` must be a plain int — bool and float are rejected
+    (SQLite binds a bool as 0/1 and a float would round, so a malformed
+    id must never mutate the wrong profile).
+
+    "today" is computed the same way the walk computes it —
+    :func:`~.dao_instances._today` — and the re-materialization range uses the
+    delete's returned effective cutoff (see
+    :meth:`~.dao_instances.QuestInstancesDao.delete_future_uncompleted_for_child`),
+    so the delete cutoff and the re-materialization window share ONE
+    execution-day anchor even when the call is queued across midnight.
+    ``today`` optionally pins a caller-resolved HA-local date, threaded
+    into the delete cutoff and the re-materialization for the same
+    host-vs-HA time-zone consistency.  ``horizon_days`` sizes the
+    re-materialization window; when omitted it falls back to
+    :data:`~.const.DEFAULT_HORIZON_DAYS`.  A bool, non-int, or sub-1 value
+    raises ValueError BEFORE any delete.  Returns the number of instances
+    the re-materialization upserted for the child across the whole
+    snapshot.
+    """
+    if isinstance(child_id, bool) or not isinstance(child_id, int):
+        raise ValueError(f"child_id must be an integer, got {child_id!r}")
+
+    horizon = _resolve_horizon_days(horizon_days)
+    today_date = _resolve_today(today)
+
+    instances = QuestInstancesDao(database)
+    # Use the delete's returned effective cutoff as the materialize anchor,
+    # NOT the independently-computed "today" above.  If the call is queued
+    # across midnight the delete clamps its cutoff to the execution-day;
+    # materializing from the stale "today" (now yesterday) would trip the
+    # walk's no-past guard and leave the child's future instances deleted
+    # but not rebuilt.  One anchor for both keeps them consistent.
+    _, effective_cutoff = await instances.delete_future_uncompleted_for_child(
+        child_id, today_date.isoformat(), today=today
+    )
+    start = datetime.date.fromisoformat(effective_cutoff)
+    end_date = (start + datetime.timedelta(days=horizon)).isoformat()
+    return await materialize(
+        database, start.isoformat(), end_date, child_ids=[child_id], today=today
+    )

custom_components/nestquest/quest_definitions.py

@@ -0,0 +1,603 @@
+"""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/edit here, and never touch the DAO SQL directly.
+: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`:
+
+- ``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, _connection_lock
+from .dao_rules import (
+    QuestDefinitionRecord,
+    QuestDefinitionSnapshot,
+    QuestDefinitionWindowRecord,
+    QuestDefinitionsDao,
+    ScheduleRuleStorage,
+    ScheduleRulesDao,
+    _UNSET,
+    _validate_date,
+    _validate_time,
+    _validate_window,
+    schedule_rule_from_storage,
+    schedule_rule_storage_from_record,
+    schedule_rule_to_storage,
+)
+from .db import NestQuestDatabase
+from .materialize import regenerate_for_definition
+from .recurrence import ScheduleRule, occurs_on
+
+
+@dataclass(frozen=True)
+class CreatedQuestDefinition:
+    """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
+    ``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
+
+
+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
+
+
+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,
+    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,
+    )
+
+
+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,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> 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.
+
+    ``today`` optionally pins the caller-resolved HA-local date the
+    regeneration's no-past guard and horizon are anchored to (threaded
+    into :func:`~.materialize.regenerate_for_definition`); it is a plain
+    ``datetime.date`` with no HA import here — the Feature 09 service
+    supplies the hass-derived value.  When omitted, regeneration falls
+    back to the host clock, same as before.  ``horizon_days`` sizes the
+    regeneration's re-materialization window the same way (threaded into
+    :func:`~.materialize.regenerate_for_definition`); it defaults to
+    :data:`~.const.DEFAULT_HORIZON_DAYS` when omitted.
+    """
+    _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)
+    try:
+        snapshot = await dao.edit_definition(
+            definition_id,
+            title=title_value,
+            description=description_value,
+            icon=icon_value,
+            rule=rule_storage,
+            windows=window_specs,
+        )
+    except ValueError as error:
+        # The DAO's missing-definition error names only the id; re-raise
+        # so the public contract names the field the caller passed (matching
+        # assign/unassign/deactivate).
+        message = str(error)
+        if message.startswith("quest definition "):
+            raise ValueError(f"definition_id: {message}") from error
+        raise
+
+    decoded_rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+
+    await regenerate_for_definition(
+        database, definition_id, today=today, horizon_days=horizon_days
+    )
+
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=decoded_rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.windows,
+    )
+
+
+async def assign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> 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.
+
+    ``today`` optionally pins the caller-resolved HA-local date threaded
+    into the assignment's regeneration (see :func:`edit_quest_definition`).
+    ``horizon_days`` optionally sizes the regeneration's re-materialization
+    window the same way.
+    """
+    _validate_definition_id(definition_id)
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    try:
+        assignees = 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
+    await regenerate_for_definition(
+        database, definition_id, today=today, horizon_days=horizon_days
+    )
+    return assignees
+
+
+async def unassign_child(
+    database: NestQuestDatabase,
+    definition_id: int,
+    child_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> 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.
+
+    ``today`` optionally pins the caller-resolved HA-local date threaded
+    into the unassignment's regeneration (see :func:`edit_quest_definition`).
+    ``horizon_days`` optionally sizes the regeneration's re-materialization
+    window the same way.
+    """
+    _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"
+        )
+    removed = await dao.remove_assignee(definition_id, child_id)
+    await regenerate_for_definition(
+        database, definition_id, today=today, horizon_days=horizon_days
+    )
+    return removed
+
+
+async def set_quest_definition_active(
+    database: NestQuestDatabase,
+    definition_id: int,
+    is_active: bool,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> 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.
+
+    ``today`` optionally pins the caller-resolved HA-local date threaded
+    into the activation change's regeneration (see
+    :func:`edit_quest_definition`).  ``horizon_days`` optionally sizes the
+    regeneration's re-materialization window the same way.
+    """
+    _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)
+    )
+    await regenerate_for_definition(
+        database, definition_id, today=today, horizon_days=horizon_days
+    )
+    return CreatedQuestDefinition(
+        definition=updated,
+        rule=rule,
+        assignees=assignees,
+        windows=windows,
+    )
+
+
+def _bundle_snapshot(
+    snapshot: QuestDefinitionSnapshot,
+) -> CreatedQuestDefinition:
+    """Decode a DAO snapshot into the layer's rich view.
+
+    The rule record carried by the snapshot is the one the definition
+    referenced at read time, so decoding it here (rather than re-reading
+    through :class:`~.dao_rules.ScheduleRulesDao`) preserves the
+    coherence the DAO's locked snapshot guarantees: the rule, assignees
+    and windows all come from one atomic read, never three separate ones.
+    """
+    rule = schedule_rule_from_storage(
+        schedule_rule_storage_from_record(snapshot.rule)
+    )
+    return CreatedQuestDefinition(
+        definition=snapshot.definition,
+        rule=rule,
+        assignees=snapshot.assignees,
+        windows=snapshot.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.  The
+    definitions and their links are read by the DAO inside one locked
+    transaction, so each bundle is a coherent snapshot that a concurrent
+    edit cannot interleave.
+    """
+    dao = QuestDefinitionsDao(database)
+    snapshots = await dao.list_snapshots_active()
+    return [_bundle_snapshot(snapshot) for snapshot in snapshots]
+
+
+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, read atomically by the DAO.
+    """
+    _validate_child_id(child_id)
+    dao = QuestDefinitionsDao(database)
+    snapshots = await dao.list_snapshots_by_child(child_id)
+    bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
+    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.
+    The definitions and their rules are read inside one locked
+    transaction, so a bundle can never filter on one rule while
+    presenting another state's windows.
+    """
+    _validate_date(date, "date")
+    target = datetime.datetime.strptime(date, "%Y-%m-%d").date()
+    dao = QuestDefinitionsDao(database)
+    snapshots = await dao.list_snapshots_active()
+    bundles = [_bundle_snapshot(snapshot) for snapshot in snapshots]
+    return [bundle for bundle in bundles if occurs_on(bundle.rule, target)]

custom_components/nestquest/services.yaml

@@ -0,0 +1,12 @@
+# Service declarations for the NestQuest integration (Home Assistant
+# services.yaml metadata). Services are domain-scoped: nested under the
+# "nestquest" domain, not repeated per config entry.
+
+regenerate:
+  name: Regenerate quest instances
+  description: >-
+    Regenerate every future quest instance for the household over the rolling
+    horizon, using the same idempotent materialization as the daily rollover
+    and the startup backfill. Safe to call at any time: a re-run refreshes the
+    same rows instead of duplicating them, and completed instances are never
+    touched.

tests/conftest.py

@@ -44,6 +44,7 @@ _HA_MODULES = (
     "homeassistant.exceptions",
     "homeassistant.helpers",
     "homeassistant.helpers.config_validation",
+    "homeassistant.helpers.event",
 )
 
 def _find_spec(name: str):
@@ -83,6 +84,7 @@ for _parent, _child in (
     ("homeassistant", "exceptions"),
     ("homeassistant", "helpers"),
     ("homeassistant.helpers", "config_validation"),
+    ("homeassistant.helpers", "event"),
 ):
     setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
 
@@ -242,6 +244,113 @@ class ListenerRegistry:
         return sum(len(listeners) for listeners in self._listeners.values())
 
 
+class TimeChangeRegistry:
+    """Models HA's time-change listener bookkeeping.
+
+    ``track`` mirrors ``homeassistant.helpers.event.async_track_time_change``:
+    each call appends a record (action + hour/minute/second) and returns
+    a remove callable that reverses the registration.  Tests inspect
+    ``registrations`` and exercise the newest action through :meth:`fire`.
+    """
+
+    def __init__(self, hass=None):
+        self._hass = hass
+        self._registrations: list[dict] = []
+
+    def track(self, action, hour=None, minute=None, second=None):
+        record = {
+            "action": action,
+            "hour": hour,
+            "minute": minute,
+            "second": second,
+        }
+
+        def _remove():
+            self._registrations.remove(record)
+
+        record["remove"] = _remove
+        self._registrations.append(record)
+        return _remove
+
+    @property
+    def registrations(self):
+        """Return a copy of the live registrations."""
+        return list(self._registrations)
+
+    @property
+    def size(self) -> int:
+        """Number of live registrations."""
+        return len(self._registrations)
+
+    async def fire(self, now=None, index: int = -1):
+        """Run a registered action as HA's scheduler would, awaiting coroutines."""
+        result = self._registrations[index]["action"](now)
+        if inspect.isawaitable(result):
+            await result
+
+
+def _async_track_time_change(hass, action, hour=None, minute=None, second=None):
+    """Mirror homeassistant.helpers.event.async_track_time_change.
+
+    Exact HA 2024.6 signature — no ``local`` keyword: the local-time
+    wrapper is already this function.  Routes to the registry attached to
+    ``hass`` (created on demand), so tests reach the registrations via
+    ``hass.time_change``.
+    """
+    registry = getattr(hass, "time_change", None)
+    if registry is None:
+        registry = TimeChangeRegistry(hass)
+        hass.time_change = registry
+    return registry.track(action, hour=hour, minute=minute, second=second)
+
+
+_ha_mock("homeassistant.helpers.event").async_track_time_change = _async_track_time_change
+
+
+class ServiceRegistry:
+    """Models HA's ``hass.services`` service registry.
+
+    ``async_register`` stores a handler under its (domain, service) key,
+    ``has_service`` reports registration, and ``async_remove`` removes it
+    — exactly the surface Home Assistant's ``ServiceRegistry`` exposes
+    (there is deliberately NO ``async_unregister`` alias here, so a caller
+    that uses the wrong method name fails instead of being silently
+    papered over).  ``call`` invokes a stored handler the way HA's service
+    bus would — awaiting a coroutine result — so tests can exercise a
+    registered service end-to-end.
+    """
+
+    def __init__(self, hass=None):
+        self._hass = hass
+        self._services: dict[tuple[str, str], object] = {}
+
+    def async_register(self, domain, service, func, schema=None, supports_response=None):
+        self._services[(domain, service)] = func
+        return None
+
+    def has_service(self, domain, service):
+        return (domain, service) in self._services
+
+    def async_remove(self, domain, service):
+        self._services.pop((domain, service), None)
+        return None
+
+    async def call(self, domain, service, data=None):
+        func = self._services.get((domain, service))
+        if func is None:
+            raise KeyError(f"service {domain}.{service} not registered")
+        call = SimpleNamespace(domain=domain, service=service, data=data or {})
+        result = func(call)
+        if inspect.isawaitable(result):
+            await result
+        return result
+
+    @property
+    def registered_services(self):
+        """The registered (domain, service) pairs."""
+        return list(self._services)
+
+
 def make_config_entry(
     entry_id: str = "test_entry",
     options: dict | None = None,
@@ -303,6 +412,11 @@ def make_hass() -> tuple:
 
     hass.async_add_executor_job = _run_executor_job
     registry = ListenerRegistry(hass)
+    hass.time_change = TimeChangeRegistry(hass)
+    hass.services = ServiceRegistry(hass)
+    # A valid IANA time zone (the day-rollover listener reads it to
+    # compute "today" in HA local time).
+    hass.config.time_zone = "UTC"
 
     async def _async_reload(entry_id: str) -> None:
         registry.reloaded.append(entry_id)

tests/test_completion.py

@@ -0,0 +1,686 @@
+"""Tests for completion.py: the Feature 08 append-only business layer."""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import inspect
+import re
+
+import pytest
+
+from custom_components.nestquest.completion import (
+    append_event,
+    complete_instance,
+    instance_state,
+)
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_instances import (
+    EVENT_COMPLETED,
+    EVENT_UNCOMPLETED,
+    CompletionEventRecord,
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from custom_components.nestquest.dao_rules import (
+    QuestDefinitionsDao,
+    ScheduleRulesDao,
+)
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+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
+
+
+def _today_iso() -> str:
+    return datetime.date.today().isoformat()
+
+
+def _now_stamp() -> str:
+    return f"{_today_iso()}T08:00:00+00:00"
+
+
+def _future_day(offset: int) -> str:
+    return (
+        datetime.date.today() + datetime.timedelta(days=offset)
+    ).isoformat()
+
+
+D1 = _future_day(1)
+
+
+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)
+    instances = QuestInstancesDao(database)
+    child = await children.create("Ada", _now_stamp())
+    rule = await rules.create("daily", D1)
+    definition = await definitions.create(
+        "Brush teeth", rule.id, _now_stamp(), assignee_child_ids=[child.id]
+    )
+    instance = await instances.upsert(
+        definition.id, child.id, D1, _now_stamp(), window="morning"
+    )
+    return database, child, instance
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            prepared = await _prepare(tmp_path / name)
+            try:
+                return await body(*prepared)
+            finally:
+                await prepared[0].close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# inverted actor shapes: field-naming ValueErrors, no write
+# ---------------------------------------------------------------------------
+
+
+def test_user_without_actor_user_id_rejected() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                was_on_time=True,
+            )
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id=None,
+                was_on_time=True,
+            )
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_user_must_not_carry_actor_child_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_child_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="user-1",
+                actor_child_id=1,
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_panel_must_not_carry_actor_user_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="panel",
+                actor_user_id="user-1",
+                actor_child_id=1,
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_panel_requires_actor_child_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_child_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="panel",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_actor_source_service_rejected() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_source"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="service",
+                actor_user_id="user-1",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_completed_requires_was_on_time() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="was_on_time"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _run(_body())
+
+
+# ---------------------------------------------------------------------------
+# happy paths
+# ---------------------------------------------------------------------------
+
+
+def test_user_append_round_trips(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        assert isinstance(event, CompletionEventRecord)
+        assert event.event_type == EVENT_COMPLETED
+        assert event.actor_source == "user"
+        assert event.actor_user_id == "user-1"
+        assert event.actor_child_id is None
+        assert event.was_on_time is True
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", event.occurred_at
+        )
+        return event
+
+    _with_db(tmp_path, "user-append.db")(_body)
+
+
+def test_panel_append_round_trips(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="panel",
+            actor_child_id=child.id,
+            was_on_time=False,
+        )
+        assert event.actor_source == "panel"
+        assert event.actor_user_id is None
+        assert event.actor_child_id == child.id
+        assert event.was_on_time is False
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", event.occurred_at
+        )
+        return event
+
+    _with_db(tmp_path, "panel-append.db")(_body)
+
+
+def test_uncompleted_allows_null_was_on_time(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=None,
+        )
+        assert event.event_type == EVENT_UNCOMPLETED
+        assert event.was_on_time is None
+        return event
+
+    _with_db(tmp_path, "uncompleted-null.db")(_body)
+
+
+def test_now_pins_occurred_at_utc(tmp_path) -> None:
+    async def _body(database, child, instance):
+        pinned = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+            now=pinned,
+        )
+        assert event.occurred_at == "2026-09-18T08:00:00+00:00"
+        return event
+
+    _with_db(tmp_path, "pinned-now.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# append-only: the module never exposes update/delete
+# ---------------------------------------------------------------------------
+
+
+def test_module_never_exposes_update_or_delete() -> None:
+    import custom_components.nestquest.completion as completion
+
+    public = {
+        name
+        for name, _ in inspect.getmembers(completion, inspect.isfunction)
+        if not name.startswith("_")
+    }
+    forbidden = {
+        name
+        for name in public
+        if name.startswith(
+            (
+                "update",
+                "delete",
+                "set",
+                "clear",
+                "remove",
+                "modify",
+                "edit",
+                "upsert",
+            )
+        )
+    }
+    assert forbidden == set(), (
+        f"completion.py exposed mutation methods: {forbidden}"
+    )
+    assert "append_event" in public
+    assert "instance_state" in public
+    assert "complete_instance" in public
+    assert "uncomplete_instance" not in public
+    source = inspect.getsource(completion)
+    assert "UPDATE" not in source
+    assert "DELETE" not in source
+    assert "INSERT" not in source
+
+
+# ---------------------------------------------------------------------------
+# instance_state: derived from the latest event
+# ---------------------------------------------------------------------------
+
+
+def test_instance_state_open_when_no_events(tmp_path) -> None:
+    async def _body(database, child, instance):
+        assert await instance_state(database, instance.id) == "open"
+
+    _with_db(tmp_path, "state-none.db")(_body)
+
+
+def test_instance_state_open_when_latest_uncompleted(tmp_path) -> None:
+    async def _body(database, child, instance):
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        assert await instance_state(database, instance.id) == "open"
+
+    _with_db(tmp_path, "state-uncompleted.db")(_body)
+
+
+def test_instance_state_unknown_instance_id_raises(tmp_path) -> None:
+    async def _body(database, child, instance):
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(database, instance.id + 999)
+
+    _with_db(tmp_path, "state-unknown.db")(_body)
+
+
+def test_instance_state_rejects_non_integer_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(object(), "1")
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(object(), True)
+
+    _run(_body())
+
+
+def test_instance_state_three_events_returns_done_and_preserves_rows(
+    tmp_path,
+) -> None:
+    async def _body(database, child, instance):
+        base = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+            now=base,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            now=base + datetime.timedelta(seconds=1),
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=False,
+            now=base + datetime.timedelta(seconds=2),
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert [e.event_type for e in before] == [
+            EVENT_COMPLETED,
+            EVENT_UNCOMPLETED,
+            EVENT_COMPLETED,
+        ]
+        assert await instance_state(database, instance.id) == "done"
+        after = await events.list_by_instance(instance.id)
+        assert after == before
+        assert len(after) == 3
+
+    _with_db(tmp_path, "state-three.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# complete_instance: append completed, no-op if already done
+# ---------------------------------------------------------------------------
+
+
+def test_complete_instance_appends_and_returns_done(tmp_path) -> None:
+    async def _body(database, child, instance):
+        events = CompletionEventsDao(database)
+        assert await events.list_by_instance(instance.id) == []
+        state = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        assert state == "done"
+        assert await instance_state(database, instance.id) == "done"
+        rows = await events.list_by_instance(instance.id)
+        assert len(rows) == 1
+        assert rows[0].event_type == EVENT_COMPLETED
+        assert rows[0].actor_source == "user"
+        assert rows[0].actor_user_id == "user-1"
+        assert rows[0].was_on_time is True
+
+    _with_db(tmp_path, "complete-open.db")(_body)
+
+
+def test_complete_instance_already_done_is_noop(tmp_path) -> None:
+    async def _body(database, child, instance):
+        first = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert first == "done"
+        assert len(before) == 1
+        second = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-2",
+        )
+        after = await events.list_by_instance(instance.id)
+        assert second == "done"
+        assert after == before
+        assert after[0].actor_user_id == "user-1"
+
+    _with_db(tmp_path, "complete-noop.db")(_body)
+
+
+def test_complete_instance_unknown_instance_id_raises(tmp_path) -> None:
+    async def _body(database, child, instance):
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                database,
+                instance.id + 999,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _with_db(tmp_path, "complete-unknown.db")(_body)
+
+
+def test_complete_instance_rejects_non_integer_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                object(),
+                "1",
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                object(),
+                True,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _run(_body())
+
+
+def test_complete_instance_after_uncompleted_appends(tmp_path) -> None:
+    async def _body(database, child, instance):
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert await instance_state(database, instance.id) == "open"
+        state = await complete_instance(
+            database,
+            instance.id,
+            actor_source="panel",
+            actor_child_id=child.id,
+        )
+        assert state == "done"
+        after = await events.list_by_instance(instance.id)
+        assert len(after) == len(before) + 1
+        assert after[-1].event_type == EVENT_COMPLETED
+        assert after[-1].actor_source == "panel"
+        assert after[-1].was_on_time is True
+
+    _with_db(tmp_path, "complete-reopen.db")(_body)
+
+
+def test_complete_instance_now_pins_occurred_at(tmp_path) -> None:
+    async def _body(database, child, instance):
+        pinned = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+            now=pinned,
+        )
+        rows = await CompletionEventsDao(database).list_by_instance(
+            instance.id
+        )
+        assert rows[0].occurred_at == "2026-09-18T08:00:00+00:00"
+
+    _with_db(tmp_path, "complete-pinned.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 INSERT of whichever ``complete_instance`` starts first
+    — proving that caller holds the connection lock 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_complete_instance_concurrent_second_call_is_noop(tmp_path) -> None:
+    """Two racing completes cannot both append.
+
+    Deterministic gate: task A is paused at its INSERT (it holds the
+    connection lock); task B is then started and must queue.  A
+    finishes, B sees done and appends nothing.
+    """
+    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 / "complete-race.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            rules = ScheduleRulesDao(database)
+            definitions = QuestDefinitionsDao(database)
+            instances = QuestInstancesDao(database)
+            child = await children.create("Ada", _now_stamp())
+            rule = await rules.create("daily", D1)
+            definition = await definitions.create(
+                "Brush teeth",
+                rule.id,
+                _now_stamp(),
+                assignee_child_ids=[child.id],
+            )
+            instance = await instances.upsert(
+                definition.id,
+                child.id,
+                D1,
+                _now_stamp(),
+                window="morning",
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                complete_instance(
+                    database,
+                    instance.id,
+                    actor_source="user",
+                    actor_user_id="user-1",
+                )
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                complete_instance(
+                    database,
+                    instance.id,
+                    actor_source="user",
+                    actor_user_id="user-2",
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            state_a, state_b = await asyncio.gather(task_a, task_b)
+            assert state_a == "done"
+            assert state_b == "done"
+            rows = await CompletionEventsDao(database).list_by_instance(
+                instance.id
+            )
+            assert len(rows) == 1
+            assert rows[0].actor_user_id == "user-1"
+        finally:
+            await database.close()
+
+    _run(_main())

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_lifecycle.py

@@ -8,7 +8,13 @@ import pytest
 from conftest import make_config_entry, make_hass, wire_entry_to_registry
 
 from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
-from custom_components.nestquest.const import DOMAIN as DOMAIN_CONST
+from custom_components.nestquest.const import (
+    CONF_DAY_ROLLOVER_TIME,
+    CONF_HORIZON_DAYS,
+    DEFAULT_HORIZON_DAYS,
+    SERVICE_REGENERATE,
+    DOMAIN as DOMAIN_CONST,
+)
 
 
 def _wire(entry, registry):
@@ -237,6 +243,163 @@ async def test_unload_closes_database_even_when_listener_removal_raises() -> Non
     assert database.connected is False, "database leaked despite remover error"
     assert entry.runtime_data is None
     assert DOMAIN not in hass.data
+    assert hass.time_change.size == 0, "time-change listener leaked"
+
+
+async def test_unload_cancels_time_listener_even_when_update_remover_raises() -> None:
+    """A raising update-listener remover must not skip the time-change cancel.
+
+    Both removers are independent: the update-listener remover raising must
+    still cancel the daily time-change callback, whose action would otherwise
+    fire after unload against a closed database.
+    """
+
+    class _RemoveBoom(Exception):
+        pass
+
+    def _remover():
+        raise _RemoveBoom()
+
+    entry = make_config_entry(entry_id="update-boom-time-cancel")
+    entry.add_update_listener = lambda listener: _remover
+    hass, _registry = make_hass()
+    await async_setup_entry(hass, entry)
+    assert hass.time_change.size == 1
+    with pytest.raises(_RemoveBoom):
+        await async_unload_entry(hass, entry)
+    assert hass.time_change.size == 0, "time-change listener not cancelled"
+    assert entry.runtime_data is None
+    assert DOMAIN not in hass.data
+
+
+async def test_setup_registers_day_rollover_listener(hass, make_entry) -> None:
+    """Setup registers a daily rollover listener and stores its cancel fn."""
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    assert hass.time_change.size == 1
+    registration = hass.time_change.registrations[0]
+    assert registration["hour"] == 0
+    assert registration["minute"] == 0
+    assert callable(entry.runtime_data.remove_time_change_listener)
+
+
+async def test_setup_registers_day_rollover_at_configured_time(hass, make_entry) -> None:
+    """The listener fires at the entry option's rollover time, not the default."""
+    entry = _wire(make_entry(options={CONF_DAY_ROLLOVER_TIME: "06:30"}), hass.registry)
+    await async_setup_entry(hass, entry)
+    registration = hass.time_change.registrations[0]
+    assert registration["hour"] == 6
+    assert registration["minute"] == 30
+
+
+async def test_unload_cancels_day_rollover_listener(hass, make_entry) -> None:
+    """Unload removes the time-change listener alongside the update listener."""
+    entry = _wire(make_entry(), hass.registry)
+    await async_setup_entry(hass, entry)
+    assert hass.time_change.size == 1
+    assert await async_unload_entry(hass, entry) is True
+    assert hass.time_change.size == 0
+
+
+async def test_rollover_time_change_reregisters_listener_at_new_time(
+    hass, make_entry
+) -> None:
+    """A rollover-time options change re-registers the listener at the new time."""
+    entry = _wire(
+        make_entry(options={CONF_DAY_ROLLOVER_TIME: "00:00"}), hass.registry
+    )
+    await async_setup_entry(hass, entry)
+    assert (hass.time_change.registrations[0]["hour"], hass.time_change.registrations[0]["minute"]) == (0, 0)
+
+    # The options flow writes the new rollover time, then HA reloads the
+    # entry: unload tears down the old listener, setup re-registers it.
+    entry.options = {CONF_DAY_ROLLOVER_TIME: "06:30"}
+    await async_unload_entry(hass, entry)
+    assert hass.time_change.size == 0
+
+    assert await async_setup_entry(hass, entry) is True
+    assert hass.time_change.size == 1
+    registration = hass.time_change.registrations[0]
+    assert (registration["hour"], registration["minute"]) == (6, 30)
+
+
+async def test_day_rollover_listener_materializes_ha_local_horizon(
+    hass, make_entry, monkeypatch
+) -> None:
+    """Firing the listener materializes today..today+horizon in HA local time."""
+    import datetime
+    from unittest.mock import AsyncMock
+    from zoneinfo import ZoneInfo
+
+    import custom_components.nestquest as nestquest
+
+    entry = _wire(make_entry(), hass.registry)
+    await async_setup_entry(hass, entry)
+    database = entry.runtime_data.database
+
+    fake = AsyncMock()
+    monkeypatch.setattr(nestquest, "_materialize_run", fake)
+
+    time_zone = ZoneInfo(hass.config.time_zone)
+    today = datetime.datetime.now(time_zone).date()
+    expected_start = today.isoformat()
+    expected_end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+
+    await hass.time_change.fire()
+
+    fake.assert_awaited_once_with(
+        database, expected_start, expected_end, today=today
+    )
+
+
+async def test_day_rollover_uses_configured_horizon_days(hass, make_entry) -> None:
+    """The daily run materializes only the configured horizon window."""
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    entry = _wire(make_entry(options={CONF_HORIZON_DAYS: 5}), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+
+    # A definition created AFTER setup has no instances yet (create does not
+    # materialize); firing the daily listener must generate only the horizon.
+    child_id = await _seed_daily_child_and_definition(
+        database, _local_today(hass)
+    )
+    await hass.time_change.fire()
+
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=5)).isoformat()
+    records = await QuestInstancesDao(database).list_by_date_range(
+        child_id, today.isoformat(), end
+    )
+    assert len(records) == 6
+
+    beyond = (today + datetime.timedelta(days=6)).isoformat()
+    beyond_end = (
+        today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    assert await QuestInstancesDao(database).list_by_date_range(
+        child_id, beyond, beyond_end
+    ) == []
+
+
+def test_day_rollover_time_change_tracker_rejects_local_kwarg() -> None:
+    """The fake async_track_time_change mirrors HA 2024.6 (no ``local`` kwarg).
+
+    Setup drives the local-time wrapper directly, so passing ``local=`` must
+    raise TypeError — otherwise a regression in the production call would be
+    silently masked by the fake.
+    """
+    import inspect as _inspect
+
+    from homeassistant.helpers.event import async_track_time_change
+
+    params = _inspect.signature(async_track_time_change).parameters
+    assert list(params) == ["hass", "action", "hour", "minute", "second"]
+    with pytest.raises(TypeError):
+        async_track_time_change(None, lambda _now: None, hour=0, minute=0, local=True)
 
 
 async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
@@ -255,4 +418,264 @@ async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
         "quest_definitions",
         "schedule_rules",
     ]
-    await async_unload_entry(hass, entry)
+    await async_unload_entry(hass, entry)
+
+
+async def _seed_daily_child_and_definition(database, today) -> int:
+    """Create one always-present child + a daily definition starting today.
+
+    Returns the child's id.  No presence schedule means the child is present
+    every day, so the daily rule yields exactly one instance per horizon day.
+    """
+    from custom_components.nestquest.dao_children import ChildrenDao
+    from custom_components.nestquest.quest_definitions import (
+        create_quest_definition,
+    )
+    from custom_components.nestquest.recurrence import RuleType, ScheduleRule
+
+    NOW = "2026-09-14T12:00:00+00:00"
+    child = await ChildrenDao(database).create("Ada", NOW)
+    await create_quest_definition(
+        database,
+        "Daily chore",
+        ScheduleRule(rule_type=RuleType.DAILY, start_date=today.isoformat()),
+        [child.id],
+        ["morning"],
+    )
+    return child.id
+
+
+def _local_today(hass) -> "datetime.date":
+    import datetime
+    from zoneinfo import ZoneInfo
+
+    return datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
+
+
+async def test_setup_backfills_instances_over_horizon(hass, make_entry) -> None:
+    """Setup materializes once so a restart backfills any missed days."""
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+    from custom_components.nestquest.db import NestQuestDatabase
+    from custom_components.nestquest.migrations import apply_migrations
+    from custom_components.nestquest.store import async_get_db_path
+
+    # Seed the database BEFORE setup: a restart reopens an existing file that
+    # already holds definitions.  The startup backfill must pick them up.
+    db_path = await async_get_db_path(hass)
+    pre = NestQuestDatabase(hass)
+    await pre.open(db_path)
+    await apply_migrations(pre)
+    child_id = await _seed_daily_child_and_definition(pre, _local_today(hass))
+    await pre.close()
+
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+    records = await QuestInstancesDao(entry.runtime_data.database).list_by_date_range(
+        child_id, today.isoformat(), end
+    )
+    assert len(records) == DEFAULT_HORIZON_DAYS + 1
+
+
+async def test_setup_backfills_configured_horizon_days(hass, make_entry) -> None:
+    """Setup backfills only the configured (non-default) horizon window."""
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+    from custom_components.nestquest.db import NestQuestDatabase
+    from custom_components.nestquest.migrations import apply_migrations
+    from custom_components.nestquest.store import async_get_db_path
+
+    db_path = await async_get_db_path(hass)
+    pre = NestQuestDatabase(hass)
+    await pre.open(db_path)
+    await apply_migrations(pre)
+    child_id = await _seed_daily_child_and_definition(pre, _local_today(hass))
+    await pre.close()
+
+    entry = _wire(
+        make_entry(options={CONF_HORIZON_DAYS: 5}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=5)).isoformat()
+    records = await QuestInstancesDao(
+        entry.runtime_data.database
+    ).list_by_date_range(child_id, today.isoformat(), end)
+    assert len(records) == 6
+
+    beyond = (today + datetime.timedelta(days=6)).isoformat()
+    beyond_end = (
+        today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    assert await QuestInstancesDao(
+        entry.runtime_data.database
+    ).list_by_date_range(child_id, beyond, beyond_end) == []
+
+
+def test_configured_horizon_days_validates_and_falls_back() -> None:
+    """The configured horizon is validated; a malformed value uses the default."""
+    import custom_components.nestquest as nestquest
+
+    assert nestquest._configured_horizon_days({CONF_HORIZON_DAYS: 5}) == 5
+    assert nestquest._configured_horizon_days({}) == DEFAULT_HORIZON_DAYS
+    assert (
+        nestquest._configured_horizon_days({CONF_HORIZON_DAYS: True})
+        == DEFAULT_HORIZON_DAYS
+    )
+    assert (
+        nestquest._configured_horizon_days({CONF_HORIZON_DAYS: "5"})
+        == DEFAULT_HORIZON_DAYS
+    )
+    assert (
+        nestquest._configured_horizon_days({CONF_HORIZON_DAYS: 0})
+        == DEFAULT_HORIZON_DAYS
+    )
+
+
+async def test_regenerate_service_registered_and_materializes(
+    hass, make_entry
+) -> None:
+    """``nestquest.regenerate`` is registered and calling it regenerates."""
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+    # A definition created AFTER setup has no instances yet (create does not
+    # materialize); the service must generate them on demand.
+    database = entry.runtime_data.database
+    child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
+
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+    records = await QuestInstancesDao(database).list_by_date_range(
+        child_id, today.isoformat(), end
+    )
+    assert len(records) == DEFAULT_HORIZON_DAYS + 1
+
+
+async def test_regenerate_service_uses_configured_horizon_days(
+    hass, make_entry
+) -> None:
+    """``nestquest.regenerate`` regenerates only the configured horizon."""
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    entry = _wire(
+        make_entry(options={CONF_HORIZON_DAYS: 5}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+
+    database = entry.runtime_data.database
+    child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
+
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=5)).isoformat()
+    records = await QuestInstancesDao(database).list_by_date_range(
+        child_id, today.isoformat(), end
+    )
+    assert len(records) == 6
+
+    beyond = (today + datetime.timedelta(days=6)).isoformat()
+    beyond_end = (
+        today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    assert await QuestInstancesDao(database).list_by_date_range(
+        child_id, beyond, beyond_end
+    ) == []
+
+
+async def test_unload_deregisters_regenerate_service(hass, make_entry) -> None:
+    """Unload removes the ``regenerate`` service so no stale entry leaks."""
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+    assert await async_unload_entry(hass, entry) is True
+    assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+
+async def test_regenerate_service_is_domain_scoped_across_entries(
+    hass, make_entry
+) -> None:
+    """The ``regenerate`` service is domain-global, not per-entry.
+
+    With two loaded entries the service must be registered and functional;
+    unloading one entry must keep it registered and functional for the
+    remaining entry; unloading the final entry must remove it.
+    """
+    import datetime
+
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    entry_a = _wire(make_entry(entry_id="entry_a"), hass.registry)
+    entry_b = _wire(make_entry(entry_id="entry_b"), hass.registry)
+
+    assert await async_setup_entry(hass, entry_a) is True
+    assert await async_setup_entry(hass, entry_b) is True
+    assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+    # The domain-global handler resolves a live database from any entry.
+    database_b = entry_b.runtime_data.database
+    child_id = await _seed_daily_child_and_definition(
+        database_b, _local_today(hass)
+    )
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    today = _local_today(hass)
+    end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
+    records = await QuestInstancesDao(database_b).list_by_date_range(
+        child_id, today.isoformat(), end
+    )
+    assert len(records) == DEFAULT_HORIZON_DAYS + 1
+
+    # Unloading one entry must not tear down the shared service.
+    assert await async_unload_entry(hass, entry_a) is True
+    assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+    # It still works for the remaining entry.
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+
+    # Unloading the final entry removes the domain-global service.
+    assert await async_unload_entry(hass, entry_b) is True
+    assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+
+
+async def test_setup_cleans_listeners_when_materialization_raises(
+    hass, make_entry, monkeypatch
+) -> None:
+    """A startup-materialization failure rolls back every registered listener.
+
+    The daily and update listeners are installed before the startup
+    materialization runs; when that walk raises, the database is closed but
+    the remover callables must also be invoked so no listener survives
+    firing against a closed database.
+    """
+    import custom_components.nestquest as nestquest
+
+    async def _boom(*_args, **_kwargs):
+        raise RuntimeError("materialization failed")
+
+    monkeypatch.setattr(nestquest, "_materialize_run", _boom)
+
+    entry = _wire(make_entry(), hass.registry)
+    with pytest.raises(RuntimeError, match="materialization failed"):
+        await async_setup_entry(hass, entry)
+
+    assert hass.time_change.size == 0
+    assert hass.registry.size == 0
+    assert not hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
+    assert not hasattr(entry, "runtime_data")

tests/test_services.py

@@ -0,0 +1,41 @@
+"""Tests for the NestQuest services.yaml declaration file."""
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+SERVICES_PATH = (
+    Path(__file__).parent.parent
+    / "custom_components"
+    / "nestquest"
+    / "services.yaml"
+)
+
+
+def test_services_yaml_exists() -> None:
+    assert SERVICES_PATH.exists(), f"services.yaml missing at {SERVICES_PATH}"
+
+
+def test_services_yaml_declares_regenerate() -> None:
+    text = SERVICES_PATH.read_text(encoding="utf-8")
+    assert re.search(r"^regenerate:\s*$", text, re.MULTILINE), (
+        "services.yaml must declare the 'regenerate' service"
+    )
+
+
+def test_services_yaml_regenerate_has_description_and_no_fields() -> None:
+    """The regenerate service is documented with a clear description and no
+    required fields (there is no ``fields`` block under it)."""
+    text = SERVICES_PATH.read_text(encoding="utf-8")
+    match = re.search(r"^regenerate:\s*$", text, re.MULTILINE)
+    assert match is not None
+    section = text[match.end():]
+    next_service = re.search(r"^(?![#\s])\w+:\s*$", section, re.MULTILINE)
+    if next_service is not None:
+        section = section[: next_service.start()]
+    assert re.search(r"description:", section), (
+        "regenerate must carry a description"
+    )
+    assert "fields:" not in section, (
+        "regenerate declares no required fields"
+    )

tests/test_timezones.py

@@ -0,0 +1,370 @@
+"""Timezone tests for the single HA-local date source in the generation path.
+
+These pin the done-condition that every "today" read in the generation path
+derives from ``hass.config.time_zone`` — not the host clock and not UTC —
+and that the rolling horizon ``[today, today + DEFAULT_HORIZON_DAYS]`` and the
+no-past guard resolve the correct local dates around midnight and across the
+daylight-saving transitions.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_instances import QuestInstancesDao
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+from custom_components.nestquest.quest_definitions import (
+    assign_child,
+    create_quest_definition,
+    edit_quest_definition,
+)
+from custom_components.nestquest.recurrence import RuleType, ScheduleRule
+
+NOW = "2026-09-14T12:00:00+00:00"
+NEW_YORK = "America/New_York"
+
+
+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
+
+
+async def _prepare(path):
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    return database
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _freeze_clock(
+    monkeypatch, instant: datetime.datetime
+) -> tuple[datetime.datetime, list]:
+    """Freeze the integration's ``datetime`` clock to a fixed aware instant.
+
+    Only ``custom_components.nestquest``'s ``datetime`` binding is patched, so
+    the generation path's "today" reads this instant (localized to whatever
+    time zone it asks for) while every other module and the test harness keep
+    the real clock.  Returns ``(frozen, requested)`` where ``frozen`` is the
+    instant normalized to UTC and ``requested`` is the list of ``tzinfo``
+    objects the code under test passed to ``now`` — so a test can prove the
+    generation path actually requested the configured HA time zone rather
+    than silently reading UTC.
+    """
+    import custom_components.nestquest as nestquest
+
+    frozen = instant.astimezone(datetime.timezone.utc)
+    requested: list = []
+
+    class _FrozenDatetime(datetime.datetime):
+        @classmethod
+        def now(cls, tz=None):
+            requested.append(tz)
+            if tz is None:
+                return frozen.replace(tzinfo=None)
+            return frozen.astimezone(tz)
+
+    fake_module = SimpleNamespace(
+        datetime=_FrozenDatetime,
+        timedelta=datetime.timedelta,
+        date=datetime.date,
+    )
+    monkeypatch.setattr(nestquest, "datetime", fake_module)
+    return frozen, requested
+
+
+def _assert_requested_ha_timezone(requested, time_zone: str) -> None:
+    """Assert ``now`` was called with (only) the configured HA time zone.
+
+    This is what turns the DST tests into a real proof of the HA-local date
+    source: if the generation path regressed to ``now(timezone.utc)``, its
+    requested tz would be UTC and this fails.
+    """
+    assert requested, "the generation path never called now(tz)"
+    assert all(tz == ZoneInfo(time_zone) for tz in requested)
+
+
+def _tz_hass(time_zone: str):
+    """A minimal hass exposing only the ``config.time_zone`` the walk reads."""
+    return SimpleNamespace(config=SimpleNamespace(time_zone=time_zone))
+
+
+async def _seed_daily(database, start_iso: str) -> int:
+    """One always-present child + a daily definition starting ``start_iso``."""
+    child = await ChildrenDao(database).create("Ada", NOW)
+    await create_quest_definition(
+        database,
+        "Daily chore",
+        ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso),
+        [child.id],
+        ["morning"],
+    )
+    return child.id
+
+
+def _local_date(instant_utc: datetime.datetime, time_zone: str) -> datetime.date:
+    return instant_utc.astimezone(ZoneInfo(time_zone)).date()
+
+
+def _assert_contiguous_horizon(records, local_today: datetime.date) -> None:
+    """The horizon is exactly today..today+horizon, no off-by-one, no gap."""
+    expected = {
+        (local_today + datetime.timedelta(days=offset)).isoformat()
+        for offset in range(DEFAULT_HORIZON_DAYS + 1)
+    }
+    assert {r.due_date for r in records} == expected
+    assert len(records) == DEFAULT_HORIZON_DAYS + 1
+
+
+def test_horizon_uses_newyork_local_date_after_local_midnight(
+    tmp_path, monkeypatch
+) -> None:
+    """A run just after New York local midnight uses the local date.
+
+    The frozen instant is 00:05 New York local time on 2026-03-10 (EDT, so
+    04:05 UTC).  The horizon must start on 2026-03-10 — the local calendar
+    date — not any host/UTC-derived value.
+    """
+    instant = datetime.datetime(
+        2026, 3, 10, 0, 5, tzinfo=ZoneInfo(NEW_YORK)
+    )
+    local_today = instant.date()
+
+    async def _body(database):
+        hass = _tz_hass(NEW_YORK)
+        _frozen, requested = _freeze_clock(monkeypatch, instant)
+        import custom_components.nestquest as nestquest
+
+        child_id = await _seed_daily(database, local_today.isoformat())
+        await nestquest._run_horizon_materialization(hass, database)
+        records = await QuestInstancesDao(database).list_by_date_range(
+            child_id,
+            local_today.isoformat(),
+            (local_today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat(),
+        )
+        _assert_contiguous_horizon(records, local_today)
+        _assert_requested_ha_timezone(requested, NEW_YORK)
+        return len(records)
+
+    async def _main():
+        database = await _prepare(tmp_path / "tz-midnight.db")
+        try:
+            return await _body(database)
+        finally:
+            await database.close()
+
+    assert _run(_main()) == DEFAULT_HORIZON_DAYS + 1
+
+
+def test_horizon_uses_local_date_not_utc_when_host_ahead(
+    tmp_path, monkeypatch
+) -> None:
+    """The walk's "today" is HA-local, never the UTC/system calendar date.
+
+    FROZEN at 2026-01-15 03:00 UTC: the UTC calendar date is 2026-01-15, but
+    New York local is still 2026-01-14 22:00 EST.  The horizon must start on
+    2026-01-14 (local), proving the UTC date did not win.
+    """
+    instant = datetime.datetime(2026, 1, 15, 3, 0, tzinfo=datetime.timezone.utc)
+    local_today = _local_date(instant, NEW_YORK)
+    assert local_today == datetime.date(2026, 1, 14)
+    assert instant.date() == datetime.date(2026, 1, 15)
+
+    async def _body(database):
+        hass = _tz_hass(NEW_YORK)
+        _frozen, requested = _freeze_clock(monkeypatch, instant)
+        import custom_components.nestquest as nestquest
+
+        child_id = await _seed_daily(database, local_today.isoformat())
+        await nestquest._run_horizon_materialization(hass, database)
+        records = await QuestInstancesDao(database).list_by_date_range(
+            child_id,
+            local_today.isoformat(),
+            (local_today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat(),
+        )
+        _assert_contiguous_horizon(records, local_today)
+        _assert_requested_ha_timezone(requested, NEW_YORK)
+        return len(records)
+
+    async def _main():
+        database = await _prepare(tmp_path / "tz-host-ahead.db")
+        try:
+            return await _body(database)
+        finally:
+            await database.close()
+
+    assert _run(_main()) == DEFAULT_HORIZON_DAYS + 1
+
+
+def test_horizon_resolves_spring_forward_day_without_off_by_one(
+    tmp_path, monkeypatch
+) -> None:
+    """Across the 2026-03-08 spring-forward, today and the horizon stay local.
+
+    Before the jump ("01:30 EST") and after it ("03:30 EDT") both resolve to
+    2026-03-08 local — the lost hour must not skip or repeat a calendar day —
+    and the full horizon remains 15 contiguous days.
+    """
+    zone = ZoneInfo(NEW_YORK)
+    before = datetime.datetime(2026, 3, 8, 1, 30, tzinfo=zone)  # EST
+    after = datetime.datetime(2026, 3, 8, 3, 30, tzinfo=zone)   # EDT
+    assert before.date() == after.date() == datetime.date(2026, 3, 8)
+
+    async def _body(database):
+        hass = _tz_hass(NEW_YORK)
+        import custom_components.nestquest as nestquest
+
+        requested_batches: list[list] = []
+        for instant in (before, after):
+            _frozen, requested = _freeze_clock(monkeypatch, instant)
+            requested_batches.append(requested)
+            local_today = instant.date()
+            child_id = await _seed_daily(database, local_today.isoformat())
+            await nestquest._run_horizon_materialization(hass, database)
+            records = await QuestInstancesDao(database).list_by_date_range(
+                child_id,
+                local_today.isoformat(),
+                (local_today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat(),
+            )
+            _assert_contiguous_horizon(records, local_today)
+        requested_all = [tz for batch in requested_batches for tz in batch]
+        _assert_requested_ha_timezone(requested_all, NEW_YORK)
+        return local_today
+
+    async def _main():
+        database = await _prepare(tmp_path / "tz-spring-forward.db")
+        try:
+            return await _body(database)
+        finally:
+            await database.close()
+
+    assert _run(_main()) == datetime.date(2026, 3, 8)
+
+
+def test_horizon_resolves_fall_back_day_without_off_by_one(
+    tmp_path, monkeypatch
+) -> None:
+    """Across the 2026-11-01 fall-back, today and the horizon stay local.
+
+    The repeated 01:00-02:00 hour must not resolve to the wrong calendar
+    day: the first 01:30 (``fold=0``, EDT) and the repeated 01:30
+    (``fold=1``, EST) are distinct instants yet both resolve to 2026-11-01
+    — and both, along with 03:30 EST after the jump, materialize the same
+    15 contiguous-day horizon with no off-by-one.
+    """
+    zone = ZoneInfo(NEW_YORK)
+    first = datetime.datetime(2026, 11, 1, 1, 30, tzinfo=zone, fold=0)  # EDT
+    second = datetime.datetime(2026, 11, 1, 1, 30, tzinfo=zone, fold=1)  # EST
+    after = datetime.datetime(2026, 11, 1, 3, 30, tzinfo=zone)  # EST
+    assert first.date() == second.date() == after.date() == datetime.date(2026, 11, 1)
+    assert first.utcoffset() != second.utcoffset()  # the two repeated-hour folds
+
+    async def _body(database):
+        hass = _tz_hass(NEW_YORK)
+        import custom_components.nestquest as nestquest
+
+        requested_batches: list[list] = []
+        for instant in (first, second, after):
+            _frozen, requested = _freeze_clock(monkeypatch, instant)
+            requested_batches.append(requested)
+            local_today = instant.date()
+            child_id = await _seed_daily(database, local_today.isoformat())
+            await nestquest._run_horizon_materialization(hass, database)
+            records = await QuestInstancesDao(database).list_by_date_range(
+                child_id,
+                local_today.isoformat(),
+                (local_today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat(),
+            )
+            _assert_contiguous_horizon(records, local_today)
+        requested_all = [tz for batch in requested_batches for tz in batch]
+        _assert_requested_ha_timezone(requested_all, NEW_YORK)
+        return local_today
+
+    async def _main():
+        database = await _prepare(tmp_path / "tz-fall-back.db")
+        try:
+            return await _body(database)
+        finally:
+            await database.close()
+
+    assert _run(_main()) == datetime.date(2026, 11, 1)
+
+
+def test_definition_business_regeneration_uses_pinned_today_not_host(
+    tmp_path, monkeypatch
+) -> None:
+    """``edit_quest_definition``/``assign_child`` anchor regeneration on
+    the caller-pinned HA-local ``today``, not the host clock.
+
+    The host clock (the ``_today`` fallback) is pinned one day AHEAD of the
+    HA-local ``today``; the guard would reject the local "today" as past
+    (and the horizon would skip it) if it consulted the host clock.  Both
+    business operations must regenerate against the pinned date and keep
+    that first local day, proving the threaded ``today`` wins.
+    """
+    import custom_components.nestquest.dao_instances as dao_instances_module
+
+    pinned = datetime.date(2026, 1, 14)
+    host_today = datetime.date(2026, 1, 15)
+    monkeypatch.setattr(dao_instances_module, "_today", lambda: host_today)
+
+    async def _body(database):
+        children = ChildrenDao(database)
+        dao = QuestInstancesDao(database)
+        horizon_end = pinned + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+
+        a = await children.create("Ada", NOW)
+        created = await create_quest_definition(
+            database,
+            "Daily chore",
+            ScheduleRule(rule_type=RuleType.DAILY, start_date=pinned.isoformat()),
+            [a.id],
+            [("morning", "09:00")],
+        )
+        definition_id = created.definition.id
+
+        # edit anchors on pinned: the horizon starts at the local day the
+        # host clock calls "past", so its presence proves pinned won.
+        await edit_quest_definition(
+            database,
+            definition_id,
+            windows=[("morning", "10:15")],
+            today=pinned,
+        )
+        a_records = await dao.list_by_date_range(
+            a.id, pinned.isoformat(), horizon_end.isoformat()
+        )
+        assert [r.due_date for r in a_records][0] == pinned.isoformat()
+        assert {r.due_time for r in a_records} == {"10:15"}
+        assert len(a_records) == DEFAULT_HORIZON_DAYS + 1
+
+        # assign anchors on pinned too: the new child's regeneration keeps
+        # the same pinned-local first day rather than skipping to the host day.
+        b = await children.create("Bo", NOW)
+        await assign_child(database, definition_id, b.id, today=pinned)
+        b_records = await dao.list_by_date_range(
+            b.id, pinned.isoformat(), horizon_end.isoformat()
+        )
+        assert [r.due_date for r in b_records][0] == pinned.isoformat()
+        assert len(b_records) == DEFAULT_HORIZON_DAYS + 1
+        return definition_id
+
+    async def _main():
+        database = await _prepare(tmp_path / "tz-definition-regen.db")
+        try:
+            return await _body(database)
+        finally:
+            await database.close()
+
+    assert _run(_main()) is not None