Serialize the missed sweep's watermark claim

The whole watermark read-query-fire-write span runs under a
per-database sweep lock; a racing second sweep re-reads the watermark
inside the lock and announces nothing.

README.md

@@ -38,6 +38,35 @@ After installation and restart:
 2. Click **Add Integration** and search for **NestQuest**.
 3. Follow the on-screen configuration flow.
 
+## Events
+
+NestQuest fires Home Assistant bus events so automations can react to a
+child's progress. Each payload carries `child_id`, `child_name`,
+`instance_id`, `quest_title`, the instance's `window`, `due_date`,
+`due_time`, and `occurred_at` (a strict UTC ISO-8601 timestamp).
+
+| Event | Fires when | Extra payload fields |
+|---|---|---|
+| `nestquest_quest_completed` | a quest is completed (an actual completion, never a re-complete no-op) | `was_on_time` (bool) |
+| `nestquest_quest_uncompleted` | an admin reverses a completion | — |
+| `nestquest_child_day_complete` | a completion clears the child's whole day (quests were owed and none remain; a zero-quest day never fires it) | `quests_due`, `quests_completed` |
+| `nestquest_quest_missed` | the nightly sweep marks a still-open past-due quest missed (integrated in a later release) | — |
+
+Example automation trigger:
+
+```yaml
+triggers:
+  - trigger: event
+    event_type: nestquest_child_day_complete
+conditions:
+  - condition: template
+    value_template: "{{ trigger.payload.child_name == 'Declan' }}"
+actions:
+  - action: notify.mobile_app_declans_phone
+    data:
+      message: "Every quest cleared — legendary!"
+```
+
 ## Developer Setup and Testing
 
 This project uses [`uv`](https://github.com/astral-sh/uv) for fast Python package and dependency management.

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/binary_sensor.py

@@ -0,0 +1,191 @@
+"""Per-child binary sensors (Feature 10, task 4).
+
+Two binary sensors per active child, both reading only the coordinator
+snapshot (the same ChildDaySnapshot the day sensors read):
+
+- ``binary_sensor.nestquest_<child>_all_done`` — ON exactly when the
+  child owes quests today (due today > 0) AND owes none of them any
+  more (remaining today == 0).  A zero-quest day is OFF: the
+  celebration automation must never fire on an empty day (the
+  day-complete event carries the same rule, Feature 10 task 7).
+- ``binary_sensor.nestquest_<child>_present_today`` — ON when the
+  presence engine resolves the child present on the HA-local date
+  (schedules and overrides, D-004 anchor-date arithmetic).
+
+Naming and registry identity follow the day sensors' contract: names
+slug to the documented entity_id shape; unique_ids are built from the
+child's database id, never the display name, so a rename never
+orphans an entity's history.
+"""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from typing import Any
+
+from homeassistant.components.binary_sensor import BinarySensorEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import DOMAIN
+from .coordinator import ChildDaySnapshot, NestQuestCoordinator
+
+
+class _NestQuestChildBinarySensor(CoordinatorEntity, BinarySensorEntity):
+    """Base for the per-child binary sensors.
+
+    Registry identity is built from the child's database id and never
+    changes; the displayed name and device label re-derive from the
+    child's CURRENT snapshot on every read, so a rename propagates on
+    the next refresh without orphaning history.  A child absent from
+    the latest snapshot (a deactivation; removal is a later task's
+    scope) yields ``None`` states rather than stale values.
+    """
+
+    _name_suffix: str = ""
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator)
+        self._child_id = child.child_id
+        self._fallback_name = child.child_name
+
+    @property
+    def _child(self) -> ChildDaySnapshot | None:
+        """Return the child's current day snapshot, or None."""
+        data = self.coordinator.data
+        if data is None:
+            return None
+        for child in data.children:
+            if child.child_id == self._child_id:
+                return child
+        return None
+
+    def _displayed_child_name(self) -> str:
+        child = self._child
+        return child.child_name if child is not None else self._fallback_name
+
+    @property
+    def name(self) -> str | None:
+        return f"NestQuest {self._displayed_child_name()} {self._name_suffix}"
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        return DeviceInfo(
+            identifiers={(DOMAIN, f"child_{self._child_id}")},
+            name=f"NestQuest {self._displayed_child_name()}",
+            manufacturer="NestQuest",
+        )
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        child = self._child
+        if child is None:
+            return None
+        return {
+            "child_id": child.child_id,
+            "child_name": child.child_name,
+        }
+
+
+class NestQuestAllDoneSensor(_NestQuestChildBinarySensor):
+    """ON when the child had quests today and owes none of them."""
+
+    _name_suffix = "all done"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = f"{DOMAIN}_child_{child.child_id}_all_done"
+
+    @property
+    def is_on(self) -> bool | None:
+        child = self._child
+        if child is None:
+            return None
+        # Zero-quest days are OFF, not ON: an empty day must never fire
+        # the celebration automation.
+        return child.due_today > 0 and child.remaining_today == 0
+
+
+class NestQuestPresentTodaySensor(_NestQuestChildBinarySensor):
+    """ON when the child is at this house today (presence engine)."""
+
+    _name_suffix = "present today"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = f"{DOMAIN}_child_{child.child_id}_present_today"
+
+    @property
+    def is_on(self) -> bool | None:
+        child = self._child
+        if child is None:
+            return None
+        return child.present
+
+
+def _child_binary_sensors(
+    coordinator: NestQuestCoordinator, child: ChildDaySnapshot
+) -> tuple[_NestQuestChildBinarySensor, ...]:
+    """Build the two documented binary sensors for one active child."""
+    return (
+        NestQuestAllDoneSensor(coordinator, child),
+        NestQuestPresentTodaySensor(coordinator, child),
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: Callable[..., None],
+) -> None:
+    """Create the two binary sensors per active child.
+
+    Entities are created for the snapshot's children at platform
+    setup; a coordinator listener adds sensors for children that first
+    appear on a later refresh (the day sensors' same pattern, so a new
+    child's whole entity set lands together).  Deactivated children
+    are NOT removed here — removal is a later task's scope.
+    """
+    coordinator = entry.runtime_data.coordinator
+    known_child_ids: set[int] = set()
+
+    def _add_new_children(children: Iterable[ChildDaySnapshot]) -> None:
+        new_children = [
+            child
+            for child in children
+            if child.child_id not in known_child_ids
+        ]
+        if not new_children:
+            return
+        known_child_ids.update(child.child_id for child in new_children)
+        async_add_entities(
+            [
+                sensor
+                for child in new_children
+                for sensor in _child_binary_sensors(coordinator, child)
+            ]
+        )
+
+    data = coordinator.data
+    if data is not None:
+        _add_new_children(data.children)
+
+    def _handle_coordinator_update() -> None:
+        data = coordinator.data
+        if data is not None:
+            _add_new_children(data.children)
+
+    coordinator.async_add_listener(_handle_coordinator_update)

