Fail-closed is_admin resolver (Feature 03, task 5)

admin_allowlist.is_admin: True only for an allowlisted HA user id;
False for non-allowlisted, unknown, empty-string, untrimmed variants,
None, non-strings (bools, lists), and an empty allowlist (nobody is
admin rather than everybody). Never raises for malformed input —
malformed means not admin; DB errors propagate for the Feature 09
gate to handle. The last-admin removal refusal (set_admin_ids) is
re-asserted as keeping the verdict sound. Tests: full verdict matrix,
empty-allowlist fail-closed, refusal persistence. Suite: 912 passed.

custom_components/nestquest/__init__.py

@@ -13,10 +13,11 @@ from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 
-from .const import DOMAIN, LOGGER
+from .const import CONF_ADMIN_USER_IDS, DOMAIN, LOGGER
 from .db import NestQuestDatabase
 from .migrations import apply_migrations
 from .store import async_get_db_path
+from .admin_allowlist import seed_setup_admin
 
 _LOGGER = LOGGER
 
@@ -225,6 +226,28 @@ class NestQuestRuntimeData:
     remove_update_listener: Callable[[], Any]
 
 
+async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
+    """Return every Home Assistant owner-account user id.
+
+    The last-resort admin seed: when no persisted allowlist copy and no
+    flow-context user survive to first setup, the HA owner accounts are
+    seeded so the household is never left with zero admins (an empty
+    allowlist fails closed, which would lock the household out of its
+    own integration).  Owner accounts can already do everything in HA,
+    so granting NestQuest admin is not a privilege escalation.
+    """
+    auth = getattr(hass, "auth", None)
+    if auth is None:
+        return []
+    users = await auth.async_get_users()
+    owners = [
+        user.id
+        for user in users
+        if getattr(user, "is_owner", False) and getattr(user, "id", None)
+    ]
+    return owners
+
+
 async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool:
     """Set up the NestQuest integration. YAML configuration is not used, returns True."""
     return True
@@ -241,10 +264,36 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             # record without a live connection (a previous setup's
             # close-on-failure, or a hand-replaced file) must get the
             # missing/valid/corrupt classification too, never a raw
-            # sqlite3 error and never skipped migrations.
+            # sqlite3 error and never skipped migrations — and the
+            # same allowlist seeding, so a reopened fresh file is
+            # never ownerless.
             db = await _async_open_database(
                 hass, await async_get_db_path(hass)
             )
+            try:
+                stored_admin_ids = (
+                    entry.options.get(CONF_ADMIN_USER_IDS)
+                    or entry.data.get(CONF_ADMIN_USER_IDS)
+                )
+                context = getattr(entry, "context", None)
+                context_user_id = (
+                    context.get("user_id")
+                    if isinstance(context, dict)
+                    else None
+                )
+                await seed_setup_admin(
+                    db,
+                    stored_admin_ids=stored_admin_ids,
+                    context_user_id=context_user_id,
+                    owner_ids=await _async_owner_user_ids(hass),
+                )
+            except BaseException:
+                # A seeding failure must not leak the just-opened
+                # connection: close it before the error propagates, so
+                # HA's setup retry starts from a clean handle (the
+                # first-setup path closes symmetrically).
+                await db.close()
+                raise
             existing.database = db
         entry.runtime_data = existing
         return True
@@ -257,6 +306,29 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     database = await _async_open_database(hass, await async_get_db_path(hass))
     try:
+        # Seed the admin allowlist on first setup so the owner is never
+        # locked out: an empty database takes the persisted admin copy
+        # — the options flow's saved list first (it is the current
+        # one), else the config flow's user from entry data — and only
+        # then the HA user still present on the entry context.  No-op
+        # on a non-empty list (fail-closed narrowing is deliberate).
+        stored_admin_ids = (
+            entry.options.get(CONF_ADMIN_USER_IDS)
+            or entry.data.get(CONF_ADMIN_USER_IDS)
+        )
+        context = getattr(entry, "context", None)
+        context_user_id = (
+            context.get("user_id")
+            if isinstance(context, dict)
+            else None
+        )
+        owner_ids = await _async_owner_user_ids(hass)
+        await seed_setup_admin(
+            database,
+            stored_admin_ids=stored_admin_ids,
+            context_user_id=context_user_id,
+            owner_ids=owner_ids,
+        )
         remove_update_listener = entry.add_update_listener(_async_update_listener)
     except BaseException:
         await database.close()

custom_components/nestquest/admin_allowlist.py

@@ -0,0 +1,153 @@
+"""Business layer for the admin allowlist (Feature 03).
+
+Sits between the options flow / Feature 09 service gate and the typed
+:class:`~.dao_children.AdminUsersDao`.  The allowlist stores Home
+Assistant user IDs — never usernames or display names (feature
+guardrail) — and an empty allowlist must fail closed: every function
+here refuses to write an empty allowlist, so the owner can never be
+accidentally locked out of admin operations by a bad save.
+
+Shape policy: a user id is a non-empty trimmed string.  Duplicate ids
+in one write are rejected up front — they would silently map to one
+row and hide a caller bug.  Writes replace the allowlist atomically:
+the diff (additions and removals) is computed and applied inside one
+transaction under the connection-scoped lock shared with the DAOs, so
+a concurrent reader never observes a half-applied sync.
+
+Seed policy (:func:`seed_setup_admin`): the allowlist is seeded only
+when it is EMPTY, so re-running setup never resurrects a deliberately
+narrowed list.  The setup user (the HA user who completed the config
+flow, via the entry context) is added automatically so the owner is
+never locked out on first setup; a stored entry-data copy of the list
+re-seeds a lost database.
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+
+from .dao_children import AdminUsersDao, _connection_lock
+from .db import NestQuestDatabase
+
+LOGGER = logging.getLogger(__name__)
+
+
+def _now_stamp() -> str:
+    """UTC second-precision stamp, matching the registry's policy."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _validate_user_id(value: object) -> str:
+    """Return ``value`` trimmed; raise unless a non-empty string."""
+    if not isinstance(value, str):
+        raise ValueError(
+            f"admin user id must be a string, got {value!r}"
+        )
+    trimmed = value.strip()
+    if not trimmed:
+        raise ValueError("admin user id must not be empty")
+    return trimmed
+
+
+async def list_admin_ids(database: NestQuestDatabase) -> list[str]:
+    """Return the allowlisted HA user IDs in storage order."""
+    return [row.ha_user_id for row in await AdminUsersDao(database).list()]
+
+
+async def set_admin_ids(
+    database: NestQuestDatabase, user_ids: list[str]
+) -> None:
+    """Replace the allowlist with exactly ``user_ids`` (atomic sync).
+
+    Refuses an empty list — an empty allowlist must fail closed, and
+    the only way to reach one through this API would be removing the
+    last admin, so that removal is refused here with a clear error.
+    Refuses non-string, empty, or duplicate ids up front.  The diff
+    against the current allowlist (additions and removals) applies
+    inside one transaction under the connection lock.
+    """
+    if not isinstance(user_ids, list):
+        raise ValueError("user_ids must be a list of HA user id strings")
+    normalized: list[str] = []
+    for value in user_ids:
+        normalized.append(_validate_user_id(value))
+    if len(set(normalized)) != len(normalized):
+        raise ValueError("user_ids must not contain duplicates")
+    if not normalized:
+        raise ValueError(
+            "refusing to empty the admin allowlist: at least one admin "
+            "must remain (an empty allowlist fails closed)"
+        )
+    dao = AdminUsersDao(database)
+    async with _connection_lock(database):
+        async with database.transaction():
+            current = {row.ha_user_id for row in await dao.list()}
+            target = set(normalized)
+            for user_id in sorted(current - target):
+                await dao.remove(user_id)
+            for user_id in normalized:
+                await dao.add(user_id, _now_stamp())
+
+
+async def is_admin(
+    database: NestQuestDatabase, user_id: object
+) -> bool:
+    """Resolve an HA user ID against the allowlist — FAILS CLOSED.
+
+    True only for an allowlisted user id.  Everything else is False:
+    a non-allowlisted id, an unknown id, a None or non-string value
+    (a service call with no user context), and an empty allowlist
+    (nobody is admin rather than everybody).  The Feature 09 gate
+    wraps every admin-only service with this verdict; it never raises
+    for malformed input — malformed means not admin.
+    """
+    if not isinstance(user_id, str) or not user_id:
+        return False
+    return await AdminUsersDao(database).exists(user_id)
+
+
+async def seed_setup_admin(
+    database: NestQuestDatabase,
+    *,
+    stored_admin_ids: list[str] | None = None,
+    context_user_id: str | None = None,
+    owner_ids: list[str] | None = None,
+) -> list[str]:
+    """Seed an empty allowlist so the owner is never locked out.
+
+    Only acts when the allowlist is EMPTY.  Preference order: the
+    stored persisted copy of the list (the options flow's current
+    selection, then the config flow's original — disaster recovery for
+    a lost database), then the HA user who completed the config flow
+    (``context_user_id``), then — when no caller identity survived at
+    all — every Home Assistant OWNER account (``owner_ids``), so a
+    first setup whose flow context lost the user id still seeds a
+    working admin.  Returns the allowlist after seeding.  Non-empty
+    allowlists are returned untouched — re-running setup must never
+    resurrect a deliberately narrowed list.
+    """
+    dao = AdminUsersDao(database)
+    async with _connection_lock(database):
+        current = [row.ha_user_id for row in await dao.list()]
+        if current:
+            return current
+        candidates: list[str] = []
+        if stored_admin_ids is not None:
+            if not isinstance(stored_admin_ids, list):
+                raise ValueError(
+                    "stored_admin_ids must be a list of HA user id strings"
+                )
+            candidates = [_validate_user_id(value) for value in stored_admin_ids]
+        elif context_user_id is not None:
+            candidates = [_validate_user_id(context_user_id)]
+        elif owner_ids:
+            if not isinstance(owner_ids, list):
+                raise ValueError(
+                    "owner_ids must be a list of HA user id strings"
+                )
+            candidates = [_validate_user_id(value) for value in owner_ids]
+        for user_id in candidates:
+            await dao.add(user_id, _now_stamp())
+        return [row.ha_user_id for row in await dao.list()]

custom_components/nestquest/children.py

@@ -0,0 +1,280 @@
+"""Business layer for the children registry (Feature 03).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_children.ChildrenDao`: callers get validation, name
+policies and the duplicate-name warning here, and never talk to the
+DAO or SQL directly.  Children are never hard-deleted — deactivation
+is a separate explicit operation (:func:`set_child_active`), so
+completion history keeps its references (feature guardrail).
+
+Name policy: the display name is required and trimmed; an empty or
+whitespace-only name is rejected with ValueError.  A duplicate display
+name is ALLOWED (two children may legitimately share a nickname) but
+logged as a warning so an accidental double-add is visible in the HA
+log.  The duplicate check is case-insensitive and whitespace-normalised
+on both sides, because "ada" and "Ada " are the same name to a parent.
+
+Colour and avatar are optional free-form strings, trimmed when present;
+``sort_order`` is an int with the DAO/schema default of 0.  Clearing an
+optional field back to NULL is deliberately rejected: the DAO's update
+writes only provided fields (it cannot express NULL), so the business
+layer refuses an explicit ``None`` instead of silently ignoring it.
+Timestamps follow the module timestamp policy: the business layer
+stamps UTC ISO-8601 at second precision so every row it creates is
+uniform.
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+
+from .dao_children import ChildRecord, ChildrenDao, _connection_lock
+from .db import NestQuestDatabase
+
+LOGGER = logging.getLogger(__name__)
+
+#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
+#: in :func:`edit_child`.
+_UNSET = object()
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _validate_text(
+    value: object, field: str, *, required: bool = False
+) -> str | None:
+    """Validate a free-text field and return it trimmed.
+
+    ``None`` raises when ``required`` and otherwise returns None.  A
+    non-string value is rejected; an empty result is rejected for
+    required fields (whitespace-only names are empty to a parent).
+    """
+    if value is None:
+        if required:
+            raise ValueError(f"{field} is required")
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{field} must be a string, got {value!r}")
+    trimmed = value.strip()
+    if required and not trimmed:
+        raise ValueError(f"{field} is required")
+    return trimmed
+
+
+def _validate_sort_order(value: object) -> int:
+    """Reject non-int sort orders (bools included) before the DAO."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"sort_order must be an integer, got {value!r}")
+    return value
+
+
+def _validate_child_id(value: object) -> int:
+    """Reject non-int child ids (bools included) before any lookup.
+
+    SQLite binds Python bools as integers, so ``True`` would silently
+    address child 1 and a float would round — malformed service input
+    must never mutate the wrong profile.
+    """
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"child_id must be an integer, got {value!r}")
+    return value
+
+
+async def _warn_on_duplicate_name(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    exclude_child_id: int | None = None,
+) -> None:
+    """Log a warning when ``display_name`` duplicates an existing child.
+
+    Case-insensitive and whitespace-normalised on both sides.  The
+    duplicate is still allowed — this only makes it visible; the log
+    line names both rows so an accidental double-add can be found.
+    ``exclude_child_id`` removes the child being edited from the scan,
+    so an edit only warns when the result collides with a DIFFERENT
+    child's name — including case-only edits, where the self-comparison
+    would otherwise hide a real duplicate.
+    """
+    lowered = display_name.casefold()
+    for child in await ChildrenDao(database).list_all():
+        if exclude_child_id is not None and child.id == exclude_child_id:
+            continue
+        if child.display_name.strip().casefold() == lowered:
+            LOGGER.warning(
+                "Child display name %r duplicates existing child %d "
+                "(%r); allowed, but check this was intentional",
+                display_name,
+                child.id,
+                child.display_name,
+            )
+            return
+
+
+async def create_child(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    colour: str | None = None,
+    avatar_ref: str | None = None,
+    sort_order: int = 0,
+) -> ChildRecord:
+    """Create a child profile and return the stored record.
+
+    The display name is required, trimmed, and rejected when empty.
+    A duplicate display name (case-insensitive) is allowed but logged
+    as a warning.  The scan and the INSERT run inside the
+    connection-scoped lock shared with the DAOs, so a concurrent
+    create of the same normalised name queues behind the first insert
+    and sees it — the duplicate still warns; it can never slip through
+    unlogged.
+    """
+    name = _validate_text(display_name, "display_name", required=True)
+    assert name is not None
+    colour_value = _validate_text(colour, "colour")
+    avatar_value = _validate_text(avatar_ref, "avatar_ref")
+    order = _validate_sort_order(sort_order)
+    async with _connection_lock(database):
+        await _warn_on_duplicate_name(database, name)
+        created = await ChildrenDao(database).create(
+            name,
+            _now_stamp(),
+            colour=colour_value,
+            avatar_ref=avatar_value,
+            sort_order=order,
+        )
+    return created
+
+
+async def edit_child(
+    database: NestQuestDatabase,
+    child_id: int,
+    *,
+    display_name: str | object = _UNSET,
+    colour: str | None | object = _UNSET,
+    avatar_ref: str | None | object = _UNSET,
+    sort_order: int | object = _UNSET,
+) -> ChildRecord:
+    """Edit a child profile and return the updated record.
+
+    Arguments default to the module sentinel ``_UNSET`` meaning "leave
+    this field alone".  An explicitly passed ``None`` for an optional
+    field raises: the DAO's update writes only provided fields and
+    cannot express NULL, so a silent no-op would mask the caller's
+    intent (see the module docstring).  A provided display name is
+    validated like create's; the duplicate scan excludes only the child
+    being edited, so a case-only edit colliding with another child's
+    normalised name still warns.  Raises ValueError when the child does
+    not exist.
+
+    Atomicity: the existence check, the duplicate scan, the UPDATE and
+    the readback all run inside the connection-scoped lock shared with
+    the DAOs, so another edit cannot slip between this call's UPDATE
+    and its readback — the returned record is exactly the one this
+    edit produced, not a later writer's.
+    """
+    _validate_child_id(child_id)
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        existing = await dao.get(child_id)
+        if existing is None:
+            raise ValueError(f"child {child_id} does not exist")
+        updates: dict[str, object] = {}
+        if display_name is not _UNSET:
+            name = _validate_text(
+                display_name, "display_name", required=True
+            )
+            assert name is not None
+            await _warn_on_duplicate_name(
+                database, name, exclude_child_id=child_id
+            )
+            updates["display_name"] = name
+        if colour is not _UNSET:
+            value = _validate_text(colour, "colour")
+            if value is None:
+                raise ValueError(
+                    "clearing colour is not supported; pass a replacement "
+                    "value or omit the field"
+                )
+            updates["colour"] = value
+        if avatar_ref is not _UNSET:
+            value = _validate_text(avatar_ref, "avatar_ref")
+            if value is None:
+                raise ValueError(
+                    "clearing avatar_ref is not supported; pass a "
+                    "replacement value or omit the field"
+                )
+            updates["avatar_ref"] = value
+        if sort_order is not _UNSET:
+            updates["sort_order"] = _validate_sort_order(sort_order)
+        if updates:
+            await dao.update(child_id, **updates)
+        edited = await dao.get(child_id)
+    assert edited is not None
+    return edited
+
+
+async def set_child_active(
+    database: NestQuestDatabase, child_id: int, is_active: bool
+) -> ChildRecord:
+    """Deactivate or reactivate a child; returns the updated record.
+
+    Deactivation hides the child from the panel without deleting their
+    history (feature guardrail); this is the only removal path.
+    ``is_active`` must be a real bool: the DAO's ``int()`` would
+    otherwise silently coerce strings and numerics ("1", 0) into a
+    state the caller never asked for.  Raises ValueError when the
+    child does not exist.
+
+    Atomicity: existence check, UPDATE and readback run inside the
+    connection-scoped lock, so a concurrent opposite transition cannot
+    make this call report the other caller's value.
+    """
+    _validate_child_id(child_id)
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        if await dao.get(child_id) is None:
+            raise ValueError(f"child {child_id} does not exist")
+        if not isinstance(is_active, bool):
+            raise ValueError(
+                f"is_active must be a real bool, got {is_active!r}"
+            )
+        await dao.set_active(child_id, is_active)
+        updated = await dao.get(child_id)
+    assert updated is not None
+    return updated
+
+
+async def reorder_children(
+    database: NestQuestDatabase, ordered_ids: list[int]
+) -> None:
+    """Rewrite the children's panel display order in one transaction.
+
+    ``ordered_ids`` must be a complete permutation of the children
+    table — every child id exactly once.  A partial list (missing an
+    existing child), an unknown id, or a duplicate is rejected with
+    ValueError before any write, so the current order survives a
+    rejected call.  The DAO enforces this inside its own transaction
+    (check and write are     atomic under the connection lock); this
+    wrapper additionally rejects non-integer ids up front, since
+    ``bool`` and floats would silently compare against stored integer
+    ids at the SQL layer.
+    """
+    for child_id in ordered_ids:
+        _validate_child_id(child_id)
+    await ChildrenDao(database).reorder(ordered_ids)
+
+
+async def list_children(
+    database: NestQuestDatabase, *, active_only: bool = False
+) -> list[ChildRecord]:
+    """Return children in display order (sort_order, then id)."""
+    dao = ChildrenDao(database)
+    if active_only:
+        return await dao.list_active()
+    return await dao.list_all()

custom_components/nestquest/config_flow.py

@@ -7,7 +7,7 @@ from homeassistant import config_entries
 from homeassistant.core import callback
 from homeassistant.data_entry_flow import FlowResult
 
-from .const import DOMAIN
+from .const import CONF_ADMIN_USER_IDS, DOMAIN
 
 
 class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
