Review fix: hardcode independently derived boundary expectations

Review round 1 (Codex gpt-5.6-terra, LOW): the boundary walk asserted
transitions in {(same, +1)} which even a reset/inverting
implementation could satisfy on a two-week cycle. The test now
asserts a hand-verified table of expected indices for every day
around each of the three Dec-31/Jan-1 boundaries (2019/2020/2021,
anchor 2019-01-06), so a parity bug cannot pass by coincidence.
Suite: 980 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/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

@@ -21,6 +21,12 @@ 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

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:
+def _build_schema(
-    """Build the form schema, pre-filled with the current values."""
+    current: dict[str, Any],
-    return vol.Schema(
+    *,
-        {
+    admin_choices: dict[str, str] | None = None,
-            vol.Required(
+    admin_default: list[str] | None = None,
-                CONF_HORIZON_DAYS, default=current[CONF_HORIZON_DAYS]
+) -> vol.Schema:
-            ): int,
+    """Build the form schema, pre-filled with the current values.
-            vol.Required(
+
-                CONF_DAY_ROLLOVER_TIME, default=current[CONF_DAY_ROLLOVER_TIME]
+    The admin picker is a multi-select of EXISTING Home Assistant users
-            ): str,
+    (never a free-text field — the allowlist stores user IDs); its
-            vol.Required(
+    default is the current database allowlist, so an untouched submit
-                CONF_PANEL_IDLE_TIMEOUT, default=current[CONF_PANEL_IDLE_TIMEOUT]
+    re-saves exactly who is already allowed.
-            ): int,
+    """
-        }
+    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(
+        # Apply the picker's selection to the allowlist when the
-            title="NestQuest",
+        # runtime database is reachable; the selection is always
-            data={
+        # stored on the entry as the disaster-recovery copy.
-                CONF_HORIZON_DAYS: user_input[CONF_HORIZON_DAYS],
+        database = self._async_database()
-                CONF_DAY_ROLLOVER_TIME: user_input[CONF_DAY_ROLLOVER_TIME],
+        if database is not None and admin_ids_present:
-                CONF_PANEL_IDLE_TIMEOUT: user_input[CONF_PANEL_IDLE_TIMEOUT],
+            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/presence.py

@@ -0,0 +1,319 @@
+"""The presence and custody model (Feature 05).
+
+PURE module: no Home Assistant imports, no database access, no
+timezone logic — it operates on plain calendar dates only (feature
+guardrail).  The later engine (cycle position, ``is_present``,
+overrides, previews) builds directly on this model.
+
+A :class:`PresenceSchedule` answers "on which weekdays of which cycle
+week is this child at this house?" for an N-week repeating custody
+pattern.  Two of the household's three children follow a two-week
+rotation; a child with NO schedule row is present every day (that
+default lives in the engine, not here).
+
+Pattern encoding: the model's canonical serialization is the SAME
+pipe-separated CSV the database stores (see schema.py's
+``presence_schedules.pattern``): segment ``i`` (week index ``i``) is a
+comma-separated list of present weekdays, 0 = Monday .. 6 = Sunday,
+e.g. ``'0,2,4|1,3'`` = Mon/Wed/Fri in week 0, Tue/Thu in week 1.  An
+empty segment means absent every day of that week.  Encoding sorts
+each segment's weekdays and uses the exact segment order, so
+``decode(encode(schedule))`` round-trips losslessly and two schedules
+with equal patterns produce equal strings.
+
+Anchor-date policy: ``anchor_date`` pins the cycle.  Cycle position is
+computed from anchor-date arithmetic ONLY — never ISO week numbers or
+week parity (D-004): 53-week years silently invert parity on January
+1st.  The arithmetic lives in the engine task; this module validates
+that the anchor is a strict calendar date.
+
+Validation policy: construction is total — every reject happens in
+``__init__`` so no invalid schedule can ever exist:
+- ``cycle_length_weeks`` must be an integer 1..4.  The 4 cap matches
+  the database CHECK (schema.py) and keeps anchor arithmetic and the
+  encoded pattern human-readable; revisit both together if a longer
+  cycle is ever needed.
+- ``pattern`` must cover week indices 0..cycle_length_weeks-1 exactly
+  (no gaps, no extra weeks) with sets of integers 0..6 (bools
+  rejected: True would silently mean Monday).
+- ``anchor_date`` must be a strict ISO YYYY-MM-DD calendar date (or a
+  ``datetime.date``).
+"""
+from __future__ import annotations
+
+import datetime
+from dataclasses import dataclass
+from types import MappingProxyType
+
+#: The schema's documented cap for cycle_length_weeks (schema.py keeps
+#: the pattern human-readable and the anchor arithmetic small).
+MAX_CYCLE_LENGTH_WEEKS = 4
+
+_DATE_FORMAT = "%Y-%m-%d"
+
+
+def _parse_date(value: object, field_name: str) -> datetime.date:
+    """Accept a datetime.date or a strict YYYY-MM-DD string.
+
+    ``datetime.datetime`` is rejected explicitly: it SUBCLASSES
+    datetime.date, so a bare isinstance check would store a value with
+    a time component and every later calendar-date comparison (e.g.
+    ``PresenceOverride.covers``) would raise TypeError.
+    """
+    if isinstance(value, datetime.datetime):
+        raise ValueError(
+            f"{field_name} must be a plain calendar date, not a "
+            f"datetime with a time component: {value!r}"
+        )
+    if isinstance(value, datetime.date):
+        return value
+    if not isinstance(value, str):
+        raise ValueError(
+            f"{field_name} must be a datetime.date or an ISO date string, "
+            f"got {value!r}"
+        )
+    try:
+        parsed = datetime.datetime.strptime(value, _DATE_FORMAT).date()
+    except (TypeError, ValueError):
+        raise ValueError(
+            f"{field_name} must be an ISO date (YYYY-MM-DD), got {value!r}"
+        ) from None
+    if parsed.isoformat() != value:
+        raise ValueError(
+            f"{field_name} must be a strict YYYY-MM-DD date, got {value!r}"
+        )
+    return parsed
+
+
+def _validate_week_index(week: object, cycle_length_weeks: int) -> int:
+    """Reject non-int week indices (bools included) and out-of-cycle."""
+    if isinstance(week, bool) or not isinstance(week, int):
+        raise ValueError(
+            f"pattern week index must be an integer, got {week!r}"
+        )
+    if not 0 <= week < cycle_length_weeks:
+        raise ValueError(
+            f"pattern week index must be in 0..{cycle_length_weeks - 1}, "
+            f"got {week}"
+        )
+    return week
+
+
+def _validate_weekday_set(week: int, weekdays: object) -> frozenset[int]:
+    """Validate one week's present-weekday set."""
+    if isinstance(weekdays, (str, bytes)) or weekdays is None:
+        raise ValueError(
+            f"pattern[{week}] must be an iterable of weekday ints 0..6, "
+            f"got {weekdays!r}"
+        )
+    try:
+        members = list(weekdays)
+    except TypeError:
+        raise ValueError(
+            f"pattern[{week}] must be an iterable of weekday ints 0..6, "
+            f"got {weekdays!r}"
+        ) from None
+    for member in members:
+        if isinstance(member, bool) or not isinstance(member, int):
+            raise ValueError(
+                f"pattern[{week}] weekday must be an int 0..6, got {member!r}"
+            )
+        if not 0 <= member <= 6:
+            raise ValueError(
+                f"pattern[{week}] weekday must be in 0..6 (Monday=0), "
+                f"got {member}"
+            )
+    return frozenset(members)
+
+
+def _validate_cycle_length(value: object) -> int:
+    """Reject non-int and out-of-range cycle lengths (bools included)."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(
+            f"cycle_length_weeks must be an integer, got {value!r}"
+        )
+    if not 1 <= value <= MAX_CYCLE_LENGTH_WEEKS:
+        raise ValueError(
+            f"cycle_length_weeks must be between 1 and "
+            f"{MAX_CYCLE_LENGTH_WEEKS}, got {value}"
+        )
+    return value
+
+
+def _validate_child_id(value: object) -> int:
+    """Reject non-int child ids (bools would silently address child 1)."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"child_id must be an integer, got {value!r}")
+    return value
+
+
+@dataclass(frozen=True, init=False)
+class PresenceSchedule:
+    """One child's N-week repeating presence pattern.
+
+    ``pattern`` maps each week index of the cycle (``0`` .. ``n-1``)
+    to the set of present weekdays (ints, Monday=0).  A week with an
+    empty set means the child is absent every day of that week — which
+    is distinct from having no schedule at all (the engine treats a
+    missing schedule as present every day).
+    """
+
+    child_id: int
+    cycle_length_weeks: int
+    anchor_date: datetime.date
+    pattern: MappingProxyType  # week index -> frozenset of weekdays
+
+    def __init__(
+        self,
+        child_id: int,
+        cycle_length_weeks: int,
+        anchor_date: object,
+        pattern: dict[int, object],
+    ) -> None:
+        object.__setattr__(self, "child_id", _validate_child_id(child_id))
+        cycle = _validate_cycle_length(cycle_length_weeks)
+        object.__setattr__(self, "cycle_length_weeks", cycle)
+        object.__setattr__(
+            self, "anchor_date", _parse_date(anchor_date, "anchor_date")
+        )
+        if not isinstance(pattern, dict):
+            raise ValueError(
+                "pattern must map week index to a set of present "
+                f"weekdays, got {pattern!r}"
+            )
+        normalised: dict[int, frozenset[int]] = {}
+        for week, weekdays in pattern.items():
+            index = _validate_week_index(week, cycle)
+            normalised[index] = _validate_weekday_set(index, weekdays)
+        missing = [
+            week for week in range(cycle) if week not in normalised
+        ]
+        if missing:
+            raise ValueError(
+                f"pattern must cover week indices 0..{cycle - 1}; "
+                f"missing {missing}"
+            )
+        # Frozen must mean frozen: a bare dict would let callers inject
+        # unvalidated weekdays (or drop required weeks) AFTER
+        # construction, bypassing total validation.  The mapping
+        # proxy exposes the validated contents read-only; the
+        # per-week sets are already frozensets.
+        object.__setattr__(self, "pattern", MappingProxyType(normalised))
+
+    def encode(self) -> str:
+        """Serialize to the database's pipe-separated CSV shape.
+
+        Segments appear in week order; each segment is the week's
+        present weekdays as a sorted comma-separated list (an empty set
+        encodes as an empty segment).  Two schedules with equal
+        patterns encode identically.
+        """
+        segments = [
+            ",".join(str(day) for day in sorted(self.pattern[week]))
+            for week in range(self.cycle_length_weeks)
+        ]
+        return "|".join(segments)
+
+    @classmethod
+    def decode(
+        cls,
+        child_id: int,
+        anchor_date: object,
+        encoded: str,
+    ) -> "PresenceSchedule":
+        """Build a schedule from its encoded pattern.
+
+        The cycle length is derived from the segment count, mirroring
+        the database's segment-count CHECK.  Decoding is strict: wrong
+        shapes, non-numeric or out-of-range weekdays are rejected —
+        the database CHECKs make such rows unrepresentable, and this
+        is the same contract on the model side.  A segment count above
+        the documented cycle cap is rejected by the ordinary
+        construction validation.
+        """
+        if not isinstance(encoded, str):
+            raise ValueError(
+                f"encoded pattern must be a string, got {encoded!r}"
+            )
+        segments = encoded.split("|")
+        pattern: dict[int, set[int]] = {}
+        for week, segment in enumerate(segments):
+            weekdays: set[int] = set()
+            if segment:
+                for part in segment.split(","):
+                    if len(part) != 1 or part not in "0123456":
+                        raise ValueError(
+                            f"encoded segment {week} carries an invalid "
+                            f"weekday {part!r} (single digits 0-6 only)"
+                        )
+                    weekdays.add(int(part))
+            pattern[week] = weekdays
+        return cls(child_id, len(segments), anchor_date, pattern)
+
+
+@dataclass(frozen=True, init=False)
+class PresenceOverride:
+    """A date-specific present/absent flag beating any pattern.
+
+    Covers a single date (``end_date == start_date``) or an inclusive
+    range.  A concrete range is the storage shape
+    (``presence_overrides.end_date`` is NOT NULL), so the model
+    requires both bounds and validates start <= end.  Construction and
+    validation live here so the engine can trust every instance.
+    """
+
+    child_id: int
+    start_date: datetime.date
+    end_date: datetime.date
+    is_present: bool
+    note: str | None
+
+    def __init__(
+        self,
+        child_id: int,
+        start_date: object,
+        end_date: object,
+        is_present: bool,
+        note: str | None = None,
+    ) -> None:
+        object.__setattr__(self, "child_id", _validate_child_id(child_id))
+        start = _parse_date(start_date, "start_date")
+        end = _parse_date(end_date, "end_date")
+        if end < start:
+            raise ValueError(
+                f"end_date must be on or after start_date, got "
+                f"{end.isoformat()} < {start.isoformat()}"
+            )
+        object.__setattr__(self, "start_date", start)
+        object.__setattr__(self, "end_date", end)
+        if not isinstance(is_present, bool):
+            raise ValueError(
+                f"is_present must be a real bool, got {is_present!r}"
+            )
+        object.__setattr__(self, "is_present", is_present)
+        if note is not None and not isinstance(note, str):
+            raise ValueError(f"note must be a string or None, got {note!r}")
+        object.__setattr__(self, "note", note)
+
+    def covers(self, date: datetime.date) -> bool:
+        """Return True when the override applies to ``date``."""
+        return self.start_date <= date <= self.end_date
+
+
+def cycle_week_index(
+    schedule: PresenceSchedule, target_date: object
+) -> int:
+    """Return the cycle week index ``target_date`` falls in.
+
+    Anchor-date arithmetic ONLY (D-004): ``((target_date -
+    anchor_date).days // 7) % cycle_length_weeks``.  Python's floor
+    division makes dates BEFORE the anchor behave correctly — they
+    walk the cycle backwards (day -1 is the last week of the cycle,
+    day -8 wraps one further) instead of inverting, which is exactly
+    what ISO week parity does wrong in 53-week years.  The modulo of
+    a negative in Python is non-negative, so the index is always a
+    valid week index.
+    """
+    target = _parse_date(target_date, "target_date")
+    days = (target - schedule.anchor_date).days
+    return (days // 7) % schedule.cycle_length_weeks

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):
+    def _factory(
-        return make_config_entry(entry_id=entry_id, options=options, data=data)
+        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_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_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:
+def _make_flow(entry, *, users=None, database=None) -> options_flow.NestQuestOptionsFlow:
-    """Create an options flow instance bound to a stubbed config entry."""
+    """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_presence.py

@@ -0,0 +1,373 @@
+"""Tests for presence.py: the Feature 05 custody model."""
+from __future__ import annotations
+
+import datetime
+
+import pytest
+
+from custom_components.nestquest.presence import (
+    MAX_CYCLE_LENGTH_WEEKS,
+    PresenceOverride,
+    PresenceSchedule,
+    cycle_week_index,
+)
+
+
+ANCHOR = datetime.date(2026, 1, 5)  # a Monday
+
+TWO_WEEK_PATTERN = {0: {0, 2, 4}, 1: {1, 3}}
+
+
+# ---------------------------------------------------------------------------
+# PresenceSchedule construction
+# ---------------------------------------------------------------------------
+
+
+def test_schedule_holds_validated_fields() -> None:
+    schedule = PresenceSchedule(
+        3, 2, "2026-01-05", {0: {0, 2, 4}, 1: {1, 3}}
+    )
+    assert schedule.child_id == 3
+    assert schedule.cycle_length_weeks == 2
+    assert schedule.anchor_date == ANCHOR
+    assert schedule.pattern == {0: frozenset({0, 2, 4}), 1: frozenset({1, 3})}
+
+
+def test_schedule_accepts_date_objects() -> None:
+    schedule = PresenceSchedule(1, 1, ANCHOR, {0: {0}})
+    assert schedule.anchor_date is ANCHOR
+
+
+def test_schedule_rejects_cycle_length_below_one() -> None:
+    with pytest.raises(ValueError, match="between 1 and"):
+        PresenceSchedule(1, 0, ANCHOR, {})
+    with pytest.raises(ValueError, match="between 1 and"):
+        PresenceSchedule(1, -1, ANCHOR, {})
+
+
+def test_schedule_rejects_cycle_length_above_documented_cap() -> None:
+    """The 4 cap matches the schema's CHECK — both documents agree."""
+    with pytest.raises(ValueError, match="between 1 and"):
+        PresenceSchedule(1, MAX_CYCLE_LENGTH_WEEKS + 1, ANCHOR, {})
+    assert MAX_CYCLE_LENGTH_WEEKS == 4
+
+
+@pytest.mark.parametrize("bad", [True, "2", 1.0, None])
+def test_schedule_rejects_non_int_cycle_length(bad) -> None:
+    with pytest.raises(ValueError, match="cycle_length_weeks must be an integer"):
+        PresenceSchedule(1, bad, ANCHOR, {0: {0}})
+
+
+def test_schedule_rejects_pattern_missing_week_indices() -> None:
+    with pytest.raises(ValueError, match="missing \\[1\\]"):
+        PresenceSchedule(1, 2, ANCHOR, {0: {0, 2}})
+
+
+def test_schedule_rejects_pattern_with_extra_week_indices() -> None:
+    with pytest.raises(ValueError, match="week index must be in 0"):
+        PresenceSchedule(1, 2, ANCHOR, {0: {0}, 1: {1}, 2: {2}})
+
+
+def test_schedule_rejects_pattern_with_non_int_week_keys() -> None:
+    with pytest.raises(ValueError, match="week index must be an integer"):
+        PresenceSchedule(1, 1, ANCHOR, {"0": {0}})
+
+
+def test_schedule_rejects_pattern_with_bool_week_key() -> None:
+    """True would silently be week 1; bools are never valid indices."""
+    with pytest.raises(ValueError, match="week index must be an integer"):
+        PresenceSchedule(1, 2, ANCHOR, {0: {0}, True: {1}})
+
+
+@pytest.mark.parametrize("bad_weekday", [-1, 7, 100])
+def test_schedule_rejects_weekday_outside_zero_to_six(bad_weekday) -> None:
+    with pytest.raises(ValueError, match="weekday must be in 0..6"):
+        PresenceSchedule(1, 1, ANCHOR, {0: {bad_weekday}})
+
+
+@pytest.mark.parametrize("bad", [True, "1", 1.5, None])
+def test_schedule_rejects_non_int_weekdays(bad) -> None:
+    with pytest.raises(ValueError, match="weekday must be an int"):
+        PresenceSchedule(1, 1, ANCHOR, {0: [bad]})
+
+
+def test_schedule_rejects_string_and_none_week_sets() -> None:
+    """A string is iterable — "02" would silently mean {0, 2}."""
+    with pytest.raises(ValueError, match="iterable of weekday ints"):
+        PresenceSchedule(1, 1, ANCHOR, {0: "02"})
+    with pytest.raises(ValueError, match="iterable of weekday ints"):
+        PresenceSchedule(1, 1, ANCHOR, {0: None})
+
+
+def test_schedule_rejects_non_dict_pattern() -> None:
+    with pytest.raises(ValueError, match="must map week index"):
+        PresenceSchedule(1, 1, ANCHOR, "0,2")
+
+
+@pytest.mark.parametrize("bad", [True, 1.0, "2026-01-05", None])
+def test_schedule_rejects_non_int_child_id(bad) -> None:
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        PresenceSchedule(bad, 1, ANCHOR, {0: {0}})
+
+
+@pytest.mark.parametrize(
+    "bad", ["2026-1-5", "2026/01/05", "not-a-date", 5, None, True]
+)
+def test_schedule_rejects_non_strict_anchor_dates(bad) -> None:
+    with pytest.raises(ValueError, match="anchor_date"):
+        PresenceSchedule(1, 1, bad, {0: {0}})
+
+
+def test_schedule_empty_week_means_absent_week() -> None:
+    """An explicitly empty week is valid: absent every day of it —
+    distinct from having no schedule (the engine's default)."""
+    schedule = PresenceSchedule(1, 2, ANCHOR, {0: {0, 2}, 1: set()})
+    assert schedule.pattern[1] == frozenset()
+
+
+def test_schedule_empty_week_encodes_as_empty_segment() -> None:
+    schedule = PresenceSchedule(1, 2, ANCHOR, {0: {0, 2}, 1: set()})
+    assert schedule.encode() == "0,2|"
+
+
+def test_schedule_equal_patterns_are_equal_and_encode_identically() -> None:
+    one = PresenceSchedule(1, 2, ANCHOR, {0: {4, 2, 0}, 1: {3, 1}})
+    two = PresenceSchedule(2, 2, ANCHOR, {0: {0, 2, 4}, 1: {1, 3}})
+    assert one.encode() == two.encode() == "0,2,4|1,3"
+
+
+# ---------------------------------------------------------------------------
+# Serialization round-trip
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "encoded",
+    [
+        "0,2,4|1,3",
+        "0|1,2,3,4,5,6",
+        "|",
+        "0,2|",
+        "|0,2",
+        "6",
+        "0,1,2,3,4,5,6|",
+    ],
+)
+def test_schedule_decode_encode_round_trips_losslessly(encoded) -> None:
+    schedule = PresenceSchedule.decode(1, ANCHOR, encoded)
+    assert schedule.encode() == encoded
+    again = PresenceSchedule.decode(1, ANCHOR, schedule.encode())
+    assert again == schedule
+
+
+def test_schedule_decode_sets_cycle_length_from_segments() -> None:
+    one_week = PresenceSchedule.decode(1, ANCHOR, "0,2,4")
+    assert one_week.cycle_length_weeks == 1
+    four_week = PresenceSchedule.decode(1, ANCHOR, "0|1|2|3,4")
+    assert four_week.cycle_length_weeks == 4
+
+
+def test_schedule_decode_rejects_malformed_segments() -> None:
+    for bad in ("0,12|1", "0,,2|1", "x|1", "0|-1|2", "8", "0,|1"):
+        with pytest.raises(ValueError, match="invalid weekday"):
+            PresenceSchedule.decode(1, ANCHOR, bad)
+
+
+def test_schedule_decode_rejects_non_string() -> None:
+    with pytest.raises(ValueError, match="must be a string"):
+        PresenceSchedule.decode(1, ANCHOR, None)
+
+
+def test_schedule_decode_rejects_too_many_segments(tmp_path) -> None:
+    with pytest.raises(ValueError, match="between 1 and"):
+        PresenceSchedule.decode(1, ANCHOR, "0|1|2|3|4")
+
+
+# ---------------------------------------------------------------------------
+# PresenceOverride
+# ---------------------------------------------------------------------------
+
+
+def test_override_single_date_and_range_cover() -> None:
+    single = PresenceOverride(1, "2026-07-20", "2026-07-20", False)
+    assert single.covers(datetime.date(2026, 7, 20))
+    assert not single.covers(datetime.date(2026, 7, 21))
+    span = PresenceOverride(1, "2026-07-20", "2026-07-24", True, note="trade")
+    assert span.covers(datetime.date(2026, 7, 22))
+    assert not span.covers(datetime.date(2026, 7, 19))
+    assert not span.covers(datetime.date(2026, 7, 25))
+    assert span.note == "trade"
+
+
+def test_override_end_before_start_rejected() -> None:
+    with pytest.raises(ValueError, match="end_date must be on or after"):
+        PresenceOverride(1, "2026-07-24", "2026-07-20", True)
+
+
+@pytest.mark.parametrize("bad", ["1", 0, 1.0, None])
+def test_override_rejects_non_bool_is_present(bad) -> None:
+    with pytest.raises(ValueError, match="is_present must be a real bool"):
+        PresenceOverride(1, "2026-07-20", "2026-07-20", bad)
+
+
+@pytest.mark.parametrize("bad", [5, ["note"], 1.5])
+def test_override_rejects_non_string_note(bad) -> None:
+    with pytest.raises(ValueError, match="note must be a string or None"):
+        PresenceOverride(1, "2026-07-20", "2026-07-20", True, note=bad)
+
+
+@pytest.mark.parametrize("bad", [True, 1.0, "1"])
+def test_override_rejects_non_int_child_id(bad) -> None:
+    with pytest.raises(ValueError, match="child_id must be an integer"):
+        PresenceOverride(bad, "2026-07-20", "2026-07-20", True)
+
+
+@pytest.mark.parametrize(
+    ("start", "end"), [("2026-7-20", "2026-07-20"), ("nope", "2026-07-20")]
+)
+def test_override_rejects_non_strict_dates(start, end) -> None:
+    with pytest.raises(ValueError, match="date"):
+        PresenceOverride(1, start, end, True)
+
+
+def test_schedule_rejects_datetime_anchor() -> None:
+    """datetime.datetime subclasses date: a bare isinstance check would
+    store a time component and every date comparison would TypeError."""
+    with pytest.raises(ValueError, match="plain calendar date"):
+        PresenceSchedule(1, 1, datetime.datetime(2026, 1, 5, 9, 30), {0: {0}})
+
+
+def test_override_rejects_datetime_bounds() -> None:
+    with pytest.raises(ValueError, match="plain calendar date"):
+        PresenceOverride(
+            1, datetime.datetime(2026, 7, 20), "2026-07-20", True
+        )
+
+
+def test_schedule_pattern_is_immutable_after_construction() -> None:
+    """The frozen schedule must not leak a mutable dict: injecting
+    unvalidated weekdays or dropping required weeks after construction
+    would bypass total validation."""
+    schedule = PresenceSchedule(1, 2, ANCHOR, {0: {0, 2}, 1: {1}})
+    with pytest.raises(TypeError):
+        schedule.pattern[0] = {9}
+    with pytest.raises(TypeError):
+        del schedule.pattern[1]
+    # mappingproxy exposes no mutators at all.
+    assert not any(
+        hasattr(schedule.pattern, name)
+        for name in ("clear", "pop", "popitem", "setdefault", "update")
+    )
+    # The validated contents are still readable and encode correctly.
+    assert schedule.pattern[0] == frozenset({0, 2})
+    assert schedule.encode() == "0,2|1"
+
+
+# ---------------------------------------------------------------------------
+# Cycle arithmetic: anchor-date only, never ISO week parity
+# ---------------------------------------------------------------------------
+
+
+def test_cycle_week_index_follows_the_anchor_formula() -> None:
+    schedule = PresenceSchedule(1, 2, ANCHOR, {0: {0}, 1: {1}})
+    # The anchor day starts week 0; each whole 7 days steps the cycle.
+    offsets = {0: 0, 1: 0, 6: 0, 7: 1, 8: 1, 13: 1, 14: 0, 15: 0, 22: 1}
+    for offset, expected in offsets.items():
+        assert (
+            cycle_week_index(
+                schedule, ANCHOR + datetime.timedelta(days=offset)
+            )
+            == expected
+        ), f"offset {offset}"
+
+
+def test_cycle_week_index_before_the_anchor_walks_backwards() -> None:
+    """Python floor division: day -1 is the LAST week of the cycle,
+    wrapping backwards — never inverting."""
+    schedule = PresenceSchedule(1, 2, ANCHOR, {0: {0}, 1: {1}})
+    assert cycle_week_index(schedule, ANCHOR - datetime.timedelta(days=1)) == 1
+    assert cycle_week_index(schedule, ANCHOR - datetime.timedelta(days=7)) == 1
+    assert cycle_week_index(schedule, ANCHOR - datetime.timedelta(days=8)) == 0
+
+
+def test_cycle_week_index_three_week_pattern() -> None:
+    schedule = PresenceSchedule(
+        1, 3, ANCHOR, {0: {0}, 1: {1}, 2: {2}}
+    )
+    offsets = [0, 7, 14, 21, 28]
+    assert [cycle_week_index(schedule, ANCHOR + datetime.timedelta(days=o)) for o in offsets] == [
+        0, 1, 2, 0, 1
+    ]
+
+
+def test_cycle_week_index_stable_across_three_year_boundaries() -> None:
+    """The regression this exists for: walking day by day through a
+    December 31st -> January 1st boundary, the cycle index steps by 0
+    or +1 (mod cycle length) and NEVER resets or inverts — including
+    2020, a 53-ISO-week year, where ISO week parity flips and would
+    invert a two-week custody schedule on January 1st."""
+    schedule = PresenceSchedule(
+        1, 2, datetime.date(2019, 1, 6), {0: {0, 2, 4}, 1: {1, 3}}
+    )
+    # Independently derived expectations (anchor 2019-01-06, index =
+    # floor(days-since-anchor / 7) % 2, hand-verified) — hardcoded so
+    # even an implementation that resets or inverts at January 1st
+    # cannot pass by coincidence:
+    expected = {
+        # 2019 -> 2020 boundary
+        datetime.date(2019, 12, 26): 0,
+        datetime.date(2019, 12, 27): 0,
+        datetime.date(2019, 12, 28): 0,
+        datetime.date(2019, 12, 29): 1,
+        datetime.date(2019, 12, 30): 1,
+        datetime.date(2019, 12, 31): 1,
+        datetime.date(2020, 1, 1): 1,
+        datetime.date(2020, 1, 2): 1,
+        datetime.date(2020, 1, 3): 1,
+        datetime.date(2020, 1, 4): 1,
+        datetime.date(2020, 1, 5): 0,
+        datetime.date(2020, 1, 6): 0,
+        datetime.date(2020, 1, 7): 0,
+        # 2020 (53 ISO weeks) -> 2021 boundary
+        datetime.date(2020, 12, 27): 1,
+        datetime.date(2020, 12, 28): 1,
+        datetime.date(2020, 12, 29): 1,
+        datetime.date(2020, 12, 30): 1,
+        datetime.date(2020, 12, 31): 1,
+        datetime.date(2021, 1, 1): 1,
+        datetime.date(2021, 1, 2): 1,
+        datetime.date(2021, 1, 3): 0,
+        datetime.date(2021, 1, 4): 0,
+        datetime.date(2021, 1, 5): 0,
+        datetime.date(2021, 1, 6): 0,
+        datetime.date(2021, 1, 7): 0,
+        # 2021 -> 2022 boundary
+        datetime.date(2021, 12, 27): 1,
+        datetime.date(2021, 12, 28): 1,
+        datetime.date(2021, 12, 29): 1,
+        datetime.date(2021, 12, 30): 1,
+        datetime.date(2021, 12, 31): 1,
+        datetime.date(2022, 1, 1): 1,
+        datetime.date(2022, 1, 2): 0,
+        datetime.date(2022, 1, 3): 0,
+        datetime.date(2022, 1, 4): 0,
+        datetime.date(2022, 1, 5): 0,
+        datetime.date(2022, 1, 6): 0,
+        datetime.date(2022, 1, 7): 0,
+    }
+    for day, wanted in expected.items():
+        got = cycle_week_index(schedule, day)
+        assert got == wanted, f"{day}: expected week {wanted}, got {got}"
+    # And 2020 really is the 53-ISO-week year this guards against.
+    assert datetime.date(2020, 12, 31).isocalendar()[1] == 53
+
+
+def test_module_uses_no_iso_week_functions() -> None:
+    """D-004 guardrail: cycle math never reads ISO week numbers."""
+    from pathlib import Path
+
+    source = Path(
+        __import__("custom_components.nestquest.presence", fromlist=["__file__"]).__file__
+    ).read_text()
+    assert "isocalendar" not in source
+    assert "isoweek" not in source