custom_components/nestquest/completion.py

@@ -366,6 +366,36 @@ async def instance_state(
     return _derived_state(instance, latest, today_date)
 
 
+class CompletionResult(str):
+    """The derived instance state, carrying the append verdict.
+
+    A ``str`` subclass so every existing ``result == "done"`` contract
+    keeps working; ``appended`` reports whether THIS call actually
+    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.  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,
+        was_on_time: bool | None = None,
+    ) -> "CompletionResult":
+        obj = super().__new__(cls, state)
+        obj.appended = appended
+        obj.was_on_time = was_on_time
+        return obj
+
+
 async def complete_instance(
     database: NestQuestDatabase,
     instance_id: int,
@@ -375,14 +405,14 @@ async def complete_instance(
     actor_child_id: int | None = None,
     now: datetime.datetime | None = None,
     today: datetime.date | None = None,
-) -> str:
+) -> CompletionResult:
     """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.
+    Returns the derived state ``done``; ``appended`` reports whether
+    THIS call appended (False on an already-done no-op).  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 computed from the HA-local completion moment
     (threaded ``now`` / ``today``) against the instance ``due_date``
@@ -404,12 +434,16 @@ async def complete_instance(
             instance_id
         )
         if latest is not None and latest.event_type == EVENT_COMPLETED:
-            return "done"
+            return CompletionResult("done", appended=False)
         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}"
             )