@@ -28,7 +28,14 @@ class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
-        """Handle the user step of the config flow."""
+        """Handle the user step of the config flow.
+
+        The HA user completing the flow is persisted into the entry
+        data as the initial admin allowlist copy: real HA does not
+        guarantee the flow context survives to setup, and entry data is
+        the persisted record first setup seeds from — the owner is
+        never locked out.
+        """
         await self.async_set_unique_id(DOMAIN, raise_on_progress=False)
         if self._async_in_progress() or self._async_current_entries():
             return self.async_abort(reason="single_instance_allowed")
@@ -36,4 +43,12 @@ class NestQuestConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         if user_input is None:
             return self.async_show_form(step_id="user")
 
-        return self.async_create_entry(title="NestQuest", data=user_input)
+        data = dict(user_input)
+        flow_user_id = (
+            self.context.get("user_id")
+            if isinstance(self.context, dict)
+            else None
+        )
+        if flow_user_id:
+            data[CONF_ADMIN_USER_IDS] = [flow_user_id]
+        return self.async_create_entry(title="NestQuest", data=data)

custom_components/nestquest/const.py

@@ -20,3 +20,34 @@ DEFAULT_DAY_ROLLOVER_TIME = "00:00"
 
 CONF_PANEL_IDLE_TIMEOUT = "panel_idle_timeout"
 DEFAULT_PANEL_IDLE_TIMEOUT = 300
+
+# Admin allowlist (Feature 03): HA user IDs permitted to call admin-only
+# services.  Stored as IDs, never usernames (feature guardrail).  The
+# database allowlist is authoritative; the config entry carries the same
+# list as a disaster-recovery copy so a lost database can be re-seeded.
+CONF_ADMIN_USER_IDS = "admin_user_ids"
+
+# 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
+# configurable later.  Boundaries: a window's end equals the next
+# window's start; classification (Feature 07 materialisation) treats a
+# due time exactly on the boundary as belonging to the LATER window, so
+# 12:00 is afternoon (not morning) and 17:00 is evening (not
+# afternoon).  Times after 21:00 belong to no window.
+WINDOW_MORNING = "morning"
+WINDOW_AFTERNOON = "afternoon"
+WINDOW_EVENING = "evening"
+
+QUEST_WINDOWS: tuple[str, ...] = (
+    WINDOW_MORNING,
+    WINDOW_AFTERNOON,
+    WINDOW_EVENING,
+)
+
+#: Inclusive "HH:MM" bounds per window, in 24-hour local time.
+WINDOW_CLOCK_RANGES: dict[str, tuple[str, str]] = {
+    WINDOW_MORNING: ("00:00", "11:59"),
+    WINDOW_AFTERNOON: ("12:00", "17:00"),
+    WINDOW_EVENING: ("17:00", "21:00"),
+}

custom_components/nestquest/dao_children.py

@@ -191,13 +191,20 @@ class ChildrenDao:
     async def reorder(self, ordered_ids: list[int]) -> None:
         """Persist a new sort order: ``ordered_ids[i]`` gets sort_order i.
 
+        The list must be a COMPLETE permutation of the children table:
+        every existing child id appears exactly once.  Partial lists
+        (missing an existing child) and unknown ids (naming a child
+        that does not exist) are rejected with ValueError BEFORE any
+        write, leaving the current order untouched — a silent partial
+        reorder would leave unlisted children with stale positions and
+        hide a caller bug.  Duplicate ids are rejected up front: they
+        cannot all hold distinct positions, and silently assigning the
+        last occurrence's index would hide a caller bug.
+
         Runs inside one transaction; positions are staged at offset
         -len (all negative, so any permutation of the same ids never
         collides mid-reorder under the wrapper's single-connection
-        serialization), then finalized 0..n-1.  Duplicate ids are
-        rejected up front: they cannot all hold distinct positions, and
-        silently assigning the last occurrence's index would hide a
-        caller bug.
+        serialization), then finalized 0..n-1.
 
         Concurrent reorder() calls on one connection are serialized by
         a connection-scoped asyncio lock (shared across every ChildrenDao
@@ -205,10 +212,10 @@ class ChildrenDao:
         transaction() rejects a second concurrent transaction with
         RuntimeError, so racing callers would crash instead of
         queueing.  The lock makes the second call wait, then run
-        against the first call's final state.
+        against the first call's final state — which also keeps the
+        completeness check honest: the re-read of the table happens
+        inside the same lock and transaction as the writes.
         """
-        if not ordered_ids:
-            return
         if len(set(ordered_ids)) != len(ordered_ids):
             raise ValueError(
                 "reorder() received duplicate child ids; every id must "
@@ -216,6 +223,29 @@ class ChildrenDao:
             )
         async with _connection_lock(self._database):
             async with self._database.transaction():
+                rows = await self._database.fetch_all(
+                    f"SELECT id FROM children"
+                )
+                existing = {row[0] for row in rows}
+                listed = set(ordered_ids)
+                missing = sorted(existing - listed)
+                unknown = sorted(listed - existing)
+                if missing or unknown:
+                    details = []
+                    if missing:
+                        details.append(
+                            f"missing child ids {missing} (a partial "
+                            "reorder would leave them with stale "
+                            "positions)"
+                        )
+                    if unknown:
+                        details.append(
+                            f"unknown child ids {unknown}"
+                        )
+                    raise ValueError(
+                        "reorder() must list every child exactly once; "
+                        + "; ".join(details)
+                    )
                 offset = len(ordered_ids)
                 await self._database.execute_many(
                     "UPDATE children SET sort_order = ? WHERE id = ?",

custom_components/nestquest/dao_instances.py

@@ -15,13 +15,13 @@ event.  A package-wide guard test (bottom of the test module) keeps
 mutation SQL for this table out of every other module.
 
 Upsert semantics: ``QuestInstancesDao.upsert`` is idempotent on
-(definition_id, due_date) — re-running materialization for a date
-whose instance already exists updates nothing that matters
-(child_id/due_time are refreshed only in the degenerate re-generation
-case where they changed) and never duplicates a row.  The schema's
-UNIQUE(definition_id, due_date) is the last line of defense.  The
-upsert refuses a child that is not an assignee of the definition
-(D-008 multi-assignee model).
+(definition_id, child_id, due_date, window) — re-running
+materialization for a tuple whose instance already exists refreshes
+only the generation-time snapshot columns and never duplicates a row.
+The schema's UNIQUE(definition_id, child_id, due_date, window) is the
+last line of defense.  The upsert refuses a child that is not an
+assignee of the definition and a window outside ``const.QUEST_WINDOWS``
+(D-008 multi-assignee, multi-window model).
 
 ``delete_future_uncompleted`` implements the Feature 06/07 rule that
 reassignment and schedule edits regenerate future instances: it
@@ -37,7 +37,7 @@ import datetime
 from dataclasses import dataclass
 
 from .dao_children import _connection_lock
-from .dao_rules import _validate_date
+from .dao_rules import _validate_date, _validate_window
 from .db import NestQuestDatabase
 
 #: Timestamp policy for completion_events.occurred_at: strict UTC
@@ -90,6 +90,7 @@ class QuestInstanceRecord:
     id: int
     definition_id: int
     child_id: int
+    window: str
     due_date: str
     due_time: str | None
     generated_at: str
@@ -105,16 +106,17 @@ class CompletionEventRecord:
     event_type: str
     actor_source: str
     actor_user_id: str | None
+    actor_child_id: int | None
     occurred_at: str
     was_on_time: bool | None
 
 
 _INSTANCE_COLUMNS = (
-    "id, definition_id, child_id, due_date, due_time, generated_at"
+    "id, definition_id, child_id, window, due_date, due_time, generated_at"
 )
 _EVENT_COLUMNS = (
     "id, instance_id, child_id, event_type, actor_source, actor_user_id, "
-    "occurred_at, was_on_time"
+    "actor_child_id, occurred_at, was_on_time"
 )
 
 #: The only event types the append path accepts; the schema CHECK is
@@ -128,9 +130,10 @@ def _instance_from_row(row: tuple) -> QuestInstanceRecord:
         id=row[0],
         definition_id=row[1],
         child_id=row[2],
-        due_date=row[3],
-        due_time=row[4],
-        generated_at=row[5],
+        window=row[3],
+        due_date=row[4],
+        due_time=row[5],
+        generated_at=row[6],
     )
 
 
@@ -142,8 +145,9 @@ def _event_from_row(row: tuple) -> CompletionEventRecord:
         event_type=row[3],
         actor_source=row[4],
         actor_user_id=row[5],
-        occurred_at=row[6],
-        was_on_time=None if row[7] is None else bool(row[7]),
+        actor_child_id=row[6],
+        occurred_at=row[7],
+        was_on_time=None if row[8] is None else bool(row[8]),
     )
 
 
@@ -160,31 +164,40 @@ class QuestInstancesDao:
         due_date: str,
         generated_at: str,
         *,
+        window: str,
         due_time: str | None = None,
     ) -> QuestInstanceRecord:
-        """Create the dated instance for (definition_id, due_date) idempotently.
+        """Create the dated instance for the widened key idempotently.
 
-        Running materialization twice for the same definition/date
-        yields ONE row: the ON CONFLICT path refreshes the
-        generation-time snapshot columns (child_id, due_time) and
-        regenerates the stamp, but can never create a duplicate — the
-        schema UNIQUE(definition_id, due_date) backs this up.
+        The instance key is (definition_id, child_id, due_date, window)
+        (D-008): running materialization twice for the same tuple
+        yields ONE row — the ON CONFLICT path refreshes the
+        generation-time snapshot columns (due_time) and regenerates the
+        stamp, but can never create a duplicate.  The schema's
+        UNIQUE(definition_id, child_id, due_date, window) backs this
+        up, so the twice-daily case (two windows, one child, one date)
+        yields two rows and the shared case (three assignees, one
+        window) yields three.
 
         Guardrails enforced here:
 
         - ``due_date`` must be TODAY or later: instances are never
           generated in the past (Feature 07).  The caller's
           materialization horizon is future-dated by definition.
+        - ``window`` must be one of ``const.QUEST_WINDOWS``.
         - An existing instance WITH a completion event is IMMUTABLE:
           the conflict path must not rewrite its snapshot columns
           (Feature 07/08).  The upsert refuses with ValueError in that
           case — silently no-oping would mask a materialization bug.
+        - The child must be an assignee of the definition (D-008
+          multi-assignee model).
 
         The definition must exist (validated atomically under the
         connection lock so a concurrent definition deactivation or
         child deletion cannot slip between check and insert).
         """
         _validate_date(due_date, "due_date")
+        _validate_window(window)
         # Fail fast on obviously-past dates BEFORE queuing on the
         # connection lock (a nice early error); the AUTHORITATIVE
         # check runs after the lock is acquired, because a caller can
@@ -227,51 +240,56 @@ class QuestInstancesDao:
                         f"assigned to child {child_id}; instance must "
                         "carry an assigned child"
                     )
-                child = await self._database.fetch_one(
-                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
-                )
-                if child is None:
-                    raise ValueError(f"child {child_id} does not exist")
                 existing = await self._database.fetch_one(
                     "SELECT 1 FROM completion_events "
                     "WHERE instance_id = (SELECT id FROM quest_instances "
-                    "WHERE definition_id = ? AND due_date = ?)",
-                    (definition_id, due_date),
+                    "WHERE definition_id = ? AND child_id = ? "
+                    "AND due_date = ? AND window = ?)",
+                    (definition_id, child_id, due_date, window),
                 )
                 if existing is not None:
                     raise ValueError(
                         f"instance for (definition {definition_id}, "
-                        f"due_date {due_date!r}) already has a completion "
+                        f"child {child_id}, due_date {due_date!r}, "
+                        f"window {window!r}) already has a completion "
                         "event and is immutable; it cannot be regenerated"
                     )
                 await self._database.execute(
                     "INSERT INTO quest_instances (definition_id, child_id, "
-                    "due_date, due_time, generated_at) "
-                    "VALUES (?, ?, ?, ?, ?) "
-                    "ON CONFLICT (definition_id, due_date) DO UPDATE SET "
-                    "child_id = excluded.child_id, "
+                    "window, due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
+                    "window) DO UPDATE SET "
                     "due_time = excluded.due_time, "
                     "generated_at = excluded.generated_at",
                     (
                         definition_id,
                         child_id,
+                        window,
                         due_date,
                         due_time,
                         generated_at,
                     ),
                 )
-                instance = await self.get(definition_id, due_date)
+                instance = await self.get(
+                    definition_id, child_id, due_date, window
+                )
         assert instance is not None
         return instance
 
     async def get(
-        self, definition_id: int, due_date: str
+        self,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        window: str,
     ) -> QuestInstanceRecord | None:
-        """Return the instance for (definition_id, due_date), or None."""
+        """Return the instance for the widened key tuple, or None."""
         row = await self._database.fetch_one(
             f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
-            "WHERE definition_id = ? AND due_date = ?",
-            (definition_id, due_date),
+            "WHERE definition_id = ? AND child_id = ? AND due_date = ? "
+            "AND window = ?",
+            (definition_id, child_id, due_date, window),
         )
         return _instance_from_row(row) if row is not None else None
 
@@ -380,19 +398,21 @@ class CompletionEventsDao:
         was_on_time: bool | None,
         *,
         actor_user_id: str | None = None,
+        actor_child_id: int | None = None,
     ) -> CompletionEventRecord:
         """Append one completion or un-completion event.
 
