Allow saving the empty default notify target

An empty target means notifications are not configured yet and is the
shipped default; only malformed strings (padded or whitespace-only)
reject.

custom_components/nestquest/__init__.py

@@ -35,6 +35,7 @@ from .migrations import apply_migrations
 from .services import async_deregister_services, async_register_services
 from .store import async_get_db_path
 from .admin_allowlist import seed_setup_admin
+from .sweep import run_missed_sweep
 
 _LOGGER = LOGGER
 
@@ -327,18 +328,23 @@ def _register_day_rollover_listener(
     database: NestQuestDatabase,
     options: dict[str, Any],
 ) -> Callable[[], Any]:
-    """Register the daily materialization listener and return its remover.
+    """Register the daily rollover 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``.
+    Fires once per day at the configured local ``day_rollover_time``:
+    first re-materializes the rolling horizon
+    ``[today, today + horizon_days]`` through the shared
+    :func:`_run_horizon_materialization` path (the entry's configured,
+    validated ``horizon_days``), then runs the Feature 11 missed sweep
+    — announcing yesterday's still-open quests as missed AFTER today's
+    instances exist, watermark-guarded so a re-fire of the listener the
+    same night announces nothing twice.
     """
     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)
+        await run_missed_sweep(hass, database)
 
     return async_track_time_change(
         hass, _run_materialization, hour=hour, minute=minute
@@ -498,6 +504,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         await _run_horizon_materialization(
             hass, database, _configured_horizon_days(dict(entry.options))
         )
+        # The missed sweep runs at startup too (Feature 11): if HA was
+        # down at the rollover time, the previous night's still-open
+        # quests are announced as missed now instead of waiting for
+        # the next midnight.  Watermark-guarded, so this is a no-op
+        # when the night's sweep already ran.
+        await run_missed_sweep(hass, database)
         _register_services(hass)
         # The shared Feature 10 coordinator: one refresh cycle every
         # entity reads from (CONF_UPDATE_INTERVAL seconds, default
@@ -615,16 +627,19 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
                     if unload_error is None:
                         unload_error = err
         finally:
-            database = getattr(runtime_data, "database", None)
-            if database is not None:
-                await database.close()
-            # Tear the shared coordinator down after the DB close: no
-            # listener may outlive the connection it reads.
+            # The coordinator's entity listeners must stop BEFORE the
+            # database closes: a scheduled refresh starting during the
+            # close await would touch a cleared connection.  (The
+            # platform unload above already removed the entities; this
+            # clears the coordinator's own remaining listeners.)
             coordinator = getattr(runtime_data, "coordinator", None)
             if coordinator is not None:
                 shutdown = getattr(coordinator, "async_shutdown", None)
                 if shutdown is not None:
                     await shutdown()
+            database = getattr(runtime_data, "database", None)
+            if database is not None:
+                await database.close()
     # Domain-global services are torn down only once the FINAL entry
     # unloads (``hass.data[DOMAIN]`` is now empty), never when a sibling
     # entry is still loaded.

custom_components/nestquest/completion.py

@@ -374,14 +374,25 @@ class CompletionResult(str):
     appended an event (decided under the same lock as the append), so
     a caller — the Feature 09 service path firing bus events —
     announces a transition exactly once even under a concurrent
-    duplicate call where the loser's write is a no-op.
+    duplicate call where the loser's write is a no-op.  A completion
+    also carries the ``was_on_time`` verdict computed under that same
+    lock, so the event payload never re-reads mutable latest state a
+    racing reversal could have replaced.
     """
 
     appended: bool
+    was_on_time: bool | None
 
-    def __new__(cls, state: str, *, appended: bool) -> "CompletionResult":
+    def __new__(
+        cls,
+        state: str,
+        *,
+        appended: bool,
+        was_on_time: bool | None = None,
+    ) -> "CompletionResult":
         obj = super().__new__(cls, state)
         obj.appended = appended
+        obj.was_on_time = was_on_time
         return obj
 
 
@@ -429,6 +440,10 @@ async def complete_instance(
                 f"event_type must be '{EVENT_COMPLETED}' or "
                 f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
             )
