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

@@ -20,11 +20,15 @@ from .const import (
     CONF_ADMIN_USER_IDS,
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
+    CONF_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
+    DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
     LOGGER,
+    PLATFORMS,
 )
+from .coordinator import NestQuestCoordinator
 from .db import NestQuestDatabase
 from .materialize import materialize as _materialize_run
 from .migrations import apply_migrations
@@ -238,6 +242,7 @@ class NestQuestRuntimeData:
     database: NestQuestDatabase
     remove_update_listener: Callable[[], Any]
     remove_time_change_listener: Callable[[], Any]
+    coordinator: Any = None
 
 
 async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
@@ -456,6 +461,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     database = await _async_open_database(hass, await async_get_db_path(hass))
     remove_update_listener: Callable[[], Any] | None = None
     remove_time_change_listener: Callable[[], Any] | None = None
+    coordinator: Any = None
     try:
         # Seed the admin allowlist on first setup so the owner is never
         # locked out: an empty database takes the persisted admin copy
@@ -493,6 +499,33 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             hass, database, _configured_horizon_days(dict(entry.options))
         )
         _register_services(hass)
+        # The shared Feature 10 coordinator: one refresh cycle every
+        # entity reads from (CONF_UPDATE_INTERVAL seconds, default
+        # five minutes).  Created AFTER services so a failure below
+        # unwinds them through the except path's listener handling.
+        coordinator = NestQuestCoordinator(
+            hass,
+            entry_id=entry.entry_id,
+            database=database,
+            update_interval_seconds=entry.options.get(
+                CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
+            ),
+        )
+        await coordinator.async_config_entry_first_refresh()
+        runtime_data = NestQuestRuntimeData(
+            entry_id=entry.entry_id,
+            options=dict(entry.options),
+            database=database,
+            remove_update_listener=remove_update_listener,
+            remove_time_change_listener=remove_time_change_listener,
+            coordinator=coordinator,
+        )
+        # runtime_data is assigned BEFORE the platforms forward so each
+        # platform's async_setup_entry reads its coordinator off the
+        # entry (the HA runtime-data contract); a forward failure below
+        # unwinds it through the except path.
+        entry.runtime_data = runtime_data
+        await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     except BaseException:
         # A failure anywhere after the database is open must not leak the
         # listeners already registered against that (now-closed) database:
@@ -511,18 +544,24 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
                     await result
             except BaseException:
                 pass
+        # A coordinator created before the failure holds entity
+        # listeners pointed at the database being closed below —
+        # shut it down FIRST so no scheduled refresh can ever run
+        # against a closed connection.
+        if coordinator is not None:
+            try:
+                await coordinator.async_shutdown()
+            except BaseException:
+                pass
+        # A platform-forward failure reached this unwind AFTER
+        # runtime_data was assigned: clear it so no later unload path
+        # closes the already-closed database through the stale record.
+        if getattr(entry, "runtime_data", None) is not None:
+            entry.runtime_data = None
         await database.close()
         raise
     assert remove_update_listener is not None
     assert remove_time_change_listener is not None
-    runtime_data = NestQuestRuntimeData(
-        entry_id=entry.entry_id,
-        options=dict(entry.options),
-        database=database,
-        remove_update_listener=remove_update_listener,
-        remove_time_change_listener=remove_time_change_listener,
-    )
-    entry.runtime_data = runtime_data
     hass.data[DOMAIN][entry.entry_id] = runtime_data
     return True
 
@@ -536,6 +575,20 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         hass.data.pop(DOMAIN, None)
     if runtime_data is None:
         runtime_data = getattr(entry, "runtime_data", None)
+    # Platforms unload FIRST (mirroring HA conventions): entity
+    # listeners must not outlive the database and coordinator the
+    # teardown below closes.
+    platforms_unloaded = await hass.config_entries.async_unload_platforms(
+        entry, PLATFORMS
+    )
+    if platforms_unloaded is False and runtime_data is not None:
+        # Home Assistant KEPT one or more platforms loaded — their
+        # entities still live and must keep their database.  Restore
+        # the runtime record popped above and report failure so HA
+        # retries the unload later instead of leaving live entities
+        # backed by a closed database.
+        hass.data.setdefault(DOMAIN, {})[entry.entry_id] = runtime_data
+        return False
     unload_error: BaseException | None = None
     if runtime_data is not None:
         remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
@@ -562,6 +615,16 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
                     if unload_error is None:
                         unload_error = err
         finally:
+            # 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()

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

@@ -128,6 +128,21 @@ def _derived_state(
     )
 
 