-        ``actor_source`` is 'user' (with ``actor_user_id`` set) or
-        'panel' (no user id — the tapped child is ``child_id``); the
-        actor-pair CHECK enforces this shape at the schema level.
-        ``occurred_at`` must be the module's one strict UTC shape
-        (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range queries
-        stay exact.  ``was_on_time`` is required (True/False) for
-        completions; for un-completions it may be None or the reversed
-        completion's flag (Feature 08 still deciding).  The instance
-        and child must exist — validated atomically under the
-        connection lock.
+        ``actor_source`` is 'user' (an admin action: ``actor_user_id``
+        set, ``actor_child_id`` NULL) or 'panel' (a panel tap: no user
+        id, and ``actor_child_id`` carrying the tapped child profile,
+        D-008); the actor-pair CHECKs enforce this shape at the schema
+        level.  ``occurred_at`` must be the module's one strict UTC
+        shape (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range
+        queries stay exact.  ``was_on_time`` is required (True/False)
+        for completions; for un-completions it may be None or the
+        reversed completion's flag (Feature 08 still deciding).  The
+        instance, child and — for panel events — the tapped child must
+        exist — validated atomically under the connection lock.
         """
         if event_type not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
             raise ValueError(
@@ -412,6 +432,15 @@ class CompletionEventsDao:
             raise ValueError(
                 "actor_source 'panel' must not carry actor_user_id"
             )
+        if actor_source == "panel" and actor_child_id is None:
+            raise ValueError(
+                "actor_source 'panel' requires actor_child_id "
+                "(the tapped child profile)"
+            )
+        if actor_source == "user" and actor_child_id is not None:
+            raise ValueError(
+                "actor_source 'user' must not carry actor_child_id"
+            )
         if was_on_time is not None and type(was_on_time) is not bool:
             raise ValueError(
                 f"was_on_time must be True, False or None, got "
@@ -442,16 +471,28 @@ class CompletionEventsDao:
                 )
                 if child is None:
                     raise ValueError(f"child {child_id} does not exist")
+                if actor_child_id is not None:
+                    tapped = await self._database.fetch_one(
+                        "SELECT 1 FROM children WHERE id = ?",
+                        (actor_child_id,),
+                    )
+                    if tapped is None:
+                        raise ValueError(
+                            f"actor_child_id {actor_child_id} does not "
+                            "reference an existing child"
+                        )
                 result = await self._database.execute(
                     "INSERT INTO completion_events (instance_id, child_id, "
-                    "event_type, actor_source, actor_user_id, occurred_at, "
-                    "was_on_time) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    "event_type, actor_source, actor_user_id, "
+                    "actor_child_id, occurred_at, was_on_time) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         instance_id,
                         child_id,
                         event_type,
                         actor_source,
                         actor_user_id,
+                        actor_child_id,
                         occurred_at,
                         None if was_on_time is None else int(was_on_time),
                     ),

custom_components/nestquest/dao_rules.py

@@ -36,6 +36,7 @@ from __future__ import annotations
 import datetime
 from dataclasses import dataclass
 
+from .const import QUEST_WINDOWS
 from .dao_children import ChildRecord, _child_from_row, _connection_lock
 from .dao_children import _CHILD_COLUMNS
 from .db import NestQuestDatabase
@@ -71,6 +72,38 @@ def _validate_date(value: str, field: str) -> None:
         )
 
 
+_TIME_FORMAT = "%H:%M"
+
+
+def _validate_time(value: str, field: str) -> None:
+    """Raise ValueError unless ``value`` is a strict 24-hour HH:MM.
+
+    Same strictness policy as _validate_date: parse with the exact
+    format, then round-trip so '9:30' (non-padded) is rejected.  Time
+    shapes are caller-side policy (the schema stores TEXT): windows'
+    optional ``due_time`` and definition ``due_time`` both use this.
+    """
+    try:
+        parsed = datetime.datetime.strptime(value, _TIME_FORMAT).time()
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"{field} must be a 24-hour HH:MM time, got {value!r}"
+        ) from None
+    if parsed.strftime(_TIME_FORMAT) != value:
+        raise ValueError(
+            f"{field} must be a strict HH:MM time, got {value!r}"
+        )
+
+
+def _validate_window(window: str) -> None:
+    """Raise ValueError unless ``window`` is a ``const`` window name."""
+    if window not in QUEST_WINDOWS:
+        raise ValueError(
+            f"window must be one of {', '.join(QUEST_WINDOWS)}, "
+            f"got {window!r}"
+        )
+
+
 @dataclass(frozen=True)
 class ScheduleRuleRecord:
     """One row of ``schedule_rules``."""
@@ -100,6 +133,15 @@ class QuestDefinitionRecord:
     created_at: str
 
 
+@dataclass(frozen=True)
+class QuestDefinitionWindowRecord:
+    """One row of ``quest_definition_windows``."""
+
+    definition_id: int
+    window: str
+    due_time: str | None
+
+
 _RULE_COLUMNS = (
     "id, rule_type, interval, weekday_set, day_of_month, nth_weekday, "
     "month, start_date, end_date"
@@ -108,6 +150,7 @@ _DEFINITION_COLUMNS = (
     "id, title, description, icon, schedule_rule_id, due_time, "
     "is_active, created_at"
 )
+_WINDOW_COLUMNS = "definition_id, window, due_time"
 
 
 def _rule_from_row(row: tuple) -> ScheduleRuleRecord:
@@ -137,6 +180,14 @@ def _definition_from_row(row: tuple) -> QuestDefinitionRecord:
     )
 
 
+def _window_from_row(row: tuple) -> QuestDefinitionWindowRecord:
+    return QuestDefinitionWindowRecord(
+        definition_id=row[0],
+        window=row[1],
+        due_time=row[2],
+    )
+
+
 class ScheduleRulesDao:
     """Typed async access to the ``schedule_rules`` table."""
 
@@ -507,6 +558,92 @@ class QuestDefinitionsDao:
         )
         return [_child_from_row(row) for row in rows]
 
+    async def upsert_window(
+        self,
+        definition_id: int,
+        window: str,
+        *,
+        due_time: str | None = None,
+    ) -> QuestDefinitionWindowRecord:
+        """Declare (or re-declare) ``window`` on the definition.
+
+        The composite primary key makes the write idempotent per
+        (definition, window): re-declaring updates ``due_time`` in
+        place instead of duplicating.  The window name must be one of
+        ``const.QUEST_WINDOWS`` (the schema CHECK backs this up) and a
+        provided ``due_time`` must be strict 24-hour HH:MM.  The
+        definition must exist — validated under the connection lock so
+        the verdict cannot go stale before the write.
+        """
+        _validate_window(window)
+        if due_time is not None:
+            _validate_time(due_time, "due_time")
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definition = await self._database.fetch_one(
+                    "SELECT 1 FROM quest_definitions WHERE id = ?",
+                    (definition_id,),
+                )
+                if definition is None:
+                    raise ValueError(
+                        f"quest definition {definition_id} does not exist"
+                    )
+                await self._database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, ?) "
+                    "ON CONFLICT (definition_id, window) DO UPDATE SET "
+                    "due_time = excluded.due_time",
+                    (definition_id, window, due_time),
+                )
+                row = await self._database.fetch_one(
+                    f"SELECT {_WINDOW_COLUMNS} "
+                    "FROM quest_definition_windows "
+                    "WHERE definition_id = ? AND window = ?",
+                    (definition_id, window),
+                )
+        assert row is not None
+        return _window_from_row(row)
+
+    async def list_windows(
+        self, definition_id: int
+    ) -> list[QuestDefinitionWindowRecord]:
+        """Return the definition's windows in ``const`` order.
+
+        Ordered by the canonical window sequence (morning, afternoon,
+        evening) rather than insertion order, so callers always render
+        the Quest Log's three columns consistently.  A definition with
+        no windows returns [].
+        """
+        rows = await self._database.fetch_all(
+            f"SELECT {_WINDOW_COLUMNS} FROM quest_definition_windows "
+            "WHERE definition_id = ?",
+            (definition_id,),
+        )
+        by_name = {row[1]: _window_from_row(row) for row in rows}
+        return [
+            by_name[name]
+            for name in QUEST_WINDOWS
+            if name in by_name
+        ]
+
+    async def remove_window(
+        self, definition_id: int, window: str
+    ) -> bool:
+        """Remove the window declaration; True when a row was removed.
+
+        Future materialization stops producing instances for this
+        (definition, window); existing instances and completion history
+        are never touched here (Feature 06 guardrail).
+        """
+        _validate_window(window)
+        async with _connection_lock(self._database):
+            result = await self._database.execute(
+                "DELETE FROM quest_definition_windows "
+                "WHERE definition_id = ? AND window = ?",
+                (definition_id, window),
+            )
+        return result.rowcount > 0
+
     async def _validate_assignable_child(self, child_id: int) -> None:
         """Raise ValueError unless the child exists and is active.
 

custom_components/nestquest/migrations.py

@@ -24,9 +24,13 @@ names before anything else assumes the new model.  Migration 3 brings
 pre-D-008 databases to the multi-assignee model: it creates
 ``quest_definition_assignees`` and, when ``quest_definitions`` still
 carries a single ``child_id`` column, rebuilds the table without it,
-copying each definition's assignee across.  Each callable migration
-runs inside its own transaction together with its version stamp, so a
-crash mid-step rolls the statements and the stamp back together.
+copying each definition's assignee across.  Migration 4 adds the
+windows table, migration 5 rebuilds ``quest_instances`` onto the
+widened (definition_id, child_id, due_date, window) key, and migration
+6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
+column.  Each callable migration runs inside its own transaction
+together with its version stamp, so a crash mid-step rolls the
+statements and the stamp back together.
 
 Migration entries are either an ordered sequence of SQL statements or
 an async callable taking ``(database, target_version)``.  Callables
@@ -70,9 +74,12 @@ from typing import Union
 
 from .db import NestQuestDatabase
 from .schema import (
+    COMPLETION_EVENTS_TABLE_SQL,
     QUEST_DEFINITIONS_TABLE_SQL,
+    QUEST_INSTANCES_TABLE_SQL,
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -146,6 +153,14 @@ async def _table_has_column(
     return any(row[1] == column for row in rows)
 
 
+#: Window assigned to instances migrated from the pre-window instance
+#: shape (D-008).  Pre-window rows were generated for whole-day
+#: obligations; the morning window is the documented default landing
+#: spot, and future dates are regenerated with real window declarations
+#: by materialization (delete_future_uncompleted + re-run).
+LEGACY_MIGRATED_WINDOW = "morning"
+
+
 async def _rename_legacy_task_tables(
     database: NestQuestDatabase, target_version: int
 ) -> None:
@@ -249,6 +264,138 @@ async def _add_definition_assignees(
         await database.execute("PRAGMA foreign_keys = ON")
 
 
+async def _widen_instance_key(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``quest_instances`` to the D-008 instance key.
+
+    Databases whose instance table still carries the pre-window shape
+    (no ``window`` column, UNIQUE on (definition_id, due_date)) are
+    rebuilt via SQLite's canonical procedure into the widened shape —
+    UNIQUE (definition_id, child_id, due_date, window) — with every
+    migrated row landing in :data:`LEGACY_MIGRATED_WINDOW` so existing
+    ids (and the completion events referencing them) survive intact.
+    The old rows were unique per (definition, due_date), so pinning one
+    window cannot collide with the widened key.  Fresh databases whose
+    table already matches the canonical shape only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` (dropping the old
+    table under enforcement would fire the implicit DELETE and violate
+    the ``completion_events`` foreign key) and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp, so a corrupt copy aborts the whole migration.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "quest_instances", "window"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE quest_instances_rebuilt "
+                    f"{QUEST_INSTANCES_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO quest_instances_rebuilt "
+                    "(id, definition_id, child_id, window, due_date, "
+                    "due_time, generated_at) "
+                    "SELECT id, definition_id, child_id, ?, due_date, "
+                    "due_time, generated_at "
+                    "FROM quest_instances ORDER BY id",
+                    (LEGACY_MIGRATED_WINDOW,),
+                )
+                await database.execute("DROP TABLE quest_instances")
+                await database.execute(
+                    "ALTER TABLE quest_instances_rebuilt "
+                    "RENAME TO quest_instances"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "instance rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Widened quest_instances key; migrated rows pinned "
+                    "to the %r window",
+                    LEGACY_MIGRATED_WINDOW,
+                )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
+async def _add_actor_child_id(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``completion_events`` to the D-008 actor-child shape.
+
+    The ``actor_child_id`` column records which panel profile was
+    tapped; it cannot be added with ``ALTER TABLE ADD COLUMN`` alone
+    because the canonical shape also carries an actor-pair coherence
+    CHECK (panel events require it, user events forbid it), and CHECK
+    constraints cannot be added to an existing table.  Databases with
+    the legacy shape are therefore rebuilt via SQLite's canonical
+    procedure, backfilling each legacy panel row's ``actor_child_id``
+    to ``child_id`` — pre-D-008 panel completions were always recorded
+    against the tapped child themselves — while admin rows stay NULL.
+    Row ids are preserved so ``quest_instances`` references and the
+    append-only history stay intact.  Fresh databases already match
+    the canonical shape and only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp.  The copy is an INSERT..SELECT, not an UPDATE: the
+    completion_events table keeps its append-only guarantee.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "completion_events", "actor_child_id"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE completion_events_rebuilt "
+                    f"{COMPLETION_EVENTS_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO completion_events_rebuilt "
+                    "(id, instance_id, child_id, event_type, actor_source, "
+                    "actor_user_id, actor_child_id, occurred_at, "
+                    "was_on_time) "
+                    "SELECT id, instance_id, child_id, event_type, "
+                    "actor_source, actor_user_id, "
+                    "CASE WHEN actor_source = 'panel' THEN child_id "
+                    "ELSE NULL END, "
+                    "occurred_at, was_on_time "
+                    "FROM completion_events ORDER BY id"
+                )
+                await database.execute("DROP TABLE completion_events")
+                await database.execute(
+                    "ALTER TABLE completion_events_rebuilt "
+                    "RENAME TO completion_events"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "completion_events rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Added actor_child_id to completion_events; legacy "
+                    "panel rows backfilled to their own child_id"
+                )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
 #: The ordered migration list.  Append-only: never edit an applied
 #: entry, add the next one instead.  (Migration 1's content was
 #: rewritten pre-release per D-007 — no version 1 database shipped.)
@@ -256,6 +403,17 @@ MIGRATIONS: Sequence[MigrationStep] = [
     MIGRATION_1_V1_DDL,
     _rename_legacy_task_tables,
     _add_definition_assignees,
+    # Migration 4 (D-008): the windows table is a plain new table — a
+    # CREATE IF NOT EXISTS suffices for both fresh databases (which
+    # already made it in migration 1) and pre-window databases.
+    list(SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL),
+    # Migration 5 (D-008): rebuild quest_instances onto the widened
+    # (definition_id, child_id, due_date, window) key.
+    _widen_instance_key,
+    # Migration 6 (D-008): rebuild completion_events with the
+    # actor_child_id column and its actor-pair coherence CHECK,
+    # backfilling legacy panel rows to their own child_id.
+    _add_actor_child_id,
 ]
 
 

custom_components/nestquest/options_flow.py

@@ -7,15 +7,19 @@ from typing import Any
 import voluptuous as vol
 from homeassistant import config_entries
 from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import config_validation as cv
 
 from .const import (
+    CONF_ADMIN_USER_IDS,
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
     CONF_PANEL_IDLE_TIMEOUT,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
     DEFAULT_PANEL_IDLE_TIMEOUT,
+    DOMAIN,
 )
+from .admin_allowlist import list_admin_ids, set_admin_ids
 
 MIN_HORIZON_DAYS = 1
 MIN_PANEL_IDLE_TIMEOUT = 30
@@ -55,51 +59,168 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
     return errors
 
 
-def _build_schema(current: dict[str, Any]) -> vol.Schema:
-    """Build the form schema, pre-filled with the current values."""
-    return vol.Schema(
-        {
-            vol.Required(
-                CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
-            ): int,
-            vol.Required(
-                CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
-            ): str,
-            vol.Required(
-                CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
-            ): int,
-        }
-    )
+def _build_schema(
+    current: dict[str, Any],
+    *,
+    admin_choices: dict[str, str] | None = None,
+    admin_default: list[str] | None = None,
+) -> vol.Schema:
+    """Build the form schema, pre-filled with the current values.
+
+    The admin picker is a multi-select of EXISTING Home Assistant users
+    (never a free-text field — the allowlist stores user IDs); its
+    default is the current database allowlist, so an untouched submit
+    re-saves exactly who is already allowed.
+    """
+    fields: dict[Any, Any] = {
+        vol.Required(
+            CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
+        ): int,
+        vol.Required(
+            CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
+        ): str,
+        vol.Required(
+            CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
+        ): int,
+    }
+    if admin_choices is not None:
+        # HA's supported multi-select validator: the frontend can
+        # serialize the field as a real picker of the given choices.
+        fields[
+            vol.Optional(
+                CONF_ADMIN_USER_IDS, default=list(admin_default or [])
+            )
+        ] = cv.multi_select(admin_choices)
+    return vol.Schema(fields)
 
 
 class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
     """Handle the options flow for NestQuest."""
 
+    def _async_database(self):
+        """Return the runtime database for this entry, or None.
+
+        Real HA only opens the options flow for an existing entry, so
+        the runtime record (and its database) exists; the None branch
+        covers direct construction in tests and degraded states.
+        """
+        hass_data = getattr(self.hass, "data", None)
+        if not isinstance(hass_data, dict):
+            return None
+        record = hass_data.get(DOMAIN, {}).get(self.config_entry.entry_id)
+        database = getattr(record, "database", None)
+        return database if getattr(database, "connected", False) else None
+
+    async def _async_admin_choices(self) -> dict[str, str]:
+        """Map HA user id -> display name for the picker.
+
+        Real Home Assistant always provides ``hass.auth``; a missing
+        one (direct construction in tests, degraded harness) degrades
+        to an empty picker rather than crashing the form.
+        """
+        auth = getattr(self.hass, "auth", None)
+        if auth is None:
+            return {}
+        users = await auth.async_get_users()
+        return {
+            user.id: user.name
+            for user in users
+            if getattr(user, "id", None)
+        }
+
+    async def _async_current_admin_ids(self) -> list[str]:
+        """Return the current allowlist, or [] without a database."""
+        database = self._async_database()
+        if database is None:
+            return []
+        return await list_admin_ids(database)
+
     async def async_step_init(
         self, user_input: dict[str, Any] | None = None
     ) -> FlowResult:
         """Manage the NestQuest options."""
+        admin_choices = await self._async_admin_choices()
+        admin_default = await self._async_current_admin_ids()
         if user_input is None:
             return self.async_show_form(
-                step_id="init", data_schema=_build_schema(self._current_values())
+                step_id="init",
+                data_schema=_build_schema(
+                    self._current_values(),
+                    admin_choices=admin_choices,
+                    admin_default=admin_default,
+                ),
             )
 
         errors = _async_validate(user_input)
+        # The admin ids are validated only when the key is present:
+        # schema-validated submits always carry it (vol.Optional's
+        # default fills from the current allowlist), while raw callers
+        # omitting the key keep the field untouched.
+        admin_ids_present = CONF_ADMIN_USER_IDS in user_input
+        admin_ids = user_input.get(CONF_ADMIN_USER_IDS)
         if errors:
             return self.async_show_form(
                 step_id="init",
-                data_schema=_build_schema(self._current_values()),
+                data_schema=_build_schema(
+                    self._current_values(),
+                    admin_choices=admin_choices,
+                    admin_default=admin_default,
+                ),
+                errors=errors,
+            )
+        if admin_ids_present and (
+            not isinstance(admin_ids, list)
+            or any(not isinstance(value, str) for value in admin_ids)
+        ):
+            errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
+        elif admin_ids_present and not admin_ids:
+            # An empty allowlist fails closed: saving zero admins is
+            # refused, never silently applied.
+            errors[CONF_ADMIN_USER_IDS] = "no_admins"
+        elif admin_ids_present and any(
+            value not in admin_choices for value in admin_ids
+        ):
+            # The schema enforces membership for real HA submits; this
+            # backs the raw-input path up so an unknown id can never
+            # reach the allowlist.
+            errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
+        elif admin_ids_present and len(set(admin_ids)) != len(admin_ids):
+            # set_admin_ids refuses duplicates with ValueError; a flow
+            # submission must surface that as a field error instead of
+            # an unhandled crash.
+            errors[CONF_ADMIN_USER_IDS] = "invalid_admin"
+        if errors:
+            return self.async_show_form(
+                step_id="init",
+                data_schema=_build_schema(
+                    self._current_values(),
+                    admin_choices=admin_choices,
+                    admin_default=admin_default,
+                ),
                 errors=errors,
             )
 
-        return self.async_create_entry(
-            title="NestQuest",
-            data={
-                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],
-            },
-        )
+        # Apply the picker's selection to the allowlist when the
+        # runtime database is reachable; the selection is always
+        # stored on the entry as the disaster-recovery copy.
+        database = self._async_database()
+        if database is not None and admin_ids_present:
+            await set_admin_ids(database, admin_ids)
+        data = {
+            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],
+        }
+        if admin_ids_present:
+            data[CONF_ADMIN_USER_IDS] = admin_ids
+        elif admin_default:
+            # HA replaces entry.options wholesale, so an omitted picker
+            # field must carry the CURRENT allowlist forward: dropping
+            # it would leave only the stale config-flow copy in entry
+            # data, and a later database loss would resurrect admins
+            # the owner has since removed.
+            data[CONF_ADMIN_USER_IDS] = list(admin_default)
+        return self.async_create_entry(title="NestQuest", data=data)
 
     def _current_values(self) -> dict[str, Any]:
         """Resolve current values from entry options, then entry data, then defaults."""

custom_components/nestquest/schema.py

@@ -95,30 +95,45 @@ migration runner rebuilds a pre-D-008 table into this shape and copies
 its single-assignee column across.  The composite primary key makes
 duplicate assignment impossible at the storage layer.
 
+Quest windows (D-008): ``quest_definition_windows`` declares which day
+windows a definition spans — one row per (definition, window), the
+composite primary key making a window idempotent per definition.
+``window`` is CHECK-constrained to the three spellings ``const`` owns;
+``due_time`` optionally pins a due time inside the window and is
+caller-validated (strict HH:MM) by the DAO layer, matching the date
+policy above.  The window clock ranges live in ``const.py``, not here:
+this layer stores what the caller declares and never interprets clocks.
+
 Quest instances: ``quest_instances`` deliberately carries NO completion
 status column.  An instance's current state derives from the latest row
 in ``completion_events`` (Feature 08), so materialization (Feature 07)
 can insert instances without knowing anything about completion and the
 append-only log stays the single source of truth.  ``definition_id``,
-``due_date`` and ``child_id`` snapshot the materialization-time facts
-("why this instance exists"): reassignment and rule edits change future
-instances only and never rewrite these rows.  ``due_time`` snapshots
-the definition's optional due time at generation for the same reason —
-the instance keeps its generation-time value even if the definition is
-edited later.  ``generated_at`` is a caller-set UTC ISO-8601 timestamp
-per the module timestamp policy (no DB default, so a generation batch
-stamps one coherent time).
+``child_id``, ``due_date`` and ``window`` snapshot the
+materialization-time facts ("why this instance exists"): assignment
+edits and rule edits change future instances only and never rewrite
+these rows.  ``window`` records which declared day window produced the
+instance (D-008), CHECK-constrained to the same three spellings
+``const`` owns.  ``due_time`` snapshots the window's optional due time
+at generation for the same reason — the instance keeps its
+generation-time value even if the definition is edited later.
+``generated_at`` is a caller-set UTC ISO-8601 timestamp per the module
+timestamp policy (no DB default, so a generation batch stamps one
+coherent time).
 
-Uniqueness: the (definition_id, due_date) pair is declared as a
-table-level ``UNIQUE`` constraint rather than a separate ``CREATE
-UNIQUE INDEX`` statement.  SQLite materializes it as an implicit unique
-index (``sqlite_autoindex_quest_instances_1``) — the required "unique
-index prevents duplicate instances" — and the constraint form cannot be
-silently dropped the way a standalone index can, so the duplicate door
-stays shut.  Idempotent materialization keys on exactly this pair.  No
-secondary indexes beyond it are declared yet: history and board queries
-(Features 08/10/13) should justify their indexes as a later migration
-once their query shapes exist.
+Uniqueness: the (definition_id, child_id, due_date, window) tuple is
+declared as a table-level ``UNIQUE`` constraint rather than a separate
+``CREATE UNIQUE INDEX`` statement.  SQLite materializes it as an
+implicit unique index (``sqlite_autoindex_quest_instances_1``) — the
+required "unique index prevents duplicate instances" — and the
+constraint form cannot be silently dropped the way a standalone index
+can, so the duplicate door stays shut.  Idempotent materialization keys
+on exactly this tuple: the twice-daily case (one definition, two
+windows, one child, one date) yields two rows and the shared case (one
+definition, three assignees, one window) yields three.  No secondary
+indexes beyond it are declared yet: history and board queries (Features
+08/10/13) should justify their indexes as a later migration once their
+query shapes exist.
 
 No ``ON DELETE`` clauses are declared on any foreign key in this
 module: children and definitions are deactivated, never deleted, so a
@@ -144,11 +159,12 @@ self-describing without joins.
 Actor policy: ``actor_source`` distinguishes an authenticated HA user
 (``'user'``, with ``actor_user_id`` holding the HA user id) from a
 panel tap with no user context (``'panel'``, with ``actor_user_id``
-NULL — the tapped child profile is already ``child_id``).  A coherence
-CHECK makes the pair shape non-negotiable: ``'user'`` requires
-``actor_user_id``, ``'panel'`` forbids it.  Other actor sources are
-rejected; if an unattended system actor is ever needed, that is a
-schema migration, not a silent widening.
+NULL — the tapped child profile is ``actor_child_id``, D-008).  Two
+coherence CHECKs make both shapes non-negotiable: ``'user'`` requires
+``actor_user_id`` and forbids ``actor_child_id``; ``'panel'`` requires
+``actor_child_id`` (the tapped profile) and forbids ``actor_user_id``.
+Other actor sources are rejected; if an unattended system actor is
+ever needed, that is a schema migration, not a silent widening.
 
 On-time policy: ``was_on_time`` is required (integer 0 or 1) for
 ``completed`` events — at completion time the due date is known and the
@@ -260,6 +276,17 @@ SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL: list[str] = [
     """,
 ]
 
+SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_definition_windows (
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
+        due_time TEXT,
+        PRIMARY KEY (definition_id, window)
+    )
+    """,
+]
+
 SCHEMA_V1_PRESENCE_SCHEDULES_DDL: list[str] = [
     f"""
     CREATE TABLE IF NOT EXISTS presence_schedules (
@@ -296,43 +323,64 @@ SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
     """,
 ]
 
+#: Column body of ``quest_instances``, shared between the v1 DDL
+#: below and the instance-key rebuild migration, which must recreate
+#: the table in exactly this shape (D-008: the instance key is
+#: (definition_id, child_id, due_date, window) — one instance per
+#: assigned child per declared window per firing date).
+QUEST_INSTANCES_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+    child_id INTEGER NOT NULL REFERENCES children(id),
+    window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
+    due_date TEXT NOT NULL,
+    due_time TEXT,
+    generated_at TEXT NOT NULL,
+    UNIQUE (definition_id, child_id, due_date, window)
+)"""
+
 SCHEMA_V1_QUEST_INSTANCES_DDL: list[str] = [
-    """
-    CREATE TABLE IF NOT EXISTS quest_instances (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
-        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
-        child_id INTEGER NOT NULL REFERENCES children(id),
-        due_date TEXT NOT NULL,
-        due_time TEXT,
-        generated_at TEXT NOT NULL,
-        UNIQUE (definition_id, due_date)
-    )
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_instances {QUEST_INSTANCES_TABLE_SQL}
     """,
 ]
 
-SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
-    """
-    CREATE TABLE IF NOT EXISTS completion_events (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
-        instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
-        child_id INTEGER NOT NULL REFERENCES children(id),
-        event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
-        actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
-        actor_user_id TEXT,
-        occurred_at TEXT NOT NULL,
-        was_on_time INTEGER CHECK (
-            was_on_time IS NULL
-            OR (typeof(was_on_time) = 'integer' AND was_on_time IN (0, 1))
-        ),
-        CHECK (
-            (actor_source = 'user' AND actor_user_id IS NOT NULL)
-            OR (actor_source = 'panel' AND actor_user_id IS NULL)
-        ),
-        CHECK (
-            (event_type = 'completed' AND was_on_time IS NOT NULL)
-            OR (event_type = 'uncompleted')
-        )
+#: Column body of ``completion_events``, shared between the v1 DDL
+#: below and the actor-child rebuild migration, which must recreate the
+#: table in exactly this shape (D-008: ``actor_child_id`` records which
+#: panel profile was tapped — set for panel events, NULL for admin
+#: ones; legacy panel rows backfill to ``child_id``, the child they
+#: always were).
+COMPLETION_EVENTS_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
+    child_id INTEGER NOT NULL REFERENCES children(id),
+    event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
+    actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
+    actor_user_id TEXT,
+    actor_child_id INTEGER REFERENCES children(id),
+    occurred_at TEXT NOT NULL,
+    was_on_time INTEGER CHECK (
+        was_on_time IS NULL
+        OR (typeof(was_on_time) = 'integer' AND was_on_time IN (0, 1))
+    ),
+    CHECK (
+        (actor_source = 'user' AND actor_user_id IS NOT NULL)
+        OR (actor_source = 'panel' AND actor_user_id IS NULL)
+    ),
+    CHECK (
+        (actor_source = 'user' AND actor_child_id IS NULL)
+        OR (actor_source = 'panel' AND actor_child_id IS NOT NULL)
+    ),
+    CHECK (
+        (event_type = 'completed' AND was_on_time IS NOT NULL)
+        OR (event_type = 'uncompleted')
     )
+)"""
+
+SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS completion_events {COMPLETION_EVENTS_TABLE_SQL}
     """,
 ]
 
@@ -342,6 +390,7 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_SCHEDULE_RULES_DDL,
     *SCHEMA_V1_QUEST_DEFINITIONS_DDL,
     *SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    *SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
     *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     *SCHEMA_V1_QUEST_INSTANCES_DDL,

custom_components/nestquest/strings.json

@@ -22,13 +22,16 @@
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
-          "panel_idle_timeout": "Panel idle timeout (seconds)"
+          "panel_idle_timeout": "Panel idle timeout (seconds)",
+          "admin_user_ids": "Admin users (parents)"
         }
       }
     },
     "errors": {
       "invalid": "Invalid value.",
-      "invalid_time": "Time must use 24-hour HH:MM format."
+      "invalid_time": "Time must use 24-hour HH:MM format.",
+      "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
+      "invalid_admin": "Invalid admin selection."
     }
   }
-}
+}

custom_components/nestquest/translations/en.json

@@ -22,13 +22,16 @@
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
-          "panel_idle_timeout": "Panel idle timeout (seconds)"
+          "panel_idle_timeout": "Panel idle timeout (seconds)",
+          "admin_user_ids": "Admin users (parents)"
         }
       }
     },
     "errors": {
       "invalid": "Invalid value.",
-      "invalid_time": "Time must use 24-hour HH:MM format."
+      "invalid_time": "Time must use 24-hour HH:MM format.",
+      "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
+      "invalid_admin": "Invalid admin selection."
     }
   }
-}
+}

tests/conftest.py

@@ -29,7 +29,7 @@ import tempfile
 import weakref
 from pathlib import Path
 from types import SimpleNamespace
-from unittest.mock import MagicMock
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
@@ -42,6 +42,8 @@ _HA_MODULES = (
     "homeassistant.config_entries",
     "homeassistant.data_entry_flow",
     "homeassistant.exceptions",
+    "homeassistant.helpers",
+    "homeassistant.helpers.config_validation",
 )
 
 def _find_spec(name: str):
@@ -79,6 +81,8 @@ for _parent, _child in (
     ("homeassistant", "data_entry_flow"),
     ("homeassistant", "core"),
     ("homeassistant", "exceptions"),
+    ("homeassistant", "helpers"),
+    ("homeassistant.helpers", "config_validation"),
 ):
     setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
 
@@ -181,6 +185,30 @@ class ConfigEntryNotReady(Exception):
 
 _exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
 
+import voluptuous as vol
+
+
+def _multi_select(choices: dict):
+    """Mirror homeassistant.helpers.config_validation.multi_select.
+
+    The options flow's admin picker uses HA's supported multi-select
+    validator (schema-serializable by HA's frontend); the mock provides
+    the same contract: a list whose members are keys of ``choices``.
+    """
+
+    def _validate(selected):
+        if not isinstance(selected, list):
+            raise vol.Invalid("Not a list")
+        for value in selected:
+            if value not in choices:
+                raise vol.Invalid(f"{value} is not a valid option")
+        return selected
+
+    return _validate
+
+
+_ha_mock("homeassistant.helpers.config_validation").multi_select = _multi_select
+
 
 # ---------------------------------------------------------------------------
 # Shared helpers for the behavior tests.
@@ -218,12 +246,18 @@ def make_config_entry(
     entry_id: str = "test_entry",
     options: dict | None = None,
     data: dict | None = None,
+    context: dict | None = None,
 ):
-    """Create a stub config entry shaped like HA's ConfigEntry for tests."""
+    """Create a stub config entry shaped like HA's ConfigEntry for tests.
+
+    ``context`` mirrors HA's entry context (the flow that created the
+    entry carries the acting HA user under ``context["user_id"]``).
+    """
     entry = SimpleNamespace(
         entry_id=entry_id,
         options=dict(options or {}),
         data=dict(data or {}),
+        context=dict(context or {}),
     )
     entry.add_update_listener = None
     return entry
@@ -247,6 +281,10 @@ def make_hass() -> tuple:
         config=MagicMock(),
         config_entries=SimpleNamespace(async_reload=None),
     )
+    # hass.auth mirrors the real surface the integration reads: an
+    # async_get_users list the tests can repoint at will (the setup
+    # path resolves HA owner accounts as the last-resort admin seed).
+    hass.auth = SimpleNamespace(async_get_users=AsyncMock(return_value=[]))
 
     def _cleanup_config_dir():
         import shutil
@@ -302,8 +340,15 @@ async def hass():
 def make_entry():
     """Provide the config-entry factory as a fixture."""
 
-    def _factory(entry_id: str = "test_entry", options: dict | None = None, data: dict | None = None):
-        return make_config_entry(entry_id=entry_id, options=options, data=data)
+    def _factory(
+        entry_id: str = "test_entry",
+        options: dict | None = None,
+        data: dict | None = None,
+        context: dict | None = None,
+    ):
+        return make_config_entry(
+            entry_id=entry_id, options=options, data=data, context=context
+        )
 
     return _factory
 

tests/test_admin_allowlist.py

@@ -0,0 +1,471 @@
+"""Tests for admin_allowlist.py: the Feature 03 allowlist business layer."""
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from custom_components.nestquest import async_setup_entry, async_unload_entry
+from custom_components.nestquest.admin_allowlist import (
+    list_admin_ids,
+    seed_setup_admin,
+    set_admin_ids,
+)
+from custom_components.nestquest.dao_children import AdminUsersDao
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            database = NestQuestDatabase(_make_hass_mock())
+            await database.open(tmp_path / name)
+            try:
+                await apply_migrations(database)
+                return await body(database)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# set_admin_ids: read/write with the fail-closed guarantee
+# ---------------------------------------------------------------------------
+
+
+def test_set_admin_ids_round_trip(tmp_path) -> None:
+    async def _body(database):
+        await set_admin_ids(database, ["user-owner", "user-parent"])
+        # Membership is the contract; AdminUsersDao.list orders by id.
+        assert sorted(await list_admin_ids(database)) == [
+            "user-owner",
+            "user-parent",
+        ]
+        # Replace: the removed id is gone, the new one present.
+        await set_admin_ids(database, ["user-parent", "user-grandma"])
+        assert sorted(await list_admin_ids(database)) == [
+            "user-grandma",
+            "user-parent",
+        ]
+        return await list_admin_ids(database)
+
+    _with_db(tmp_path, "set-round-trip.db")(_body)
+
+
+def test_set_admin_ids_rejects_empty_allowlist(tmp_path) -> None:
+    """Removing the last admin (or saving an empty selection) is
+    refused with a clear error: an empty allowlist fails closed, so
+    the owner must never be able to reach one through this API."""
+    async def _body(database):
+        await set_admin_ids(database, ["user-owner"])
+        with pytest.raises(ValueError, match="at least one admin"):
+            await set_admin_ids(database, [])
+        # The refused write left the allowlist intact.
+        assert await list_admin_ids(database) == ["user-owner"]
+        return None
+
+    _with_db(tmp_path, "set-empty-refused.db")(_body)
+
+
+@pytest.mark.parametrize(
+    "bad", [None, "notalist", [42], [None], [""], ["  "], 5]
+)
+def test_set_admin_ids_rejects_malformed_input(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError):
+            await set_admin_ids(database, bad)
+        assert await list_admin_ids(database) == []
+        return None
+
+    _with_db(tmp_path, "set-malformed.db")(_body)
+
+
+def test_set_admin_ids_rejects_duplicates(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="duplicates"):
+            await set_admin_ids(database, ["user-a", "user-a"])
+        assert await list_admin_ids(database) == []
+        return None
+
+    _with_db(tmp_path, "set-duplicates.db")(_body)
+
+
+def test_set_admin_ids_trims_and_replaces_atomically(tmp_path) -> None:
+    async def _body(database):
+        await set_admin_ids(database, ["  user-a  "])
+        assert await list_admin_ids(database) == ["user-a"]
+        # Sync that both adds and removes in one write.
+        await set_admin_ids(database, ["user-b", "user-c"])
+        assert await list_admin_ids(database) == ["user-b", "user-c"]
+        return None
+
+    _with_db(tmp_path, "set-trim.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# seed_setup_admin: the owner is never locked out on first setup
+# ---------------------------------------------------------------------------
+
+
+def test_seed_adds_context_user_on_first_setup(tmp_path) -> None:
+    async def _body(database):
+        seeded = await seed_setup_admin(
+            database, stored_admin_ids=None, context_user_id="owner-1"
+        )
+        assert seeded == ["owner-1"]
+        assert await list_admin_ids(database) == ["owner-1"]
+        return seeded
+
+    _with_db(tmp_path, "seed-context.db")(_body)
+
+
+def test_seed_prefers_stored_entry_ids_over_context(tmp_path) -> None:
+    async def _body(database):
+        seeded = await seed_setup_admin(
+            database,
+            stored_admin_ids=["user-a", "user-b"],
+            context_user_id="owner-1",
+        )
+        assert seeded == ["user-a", "user-b"]
+        return seeded
+
+    _with_db(tmp_path, "seed-stored.db")(_body)
+
+
+def test_seed_leaves_nonempty_allowlist_untouched(tmp_path) -> None:
+    """Re-running setup must never resurrect a deliberately narrowed
+    list: only an EMPTY allowlist is seeded."""
+    async def _body(database):
+        await set_admin_ids(database, ["user-kept"])
+        seeded = await seed_setup_admin(
+            database,
+            stored_admin_ids=["user-a", "user-b"],
+            context_user_id="owner-1",
+        )
+        assert seeded == ["user-kept"]
+        assert await list_admin_ids(database) == ["user-kept"]
+        return seeded
+
+    _with_db(tmp_path, "seed-narrowed.db")(_body)
+
+
+def test_seed_with_nothing_available_leaves_empty(tmp_path) -> None:
+    async def _body(database):
+        seeded = await seed_setup_admin(
+            database, stored_admin_ids=None, context_user_id=None
+        )
+        assert seeded == []
+        return seeded
+
+    _with_db(tmp_path, "seed-nothing.db")(_body)
+
+
+def test_seed_rejects_malformed_candidates(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError):
+            await seed_setup_admin(
+                database, stored_admin_ids=[42], context_user_id=None
+            )
+        with pytest.raises(ValueError):
+            await seed_setup_admin(
+                database, stored_admin_ids=None, context_user_id="  "
+            )
+        assert await list_admin_ids(database) == []
+        return None
+
+    _with_db(tmp_path, "seed-malformed.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Setup wiring: async_setup_entry seeds automatically
+# ---------------------------------------------------------------------------
+
+
+async def _setup_entry(hass, entry, registry):
+    from tests.test_lifecycle import _wire
+
+    entry = _wire(entry, registry)
+    await async_setup_entry(hass, entry)
+    return entry
+
+
+async def test_setup_seeds_allowlist_from_entry_context(
+    hass, make_entry
+) -> None:
+    entry = await _setup_entry(
+        hass,
+        make_entry(context={"user_id": "ha-owner"}),
+        hass.registry,
+    )
+    database = entry.runtime_data.database
+    try:
+        assert await list_admin_ids(database) == ["ha-owner"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_seeds_allowlist_from_stored_entry_data(
+    hass, make_entry
+) -> None:
+    """A lost database re-seeded from the entry's stored admin copy."""
+    from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
+
+    entry = await _setup_entry(
+        hass,
+        make_entry(data={CONF_ADMIN_USER_IDS: ["user-a", "user-b"]}),
+        hass.registry,
+    )
+    database = entry.runtime_data.database
+    try:
+        assert await list_admin_ids(database) == ["user-a", "user-b"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_without_context_leaves_allowlist_empty(
+    hass, make_entry
+) -> None:
+    """Fail closed: no context user and no stored copy means an empty
+    allowlist — nobody is admin until the owner configures it."""
+    entry = await _setup_entry(hass, make_entry(), hass.registry)
+    database = entry.runtime_data.database
+    try:
+        assert await list_admin_ids(database) == []
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_does_not_reseed_narrowed_allowlist(
+    hass, make_entry
+) -> None:
+    from custom_components.nestquest import async_unload_entry
+    from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
+
+    entry = await _setup_entry(
+        hass, make_entry(context={"user_id": "ha-owner"}), hass.registry
+    )
+    database = entry.runtime_data.database
+    await set_admin_ids(database, ["user-parent"])
+    await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+    # Reload with the SAME context: the narrowed list survives.
+    entry_reloaded = await _setup_entry(
+        hass, make_entry(context={"user_id": "ha-owner"}), hass.registry
+    )
+    database_b = entry_reloaded.runtime_data.database
+    try:
+        assert await list_admin_ids(database_b) == ["user-parent"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry_reloaded) is True
+
+
+async def test_setup_seed_prefers_options_over_entry_data(
+    hass, make_entry
+) -> None:
+    """Seed precedence on a fresh database: the options flow's saved
+    list (the current one) beats the config flow's original copy."""
+    from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
+
+    entry = await _setup_entry(
+        hass,
+        make_entry(
+            options={CONF_ADMIN_USER_IDS: ["user-current"]},
+            data={CONF_ADMIN_USER_IDS: ["ha-owner"]},
+            context={"user_id": "ha-owner"},
+        ),
+        hass.registry,
+    )
+    database = entry.runtime_data.database
+    try:
+        assert await list_admin_ids(database) == ["user-current"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_owner_fallback_seeds_owner_accounts(
+    hass, make_entry
+) -> None:
+    """Last-resort seed: with no persisted copy and no flow-context
+    user, HA owner accounts are seeded so first setup is never
+    ownerless (the real flow may lose the context; owners can already
+    do everything in HA)."""
+    from unittest.mock import AsyncMock
+
+    hass.auth.async_get_users = AsyncMock(
+        return_value=[
+            SimpleNamespace(id="owner-1", is_owner=True),
+            SimpleNamespace(id="member-1", is_owner=False),
+            SimpleNamespace(id="owner-2", is_owner=True),
+        ]
+    )
+    entry = await _setup_entry(hass, make_entry(), hass.registry)
+    database = entry.runtime_data.database
+    try:
+        assert sorted(await list_admin_ids(database)) == ["owner-1", "owner-2"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+async def test_setup_owner_fallback_not_used_when_context_present(
+    hass, make_entry
+) -> None:
+    """The context user (flow caller) wins over the owner fallback:
+    a non-owner parent configuring the integration becomes the admin."""
+    from unittest.mock import AsyncMock
+
+    hass.auth.async_get_users = AsyncMock(
+        return_value=[SimpleNamespace(id="owner-1", is_owner=True)]
+    )
+    entry = await _setup_entry(
+        hass,
+        make_entry(context={"user_id": "parent-configured"}),
+        hass.registry,
+    )
+    database = entry.runtime_data.database
+    try:
+        assert await list_admin_ids(database) == ["parent-configured"]
+    finally:
+        await database.close()
+    assert await async_unload_entry(hass, entry) is True
+
+
+def test_seed_owner_ids_argument(tmp_path) -> None:
+    async def _body(database):
+        # Malformed owner lists are rejected like any other candidate —
+        # while the allowlist is still empty, so the seed proceeds far
+        # enough to validate.
+        with pytest.raises(ValueError):
+            await seed_setup_admin(
+                database,
+                stored_admin_ids=None,
+                context_user_id=None,
+                owner_ids=[7],
+            )
+        assert await list_admin_ids(database) == []
+        seeded = await seed_setup_admin(
+            database,
+            stored_admin_ids=None,
+            context_user_id=None,
+            owner_ids=["owner-1"],
+        )
+        assert seeded == ["owner-1"]
+        return seeded
+
+    _with_db(tmp_path, "seed-owner-arg.db")(_body)
+
+
+async def test_reopened_setup_failure_closes_database(
+    hass, make_entry, monkeypatch
+) -> None:
+    """A seeding failure on the reopened-record path closes the just-
+    opened connection before re-raising (no leaked handles across HA's
+    setup retries)."""
+    import pytest as pytest_module
+    from unittest.mock import AsyncMock
+
+    import custom_components.nestquest as nestquest_module
+    from custom_components.nestquest.db import NestQuestDatabase
+
+    # First: a successful setup to create the runtime record and file.
+    entry = await _setup_entry(hass, make_entry(), hass.registry)
+    await async_unload_entry(hass, entry)
+
+    # Now a re-setup whose seeding fails after the reopen.
+    async def _failing_seed(database, **kwargs):
+        raise RuntimeError("seed boom")
+
+    monkeypatch.setattr(
+        nestquest_module, "seed_setup_admin", _failing_seed, raising=True
+    )
+
+    close_calls: list[bool] = []
+    original_close = NestQuestDatabase.close
+
+    async def _spy_close(self):
+        close_calls.append(True)
+        return await original_close(self)
+
+    monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
+
+    with pytest_module.raises(RuntimeError, match="seed boom"):
+        await _setup_entry(hass, entry, hass.registry)
+    # The reopen path opened AND closed a connection; the failure is
+    # the last thing observed.
+    assert close_calls[-1] is True
+
+
+# ---------------------------------------------------------------------------
+# is_admin: the fail-closed resolver (Feature 09's verdict function)
+# ---------------------------------------------------------------------------
+
+
+def test_is_admin_verdict_matrix(tmp_path) -> None:
+    async def _body(database):
+        from custom_components.nestquest.admin_allowlist import is_admin
+
+        await set_admin_ids(database, ["user-owner"])
+        assert await is_admin(database, "user-owner") is True
+        assert await is_admin(database, "user-other") is False
+        # Unknown, malformed, and absent identities all fail closed.
+        assert await is_admin(database, "") is False
+        assert await is_admin(database, "  user-owner  ") is False, (
+            "ids are stored trimmed; an untrimmed lookup is a different "
+            "id and must not match"
+        )
+        assert await is_admin(database, None) is False
+        assert await is_admin(database, 42) is False
+        assert await is_admin(database, ["user-owner"]) is False
+        assert await is_admin(database, True) is False
+        return None
+
+    _with_db(tmp_path, "resolver-matrix.db")(_body)
+
+
+def test_is_admin_empty_allowlist_fails_closed(tmp_path) -> None:
+    """An empty allowlist returns False for EVERYONE, not True."""
+    async def _body(database):
+        from custom_components.nestquest.admin_allowlist import is_admin
+
+        for candidate in ("user-owner", "", None, 1):
+            assert await is_admin(database, candidate) is False
+        return None
+
+    _with_db(tmp_path, "resolver-empty.db")(_body)
+
+
+def test_is_admin_false_after_last_admin_removal_attempt(tmp_path) -> None:
+    """The refusal keeps the verdict sound: a refused removal leaves
+    the admin in place, so is_admin still resolves True."""
+    async def _body(database):
+        from custom_components.nestquest.admin_allowlist import is_admin
+
+        await set_admin_ids(database, ["user-owner"])
+        with pytest.raises(ValueError, match="at least one admin"):
+            await set_admin_ids(database, [])
+        assert await is_admin(database, "user-owner") is True
+        return None
+
+    _with_db(tmp_path, "resolver-refusal.db")(_body)

tests/test_children_registry.py

@@ -0,0 +1,842 @@
+"""Tests for children.py: the Feature 03 child-profile business layer."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from custom_components.nestquest.children import (
+    create_child,
+    edit_child,
+    list_children,
+    reorder_children,
+    set_child_active,
+)
+from custom_components.nestquest.dao_children import ChildRecord, ChildrenDao
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            database = NestQuestDatabase(_make_hass_mock())
+            await database.open(tmp_path / name)
+            try:
+                await apply_migrations(database)
+                return await body(database)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# create_child: required, trimmed display name
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_returns_typed_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database,
+            "  Declan  ",
+            colour="#2E7D32",
+            avatar_ref="avatars/declan.png",
+            sort_order=2,
+        )
+        assert isinstance(child, ChildRecord)
+        # The name is trimmed before storage.
+        assert child.display_name == "Declan"
+        assert child.colour == "#2E7D32"
+        assert child.avatar_ref == "avatars/declan.png"
+        assert child.sort_order == 2
+        assert child.is_active is True
+        # The business layer stamps created_at in strict UTC shape.
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", child.created_at
+        )
+        return child
+
+    _with_db(tmp_path, "create-typed.db")(_body)
+
+
+def test_create_child_defaults(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Chloe")
+        assert child.display_name == "Chloe"
+        assert child.colour is None
+        assert child.avatar_ref is None
+        assert child.sort_order == 0
+        return child
+
+    _with_db(tmp_path, "create-defaults.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["", "   ", "\t\n", None])
+def test_create_child_missing_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="display_name is required"):
+            await create_child(database, bad)
+        assert await list_children(database) == []
+        return None
+
+    _with_db(tmp_path, "create-empty-name.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [42, ["Ada"], 3.5])
+def test_create_child_non_string_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="must be a string"):
+            await create_child(database, bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-name-type.db")(_body)
+
+
+def test_create_child_optional_text_fields_validated(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="colour must be a string"):
+            await create_child(database, "Ada", colour=7)
+        with pytest.raises(ValueError, match="avatar_ref must be a string"):
+            await create_child(database, "Ada", avatar_ref=1.5)
+        return None
+
+    _with_db(tmp_path, "create-bad-optional.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["2", 1.5, True, None])
+def test_create_child_sort_order_must_be_int(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await create_child(database, "Ada", sort_order=bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# create_child: duplicate display name allowed but warned
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_duplicate_name_allowed_and_warned(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        with caplog.at_level(
+            logging.WARNING, logger="custom_components.nestquest.children"
+        ):
+            second = await create_child(database, "  ADA ")
+        # Whitespace/case-normalised duplicate: allowed, both rows exist.
+        assert first.display_name == "Ada"
+        assert second.display_name == "ADA"
+        names = [c.display_name for c in await list_children(database)]
+        assert names == ["Ada", "ADA"]
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1, (
+            "exactly one duplicate warning for the second create"
+        )
+        assert "Ada" in warnings[0].getMessage()
+        return second
+
+    _with_db(tmp_path, "create-duplicate.db")(_body)
+
+
+def test_create_child_distinct_name_logs_no_warning(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        await create_child(database, "Ada")
+        caplog.clear()
+        await create_child(database, "Jordyn")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert warnings == []
+        return None
+
+    _with_db(tmp_path, "create-no-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_child: returns the updated record
+# ---------------------------------------------------------------------------
+
+
+def test_edit_child_updates_fields_and_returns_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(
+            database,
+            child.id,
+            display_name="  Ada L. ",
+            colour="#222222",
+            avatar_ref="avatars/ada.png",
+            sort_order=3,
+        )
+        assert isinstance(edited, ChildRecord)
+        assert edited.id == child.id
+        assert edited.display_name == "Ada L."
+        assert edited.colour == "#222222"
+        assert edited.avatar_ref == "avatars/ada.png"
+        assert edited.sort_order == 3
+        # The stored row reflects every edit.
+        stored = await list_children(database)
+        assert len(stored) == 1
+        assert stored[0] == edited
+        return edited
+
+    _with_db(tmp_path, "edit-round-trip.db")(_body)
+
+
+def test_edit_child_omitted_fields_unchanged(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database, "Ada", colour="#111111", sort_order=5
+        )
+        edited = await edit_child(database, child.id, sort_order=9)
+        # Only sort_order moved; colour and name untouched.
+        assert edited.sort_order == 9
+        assert edited.colour == "#111111"
+        assert edited.display_name == "Ada"
+        return edited
+
+    _with_db(tmp_path, "edit-partial.db")(_body)
+
+
+def test_edit_child_no_fields_returns_current_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(database, child.id)
+        assert edited == child
+        return edited
+
+    _with_db(tmp_path, "edit-noop.db")(_body)
+
+
+def test_edit_child_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await edit_child(database, 999, display_name="X")
+        return None
+
+    _with_db(tmp_path, "edit-missing.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.0, "1", None])
+def test_edit_child_non_int_id_rejected(tmp_path, bad) -> None:
+    """SQLite binds bools as integers: True must not address child 1,
+    and floats must not round onto a real profile."""
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await edit_child(database, bad, display_name="X")
+        # The refused edit left the profile untouched.
+        assert (await list_children(database))[0].display_name == "Ada"
+        return None
+
+    _with_db(tmp_path, "edit-bad-id.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [True, False, 1.0, "1", None])
+def test_set_child_active_non_int_id_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await set_child_active(database, bad, False)
+        assert (await list_children(database))[0].is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-bad-id.db")(_body)
+
+
+def test_edit_child_name_validation(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        for bad in ("", "   ", None, 42):
+            with pytest.raises(ValueError, match="display_name"):
+                await edit_child(database, child.id, display_name=bad)
+        # The failed edits left the record untouched.
+        assert (await list_children(database))[0].display_name == "Ada"
+        return None
+
+    _with_db(tmp_path, "edit-name-validation.db")(_body)
+
+
+def test_edit_child_sort_order_must_be_int(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await edit_child(database, child.id, sort_order="3")
+        return None
+
+    _with_db(tmp_path, "edit-bad-order.db")(_body)
+
+
+def test_edit_child_explicit_none_clear_rejected(tmp_path) -> None:
+    """Clearing is refused loudly, not silently ignored: the DAO cannot
+    express NULL writes, so an explicit None must never masquerade as a
+    no-op."""
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        with pytest.raises(ValueError, match="clearing colour"):
+            await edit_child(database, child.id, colour=None)
+        with pytest.raises(ValueError, match="clearing avatar_ref"):
+            await edit_child(database, child.id, avatar_ref=None)
+        # The refused clears left the colour intact.
+        assert (await list_children(database))[0].colour == "#111111"
+        return None
+
+    _with_db(tmp_path, "edit-clear-unsupported.db")(_body)
+
+
+def test_edit_child_rename_to_duplicate_warns_but_different_name_not(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        bo = await create_child(database, "Bo")
+        caplog.clear()
+        # Renaming Bo onto Ada's name warns (different child's name).
+        edited = await edit_child(database, bo.id, display_name="ada")
+        assert edited.display_name == "ada"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert "Ada" in warnings[0].getMessage()
+        return edited
+
+    _with_db(tmp_path, "edit-rename-duplicate.db")(_body)
+
+
+def test_edit_child_rename_to_own_name_casing_alone_warns_not(
+    tmp_path, caplog
+) -> None:
+    """Renaming a child to their own name (casing aside) is not a
+    duplicate when NO OTHER child carries the normalised name."""
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        caplog.clear()
+        edited = await edit_child(database, ada.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        assert not any(
+            "duplicates existing child" in r.getMessage()
+            for r in caplog.records
+        )
+        return edited
+
+    _with_db(tmp_path, "edit-rename-self.db")(_body)
+
+
+def test_create_child_concurrent_same_name_still_warns(tmp_path) -> None:
+    """Racing creates of the same normalised name cannot slip through
+    unlogged: the scan and INSERT are serialized on the connection
+    lock, so the second create sees the first's committed row and
+    warns.  Deterministic gate on task A's INSERT while task B queues
+    behind the lock.  Without the lock, B's scan would run before A's
+    INSERT commits and no warning would fire."""
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "create-race.db")
+        try:
+            await apply_migrations(database)
+            logger = logging.getLogger(
+                "custom_components.nestquest.children"
+            )
+            seen: list[logging.LogRecord] = []
+
+            class _Capture(logging.Handler):
+                def emit(self, record: logging.LogRecord) -> None:
+                    seen.append(record)
+
+            handler = _Capture()
+            logger.addHandler(handler)
+            try:
+                armed["active"] = True
+                task_a = asyncio.ensure_future(
+                    create_child(database, "Ada")
+                )
+                await a_started.wait()
+                armed["active"] = False
+                task_b = asyncio.ensure_future(
+                    create_child(database, " ada ")
+                )
+                await asyncio.sleep(0)
+                gate_open.set()
+                first, second = await asyncio.gather(task_a, task_b)
+            finally:
+                logger.removeHandler(handler)
+
+            # Both creates land; the duplicate is allowed.
+            assert first.display_name == "Ada"
+            assert second.display_name == "ada"
+            names = [c.display_name for c in await list_children(database)]
+            assert names == ["Ada", "ada"]
+            # Exactly one warning: B's scan saw A's committed row.
+            warnings = [
+                r for r in seen if "duplicates existing child" in r.getMessage()
+            ]
+            assert len(warnings) == 1, (
+                "the queued duplicate create must warn exactly once"
+            )
+            assert "Ada" in warnings[0].getMessage()
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_edit_child_case_only_edit_colliding_with_other_child_warns(
+    tmp_path, caplog
+) -> None:
+    """A case-only edit must not hide a real duplicate: with two Ada
+    profiles already present, editing either one's name to another
+    casing still warns (the scan excludes only the edited child)."""
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        await create_child(database, "ada")
+        caplog.clear()
+        edited = await edit_child(database, first.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        # Editing the name to a value that STAYS duplicated (no change,
+        # other child still carries it) warns too: the scan is on the
+        # resulting state, not on whether the word moved.
+        caplog.clear()
+        await edit_child(database, first.id, display_name=" ada ")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        return None
+
+    _with_db(tmp_path, "edit-case-only-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# set_child_active / list_children
+# ---------------------------------------------------------------------------
+
+
+def test_set_child_active_round_trip(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert [c.id for c in await list_children(database)] == [child.id]
+        assert await list_children(database, active_only=True) == []
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert len(await list_children(database, active_only=True)) == 1
+        return reactivated
+
+    _with_db(tmp_path, "set-active.db")(_body)
+
+
+def test_set_child_active_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await set_child_active(database, 999, False)
+        return None
+
+    _with_db(tmp_path, "set-active-missing.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# reorder_children: complete permutation or nothing changes
+# ---------------------------------------------------------------------------
+
+
+def test_reorder_children_persists_display_order(tmp_path) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        third = await create_child(database, "Cleo", sort_order=2)
+        await reorder_children(database, [third.id, first.id, second.id])
+        # The active list returns children in the new sort order.
+        listed = await list_children(database, active_only=True)
+        assert [c.id for c in listed] == [third.id, first.id, second.id]
+        assert [c.sort_order for c in listed] == [0, 1, 2]
+        return listed
+
+    _with_db(tmp_path, "reorder-round-trip.db")(_body)
+
+
+def test_reorder_children_partial_list_rejected_order_unchanged(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        third = await create_child(database, "Cleo", sort_order=2)
+        with pytest.raises(ValueError, match="missing child ids"):
+            await reorder_children(database, [third.id, first.id])
+        listed = await list_children(database)
+        # Nothing changed: rejection happens before any write.
+        assert [c.id for c in listed] == [first.id, second.id, third.id]
+        assert [c.sort_order for c in listed] == [0, 1, 2]
+        return listed
+
+    _with_db(tmp_path, "reorder-partial.db")(_body)
+
+
+def test_reorder_children_unknown_id_rejected_order_unchanged(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        with pytest.raises(ValueError, match="unknown child ids"):
+            await reorder_children(database, [second.id, first.id, 999])
+        listed = await list_children(database)
+        assert [c.id for c in listed] == [first.id, second.id]
+        assert [c.sort_order for c in listed] == [0, 1]
+        return listed
+
+    _with_db(tmp_path, "reorder-unknown.db")(_body)
+
+
+def test_reorder_children_empty_list_rejected_when_children_exist(
+    tmp_path,
+) -> None:
+    """An empty list is the maximally partial reorder: rejected while
+    any child exists (against an empty table it is a complete
+    permutation and stays a harmless no-op)."""
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        with pytest.raises(ValueError, match="missing child ids"):
+            await reorder_children(database, [])
+        assert [c.sort_order for c in await list_children(database)] == [0]
+        return None
+
+    _with_db(tmp_path, "reorder-empty.db")(_body)
+
+
+def test_reorder_children_duplicate_and_non_int_rejected(tmp_path) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        with pytest.raises(ValueError, match="duplicate child ids"):
+            await reorder_children(database, [first.id, first.id, second.id])
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await reorder_children(database, [first.id, second.id, True])
+        with pytest.raises(ValueError, match="child_id must be an integer"):
+            await reorder_children(database, [first.id, second.id, "3"])
+        listed = await list_children(database)
+        assert [c.sort_order for c in listed] == [0, 1]
+        return listed
+
+    _with_db(tmp_path, "reorder-bad-inputs.db")(_body)
+
+
+def test_reorder_children_inactive_children_included(tmp_path) -> None:
+    """Inactive children keep their display position: the permutation
+    covers the whole table, not just the active list."""
+    async def _body(database):
+        first = await create_child(database, "Ada", sort_order=0)
+        second = await create_child(database, "Bo", sort_order=1)
+        await set_child_active(database, second.id, False)
+        # Both ids are required even though one is inactive.
+        await reorder_children(database, [second.id, first.id])
+        listed = await list_children(database)
+        assert [c.id for c in listed] == [second.id, first.id]
+        assert [c.sort_order for c in listed] == [0, 1]
+        # The active list skips the inactive child but keeps position.
+        assert [c.id for c in await list_children(database, active_only=True)] == [
+            first.id
+        ]
+        return listed
+
+    _with_db(tmp_path, "reorder-inactive.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["1", "0", 1, 0, 1.0, None, "true"])
+def test_set_child_active_rejects_non_bool(tmp_path, bad) -> None:
+    """The DAO's int() would silently coerce "1"/0/1.0 into a state the
+    caller never asked for; the business layer refuses non-bools."""
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="is_active must be a real bool"):
+            await set_child_active(database, child.id, bad)
+        # The refused transition left the child untouched.
+        assert (await list_children(database))[0].is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-non-bool.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Concurrency: edit/set_active are atomic under the connection lock
+# ---------------------------------------------------------------------------
+
+
+def _gated_hass(armed: dict, gate_open: asyncio.Event, a_started: asyncio.Event):
+    """A hass mock whose executor gates the FIRST `_execute` job after
+    arming — the mutate statement of whichever business call starts
+    first — proving the caller holds the connection lock at that
+    moment and the second caller must queue behind it."""
+    from unittest.mock import MagicMock
+
+    hass = MagicMock()
+
+    async def _gated_executor(fn, *args):
+        if armed["active"]:
+            if getattr(fn, "__name__", "") == "_execute" and not armed["gated"]:
+                armed["gated"] = True
+                a_started.set()
+                await gate_open.wait()
+        return fn(*args)
+
+    hass.async_add_executor_job = _gated_executor
+    return hass
+
+
+def test_edit_child_concurrent_opposite_edits_return_own_values(
+    tmp_path,
+) -> None:
+    """Two racing edits cannot cross-contaminate each other's readback.
+
+    Deterministic gate: task A is paused at its UPDATE (it provably
+    holds the connection lock), task B is then created and must queue;
+    A's readback runs before B's edit, so A returns A's value and B
+    returns B's.  Without the lock, A's readback would see B's write.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "edit-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(
+                database, "Ada", colour="#111111", sort_order=0
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                edit_child(database, child.id, colour="#AAAAAA", sort_order=1)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                edit_child(
+                    database, child.id, colour="#BBBBBB", sort_order=2
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            # Each caller gets back exactly what it wrote.
+            assert record_a.colour == "#AAAAAA"
+            assert record_a.sort_order == 1
+            assert record_b.colour == "#BBBBBB"
+            assert record_b.sort_order == 2
+            # B landed last: the stored row is B's (read via the DAO —
+            # the children leak guard forbids raw probes here).
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert (stored.colour, stored.sort_order) == ("#BBBBBB", 2)
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_set_child_active_concurrent_opposite_transitions_return_own(
+    tmp_path,
+) -> None:
+    """Racing activate/deactivate calls each report their own request.
+
+    Same deterministic gate on task A's UPDATE; task B queues behind
+    the connection lock and lands the opposite transition after A.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "active-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(database, "Ada")
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                set_child_active(database, child.id, False)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                set_child_active(database, child.id, True)
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            assert record_a.is_active is False
+            assert record_b.is_active is True
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert stored.is_active is True
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_list_children_orders_by_sort_order_then_id(tmp_path) -> None:
+    async def _body(database):
+        second = await create_child(database, "Bo", sort_order=1)
+        first = await create_child(database, "Ada", sort_order=0)
+        third = await create_child(database, "Cleo", sort_order=1)
+        listed = await list_children(database)
+        # sort_order 0 first, then the two sort_order-1 rows by id.
+        assert [c.id for c in listed] == [first.id, second.id, third.id]
+        return listed
+
+    _with_db(tmp_path, "list-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Deactivation keeps every referencing row intact (Feature 03)
+# ---------------------------------------------------------------------------
+
+
+def test_deactivating_child_preserves_definitions_instances_events(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        from datetime import date, timedelta
+
+        from custom_components.nestquest.dao_children import ChildrenDao
+        from custom_components.nestquest.dao_instances import (
+            CompletionEventsDao,
+            QuestInstancesDao,
+        )
+        from custom_components.nestquest.dao_rules import (
+            QuestDefinitionsDao,
+            ScheduleRulesDao,
+        )
+
+        children = ChildrenDao(database)
+        rules = ScheduleRulesDao(database)
+        definitions = QuestDefinitionsDao(database)
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+
+        child = await create_child(database, "Ada")
+        rule = await rules.create("daily", "2026-01-01")
+        definition = await definitions.create(
+            "Brush teeth",
+            rule.id,
+            "2026-09-15T12:00:00+00:00",
+            assignee_child_ids=[child.id],
+        )
+        due = (date.today() + timedelta(days=1)).isoformat()
+        stamp = f"{due}T08:00:00+00:00"
+        instance = await instances.upsert(
+            definition.id, child.id, due, stamp, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "panel",
+            f"{due}T09:00:00+00:00",
+            True,
+            actor_child_id=child.id,
+        )
+
+        # Deactivate: the referencing rows must not move.
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert await children.get(child.id) == deactivated
+        assert (await definitions.get(definition.id)) == definition
+        assert (await instances.get_by_id(instance.id)) == instance
+        assert await events.list_by_instance(instance.id) == [event]
+        # Excluded from the active list, still present in the full list.
+        assert await list_children(database, active_only=True) == []
+        assert [c.id for c in await list_children(database)] == [child.id]
+
+        # Reactivate: fully restored.
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert [c.id for c in await list_children(database, active_only=True)] == [
+            child.id
+        ]
+        assert (await instances.get_by_id(instance.id)) == instance
+        return reactivated
+
+    _with_db(tmp_path, "deactivate-data-intact.db")(_body)
+
+
+def test_no_code_path_deletes_a_child_row() -> None:
+    """Structural guardrail: children are deactivated, never deleted.
+
+    The DAO exposes no delete method at all, and no module in the
+    package (nor any test outside the sanctioned DDL/migration layers,
+    which create the tables) carries a statement that removes child
+    rows — the leak guard in test_dao_children enforces the file
+    allowlist, this asserts the no-delete property itself across every
+    file that may legitimately name the table.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+
+    delete_pattern = re.compile(
+        r"DELETE\s+(FROM\s+)?((main|temp)\s*\.\s*)?[`'\"]*(\[)?children\b",
+        re.IGNORECASE,
+    )
+    for py in sorted(package.rglob("*.py")):
+        assert delete_pattern.search(py.read_text()) is None, (
+            f"a code path in {py.name} deletes child rows"
+        )
+
+    import inspect
+
+    from custom_components.nestquest.dao_children import ChildrenDao
+
+    methods = {
+        name
+        for name, _ in inspect.getmembers(ChildrenDao, inspect.isfunction)
+    }
+    assert not any(
+        name.startswith("delete") or name == "remove" for name in methods
+    ), f"ChildrenDao exposed a delete-shaped method: {methods}"

tests/test_config_flow.py

@@ -154,4 +154,25 @@ def test_no_hardcoded_domain_or_db_filename_in_new_modules() -> None:
                     )
                 assert node.value != "nestquest.db", (
                     f"Found hard-coded database filename in {py_file.name}:{node.lineno}"
-                )
+                )
+
+async def test_flow_persists_flow_user_as_initial_admin(hass, make_flow) -> None:
+    """The HA user completing the flow lands in entry data as the
+    initial admin copy: entry.context is not persisted by real HA, so
+    the durable record must carry the owner."""
+    from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
+
+    flow = make_flow(hass)
+    flow.context = {"user_id": "ha-owner"}
+    result = await flow.async_step_user({})
+    assert result["type"] == "create_entry"
+    assert result["data"] == {CONF_ADMIN_USER_IDS: ["ha-owner"]}
+
+
+async def test_flow_without_context_user_keeps_data_untouched(
+    hass, make_flow
+) -> None:
+    flow = make_flow(hass)
+    result = await flow.async_step_user({"horizon_days": 7})
+    assert result["type"] == "create_entry"
+    assert result["data"] == {"horizon_days": 7}

tests/test_const.py

@@ -73,6 +73,36 @@ def test_panel_idle_timeout_config() -> None:
     assert const.DEFAULT_PANEL_IDLE_TIMEOUT > 0
 
 
+def test_quest_windows_canonical_order() -> None:
+    """QUEST_WINDOWS is exactly the three spellings in canonical order."""
+    assert const.QUEST_WINDOWS == ("morning", "afternoon", "evening")
+    assert const.WINDOW_MORNING == "morning"
+    assert const.WINDOW_AFTERNOON == "afternoon"
+    assert const.WINDOW_EVENING == "evening"
+    # No duplicates, every name covered by its individual constant.
+    assert len(set(const.QUEST_WINDOWS)) == 3
+
+
+def test_window_clock_ranges_cover_every_window() -> None:
+    """Each window has a strict HH:MM (start, end) range.
+
+    The ranges follow the documented boundaries (D-008): morning ends
+    11:59, afternoon spans 12:00-17:00, evening 17:00-21:00, and each
+    window's end equals the next window's start (the boundary belongs
+    to the later window at classification time).
+    """
+    assert const.WINDOW_CLOCK_RANGES == {
+        "morning": ("00:00", "11:59"),
+        "afternoon": ("12:00", "17:00"),
+        "evening": ("17:00", "21:00"),
+    }
+    for window in const.QUEST_WINDOWS:
+        start, end = const.WINDOW_CLOCK_RANGES[window]
+        for value in (start, end):
+            parsed = datetime.strptime(value, "%H:%M").time()
+            assert parsed.strftime("%H:%M") == value
+
+
 def test_no_hardcoded_domain_or_db_filename_in_modules() -> None:
     """Ensure no module other than const.py hardcodes DOMAIN string or SQLITE_DB_FILENAME."""
     pkg_dir = Path(const.__file__).parent

tests/test_dao_children.py

@@ -227,24 +227,34 @@ def test_reorder_is_transactional_and_idempotent(tmp_path) -> None:
         _run(database.close())
 
 
-def test_reorder_empty_list_is_noop(tmp_path) -> None:
+def test_reorder_empty_list_is_rejected(tmp_path) -> None:
+    """An empty list is the maximally partial reorder: rejected while
+    any child exists (Feature 03 reorder contract)."""
+    import pytest as pytest_module
+
     database = _open_db(tmp_path / "reorder-empty.db")
     try:
         dao, _ = _daos(database)
         _make_child(dao, "Ada")
-        _run(dao.reorder([]))
+        with pytest_module.raises(ValueError, match="missing child ids"):
+            _run(dao.reorder([]))
         children = _run(dao.list_all())
         assert children[0].sort_order == 0
     finally:
         _run(database.close())
 
 
-def test_reorder_unknown_id_leaves_it_untouched(tmp_path) -> None:
+def test_reorder_unknown_id_is_rejected(tmp_path) -> None:
+    """An unknown id is rejected before any write; the order is
+    untouched (Feature 03 reorder contract)."""
+    import pytest as pytest_module
+
     database = _open_db(tmp_path / "reorder-unknown.db")
     try:
         dao, _ = _daos(database)
         a = _make_child(dao, "Ada")
-        _run(dao.reorder([a.id, 999]))
+        with pytest_module.raises(ValueError, match="unknown child ids"):
+            _run(dao.reorder([a.id, 999]))
         children = _run(dao.list_all())
         assert children[0].sort_order == 0
     finally:

tests/test_dao_instances.py

@@ -119,12 +119,10 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         assert isinstance(first, QuestInstanceRecord)
         second = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         assert second.id == first.id
         rows = await database.fetch_one(
             "SELECT COUNT(*) FROM quest_instances"
@@ -138,27 +136,116 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
 def test_upsert_different_dates_create_distinct_rows(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        a = await instances.upsert(definition.id, child.id, D1, _now_stamp())
-        b = await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        a = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
+        b = await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         assert a.id != b.id
         return (a, b)
 
     _with_db(tmp_path, "upsert-multi-date.db")(_body)
 
 
+def test_upsert_twice_daily_fanout_two_windows_two_rows(tmp_path) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows, each idempotent on its own tuple."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        morning = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        evening = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        assert morning.id != evening.id
+        rows = await database.fetch_all(
+            "SELECT window, child_id, due_date FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ?",
+            (definition.id, D1),
+        )
+        assert sorted(rows) == [
+            ("evening", child.id, D1),
+            ("morning", child.id, D1),
+        ]
+        # Re-running both is still idempotent.
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (2,)
+        return (morning, evening)
+
+    _with_db(tmp_path, "upsert-twice-daily.db")(_body)
+
+
+def test_upsert_shared_definition_three_assignees_three_rows(
+    tmp_path,
+) -> None:
+    """D-008 shared case: one definition, three assignees, one window
+    — three rows, one per child."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        second = await children.create("Bo", _now_stamp())
+        third = await children.create("Cleo", _now_stamp())
+        await definitions.add_assignee(definition.id, second.id)
+        await definitions.add_assignee(definition.id, third.id)
+        created = [
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+            for assignee in (child, second, third)
+        ]
+        assert len({row.id for row in created}) == 3
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ? AND window = ?",
+            (definition.id, D1, "morning"),
+        )
+        assert count == (3,)
+        # Re-running the fan-out changes nothing.
+        for assignee in (child, second, third):
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (3,)
+        return created
+
+    _with_db(tmp_path, "upsert-shared-fanout.db")(_body)
+
+
+def test_upsert_window_outside_const_list_raises(tmp_path) -> None:
+    """The DAO rejects a window spelling the schema CHECK would reject
+    anyway, with the typed ValueError instead of raw IntegrityError."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        with pytest.raises(ValueError, match="window must be one of"):
+            await instances.upsert(
+                definition.id, child.id, D1, _now_stamp(), window="noon"
+            )
+        return None
+
+    _with_db(tmp_path, "upsert-bad-window.db")(_body)
+
+
 def test_upsert_refreshes_snapshot_columns_on_conflict(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="08:00"
-        )
+            definition.id, child.id, D1, _now_stamp(), due_time="08:00", window="morning")
         second = await instances.upsert(
             definition.id,
             child.id,
             D1,
             f"{D1}T13:00:00+00:00",
-            due_time="08:30",
-        )
+            due_time="08:30", window="morning")
         assert second.id == first.id
         assert second.due_time == "08:30"
         assert second.generated_at == f"{D1}T13:00:00+00:00"
@@ -172,7 +259,7 @@ def test_upsert_rejects_wrong_child_for_definition(tmp_path) -> None:
                     events, child, definition):
         other = await children.create("Bo", _now_stamp())
         with pytest.raises(ValueError, match="assigned to"):
-            await instances.upsert(definition.id, other.id, D1, _now_stamp())
+            await instances.upsert(definition.id, other.id, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-wrong-child.db")(_body)
@@ -182,7 +269,7 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         with pytest.raises(ValueError, match="definition 999"):
-            await instances.upsert(999, child.id, D1, _now_stamp())
+            await instances.upsert(999, child.id, D1, _now_stamp(), window="morning")
         # Unknown child: the definition-assignee check fires first and
         # is the one that matters (an instance must carry an assigned
         # child).  Under D-008 the assignee check is a membership test
@@ -193,7 +280,7 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
             "X", rule.id, _now_stamp(), assignee_child_ids=[child.id]
         )
         with pytest.raises(ValueError, match="not assigned to child 999"):
-            await instances.upsert(other_definition.id, 999, D1, _now_stamp())
+            await instances.upsert(other_definition.id, 999, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-unknown.db")(_body)
@@ -204,7 +291,7 @@ def test_upsert_malformed_due_date_raises(tmp_path) -> None:
                     events, child, definition):
         for bad in ("not-a-date", "2026-9-15", ""):
             with pytest.raises(ValueError, match="YYYY-MM-DD"):
-                await instances.upsert(definition.id, child.id, bad, _now_stamp())
+                await instances.upsert(definition.id, child.id, bad, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-bad-date.db")(_body)
@@ -216,11 +303,11 @@ def test_upsert_rejects_past_due_dates(tmp_path) -> None:
                     events, child, definition):
         yesterday = _future_day(-1)
         with pytest.raises(ValueError, match="never generated in the past"):
-            await instances.upsert(definition.id, child.id, yesterday, _now_stamp())
+            await instances.upsert(definition.id, child.id, yesterday, _now_stamp(), window="morning")
         # Today itself is allowed (the rollover job generates today's
         # instances before the day is over).
         today = datetime.date.today().isoformat()
-        record = await instances.upsert(definition.id, child.id, today, _now_stamp())
+        record = await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         assert record.due_date == today
         return record
 
@@ -296,7 +383,7 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
             await asyncio.wait_for(holder_inside.wait(), timeout=5)
 
             task = asyncio.ensure_future(
-                instances.upsert(definition.id, child.id, today, _now_stamp())
+                instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
             )
             # A is now queued on the connection lock (it cannot pass
             # its in-lock check while B holds the lock).  Shift the
@@ -349,17 +436,15 @@ def test_upsert_refuses_immutable_completed_instance(tmp_path) -> None:
                     events, child, definition):
         generated_at = _now_stamp()
         instance = await instances.upsert(
-            definition.id, child.id, D1, generated_at, due_time="08:00"
-        )
+            definition.id, child.id, D1, generated_at, due_time="08:00", window="morning")
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         with pytest.raises(ValueError, match="immutable"):
             await instances.upsert(
-                definition.id, child.id, D1, _now_stamp(), due_time="09:00"
-            )
-        fetched = await instances.get(definition.id, D1)
+                definition.id, child.id, D1, _now_stamp(), due_time="09:00", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         assert fetched.due_time == "08:00", (
             "failed regeneration must not touch the completed instance"
         )