+        # 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,
@@ -418,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 "done"
+    return CompletionResult("done", appended=True, was_on_time=on_time)
 
 
 async def uncomplete_instance(
@@ -433,16 +467,16 @@ async def uncomplete_instance(
     actor_child_id: int | None = None,
     now: datetime.datetime | None = None,
     today: datetime.date | None = None,
-) -> str:
+) -> CompletionResult:
     """Append an uncompleted event unless the instance is already open.
 
     Returns the derived state ``open``, or ``missed`` when the instance
-    is past due (``due_date`` before HA-local ``today``).  An
-    already-open 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 uncomplete cannot double-append.  The original completed
-    row is not modified.
+    is past due (``due_date`` before HA-local ``today``); ``appended``
+    reports whether THIS call appended (False on an already-open
+    no-op).  Unknown ``instance_id`` raises ValueError naming the
+    field.  Existence, latest-event, and the append run under one
+    connection lock so a concurrent uncomplete cannot double-append.
+    The original completed row is not modified.
     """
     instance_id = _validate_int_id(instance_id, "instance_id")
     actor_source, actor_user_id, actor_child_id = _validate_actor(
@@ -459,7 +493,10 @@ async def uncomplete_instance(
             instance_id
         )
         if latest is None or latest.event_type == EVENT_UNCOMPLETED:
-            return _derived_state(instance, latest, today_date)
+            return CompletionResult(
+                _derived_state(instance, latest, today_date),
+                appended=False,
+            )
         if latest.event_type != EVENT_COMPLETED:
             raise ValueError(
                 f"event_type must be '{EVENT_COMPLETED}' or "
@@ -476,7 +513,9 @@ async def uncomplete_instance(
             was_on_time=None,
             now=now,
         )
-    return _derived_state(instance, latest, today_date)
+    return CompletionResult(
+        _derived_state(instance, latest, today_date), appended=True
+    )
 
 
 async def list_missed_for_child(

custom_components/nestquest/const.py

@@ -8,8 +8,17 @@ LOGGER_NAME = "custom_components.nestquest"
 LOGGER = logging.getLogger(LOGGER_NAME)
 
 # Platforms set up per config entry (Feature 10): the per-child day
-# sensors. binary_sensor joins when task 4's entities land.
-PLATFORMS: list[str] = ["sensor"]
+# sensors and binary sensors.
+PLATFORMS: list[str] = ["sensor", "binary_sensor"]
+
+# Home Assistant bus events (design/ENTITIES-AND-SERVICES.md §3).  The
+# panel and admin cards listen for these; Feature 10 fires completed,
+# uncompleted, and child_day_complete from the service paths, and
+# Feature 11's nightly sweep fires quest_missed.
+EVENT_QUEST_COMPLETED = "nestquest_quest_completed"
+EVENT_QUEST_UNCOMPLETED = "nestquest_quest_uncompleted"
+EVENT_QUEST_MISSED = "nestquest_quest_missed"
+EVENT_CHILD_DAY_COMPLETE = "nestquest_child_day_complete"
 
 #: Canonical domain-global service names (design/ENTITIES-AND-SERVICES.md §2
 #: plus the existing Feature 07 regenerate service).

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

@@ -0,0 +1,130 @@
+"""HA bus events fired by the NestQuest service paths (Feature 10).
+
+The four documented events (design/ENTITIES-AND-SERVICES.md §3, names
+in :mod:`~.const`): ``nestquest_quest_completed`` and
+``nestquest_quest_uncompleted`` fire from the completion service
+handlers on an actual transition (a no-op re-complete or re-uncomplete
+fires nothing), and ``nestquest_child_day_complete`` fires when a
+completion clears the child's whole day (quests were owed and none
+remain — a zero-quest day can never fire it, matching the all-done
+binary sensor's rule).  ``nestquest_quest_missed`` is the documented
+contract Feature 11's nightly sweep fires; nothing here fires it.
+
+Payload policy: one dict carrying ``child_id``, ``child_name``,
+``instance_id``, ``quest_title``, the instance's ``window``/
+``due_date``/``due_time`` so automation authors can filter without a
+database lookup, and ``occurred_at`` — the event's own strict UTC
+ISO-8601 timestamp.  Completion payloads additionally carry
+``was_on_time``.  No other personal data: names and titles are the
+household's own records, and nothing beyond them leaks.
+"""
+from __future__ import annotations
+
+import datetime
+from zoneinfo import ZoneInfo
+
+from .completion import derive_state
+from .const import (
+    EVENT_CHILD_DAY_COMPLETE,
+    EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_UNCOMPLETED,
+)
+from .dao_children import ChildrenDao
+from .dao_instances import CompletionEventsDao, QuestInstancesDao
+from .dao_rules import QuestDefinitionsDao
+from .db import NestQuestDatabase
+
+#: Strict UTC ISO-8601 timestamp (the DAO layers' policy): one shape,
+#: explicit offset, second precision.
+_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 _instance_payload(
+    database: NestQuestDatabase, instance_id: int, now_stamp: str
+) -> dict:
+    """Build the shared event payload for one quest instance."""
+    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"
+        )
+    child = await ChildrenDao(database).get(instance.child_id)
+    definition = await QuestDefinitionsDao(database).get(
+        instance.definition_id
+    )
+    return {
+        "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,
+    }
+
+
+async def fire_quest_completed(
+    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, 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)
+    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
+    # rule as the all-done binary sensor; a zero-quest day never fires).
+    # Only a completion of TODAY'S instance evaluates it: completing a
+    # back-dated or future instance must not announce a cleared day.
+    today = datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
+    if payload["due_date"] != today.isoformat():
+        return
+    instances = await QuestInstancesDao(database).list_by_date_range(
+        payload["child_id"], today.isoformat(), today.isoformat()
+    )
+    events = CompletionEventsDao(database)
+    remaining = 0
+    for instance in instances:
+        latest_for_instance = await events.get_latest_for_instance(
+            instance.id
+        )
+        if derive_state(instance, latest_for_instance, today) != "done":
+            remaining += 1
+    if instances and remaining == 0:
+        hass.bus.async_fire(
+            EVENT_CHILD_DAY_COMPLETE,
+            {
+                "child_id": payload["child_id"],
+                "child_name": payload["child_name"],
+                "quests_due": len(instances),
+                "quests_completed": len(instances),
+                "occurred_at": now_stamp,
+            },
+        )
+
+
+async def fire_quest_uncompleted(
+    hass, database: NestQuestDatabase, instance_id: int
+) -> None:
+    """Fire ``nestquest_quest_uncompleted`` for an actual reversal."""
+    payload = await _instance_payload(database, instance_id, _now_stamp())
+    hass.bus.async_fire(EVENT_QUEST_UNCOMPLETED, payload)

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/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/sensor.py

@@ -369,6 +369,123 @@ def _child_day_sensors(
     )
 
 
+class _NestQuestHouseholdSensor(CoordinatorEntity, SensorEntity):
+    """Base for the household-level rollup sensors.
+
+    Aggregates the ACTIVE children's snapshots only (the snapshot
+    already excludes inactive children), matching the per-child
+    sensors exactly — a household number is always the sum of the
+    per-child numbers.  Grouped under one household device so the
+    three household sensors render together.
+    """
+
+    _attr_name: str
+
+    def __init__(self, coordinator: NestQuestCoordinator) -> None:
+        super().__init__(coordinator)
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, "household")},
+            name="NestQuest Household",
+            manufacturer="NestQuest",
+        )
+
+
+class NestQuestHouseholdDueTodaySensor(_NestQuestHouseholdSensor):
+    """Total quests due today across active children."""
+
+    _attr_name = "NestQuest household quests due today"
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_unique_id = f"{DOMAIN}_household_quests_due_today"
+
+    @property
+    def native_value(self) -> int | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return _state_within_limit(
+            sum(child.due_today for child in data.children)
+        )
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return {
+            "children": {
+                str(child.child_id): child.due_today
+                for child in data.children
+            }
+        }
+
+
+class NestQuestHouseholdCompletedTodaySensor(_NestQuestHouseholdSensor):
+    """Total quests completed today across active children."""
+
+    _attr_name = "NestQuest household quests completed today"
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_unique_id = f"{DOMAIN}_household_quests_completed_today"
+
+    @property
+    def native_value(self) -> int | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return _state_within_limit(
+            sum(child.completed_today for child in data.children)
+        )
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return {
+            "children": {
+                str(child.child_id): child.completed_today
+                for child in data.children
+            }
+        }
+
+
+class NestQuestCycleDaySensor(_NestQuestHouseholdSensor):
+    """Cycle day N of the custody cycle (0 when nobody is scheduled).
+
+    Computed by the coordinator from the first active child with a
+    presence schedule, anchor-date arithmetic only (D-004 — never ISO
+    week parity), so a 53-week year cannot silently invert it.
+    """
+
+    _attr_name = "NestQuest cycle day"
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_unique_id = f"{DOMAIN}_cycle_day"
+
+    @property
+    def native_value(self) -> int | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return _state_within_limit(data.cycle_day)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        data = self.coordinator.data
+        if data is None:
+            return None
+        return {"cycle_day": data.cycle_day, "today": data.today_iso}
+
+
+def _household_sensors(
+    coordinator: NestQuestCoordinator,
+) -> tuple[_NestQuestHouseholdSensor, ...]:
+    """Build the three household-level sensors (one set per entry)."""
+    return (
+        NestQuestHouseholdDueTodaySensor(coordinator),
+        NestQuestHouseholdCompletedTodaySensor(coordinator),
+        NestQuestCycleDaySensor(coordinator),
+    )
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
@@ -405,6 +522,10 @@ async def async_setup_entry(
     data = coordinator.data
     if data is not None:
         _add_new_children(data.children)
+        # The household rollups are child-independent: created exactly
+        # once at platform setup, they aggregate whatever the snapshot
+        # carries (empty included — an empty household is a valid 0).
+        async_add_entities(_household_sensors(coordinator))
 
     def _handle_coordinator_update() -> None:
         data = coordinator.data

custom_components/nestquest/services.py

@@ -42,6 +42,7 @@ from .const import (
     SERVICE_UPDATE_QUEST_DEFINITION,
 )
 from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
+from .events import fire_quest_completed, fire_quest_uncompleted
 from .materialize import regenerate_for_child
 from .permissions import permission_gate
 from .presence import PresenceSchedule
@@ -241,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()
@@ -292,12 +306,24 @@ def _complete_quest(
     hass: HomeAssistant, find_runtime: FindRuntime
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
-        database = _require_runtime(hass, find_runtime).database
-        await completion_layer.complete_instance(
-            database,
-            call.data["instance_id"],
+        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(
+            runtime.database,
+            instance_id,
             **_actor_kwargs(call),
         )
+        if result.appended:
+            await fire_quest_completed(
+                hass,
+                runtime.database,
+                instance_id,
+                was_on_time=result.was_on_time,
+            )
+            await _refresh_entities(runtime)
 
     return _with_errors(_handler)
 
@@ -306,12 +332,18 @@ def _uncomplete_quest(
     hass: HomeAssistant, find_runtime: FindRuntime
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
-        database = _require_runtime(hass, find_runtime).database
-        await completion_layer.uncomplete_instance(
-            database,
-            call.data["instance_id"],
+        runtime = _require_runtime(hass, find_runtime)
+        instance_id = call.data["instance_id"]
+        result = await completion_layer.uncomplete_instance(
+            runtime.database,
+            instance_id,
             **_actor_kwargs(call),
         )
+        if result.appended:
+            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

@@ -549,6 +549,31 @@ def _async_track_time_change(hass, action, hour=None, minute=None, second=None):
 _ha_mock("homeassistant.helpers.event").async_track_time_change = _async_track_time_change
 
 
+class EventBus:
+    """Models HA's ``hass.bus`` event bus.
+
+    ``async_fire`` is SYNCHRONOUS (matching real HA) and records
+    ``(event_type, payload)`` pairs in order so tests can assert
+    exactly which events fired with which payloads; there is no
+    listener dispatch (entities poll the coordinator, and the cards
+    are not part of this harness).
+    """
+
+    def __init__(self):
+        self.events: list[tuple[str, dict]] = []
+
+    def async_fire(self, event_type: str, payload=None):
+        self.events.append((event_type, dict(payload or {})))
+
+    def fired(self, event_type: str) -> list[dict]:
+        """Return every payload fired for ``event_type``, in order."""
+        return [
+            payload
+            for fired_type, payload in self.events
+            if fired_type == event_type
+        ]
+
+
 class ServiceRegistry:
     """Models HA's ``hass.services`` service registry.
 
@@ -671,6 +696,7 @@ def make_hass() -> tuple:
     registry = ListenerRegistry(hass)
     hass.time_change = TimeChangeRegistry(hass)
     hass.services = ServiceRegistry(hass)
+    hass.bus = EventBus()
     # A valid IANA time zone (the day-rollover listener reads it to
     # compute "today" in HA local time).
     hass.config.time_zone = "UTC"
@@ -678,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):
@@ -707,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")
@@ -725,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_binary_sensors.py

@@ -0,0 +1,134 @@
+"""Tests for the per-child binary sensors (Feature 10, task 4)."""
+from __future__ import annotations
+
+import datetime
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.children import create_child
+from custom_components.nestquest.const import DOMAIN
+from custom_components.nestquest.dao_presence import PresenceSchedulesDao
+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.completion import complete_instance
+
+SEED_HORIZON_DAYS = 3
+
+
+def _binary(hass, child_id: int, kind: str):
+    return hass.entities[f"{DOMAIN}_child_{child_id}_{kind}"]
+
+
+async def _setup_entry(hass, make_entry):
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    return entry
+
+
+async def _seed_two_children(database) -> tuple:
+    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)
+    return ada, bo
+
+
+async def test_all_done_off_with_open_quests_and_zero_quests(
+    hass, make_entry
+) -> None:
+    """all_done is OFF while quests are owed AND on a zero-quest day;
+    a child with no quests never fires the celebration automation."""
+    entry = await _setup_entry(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    ada, _bo = await _seed_two_children(database)
+    await create_child(database, "Cory")  # zero quests
+    await coordinator.async_refresh()
+
+    assert _binary(hass, ada.id, "all_done").is_on is False
+    cory = [
+        child for child in coordinator.data.children
+        if child.child_name == "Cory"
+    ][0]
+    assert _binary(hass, cory.child_id, "all_done").is_on is False
+
+
+async def test_all_done_on_only_when_day_clear(hass, make_entry) -> None:
+    entry = await _setup_entry(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    ada, _bo = await _seed_two_children(database)
+    await coordinator.async_refresh()
+    snapshot_ada = [
+        child for child in coordinator.data.children
+        if child.child_id == ada.id
+    ][0]
+    instance_id = snapshot_ada.instances[0].instance_id
+    await complete_instance(
+        database,
+        instance_id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    await coordinator.async_refresh()
+    assert _binary(hass, ada.id, "all_done").is_on is True
+
+
+async def test_present_today_on_without_schedule_off_when_absent(
+    hass, make_entry
+) -> None:
+    """No schedule means present every day; an all-absent pattern
+    today reports OFF."""
+    entry = await _setup_entry(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    ada, _bo = await _seed_two_children(database)
+    await create_child(database, "Cory")
+    today = datetime.date.today()
+    await coordinator.async_refresh()
+    cory = [
+        child for child in coordinator.data.children
+        if child.child_name == "Cory"
+    ][0]
+    await PresenceSchedulesDao(database).upsert_by_child(
+        cory.child_id, 1, today.isoformat(), ""
+    )
+    await coordinator.async_refresh()
+
+    assert _binary(hass, ada.id, "present_today").is_on is True
+    assert _binary(hass, cory.child_id, "present_today").is_on is False
+
+
+async def test_binary_sensor_identity_and_attributes(hass, make_entry) -> None:
+    entry = await _setup_entry(hass, make_entry)
+    coordinator = entry.runtime_data.coordinator
+    database = entry.runtime_data.database
+    ada, _bo = await _seed_two_children(database)
+    await coordinator.async_refresh()
+
+    all_done = _binary(hass, ada.id, "all_done")
+    present = _binary(hass, ada.id, "present_today")
+    assert all_done.unique_id == f"nestquest_child_{ada.id}_all_done"
+    assert present.unique_id == f"nestquest_child_{ada.id}_present_today"
+    assert all_done.name == "NestQuest Ada all done"
+    assert present.name == "NestQuest Ada present today"
+    for entity in (all_done, present):
+        assert entity.device_info.name == "NestQuest Ada"
+        assert entity.extra_state_attributes == {
+            "child_id": ada.id,
+            "child_name": "Ada",
+        }

tests/test_const.py

@@ -28,10 +28,10 @@ def test_logger() -> None:
 
 
 def test_platforms() -> None:
-    """Verify PLATFORMS is defined as the sensor platform list."""
+    """Verify PLATFORMS is defined as the sensor + binary_sensor list."""
     assert hasattr(const, "PLATFORMS")
     assert isinstance(const.PLATFORMS, list)
-    assert const.PLATFORMS == ["sensor"]
+    assert const.PLATFORMS == ["sensor", "binary_sensor"]
 
 
 def test_sqlite_db_filename() -> None:

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

@@ -0,0 +1,379 @@
+"""Tests for the NestQuest bus events (Feature 10, task 6)."""
+from __future__ import annotations
+
+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
+from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
+    DOMAIN,
+    EVENT_CHILD_DAY_COMPLETE,
+    EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_MISSED,
+    EVENT_QUEST_UNCOMPLETED,
+    SERVICE_COMPLETE_QUEST,
+    SERVICE_CREATE_QUEST_DEFINITION,
+    SERVICE_MANAGE_CHILD,
+    SERVICE_REGENERATE,
+    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
+
+ADMIN_CTX = {"user_id": "admin-1"}
+_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$")
+
+
+async def _setup_and_seed(hass, make_entry, *, windows=("morning",)):
+    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
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_MANAGE_CHILD,
+        {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
+    )
+    child = None
+    from custom_components.nestquest.children import list_children
+
+    child = (await list_children(database))[0]
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": __import__("datetime").date.today().isoformat()}
+        ),
+        [child.id],
+        list(windows),
+    )
+    import datetime
+
+    today = datetime.date.today().isoformat()
+    end = (datetime.date.today() + datetime.timedelta(days=3)).isoformat()
+    await materialize(database, today, end, today=datetime.date.today())
+    instances = await QuestInstancesDao(database).list_by_date_range(
+        child.id, today, today
+    )
+    return entry, child, instances
+
+
+async def test_quest_completed_fires_with_documented_payload(
+    hass, make_entry
+) -> None:
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    instance_id = instances[0].id
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
+    )
+    fired = hass.bus.fired(EVENT_QUEST_COMPLETED)
+    assert len(fired) == 1
+    payload = fired[0]
+    assert payload["child_id"] == child.id
+    assert payload["child_name"] == "Ada"
+    assert payload["instance_id"] == instance_id
+    assert payload["quest_title"] == "Brush teeth"
+    assert payload["window"] == "morning"
+    assert payload["was_on_time"] is not None
+    assert _TIMESTAMP.match(payload["occurred_at"]), payload["occurred_at"]
+
+
+async def test_recomplete_no_op_fires_nothing(hass, make_entry) -> None:
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    instance_id = instances[0].id
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
+    )
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
+    )
+    assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
+
+
+async def test_child_day_complete_fires_when_day_clears(
+    hass, make_entry
+) -> None:
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    for instance in instances:
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
+        )
+    day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
+    assert len(day_complete) == 1, "exactly one day-complete per cleared day"
+    payload = day_complete[0]
+    assert payload["child_id"] == child.id
+    assert payload["child_name"] == "Ada"
+    assert payload["quests_due"] == len(instances)
+    assert _TIMESTAMP.match(payload["occurred_at"])
+
+
+async def test_child_day_complete_does_not_fire_partway(
+    hass, make_entry
+) -> None:
+    entry, child, instances = await _setup_and_seed(
+        hass, make_entry, windows=("morning", "evening")
+    )
+    assert len(instances) == 2
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": instances[0].id,
+            "actor": "panel",
+            "actor_child_id": child.id,
+        },
+    )
+    assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": instances[1].id,
+            "actor": "panel",
+            "actor_child_id": child.id,
+        },
+    )
+    assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
+
+
+async def test_uncompleted_fires_on_reversal_only(hass, make_entry) -> None:
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    instance_id = instances[0].id
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
+    )
+    # Un-completing an open instance is a no-op and fires nothing.
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_UNCOMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "user"},
+        context=ADMIN_CTX,
+    )
+    fired = hass.bus.fired(EVENT_QUEST_UNCOMPLETED)
+    assert len(fired) == 1
+    payload = fired[0]
+    assert payload["instance_id"] == instance_id
+    assert payload["quest_title"] == "Brush teeth"
+    assert _TIMESTAMP.match(payload["occurred_at"])
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_UNCOMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "user"},
+        context=ADMIN_CTX,
+    )
+    assert len(hass.bus.fired(EVENT_QUEST_UNCOMPLETED)) == 1
+
+
+def test_missed_event_name_is_documented_contract() -> None:
+    """Feature 10 defines the missed event Feature 11's sweep fires."""
+    assert EVENT_QUEST_MISSED == "nestquest_quest_missed"
+
+
+async def test_appended_verdict_is_atomic_under_duplicate_calls(
+    hass, make_entry
+) -> None:
+    """Concurrent duplicate completes decide appended under the same
+    lock as the write: exactly one fires the transition event."""
+    import asyncio
+
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    instance_id = instances[0].id
+    calls = [
+        hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance_id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
+        )
+        for _ in range(2)
+    ]
+    await asyncio.gather(*calls)
+    assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
+
+
+async def test_day_complete_ignored_for_backdated_completion(
+    hass, make_entry
+) -> None:
+    """Completing an instance NOT due today never evaluates the
+    child's day-complete, so a cleared day is announced at most once
+    per day and only by that day's completions."""
+    import datetime
+
+    from custom_components.nestquest.completion import (
+        complete_instance as _complete,
+    )
+
+    entry, child, instances = await _setup_and_seed(hass, make_entry)
+    database = entry.runtime_data.database
+    # A future instance (tomorrow) completed after today's day is
+    # already clear.
+    today = datetime.date.today()
+    end = (today + datetime.timedelta(days=3)).isoformat()
+    all_instances = await QuestInstancesDao(database).list_by_date_range(
+        child.id, today.isoformat(), end
+    )
+    today_ids = {instance.id for instance in instances}
+    tomorrow_instance = next(
+        instance
+        for instance in all_instances
+        if instance.id not in today_ids
+    )
+    for instance in instances:
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
+        )
+    assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
+
+    await _complete(
+        database,
+        tomorrow_instance.id,
+        actor_source="user",
+        actor_user_id="admin-1",
+    )
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": tomorrow_instance.id,
+            "actor": "panel",
+            "actor_child_id": child.id,
+        },
+    )
+    assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1, (
+        "a back-dated/future completion must not re-announce today"
+    )
+    # The direct business-layer call fired nothing (only the service
+    # 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_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_sensors.py

@@ -104,13 +104,17 @@ async def test_every_active_child_gets_five_sensors_with_documented_identity(
         hass, make_entry
     )
     database = entry.runtime_data.database
-    assert hass.config_entries.forwarded_platforms == ["sensor"]
+    assert hass.config_entries.forwarded_platforms == [
+        "sensor",
+        "binary_sensor",
+    ]
     await _complete_first_instance(database, coordinator, ada)
     await coordinator.async_refresh()
 
     # Two quest children plus the zero-quest child: three children x
-    # five sensors, no duplicates.
-    assert len(hass.entities) == 15
+    # (five sensors + two binary sensors) + three household sensors,
+    # no duplicates.
+    assert len(hass.entities) == 24
     expected = {
         ada.id: ("Ada", 1, 1, 0, 100),
         bo.id: ("Bo", 1, 0, 1, 0),
@@ -324,22 +328,32 @@ async def test_repeated_refreshes_do_not_duplicate_entities(
     )
     await coordinator.async_refresh()
     await coordinator.async_refresh()
-    assert len(hass.entities) == 15
+    assert len(hass.entities) == 24
 
 
-async def test_setup_with_no_children_registers_no_entities(
+async def test_setup_with_no_children_registers_only_household(
     hass, make_entry
 ) -> None:
     entry = wire_entry_to_registry(make_entry(), hass.registry)
     assert await async_setup_entry(hass, entry) is True
-    assert hass.entities == {}
+    # No children means no per-child entities; the three household
+    # rollups still exist (an empty household is a valid 0).
+    assert set(hass.entities) == {
+        "nestquest_household_quests_due_today",
+        "nestquest_household_quests_completed_today",
+        "nestquest_cycle_day",
+    }
+    assert hass.entities["nestquest_household_quests_due_today"].native_value == 0
 
 
 async def test_unload_unloads_platforms(hass, make_entry) -> None:
     entry = wire_entry_to_registry(make_entry(), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     assert await async_unload_entry(hass, entry) is True
-    assert hass.config_entries.unloaded_platforms == ["sensor"]
+    assert hass.config_entries.unloaded_platforms == [
+        "sensor",
+        "binary_sensor",
+    ]
 
 
 async def test_rename_propagates_to_name_and_device_label(
@@ -446,3 +460,56 @@ async def test_next_quest_state_truncates_over_the_limit(
     state = next_entity.native_value
     assert len(state) <= 255
     assert next_entity.extra_state_attributes["title"] == full_title
+
+
+async def test_household_rollups_sum_the_active_children(
+    hass, make_entry
+) -> None:
+    """Household due/completed equal the sum of the per-child sensors;
+    the cycle-day sensor carries the coordinator's answer."""
+    import datetime as _dt
+
+    from custom_components.nestquest.dao_presence import (
+        PresenceSchedulesDao,
+    )
+
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+    await _complete_first_instance(database, coordinator, ada)
+    await coordinator.async_refresh()
+
+    household_due = hass.entities["nestquest_household_quests_due_today"]
+    household_completed = hass.entities[
+        "nestquest_household_quests_completed_today"
+    ]
+    cycle_day_entity = hass.entities["nestquest_cycle_day"]
+    per_child_due = [
+        _sensor(hass, child.id, "quests_due_today").native_value
+        for child in (ada, bo, cory)
+    ]
+    per_child_completed = [
+        _sensor(hass, child.id, "quests_completed_today").native_value
+        for child in (ada, bo, cory)
+    ]
+    assert household_due.native_value == sum(per_child_due)
+    assert household_completed.native_value == sum(per_child_completed)
+    assert household_due.name == "NestQuest household quests due today"
+    assert household_completed.name == (
+        "NestQuest household quests completed today"
+    )
+    assert cycle_day_entity.name == "NestQuest cycle day"
+    assert cycle_day_entity.native_value == 0, "no schedule anywhere"
+
+    # A schedule on the FIRST active child drives cycle day.
+    today = _dt.date.today()
+    await PresenceSchedulesDao(database).upsert_by_child(
+        ada.id, 2, (today - _dt.timedelta(days=3)).isoformat(),
+        "0,1,2,3,4,5,6|",
+    )
+    await coordinator.async_refresh()
+    assert cycle_day_entity.native_value == 4
+    assert cycle_day_entity.extra_state_attributes["today"] == (
+        today.isoformat()
+    )

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