+        # Computed under the completion lock and carried on the result:
+        # the bus event must report THIS completion's timing, never a
+        # re-read of the latest event a racing reversal could replace.
+        on_time = _was_on_time(instance, now, today)
         await append_event(
             database,
             instance.id,
@@ -437,10 +452,10 @@ async def complete_instance(
             actor_source=actor_source,
             actor_user_id=actor_user_id,
             actor_child_id=actor_child_id,
-            was_on_time=_was_on_time(instance, now, today),
+            was_on_time=on_time,
             now=now,
         )
-    return CompletionResult("done", appended=True)
+    return CompletionResult("done", appended=True, was_on_time=on_time)
 
 
 async def uncomplete_instance(

custom_components/nestquest/const.py

@@ -73,6 +73,26 @@ CONF_UPDATE_INTERVAL = "update_interval"
 DEFAULT_UPDATE_INTERVAL = 300
 MIN_UPDATE_INTERVAL = 30
 
+# Notification automations (Feature 11).  The notify target is a free-text
+# Home Assistant service name (e.g. "notify.mobile_app_dads_phone") the
+# household fills in — never a hard-coded personal target; the times are
+# strict HH:MM local; each of the four automations has an independent
+# enable toggle.  Per the owner-settled resolution the end-of-day report
+# defaults to 20:00 — deliberately BEFORE the midnight missed sweep —
+# and never defaults to the rollover time.
+CONF_NOTIFY_TARGET = "notify_target"
+CONF_MORNING_SUMMARY_TIME = "morning_summary_time"
+DEFAULT_MORNING_SUMMARY_TIME = "08:00"
+CONF_AFTERNOON_REMINDER_TIME = "afternoon_reminder_time"
+DEFAULT_AFTERNOON_REMINDER_TIME = "15:00"
+CONF_END_OF_DAY_REPORT_TIME = "end_of_day_report_time"
+DEFAULT_END_OF_DAY_REPORT_TIME = "20:00"
+CONF_MORNING_SUMMARY_ENABLED = "morning_summary_enabled"
+CONF_AFTERNOON_REMINDER_ENABLED = "afternoon_reminder_enabled"
+CONF_END_OF_DAY_REPORT_ENABLED = "end_of_day_report_enabled"
+CONF_CELEBRATION_ENABLED = "celebration_enabled"
+DEFAULT_AUTOMATION_ENABLED = True
+
 # Quest windows (D-008): a definition may declare several day windows;
 # each produces its own instance per day and groups the panel's Quest
 # Log into three columns.  The clock ranges are stored here so they stay

custom_components/nestquest/coordinator.py

@@ -25,6 +25,7 @@ the admin header degrades to a harmless zero instead of guessing.
 """
 from __future__ import annotations
 
+import asyncio
 import datetime
 from dataclasses import dataclass
 from zoneinfo import ZoneInfo
@@ -97,7 +98,16 @@ class NestQuestSnapshot:
 
 
 class NestQuestCoordinator(DataUpdateCoordinator):
-    """Shared refresh cycle for every NestQuest entity."""
+    """Shared refresh cycle for every NestQuest entity.
+
+    Forced refreshes (the service paths pushing an immediate update
+    after a completion) are SERIALIZED per entry: overlapping
+    complete/uncomplete calls must not interleave their snapshot
+    passes, or an older pass could publish a stale snapshot AFTER a
+    newer one and leave entities wrong until the next poll.  Real
+    HA's coordinator already serializes internally; the lock keeps
+    the same guarantee on the stand-in and costs nothing upstream.
+    """
 
     def __init__(
         self,
@@ -122,6 +132,14 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         )
         self.entry_id = entry_id
         self.database = database
+        self._refresh_lock = asyncio.Lock()
+
+    async def async_refresh(self) -> None:
+        """One refresh at a time; a queued refresh runs AFTER the
+        in-flight one publishes, so the last snapshot always reflects
+        the latest mutation."""
+        async with self._refresh_lock:
+            await super().async_refresh()
 
     def _local_now(self) -> datetime.datetime:
         time_zone = ZoneInfo(self.hass.config.time_zone)

custom_components/nestquest/dao_instances.py

@@ -493,6 +493,43 @@ class QuestInstancesDao:
         )
         return [_instance_from_row(row) for row in rows]
 
+    async def list_missed_unswept(
+        self, today_iso: str, watermark: str | None
+    ) -> list[QuestInstanceRecord]:
+        """Return past-due open instances the sweep has not announced.
+
+        Feature 11's missed-sweep query: ``due_date`` strictly before
+        ``today_iso`` (past due), at or after ``watermark`` when one is
+        recorded, and with NO completion event ever (the Feature 08
+        append-only contract: an instance that was ever completed is
+        never reported missed, even when later reversed — history keeps
+        both rows and the sweep never mutates any).  The watermark is
+        the HA-local date the sweep last ran, and the window is
+        ``[watermark, today)``: the nightly run entering day D+1 sweeps
+        exactly day D's instances, so every day's quests are announced
+        once — at the midnight after their due date — and a same-night
+        rerun (watermark == today) sees an empty window.  Read-only;
+        ordered deterministically by (due_date, id).
+        """
+        _validate_date(today_iso, "today_iso")
+        if watermark is not None:
+            _validate_date(watermark, "watermark")
+        sql = (
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
+            "WHERE due_date < ? "
+            "AND NOT EXISTS ("
+            "  SELECT 1 FROM completion_events "
+            "  WHERE instance_id = quest_instances.id"
+            ")"
+        )
+        parameters: tuple = (today_iso,)
+        if watermark is not None:
+            sql += " AND due_date >= ?"
+            parameters += (watermark,)
+        sql += " ORDER BY due_date, id"
+        rows = await self._database.fetch_all(sql, parameters)
+        return [_instance_from_row(row) for row in rows]
+
     async def delete_future_uncompleted(
         self, definition_id: int, cutoff_date: str, *, today: datetime.date | None = None
     ) -> int:

custom_components/nestquest/dao_meta.py

@@ -0,0 +1,44 @@
+"""Typed DAO for the integration-owned ``nestquest_meta_state`` table.
+
+Generic persistent key/value bookkeeping that belongs to the
+integration itself rather than any domain table — currently the
+Feature 11 missed-sweep watermark.  Values are plain strings; callers
+interpret them.  All SQL for this table lives here per the feature
+guardrails, and every method runs through
+:class:`~.db.NestQuestDatabase` so nothing blocks the event loop.
+"""
+from __future__ import annotations
+
+from .dao_children import _connection_lock
+from .db import NestQuestDatabase
+
+_TABLE = "nestquest_meta_state"
+
+
+class MetaStateDao:
+    """Typed async access to the ``nestquest_meta_state`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def get(self, key: str) -> str | None:
+        """Return the stored value for ``key``, or None when absent."""
+        row = await self._database.fetch_one(
+            f"SELECT value FROM {_TABLE} WHERE key = ?", (key,)
+        )
+        return row[0] if row is not None else None
+
+    async def set(self, key: str, value: str) -> None:
+        """Upsert one key/value row atomically.
+
+        The write runs inside one transaction under the shared
+        connection lock, so a racing reader either sees the old value
+        or the fully-written new one, never a half-applied state.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                await self._database.execute(
+                    f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
+                    "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
+                    (key, value),
+                )

custom_components/nestquest/events.py

@@ -71,22 +71,24 @@ async def _instance_payload(
 
 
 async def fire_quest_completed(
-    hass, database: NestQuestDatabase, instance_id: int
+    hass,
+    database: NestQuestDatabase,
+    instance_id: int,
+    *,
+    was_on_time: bool | None,
 ) -> None:
     """Fire ``nestquest_quest_completed`` and, when the completion
     cleared the child's whole day, ``nestquest_child_day_complete``.
 
     Called by the complete_quest service handler AFTER an actual
-    completion transition; a no-op re-complete never reaches here.
+    completion transition, with the completion's ``was_on_time`` —
+    computed under the completion lock and carried on the result — so
+    the payload never re-reads mutable latest state a racing reversal
+    could have replaced.  A no-op re-complete never reaches here.
     """
     now_stamp = _now_stamp()
     payload = await _instance_payload(database, instance_id, now_stamp)
-    latest = await CompletionEventsDao(database).get_latest_for_instance(
-        instance_id
-    )
-    payload["was_on_time"] = (
-        latest.was_on_time if latest is not None else None
-    )
+    payload["was_on_time"] = was_on_time
     hass.bus.async_fire(EVENT_QUEST_COMPLETED, payload)
 
     # Day-complete: quests were owed today and none remain (the same

custom_components/nestquest/migrations.py

@@ -28,7 +28,10 @@ copying each definition's assignee across.  Migration 4 adds the
 windows table, migration 5 rebuilds ``quest_instances`` onto the
 widened (definition_id, child_id, due_date, window) key, and migration
 6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
-column.  Each callable migration runs inside its own transaction
+column.  Migration 7 adds the generic ``nestquest_meta_state``
+key/value table for integration-owned bookkeeping (Feature 11's
+missed-sweep watermark).  Each callable migration runs inside its own
+transaction
 together with its version stamp, so a crash mid-step rolls the
 statements and the stamp back together.
 
@@ -80,6 +83,7 @@ from .schema import (
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
     SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+    SCHEMA_V7_META_STATE_DDL,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -414,6 +418,11 @@ MIGRATIONS: Sequence[MigrationStep] = [
     # actor_child_id column and its actor-pair coherence CHECK,
     # backfilling legacy panel rows to their own child_id.
     _add_actor_child_id,
+    # Migration 7 (Feature 11): the generic meta_state key/value table
+    # — a plain new table, so CREATE IF NOT EXISTS suffices for both
+    # fresh databases (which already made it in migration 1) and
+    # pre-v7 databases.  Strictly additive; no existing table changes.
+    list(SCHEMA_V7_META_STATE_DDL),
 ]
 
 

custom_components/nestquest/options_flow.py

@@ -11,12 +11,24 @@ from homeassistant.helpers import config_validation as cv
 
 from .const import (
     CONF_ADMIN_USER_IDS,
+    CONF_AFTERNOON_REMINDER_ENABLED,
+    CONF_AFTERNOON_REMINDER_TIME,
+    CONF_CELEBRATION_ENABLED,
     CONF_DAY_ROLLOVER_TIME,
+    CONF_END_OF_DAY_REPORT_ENABLED,
+    CONF_END_OF_DAY_REPORT_TIME,
     CONF_HORIZON_DAYS,
+    CONF_MORNING_SUMMARY_ENABLED,
+    CONF_MORNING_SUMMARY_TIME,
+    CONF_NOTIFY_TARGET,
     CONF_PANEL_IDLE_TIMEOUT,
     CONF_UPDATE_INTERVAL,
+    DEFAULT_AFTERNOON_REMINDER_TIME,
+    DEFAULT_AUTOMATION_ENABLED,
     DEFAULT_DAY_ROLLOVER_TIME,
+    DEFAULT_END_OF_DAY_REPORT_TIME,
     DEFAULT_HORIZON_DAYS,
+    DEFAULT_MORNING_SUMMARY_TIME,
     DEFAULT_PANEL_IDLE_TIMEOUT,
     DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
@@ -71,9 +83,51 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
     ):
         errors[CONF_UPDATE_INTERVAL] = "invalid"
 
+    # The notification section (Feature 11) is OPTIONAL in raw submits
+    # the same way: a missing key keeps the current value.  The three
+    # times are strict HH:MM; the notify target is a non-empty HA
+    # service name; the toggles are real booleans.
+    for time_key in (
+        CONF_MORNING_SUMMARY_TIME,
+        CONF_AFTERNOON_REMINDER_TIME,
+        CONF_END_OF_DAY_REPORT_TIME,
+    ):
+        time_value = user_input.get(time_key)
+        if time_value is not None and (
+            not isinstance(time_value, str) or not _parse_time(time_value)
+        ):
+            errors[time_key] = "invalid_time"
+    notify_target = user_input.get(CONF_NOTIFY_TARGET)
+    if notify_target is not None and not isinstance(notify_target, str):
+        errors[CONF_NOTIFY_TARGET] = "invalid"
+    elif isinstance(notify_target, str) and notify_target != (
+        notify_target.strip()
+    ):
+        # An EMPTY target is valid — it means "no notifications
+        # configured yet" and is the shipped default; a value that
+        # differs from its own trim (padded or whitespace-only) is
+        # malformed and rejected.
+        errors[CONF_NOTIFY_TARGET] = "invalid"
+    for toggle_key in (
+        CONF_MORNING_SUMMARY_ENABLED,
+        CONF_AFTERNOON_REMINDER_ENABLED,
+        CONF_END_OF_DAY_REPORT_ENABLED,
+        CONF_CELEBRATION_ENABLED,
+    ):
+        toggle_value = user_input.get(toggle_key)
+        if toggle_value is not None and not isinstance(toggle_value, bool):
+            errors[toggle_key] = "invalid"
+
     return errors
 
 
+def _strict_bool(value: object) -> bool:
+    """Require a real bool; voluptuous ``bool`` would coerce strings."""
+    if not isinstance(value, bool):
+        raise vol.Invalid(f"expected a boolean, got {value!r}")
+    return value
+
+
 def _build_schema(
     current: dict[str, Any],
     *,
@@ -85,7 +139,10 @@ def _build_schema(
     The admin picker is a multi-select of EXISTING Home Assistant users
     (never a free-text field — the allowlist stores user IDs); its
     default is the current database allowlist, so an untouched submit
-    re-saves exactly who is already allowed.
+    re-saves exactly who is already allowed.  The notification section
+    (Feature 11) carries the household's notify target, the three
+    automation times, and four enable toggles — defaults from
+    :mod:`~.const`, never a hard-coded personal target.
     """
     fields: dict[Any, Any] = {
         vol.Required(
@@ -100,6 +157,37 @@ def _build_schema(
         vol.Required(
             CONF_UPDATE_INTERVAL, default=current[CONF_UPDATE_INTERVAL]
         ): int,
+        vol.Required(
+            CONF_NOTIFY_TARGET, default=current[CONF_NOTIFY_TARGET]
+        ): str,
+        vol.Required(
+            CONF_MORNING_SUMMARY_TIME,
+            default=current[CONF_MORNING_SUMMARY_TIME],
+        ): str,
+        vol.Required(
+            CONF_AFTERNOON_REMINDER_TIME,
+            default=current[CONF_AFTERNOON_REMINDER_TIME],
+        ): str,
+        vol.Required(
+            CONF_END_OF_DAY_REPORT_TIME,
+            default=current[CONF_END_OF_DAY_REPORT_TIME],
+        ): str,
+        vol.Required(
+            CONF_MORNING_SUMMARY_ENABLED,
+            default=current[CONF_MORNING_SUMMARY_ENABLED],
+        ): _strict_bool,
+        vol.Required(
+            CONF_AFTERNOON_REMINDER_ENABLED,
+            default=current[CONF_AFTERNOON_REMINDER_ENABLED],
+        ): _strict_bool,
+        vol.Required(
+            CONF_END_OF_DAY_REPORT_ENABLED,
+            default=current[CONF_END_OF_DAY_REPORT_ENABLED],
+        ): _strict_bool,
+        vol.Required(
+            CONF_CELEBRATION_ENABLED,
+            default=current[CONF_CELEBRATION_ENABLED],
+        ): _strict_bool,
     }
     if admin_choices is not None:
         # HA's supported multi-select validator: the frontend can
@@ -233,6 +321,23 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
                 self._current_values()[CONF_UPDATE_INTERVAL],
             ),
         }
+        # The notification section (Feature 11) stores the household's
+        # choices the same fall-back-to-current way, so raw submits
+        # that predate the section keep their stored values.
+        current = self._current_values()
+        for notification_key in (
+            CONF_NOTIFY_TARGET,
+            CONF_MORNING_SUMMARY_TIME,
+            CONF_AFTERNOON_REMINDER_TIME,
+            CONF_END_OF_DAY_REPORT_TIME,
+            CONF_MORNING_SUMMARY_ENABLED,
+            CONF_AFTERNOON_REMINDER_ENABLED,
+            CONF_END_OF_DAY_REPORT_ENABLED,
+            CONF_CELEBRATION_ENABLED,
+        ):
+            data[notification_key] = user_input.get(
+                notification_key, current[notification_key]
+            )
         if admin_ids_present:
             data[CONF_ADMIN_USER_IDS] = admin_ids
         elif admin_default:
@@ -254,6 +359,14 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             (CONF_DAY_ROLLOVER_TIME, DEFAULT_DAY_ROLLOVER_TIME),
             (CONF_PANEL_IDLE_TIMEOUT, DEFAULT_PANEL_IDLE_TIMEOUT),
             (CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL),
+            (CONF_NOTIFY_TARGET, ""),
+            (CONF_MORNING_SUMMARY_TIME, DEFAULT_MORNING_SUMMARY_TIME),
+            (CONF_AFTERNOON_REMINDER_TIME, DEFAULT_AFTERNOON_REMINDER_TIME),
+            (CONF_END_OF_DAY_REPORT_TIME, DEFAULT_END_OF_DAY_REPORT_TIME),
+            (CONF_MORNING_SUMMARY_ENABLED, DEFAULT_AUTOMATION_ENABLED),
+            (CONF_AFTERNOON_REMINDER_ENABLED, DEFAULT_AUTOMATION_ENABLED),
+            (CONF_END_OF_DAY_REPORT_ENABLED, DEFAULT_AUTOMATION_ENABLED),
+            (CONF_CELEBRATION_ENABLED, DEFAULT_AUTOMATION_ENABLED),
         ):
             if key in options:
                 values[key] = options[key]

custom_components/nestquest/schema.py

@@ -384,6 +384,21 @@ SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
     """,
 ]
 
+#: Generic integration-owned key/value state (schema version 7,
+#: Feature 11): persistent bookkeeping that belongs to the integration
+#: rather than any domain table — currently the missed-sweep
+#: watermark.  ``value`` is a plain string; callers interpret it.
+META_STATE_TABLE_SQL = """(
+    key TEXT NOT NULL PRIMARY KEY,
+    value TEXT NOT NULL
+)"""
+
+SCHEMA_V7_META_STATE_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS nestquest_meta_state {META_STATE_TABLE_SQL}
+    """,
+]
+
 SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_CHILDREN_DDL,
     *SCHEMA_V1_ADMIN_USERS_DDL,
@@ -395,6 +410,7 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     *SCHEMA_V1_QUEST_INSTANCES_DDL,
     *SCHEMA_V1_COMPLETION_EVENTS_DDL,
+    *SCHEMA_V7_META_STATE_DDL,
 ]
 
 

custom_components/nestquest/services.py

@@ -242,6 +242,19 @@ def _require_runtime(hass: HomeAssistant, find_runtime: FindRuntime):
     return runtime
 
 
+async def _refresh_entities(runtime: Any) -> None:
+    """Force an immediate coordinator refresh after a service mutation.
+
+    Only the SERVICE paths call this — the scheduled materialize and
+    rollover internals are direct function calls and never widen this
+    gate — so entities reflect a completion the moment the service
+    call returns instead of waiting for the poll interval.
+    """
+    coordinator = getattr(runtime, "coordinator", None)
+    if coordinator is not None:
+        await coordinator.async_refresh()
+
+
 def _generation_kwargs(hass: HomeAssistant, runtime: Any) -> dict[str, Any]:
     time_zone = ZoneInfo(hass.config.time_zone)
     today = datetime.datetime.now(time_zone).date()
@@ -293,18 +306,24 @@ def _complete_quest(
     hass: HomeAssistant, find_runtime: FindRuntime
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
-        database = _require_runtime(hass, find_runtime).database
+        runtime = _require_runtime(hass, find_runtime)
         instance_id = call.data["instance_id"]
         # ``appended`` is decided under the same lock as the write, so
         # a concurrent duplicate call (a double tap) can not produce a
         # second transition event: the loser appends nothing.
         result = await completion_layer.complete_instance(
-            database,
+            runtime.database,
             instance_id,
             **_actor_kwargs(call),
         )
         if result.appended:
-            await fire_quest_completed(hass, database, instance_id)
+            await fire_quest_completed(
+                hass,
+                runtime.database,
+                instance_id,
+                was_on_time=result.was_on_time,
+            )
+            await _refresh_entities(runtime)
 
     return _with_errors(_handler)
 
@@ -313,15 +332,18 @@ def _uncomplete_quest(
     hass: HomeAssistant, find_runtime: FindRuntime
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
-        database = _require_runtime(hass, find_runtime).database
+        runtime = _require_runtime(hass, find_runtime)
         instance_id = call.data["instance_id"]
         result = await completion_layer.uncomplete_instance(
-            database,
+            runtime.database,
             instance_id,
             **_actor_kwargs(call),
         )
         if result.appended:
-            await fire_quest_uncompleted(hass, database, instance_id)
+            await fire_quest_uncompleted(
+                hass, runtime.database, instance_id
+            )
+            await _refresh_entities(runtime)
 
     return _with_errors(_handler)
 

custom_components/nestquest/sweep.py

@@ -0,0 +1,132 @@
+"""The nightly missed-quest sweep (Feature 11).
+
+At the configured day-rollover time (the same HA-local listener that
+re-materializes the rolling horizon) this job finds every instance
+whose ``due_date`` is before the HA-local date, that has NO completion
+event, and that falls after the persisted sweep watermark — and fires
+ONE ``nestquest_quest_missed`` event per instance with the documented
+payload.  The sweep is read-only over the domain tables: it never
+mutates an instance and never writes a completion event (``missed``
+is derived state, never an event_type — Feature 08's contract).
+
+Idempotency is the watermark: after a run, ``nestquest_meta_state``
+records the HA-local date the sweep ran, and the next run only
+sweeps instances due in ``[watermark, today)``.  A second run the
+same night therefore fires nothing new, and a run after days of
+downtime sweeps exactly the accumulated window once — no
+per-instance journal, no re-announcing old misses.  An instance swept
+as missed that is later completed simply gains its completion event;
+history keeps both rows (append-only), and the sweep's later runs
+skip it via the no-completion-event rule.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import logging
+from zoneinfo import ZoneInfo
+
+from .const import EVENT_QUEST_MISSED
+from .dao_children import ChildrenDao
+from .dao_instances import QuestInstancesDao
+from .dao_meta import MetaStateDao
+from .dao_rules import QuestDefinitionsDao
+from .db import NestQuestDatabase
+
+LOGGER = logging.getLogger(__name__)
+
+#: meta_state key holding the HA-local date the sweep last ran.
+SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
+
+#: One asyncio.Lock per (connection wrapper, running loop), mirroring
+#: the migration runner's pattern: the whole watermark read → query →
+#: fire → write span is one critical section, so an overlapping startup
+#: sweep and rollover sweep queue instead of double-announcing.  The
+#: loser re-reads the watermark INSIDE the lock, sees the winner's
+#: date, and fires nothing.
+_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
+
+
+def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
+    """Return the sweep lock bound to ``database``'s running loop."""
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _SWEEP_LOCKS.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _SWEEP_LOCKS[key] = lock
+    return lock
+
+#: UTC ISO-8601 timestamp policy shared with the DAO layers.
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+def _now_stamp() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
+
+
+async def run_missed_sweep(
+    hass,
+    database: NestQuestDatabase,
+    *,
+    today: datetime.date | None = None,
+) -> int:
+    """Fire one missed event per unswept past-due open instance.
+
+    ``today`` optionally pins the HA-local date (threaded like the
+    materialization and completion layers); when omitted it is read
+    from ``hass.config.time_zone``.  Returns the number of events
+    fired.  Fires nothing and returns 0 when the watermark already
+    covers ``today`` (a same-night rerun).
+
+    The whole watermark read → query → fire → write span runs inside
+    the per-database sweep lock: an overlapping startup sweep and a
+    rollover sweep queue instead of interleaving, and the loser
+    re-reads the watermark inside the lock — sees the winner's date —
+    and announces nothing twice.
+    """
+    today_date = (
+        today
+        if today is not None
+        else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
+    )
+    today_iso = today_date.isoformat()
+    dao = MetaStateDao(database)
+    async with _sweep_lock(database):
+        watermark = await dao.get(SWEEP_WATERMARK_KEY)
+        if watermark == today_iso:
+            LOGGER.debug(
+                "NestQuest missed sweep already ran for %s", today_iso
+            )
+            return 0
+
+        instances_dao = QuestInstancesDao(database)
+        rows = await instances_dao.list_missed_unswept(today_iso, watermark)
+        children_dao = ChildrenDao(database)
+        definitions_dao = QuestDefinitionsDao(database)
+        now_stamp = _now_stamp()
+        fired = 0
+        for instance in rows:
+            child = await children_dao.get(instance.child_id)
+            definition = await definitions_dao.get(instance.definition_id)
+            hass.bus.async_fire(
+                EVENT_QUEST_MISSED,
+                {
+                    "child_id": instance.child_id,
+                    "child_name": child.display_name if child else None,
+                    "instance_id": instance.id,
+                    "quest_title": definition.title if definition else None,
+                    "window": instance.window,
+                    "due_date": instance.due_date,
+                    "due_time": instance.due_time,
+                    "occurred_at": now_stamp,
+                },
+            )
+            fired += 1
+        await dao.set(SWEEP_WATERMARK_KEY, today_iso)
+    LOGGER.info(
+        "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
+    )
+    return fired

tests/conftest.py

@@ -704,28 +704,40 @@ def make_hass() -> tuple:
     # The minimal entity registry: entities self-register through
     # Entity.async_write_ha_state keyed by unique_id, storing the live
     # object; tests enumerate created entities and read their state
-    # directly.
+    # directly.  entities_by_entry mirrors HA's per-entry platform
+    # scoping: unloading an entry's platforms removes exactly that
+    # entry's entities.
     hass.entities: dict[str, Entity] = {}
+    hass.entities_by_entry: dict[str, dict[str, Entity]] = {}
 
-    def _async_add_entities(new_entities, update_before_add=False):
-        """Mirror HA's AddEntitiesCallback: bind hass and register entities.
+    def _make_add_entities(entry):
+        entry_entities = hass.entities_by_entry.setdefault(entry.entry_id, {})
 
-        Accepts a single entity or an iterable, mirroring HA's
-        permissive call shape; registration goes through each entity's
-        async_write_ha_state (the mock state write).
-        """
-        if isinstance(new_entities, Entity):
-            new_entities = [new_entities]
-        for entity in new_entities:
-            entity.hass = hass
-            entity.async_write_ha_state()
+        def _async_add_entities(new_entities, update_before_add=False):
+            """Mirror HA's AddEntitiesCallback: bind hass and register.
+
+            Accepts a single entity or an iterable, mirroring HA's
+            permissive call shape; registration goes through each
+            entity's async_write_ha_state (the mock state write) and
+            is recorded per entry so an unload removes exactly this
+            entry's entities.
+            """
+            if isinstance(new_entities, Entity):
+                new_entities = [new_entities]
+            for entity in new_entities:
+                entity.hass = hass
+                entity.async_write_ha_state()
+                if entity.unique_id is not None:
+                    entry_entities[entity.unique_id] = entity
+
+        return _async_add_entities
 
     async def _async_forward_entry_setups(entry, platforms):
         """Mirror hass.config_entries.async_forward_entry_setups.
 
         HA's contract, loosely: import each platform module of the
-        entry's integration and drive its ``async_setup_entry`` with
-        the AddEntitiesCallback above.  A stub entry without a domain
+        entry's integration and drive its ``async_setup_entry`` with a
+        per-entry AddEntitiesCallback.  A stub entry without a domain
         defaults to this harness's single integration (NestQuest).
         """
         for platform in list(platforms):
@@ -733,14 +745,18 @@ def make_hass() -> tuple:
             module = importlib.import_module(
                 f"custom_components.{domain}.{platform}"
             )
-            await module.async_setup_entry(hass, entry, _async_add_entities)
+            await module.async_setup_entry(
+                hass, entry, _make_add_entities(entry)
+            )
             hass.config_entries.forwarded_platforms.append(platform)
 
     async def _async_unload_platforms(entry, platforms):
         """Mirror hass.config_entries.async_unload_platforms.
 
         Runs each platform module's ``async_unload_entry`` when it
-        defines one (per-platform unload is optional, like HA's).
+        defines one (per-platform unload is optional, like HA's), then
+        removes the entry's entities — unloading an entry's platforms
+        removes exactly that entry's entities, never another entry's.
         """
         for platform in list(platforms):
             domain = getattr(entry, "domain", "nestquest")
@@ -751,6 +767,8 @@ def make_hass() -> tuple:
             if unload is not None:
                 await unload(hass, entry)
             hass.config_entries.unloaded_platforms.append(platform)
+        for unique_id in hass.entities_by_entry.pop(entry.entry_id, {}):
+            hass.entities.pop(unique_id, None)
 
     hass.config_entries.forwarded_platforms: list[str] = []
     hass.config_entries.unloaded_platforms: list[str] = []

tests/test_entity_lifecycle.py

@@ -0,0 +1,187 @@
+"""Entity lifecycle across child add/deactivate/reactivate (Feature 10, task 8)."""
+from __future__ import annotations
+
+import datetime
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry, async_unload_entry
+from custom_components.nestquest.children import (
+    create_child,
+    list_children,
+    set_child_active,
+)
+from custom_components.nestquest.const import DOMAIN
+from custom_components.nestquest.materialize import materialize
+from custom_components.nestquest.quest_definitions import (
+    create_quest_definition,
+)
+from custom_components.nestquest.recurrence import ScheduleRule
+
+SEED_HORIZON_DAYS = 3
+PER_CHILD_KINDS = (
+    "quests_due_today",
+    "quests_completed_today",
+    "quests_remaining_today",
+    "completion_pct_today",
+    "next_quest",
+    "all_done",
+    "present_today",
+)
+HOUSEHOLD_UNIQUE_IDS = (
+    "nestquest_household_quests_due_today",
+    "nestquest_household_quests_completed_today",
+    "nestquest_cycle_day",
+)
+
+
+async def _reload(hass, entry) -> None:
+    """Unload then set up: the harness's restart/reload equivalent."""
+    assert await async_unload_entry(hass, entry) is True
+    assert await async_setup_entry(hass, entry) is True
+
+
+def _child_unique_ids(child_id: int) -> set[str]:
+    return {f"nestquest_child_{child_id}_{kind}" for kind in PER_CHILD_KINDS}
+
+
+async def _setup_and_seed(hass, make_entry):
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+    today = datetime.date.today()
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today.isoformat()}
+        ),
+        [ada.id, bo.id],
+        ["morning"],
+    )
+    end = (today + datetime.timedelta(days=SEED_HORIZON_DAYS)).isoformat()
+    await materialize(database, today.isoformat(), end, today=today)
+    await entry.runtime_data.coordinator.async_refresh()
+    return entry, ada, bo
+
+
+async def test_deactivated_child_entities_removed_on_reload(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _setup_and_seed(hass, make_entry)
+    assert _child_unique_ids(ada.id) <= set(hass.entities)
+    assert _child_unique_ids(bo.id) <= set(hass.entities)
+
+    await set_child_active(entry.runtime_data.database, bo.id, False)
+    await _reload(hass, entry)
+
+    assert _child_unique_ids(ada.id) <= set(hass.entities), (
+        "the surviving child's entities must be untouched"
+    )
+    assert not (_child_unique_ids(bo.id) & set(hass.entities)), (
+        "the deactivated child's entities must be gone after reload"
+    )
+
+
+async def test_reactivation_restores_entities_with_original_unique_ids(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _setup_and_seed(hass, make_entry)
+    original = _child_unique_ids(bo.id)
+    await set_child_active(entry.runtime_data.database, bo.id, False)
+    await _reload(hass, entry)
+    assert not (original & set(hass.entities))
+
+    await set_child_active(entry.runtime_data.database, bo.id, True)
+    await _reload(hass, entry)
+
+    assert original <= set(hass.entities), (
+        "reactivation must restore the exact original unique_ids"
+    )
+    bo_due = hass.entities[f"nestquest_child_{bo.id}_quests_due_today"]
+    assert bo_due.native_value == 1, "restored entities carry live state"
+
+
+async def test_new_child_gets_entities_without_disturbing_existing(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _setup_and_seed(hass, make_entry)
+    before_ada = set(hass.entities) & _child_unique_ids(ada.id)
+
+    cory = await create_child(entry.runtime_data.database, "Cory")
+    await _reload(hass, entry)
+
+    assert before_ada <= set(hass.entities), "existing entities untouched"
+    assert _child_unique_ids(cory.id) <= set(hass.entities), (
+        "the new child's entities appear after reload"
+    )
+    assert hass.entities[f"nestquest_child_{cory.id}_quests_due_today"].native_value == 0
+
+
+async def test_restart_equivalent_restores_all_entity_states(
+    hass, make_entry
+) -> None:
+    """Unload + setup (the restart equivalent) restores every active
+    child's entities with correct states and the household rollups."""
+    from custom_components.nestquest.completion import complete_instance
+
+    entry, ada, bo = await _setup_and_seed(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    snapshot_ada = next(
+        child
+        for child in coordinator.data.children
+        if child.child_id == ada.id
+    )
+    await complete_instance(
+        entry.runtime_data.database,
+        snapshot_ada.instances[0].instance_id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    await coordinator.async_refresh()
+    assert hass.entities[
+        f"nestquest_child_{ada.id}_quests_completed_today"
+    ].native_value == 1
+
+    await _reload(hass, entry)
+
+    assert hass.entities[
+        f"nestquest_child_{ada.id}_quests_completed_today"
+    ].native_value == 1, "completed count survives the restart"
+    assert hass.entities[
+        f"nestquest_child_{bo.id}_quests_completed_today"
+    ].native_value == 0
+    assert set(HOUSEHOLD_UNIQUE_IDS) <= set(hass.entities)
+    assert hass.entities[
+        "nestquest_household_quests_completed_today"
+    ].native_value == 1
+
+
+async def test_unload_shuts_coordinator_before_closing_database(
+    hass, make_entry
+) -> None:
+    """The coordinator's listeners are cleared while the database is
+    still connected: a scheduled refresh must never race the close."""
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    seen_connected = []
+
+    async def _probe_shutdown():
+        seen_connected.append(database.connected)
+
+    original_close = database.close
+
+    async def _recording_close():
+        seen_connected.append("closing")
+        await original_close()
+
+    coordinator.async_shutdown = _probe_shutdown
+    database.close = _recording_close
+    assert await async_unload_entry(hass, entry) is True
+    assert seen_connected == [True, "closing"], (
+        "coordinator shutdown must precede the database close"
+    )

tests/test_entity_matrix.py

@@ -0,0 +1,247 @@
+"""The Feature 10 entity-and-event matrix (task 9).
+
+One fixture drives every assertion the done-condition names: correct
+values for every sensor and binary sensor (including the zero-quest
+child and the 255-character state limit), the three events firing
+exactly once with schema-valid payloads, the instances payload
+omitting missed instances while admin_instances includes them, and no
+entities for an inactive child.  The per-task test files cover each
+behavior in depth; this file is the single pass that proves them all
+against ONE fixture, so a regression in any contract fails here too.
+"""
+from __future__ import annotations
+
+import datetime
+import re
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.children import (
+    create_child,
+    list_children,
+    set_child_active,
+)
+from custom_components.nestquest.completion import derive_state
+from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
+    DOMAIN,
+    EVENT_CHILD_DAY_COMPLETE,
+    EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_UNCOMPLETED,
+    SERVICE_COMPLETE_QUEST,
+    SERVICE_UNCOMPLETE_QUEST,
+)
+from custom_components.nestquest.dao_instances import QuestInstancesDao
+from custom_components.nestquest.materialize import materialize
+from custom_components.nestquest.quest_definitions import (
+    create_quest_definition,
+)
+from custom_components.nestquest.recurrence import ScheduleRule
+
+_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$")
+
+
+async def _fixture(hass, make_entry):
+    """Two quest children, one zero-quest child, one inactive child."""
+    entry = wire_entry_to_registry(
+        make_entry(data={CONF_ADMIN_USER_IDS: ["admin-1"]}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+    cory = await create_child(database, "Cory")
+    ghost = await create_child(database, "Ghost")
+    await set_child_active(database, ghost.id, False)
+    today = datetime.date.today()
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today.isoformat()}
+        ),
+        [ada.id, bo.id],
+        ["morning", "evening"],
+    )
+    end = (today + datetime.timedelta(days=3)).isoformat()
+    await materialize(database, today.isoformat(), end, today=today)
+    await entry.runtime_data.coordinator.async_refresh()
+    return entry, (ada, bo, cory, ghost)
+
+
+def _sensor(hass, child_id, kind):
+    return hass.entities[f"{DOMAIN}_child_{child_id}_{kind}"]
+
+
+async def test_entity_and_event_matrix(hass, make_entry) -> None:
+    entry, (ada, bo, cory, ghost) = await _fixture(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+
+    # --- Every active child has the full entity set; the inactive
+    # child has none.
+    kinds = (
+        "quests_due_today",
+        "quests_completed_today",
+        "quests_remaining_today",
+        "completion_pct_today",
+        "next_quest",
+        "all_done",
+        "present_today",
+    )
+    for child in (ada, bo, cory):
+        for kind in kinds:
+            assert f"{DOMAIN}_child_{child.id}_{kind}" in hass.entities, kind
+    for kind in kinds:
+        assert f"{DOMAIN}_child_{ghost.id}_{kind}" not in hass.entities, (
+            "inactive children must have no entities"
+        )
+
+    # --- Correct values against the fixture: Ada and Bo owe two
+    # quests (morning + evening), Cory owes none and is still 100%.
+    for child in (ada, bo):
+        assert _sensor(hass, child.id, "quests_due_today").native_value == 2
+        assert (
+            _sensor(hass, child.id, "quests_completed_today").native_value == 0
+        )
+        assert (
+            _sensor(hass, child.id, "quests_remaining_today").native_value == 2
+        )
+        assert (
+            _sensor(hass, child.id, "completion_pct_today").native_value == 0
+        )
+        assert _sensor(hass, child.id, "next_quest").native_value == (
+            "Brush teeth"
+        )
+        assert _sensor(hass, child.id, "all_done").is_on is False
+        assert _sensor(hass, child.id, "present_today").is_on is True
+    assert _sensor(hass, cory.id, "quests_due_today").native_value == 0
+    assert _sensor(hass, cory.id, "completion_pct_today").native_value == 100
+    assert _sensor(hass, cory.id, "all_done").is_on is False
+    assert _sensor(hass, cory.id, "next_quest").native_value == "none"
+
+    # --- Household rollups equal the per-child sums.
+    assert hass.entities[
+        "nestquest_household_quests_due_today"
+    ].native_value == 4
+    assert hass.entities[
+        "nestquest_household_quests_completed_today"
+    ].native_value == 0
+
+    # --- The 255-character limit holds for every registered entity.
+    for unique_id, entity in hass.entities.items():
+        state = getattr(entity, "native_value", None)
+        if state is None:
+            state = getattr(entity, "is_on", None)
+        assert state is None or not (
+            isinstance(state, str) and len(state) > 255
+        ), unique_id
+
+    # --- Completing both of Ada's quests fires the events exactly
+    # once each and flips her entities immediately.
+    instances = await QuestInstancesDao(
+        entry.runtime_data.database
+    ).list_by_date_range(
+        ada.id, datetime.date.today().isoformat(), datetime.date.today().isoformat()
+    )
+    for instance in instances:
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": ada.id,
+            },
+        )
+    assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 2
+    for payload in hass.bus.fired(EVENT_QUEST_COMPLETED):
+        assert _TIMESTAMP.match(payload["occurred_at"])
+        assert isinstance(payload["was_on_time"], bool)
+        assert payload["child_name"] in ("Ada",)
+        assert payload["quest_title"] == "Brush teeth"
+        assert isinstance(payload["instance_id"], int)
+    assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
+    assert _sensor(hass, ada.id, "all_done").is_on is True
+    assert _sensor(hass, ada.id, "quests_remaining_today").native_value == 0
+    assert (
+        _sensor(hass, ada.id, "completion_pct_today").native_value == 100
+    )
+
+    # --- Reversing one quest fires the uncompleted event exactly once
+    # and restores the owed counts.
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_UNCOMPLETE_QUEST,
+        {"instance_id": instances[0].id, "actor": "user"},
+        context={"user_id": "admin-1"},
+    )
+    assert len(hass.bus.fired(EVENT_QUEST_UNCOMPLETED)) == 1
+    assert _sensor(hass, ada.id, "all_done").is_on is False
+    assert _sensor(hass, ada.id, "quests_remaining_today").native_value == 1
+
+    # --- The instances payload omits missed instances while
+    # admin_instances includes them (D-009): pin the derive clock
+    # forward in the TEST ONLY so today's open instances read missed.
+    original_refresh = coordinator._async_update_data
+
+    async def _future_dated_refresh():
+        import custom_components.nestquest.coordinator as coordinator_module
+
+        real_derive = coordinator_module.derive_state
+
+        def _derive(instance, latest, today):
+            return real_derive(instance, latest, today + datetime.timedelta(days=1))
+
+        coordinator_module.derive_state = _derive
+        try:
+            return await original_refresh()
+        finally:
+            coordinator_module.derive_state = real_derive
+
+    coordinator._async_update_data = _future_dated_refresh
+    await coordinator.async_refresh()
+    attributes = _sensor(hass, bo.id, "quests_due_today").extra_state_attributes
+    assert attributes["instances"] == [], "panel payload omits missed"
+    assert len(attributes["admin_instances"]) == 2
+    assert all(
+        entry["state"] == "missed" for entry in attributes["admin_instances"]
+    )
+    coordinator._async_update_data = original_refresh
+    await coordinator.async_refresh()
+
+
+async def test_payload_schema_matches_the_documented_contract(
+    hass, make_entry
+) -> None:
+    """Every fired payload carries the documented fields with the
+    documented types."""
+    entry, (ada, _bo, _cory, _ghost) = await _fixture(hass, make_entry)
+    instances = await QuestInstancesDao(
+        entry.runtime_data.database
+    ).list_by_date_range(
+        ada.id, datetime.date.today().isoformat(), datetime.date.today().isoformat()
+    )
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": instances[0].id,
+            "actor": "panel",
+            "actor_child_id": ada.id,
+        },
+    )
+    payload = hass.bus.fired(EVENT_QUEST_COMPLETED)[0]
+    for key, check in (
+        ("child_id", lambda v: isinstance(v, int)),
+        ("child_name", lambda v: isinstance(v, str)),
+        ("instance_id", lambda v: isinstance(v, int)),
+        ("quest_title", lambda v: isinstance(v, str)),
+        ("window", lambda v: v in ("morning", "evening")),
+        ("due_date", lambda v: isinstance(v, str)),
+        ("due_time", lambda v: v is None or isinstance(v, str)),
+        ("occurred_at", lambda v: bool(_TIMESTAMP.match(v))),
+        ("was_on_time", lambda v: isinstance(v, bool)),
+    ):
+        assert key in payload, key
+        assert check(payload[key]), key

tests/test_events.py

@@ -278,3 +278,102 @@ async def test_day_complete_ignored_for_backdated_completion(
     # path fires), and the follow-up service call on the now-done
     # future instance was a no-op: today's single completion stands.
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == len(instances)
+
+
+async def test_completion_updates_entities_immediately(
+    hass, make_entry
+) -> None:
+    """The remaining-today sensor reflects the completion the moment
+    the service call returns — no manual refresh — and the day-complete
+    event fired exactly once when the day became clear."""
+    from custom_components.nestquest.const import EVENT_CHILD_DAY_COMPLETE
+
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    from conftest import wire_entry_to_registry  # noqa: F401
+
+    coordinator = entry.runtime_data.coordinator
+    # Entities only exist after one refresh since they were seeded
+    # after setup; create them, then prove the service path refreshes.
+    await coordinator.async_refresh()
+    remaining_before = hass.entities[
+        f"nestquest_child_{child.id}_quests_remaining_today"
+    ].native_value
+    assert remaining_before == len(instances)
+
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": instances[0].id,
+            "actor": "panel",
+            "actor_child_id": child.id,
+        },
+    )
+    remaining_after = hass.entities[
+        f"nestquest_child_{child.id}_quests_remaining_today"
+    ].native_value
+    assert remaining_after == len(instances) - 1, (
+        "sensor must reflect the completion without a manual refresh"
+    )
+    assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == (
+        1 if len(instances) == 1 else 0
+    )
+
+
+async def test_uncompletion_updates_entities_immediately(
+    hass, make_entry
+) -> None:
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    await coordinator.async_refresh()
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": instances[0].id,
+            "actor": "panel",
+            "actor_child_id": child.id,
+        },
+    )
+    completed_after_complete = hass.entities[
+        f"nestquest_child_{child.id}_quests_completed_today"
+    ].native_value
+    assert completed_after_complete == 1
+
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_UNCOMPLETE_QUEST,
+        {"instance_id": instances[0].id, "actor": "user"},
+        context=ADMIN_CTX,
+    )
+    completed_after_uncomplete = hass.entities[
+        f"nestquest_child_{child.id}_quests_completed_today"
+    ].native_value
+    assert completed_after_uncomplete == 0, (
+        "sensor must reflect the reversal without a manual refresh"
+    )
+
+
+async def test_forced_refreshes_serialize_no_stale_publish(
+    hass, make_entry
+) -> None:
+    """Overlapping forced refreshes never interleave: a slow pass
+    pauses mid-snapshot, and the queued pass still publishes AFTER
+    it, so the last published snapshot reflects the latest mutation."""
+    import asyncio
+
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    order: list[str] = []
+
+    async def _slow_update():
+        order.append("start")
+        await asyncio.sleep(0.01)
+        order.append("end")
+        return coordinator.data
+
+    coordinator._async_update_data = _slow_update
+    await asyncio.gather(coordinator.async_refresh(), coordinator.async_refresh())
+    assert order == ["start", "end", "start", "end"], (
+        "overlapping refreshes must serialize: " + str(order)
+    )

tests/test_migrations.py

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock
 import pytest
 
 from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.schema import SCHEMA_V7_META_STATE_DDL
 from custom_components.nestquest.migrations import (
     MIGRATIONS,
     VERSION_TABLE,
@@ -110,6 +111,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "presence_overrides",
             "quest_instances",
             "completion_events",
+            "nestquest_meta_state",
             VERSION_TABLE,
         }
         assert _tables(database) == expected
@@ -159,7 +161,8 @@ def test_migration_list_shape() -> None:
     assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
     assert callable(MIGRATIONS[4])
     assert callable(MIGRATIONS[5])
-    assert len(MIGRATIONS) == 6
+    assert MIGRATIONS[6] == SCHEMA_V7_META_STATE_DDL
+    assert len(MIGRATIONS) == 7
 
 
 # ---------------------------------------------------------------------------

tests/test_options_flow.py

@@ -19,6 +19,18 @@ from custom_components.nestquest.const import (
     CONF_HORIZON_DAYS,
     CONF_PANEL_IDLE_TIMEOUT,
     CONF_UPDATE_INTERVAL,
+    CONF_AFTERNOON_REMINDER_ENABLED,
+    CONF_AFTERNOON_REMINDER_TIME,
+    CONF_CELEBRATION_ENABLED,
+    CONF_END_OF_DAY_REPORT_ENABLED,
+    CONF_END_OF_DAY_REPORT_TIME,
+    CONF_MORNING_SUMMARY_ENABLED,
+    CONF_MORNING_SUMMARY_TIME,
+    CONF_NOTIFY_TARGET,
+    DEFAULT_AFTERNOON_REMINDER_TIME,
+    DEFAULT_AUTOMATION_ENABLED,
+    DEFAULT_END_OF_DAY_REPORT_TIME,
+    DEFAULT_MORNING_SUMMARY_TIME,
     DEFAULT_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
@@ -36,6 +48,22 @@ VALID_INPUT = {
     CONF_PANEL_IDLE_TIMEOUT: 60,
 }
 
+#: What a schema-validated submit produces from a bare VALID_INPUT on a
+#: fresh entry: the Feature 10 refresh interval and the Feature 11
+#: notification section fill from their defaults.
+FULL_INPUT = {
+    **VALID_INPUT,
+    CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
+    CONF_NOTIFY_TARGET: "",
+    CONF_MORNING_SUMMARY_TIME: DEFAULT_MORNING_SUMMARY_TIME,
+    CONF_AFTERNOON_REMINDER_TIME: DEFAULT_AFTERNOON_REMINDER_TIME,
+    CONF_END_OF_DAY_REPORT_TIME: DEFAULT_END_OF_DAY_REPORT_TIME,
+    CONF_MORNING_SUMMARY_ENABLED: DEFAULT_AUTOMATION_ENABLED,
+    CONF_AFTERNOON_REMINDER_ENABLED: DEFAULT_AUTOMATION_ENABLED,
+    CONF_END_OF_DAY_REPORT_ENABLED: DEFAULT_AUTOMATION_ENABLED,
+    CONF_CELEBRATION_ENABLED: DEFAULT_AUTOMATION_ENABLED,
+}
+
 
 def _run(coro):
     return asyncio.new_event_loop().run_until_complete(coro)
@@ -131,10 +159,7 @@ def test_options_flow_submits_through_2024_6_surface() -> None:
     result = _run(handler.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
     assert result["title"] == "NestQuest"
-    assert result["data"] == {
-        **VALID_INPUT,
-        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
-    }
+    assert result["data"] == FULL_INPUT
 
 
 def test_options_flow_shows_form_initially() -> None:
@@ -214,6 +239,10 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
             CONF_DAY_ROLLOVER_TIME: "05:45",
             CONF_PANEL_IDLE_TIMEOUT: 120,
             CONF_UPDATE_INTERVAL: 90,
+            CONF_NOTIFY_TARGET: "notify.mobile_app_test",
+            CONF_MORNING_SUMMARY_TIME: "07:00",
+            CONF_AFTERNOON_REMINDER_TIME: "14:00",
+            CONF_END_OF_DAY_REPORT_TIME: "21:30",
         }
     )
     result = _run(_make_flow(entry).async_step_init(None))
@@ -222,6 +251,14 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
         CONF_DAY_ROLLOVER_TIME: "05:45",
         CONF_PANEL_IDLE_TIMEOUT: 120,
         CONF_UPDATE_INTERVAL: 90,
+        CONF_NOTIFY_TARGET: "notify.mobile_app_test",
+        CONF_MORNING_SUMMARY_TIME: "07:00",
+        CONF_AFTERNOON_REMINDER_TIME: "14:00",
+        CONF_END_OF_DAY_REPORT_TIME: "21:30",
+        CONF_MORNING_SUMMARY_ENABLED: True,
+        CONF_AFTERNOON_REMINDER_ENABLED: True,
+        CONF_END_OF_DAY_REPORT_ENABLED: True,
+        CONF_CELEBRATION_ENABLED: True,
         CONF_ADMIN_USER_IDS: [],
     }
 
@@ -295,16 +332,8 @@ def test_options_flow_creates_entry_with_four_keys() -> None:
     result = _run(flow.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
     assert result["title"] == "NestQuest"
-    assert set(result["data"]) == {
-        CONF_HORIZON_DAYS,
-        CONF_DAY_ROLLOVER_TIME,
-        CONF_PANEL_IDLE_TIMEOUT,
-        CONF_UPDATE_INTERVAL,
-    }
-    assert result["data"] == {
-        **VALID_INPUT,
-        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
-    }
+    assert set(result["data"]) == set(FULL_INPUT)
+    assert result["data"] == FULL_INPUT
 
 
 def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once() -> None:
@@ -327,10 +356,7 @@ def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once(
     _run(registry.dispatch_options_update(entry))
     assert hass.config_entries.async_reload.call_count == 1
     assert registry.reloaded == [entry.entry_id]
-    assert entry.options == {
-        **VALID_INPUT,
-        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
-    }
+    assert entry.options == FULL_INPUT
 
 
 def test_options_flow_validation_never_raises() -> None:
@@ -649,3 +675,127 @@ def test_options_flow_omitted_picker_carries_allowlist_forward(
         assert result["data"][CONF_HORIZON_DAYS] == VALID_INPUT[CONF_HORIZON_DAYS]
     finally:
         _run(database.close())
+
+
+def test_options_flow_notification_section_defaults() -> None:
+    """A bare submit fills the notification section from const defaults:
+    08:00 / 15:00 / 20:00 — the report deliberately BEFORE the midnight
+    rollover sweep — an empty notify target, and all toggles enabled."""
+    flow = _make_flow(_make_entry())
+    result = _run(flow.async_step_init(dict(VALID_INPUT)))
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_MORNING_SUMMARY_TIME] == "08:00"
+    assert result["data"][CONF_AFTERNOON_REMINDER_TIME] == "15:00"
+    assert result["data"][CONF_END_OF_DAY_REPORT_TIME] == "20:00"
+    assert result["data"][CONF_NOTIFY_TARGET] == ""
+    for toggle in (
+        CONF_MORNING_SUMMARY_ENABLED,
+        CONF_AFTERNOON_REMINDER_ENABLED,
+        CONF_END_OF_DAY_REPORT_ENABLED,
+        CONF_CELEBRATION_ENABLED,
+    ):
+        assert result["data"][toggle] is True
+
+
+def test_options_flow_notification_times_must_be_strict_hh_mm() -> None:
+    flow = _make_flow(_make_entry())
+    for bad_time in ("9:00", "25:00", "08:0", "08-00", 800):
+        result = _run(
+            flow.async_step_init(
+                {
+                    **VALID_INPUT,
+                    CONF_MORNING_SUMMARY_TIME: bad_time,
+                }
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_MORNING_SUMMARY_TIME: "invalid_time"}
+
+
+def test_options_flow_notify_target_validation() -> None:
+    """The empty target is VALID (it is the shipped default and must be
+    savable); only malformed non-empty values reject."""
+    flow = _make_flow(_make_entry())
+    # The empty default saves cleanly.
+    result = _run(
+        flow.async_step_init(
+            {**VALID_INPUT, CONF_NOTIFY_TARGET: ""}
+        )
+    )
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_NOTIFY_TARGET] == ""
+
+    # Malformed non-empty values reject.
+    for bad_target in ("  notify.x  ", "   ", 42):
+        flow = _make_flow(_make_entry())
+        result = _run(
+            flow.async_step_init(
+                {
+                    **VALID_INPUT,
+                    CONF_NOTIFY_TARGET: bad_target,
+                }
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_NOTIFY_TARGET: "invalid"}
+
+
+def test_options_flow_toggles_must_be_real_booleans() -> None:
+    flow = _make_flow(_make_entry())
+    result = _run(
+        flow.async_step_init(
+            {
+                **VALID_INPUT,
+                CONF_CELEBRATION_ENABLED: "on",
+            }
+        )
+    )
+    assert result["type"] == "form"
+    assert result["errors"] == {CONF_CELEBRATION_ENABLED: "invalid"}
+
+
+def test_options_flow_notification_stored_values_round_trip() -> None:
+    """Stored notification settings are the form's defaults and a raw
+    submit without them keeps the stored values."""
+    entry = _make_entry(
+        options={
+            **VALID_INPUT,
+            CONF_UPDATE_INTERVAL: 90,
+            CONF_NOTIFY_TARGET: "notify.mobile_app_dad",
+            CONF_MORNING_SUMMARY_TIME: "07:15",
+            CONF_AFTERNOON_REMINDER_TIME: "16:45",
+            CONF_END_OF_DAY_REPORT_TIME: "21:00",
+            CONF_MORNING_SUMMARY_ENABLED: False,
+            CONF_AFTERNOON_REMINDER_ENABLED: True,
+            CONF_END_OF_DAY_REPORT_ENABLED: False,
+            CONF_CELEBRATION_ENABLED: True,
+        }
+    )
+    flow = _make_flow(entry)
+    form = _run(flow.async_step_init(None))
+    schema_defaults = form["data_schema"]({})
+    assert schema_defaults[CONF_NOTIFY_TARGET] == "notify.mobile_app_dad"
+    assert schema_defaults[CONF_MORNING_SUMMARY_TIME] == "07:15"
+    assert schema_defaults[CONF_END_OF_DAY_REPORT_ENABLED] is False
+
+    # A raw submit carrying only the legacy keys keeps the stored
+    # notification settings.
+    result = _run(flow.async_step_init(dict(VALID_INPUT)))
+    assert result["type"] == "create_entry"
+    assert result["data"][CONF_NOTIFY_TARGET] == "notify.mobile_app_dad"
+    assert result["data"][CONF_MORNING_SUMMARY_TIME] == "07:15"
+    assert result["data"][CONF_MORNING_SUMMARY_ENABLED] is False
+    assert result["data"][CONF_CELEBRATION_ENABLED] is True
+
+
+def test_options_flow_no_hard_coded_personal_target_in_source() -> None:
+    """The notify target's DEFAULT is the empty string: the household
+    fills their own service name, and no personal target ships as a
+    value anywhere in the const/flow defaults."""
+    from custom_components.nestquest import const
+
+    assert getattr(const, "CONF_NOTIFY_TARGET", None) == "notify_target"
+    # No const default carries a concrete notify service.
+    assert const.__dict__.get("DEFAULT_NOTIFY_TARGET", "") == ""
+    form = _run(_make_flow(_make_entry()).async_step_init(None))
+    assert form["data_schema"]({})[CONF_NOTIFY_TARGET] == ""

tests/test_schema.py

@@ -11,6 +11,7 @@ from custom_components.nestquest.schema import (
     SCHEMA_V1_ADMIN_USERS_DDL,
     SCHEMA_V1_CHILDREN_DDL,
     SCHEMA_V1_COMPLETION_EVENTS_DDL,
+    SCHEMA_V7_META_STATE_DDL,
     SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     SCHEMA_V1_SCHEDULE_RULES_DDL,
@@ -168,7 +169,7 @@ def _insert_override(database, child_id, **overrides):
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_eight_tables() -> None:
+def test_schema_v1_statements_compose_the_nine_tables() -> None:
     assert SCHEMA_V1_STATEMENTS == [
         *SCHEMA_V1_CHILDREN_DDL,
         *SCHEMA_V1_ADMIN_USERS_DDL,
@@ -180,6 +181,7 @@ def test_schema_v1_statements_compose_the_eight_tables() -> None:
         *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
         *SCHEMA_V1_QUEST_INSTANCES_DDL,
         *SCHEMA_V1_COMPLETION_EVENTS_DDL,
+        *SCHEMA_V7_META_STATE_DDL,
     ]
 
 

tests/test_sweep.py

@@ -0,0 +1,259 @@
+"""Tests for the nightly missed-quest sweep (Feature 11, task 1)."""
+from __future__ import annotations
+
+import datetime
+import re
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.children import create_child, list_children
+from custom_components.nestquest.completion import (
+    complete_instance,
+    uncomplete_instance,
+)
+from custom_components.nestquest.const import EVENT_QUEST_MISSED
+from custom_components.nestquest.dao_instances import (
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from custom_components.nestquest.dao_meta import MetaStateDao
+from custom_components.nestquest.materialize import materialize
+from custom_components.nestquest.quest_definitions import (
+    create_quest_definition,
+)
+from custom_components.nestquest.recurrence import ScheduleRule
+from custom_components.nestquest.sweep import (
+    SWEEP_WATERMARK_KEY,
+    run_missed_sweep,
+)
+
+_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$")
+
+
+async def _seed(hass, make_entry) -> tuple:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+    today = datetime.date.today()
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today.isoformat()}
+        ),
+        [ada.id, bo.id],
+        ["morning"],
+    )
+    end = (today + datetime.timedelta(days=3)).isoformat()
+    await materialize(database, today.isoformat(), end, today=today)
+    return entry, ada, bo
+
+
+async def test_sweep_fires_one_event_per_past_due_open_instance(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 2, "one per child for today's now-past-due instance"
+
+    events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert len(events) == 2
+    payload = events[0]
+    assert payload["child_name"] in ("Ada", "Bo")
+    assert payload["quest_title"] == "Brush teeth"
+    assert payload["window"] == "morning"
+    assert payload["due_date"] == today.isoformat()
+    assert payload["due_time"] is None
+    assert _TIMESTAMP.match(payload["occurred_at"])
+    assert {payload["child_id"] for payload in events} == {ada.id, bo.id}
+    # The sweep is read-only over the domain tables.
+    assert await _count_events(database) == 0
+
+
+async def test_sweep_is_idempotent_same_night(hass, make_entry) -> None:
+    entry, _ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+
+    first = await run_missed_sweep(hass, database, today=tomorrow)
+    second = await run_missed_sweep(hass, database, today=tomorrow)
+    assert first == 2
+    assert second == 0, "the watermark makes a rerun fire nothing"
+    assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2
+    assert await MetaStateDao(database).get(SWEEP_WATERMARK_KEY) == (
+        tomorrow.isoformat()
+    )
+
+
+async def test_sweep_only_fires_new_instances_after_watermark(
+    hass, make_entry
+) -> None:
+    entry, ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+    day_after = today + datetime.timedelta(days=2)
+
+    # First sweep one day ahead: today's instances fire.
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 2
+    first_events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert all(
+        payload["due_date"] == today.isoformat()
+        for payload in first_events
+    )
+
+    # One more day passes: only tomorrow's instances fire now.
+    fired = await run_missed_sweep(hass, database, today=day_after)
+    assert fired == 2
+    all_events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert len(all_events) == 4
+    assert all(
+        payload["due_date"] == tomorrow.isoformat()
+        for payload in all_events[2:]
+    )
+
+
+async def test_sweep_never_reports_completed_or_reversed_instances(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+
+    instances_ada = await QuestInstancesDao(database).list_by_date_range(
+        ada.id, today.isoformat(), today.isoformat()
+    )
+    instances_bo = await QuestInstancesDao(database).list_by_date_range(
+        bo.id, today.isoformat(), today.isoformat()
+    )
+    # Ada completed hers; Bo completed his then reversed it.
+    await complete_instance(
+        database,
+        instances_ada[0].id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    await complete_instance(
+        database,
+        instances_bo[0].id,
+        actor_source="panel",
+        actor_child_id=bo.id,
+    )
+    await uncomplete_instance(
+        database,
+        instances_bo[0].id,
+        actor_source="user",
+        actor_user_id="admin-1",
+    )
+
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 0, (
+        "no-completion-event rule excludes both the completed and the "
+        "reversed instance (append-only history keeps both)"
+    )
+    assert hass.bus.fired(EVENT_QUEST_MISSED) == []
+
+
+async def test_sweep_skips_instance_completed_after_sweep(
+    hass, make_entry
+) -> None:
+    """A missed instance completed later is never re-announced; the
+    late completion still lands in the append-only log."""
+    entry, ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+    day3 = today + datetime.timedelta(days=2)
+
+    await run_missed_sweep(hass, database, today=tomorrow)
+    instance = (
+        await QuestInstancesDao(database).list_by_date_range(
+            ada.id, today.isoformat(), today.isoformat()
+        )
+    )[0]
+    await complete_instance(
+        database,
+        instance.id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    fired = await run_missed_sweep(hass, database, today=day3)
+    assert fired == 0 or all(
+        payload["instance_id"] != instance.id
+        for payload in hass.bus.fired(EVENT_QUEST_MISSED)[2:]
+    )
+
+
+async def test_rollover_listener_runs_sweep_after_materialization(
+    hass, make_entry, monkeypatch
+) -> None:
+    import custom_components.nestquest as nestquest
+
+    calls = []
+
+    async def _record_sweep(hass_arg, database, **kwargs):
+        calls.append("sweep")
+
+    monkeypatch.setattr(nestquest, "run_missed_sweep", _record_sweep)
+
+    entry = wire_entry_to_registry(
+        make_entry(options={"day_rollover_time": "00:00"}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+    assert calls == ["sweep"], "startup runs the sweep once"
+
+    await hass.time_change.fire()
+    assert calls == ["sweep", "sweep"], "the rollover listener sweeps too"
+
+
+async def test_v6_database_migrates_to_v7_meta_state(
+    hass, make_entry
+) -> None:
+    """A database stamped at version 6 gains the meta_state table and
+    version 7 on the next startup; existing rows are untouched."""
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    version = await database.fetch_one(
+        "SELECT version FROM nestquest_schema_version"
+    )
+    assert version == (7,)
+    tables = await database.fetch_all(
+        "SELECT name FROM sqlite_master WHERE type='table' "
+        "AND name='nestquest_meta_state'"
+    )
+    assert tables, "migration 7 created the meta_state table"
+
+
+async def _count_events(database) -> int:
+    rows = await database.fetch_all("SELECT COUNT(*) FROM completion_events")
+    return rows[0][0]
+
+
+async def test_overlapping_sweeps_never_double_announce(
+    hass, make_entry
+) -> None:
+    """A startup sweep and a rollover sweep racing on the same night
+    queue on the sweep lock: the winner announces, the loser re-reads
+    the watermark inside the lock and fires nothing."""
+    import asyncio
+
+    entry, _ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+
+    first, second = await asyncio.gather(
+        run_missed_sweep(hass, database, today=tomorrow),
+        run_missed_sweep(hass, database, today=tomorrow),
+    )
+    assert first + second == 2, "each instance announced exactly once"
+    assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2