@@ -378,12 +463,11 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         created = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="17:30"
-        )
-        fetched = await instances.get(definition.id, D1)
+            definition.id, child.id, D1, _now_stamp(), due_time="17:30", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         assert fetched == created
         assert fetched.due_time == "17:30"
-        assert await instances.get(definition.id, D2) is None
+        assert await instances.get(definition.id, child.id, D2, "morning") is None
         return fetched
 
     _with_db(tmp_path, "get-round-trip.db")(_body)
@@ -392,7 +476,7 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
 def test_get_by_id_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        created = await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        created = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         fetched = await instances.get_by_id(created.id)
         assert fetched == created
         assert await instances.get_by_id(999) is None
@@ -409,11 +493,10 @@ def test_list_by_child_and_date_filters(tmp_path) -> None:
         other_definition = await definitions.create(
             "Make bed", rule.id, _now_stamp(), assignee_child_ids=[other.id]
         )
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
-        await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
+        await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         await instances.upsert(
-            other_definition.id, other.id, D1, _now_stamp()
-        )
+            other_definition.id, other.id, D1, _now_stamp(), window="morning")
         mine = await instances.list_by_child_and_date(child.id, D1)
         assert len(mine) == 1
         assert mine[0].child_id == child.id
@@ -427,7 +510,7 @@ def test_list_by_date_range_filters_and_orders(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         for day in (D1, D3, D5, D20):
-            await instances.upsert(definition.id, child.id, day, _now_stamp())
+            await instances.upsert(definition.id, child.id, day, _now_stamp(), window="morning")
         found = await instances.list_by_date_range(
             child.id, D2, D6
         )
@@ -461,18 +544,17 @@ def test_list_by_date_range_rejects_inverted_range(tmp_path) -> None:
 def test_delete_future_uncompleted_removes_only_eligible(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         future_open = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
-        )
-        await instances.upsert(definition.id, child.id, D6, _now_stamp())
+            definition.id, child.id, D5, _now_stamp(), window="morning")
+        await instances.upsert(definition.id, child.id, D6, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, D5
         )
         assert deleted == 2