+def derive_state(
+    instance: QuestInstanceRecord,
+    latest: CompletionEventRecord | None,
+    today: datetime.date,
+) -> str:
+    """Public state derivation for the Feature 10 coordinator.
+
+    Thin wrapper over :func:`_derived_state`: ``open``, ``done``, or
+    ``missed`` from the latest event plus the due date (``missed`` is
+    derived, never an event_type).  ``today`` is the caller-resolved
+    HA-local date.
+    """
+    return _derived_state(instance, latest, today)
+
+
 def _was_on_time(
     instance: QuestInstanceRecord,
     now: datetime.datetime | None,
@@ -351,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,
@@ -360,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``
@@ -389,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,
@@ -403,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(
@@ -418,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(
@@ -444,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 "
@@ -461,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

@@ -7,7 +7,18 @@ DOMAIN = "nestquest"
 LOGGER_NAME = "custom_components.nestquest"
 LOGGER = logging.getLogger(LOGGER_NAME)
 
-PLATFORMS: list[str] = []
+# Platforms set up per config entry (Feature 10): the per-child day
+# 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).
@@ -55,6 +66,13 @@ DEFAULT_PANEL_IDLE_TIMEOUT = 300
 # list as a disaster-recovery copy so a lost database can be re-seeded.
 CONF_ADMIN_USER_IDS = "admin_user_ids"
 
+# Shared coordinator refresh interval (Feature 10), in SECONDS.  Every
+# entity reads one snapshot per refresh; completions push an immediate
+# refresh so this poll is only the fallback cadence.
+CONF_UPDATE_INTERVAL = "update_interval"
+DEFAULT_UPDATE_INTERVAL = 300
+MIN_UPDATE_INTERVAL = 30
+
 # Quest windows (D-008): a definition may declare several day windows;
 # each produces its own instance per day and groups the panel's Quest
 # Log into three columns.  The clock ranges are stored here so they stay

custom_components/nestquest/coordinator.py

@@ -0,0 +1,272 @@
+"""The shared NestQuest DataUpdateCoordinator (Feature 10).
+
+ONE refresh pass builds the whole entity snapshot: active children,
+today's instances with their derived states (open / done / missed via
+:func:`~.completion.derive_state`), each child's presence today, and
+per-child rollup counts.  Every Feature 10 entity reads
+``coordinator.data`` — no entity queries the database itself, so one
+refresh updates the whole board coherently.
+
+"Today" and "now" are HA-local (``hass.config.time_zone``), never the
+host clock and never UTC, matching the materialization and completion
+layers.
+
+Presence is resolved through the SAME engine the materializer uses
+(:class:`~.presence.PresenceEngine`): schedules and overrides are read
+with the public DAO methods and the engine's anchor-date arithmetic
+answers ``is this child at this house today`` (D-004 — never ISO week
+parity).
+
+``cycle_day`` reports where today falls in the custody cycle of the
+first active child (display order) that HAS a schedule — 1-based day
+``N`` of ``cycle_length_weeks * 7``, computed from that schedule's
+anchor date.  ``0`` means no active child has a schedule at all, so
+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
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .completion import derive_state
+from .const import (
+    CONF_UPDATE_INTERVAL,
+    DEFAULT_UPDATE_INTERVAL,
+    DOMAIN,
+    LOGGER,
+    MIN_UPDATE_INTERVAL,
+)
+from .dao_children import ChildrenDao
+from .dao_instances import (
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from .dao_rules import QuestDefinitionsDao
+from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
+from .db import NestQuestDatabase
+from .presence import (
+    PresenceEngine,
+    PresenceOverride,
+    PresenceSchedule,
+)
+
+
+@dataclass(frozen=True)
+class QuestInstanceView:
+    """One instance plus everything the entities render from it."""
+
+    instance_id: int
+    definition_id: int
+    child_id: int
+    title: str
+    icon: str | None
+    window: str
+    due_date: str
+    due_time: str | None
+    state: str  # open | done | missed
+    overdue: bool
+    completed_at: str | None
+    was_on_time: bool | None
+
+
+@dataclass(frozen=True)
+class ChildDaySnapshot:
+    """One active child's day: instances, presence, and rollup counts."""
+
+    child_id: int
+    child_name: str
+    instances: tuple[QuestInstanceView, ...]
+    present: bool
+    due_today: int
+    completed_today: int
+    remaining_today: int
+    completion_pct: int
+
+
+@dataclass(frozen=True)
+class NestQuestSnapshot:
+    """The coordinator's whole-entity data payload."""
+
+    today_iso: str
+    children: tuple[ChildDaySnapshot, ...]
+    cycle_day: int
+
+
+class NestQuestCoordinator(DataUpdateCoordinator):
+    """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,
+        hass: HomeAssistant,
+        *,
+        entry_id: str,
+        database: NestQuestDatabase,
+        update_interval_seconds: int | None = None,
+    ) -> None:
+        interval = update_interval_seconds
+        if (
+            not isinstance(interval, int)
+            or isinstance(interval, bool)
+            or interval < MIN_UPDATE_INTERVAL
+        ):
+            interval = DEFAULT_UPDATE_INTERVAL
+        super().__init__(
+            hass,
+            LOGGER,
+            name=DOMAIN,
+            update_interval=datetime.timedelta(seconds=interval),
+        )
+        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)
+        return datetime.datetime.now(time_zone)
+
+    async def _async_update_data(self) -> NestQuestSnapshot:
+        now = self._local_now()
+        today = now.date()
+        today_iso = today.isoformat()
+
+        children = await ChildrenDao(self.database).list_active()
+
+        schedules: dict[int, PresenceSchedule] = {}
+        overrides: dict[int, list[PresenceOverride]] = {}
+        for child in children:
+            schedule = await PresenceSchedulesDao(self.database).get_by_child(
+                child.id
+            )
+            if schedule is not None:
+                schedules[child.id] = PresenceSchedule.decode(
+                    child.id, schedule.anchor_date, schedule.pattern
+                )
+            todays = await PresenceOverridesDao(
+                self.database
+            ).list_by_child_and_range(child.id, today_iso, today_iso)
+            if todays:
+                # The engine validates MODEL objects, not DAO records;
+                # each record converts losslessly (the schema guarantees
+                # end >= start and a real is_present).
+                overrides[child.id] = [
+                    PresenceOverride(
+                        record.child_id,
+                        record.start_date,
+                        record.end_date,
+                        record.is_present,
+                        record.note,
+                    )
+                    for record in todays
+                ]
+        engine = PresenceEngine(schedules, overrides)
+
+        instances_dao = QuestInstancesDao(self.database)
+        events_dao = CompletionEventsDao(self.database)
+        definitions_dao = QuestDefinitionsDao(self.database)
+
+        snapshots: list[ChildDaySnapshot] = []
+        for child in children:
+            instances = await instances_dao.list_by_date_range(
+                child.id, today_iso, today_iso
+            )
+            views: list[QuestInstanceView] = []
+            for instance in instances:
+                latest = await events_dao.get_latest_for_instance(instance.id)
+                definition = await definitions_dao.get(instance.definition_id)
+                state = derive_state(instance, latest, today)
+                overdue = False
+                if state == "open" and instance.due_date == today_iso:
+                    if instance.due_time is not None:
+                        due_moment = datetime.datetime.strptime(
+                            instance.due_time, "%H:%M"
+                        ).time()
+                        overdue = now.time() > due_moment
+                views.append(
+                    QuestInstanceView(
+                        instance_id=instance.id,
+                        definition_id=instance.definition_id,
+                        child_id=instance.child_id,
+                        title=definition.title if definition else "Quest",
+                        icon=definition.icon if definition else None,
+                        window=instance.window,
+                        due_date=instance.due_date,
+                        due_time=instance.due_time,
+                        state=state,
+                        overdue=overdue,
+                        completed_at=(
+                            latest.occurred_at
+                            if latest is not None
+                            and latest.event_type == "completed"
+                            else None
+                        ),
+                        was_on_time=(
+                            latest.was_on_time
+                            if latest is not None
+                            and latest.event_type == "completed"
+                            else None
+                        ),
+                    )
+                )
+            due = len(views)
+            completed = sum(1 for view in views if view.state == "done")
+            remaining = due - completed
+            snapshots.append(
+                ChildDaySnapshot(
+                    child_id=child.id,
+                    child_name=child.display_name,
+                    instances=tuple(views),
+                    present=engine.is_present(child.id, today),
+                    due_today=due,
+                    completed_today=completed,
+                    remaining_today=remaining,
+                    completion_pct=(
+                        round(completed * 100 / due) if due else 100
+                    ),
+                )
+            )
+
+        return NestQuestSnapshot(
+            today_iso=today_iso,
+            children=tuple(snapshots),
+            cycle_day=_cycle_day(schedules, children, today),
+        )
+
+
+def _cycle_day(
+    schedules: dict[int, PresenceSchedule], children, today: datetime.date
+) -> int:
+    """Return today's 1-based day in the first scheduled child's cycle.
+
+    Anchor-date arithmetic ONLY (D-004): the day offset is computed from
+    the schedule's own anchor, never ISO week numbers or parity, so a
+    53-week year cannot silently invert the cycle.
+    """
+    for child in children:
+        schedule = schedules.get(child.id)
+        if schedule is None:
+            continue
+        days = (today - schedule.anchor_date).days
+        cycle_length = schedule.cycle_length_weeks * 7
+        return (days % cycle_length) + 1
+    return 0

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

@@ -14,10 +14,13 @@ from .const import (
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
     CONF_PANEL_IDLE_TIMEOUT,
+    CONF_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
     DEFAULT_PANEL_IDLE_TIMEOUT,
+    DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
+    MIN_UPDATE_INTERVAL,
 )
 from .admin_allowlist import list_admin_ids, set_admin_ids
 
@@ -56,6 +59,18 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
     ):
         errors[CONF_PANEL_IDLE_TIMEOUT] = "invalid"
 
+    # The coordinator refresh interval (Feature 10) is OPTIONAL in raw
+    # submits: a missing key keeps the current value, so pre-Feature-10
+    # form payloads (and API callers) are unaffected.  Schema-validated
+    # submits always carry it.
+    update_interval = user_input.get(CONF_UPDATE_INTERVAL)
+    if update_interval is not None and (
+        isinstance(update_interval, bool)
+        or not isinstance(update_interval, int)
+        or update_interval < MIN_UPDATE_INTERVAL
+    ):
+        errors[CONF_UPDATE_INTERVAL] = "invalid"
+
     return errors
 
 
@@ -82,6 +97,9 @@ def _build_schema(
         vol.Required(
             CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
         ): int,
+        vol.Required(
+            CONF_UPDATE_INTERVAL, default=current[CONF_UPDATE_INTERVAL]
+        ): int,
     }
     if admin_choices is not None:
         # HA's supported multi-select validator: the frontend can
@@ -210,6 +228,10 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             CONF_HORIZON_DAYS: user_input[CONF_HORIZON_DAYS],
             CONF_DAY_ROLLOVER_TIME: user_input[CONF_DAY_ROLLOVER_TIME],
             CONF_PANEL_IDLE_TIMEOUT: user_input[CONF_PANEL_IDLE_TIMEOUT],
+            CONF_UPDATE_INTERVAL: user_input.get(
+                CONF_UPDATE_INTERVAL,
+                self._current_values()[CONF_UPDATE_INTERVAL],
+            ),
         }
         if admin_ids_present:
             data[CONF_ADMIN_USER_IDS] = admin_ids
@@ -231,6 +253,7 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             (CONF_HORIZON_DAYS, DEFAULT_HORIZON_DAYS),
             (CONF_DAY_ROLLOVER_TIME, DEFAULT_DAY_ROLLOVER_TIME),
             (CONF_PANEL_IDLE_TIMEOUT, DEFAULT_PANEL_IDLE_TIMEOUT),
+            (CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL),
         ):
             if key in options:
                 values[key] = options[key]

custom_components/nestquest/sensor.py

@@ -0,0 +1,535 @@
+"""Per-child daily count and percentage sensors (Feature 10, task 2).
+
+Naming contract (design/ENTITIES-AND-SERVICES.md §1): the design table
+documents the entity_id shape ``sensor.nestquest_<child>_quests_due_today``.
+Home Assistant derives entity_id from the entity NAME and exact
+entity_id strings are not directly controllable, so names are spelled
+to slug into the documented shape — ``NestQuest Ada quests due today``
+slugs to ``sensor.nestquest_ada_quests_due_today`` — while the
+registry identity (``unique_id``) IS exact: built from the child's
+database id, never the display name, so a rename never orphans an
+entity's history.  The design table is the visual contract; this
+module spells names so HA's default slugging produces it.
+
+Every sensor reads ONLY ``coordinator.data`` (one
+:class:`~.coordinator.ChildDaySnapshot` per active child); no entity
+queries the database, so one refresh updates the whole board
+coherently.  The due-today sensor additionally carries the child's
+day payload as attributes: ``instances`` is the panel shape of
+ENTITIES-AND-SERVICES.md §1 (state ``open``/``completed``) OMITTING
+missed instances, and ``admin_instances`` includes them (D-009).
+"""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable
+from typing import Any
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorStateClass,
+)
+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,
+    QuestInstanceView,
+)
+
+#: Home Assistant's state machine refuses states longer than 255
+#: characters (its hard recorder/state limit).  These sensors' states
+#: are numeric, but the guard is applied uniformly: a payload-bearing
+#: sensor can never crash the state write, and later string-state
+#: sensors reuse the same helper.
+MAX_STATE_LENGTH = 255
+
+
+def _state_within_limit(state: Any) -> Any:
+    """Return ``state`` unchanged, refusing strings over HA's limit.
+
+    Numeric states pass through untouched; a string state longer than
+    :data:`MAX_STATE_LENGTH` fails fast with ValueError at the property
+    read — before HA's state machine rejects the write.
+    """
+    if isinstance(state, str) and len(state) > MAX_STATE_LENGTH:
+        raise ValueError(
+            f"state exceeds Home Assistant's {MAX_STATE_LENGTH}-character "
+            f"limit ({len(state)} characters)"
+        )
+    return state
+
+
+def _instance_payload(
+    instances: Iterable[QuestInstanceView], *, include_missed: bool
+) -> list[dict[str, Any]]:
+    """Shape instance dicts per ENTITIES-AND-SERVICES.md §1.
+
+    ``state`` maps the coordinator's derived ``done`` to the panel's
+    ``completed`` spelling; ``missed`` instances are omitted unless
+    ``include_missed`` (the admin payload, D-009, keeps them with
+    their ``missed`` state).
+    """
+    payload: list[dict[str, Any]] = []
+    for view in instances:
+        if view.state == "missed" and not include_missed:
+            continue
+        payload.append(
+            {
+                "id": view.instance_id,
+                "definition_id": view.definition_id,
+                "child_id": view.child_id,
+                "title": view.title,
+                "icon": view.icon,
+                "window": view.window,
+                "due_time": view.due_time,
+                "state": (
+                    "completed" if view.state == "done" else view.state
+                ),
+                "overdue": view.overdue,
+                "completed_at": view.completed_at,
+                "on_time": view.was_on_time,
+            }
+        )
+    return payload
+
+
+#: State reported by the next-quest sensor when the child owes nothing
+#: else today.  A documented sentinel, never an empty string (an empty
+#: state reads as "sensor went unavailable" in HA).
+NEXT_QUEST_NONE_STATE = "none"
+
+
+def _next_open_quest(
+    child: ChildDaySnapshot,
+) -> QuestInstanceView | None:
+    """Return the child's earliest open quest from today's snapshot.
+
+    The snapshot carries TODAY's instances, so "next" is the earliest
+    open instance by (due_date, due_time), with ``due_time`` ordering
+    ahead of ``None`` (a window with no due time comes first only
+    within the same date — the materializer snapshots each window's
+    due_time, so a missing one means the window is unscheduled).
+    Missed and completed instances never qualify.
+    """
+    open_views = [
+        view
+        for view in child.instances
+        if view.state == "open"
+    ]
+    if not open_views:
+        return None
+    return min(
+        open_views,
+        key=lambda view: (
+            view.due_date,
+            view.due_time if view.due_time is not None else "",
+        ),
+    )
+
+
+class _NestQuestChildDaySensor(CoordinatorEntity, SensorEntity):
+    """Base for the per-child day sensors.
+
+    Registry identity is built from the child's database id and never
+    changes; the DISPLAYED name and device label are derived from the
+    child's CURRENT snapshot on every read, so a rename through
+    ``manage_child`` propagates on the next refresh without orphaning
+    the entity's history.  A child absent from the latest snapshot (a
+    deactivation; removal is a later task's scope) yields ``None``
+    states rather than stale counts.
+    """
+
+    #: Per-subclass label appended to the name (e.g. ``quests due
+    #: today``), set as a class attribute so ``name`` composes from the
+    #: current snapshot.
+    _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
+        self._attr_unique_id = None  # set by the subclass
+
+    @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",
+        )
+
+
+class NestQuestQuestsDueTodaySensor(_NestQuestChildDaySensor):
+    """How many quests the child has due today, with the day payload."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _name_suffix = "quests due today"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = (
+            f"{DOMAIN}_child_{child.child_id}_quests_due_today"
+        )
+
+    @property
+    def native_value(self) -> int | None:
+        child = self._child
+        if child is None:
+            return None
+        return _state_within_limit(child.due_today)
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        child = self._child
+        if child is None:
+            return None
+        return {
+            "instances": _instance_payload(
+                child.instances, include_missed=False
+            ),
+            "admin_instances": _instance_payload(
+                child.instances, include_missed=True
+            ),
+            "child_id": child.child_id,
+            "child_name": child.child_name,
+            "present": child.present,
+        }
+
+
+class NestQuestQuestsCompletedTodaySensor(_NestQuestChildDaySensor):
+    """How many of the child's quests were completed today."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _name_suffix = "quests completed today"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = (
+            f"{DOMAIN}_child_{child.child_id}_quests_completed_today"
+        )
+
+    @property
+    def native_value(self) -> int | None:
+        child = self._child
+        if child is None:
+            return None
+        return _state_within_limit(child.completed_today)
+
+
+class NestQuestQuestsRemainingTodaySensor(_NestQuestChildDaySensor):
+    """How many of the child's quests are still owed today."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _name_suffix = "quests remaining today"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = (
+            f"{DOMAIN}_child_{child.child_id}_quests_remaining_today"
+        )
+
+    @property
+    def native_value(self) -> int | None:
+        child = self._child
+        if child is None:
+            return None
+        return _state_within_limit(child.remaining_today)
+
+
+class NestQuestCompletionPctTodaySensor(_NestQuestChildDaySensor):
+    """The child's completion percentage for today (100 on a zero-quest day)."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_device_class = SensorDeviceClass.PERCENTAGE
+    _attr_native_unit_of_measurement = "%"
+    _name_suffix = "completion pct today"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = (
+            f"{DOMAIN}_child_{child.child_id}_completion_pct_today"
+        )
+
+    @property
+    def native_value(self) -> int | None:
+        child = self._child
+        if child is None:
+            return None
+        return _state_within_limit(child.completion_pct)
+
+
+class NestQuestNextQuestSensor(_NestQuestChildDaySensor):
+    """The child's next open quest today (title in the state).
+
+    State is the quest TITLE, truncated to HA's 255-character state
+    limit (attributes always carry the full title); when the child
+    owes nothing else today the state is the documented sentinel
+    ``none`` (NEXT_QUEST_NONE_STATE), never an empty string.
+    """
+
+    _attr_state_class = None  # a label, not a measurement
+    _name_suffix = "next quest"
+
+    def __init__(
+        self,
+        coordinator: NestQuestCoordinator,
+        child: ChildDaySnapshot,
+    ) -> None:
+        super().__init__(coordinator, child)
+        self._attr_unique_id = (
+            f"{DOMAIN}_child_{child.child_id}_next_quest"
+        )
+
+    @property
+    def native_value(self) -> str | None:
+        child = self._child
+        if child is None:
+            return None
+        quest = _next_open_quest(child)
+        if quest is None:
+            return NEXT_QUEST_NONE_STATE
+        return quest.title[:MAX_STATE_LENGTH]
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any] | None:
+        child = self._child
+        if child is None:
+            return None
+        quest = _next_open_quest(child)
+        if quest is None:
+            return {"child_id": child.child_id, "child_name": child.child_name}
+        return {
+            "instance_id": quest.instance_id,
+            "definition_id": quest.definition_id,
+            "child_id": quest.child_id,
+            "child_name": child.child_name,
+            "title": quest.title,
+            "icon": quest.icon,
+            "window": quest.window,
+            "due_date": quest.due_date,
+            "due_time": quest.due_time,
+            "overdue": quest.overdue,
+        }
+
+
+def _child_day_sensors(
+    coordinator: NestQuestCoordinator, child: ChildDaySnapshot
+) -> tuple[_NestQuestChildDaySensor, ...]:
+    """Build the per-child day sensors for one active child."""
+    return (
+        NestQuestQuestsDueTodaySensor(coordinator, child),
+        NestQuestQuestsCompletedTodaySensor(coordinator, child),
+        NestQuestQuestsRemainingTodaySensor(coordinator, child),
+        NestQuestCompletionPctTodaySensor(coordinator, child),
+        NestQuestNextQuestSensor(coordinator, child),
+    )
+
+
+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,
+    async_add_entities: Callable[..., None],
+) -> None:
+    """Create the five day 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.  Deactivated children are NOT
+    removed here — their sensors hold ``None`` states until the
+    entity-removal task lands.
+    """
+    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_day_sensors(coordinator, child)
+            ]
+        )
+
+    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
+        if data is not None:
+            _add_new_children(data.children)
+
+    coordinator.async_add_listener(_handle_coordinator_update)

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)
 

tests/conftest.py

@@ -27,6 +27,7 @@ import inspect
 import sys
 import tempfile
 import weakref
+from dataclasses import dataclass
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
@@ -44,7 +45,12 @@ _HA_MODULES = (
     "homeassistant.exceptions",
     "homeassistant.helpers",
     "homeassistant.helpers.config_validation",
+    "homeassistant.helpers.entity",
     "homeassistant.helpers.event",
+    "homeassistant.helpers.update_coordinator",
+    "homeassistant.components",
+    "homeassistant.components.sensor",
+    "homeassistant.components.binary_sensor",
 )
 
 def _find_spec(name: str):
@@ -84,7 +90,12 @@ for _parent, _child in (
     ("homeassistant", "exceptions"),
     ("homeassistant", "helpers"),
     ("homeassistant.helpers", "config_validation"),
+    ("homeassistant.helpers", "entity"),
     ("homeassistant.helpers", "event"),
+    ("homeassistant.helpers", "update_coordinator"),
+    ("homeassistant", "components"),
+    ("homeassistant.components", "sensor"),
+    ("homeassistant.components", "binary_sensor"),
 ):
     setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
 
@@ -197,6 +208,227 @@ _exceptions_mock.HomeAssistantError = HomeAssistantError
 _exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
 _exceptions_mock.Unauthorized = Unauthorized
 
+
+@dataclass
+class DeviceInfo:
+    """Stand-in for homeassistant.helpers.entity.DeviceInfo.
+
+    The kwargs-only shape the integration builds per-child devices
+    with: ``identifiers``, ``name``, ``manufacturer``, ``model``.
+    """
+
+    identifiers: object = None
+    name: str | None = None
+    manufacturer: str | None = None
+    model: str | None = None
+
+
+class Entity:
+    """Stand-in for homeassistant.helpers.entity.Entity.
+
+    Mirrors the ``_attr_*`` attribute contract through the ``name``,
+    ``unique_id``, ``device_info`` and ``extra_state_attributes``
+    properties, and ``async_write_ha_state`` registering the entity
+    into the hass entity registry (``hass.entities``, a dict keyed by
+    ``unique_id`` storing the live object) — the harness's stand-in
+    for HA's state machine, so tests enumerate entities and read
+    their state directly off the registered object.
+    """
+
+    hass = None
+    _attr_name: str | None = None
+    _attr_unique_id: str | None = None
+    _attr_device_info: DeviceInfo | None = None
+    _attr_extra_state_attributes: dict | None = None
+
+    @property
+    def name(self) -> str | None:
+        return self._attr_name
+
+    @property
+    def unique_id(self) -> str | None:
+        return self._attr_unique_id
+
+    @property
+    def device_info(self) -> DeviceInfo | None:
+        return self._attr_device_info
+
+    @property
+    def extra_state_attributes(self) -> dict | None:
+        return self._attr_extra_state_attributes
+
+    def async_write_ha_state(self) -> None:
+        """Register the entity under its unique_id (the mock state write)."""
+        if self.hass is not None and self.unique_id is not None:
+            self.hass.entities[self.unique_id] = self
+
+
+class CoordinatorEntity(Entity):
+    """Stand-in for homeassistant.helpers.update_coordinator.CoordinatorEntity.
+
+    Real HA subscribes the entity to coordinator notifications in the
+    platform-scheduler-driven ``async_added_to_hass`` lifecycle hook;
+    the stand-in subscribes eagerly in ``__init__`` so the synchronous
+    ``AddEntitiesCallback`` surface stays synchronous (the mock
+    harness has no lifecycle scheduler).  ``_handle_coordinator_update``
+    mirrors HA: one state write per coordinator notification.
+    """
+
+    def __init__(self, coordinator) -> None:
+        self.coordinator = coordinator
+        self.coordinator.async_add_listener(self._handle_coordinator_update)
+
+    def _handle_coordinator_update(self) -> None:
+        self.async_write_ha_state()
+
+
+class SensorStateClass:
+    """Stand-in namespace for homeassistant.components.sensor.SensorStateClass.
+
+    Only the members the integration uses (plus TOTAL for the
+    long-history sensors later tasks add).
+    """
+
+    MEASUREMENT = "measurement"
+    TOTAL = "total"
+
+
+class SensorDeviceClass:
+    """Stand-in namespace for homeassistant.components.sensor.SensorDeviceClass.
+
+    Only the members the integration uses.
+    """
+
+    PERCENTAGE = "percentage"
+
+
+class SensorEntity(Entity):
+    """Stand-in for homeassistant.components.sensor.SensorEntity.
+
+    The ``_attr_*`` passthrough for ``native_value``,
+    ``native_unit_of_measurement``, ``state_class``, ``device_class``
+    and ``suggested_display_precision``.
+    """
+
+    _attr_native_value: object = None
+    _attr_native_unit_of_measurement: str | None = None
+    _attr_state_class: str | None = None
+    _attr_device_class: str | None = None
+    _attr_suggested_display_precision: int | None = None
+
+    @property
+    def native_value(self) -> object:
+        return self._attr_native_value
+
+    @property
+    def native_unit_of_measurement(self) -> str | None:
+        return self._attr_native_unit_of_measurement
+
+    @property
+    def state_class(self) -> str | None:
+        return self._attr_state_class
+
+    @property
+    def device_class(self) -> str | None:
+        return self._attr_device_class
+
+    @property
+    def suggested_display_precision(self) -> int | None:
+        return self._attr_suggested_display_precision
+
+
+class BinarySensorEntity(Entity):
+    """Stand-in for homeassistant.components.binary_sensor.BinarySensorEntity.
+
+    Wired now so PLATFORMS can grow (task 4's all-done / present
+    sensors) without more harness surgery; only the surface the
+    integration uses.
+    """
+
+    _attr_is_on: bool | None = None
+    _attr_device_class: str | None = None
+
+    @property
+    def is_on(self) -> bool | None:
+        return self._attr_is_on
+
+    @property
+    def device_class(self) -> str | None:
+        return self._attr_device_class
+
+
+class DataUpdateCoordinator:
+    """Functional stand-in for HA's DataUpdateCoordinator.
+
+    Mirrors the surface NestQuest's coordinator subclass uses:
+    ``__init__(hass, logger, name=..., update_interval=...)``,
+    ``async_add_listener``/``async_update_listeners`` for entity
+    subscription, ``async_refresh``/``async_config_entry_first_refresh``
+    driving the subclass's ``_async_update_data``, ``last_update_success``,
+    and ``data``.  Interval scheduling is NOT simulated (no real clock
+    ticks in tests) — tests drive refreshes directly.
+    """
+
+    def __init__(self, hass, logger, *, name=None, update_interval=None):
+        self.hass = hass
+        self.logger = logger
+        self.name = name
+        self.update_interval = update_interval
+        self.data = None
+        self.last_update_success = False
+        self._listeners = []
+
+    def async_add_listener(self, update_callback, context=None):
+        """Subscribe; returns an unsubscribe callable like HA's."""
+        entry = (update_callback, context)
+        self._listeners.append(entry)
+
+        def _remove():
+            if entry in self._listeners:
+                self._listeners.remove(entry)
+
+        return _remove
+
+    def async_update_listeners(self):
+        """Notify every listener (entities schedule their writes)."""
+        for callback, _context in list(self._listeners):
+            callback()
+
+    async def async_refresh(self):
+        """Run one update pass and notify listeners."""
+        try:
+            self.data = await self._async_update_data()
+            self.last_update_success = True
+        except BaseException:
+            self.last_update_success = False
+            raise
+        self.async_update_listeners()
+
+    async def async_config_entry_first_refresh(self):
+        """First refresh at config-entry setup (alias of refresh)."""
+        await self.async_refresh()
+
+    async def async_shutdown(self):
+        """Release listeners on teardown."""
+        self._listeners.clear()
+
+
+_update_coordinator_mock = _ha_mock("homeassistant.helpers.update_coordinator")
+_update_coordinator_mock.DataUpdateCoordinator = DataUpdateCoordinator
+_update_coordinator_mock.CoordinatorEntity = CoordinatorEntity
+
+_entity_mock = _ha_mock("homeassistant.helpers.entity")
+_entity_mock.DeviceInfo = DeviceInfo
+_entity_mock.Entity = Entity
+
+_sensor_mock = _ha_mock("homeassistant.components.sensor")
+_sensor_mock.SensorEntity = SensorEntity
+_sensor_mock.SensorDeviceClass = SensorDeviceClass
+_sensor_mock.SensorStateClass = SensorStateClass
+
+_binary_sensor_mock = _ha_mock("homeassistant.components.binary_sensor")
+_binary_sensor_mock.BinarySensorEntity = BinarySensorEntity
+
 import voluptuous as vol
 
 
@@ -317,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.
 
@@ -439,10 +696,85 @@ 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"
 
+    # 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.  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 _make_add_entities(entry):
+        entry_entities = hass.entities_by_entry.setdefault(entry.entry_id, {})
+
+        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 a
+        per-entry AddEntitiesCallback.  A stub entry without a domain
+        defaults to this harness's single integration (NestQuest).
+        """
+        for platform in list(platforms):
+            domain = getattr(entry, "domain", "nestquest")
+            module = importlib.import_module(
+                f"custom_components.{domain}.{platform}"
+            )
+            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), 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")
+            module = importlib.import_module(
+                f"custom_components.{domain}.{platform}"
+            )
+            unload = getattr(module, "async_unload_entry", None)
+            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] = []
+    hass.config_entries.async_forward_entry_setups = _async_forward_entry_setups
+    hass.config_entries.async_unload_platforms = _async_unload_platforms
+
     async def _async_reload(entry_id: str) -> None:
         registry.reloaded.append(entry_id)
 

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 an empty list."""
+    """Verify PLATFORMS is defined as the sensor + binary_sensor list."""
     assert hasattr(const, "PLATFORMS")
     assert isinstance(const.PLATFORMS, list)
-    assert const.PLATFORMS == []
+    assert const.PLATFORMS == ["sensor", "binary_sensor"]
 
 
 def test_sqlite_db_filename() -> None:

tests/test_coordinator.py

@@ -0,0 +1,281 @@
+"""Tests for the shared NestQuest coordinator (Feature 10, task 1)."""
+from __future__ import annotations
+
+import datetime
+
+import pytest
+
+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.completion import complete_instance
+from custom_components.nestquest.const import (
+    CONF_UPDATE_INTERVAL,
+    DEFAULT_UPDATE_INTERVAL,
+    DEFAULT_HORIZON_DAYS,
+    DOMAIN,
+)
+from custom_components.nestquest.coordinator import NestQuestCoordinator
+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
+
+ADMIN_CTX = {"user_id": "admin-1"}
+
+
+def _today_iso() -> str:
+    return datetime.date.today().isoformat()
+
+
+async def _seed_day(database, *, completed: bool = False) -> None:
+    """Two active children + one inactive, one daily definition each."""
+    await create_child(database, "Ada")
+    await create_child(database, "Bo")
+    inactive = await create_child(database, "Ghost")
+    await set_child_active(database, inactive.id, False)
+    today = _today_iso()
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict({"rule_type": "daily", "start_date": today}),
+        [child.id for child in await list_children(database, active_only=True)],
+        ["morning"],
+    )
+    end = (
+        datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    await materialize(database, today, end, today=datetime.date.today())
+
+
+async def test_coordinator_snapshot_counts_and_states(hass, make_entry) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    assert coordinator is not None
+    database = entry.runtime_data.database
+    await _seed_day(database)
+
+    snapshot = await coordinator._async_update_data()
+    assert snapshot.today_iso == _today_iso()
+    assert len(snapshot.children) == 2, "inactive child must be absent"
+    by_name = {child.child_name: child for child in snapshot.children}
+    ada = by_name["Ada"]
+    assert ada.due_today == 1
+    assert ada.completed_today == 0
+    assert ada.remaining_today == 1
+    assert ada.completion_pct == 0
+    instance = ada.instances[0]
+    assert instance.state == "open"
+    assert instance.overdue is False
+    assert instance.title == "Brush teeth"
+    assert instance.window == "morning"
+    assert ada.present is True, "no schedule means present every day"
+    assert snapshot.cycle_day == 0, "no schedule anywhere"
+
+
+async def test_coordinator_reflects_completion_and_uncompletion(
+    hass, make_entry
+) -> None:
+    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
+    await _seed_day(database)
+    ada = (
+        await coordinator._async_update_data()
+    ).children[0]
+    instance_id = ada.instances[0].instance_id
+
+    await complete_instance(
+        database,
+        instance_id,
+        actor_source="panel",
+        actor_child_id=ada.child_id,
+    )
+    snapshot = await coordinator._async_update_data()
+    child = snapshot.children[0]
+    assert child.completed_today == 1
+    assert child.remaining_today == 0
+    assert child.completion_pct == 100
+    view = child.instances[0]
+    assert view.state == "done"
+    assert view.completed_at is not None
+    assert view.was_on_time is not None
+
+    from custom_components.nestquest.completion import uncomplete_instance
+
+    await uncomplete_instance(
+        database, instance_id, actor_source="user", actor_user_id="admin-1"
+    )
+    snapshot = await coordinator._async_update_data()
+    child = snapshot.children[0]
+    assert child.completed_today == 0
+    assert child.remaining_today == 1
+    assert child.instances[0].state == "open"
+
+
+async def test_coordinator_zero_quests_reports_full_percentage(
+    hass, make_entry
+) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    await create_child(entry.runtime_data.database, "Ada")
+    snapshot = await coordinator._async_update_data()
+    child = snapshot.children[0]
+    assert child.due_today == 0
+    assert child.completion_pct == 100, "zero quests is a perfect day"
+    assert child.instances == ()
+
+
+async def test_coordinator_cycle_day_from_first_scheduled_child(
+    hass, make_entry
+) -> None:
+    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
+    await create_child(database, "Ada")
+    ada = (await list_children(database))[0]
+    today = datetime.date.today()
+    # A 2-week cycle anchored 3 days ago: today is day 4.
+    anchor = today - datetime.timedelta(days=3)
+    await PresenceSchedulesDao(database).upsert_by_child(
+        ada.id, 2, anchor.isoformat(), "0,1,2,3,4,5,6|"
+    )
+    snapshot = await coordinator._async_update_data()
+    assert snapshot.cycle_day == 4
+
+
+async def test_coordinator_presence_resolves_schedule_and_override(
+    hass, make_entry
+) -> None:
+    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
+    await create_child(database, "Ada")
+    ada = (await list_children(database))[0]
+    today = datetime.date.today()
+    # Scheduled absent today (empty weekday set for this cycle week).
+    await PresenceSchedulesDao(database).upsert_by_child(
+        ada.id, 1, today.isoformat(), ""
+    )
+    snapshot = await coordinator._async_update_data()
+    assert snapshot.children[0].present is False
+
+    # An override beats the pattern.
+    from custom_components.nestquest.dao_presence import (
+        PresenceOverridesDao,
+    )
+
+    await PresenceOverridesDao(database).create(
+        ada.id, today.isoformat(), today.isoformat(), True
+    )
+    snapshot = await coordinator._async_update_data()
+    assert snapshot.children[0].present is True
+
+
+async def test_coordinator_derives_missed_for_past_due_open(
+    hass, make_entry
+) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    snapshot = await coordinator._async_update_data()
+    # No children: empty but valid snapshot.
+    assert snapshot.children == ()
+
+
+async def test_setup_creates_coordinator_with_configured_interval(
+    hass, make_entry
+) -> None:
+    entry = wire_entry_to_registry(
+        make_entry(options={CONF_UPDATE_INTERVAL: 60}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    assert isinstance(coordinator, NestQuestCoordinator)
+    assert coordinator.update_interval == datetime.timedelta(seconds=60)
+    assert coordinator.data is not None, "first refresh ran at setup"
+
+
+async def test_setup_defaults_to_five_minute_interval(
+    hass, make_entry
+) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    assert coordinator.update_interval == datetime.timedelta(
+        seconds=DEFAULT_UPDATE_INTERVAL
+    )
+
+
+async def test_unload_shuts_coordinator_down(hass, make_entry) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    torn_down = []
+
+    async def _shutdown():
+        torn_down.append(coordinator)
+
+    coordinator.async_shutdown = _shutdown
+    assert await async_unload_entry(hass, entry) is True
+    assert torn_down == [coordinator]
+
+
+async def test_coordinator_notifies_listeners_on_refresh(
+    hass, make_entry
+) -> None:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    coordinator = entry.runtime_data.coordinator
+    called = []
+    coordinator.async_add_listener(lambda: called.append("entity"))
+    await coordinator.async_refresh()
+    assert called == ["entity"]
+    assert coordinator.last_update_success is True
+
+
+async def test_failed_first_refresh_shuts_coordinator_down(
+    hass, make_entry, monkeypatch
+) -> None:
+    """A setup failure after coordinator creation must not leak its
+    listeners against the database the except path closes."""
+    from custom_components.nestquest import async_setup_entry
+    import custom_components.nestquest.coordinator as coordinator_module
+
+    shutdown_calls = []
+
+    async def _record_shutdown(self):
+        shutdown_calls.append(self.entry_id)
+
+    monkeypatch.setattr(
+        coordinator_module.NestQuestCoordinator,
+        "async_shutdown",
+        _record_shutdown,
+    )
+
+    async def _boom(self):
+        raise RuntimeError("refresh failed")
+
+    monkeypatch.setattr(
+        coordinator_module.NestQuestCoordinator,
+        "_async_update_data",
+        _boom,
+    )
+
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    with pytest.raises(RuntimeError, match="refresh failed"):
+        await async_setup_entry(hass, entry)
+    assert shutdown_calls == [entry.entry_id]
+    assert getattr(entry, "runtime_data", None) is 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_harness.py

@@ -31,6 +31,23 @@ def test_mock_only_harness_contract(monkeypatch):
     assert sys.modules["homeassistant.core"].callback is conftest._callback_decorator
     assert sys.modules["homeassistant.data_entry_flow"].RESULT_TYPE_FORM == "form"
 
+    # Entity-platform stand-ins are wired with the shaped classes, and
+    # parent-package attributes resolve to the mocked submodules.
+    entity_module = sys.modules["homeassistant.helpers.entity"]
+    assert entity_module.DeviceInfo is conftest.DeviceInfo
+    assert entity_module.Entity is conftest.Entity
+    sensor_module = sys.modules["homeassistant.components.sensor"]
+    assert sensor_module.SensorEntity is conftest.SensorEntity
+    assert sensor_module.SensorStateClass.MEASUREMENT == "measurement"
+    assert sensor_module.SensorDeviceClass.PERCENTAGE == "percentage"
+    assert sys.modules[
+        "homeassistant.components.binary_sensor"
+    ].BinarySensorEntity is conftest.BinarySensorEntity
+    assert sys.modules[
+        "homeassistant.helpers.update_coordinator"
+    ].CoordinatorEntity is conftest.CoordinatorEntity
+    assert sys.modules["homeassistant"].components.sensor is sensor_module
+
     # No homeassistant distribution is installed in this suite's own venv.
     with pytest.raises(importlib.metadata.PackageNotFoundError):
         importlib.metadata.version("homeassistant")

tests/test_options_flow.py

@@ -18,6 +18,8 @@ from custom_components.nestquest.const import (
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
     CONF_PANEL_IDLE_TIMEOUT,
+    CONF_UPDATE_INTERVAL,
+    DEFAULT_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
     DEFAULT_PANEL_IDLE_TIMEOUT,
@@ -129,7 +131,10 @@ def test_options_flow_submits_through_2024_6_surface() -> None:
     result = _run(handler.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
     assert result["title"] == "NestQuest"
-    assert result["data"] == VALID_INPUT
+    assert result["data"] == {
+        **VALID_INPUT,
+        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
+    }
 
 
 def test_options_flow_shows_form_initially() -> None:
@@ -208,6 +213,7 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
             CONF_HORIZON_DAYS: 21,
             CONF_DAY_ROLLOVER_TIME: "05:45",
             CONF_PANEL_IDLE_TIMEOUT: 120,
+            CONF_UPDATE_INTERVAL: 90,
         }
     )
     result = _run(_make_flow(entry).async_step_init(None))
@@ -215,6 +221,7 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
         CONF_HORIZON_DAYS: 21,
         CONF_DAY_ROLLOVER_TIME: "05:45",
         CONF_PANEL_IDLE_TIMEOUT: 120,
+        CONF_UPDATE_INTERVAL: 90,
         CONF_ADMIN_USER_IDS: [],
     }
 
@@ -282,8 +289,8 @@ def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
     assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
 
 
-def test_options_flow_creates_entry_with_three_keys() -> None:
-    """A valid submit creates an entry with exactly the three validated keys."""
+def test_options_flow_creates_entry_with_four_keys() -> None:
+    """A valid submit creates an entry with exactly the four validated keys."""
     flow = _make_flow(_make_entry())
     result = _run(flow.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
@@ -292,8 +299,12 @@ def test_options_flow_creates_entry_with_three_keys() -> None:
         CONF_HORIZON_DAYS,
         CONF_DAY_ROLLOVER_TIME,
         CONF_PANEL_IDLE_TIMEOUT,
+        CONF_UPDATE_INTERVAL,
+    }
+    assert result["data"] == {
+        **VALID_INPUT,
+        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
     }
-    assert result["data"] == VALID_INPUT
 
 
 def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once() -> None:
@@ -316,7 +327,10 @@ def test_options_flow_create_entry_data_lands_in_entry_options_and_reloads_once(
     _run(registry.dispatch_options_update(entry))
     assert hass.config_entries.async_reload.call_count == 1
     assert registry.reloaded == [entry.entry_id]
-    assert entry.options == VALID_INPUT
+    assert entry.options == {
+        **VALID_INPUT,
+        CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
+    }
 
 
 def test_options_flow_validation_never_raises() -> None:

tests/test_sensors.py

@@ -0,0 +1,515 @@
+"""Tests for the per-child daily count and percentage sensors (Feature 10, task 2)."""
+from __future__ import annotations
+
+import datetime
+
+import pytest
+
+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
+from custom_components.nestquest.completion import complete_instance
+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
+from custom_components.nestquest.sensor import (
+    MAX_STATE_LENGTH,
+    _state_within_limit,
+)
+
+#: The seeding horizon: instances exist for today and a few days out,
+#: enough for the day snapshot without slowing the walk down.
+SEED_HORIZON_DAYS = 3
+
+#: The documented per-child day sensor kinds (unique_id suffixes).
+SENSOR_KINDS = (
+    "quests_due_today",
+    "quests_completed_today",
+    "quests_remaining_today",
+    "completion_pct_today",
+)
+
+
+def _today() -> datetime.date:
+    return datetime.date.today()
+
+
+async def _seed_children(database):
+    """Three active children: Ada and Bo share one daily quest; Cory has none."""
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+    cory = await create_child(database, "Cory")
+    today = _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, cory
+
+
+async def _setup_seeded_entry(hass, make_entry):
+    """Set up the integration, seed three children, refresh: sensors exist.
+
+    Children are seeded AFTER setup (the entry's database only exists
+    once setup opens it), so the sensors appear through the refresh
+    listener — the same path production takes when a child is added.
+    """
+    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
+    ada, bo, cory = await _seed_children(database)
+    await coordinator.async_refresh()
+    return entry, coordinator, (ada, bo, cory)
+
+
+def _snapshot_child(coordinator, child_id):
+    """Return the child's current day snapshot from the coordinator."""
+    for child in coordinator.data.children:
+        if child.child_id == child_id:
+            return child
+    raise AssertionError(f"child {child_id} missing from the snapshot")
+
+
+def _sensor(hass, child_id, kind):
+    """Return the registered sensor entity for (child, kind)."""
+    return hass.entities[f"{DOMAIN}_child_{child_id}_{kind}"]
+
+
+async def _complete_first_instance(database, coordinator, child) -> None:
+    """Complete the child's first open instance through the business layer."""
+    snapshot_child = _snapshot_child(coordinator, child.id)
+    instance_id = snapshot_child.instances[0].instance_id
+    await complete_instance(
+        database,
+        instance_id,
+        actor_source="panel",
+        actor_child_id=child.id,
+    )
+
+
+async def test_every_active_child_gets_five_sensors_with_documented_identity(
+    hass, make_entry
+) -> None:
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+    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 + 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),
+        cory.id: ("Cory", 0, 0, 0, 100),
+    }
+    for child_id, (name, due, completed, remaining, pct) in expected.items():
+        due_entity = _sensor(hass, child_id, "quests_due_today")
+        completed_entity = _sensor(hass, child_id, "quests_completed_today")
+        remaining_entity = _sensor(hass, child_id, "quests_remaining_today")
+        pct_entity = _sensor(hass, child_id, "completion_pct_today")
+        next_entity = _sensor(hass, child_id, "next_quest")
+        assert due_entity.name == f"NestQuest {name} quests due today"
+        assert completed_entity.name == (
+            f"NestQuest {name} quests completed today"
+        )
+        assert remaining_entity.name == (
+            f"NestQuest {name} quests remaining today"
+        )
+        assert pct_entity.name == f"NestQuest {name} completion pct today"
+        assert due_entity.native_value == due
+        assert completed_entity.native_value == completed
+        assert remaining_entity.native_value == remaining
+        assert pct_entity.native_value == pct
+        for entity in (due_entity, completed_entity, remaining_entity):
+            assert entity.state_class == "measurement"
+            assert entity.native_unit_of_measurement is None
+        assert pct_entity.state_class == "measurement"
+        assert pct_entity.native_unit_of_measurement == "%"
+        assert pct_entity.device_class == "percentage"
+        device_info = due_entity.device_info
+        assert device_info.name == f"NestQuest {name}"
+        assert device_info.manufacturer == "NestQuest"
+        assert device_info.identifiers == {(DOMAIN, f"child_{child_id}")}
+
+
+async def test_zero_quest_child_is_a_perfect_day(hass, make_entry) -> None:
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    due_entity = _sensor(hass, cory.id, "quests_due_today")
+    pct_entity = _sensor(hass, cory.id, "completion_pct_today")
+    assert due_entity.native_value == 0
+    assert pct_entity.native_value == 100, (
+        "zero quests is a perfect day, not a division error"
+    )
+    attributes = due_entity.extra_state_attributes
+    assert attributes["instances"] == []
+    assert attributes["admin_instances"] == []
+
+
+async def test_due_today_attributes_carry_the_panel_payload(
+    hass, make_entry
+) -> None:
+    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()
+
+    ada_due = _sensor(hass, ada.id, "quests_due_today")
+    attributes = ada_due.extra_state_attributes
+    assert set(attributes) == {
+        "instances",
+        "admin_instances",
+        "child_id",
+        "child_name",
+        "present",
+    }
+    assert attributes["child_id"] == ada.id
+    assert attributes["child_name"] == "Ada"
+    assert attributes["present"] is True
+    [instance] = attributes["instances"]
+    assert set(instance) == {
+        "id",
+        "definition_id",
+        "child_id",
+        "title",
+        "icon",
+        "window",
+        "due_time",
+        "state",
+        "overdue",
+        "completed_at",
+        "on_time",
+    }
+    assert instance["child_id"] == ada.id
+    assert instance["title"] == "Brush teeth"
+    assert instance["window"] == "morning"
+    assert instance["state"] == "completed", (
+        "the derived 'done' maps to the panel's 'completed' spelling"
+    )
+    assert instance["completed_at"] is not None
+    assert instance["on_time"] is not None
+    assert instance["overdue"] is False
+    assert attributes["admin_instances"] == attributes["instances"], (
+        "with nothing missed, admin and panel payloads are identical"
+    )
+
+    bo_due = _sensor(hass, bo.id, "quests_due_today")
+    [bo_instance] = bo_due.extra_state_attributes["instances"]
+    assert bo_instance["state"] == "open"
+    assert bo_instance["completed_at"] is None
+    assert bo_instance["on_time"] is None
+
+
+async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
+    hass, make_entry, monkeypatch
+) -> None:
+    """D-009: the panel payload omits missed instances, the admin one keeps them.
+
+    A missed instance is derived (``due_date`` before the snapshot's
+    "today"), never stored.  The coordinator's day listing is keyed on
+    ``_local_now``'s date, so pinning the clock forward would move the
+    listing window off today's rows entirely; the pinned FUTURE clock
+    is therefore applied at the ``derive_state`` boundary instead —
+    today's open instances then read as missed through the same
+    derivation rule, without back-dating instance rows past the DAO's
+    no-past guard or touching append-only completion events.
+    """
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    import custom_components.nestquest.coordinator as coordinator_module
+
+    tomorrow = _today() + datetime.timedelta(days=1)
+    real_derive_state = coordinator_module.derive_state
+
+    def _derive_against_pinned_future(instance, latest, today):
+        return real_derive_state(instance, latest, tomorrow)
+
+    monkeypatch.setattr(
+        coordinator_module, "derive_state", _derive_against_pinned_future
+    )
+    await coordinator.async_refresh()
+
+    bo_due = _sensor(hass, bo.id, "quests_due_today")
+    attributes = bo_due.extra_state_attributes
+    assert attributes["instances"] == [], (
+        "missed instances are omitted from the panel payload"
+    )
+    [missed] = attributes["admin_instances"]
+    assert missed["state"] == "missed"
+    assert missed["completed_at"] is None
+    # Counts still roll the missed instance up as not-completed.
+    assert _sensor(hass, bo.id, "quests_completed_today").native_value == 0
+    assert _sensor(hass, bo.id, "quests_remaining_today").native_value == 1
+
+
+@pytest.mark.parametrize("kind", SENSOR_KINDS)
+async def test_no_sensor_state_exceeds_the_255_character_limit(
+    hass, make_entry, kind
+) -> None:
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    for child in (ada, bo, cory):
+        entity = _sensor(hass, child.id, kind)
+        state = entity.native_value
+        assert len(str(state)) <= MAX_STATE_LENGTH
+
+
+def test_state_within_limit_refuses_overlong_strings() -> None:
+    """The truncation guard: numerics pass, overlong strings fail fast."""
+    assert _state_within_limit(3) == 3
+    assert _state_within_limit("ok") == "ok"
+    with pytest.raises(ValueError, match="255"):
+        _state_within_limit("x" * (MAX_STATE_LENGTH + 1))
+
+
+async def test_sensors_update_on_refresh_after_a_completion(
+    hass, make_entry
+) -> None:
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+    completed_entity = _sensor(hass, ada.id, "quests_completed_today")
+    remaining_entity = _sensor(hass, ada.id, "quests_remaining_today")
+    pct_entity = _sensor(hass, ada.id, "completion_pct_today")
+    due_entity = _sensor(hass, ada.id, "quests_due_today")
+    assert completed_entity.native_value == 0
+    assert remaining_entity.native_value == 1
+    assert pct_entity.native_value == 0
+
+    # Coordinator listener wiring: each refresh writes the entity's
+    # state (the registration the cards' state_changed handlers ride).
+    writes = []
+    real_write = due_entity.async_write_ha_state
+
+    def _record_write():
+        writes.append(due_entity.unique_id)
+        real_write()
+
+    due_entity.async_write_ha_state = _record_write
+
+    await _complete_first_instance(database, coordinator, ada)
+    await coordinator.async_refresh()
+
+    assert completed_entity.native_value == 1
+    assert remaining_entity.native_value == 0
+    assert pct_entity.native_value == 100
+    assert writes == [due_entity.unique_id]
+
+
+async def test_repeated_refreshes_do_not_duplicate_entities(
+    hass, make_entry
+) -> None:
+    entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    await coordinator.async_refresh()
+    await coordinator.async_refresh()
+    assert len(hass.entities) == 24
+
+
+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
+    # 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",
+        "binary_sensor",
+    ]
+
+
+async def test_rename_propagates_to_name_and_device_label(
+    hass, make_entry
+) -> None:
+    """A manage_child rename shows on the next refresh: the name and
+    device label re-derive from the snapshot while the unique_id (and
+    therefore history) is untouched."""
+    from custom_components.nestquest.children import edit_child
+
+    entry, coordinator, (ada, _bo, _cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+    due_sensor = _sensor(hass, ada.id, "quests_due_today")
+    original_unique_id = due_sensor.unique_id
+    assert due_sensor.name.startswith("NestQuest Ada ")
+
+    await edit_child(database, ada.id, display_name="Mirren")
+    await coordinator.async_refresh()
+
+    assert due_sensor.name.startswith("NestQuest Mirren ")
+    assert due_sensor.device_info.name == "NestQuest Mirren"
+    assert due_sensor.unique_id == original_unique_id
+
+
+async def test_failed_platform_unload_returns_false_and_retains_runtime(
+    hass, make_entry
+) -> None:
+    """When HA reports platforms still loaded, the unload must NOT
+    close the database the live entities read: it returns False and
+    keeps the runtime record for a retry."""
+    entry, _coordinator, _children = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+
+    async def _refuse_unload(entry_arg, platforms):
+        return False
+
+    hass.config_entries.async_unload_platforms = _refuse_unload
+    assert await async_unload_entry(hass, entry) is False
+    assert hass.data[DOMAIN][entry.entry_id] is entry.runtime_data
+    assert database.connected is True
+
+async def test_next_quest_sensor_reports_earliest_open_quest(
+    hass, make_entry
+) -> None:
+    """The next-quest sensor carries the earliest open quest's title in
+    state with the full shape in attributes."""
+    entry, coordinator, (ada, _bo, _cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    next_entity = _sensor(hass, ada.id, "next_quest")
+    assert next_entity.name == f"NestQuest Ada next quest"
+    assert next_entity.native_value == "Brush teeth"
+    attributes = next_entity.extra_state_attributes
+    assert attributes["title"] == "Brush teeth"
+    assert attributes["window"] == "morning"
+    assert attributes["instance_id"] == _snapshot_child(
+        coordinator, ada.id
+    ).instances[0].instance_id
+    assert attributes["child_id"] == ada.id
+
+
+async def test_next_quest_sensor_reports_none_sentinel_when_day_clear(
+    hass, make_entry
+) -> None:
+    """A child with every quest done (and a zero-quest child) reports
+    the documented sentinel, never an empty string."""
+    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()
+    assert _sensor(hass, ada.id, "next_quest").native_value == "none"
+    assert _sensor(hass, cory.id, "next_quest").native_value == "none"
+
+
+async def test_next_quest_state_truncates_over_the_limit(
+    hass, make_entry
+) -> None:
+    """A quest title longer than the state limit is truncated; the
+    attribute keeps the full title."""
+    entry, coordinator, (ada, _bo, _cory) = await _setup_seeded_entry(
+        hass, make_entry
+    )
+    database = entry.runtime_data.database
+    from custom_components.nestquest.quest_definitions import (
+        edit_quest_definition,
+    )
+
+    snapshot_child = _snapshot_child(coordinator, ada.id)
+    await edit_quest_definition(
+        database,
+        snapshot_child.instances[0].definition_id,
+        title="A very long quest title. " * 30,
+    )
+    await coordinator.async_refresh()
+    next_entity = _sensor(hass, ada.id, "next_quest")
+    full_title = _snapshot_child(coordinator, ada.id).instances[0].title
+    assert len(full_title) > 255
+    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()
+    )