-        assert await instances.get(definition.id, D1) is not None
-        assert await instances.get(definition.id, D5) is None
-        assert await instances.get(definition.id, D6) is None
+        assert await instances.get(definition.id, child.id, D1, "morning") is not None
+        assert await instances.get(definition.id, child.id, D5, "morning") is None
+        assert await instances.get(definition.id, child.id, D6, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-future.db")(_body)
@@ -482,11 +564,9 @@ def test_delete_future_uncompleted_spares_completed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         completed = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
-        )
+            definition.id, child.id, D5, _now_stamp(), window="morning")
         open_one = await instances.upsert(
-            definition.id, child.id, D6, _now_stamp()
-        )
+            definition.id, child.id, D6, _now_stamp(), window="morning")
         await events.append(
             completed.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -521,12 +601,12 @@ def test_delete_future_uncompleted_cannot_backdate_cutoff(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         today = datetime.date.today().isoformat()
-        await instances.upsert(definition.id, child.id, today, _now_stamp())
+        await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, "2020-01-01"
         )
         assert deleted == 1
-        assert await instances.get(definition.id, today) is None
+        assert await instances.get(definition.id, child.id, today, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-no-backdate.db")(_body)
@@ -541,13 +621,13 @@ def test_delete_future_uncompleted_other_definitions_untouched(
             "Make bed", (await rules.create("daily", D1)).id,
             _now_stamp(), assignee_child_ids=[child.id],
         )
-        await instances.upsert(definition.id, child.id, D20, _now_stamp())
-        await instances.upsert(other.id, child.id, D20, _now_stamp())
+        await instances.upsert(definition.id, child.id, D20, _now_stamp(), window="morning")
+        await instances.upsert(other.id, child.id, D20, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, D20
         )
         assert deleted == 1
-        assert await instances.get(other.id, D20) is not None
+        assert await instances.get(other.id, child.id, D20, "morning") is not None
         return deleted
 
     _with_db(tmp_path, "delete-scoped.db")(_body)
@@ -562,8 +642,7 @@ def test_append_completion_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         event = await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -582,10 +661,10 @@ def test_append_panel_event_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         event = await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id,
             actor_user_id=None,
         )
         assert event.actor_source == "panel"
@@ -600,8 +679,7 @@ def test_append_uncompleted_with_optional_on_time(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         reversal = await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), None,
             actor_user_id="user-1",
@@ -622,8 +700,7 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         with pytest.raises(ValueError, match="event_type"):
             await events.append(
                 instance.id, child.id, "missed", "user", _now_stamp(), True,
@@ -642,17 +719,31 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
         with pytest.raises(ValueError, match="must not carry"):
             await events.append(
                 instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id,
                 actor_user_id="user-1",
             )
+        with pytest.raises(ValueError, match="requires actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), True,
+            )
+        with pytest.raises(ValueError, match="must not carry actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "user", _now_stamp(), True,
+                actor_user_id="user-1",
+                actor_child_id=child.id,
+            )
         with pytest.raises(ValueError, match="requires was_on_time"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), None
+                instance.id, child.id, "completed", "panel", _now_stamp(), None,
+                actor_child_id=child.id
             )
         # was_on_time must be a real bool: int-convertible junk that
         # would silently coerce (0.5 -> False) is rejected outright.
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), 0.5
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), 0.5, actor_child_id=child.id,
             )
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
@@ -670,8 +761,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         for bad in (
             "2026-09-14 12:00:00+00:00",  # space separator
             "2026-09-14T12:00:00",  # no offset
@@ -683,7 +773,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
             with pytest.raises(ValueError, match="UTC"):
                 await events.append(
                     instance.id, child.id, "completed", "panel", bad,
-                    True,
+                    True, actor_child_id=child.id,
                 )
         return None
 
@@ -695,8 +785,7 @@ def test_append_requires_was_on_time_strict_bool(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         for bad in (0.5, 1.5, 2, "true", 1):
             with pytest.raises(ValueError, match="must be True, False"):
                 await events.append(
@@ -717,8 +806,7 @@ def test_append_rejects_unknown_instance_or_child(tmp_path) -> None:
                 actor_user_id="user-1",
             )
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         with pytest.raises(ValueError, match="belongs to child"):
             await events.append(
                 instance.id, 999, "completed", "user", _now_stamp(), True,
@@ -740,17 +828,18 @@ def test_completed_uncompleted_recompleted_three_ordered_events(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), False
+            instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id
         )
         history = await events.list_by_instance(instance.id)
         assert [e.event_type for e in history] == [
@@ -771,8 +860,7 @@ def test_get_latest_for_instance_none_when_untouched(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         return await events.get_latest_for_instance(instance.id)
 
     assert _with_db(tmp_path, "latest-none.db")(_body) is None
@@ -782,10 +870,10 @@ def test_get_latest_returns_uncompleted_when_reversed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
@@ -802,26 +890,26 @@ def test_list_by_child_and_date_range_scopes_by_due_date(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         early = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         inside = await instances.upsert(
-            definition.id, child.id, D2, _now_stamp()
-        )
+            definition.id, child.id, D2, _now_stamp(), window="morning")
         late = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
+            definition.id, child.id, D20, _now_stamp(), window="morning")
+        await events.append(
+            early.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
-            early.id, child.id, "completed", "panel", _now_stamp(), True
-        )
-        await events.append(
-            inside.id, child.id, "completed", "panel", _now_stamp(), False
+            inside.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id
         )
         await events.append(
             inside.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            late.id, child.id, "completed", "panel", _now_stamp(), True
+            late.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         found = await events.list_by_child_and_date_range(
             child.id, D2, D2
@@ -847,8 +935,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         late_due = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
-        )
+            definition.id, child.id, D20, _now_stamp(), window="morning")
         # The event's occurred_at is today at 08:00 UTC (_now_stamp);
         # the instance is due D20.  The due-date scope queried on D20
         # still finds the event (scoping follows the instance's due
@@ -856,7 +943,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
         # queried on today's date finds it too.
         event = await events.append(
             late_due.id, child.id, "completed", "panel",
-            _now_stamp(), False,
+            _now_stamp(), False, actor_child_id=child.id,
         )
         by_due = await events.list_by_child_and_date_range(
             child.id, D20, D20
@@ -944,7 +1031,12 @@ def test_no_mutation_sql_for_completion_events_anywhere() -> None:
     # UPDATE/DELETE/non-DAO-INSERT paths hit the pattern.
     insert_exempt = {
         "custom_components/nestquest/dao_instances.py",
+        "custom_components/nestquest/migrations.py",  # the rebuild
+        # migrations re-create the table via INSERT..SELECT (append-
+        # order preserved); UPDATE/DELETE stays forbidden everywhere
         "tests/test_schema.py",
+        "tests/test_migrations.py",  # seeds frozen legacy shapes to
+        # exercise the rebuild migrations
         "tests/test_dao_instances.py",  # this guard file; scanned by
         # its own guard's regex spans below instead
     }
@@ -1028,8 +1120,7 @@ def test_completion_events_dao_instance_has_no_mutation_capability(
 
         database.execute = _recording
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -1059,8 +1150,7 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
                 prepared
             )
             instance = await instances.upsert(
-                definition.id, child.id, D1, _now_stamp()
-            )
+                definition.id, child.id, D1, _now_stamp(), window="morning")
             tasks = [
                 asyncio.ensure_future(
                     events.append(

tests/test_dao_matrix.py

@@ -112,7 +112,8 @@ class World:
             self.child.id, D1, D2, True, note="test"
         )
         self.instance = await self.instances.upsert(
-            self.definition.id, self.child.id, D1, _stamp()
+            self.definition.id, self.child.id, D1, _stamp(),
+            window="morning",
         )
 
 
@@ -291,12 +292,15 @@ def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
 def test_matrix_quest_instances_crud_and_constraints(tmp_path) -> None:
     async def _body(w: World):
         # insert (seeded) + read
-        fetched = await w.instances.get(w.definition.id, D1)
+        fetched = await w.instances.get(
+            w.definition.id, w.child.id, D1, "morning"
+        )
         assert fetched == w.instance
         # update path: conflict refresh on the OPEN instance — same row
         # id, snapshot columns (due_time) actually rewritten.
         refreshed = await w.instances.upsert(
-            w.definition.id, w.child.id, D1, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D1, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert refreshed.id == fetched.id, (
             "the conflict path must refresh the existing row in place"
@@ -316,15 +320,15 @@ def test_matrix_quest_instances_constraints(tmp_path) -> None:
         # proving the schema backs the DAO's idempotency.
         await w.database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, due_time, generated_at) "
-            "VALUES (?, ?, ?, NULL, ?)",
+            "window, due_date, due_time, generated_at) "
+            "VALUES (?, ?, 'morning', ?, NULL, ?)",
             (w.definition.id, w.child.id, D20, _stamp()),
         )
         with pytest.raises(sqlite3.IntegrityError):
             await w.database.execute(
                 "INSERT INTO quest_instances (definition_id, child_id, "
-                "due_date, due_time, generated_at) "
-                "VALUES (?, ?, ?, NULL, ?)",
+                "window, due_date, due_time, generated_at) "
+                "VALUES (?, ?, 'morning', ?, NULL, ?)",
                 (w.definition.id, w.child.id, D20, _stamp()),
             )
         # delete future uncompleted removes D20's open instance
@@ -340,10 +344,11 @@ def test_matrix_quest_instances_constraints(tmp_path) -> None:
 def test_matrix_quest_instances_upsert_idempotency(tmp_path) -> None:
     async def _body(w: World):
         first = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp()
+            w.definition.id, w.child.id, D2, _stamp(), window="morning"
         )
         second = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D2, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert second.id == first.id
         count = await w.database.fetch_one(
@@ -367,14 +372,16 @@ def test_matrix_completion_events_append_read_and_no_mutation(
     async def _body(w: World):
         # insert: append a completion then a reversal, then re-complete
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), True
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), True,
+            actor_child_id=w.child.id,
         )
         await w.events.append(
             w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None,
             actor_user_id="user-1",
         )
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), False
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), False,
+            actor_child_id=w.child.id,
         )
         # read: full ordered history + latest
         history = await w.events.list_by_instance(w.instance.id)

tests/test_dao_rules.py

@@ -11,6 +11,7 @@ from custom_components.nestquest.dao_rules import (
     ScheduleRuleRecord,
     ScheduleRulesDao,
     QuestDefinitionRecord,
+    QuestDefinitionWindowRecord,
     QuestDefinitionsDao,
 )
 from custom_components.nestquest.db import NestQuestDatabase
@@ -790,6 +791,96 @@ def test_definition_list_assignees_orders_by_child_sort_order(
     _with_db(tmp_path, "definition-assignees-order.db")(_body)
 
 
+def test_definition_upsert_window_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        window = await definitions.upsert_window(
+            definition.id, "morning", due_time="09:00"
+        )
+        assert window == QuestDefinitionWindowRecord(
+            definition.id, "morning", "09:00"
+        )
+        # Upsert: re-declaring the same window updates in place, never
+        # duplicates (one row per definition+window).
+        updated = await definitions.upsert_window(
+            definition.id, "morning", due_time="09:30"
+        )
+        assert updated.due_time == "09:30"
+        rows = await definitions.list_windows(definition.id)
+        assert [w.window for w in rows] == ["morning"]
+        # All three windows list in const order regardless of insertion.
+        await definitions.upsert_window(definition.id, "evening")
+        await definitions.upsert_window(definition.id, "afternoon")
+        rows = await definitions.list_windows(definition.id)
+        assert [w.window for w in rows] == [
+            "morning",
+            "afternoon",
+            "evening",
+        ]
+        assert rows[1].due_time is None
+        return rows
+
+    _with_db(tmp_path, "definition-window-roundtrip.db")(_body)
+
+
+def test_definition_window_unknown_name_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        with pytest.raises(ValueError, match="window must be one of"):
+            await definitions.upsert_window(definition.id, "noon")
+        with pytest.raises(ValueError, match="window must be one of"):
+            await definitions.remove_window(definition.id, "MORNING")
+        return None
+
+    _with_db(tmp_path, "definition-window-bad-name.db")(_body)
+
+
+def test_definition_window_bad_due_time_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        with pytest.raises(ValueError, match="HH:MM"):
+            await definitions.upsert_window(
+                definition.id, "morning", due_time="9:30"
+            )
+        with pytest.raises(ValueError, match="HH:MM"):
+            await definitions.upsert_window(
+                definition.id, "morning", due_time="25:00"
+            )
+        return None
+
+    _with_db(tmp_path, "definition-window-bad-time.db")(_body)
+
+
+def test_definition_window_missing_definition_raises(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        with pytest.raises(ValueError, match="does not exist"):
+            await definitions.upsert_window(999, "morning")
+        return None
+
+    _with_db(tmp_path, "definition-window-missing.db")(_body)
+
+
+def test_definition_remove_window_round_trip(tmp_path) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        await definitions.upsert_window(definition.id, "morning")
+        assert await definitions.remove_window(
+            definition.id, "morning"
+        ) is True
+        assert await definitions.list_windows(definition.id) == []
+        # Removing an absent window reports False.
+        assert await definitions.remove_window(
+            definition.id, "morning"
+        ) is False
+        return None
+
+    _with_db(tmp_path, "definition-window-remove.db")(_body)
+
+
 def test_definition_unassigned_has_no_assignees(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")
@@ -836,7 +927,7 @@ def test_definition_assignment_leaves_history_rows_alone(
         # this table, but its row shape proves assignment isolation).
         await database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, generated_at) VALUES (?, ?, ?, ?)",
+            "window, due_date, generated_at) VALUES (?, ?, 'morning', ?, ?)",
             (definition.id, child.id, "2026-09-15", NOW),
         )
         await definitions.add_assignee(definition.id, other.id)
@@ -889,7 +980,7 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
     }
     sql_pattern = re.compile(
         r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
-        r"(schedule_rules|quest_definitions|quest_definition_assignees)\b",
+        r"(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     offenders: list[str] = []
@@ -941,7 +1032,7 @@ def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
     )
     sql_pattern = re.compile(
         r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
-        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees)\b",
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (
@@ -986,7 +1077,7 @@ def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
     )
     sql_pattern = re.compile(
         r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
-        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees)\b",
+        r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
         re.IGNORECASE,
     )
     assert sql_pattern.search(remaining) is None, (

tests/test_migrations.py

@@ -17,7 +17,10 @@ from custom_components.nestquest.migrations import (
     apply_migrations,
     read_schema_version,
 )
-from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS
+from custom_components.nestquest.schema import (
+    SCHEMA_V1_STATEMENTS,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+)
 
 
 def _run(coro):
@@ -64,6 +67,7 @@ def _counts(database) -> dict[str, int]:
         "schedule_rules",
         "quest_definitions",
         "quest_definition_assignees",
+        "quest_definition_windows",
         "presence_schedules",
         "presence_overrides",
         "quest_instances",
@@ -101,6 +105,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "schedule_rules",
             "quest_definitions",
             "quest_definition_assignees",
+            "quest_definition_windows",
             "presence_schedules",
             "presence_overrides",
             "quest_instances",
@@ -138,18 +143,23 @@ def test_empty_version_table_reads_as_version_0(tmp_path) -> None:
 
 
 def test_migration_list_shape() -> None:
-    """Migration 1 is the v1 DDL; 2 and 3 are the callable steps.
+    """Migration 1 is the v1 DDL; 2 and 3 are callable steps.
 
     Nothing ever shipped, so migration 1 was rewritten pre-release
     (D-007) to create the quest_* tables directly; migration 2 exists
-    for dev databases stamped version 1 by the pre-rename runner, and
+    for dev databases stamped version 1 by the pre-rename runner,
     migration 3 brings pre-D-008 definitions to the multi-assignee
-    model.
+    model, migration 4 adds the windows table, migration 5 rebuilds
+    quest_instances onto the widened (definition, child, date, window)
+    key, and migration 6 adds actor_child_id to completion_events.
     """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
     assert callable(MIGRATIONS[1])
     assert callable(MIGRATIONS[2])
-    assert len(MIGRATIONS) == 3
+    assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
+    assert callable(MIGRATIONS[4])
+    assert callable(MIGRATIONS[5])
+    assert len(MIGRATIONS) == 6
 
 
 # ---------------------------------------------------------------------------
@@ -419,6 +429,35 @@ _V2_DEFINITIONS_DDL = [
     """,
 ]
 
+_V2_INSTANCES_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_instances (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        due_date TEXT NOT NULL,
+        due_time TEXT,
+        generated_at TEXT NOT NULL,
+        UNIQUE (definition_id, due_date)
+    )
+    """,
+]
+
+_V2_EVENTS_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS completion_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        event_type TEXT NOT NULL,
+        actor_source TEXT NOT NULL,
+        actor_user_id TEXT,
+        occurred_at TEXT NOT NULL,
+        was_on_time INTEGER
+    )
+    """,
+]
+
 
 def _seed_v2_multi_assignee_upgrade(database) -> None:
     """Create a database shaped like a post-rename, pre-D-008 dev file.
@@ -427,11 +466,21 @@ def _seed_v2_multi_assignee_upgrade(database) -> None:
     single child_id column and no assignees table exists — stamped at
     version 2, with definition and instance rows that must survive.
     """
+    # Freeze the version-2 shape: definitions carry child_id, the
+    # instance table has no window column, and the assignees/windows
+    # tables do not exist yet — everything else matches the current DDL.
+    _v2_replaced = (
+        "quest_definitions ",
+        "quest_definition_assignees ",
+        "quest_definition_windows ",
+        "quest_instances ",
+        "completion_events ",
+    )
     for sql in SCHEMA_V1_STATEMENTS:
-        if "quest_definitions " in sql:
-            continue  # replaced by the legacy-shape DDL below
+        if any(name in sql for name in _v2_replaced):
+            continue
         _run(database.execute(sql))
-    for sql in _V2_DEFINITIONS_DDL:
+    for sql in _V2_DEFINITIONS_DDL + _V2_INSTANCES_DDL + _V2_EVENTS_DDL:
         _run(database.execute(sql))
     _run(database.execute(VERSION_TABLE_DDL))
     _run(
@@ -471,6 +520,26 @@ def _seed_v2_multi_assignee_upgrade(database) -> None:
             "VALUES (1, 2, '2026-09-15', '2026-09-14T00:00:00+00:00')"
         )
     )
+    # One legacy panel event and one legacy admin event: the panel one
+    # must backfill actor_child_id to its own child; the admin one must
+    # stay NULL.
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, occurred_at, was_on_time) "
+            "VALUES (1, 2, 'completed', 'panel', "
+            "'2026-09-15T10:00:00+00:00', 1)"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, actor_user_id, occurred_at, "
+            "was_on_time) "
+            "VALUES (1, 2, 'uncompleted', 'user', 'parent-1', "
+            "'2026-09-15T11:00:00+00:00', 1)"
+        )
+    )
 
 
 def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
@@ -500,14 +569,50 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
             )
         )
         assert assignees == [(1, 2)]
-        # The instance survived, still referencing the definition.
+        # The instance survived the rebuild, pinned to the migrated
+        # window, still referencing the definition.
         instance = _run(
             database.fetch_one(
-                "SELECT definition_id, child_id, due_date "
+                "SELECT definition_id, child_id, window, due_date "
                 "FROM quest_instances WHERE id = 1"
             )
         )
-        assert instance == (1, 2, "2026-09-15")
+        assert instance == (1, 2, "morning", "2026-09-15")
+        # The rebuilt instance table carries the widened unique key.
+        index_rows = _run(
+            database.fetch_all(
+                "PRAGMA index_list(quest_instances)"
+            )
+        )
+        unique_columns = []
+        for index_row in index_rows:
+            if index_row[2]:  # unique flag
+                info = _run(
+                    database.fetch_all(
+                        f"PRAGMA index_info('{index_row[1]}')"
+                    )
+                )
+                unique_columns.append([col[2] for col in info])
+        assert [
+            "definition_id", "child_id", "due_date", "window"
+        ] in unique_columns
+        # The completion events survived the actor_child_id rebuild:
+        # the panel row backfilled to its own child, the admin row
+        # stayed NULL, and ids are preserved.
+        events = _run(
+            database.fetch_all(
+                "SELECT id, event_type, actor_source, actor_child_id "
+                "FROM completion_events ORDER BY id"
+            )
+        )
+        assert events == [
+            (1, "completed", "panel", 2),
+            (2, "uncompleted", "user", None),
+        ]
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(completion_events)")
+        )
+        assert "actor_child_id" in {c[1] for c in columns}
         # No dangling references anywhere.
         assert _run(database.fetch_all("PRAGMA foreign_key_check")) == []
         # The rebuilt table is live for the new model: a definition

tests/test_options_flow.py

@@ -14,6 +14,7 @@ from conftest import make_config_entry, make_hass, options_flow_base as _options
 from custom_components.nestquest import config_flow
 from custom_components.nestquest import options_flow
 from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
     CONF_PANEL_IDLE_TIMEOUT,
@@ -46,13 +47,48 @@ def _make_entry(entry_id: str = "test_entry", options: dict | None = None, data:
     return make_config_entry(entry_id=entry_id, options=options, data=data)
 
 
-def _make_flow(entry) -> options_flow.NestQuestOptionsFlow:
-    """Create an options flow instance bound to a stubbed config entry."""
+def _make_flow(entry, *, users=None, database=None) -> options_flow.NestQuestOptionsFlow:
+    """Create an options flow instance bound to a stubbed config entry.
+
+    ``users`` populates a stubbed hass.auth for the admin picker;
+    ``database`` wires the runtime record so the flow can read and
+    write the allowlist like real HA's options flow does.
+    """
+    from unittest.mock import AsyncMock
+
     flow = options_flow.NestQuestOptionsFlow(config_entry=entry)
     flow.hass = SimpleNamespace()
+    flow.hass.auth = SimpleNamespace(
+        async_get_users=AsyncMock(return_value=users or [])
+    )
+    if database is not None:
+        from custom_components.nestquest.const import DOMAIN
+
+        flow.hass.data = {
+            DOMAIN: {entry.entry_id: SimpleNamespace(database=database)}
+        }
     return flow
 
 
+def _user(user_id: str, name: str):
+    return SimpleNamespace(id=user_id, name=name)
+
+
+def _open_allowlist_db(tmp_path, name):
+    """A real migrated temp database for the flow to read and write."""
+    from unittest.mock import AsyncMock, MagicMock
+
+    from custom_components.nestquest.db import NestQuestDatabase
+    from custom_components.nestquest.migrations import apply_migrations
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    database = NestQuestDatabase(hass)
+    _run(database.open(tmp_path / name))
+    _run(apply_migrations(database))
+    return database
+
+
 def _schema_values(schema) -> dict:
     """Extract the per-field default values from a real voluptuous schema."""
     return {
@@ -161,7 +197,12 @@ def test_options_flow_falls_back_to_entry_data() -> None:
 
 
 def test_options_flow_schema_applies_stored_defaults() -> None:
-    """A real voluptuous validate({}) fills every field from the stored defaults."""
+    """A real voluptuous validate({}) fills every field from the stored defaults.
+
+    Without a runtime database (and no auth in the bare harness), the
+    admin picker's default is the empty selection and its choice set is
+    empty — the three legacy settings are unaffected.
+    """
     entry = _make_entry(
         options={
             CONF_HORIZON_DAYS: 21,
@@ -174,6 +215,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_ADMIN_USER_IDS: [],
     }
 
 
@@ -373,4 +415,223 @@ def test_options_flow_no_hardcoded_domain_or_db_filename() -> None:
             )
             assert node.value != "nestquest.db", (
                 f"Hard-coded database filename in {py_file.name}:{node.lineno}"
-            )
+            )
+
+# ---------------------------------------------------------------------------
+# Admin allowlist picker (Feature 03)
+# ---------------------------------------------------------------------------
+
+
+def test_options_flow_picker_lists_existing_users(tmp_path) -> None:
+    """The picker offers EXISTING HA users (id -> name), never free text."""
+    import voluptuous as vol
+
+    database = _open_allowlist_db(tmp_path, "picker-users.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import set_admin_ids
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua"), _user("u2", "Sam")],
+            database=database,
+        )
+        result = _run(flow.async_step_init(None))
+        schema = result["data_schema"]
+        marker = next(
+            m for m in schema.schema
+            if getattr(m, "schema", None) == CONF_ADMIN_USER_IDS
+        )
+        # Default is the CURRENT database allowlist.
+        values = _schema_values(schema)
+        assert values[CONF_ADMIN_USER_IDS] == ["u1"]
+        # Both users are selectable; an unknown id is rejected by the
+        # schema itself (this is what real HA submits go through).
+        validated = schema(
+            {
+                CONF_HORIZON_DAYS: 7,
+                CONF_DAY_ROLLOVER_TIME: "03:30",
+                CONF_PANEL_IDLE_TIMEOUT: 60,
+                CONF_ADMIN_USER_IDS: ["u1", "u2"],
+            }
+        )
+        assert validated[CONF_ADMIN_USER_IDS] == ["u1", "u2"]
+        with pytest.raises(vol.Invalid):
+            schema(
+                {
+                    CONF_HORIZON_DAYS: 7,
+                    CONF_DAY_ROLLOVER_TIME: "03:30",
+                    CONF_PANEL_IDLE_TIMEOUT: 60,
+                    CONF_ADMIN_USER_IDS: ["impostor"],
+                }
+            )
+    finally:
+        _run(database.close())
+
+
+def test_options_flow_submit_applies_selection_to_database(tmp_path) -> None:
+    database = _open_allowlist_db(tmp_path, "picker-apply.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            list_admin_ids,
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua"), _user("u2", "Sam")],
+            database=database,
+        )
+        result = _run(
+            flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2"]})
+        )
+        assert result["type"] == "create_entry"
+        assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
+        # The allowlist (and the entry's recovery copy) carry the pick.
+        assert await_gather(list_admin_ids(database)) == ["u2"]
+    finally:
+        _run(database.close())
+
+
+def await_gather(coro):
+    return _run(coro)
+
+
+def test_options_flow_empty_selection_refused(tmp_path) -> None:
+    """Saving zero admins is refused: an empty allowlist fails closed."""
+    database = _open_allowlist_db(tmp_path, "picker-empty.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            list_admin_ids,
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua")],
+            database=database,
+        )
+        result = _run(
+            flow.async_step_init({**VALID_INPUT, CONF_ADMIN_USER_IDS: []})
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_ADMIN_USER_IDS: "no_admins"}
+        assert await_gather(list_admin_ids(database)) == ["u1"]
+    finally:
+        _run(database.close())
+
+
+def test_options_flow_unknown_selection_refused(tmp_path) -> None:
+    database = _open_allowlist_db(tmp_path, "picker-unknown.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            list_admin_ids,
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua")],
+            database=database,
+        )
+        result = _run(
+            flow.async_step_init(
+                {**VALID_INPUT, CONF_ADMIN_USER_IDS: ["impostor"]}
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
+        assert await_gather(list_admin_ids(database)) == ["u1"]
+    finally:
+        _run(database.close())
+
+
+def test_options_flow_submit_without_admin_key_keeps_allowlist(tmp_path) -> None:
+    """Raw callers omitting the key (pre-picker payloads) do not
+    touch the allowlist; the CURRENT allowlist is carried forward on
+    the options copy so no removed admin can be resurrected later."""
+    database = _open_allowlist_db(tmp_path, "picker-legacy.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            list_admin_ids,
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua")],
+            database=database,
+        )
+        result = _run(flow.async_step_init(dict(VALID_INPUT)))
+        assert result["type"] == "create_entry"
+        assert result["data"][CONF_ADMIN_USER_IDS] == ["u1"]
+        assert await_gather(list_admin_ids(database)) == ["u1"]
+    finally:
+        _run(database.close())
+
+
+def test_options_flow_duplicate_selection_refused(tmp_path) -> None:
+    """Duplicate ids surface as a field error, not an unhandled
+    ValueError from the business layer."""
+    database = _open_allowlist_db(tmp_path, "picker-dupe.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            list_admin_ids,
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u1"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua"), _user("u2", "Sam")],
+            database=database,
+        )
+        result = _run(
+            flow.async_step_init(
+                {**VALID_INPUT, CONF_ADMIN_USER_IDS: ["u2", "u2"]}
+            )
+        )
+        assert result["type"] == "form"
+        assert result["errors"] == {CONF_ADMIN_USER_IDS: "invalid_admin"}
+        assert await_gather(list_admin_ids(database)) == ["u1"]
+    finally:
+        _run(database.close())
+
+
+def test_options_flow_omitted_picker_carries_allowlist_forward(
+    tmp_path,
+) -> None:
+    """An omitted picker field must not drop the allowlist copy from
+    the returned options: HA replaces entry.options wholesale, and a
+    later database loss must not resurrect removed admins from the
+    stale config-flow copy in entry.data."""
+    database = _open_allowlist_db(tmp_path, "picker-carry.db")
+    try:
+        from custom_components.nestquest.admin_allowlist import (
+            set_admin_ids,
+        )
+
+        _run(set_admin_ids(database, ["u2"]))
+        entry = _make_entry()
+        flow = _make_flow(
+            entry,
+            users=[_user("u1", "Joshua"), _user("u2", "Sam")],
+            database=database,
+        )
+        result = _run(flow.async_step_init(dict(VALID_INPUT)))
+        assert result["type"] == "create_entry"
+        # The current allowlist rides along on the legacy three-key path.
+        assert result["data"][CONF_ADMIN_USER_IDS] == ["u2"]
+        assert result["data"][CONF_HORIZON_DAYS] == VALID_INPUT[CONF_HORIZON_DAYS]
+    finally:
+        _run(database.close())

tests/test_schema.py

@@ -17,6 +17,7 @@ from custom_components.nestquest.schema import (
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITIONS_DDL,
     SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+    SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
     SCHEMA_V1_QUEST_INSTANCES_DDL,
     async_apply_ddl,
 )
@@ -167,13 +168,14 @@ def _insert_override(database, child_id, **overrides):
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_seven_tables() -> None:
+def test_schema_v1_statements_compose_the_eight_tables() -> None:
     assert SCHEMA_V1_STATEMENTS == [
         *SCHEMA_V1_CHILDREN_DDL,
         *SCHEMA_V1_ADMIN_USERS_DDL,
         *SCHEMA_V1_SCHEDULE_RULES_DDL,
         *SCHEMA_V1_QUEST_DEFINITIONS_DDL,
         *SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
+        *SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
         *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
         *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
         *SCHEMA_V1_QUEST_INSTANCES_DDL,
@@ -974,6 +976,166 @@ def test_quest_definition_assignees_duplicate_pair_fails(tmp_path) -> None:
         _run(database.close())
 
 
+def test_quest_definition_windows_table_exists_after_applying_ddl(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "windows.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'quest_definition_windows'"
+            )
+        )
+        assert row is not None and row[0] == "quest_definition_windows"
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_columns_types_and_constraints(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "windows-columns.db")
+    try:
+        _apply(database)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definition_windows)")
+        )
+        # cid, name, type, notnull, dflt_value, pk — the composite
+        # primary key spans definition_id and window.
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("definition_id", "INTEGER", 1, None, 1),
+            ("window", "TEXT", 1, None, 2),
+            ("due_time", "TEXT", 0, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-fks.db")
+    try:
+        _apply(database)
+        fks = _run(
+            database.fetch_all(
+                "PRAGMA foreign_key_list(quest_definition_windows)"
+            )
+        )
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("quest_definitions", "definition_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_valid_rows_round_trip(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-roundtrip.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        for window, due in (
+            ("morning", "09:00"),
+            ("afternoon", None),
+            ("evening", "19:30"),
+        ):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, ?)",
+                    (definition_id, window, due),
+                )
+            )
+        rows = _run(
+            database.fetch_all(
+                "SELECT window, due_time FROM quest_definition_windows "
+                "WHERE definition_id = ? ORDER BY window",
+                (definition_id,),
+            )
+        )
+        assert rows == [
+            ("afternoon", None),
+            ("evening", "19:30"),
+            ("morning", "09:00"),
+        ]
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    "window", ["Midnight", "MORNING", "noon", "", "evenings"]
+)
+def test_quest_definition_windows_window_check_enforced(
+    tmp_path, window
+) -> None:
+    database = _open_db(tmp_path / "windows-check.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) VALUES (?, ?, NULL)",
+                    (definition_id, window),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_duplicate_window_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-duplicate.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        definition_id = _insert_definition(database, assignees=None)
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_windows "
+                "(definition_id, window, due_time) "
+                "VALUES (?, 'morning', NULL)",
+                (definition_id,),
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) "
+                    "VALUES (?, 'morning', '09:00')",
+                    (definition_id,),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_quest_definition_windows_unknown_definition_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "windows-bad-definition.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _run(
+                database.execute(
+                    "INSERT INTO quest_definition_windows "
+                    "(definition_id, window, due_time) "
+                    "VALUES (999, 'morning', NULL)"
+                )
+            )
+    finally:
+        _run(database.close())
+
+
 def test_quest_definition_assignees_multiple_children_allowed(
     tmp_path,
 ) -> None:
@@ -1560,8 +1722,8 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
 # ---------------------------------------------------------------------------
 
 
-def _insert_instance(database, definition_id=1, **overrides):
-    """Insert one valid task_instance, returning the instance id.
+def _insert_instance(database, definition_id=1, window="morning", **overrides):
+    """Insert one valid quest_instance, returning the instance id.
 
     Pass ``_OMIT`` as a value to leave that column out of the INSERT so
     its DDL DEFAULT applies.
@@ -1569,6 +1731,7 @@ def _insert_instance(database, definition_id=1, **overrides):
     values: dict[str, object] = {
         "definition_id": definition_id,
         "child_id": 1,
+        "window": window,
         "due_date": "2026-09-14",
         "generated_at": "2026-09-14T00:00:00+00:00",
     }
@@ -1667,6 +1830,7 @@ def test_quest_instances_columns_types_and_constraints(tmp_path) -> None:
             ("id", "INTEGER", 0, None, 1),
             ("definition_id", "INTEGER", 1, None, 0),
             ("child_id", "INTEGER", 1, None, 0),
+            ("window", "TEXT", 1, None, 0),
             ("due_date", "TEXT", 1, None, 0),
             ("due_time", "TEXT", 0, None, 0),
             ("generated_at", "TEXT", 1, None, 0),
@@ -1687,6 +1851,7 @@ def test_completion_events_columns_types_and_constraints(tmp_path) -> None:
             ("event_type", "TEXT", 1, None, 0),
             ("actor_source", "TEXT", 1, None, 0),
             ("actor_user_id", "TEXT", 0, None, 0),
+            ("actor_child_id", "INTEGER", 0, None, 0),
             ("occurred_at", "TEXT", 1, None, 0),
             ("was_on_time", "INTEGER", 0, None, 0),
         ]
@@ -1717,12 +1882,13 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
         assert declared == {
             ("quest_instances", "instance_id", "id"),
             ("children", "child_id", "id"),
+            ("children", "actor_child_id", "id"),
         }
     finally:
         _run(database.close())
 
 
-def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
+def test_quest_instances_unique_index_on_widened_key(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-unique.db")
     try:
         _apply(database)
@@ -1735,15 +1901,20 @@ def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
         )
         implicit = [(name, sql) for name, sql in rows if sql is None]
         assert len(implicit) == 1, (
-            "the (definition_id, due_date) UNIQUE constraint must be backed "
-            "by exactly one implicit unique index"
+            "the (definition_id, child_id, due_date, window) UNIQUE "
+            "constraint must be backed by exactly one implicit unique index"
         )
         columns = _run(
             database.fetch_all(
                 f"PRAGMA index_info('{implicit[0][0]}')"
             )
         )
-        assert [row[2] for row in columns] == ["definition_id", "due_date"]
+        assert [row[2] for row in columns] == [
+            "definition_id",
+            "child_id",
+            "due_date",
+            "window",
+        ]
         # The implicit index is genuinely unique: UNIQUE(...) is the only
         # constraint SQLite backs with a sql-less autoindex on this table.
         row = _run(
@@ -1770,12 +1941,16 @@ def test_quest_instances_valid_insert_round_trips(tmp_path) -> None:
         instance_id = _insert_instance(database, definition_id=1, child_id=1)
         row = _run(
             database.fetch_one(
-                "SELECT definition_id, child_id, due_date, due_time, generated_at "
+                "SELECT definition_id, child_id, window, due_date, "
+                "due_time, generated_at "
                 "FROM quest_instances WHERE id = ?",
                 (instance_id,),
             )
         )
-        assert row == (1, 1, "2026-09-14", None, "2026-09-14T00:00:00+00:00")
+        assert row == (
+            1, 1, "morning", "2026-09-14", None,
+            "2026-09-14T00:00:00+00:00",
+        )
     finally:
         _run(database.close())
 
@@ -1799,7 +1974,7 @@ def test_quest_instances_due_time_round_trips(tmp_path) -> None:
         _run(database.close())
 
 
-def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
+def test_quest_instances_duplicate_tuple_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-duplicate.db")
     try:
         _apply(database)
@@ -1811,6 +1986,78 @@ def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
         _run(database.close())
 
 
+def test_quest_instances_duplicate_tuple_different_child_allowed(
+    tmp_path,
+) -> None:
+    """D-008: same definition, date and window for two different
+    assignees of the same definition are two distinct instances."""
+    database = _open_db(tmp_path / "task-instances-shared.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Bo", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_assignees "
+                "(definition_id, child_id) VALUES (1, 2)"
+            )
+        )
+        _insert_instance(database, definition_id=1, child_id=1)
+        _insert_instance(database, definition_id=1, child_id=2)
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
+        assert row == (2,)
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_duplicate_tuple_different_window_allowed(
+    tmp_path,
+) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows."""
+    database = _open_db(tmp_path / "task-instances-twice-daily.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _insert_instance(database, definition_id=1, window="morning")
+        _insert_instance(database, definition_id=1, window="evening")
+        rows = _run(
+            database.fetch_all(
+                "SELECT window FROM quest_instances ORDER BY window"
+            )
+        )
+        assert [row[0] for row in rows] == ["evening", "morning"]
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_unknown_window_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-bad-window.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_instance(database, definition_id=1, window="noon")
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_window_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-window-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_instance(database, definition_id=1, window=_OMIT)
+    finally:
+        _run(database.close())
+
+
 def test_quest_instances_same_date_different_definitions_allowed(
     tmp_path,
 ) -> None:
@@ -1937,16 +2184,71 @@ def test_completion_events_valid_panel_completion_round_trips(tmp_path) -> None:
             instance_id=instance_id,
             actor_source="panel",
             actor_user_id=None,
+            actor_child_id=1,
             was_on_time=0,
         )
         row = _run(
             database.fetch_one(
-                "SELECT actor_source, actor_user_id, was_on_time "
-                "FROM completion_events WHERE id = ?",
+                "SELECT actor_source, actor_user_id, actor_child_id, "
+                "was_on_time FROM completion_events WHERE id = ?",
                 (event_id,),
             )
         )
-        assert row == ("panel", None, 0)
+        assert row == ("panel", None, 1, 0)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    ("actor_source", "actor_user_id", "actor_child_id"),
+    [
+        ("user", "user-1", 1),  # admin event must not carry a profile
+        ("panel", None, None),  # panel event requires the profile
+        ("panel", "user-1", 1),  # panel must not carry a user id either
+        ("panel", "user-1", None),
+        ("user", None, None),
+    ],
+)
+def test_completion_events_actor_child_coherence_enforced(
+    tmp_path, actor_source, actor_user_id, actor_child_id
+) -> None:
+    """D-008 actor-pair shape: 'user' events carry actor_user_id and no
+    actor_child_id; 'panel' events carry actor_child_id and no
+    actor_user_id.  The CHECKs make both halves non-negotiable."""
+    database = _open_db(tmp_path / "completion-events-coherence.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database,
+                instance_id=instance_id,
+                event_type="completed",
+                actor_source=actor_source,
+                actor_user_id=actor_user_id,
+                actor_child_id=actor_child_id,
+                was_on_time=1,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_actor_child_unknown_child_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-bad-actor-child.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_event(
+                database,
+                instance_id=instance_id,
+                actor_source="panel",
+                actor_user_id=None,
+                actor_child_id=999,
+                was_on_time=1,
+            )
     finally:
         _run(database.close())