Admin routes: history query and CSV export

The admin History screen (ADMIN-SPEC 1g) had no API: the read
primitives existed (append-only completion-event reads, derived-state
missed logic) but nothing assembled them, so the admin PWA could not
render the event log or export it.  Feature 13's export service was
superseded by the Admin PWA and remains unimplemented by design.

Add a Home-Assistant-free core history layer (core/history.py) that
owns every rule so the two routes cannot drift: one query over the
closed due-date range [start, end] for filter all (every completion
event) / reversals (uncompleted only) / missed (DERIVED past-due open
instances via the existing completion.list_missed_for_child rule --
never stored, never appended); strict ISO date and filter validation
raising ValueError naming filter/start/end; and a csv-module serializer
with a stable documented header (ADMIN-SPEC 5 event-row fields) so the
CSV export is the SAME filtered range as the query, through one
export_history_csv call.  Admin routes GET /api/v1/admin/history and
/history.csv are thin adapters on the existing admin router (the one
require_admin dependency; no new auth check), mapping malformed input
to 422 and never 500.  completion_events stays append-only: history
only reads.

Tests (tests/test_api_admin_history.py) drive the ASGI app as an admin
via the shared JWT harness: seeded completion + reversal + past-due
open instances prove each filter's rows, the CSV header/rows via
csv.reader, due-date range scoping, 403/401 refusals, and 422s for
unknown filter, non-strict dates and inverted ranges.

api/nestquest_core.py

@@ -116,6 +116,11 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
 #: ``uncomplete_instance``) the panel's write routes call.
 core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion")
 
+#: The read-only history layer (``core.history.query_history``,
+#: ``export_history_csv``, ``rows_to_csv``) the admin plane's history
+#: query and CSV export routes call.
+core_history = importlib.import_module(f"{_CORE_MODULE_NAME}.history")
+
 #: The transition-event builders (``core.events.build_quest_completed_event``
 #: et al.) the panel's write routes build their SSE events with.
 core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
@@ -124,12 +129,81 @@ core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
 #: the snapshot builder's ``settings`` parameter takes.
 core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
 
+#: The children business layer (``core.children.create_child`` et al.)
+#: the admin plane's children routes call.
+core_children = importlib.import_module(f"{_CORE_MODULE_NAME}.children")
+
+#: The quest-definition business layer (``core.quest_definitions`` —
+#: ``create_quest_definition``, ``edit_quest_definition``,
+#: ``set_quest_definition_active``) the admin plane's quest-definition
+#: routes call.
+core_quest_definitions = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.quest_definitions"
+)
+
+#: The recurrence model (``core.recurrence.ScheduleRule`` and
+#: ``RuleValidationError``) the quest-definition routes build request
+#: rules with — the model validates at construction, the routes never
+#: re-implement rule validation.
+core_recurrence = importlib.import_module(f"{_CORE_MODULE_NAME}.recurrence")
+
+#: The presence model (``core.presence.PresenceEngine``,
+#: ``PresenceSchedule``, ``PresenceOverride``) the admin plane's
+#: presence routes serialize and the tests resolve presence with.
+core_presence = importlib.import_module(f"{_CORE_MODULE_NAME}.presence")
+
+#: The presence-write business layer
+#: (``core.presence_management.set_presence_schedule`` et al.) the
+#: admin plane's presence routes call — the ONE implementation the HA
+#: services share.
+core_presence_management = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.presence_management"
+)
+
+#: The typed presence DAOs (``core.dao_presence``) the admin-plane tests
+#: read stored schedules and overrides back through.
+core_dao_presence = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.dao_presence"
+)
+
+#: The materialization walk (``core.materialize.materialize``,
+#: ``regenerate_for_definition``, ``regenerate_for_child``) the admin
+#: plane's regenerate route calls — the same walk the integration's
+#: config/presence changes trigger.
+core_materialize = importlib.import_module(f"{_CORE_MODULE_NAME}.materialize")
+
+#: The core constants module (``core.const.DEFAULT_HORIZON_DAYS``) the
+#: regenerate route sizes the household rolling horizon with.
+core_const = importlib.import_module(f"{_CORE_MODULE_NAME}.const")
+
+#: The typed child DAO (``core.dao_children.ChildrenDao.get``) the
+#: regenerate route reads a child's existence with before delegating.
+core_dao_children = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.dao_children"
+)
+
+#: The typed definition DAO (``core.dao_rules.QuestDefinitionsDao.get``)
+#: the regenerate route reads a definition's existence with before
+#: delegating.
+core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules")
+
 __all__ = [
     "core",
+    "core_children",
     "core_completion",
+    "core_const",
+    "core_dao_children",
+    "core_dao_presence",
+    "core_dao_rules",
     "core_db",
     "core_events",
+    "core_history",
+    "core_materialize",
     "core_migrations",
+    "core_presence",
+    "core_presence_management",
+    "core_quest_definitions",
+    "core_recurrence",
     "core_snapshot",
     "core_settings",
 ]

custom_components/nestquest/core/history.py

@@ -0,0 +1,405 @@
+"""Read-only history layer for the admin plane (Admin spec §5, task 2d2dda53).
+
+Assembles the History screen's event rows over the append-only
+completion log: a query over a CLOSED date range ``[start_date,
+end_date]`` for one of three filters —
+
+- ``all`` — every completion event (completed AND uncompleted) whose
+  instance is due in the range;
+- ``reversals`` — the uncompleted events only;
+- ``missed`` — the DERIVED missed instances in the range.
+
+Range semantics (Feature 13's rule, kept): the range scopes by the
+INSTANCE's ``due_date`` — the day the task was owed — not the event's
+wall-clock ``occurred_at``, because history reporting asks "what
+happened to the tasks due that day".  ``missed`` is derived, never
+stored: rows come from :func:`~.completion.list_missed_for_child`
+(past-due open instances, the same ``_derived_state`` rule the
+coordinator uses), and nothing is ever appended for them.
+
+Row shape: the Admin spec §5 event-row fields, fixed and documented in
+:data:`CSV_COLUMNS` — ``occurred_at``, ``event_type``,
+``child_name``, ``child_id``, ``quest_title``, ``instance_id``,
+``window``, ``due_date``, ``due_time``, ``actor``, ``was_on_time``.
+Derived missed rows carry ``event_type`` ``'missed'`` (never a stored
+event_type), no ``occurred_at`` (there is no event), ``was_on_time``
+``None`` and the spec's ``nightly sweep`` actor.  A stored event's
+``actor`` follows the spec's meta shape: the ``actor_user_id`` string
+for a user event (``joshua``), ``panel (Child)`` naming the tapped
+profile for a panel event.
+
+Validation policy (module-level, shared by both the JSON query and the
+CSV export): the filter must be one of :data:`FILTERS` and the range
+must be two strict ``YYYY-MM-DD`` dates with ``end >= start``;
+violations raise ``ValueError`` naming the offending field
+(``filter`` / ``start`` / ``end``) BEFORE any read.  The whole
+household read runs under the task-reentrant connection lock so every
+row of one answer is one snapshot — the same discipline
+:mod:`.completion` gives its read-decide-write paths.
+
+:func:`rows_to_csv` serializes rows with the :mod:`csv` module (proper
+quoting, one stable header row) and :func:`export_history_csv` runs
+query + serialization as ONE call, so the CSV export can never drift
+from the JSON query's filtered range.  No permission check lives here
+(that is the admin router's single ``require_admin`` dependency).
+"""
+from __future__ import annotations
+
+import csv
+import datetime
+import io
+from dataclasses import dataclass
+
+from .completion import _reentrant_connection_lock, list_missed_for_child
+from .dao_children import ChildRecord, ChildrenDao
+from .dao_instances import (
+    EVENT_UNCOMPLETED,
+    CompletionEventRecord,
+    CompletionEventsDao,
+    QuestInstanceRecord,
+    QuestInstancesDao,
+)
+from .dao_rules import QuestDefinitionsDao, _validate_date
+from .db import NestQuestDatabase
+
+#: The ``all`` filter: every completion event in the range.
+FILTER_ALL = "all"
+
+#: The ``reversals`` filter: uncompleted events only.
+FILTER_REVERSALS = "reversals"
+
+#: The ``missed`` filter: derived missed instances in the range.
+FILTER_MISSED = "missed"
+
+#: The filters the query accepts, in the Admin spec's chip order.
+FILTERS = (FILTER_ALL, FILTER_REVERSALS, FILTER_MISSED)
+
+#: The ``event_type`` a derived missed row reports.  ``missed`` is a
+#: DERIVED state, never a stored event_type (Feature 08).
+EVENT_MISSED = "missed"
+
+#: The actor a derived missed row reports (Admin spec §5 meta:
+#: ``missed · nightly sweep``).
+ACTOR_NIGHTLY_SWEEP = "nightly sweep"
+
+#: The CSV column set, in header order — the Admin spec §5 event-row
+#: fields.  The header row is exactly this tuple, always, so a consumer
+#: can key on column names.
+CSV_COLUMNS = (
+    "occurred_at",
+    "event_type",
+    "child_name",
+    "child_id",
+    "quest_title",
+    "instance_id",
+    "window",
+    "due_date",
+    "due_time",
+    "actor",
+    "was_on_time",
+)
+
+
+@dataclass(frozen=True)
+class HistoryRow:
+    """One history row: a stored event or a derived missed instance.
+
+    The fields are :data:`CSV_COLUMNS` in order.  For a derived missed
+    row ``occurred_at`` is ``None`` (no event exists) and ``was_on_time``
+    is ``None``; for a stored event ``event_type`` is
+    ``'completed'``/``'uncompleted'`` and the fields come from the event
+    joined with its instance, child and definition.
+    """
+
+    occurred_at: str | None
+    event_type: str
+    child_name: str
+    child_id: int
+    quest_title: str
+    instance_id: int
+    window: str
+    due_date: str
+    due_time: str | None
+    actor: str
+    was_on_time: bool | None
+
+
+def _validate_filter(filter_name: str) -> str:
+    """Return the filter, or raise ValueError naming the field."""
+    if filter_name not in FILTERS:
+        raise ValueError(
+            f"filter must be one of {', '.join(repr(f) for f in FILTERS)}, "
+            f"got {filter_name!r}"
+        )
+    return filter_name
+
+
+def _validate_range(start_date: str, end_date: str) -> tuple[str, str]:
+    """Validate the closed range and return it, else raise naming the field.
+
+    Both bounds must be strict ``YYYY-MM-DD`` dates (the shared
+    :func:`~.dao_rules._validate_date`, so non-padded or re-ordered
+    shapes are rejected and a non-string raises naming the field) and
+    ``end`` must be on or after ``start``.
+    """
+    _validate_date(start_date, "start")
+    _validate_date(end_date, "end")
+    if end_date < start_date:
+        raise ValueError(
+            f"end must be on or after start, got {end_date!r} < "
+            f"{start_date!r}"
+        )
+    return start_date, end_date
+
+
+def _resolve_today(today: datetime.date | None) -> datetime.date:
+    """Return the threaded HA-local date, or the host calendar date."""
+    if today is None:
+        return datetime.date.today()
+    return today
+
+
+def _event_actor(
+    event: CompletionEventRecord, actor_child: ChildRecord | None
+) -> str:
+    """Return the spec-meta actor string for one stored event.
+
+    A user event reports the ``actor_user_id`` string (the spec's
+    ``uncompleted · joshua (admin)`` shape); a panel event reports
+    ``panel (Child)`` naming the tapped profile.  The tapped child is
+    never hard-deleted, so the lookup below is a belt-and-braces
+    fallback to the stored id, never a silent rewrite of policy.
+    """
+    if event.actor_source == "user":
+        return event.actor_user_id or ""
+    if actor_child is not None:
+        return f"panel ({actor_child.display_name})"
+    return f"panel ({event.actor_child_id})"
+
+
+def _event_row(
+    event: CompletionEventRecord,
+    instance: QuestInstanceRecord,
+    child: ChildRecord,
+    quest_title: str,
+    actor_child: ChildRecord | None,
+) -> HistoryRow:
+    """Join one stored event with its instance, child and definition."""
+    return HistoryRow(
+        occurred_at=event.occurred_at,
+        event_type=event.event_type,
+        child_name=child.display_name,
+        child_id=event.child_id,
+        quest_title=quest_title,
+        instance_id=event.instance_id,
+        window=instance.window,
+        due_date=instance.due_date,
+        due_time=instance.due_time,
+        actor=_event_actor(event, actor_child),
+        was_on_time=event.was_on_time,
+    )
+
+
+def _missed_row(
+    instance: QuestInstanceRecord,
+    child: ChildRecord,
+    quest_title: str,
+) -> HistoryRow:
+    """Build the derived row for one past-due open instance."""
+    return HistoryRow(
+        occurred_at=None,
+        event_type=EVENT_MISSED,
+        child_name=child.display_name,
+        child_id=instance.child_id,
+        quest_title=quest_title,
+        instance_id=instance.id,
+        window=instance.window,
+        due_date=instance.due_date,
+        due_time=instance.due_time,
+        actor=ACTOR_NIGHTLY_SWEEP,
+        was_on_time=None,
+    )
+
+
+def _csv_values(row: HistoryRow) -> list[str]:
+    """Return one CSV row's cell values in :data:`CSV_COLUMNS` order.
+
+    ``None`` cells serialize as empty (a missed row has no
+    ``occurred_at``); ``was_on_time`` serializes as ``true``/``false``
+    when the event carries the flag and empty when it does not.
+    """
+    return [
+        row.occurred_at or "",
+        row.event_type,
+        row.child_name,
+        str(row.child_id),
+        row.quest_title,
+        str(row.instance_id),
+        row.window,
+        row.due_date,
+        row.due_time or "",
+        row.actor,
+        "" if row.was_on_time is None else str(row.was_on_time).lower(),
+    ]
+
+
+def rows_to_csv(rows: list[HistoryRow]) -> str:
+    """Serialize history rows to CSV text with the documented header.
+
+    The FIRST row is always :data:`CSV_COLUMNS`, so the header is
+    stable and deterministic; the :mod:`csv` module handles quoting
+    (titles or names containing commas or quotes are quoted per RFC
+    4180) and CRLF line endings.
+    """
+    buffer = io.StringIO()
+    writer = csv.writer(buffer)
+    writer.writerow(CSV_COLUMNS)
+    for row in rows:
+        writer.writerow(_csv_values(row))
+    return buffer.getvalue()
+
+
+async def query_history(
+    database: NestQuestDatabase,
+    filter_name: str,
+    start_date: str,
+    end_date: str,
+    *,
+    today: datetime.date | None = None,
+) -> list[HistoryRow]:
+    """Return the history rows for ``filter_name`` over the closed range.
+
+    Validates the filter and the strict ``YYYY-MM-DD`` range (raising
+    ``ValueError`` naming ``filter`` / ``start`` / ``end``) BEFORE any
+    read, then builds the rows:
+
+    - ``all`` / ``reversals``: every child's events whose instances are
+      due in the range (:meth:`~.dao_instances.CompletionEventsDao.list_by_child_and_date_range`,
+      the Feature 13 due-date scope), ``reversals`` keeping only the
+      ``uncompleted`` rows.
+    - ``missed``: :func:`~.completion.list_missed_for_child` per child —
+      the SAME derived-state rule the coordinator uses, never a new one,
+      and never a stored event.
+
+    Rows are ordered deterministically by ``due_date``, then instance,
+    then event id (the per-instance append order, so a completion and
+    its reversal keep their order).  ``today`` pins the HA-local date
+    the missed derivation compares due dates against (the host calendar
+    date when omitted).  The whole read runs under the task-reentrant
+    connection lock — :func:`list_missed_for_child` re-enters it — so
+    one answer is one snapshot.
+    """
+    filter_name = _validate_filter(filter_name)
+    start_date, end_date = _validate_range(start_date, end_date)
+    today_date = _resolve_today(today)
+
+    keyed: list[tuple[tuple[str, int, int], HistoryRow]] = []
+    async with _reentrant_connection_lock(database):
+        children = await ChildrenDao(database).list_all()
+        children_by_id = {child.id: child for child in children}
+        instances = QuestInstancesDao(database)
+        definitions = QuestDefinitionsDao(database)
+        events = CompletionEventsDao(database)
+        titles: dict[int, str] = {}
+
+        if filter_name == FILTER_MISSED:
+            for child in children:
+                for instance in await list_missed_for_child(
+                    database, child.id, start_date, end_date, today=today_date
+                ):
+                    if instance.definition_id not in titles:
+                        definition = await definitions.get(
+                            instance.definition_id
+                        )
+                        assert definition is not None, (
+                            "a materialized instance's definition exists"
+                        )
+                        titles[instance.definition_id] = definition.title
+                    keyed.append(
+                        (
+                            (instance.due_date, instance.id, 0),
+                            _missed_row(
+                                instance,
+                                child,
+                                titles[instance.definition_id],
+                            ),
+                        )
+                    )
+            keyed.sort(key=lambda pair: pair[0])
+            return [row for _, row in keyed]
+
+        for child in children:
+            for event in await events.list_by_child_and_date_range(
+                child.id, start_date, end_date
+            ):
+                if filter_name == FILTER_REVERSALS and (
+                    event.event_type != EVENT_UNCOMPLETED
+                ):
+                    continue
+                instance = await instances.get_by_id(event.instance_id)
+                assert instance is not None, (
+                    "an event's instance is never deleted (append-only "
+                    "history keeps its references)"
+                )
+                actor_child = (
+                    children_by_id.get(event.actor_child_id)
+                    if event.actor_child_id is not None
+                    else None
+                )
+                if instance.definition_id not in titles:
+                    definition = await definitions.get(instance.definition_id)
+                    assert definition is not None, (
+                        "a materialized instance's definition exists"
+                    )
+                    titles[instance.definition_id] = definition.title
+                keyed.append(
+                    (
+                        (instance.due_date, instance.id, event.id),
+                        _event_row(
+                            event,
+                            instance,
+                            child,
+                            titles[instance.definition_id],
+                            actor_child,
+                        ),
+                    )
+                )
+        keyed.sort(key=lambda pair: pair[0])
+        return [row for _, row in keyed]
+
+
+async def export_history_csv(
+    database: NestQuestDatabase,
+    filter_name: str,
+    start_date: str,
+    end_date: str,
+    *,
+    today: datetime.date | None = None,
+) -> str:
+    """Return the filtered history as CSV text, header row included.
+
+    ONE query + serialization call so the CSV export carries exactly
+    the JSON query's filtered range and order — the same
+    :func:`query_history` validation, rows and :func:`rows_to_csv`
+    serializer, never a second implementation.
+    """
+    rows = await query_history(
+        database, filter_name, start_date, end_date, today=today
+    )
+    return rows_to_csv(rows)
+
+
+__all__ = [
+    "ACTOR_NIGHTLY_SWEEP",
+    "CSV_COLUMNS",
+    "EVENT_MISSED",
+    "EVENT_UNCOMPLETED",
+    "FILTERS",
+    "FILTER_ALL",
+    "FILTER_MISSED",
+    "FILTER_REVERSALS",
+    "HistoryRow",
+    "export_history_csv",
+    "query_history",
+    "rows_to_csv",
+]

custom_components/nestquest/core/presence_management.py

@@ -0,0 +1,161 @@
+"""Business layer for presence writes (Feature 09, API admin plane).
+
+Sits between the write callers (the Feature 09 HA service handlers and
+the API admin plane's presence routes) and the typed presence DAOs plus
+the pure presence model: callers get the model validation, the upsert /
+create / delete operations and the regeneration side effect here, and
+never touch the DAO or SQL directly.  This is the ONE implementation of
+these business rules — every caller constructs the same
+:class:`~.presence.PresenceSchedule` / :class:`~.presence.PresenceOverride`
+models (whose constructors are total, so no re-validation happens in a
+handler) and gets the same post-write
+:func:`~.materialize.regenerate_for_child`, so a child's future quest
+instances track its changed presence no matter which plane issued the
+write.  Like the model it is Home-Assistant-free: ``today`` is a plain
+calendar date threaded through to the regeneration, defaulting to the
+host clock there.
+
+Upsert policy (Feature 05, unchanged): a child has AT MOST ONE presence
+schedule — setting again REPLACES the existing row rather than
+duplicating it.  Overrides are deletable by design (children are not —
+the children layer owns that guardrail).  Deleting a nonexistent
+override raises ValueError prefixed ``override_id:`` so a caller can
+distinguish "unknown id" from a rejected argument the same way the
+quest-definitions layer prefixes ``definition_id:``.
+"""
+from __future__ import annotations
+
+import datetime
+
+from .dao_presence import (
+    PresenceOverrideRecord,
+    PresenceOverridesDao,
+    PresenceSchedulesDao,
+)
+from .db import NestQuestDatabase
+from .materialize import regenerate_for_child
+from .presence import PresenceOverride, PresenceSchedule
+
+
+def _validate_override_id(value: object) -> int:
+    """Reject non-int override ids (bools would silently address id 1)."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"override_id must be an integer, got {value!r}")
+    return value
+
+
+async def set_presence_schedule(
+    database: NestQuestDatabase,
+    child_id: int,
+    cycle_length_weeks: int,
+    anchor_date: object,
+    pattern: dict[int, object],
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceSchedule:
+    """Set (upsert) a child's repeating presence schedule.
+
+    The :class:`~.presence.PresenceSchedule` model validates at
+    construction — cycle length 1..4, a strict calendar anchor date,
+    and a pattern covering week indices 0..n-1 with weekday sets — so
+    no invalid schedule reaches the DAO.  The encoded pattern is
+    UPSERTed by child (a second set REPLACES the child's existing
+    schedule; the schema's UNIQUE(child_id) makes duplication
+    unrepresentable), then :func:`~.materialize.regenerate_for_child`
+    re-materializes the child's future instances against the new
+    presence.  ``today``/``horizon_days`` thread through to that
+    regeneration; raising ValueError names what to fix (an unknown
+    child reads ``child N does not exist``).
+
+    Returns the STORED schedule decoded back through the model — the
+    round-trip proves the encoded pattern the database now holds is
+    exactly the one the caller asked for.
+    """
+    schedule = PresenceSchedule(
+        child_id, cycle_length_weeks, anchor_date, pattern
+    )
+    record = await PresenceSchedulesDao(database).upsert_by_child(
+        schedule.child_id,
+        schedule.cycle_length_weeks,
+        schedule.anchor_date.isoformat(),
+        schedule.encode(),
+    )
+    await regenerate_for_child(
+        database, schedule.child_id, today=today, horizon_days=horizon_days
+    )
+    return PresenceSchedule.decode(
+        record.child_id, record.anchor_date, record.pattern
+    )
+
+
+async def create_presence_override(
+    database: NestQuestDatabase,
+    child_id: int,
+    start_date: object,
+    end_date: object,
+    is_present: bool,
+    note: str | None = None,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceOverrideRecord:
+    """Create one date-range presence override and return it as stored.
+
+    The :class:`~.presence.PresenceOverride` model validates at
+    construction — strict calendar dates, ``end_date`` on or after
+    ``start_date``, a real ``is_present`` bool, a string-or-None note —
+    then the DAO insert applies the same-child no-overlap policy (a
+    conflicting range raises ValueError naming the existing override)
+    and :func:`~.materialize.regenerate_for_child` re-materializes the
+    child's future instances against the override's presence answer.
+    An unknown child reads ``child N does not exist``.
+
+    Returns the stored record WITH its ``id`` — the handle a later
+    :func:`delete_presence_override` takes.
+    """
+    override = PresenceOverride(
+        child_id, start_date, end_date, is_present, note=note
+    )
+    record = await PresenceOverridesDao(database).create(
+        override.child_id,
+        override.start_date.isoformat(),
+        override.end_date.isoformat(),
+        override.is_present,
+        note=override.note,
+    )
+    await regenerate_for_child(
+        database, override.child_id, today=today, horizon_days=horizon_days
+    )
+    return record
+
+
+async def delete_presence_override(
+    database: NestQuestDatabase,
+    override_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceOverrideRecord:
+    """Delete one presence override by id and return it as it was stored.
+
+    The override is looked up FIRST and an unknown id raises ValueError
+    prefixed ``override_id:`` (naming the id) BEFORE any delete, so a
+    mistyped id can never be silently swallowed by a no-op.  The
+    deletion is followed by :func:`~.materialize.regenerate_for_child`
+    for the override's own child — the id decides the child, never a
+    caller-supplied one.  ``today``/``horizon_days`` thread through to
+    that regeneration.
+    """
+    _validate_override_id(override_id)
+    dao = PresenceOverridesDao(database)
+    override = await dao.get(override_id)
+    if override is None:
+        raise ValueError(
+            f"override_id: presence override {override_id} does not exist"
+        )
+    await dao.delete(override_id)
+    await regenerate_for_child(
+        database, override.child_id, today=today, horizon_days=horizon_days
+    )
+    return override

custom_components/nestquest/presence_management.py

@@ -0,0 +1,10 @@
+"""Re-export of :mod:`custom_components.nestquest.core.presence_management` for the integration package.
+
+The presence-write business layer lives in
+:mod:`custom_components.nestquest.core.presence_management` (added for
+task e4a9d31b); this module re-exports it so the integration's service
+handlers call the ONE implementation the API admin plane shares.
+"""
+from __future__ import annotations
+
+from .core.presence_management import *  # noqa: F401,F403

custom_components/nestquest/services.py

@@ -4,9 +4,11 @@ Registration is domain-scoped and happens once (see
 :func:`async_register_services`); handlers resolve a live database at
 call time through the injected ``find_runtime`` callable.  Permission
 checks are Feature 09 later tasks — this module only validates payload
-shape and forwards to the Feature 03/06/08 business layers (and the
-presence DAOs).  ``export_history_csv`` is registered with a schema but
-raises until Feature 13.  ``regenerate`` is supplied by the caller so
+shape and forwards to the Feature 03/06/08 business layers (the
+presence writes included, through :mod:`.presence_management` — the
+ONE implementation the API admin plane shares).
+``export_history_csv`` is registered with a schema but raises until
+Feature 13.  ``regenerate`` is supplied by the caller so
 the Feature 07 handler stays the one in ``__init__``.
 """
 from __future__ import annotations
@@ -22,6 +24,7 @@ from homeassistant.exceptions import HomeAssistantError
 
 from . import children as children_layer
 from . import completion as completion_layer
+from . import presence_management as presence_management_layer
 from . import quest_definitions as quest_definitions_layer
 from .const import (
     DOMAIN,
@@ -39,11 +42,8 @@ from .const import (
     SERVICE_UNCOMPLETE_QUEST,
     SERVICE_UPDATE_QUEST_DEFINITION,
 )
-from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
 from .events import fire_quest_completed, fire_quest_uncompleted
-from .materialize import regenerate_for_child
 from .permissions import permission_gate
-from .presence import PresenceSchedule
 from .recurrence import ScheduleRule
 from .service_policy import SERVICE_POLICY
 
@@ -405,21 +405,12 @@ def _set_presence_pattern(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         data = call.data
-        schedule = PresenceSchedule(
+        await presence_management_layer.set_presence_schedule(
+            runtime.database,
             data["child_id"],
             data["cycle_length_weeks"],
             data["anchor_date"],
             data["pattern"],
-        )
-        await PresenceSchedulesDao(runtime.database).upsert_by_child(
-            schedule.child_id,
-            schedule.cycle_length_weeks,
-            schedule.anchor_date.isoformat(),
-            schedule.encode(),
-        )
-        await regenerate_for_child(
-            runtime.database,
-            schedule.child_id,
             **_generation_kwargs(hass, runtime),
         )
 
@@ -432,16 +423,13 @@ def _create_presence_override(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         data = call.data
-        await PresenceOverridesDao(runtime.database).create(
+        await presence_management_layer.create_presence_override(
+            runtime.database,
             data["child_id"],
             data["start_date"],
             data["end_date"],
             data["is_present"],
             note=data.get("note"),
-        )
-        await regenerate_for_child(
-            runtime.database,
-            data["child_id"],
             **_generation_kwargs(hass, runtime),
         )
 
@@ -453,17 +441,9 @@ def _delete_presence_override(
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
-        dao = PresenceOverridesDao(runtime.database)
-        override_id = call.data["override_id"]
-        override = await dao.get(override_id)
-        if override is None:
-            raise HomeAssistantError(
-                f"override_id: presence override {override_id} does not exist"
-            )
-        await dao.delete(override_id)
-        await regenerate_for_child(
+        await presence_management_layer.delete_presence_override(
             runtime.database,
-            override.child_id,
+            call.data["override_id"],
             **_generation_kwargs(hass, runtime),
         )
 

tests/admin_jwt_harness.py

@@ -0,0 +1,189 @@
+"""Shared admin-plane JWT test harness: local keys, stubbed JWKS.
+
+The admin-plane API test modules share ONE harness instead of
+duplicating it: the auth contract tests (:mod:`tests.test_api_admin_auth`)
+and the children route tests (:mod:`tests.test_api_admin_children`) both
+drive the ASGI app with httpx using a locally generated RSA key pair and
+a stubbed JWKS fetcher — NO network access:
+
+- ``ADMIN_KEY`` signs the test tokens; :func:`jwks_for` builds the JWKS
+  document publishing its public half, so genuinely signed tokens
+  verify, and tokens signed with the second pair (``OTHER_KEY``) fail
+  the signature check for real.
+- :func:`make_token` mints one RS256 JWT the way Authentik would (its
+  claims, our key), defaulting ``groups`` to the admin group a real
+  admin's token carries.
+- :class:`StubbedFetch` is the counting JWKS fetcher stub.
+- :class:`AdminRunner` runs the app's lifespan around an httpx
+  AsyncClient (httpx's ASGI transport does not run the lifespan, so it
+  is driven explicitly and shut down in ``__aexit__``'s finally block
+  even on failure) with the app's JWKS cache swapped for one built
+  around the stub fetcher.
+
+The harness carries no tests itself; the ``admin_client`` fixture in
+``tests/conftest.py`` hands its runner to the test modules.
+"""
+from __future__ import annotations
+
+import base64
+import time
+
+import httpx
+import jwt
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from api.app import create_app
+from api.auth import ADMIN_GROUP, JwksCache
+from api.config import ApiConfig
+
+#: The configured OIDC settings the test apps run with (values a real
+#: deployment would get from the NESTQUEST_OIDC_* environment vars).
+ISSUER = "https://authentik.example.com/application/o/nestquest/"
+AUDIENCE = "nestquest-api"
+JWKS_URL = "https://authentik.example.com/application/o/nestquest/jwks/"
+
+#: The ``kid`` the test tokens carry, matching the JWKS entry.
+KID = "admin-test-key-1"
+
+#: The subject a valid token authenticates as.
+SUBJECT = "admin-user-1"
+
+
+def _generate_key() -> rsa.RSAPrivateKey:
+    """Generate one local RSA key pair (no network, no fixtures)."""
+    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
+
+
+#: The key pair whose PUBLIC half the stubbed JWKS publishes — tokens
+#: signed with it verify; the second pair below proves they don't when
+#: signed otherwise.
+ADMIN_KEY = _generate_key()
+OTHER_KEY = _generate_key()
+
+
+def _b64url_uint(value: int) -> str:
+    """Base64url-encode a non-negative integer (no padding), per JWK."""
+    raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
+
+
+def jwks_for(private_key: rsa.RSAPrivateKey, kid: str) -> dict[str, object]:
+    """Build a JWKS document publishing ``private_key``'s public half."""
+    numbers = private_key.public_key().public_numbers()
+    return {
+        "keys": [
+            {
+                "kty": "RSA",
+                "use": "sig",
+                "alg": "RS256",
+                "kid": kid,
+                "n": _b64url_uint(numbers.n),
+                "e": _b64url_uint(numbers.e),
+            }
+        ]
+    }
+
+
+class StubbedFetch:
+    """A JWKS fetcher stub: returns a fixed document, counts calls."""
+
+    def __init__(self, document: dict[str, object]) -> None:
+        self.document = document
+        self.calls = 0
+
+    async def __call__(self, url: str) -> dict[str, object]:
+        self.calls += 1
+        return self.document
+
+
+def make_token(
+    signing_key: rsa.RSAPrivateKey = ADMIN_KEY,
+    *,
+    kid: str | None = KID,
+    iss: str = ISSUER,
+    aud: object = AUDIENCE,
+    sub: str = SUBJECT,
+    expires_in: int | None = 300,
+    nbf_in: int | None = -1,
+    groups: object = (ADMIN_GROUP,),
+) -> str:
+    """Mint one RS256 JWT the way Authentik would (its claims, our key).
+
+    ``groups`` defaults to the admin group (what a real admin user's
+    token carries); tests pass a different value — or ``None`` for NO
+    groups claim at all — to prove the group gate rejects it.
+    """
+    now = int(time.time())
+    claims: dict[str, object] = {"iss": iss, "aud": aud, "sub": sub, "iat": now}
+    if groups is not None:
+        claims["groups"] = (
+            list(groups) if isinstance(groups, (list, tuple)) else groups
+        )
+    if nbf_in is not None:
+        claims["nbf"] = now + nbf_in
+    if expires_in is not None:
+        claims["exp"] = now + expires_in
+    headers: dict[str, str] = {}
+    if kid is not None:
+        headers["kid"] = kid
+    return jwt.encode(
+        claims, signing_key, algorithm="RS256", headers=headers
+    )
+
+
+class AdminRunner:
+    """Run the app's lifespan around an httpx AsyncClient.
+
+    Same pattern as test_api_health.py's ``_AppRunner``: httpx's ASGI
+    transport does not run the lifespan, so it is driven explicitly and
+    shut down in ``__aexit__``'s finally block even on failure.
+
+    The app's JWKS cache is replaced (before the lifespan runs) with
+    one built around a counting stub fetcher, so verification goes
+    through the REAL cache logic but never the network.
+    """
+
+    def __init__(
+        self,
+        db_path: str,
+        jwks_document: dict[str, object],
+        *,
+        ttl_seconds: float | None = None,
+        backoff_seconds: float | None = None,
+        kid_refetch_seconds: float | None = None,
+    ) -> None:
+        self.app = create_app(
+            ApiConfig(
+                db_path=db_path,
+                panel_token="admin-test-panel-token",
+                oidc_issuer=ISSUER,
+                oidc_audience=AUDIENCE,
+                oidc_jwks_url=JWKS_URL,
+            )
+        )
+        self.fetch = StubbedFetch(jwks_document)
+        cache_kwargs: dict[str, float] = {}
+        if ttl_seconds is not None:
+            cache_kwargs["ttl_seconds"] = ttl_seconds
+        if backoff_seconds is not None:
+            cache_kwargs["backoff_seconds"] = backoff_seconds
+        if kid_refetch_seconds is not None:
+            cache_kwargs["kid_refetch_seconds"] = kid_refetch_seconds
+        self.app.state.jwks_cache = JwksCache(
+            JWKS_URL, fetcher=self.fetch, **cache_kwargs
+        )
+
+    async def __aenter__(self) -> httpx.AsyncClient:
+        self._lifespan = self.app.router.lifespan_context(self.app)
+        await self._lifespan.__aenter__()
+        self._client = httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=self.app),
+            base_url="http://testserver",
+        )
+        return await self._client.__aenter__()
+
+    async def __aexit__(self, *exc) -> None:
+        try:
+            await self._client.__aexit__(*exc)
+        finally:
+            await self._lifespan.__aexit__(*exc)

tests/conftest.py

@@ -33,8 +33,11 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
+import httpx
 import pytest
 
+from tests.admin_jwt_harness import ADMIN_KEY, KID, AdminRunner, jwks_for
+
 # ---------------------------------------------------------------------------
 # Homeassistant module provisioning (explicitly mock-only).
 # ---------------------------------------------------------------------------
@@ -953,4 +956,25 @@ def freeze_nestquest_clock(
         date=datetime.date,
     )
     monkeypatch.setattr(nestquest, "datetime", fake_module)
-    return frozen, requested
+    return frozen, requested
+
+
+# ---------------------------------------------------------------------------
+# Admin-plane API fixtures (shared by the admin-plane test modules).
+#
+# The local-key/stubbed-JWKS harness itself lives in
+# :mod:`tests.admin_jwt_harness` (imported by name, the same pattern as
+# tests.blueprint_helpers); these fixtures hand its runner to every test
+# module in tests/ without re-declaring it.
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def temp_db_path(tmp_path: Path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def admin_client(temp_db_path: str) -> httpx.AsyncClient:
+    """An app client whose provider publishes the admin test key."""
+    async with AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID)) as client:
+        yield client

tests/test_api_admin_auth.py

@@ -2,7 +2,10 @@
 
 These tests exercise the NestQuest API's admin plane against the ASGI
 app with httpx, using a locally generated RSA key pair and a stubbed
-JWKS fetcher — NO network access:
+JWKS fetcher — NO network access.  That harness (the key pairs, the
+JWKS builder, the token minting and the app runner) is shared with the
+other admin-plane route tests in :mod:`tests.admin_jwt_harness`; this
+module imports it and proves the auth CONTRACT itself:
 
 - ``GET /api/v1/admin/ping`` requires a valid Authentik OIDC JWT: a
   request with NO ``Authorization`` header and a MALFORMED header
@@ -41,28 +44,24 @@ import asyncio
 import base64
 import json
 import time
-from pathlib import Path
 
 import httpx
-import jwt
 import pytest
-from cryptography.hazmat.primitives.asymmetric import rsa
 
-from api.app import create_app
-from api.auth import ADMIN_GROUP, JwksCache
+from api.auth import JwksCache
 from api.config import ApiConfig, ConfigError
-
-#: The configured OIDC settings the test apps run with (values a real
-#: deployment would get from the NESTQUEST_OIDC_* environment vars).
-ISSUER = "https://authentik.example.com/application/o/nestquest/"
-AUDIENCE = "nestquest-api"
-JWKS_URL = "https://authentik.example.com/application/o/nestquest/jwks/"
-
-#: The ``kid`` the test tokens carry, matching the JWKS entry.
-KID = "admin-test-key-1"
-
-#: The subject a valid token authenticates as.
-SUBJECT = "admin-user-1"
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    AUDIENCE,
+    ISSUER,
+    JWKS_URL,
+    KID,
+    OTHER_KEY,
+    SUBJECT,
+    AdminRunner as _AdminRunner,
+    jwks_for as _jwks_for,
+    make_token as _make_token,
+)
 
 #: The uniform 401 detail every rejection must carry.
 REJECTION_DETAIL = "Missing or invalid admin credentials"
@@ -78,53 +77,6 @@ FORBIDDEN_DETAIL = (
 )
 
 
-def _generate_key() -> rsa.RSAPrivateKey:
-    """Generate one local RSA key pair (no network, no fixtures)."""
-    return rsa.generate_private_key(public_exponent=65537, key_size=2048)
-
-
-#: The key pair whose PUBLIC half the stubbed JWKS publishes — tokens
-#: signed with it verify; the second pair below proves they don't when
-#: signed otherwise.
-ADMIN_KEY = _generate_key()
-OTHER_KEY = _generate_key()
-
-
-def _b64url_uint(value: int) -> str:
-    """Base64url-encode a non-negative integer (no padding), per JWK."""
-    raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
-    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
-
-
-def _jwks_for(private_key: rsa.RSAPrivateKey, kid: str) -> dict[str, object]:
-    """Build a JWKS document publishing ``private_key``'s public half."""
-    numbers = private_key.public_key().public_numbers()
-    return {
-        "keys": [
-            {
-                "kty": "RSA",
-                "use": "sig",
-                "alg": "RS256",
-                "kid": kid,
-                "n": _b64url_uint(numbers.n),
-                "e": _b64url_uint(numbers.e),
-            }
-        ]
-    }
-
-
-class _StubbedFetch:
-    """A JWKS fetcher stub: returns a fixed document, counts calls."""
-
-    def __init__(self, document: dict[str, object]) -> None:
-        self.document = document
-        self.calls = 0
-
-    async def __call__(self, url: str) -> dict[str, object]:
-        self.calls += 1
-        return self.document
-
-
 class _FailingFetch:
     """A JWKS fetcher stub that always fails, counting the attempts."""
 
@@ -164,41 +116,6 @@ class _FailThenServe:
         return self.document
 
 
-def _make_token(
-    signing_key: rsa.RSAPrivateKey = ADMIN_KEY,
-    *,
-    kid: str | None = KID,
-    iss: str = ISSUER,
-    aud: object = AUDIENCE,
-    sub: str = SUBJECT,
-    expires_in: int | None = 300,
-    nbf_in: int | None = -1,
-    groups: object = (ADMIN_GROUP,),
-) -> str:
-    """Mint one RS256 JWT the way Authentik would (its claims, our key).
-
-    ``groups`` defaults to the admin group (what a real admin user's
-    token carries); tests pass a different value — or ``None`` for NO
-    groups claim at all — to prove the group gate rejects it.
-    """
-    now = int(time.time())
-    claims: dict[str, object] = {"iss": iss, "aud": aud, "sub": sub, "iat": now}
-    if groups is not None:
-        claims["groups"] = (
-            list(groups) if isinstance(groups, (list, tuple)) else groups
-        )
-    if nbf_in is not None:
-        claims["nbf"] = now + nbf_in
-    if expires_in is not None:
-        claims["exp"] = now + expires_in
-    headers: dict[str, str] = {}
-    if kid is not None:
-        headers["kid"] = kid
-    return jwt.encode(
-        claims, signing_key, algorithm="RS256", headers=headers
-    )
-
-
 def _token_with_header(
     header: dict[str, object], payload: dict[str, object]
 ) -> str:
@@ -226,79 +143,6 @@ def _verified_payload() -> dict[str, object]:
     }
 
 
-class _AdminRunner:
-    """Run the app's lifespan around an httpx AsyncClient.
-
-    Same pattern as test_api_health.py's ``_AppRunner``: httpx's ASGI
-    transport does not run the lifespan, so it is driven explicitly and
-    shut down in ``__aexit__``'s finally block even on failure.
-
-    The app's JWKS cache is replaced (before the lifespan runs) with
-    one built around a counting stub fetcher, so verification goes
-    through the REAL cache logic but never the network.
-    """
-
-    def __init__(
-        self,
-        db_path: str,
-        jwks_document: dict[str, object],
-        *,
-        ttl_seconds: float | None = None,
-        backoff_seconds: float | None = None,
-        kid_refetch_seconds: float | None = None,
-    ) -> None:
-        self.app = create_app(
-            ApiConfig(
-                db_path=db_path,
-                panel_token="admin-test-panel-token",
-                oidc_issuer=ISSUER,
-                oidc_audience=AUDIENCE,
-                oidc_jwks_url=JWKS_URL,
-            )
-        )
-        self.fetch = _StubbedFetch(jwks_document)
-        cache_kwargs: dict[str, float] = {}
-        if ttl_seconds is not None:
-            cache_kwargs["ttl_seconds"] = ttl_seconds
-        if backoff_seconds is not None:
-            cache_kwargs["backoff_seconds"] = backoff_seconds
-        if kid_refetch_seconds is not None:
-            cache_kwargs["kid_refetch_seconds"] = kid_refetch_seconds
-        self.app.state.jwks_cache = JwksCache(
-            JWKS_URL, fetcher=self.fetch, **cache_kwargs
-        )
-
-    async def __aenter__(self) -> httpx.AsyncClient:
-        self._lifespan = self.app.router.lifespan_context(self.app)
-        await self._lifespan.__aenter__()
-        self._client = httpx.AsyncClient(
-            transport=httpx.ASGITransport(app=self.app),
-            base_url="http://testserver",
-        )
-        return await self._client.__aenter__()
-
-    async def __aexit__(self, *exc) -> None:
-        try:
-            await self._client.__aexit__(*exc)
-        finally:
-            await self._lifespan.__aexit__(*exc)
-
-
-@pytest.fixture
-def temp_db_path(tmp_path: Path) -> str:
-    """A fresh per-test SQLite path under pytest's tmp_path."""
-    return str(tmp_path / "nestquest.db")
-
-
-@pytest.fixture
-async def admin_client(
-    temp_db_path: str,
-) -> httpx.AsyncClient:
-    """An app client whose provider publishes the admin test key."""
-    async with _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID)) as client:
-        yield client
-
-
 # --- the happy path -------------------------------------------------------
 
 

tests/test_api_admin_children.py

@@ -0,0 +1,464 @@
+"""Admin-plane children route tests (task e4845baa).
+
+These tests exercise the admin plane's children routes against the
+ASGI app with httpx, reusing the shared local-RSA-key/stubbed-JWKS
+harness (:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.
+
+Every done-condition case is proven against the DATABASE, read through
+the SAME core copy the app uses (``nestquest_core.*``; see the two-copy
+caveat in api/nestquest_core.py):
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and no row is written.
+- ``POST /api/v1/admin/children``: the created child EXISTS with the
+  given display name (and the optional colour/avatar_ref/sort_order).
+- ``PATCH /api/v1/admin/children/{id}``: an edit PERSISTS the supplied
+  fields and leaves omitted fields untouched (exclude_unset semantics —
+  an omitted field is never passed to the core).
+- ``PATCH /api/v1/admin/children/{id}/active`` with ``is_active: false``
+  sets the active flag FALSE and the row STILL EXISTS (children are
+  never hard-deleted).
+- ``POST /api/v1/admin/children/reorder``: the posted order becomes the
+  rows' sort_order; a list that is not a complete permutation is 422
+  and leaves the stored order untouched.
+
+Error mapping (documented in api/routes_admin.py): a core ValueError
+for a non-existent child is 404; every other core ValueError (an empty
+display name, an explicit null colour, a bad permutation) is 422 — the
+same failure class FastAPI reports for malformed bodies.
+"""
+from __future__ import annotations
+
+import importlib
+from types import SimpleNamespace
+
+import httpx
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The children business layer through the SAME core copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy) —
+# used for the tests' database assertions only; the routes reach it
+# through api.nestquest_core.
+_children = importlib.import_module("nestquest_core.children")
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+async def _children_by_id(database: object) -> dict[int, object]:
+    """Every child row through the app's core copy, keyed by id."""
+    records = await _children.list_children(database)
+    return {record.id: record for record in records}
+
+
+@pytest.fixture
+async def admin_children(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth tests use (stubbed JWKS, no network); the
+    fixture also exposes the app's ONE database connection so tests
+    assert the routes' effects on the stored rows themselves.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_children_routes_refuse_non_admin_and_absent_credentials(
+    admin_children: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_children.client
+    body = {"display_name": "Ada"}
+    non_admin = make_token(groups=("some-other-group",))
+
+    forbidden = await client.post(
+        "/api/v1/admin/children",
+        json=body,
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+    unauthenticated = await client.post("/api/v1/admin/children", json=body)
+    assert unauthenticated.status_code == 401
+
+    unauthenticated_list = await client.get("/api/v1/admin/children")
+    assert unauthenticated_list.status_code == 401
+
+    # No route ran its handler: the household is still empty.
+    assert await _children_by_id(admin_children.database) == {}
+
+
+# --- create -----------------------------------------------------------------
+
+
+async def test_create_child_persists_the_given_name(
+    admin_children: SimpleNamespace,
+) -> None:
+    """POST /children creates the row; the child exists with that name."""
+    response = await admin_children.client.post(
+        "/api/v1/admin/children",
+        json={
+            "display_name": "Ada",
+            "colour": "#ff8800",
+            "avatar_ref": "avatar:bee",
+            "sort_order": 2,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 201
+    body = response.json()
+    assert body["display_name"] == "Ada"
+    assert body["colour"] == "#ff8800"
+    assert body["avatar_ref"] == "avatar:bee"
+    assert body["sort_order"] == 2
+    assert body["is_active"] is True
+
+    stored = (await _children_by_id(admin_children.database))[body["id"]]
+    assert stored.display_name == "Ada"
+    assert stored.colour == "#ff8800"
+    assert stored.avatar_ref == "avatar:bee"
+    assert stored.sort_order == 2
+    assert stored.is_active is True
+
+
+async def test_create_child_with_only_a_name_uses_core_defaults(
+    admin_children: SimpleNamespace,
+) -> None:
+    """Omitted optional fields default through the core layer."""
+    response = await admin_children.client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Bo"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 201
+    stored = (await _children_by_id(admin_children.database))[
+        response.json()["id"]
+    ]
+    assert stored.display_name == "Bo"
+    assert stored.colour is None
+    assert stored.avatar_ref is None
+    assert stored.sort_order == 0
+    assert stored.is_active is True
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        {},  # display_name missing
+        {"display_name": 5},  # wrong type
+        {"display_name": ""},  # empty: the core rejects it
+        {"display_name": "   "},  # whitespace-only: empty to a parent
+        {"display_name": "Ada", "sort_order": "first"},  # wrong type
+    ],
+)
+async def test_create_child_malformed_input_is_422(
+    admin_children: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """Malformed bodies are 422 and never create a row."""
+    response = await admin_children.client.post(
+        "/api/v1/admin/children", json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+    assert await _children_by_id(admin_children.database) == {}
+
+
+# --- edit -------------------------------------------------------------------
+
+
+async def test_edit_child_persists_supplied_fields_and_skips_omitted_ones(
+    admin_children: SimpleNamespace,
+) -> None:
+    """PATCH persists exactly the supplied fields (exclude_unset)."""
+    client = admin_children.client
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={
+            "display_name": "Ada",
+            "colour": "#ff8800",
+            "avatar_ref": "avatar:bee",
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    child_id = created.json()["id"]
+
+    # Only display_name and sort_order: colour/avatar_ref must survive.
+    edited = await client.patch(
+        f"/api/v1/admin/children/{child_id}",
+        json={"display_name": "Ada Lovelace", "sort_order": 4},
+        headers=_admin_headers(),
+    )
+    assert edited.status_code == 200
+    assert edited.json()["display_name"] == "Ada Lovelace"
+    assert edited.json()["sort_order"] == 4
+    stored = (await _children_by_id(admin_children.database))[child_id]
+    assert stored.display_name == "Ada Lovelace"
+    assert stored.sort_order == 4
+    # The omitted fields were never passed to the core: unchanged.
+    assert stored.colour == "#ff8800"
+    assert stored.avatar_ref == "avatar:bee"
+
+    # A second edit touching only colour leaves the name alone.
+    recoloured = await client.patch(
+        f"/api/v1/admin/children/{child_id}",
+        json={"colour": "#00aa55"},
+        headers=_admin_headers(),
+    )
+    assert recoloured.status_code == 200
+    stored = (await _children_by_id(admin_children.database))[child_id]
+    assert stored.colour == "#00aa55"
+    assert stored.display_name == "Ada Lovelace"
+
+
+async def test_edit_unknown_child_is_404(
+    admin_children: SimpleNamespace,
+) -> None:
+    """An edit naming a child that does not exist is 404."""
+    response = await admin_children.client.patch(
+        "/api/v1/admin/children/999",
+        json={"display_name": "Nobody"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Child not found"
+
+
+async def test_edit_explicit_null_colour_is_422_and_changes_nothing(
+    admin_children: SimpleNamespace,
+) -> None:
+    """An explicit null (clearing) is passed through and rejected by the core.
+
+    Only the fields the request SUPPLIES are forwarded: a null colour is
+    a supplied None, which the core refuses (the DAO cannot express
+    NULL) — mapped to 422, with the stored colour left untouched.
+    """
+    created = await admin_children.client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Ada", "colour": "#ff8800"},
+        headers=_admin_headers(),
+    )
+    child_id = created.json()["id"]
+    response = await admin_children.client.patch(
+        f"/api/v1/admin/children/{child_id}",
+        json={"colour": None},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    stored = (await _children_by_id(admin_children.database))[child_id]
+    assert stored.colour == "#ff8800"
+
+
+# --- set_active -------------------------------------------------------------
+
+
+async def test_deactivation_sets_the_flag_false_and_never_deletes(
+    admin_children: SimpleNamespace,
+) -> None:
+    """is_active=false deactivates the row; the row STILL exists."""
+    client = admin_children.client
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Ada"},
+        headers=_admin_headers(),
+    )
+    child_id = created.json()["id"]
+
+    deactivated = await client.patch(
+        f"/api/v1/admin/children/{child_id}/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert deactivated.status_code == 200
+    assert deactivated.json()["is_active"] is False
+
+    # Database proof: the flag is False and the row was NOT hard-deleted.
+    rows = await _children_by_id(admin_children.database)
+    stored = rows[child_id]
+    assert stored.is_active is False
+    assert stored.display_name == "Ada"
+
+    # Reactivation goes through the same route: the flag is set, not
+    # one-way.
+    reactivated = await client.patch(
+        f"/api/v1/admin/children/{child_id}/active",
+        json={"is_active": True},
+        headers=_admin_headers(),
+    )
+    assert reactivated.status_code == 200
+    assert reactivated.json()["is_active"] is True
+    rows = await _children_by_id(admin_children.database)
+    assert rows[child_id].is_active is True
+
+
+async def test_set_active_unknown_child_is_404(
+    admin_children: SimpleNamespace,
+) -> None:
+    """A set_active naming a child that does not exist is 404."""
+    response = await admin_children.client.patch(
+        "/api/v1/admin/children/999/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Child not found"
+
+
+async def test_set_active_malformed_body_is_422(
+    admin_children: SimpleNamespace,
+) -> None:
+    """A non-bool is_active is rejected by the body model (422)."""
+    response = await admin_children.client.patch(
+        "/api/v1/admin/children/1/active",
+        json={"is_active": "maybe"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+
+
+# --- reorder ----------------------------------------------------------------
+
+
+async def test_reorder_writes_the_posted_order_to_sort_order(
+    admin_children: SimpleNamespace,
+) -> None:
+    """The posted id order becomes the rows' sort_order."""
+    client = admin_children.client
+    ids = []
+    for name in ("Ada", "Bo", "Cyd"):
+        created = await client.post(
+            "/api/v1/admin/children",
+            json={"display_name": name},
+            headers=_admin_headers(),
+        )
+        assert created.status_code == 201
+        ids.append(created.json()["id"])
+
+    response = await client.post(
+        "/api/v1/admin/children/reorder",
+        json={"ordered_ids": [ids[2], ids[0], ids[1]]},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}
+
+    # Database proof: each posted position became that child's sort_order.
+    rows = await _children_by_id(admin_children.database)
+    assert rows[ids[2]].sort_order == 0
+    assert rows[ids[0]].sort_order == 1
+    assert rows[ids[1]].sort_order == 2
+
+
+async def test_reorder_rejects_non_permutations_and_writes_nothing(
+    admin_children: SimpleNamespace,
+) -> None:
+    """A list that is not a complete permutation is 422, order intact.
+
+    The core rejects the list BEFORE any write, so the stored sort_order
+    values survive every rejected call unchanged.
+    """
+    client = admin_children.client
+    ids = []
+    for name in ("Ada", "Bo"):
+        created = await client.post(
+            "/api/v1/admin/children",
+            json={"display_name": name},
+            headers=_admin_headers(),
+        )
+        assert created.status_code == 201
+        ids.append(created.json()["id"])
+
+    bad_lists = [
+        ids[:1],  # partial: missing an existing child
+        [ids[0], ids[0], ids[1]],  # duplicate id
+        [*ids, ids[1] + 999],  # unknown id
+    ]
+    for ordered_ids in bad_lists:
+        response = await client.post(
+            "/api/v1/admin/children/reorder",
+            json={"ordered_ids": ordered_ids},
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 422, ordered_ids
+
+    # A non-integer id never reaches the core either (body model).
+    malformed = await client.post(
+        "/api/v1/admin/children/reorder",
+        json={"ordered_ids": ["first", "second"]},
+        headers=_admin_headers(),
+    )
+    assert malformed.status_code == 422
+
+    # Database proof: no write happened for any rejected call.
+    rows = await _children_by_id(admin_children.database)
+    assert [rows[child_id].sort_order for child_id in ids] == [0, 0]
+
+
+# --- list (the read model the admin PWA re-renders from) ---------------------
+
+
+async def test_list_children_returns_all_in_display_order(
+    admin_children: SimpleNamespace,
+) -> None:
+    """GET /children returns every child (inactive included) in order."""
+    client = admin_children.client
+    ada = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Ada", "sort_order": 5},
+        headers=_admin_headers(),
+    )
+    bo = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Bo", "sort_order": 1},
+        headers=_admin_headers(),
+    )
+    deactivated = await client.patch(
+        f"/api/v1/admin/children/{ada.json()['id']}/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert deactivated.status_code == 200
+
+    response = await client.get(
+        "/api/v1/admin/children", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    children = response.json()["children"]
+    assert [child["id"] for child in children] == [
+        bo.json()["id"],
+        ada.json()["id"],
+    ]
+    by_id = {child["id"]: child for child in children}
+    assert by_id[ada.json()["id"]]["is_active"] is False
+    assert by_id[bo.json()["id"]]["is_active"] is True
+    # The serialized shape is the documented child payload.
+    assert set(by_id[bo.json()["id"]]) == {
+        "id",
+        "display_name",
+        "colour",
+        "avatar_ref",
+        "sort_order",
+        "is_active",
+    }

tests/test_api_admin_history.py

@@ -0,0 +1,547 @@
+"""Admin-plane history route tests (task 2d2dda53).
+
+These tests exercise the admin plane's history query and CSV export
+routes against the ASGI app with httpx, reusing the shared
+local-RSA-key/stubbed-JWKS harness (:mod:`tests.admin_jwt_harness`) so
+an admin JWT genuinely authenticates — signature, issuer, audience,
+expiry and the nestquest-admins group — with NO network access.  The
+household is seeded with :func:`tests.test_api_panel._seed_household`
+through the SAME core copy the app uses (``nestquest_core.*``; the
+two-copy caveat in api/nestquest_core.py), and the reversal the
+history tests read is appended through that same copy's
+:func:`nestquest_core.completion.uncomplete_instance` — so the rows
+the routes report are the rows the core's own write paths produced.
+
+The seeded household (clock pinned to noon, the route's ``_local_now``
+pinned to the same instant so nothing straddles midnight):
+
+- Bo's ``Brush teeth`` TODAY instance: a ``completed`` event (panel
+  tap, on time) plus the appended ``uncompleted`` reversal — the
+  completion and the reversal the ``all`` and ``reversals`` filters
+  report.
+- Ada's ``Pack bag`` and ``Stale chore`` YESTERDAY instances and Bo's
+  ``Brush teeth`` YESTERDAY instance: open, past due, no events — the
+  DERIVED missed rows the ``missed`` filter reports.
+
+Proven per the done-condition, everything read back through the API:
+
+- ``GET /api/v1/admin/history`` returns the expected rows for EACH
+  filter — ``all`` (the completed event AND the reversal),
+  ``reversals`` (the uncompleted event only) and ``missed`` (the
+  derived past-due open instances, actor ``nightly sweep``, no
+  ``occurred_at``) — and the range scopes by the INSTANCE's due date
+  (the Feature 13 rule), so the events sit under their instance's due
+  day, not the wall-clock day.
+- ``GET /api/v1/admin/history.csv`` answers ``text/csv`` with a
+  ``Content-Disposition`` attachment filename, and the parsed body
+  (csv.reader) carries the documented header columns and exactly the
+  filtered rows.
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: a non-admin JWT is 403 and an absent credential is 401
+  on both routes.
+- Malformed input — an unknown filter, a non-strict date, an inverted
+  range, a missing parameter — is 422 on both routes (never a 500).
+"""
+from __future__ import annotations
+
+import csv
+import datetime
+import importlib
+import io
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from api import routes_admin
+from api.database import _executor
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    SUBJECT,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+from tests.test_api_panel import _seed_household
+
+# The completion, instance-DAO and db layers through the SAME core copy
+# the app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' seeding only; the
+# routes reach them through api.nestquest_core.
+_completion = importlib.import_module("nestquest_core.completion")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_core_db = importlib.import_module("nestquest_core.db")
+
+#: The documented CSV columns (Admin spec §5 event-row fields), in
+#: header order — asserted literally so the contract is pinned here,
+#: independent of the core constant.
+DOCUMENTED_CSV_COLUMNS = [
+    "occurred_at",
+    "event_type",
+    "child_name",
+    "child_id",
+    "quest_title",
+    "instance_id",
+    "window",
+    "due_date",
+    "due_time",
+    "actor",
+    "was_on_time",
+]
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _pinned_noon() -> tuple[datetime.date, datetime.datetime]:
+    """One host clock read pinned to 12:00 UTC that day.
+
+    The seed's completion, the reversal and the route's clock all
+    anchor to this instant (the reversal five minutes later), so
+    nothing can straddle midnight mid-test.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    return today, datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
+    )
+
+
+@pytest.fixture
+async def history_household(temp_db_path: str, monkeypatch) -> SimpleNamespace:
+    """A seeded household served by the admin-plane app, clock pinned.
+
+    Beyond :func:`_seed_household`'s rows (Bo's today completion, the
+    past-due open yesterday instances), the fixture appends ONE
+    ``uncompleted`` reversal on Bo's today instance through the core
+    completion layer — attributed to the harness subject, stamped five
+    minutes after the pinned completion — and exposes the yesterday
+    instances the missed filter must report.
+    """
+    today, pinned_now = _pinned_noon()
+    yesterday = today - datetime.timedelta(days=1)
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+
+    database = _core_db.NestQuestDatabase(_executor)
+    await database.open(temp_db_path)
+    try:
+        instances = _dao_instances.QuestInstancesDao(database)
+        ada_yesterday = await instances.list_by_child_and_date(
+            seed.ada.id, yesterday.isoformat()
+        )
+        pack_bag_yesterday = next(
+            record for record in ada_yesterday if record.due_time == "10:00"
+        )
+        bo_yesterday = await instances.list_by_child_and_date(
+            seed.bo.id, yesterday.isoformat()
+        )
+        brush_teeth_yesterday = next(
+            record for record in bo_yesterday if record.due_time == "17:00"
+        )
+        await _completion.uncomplete_instance(
+            database,
+            seed.brush_teeth_instance.id,
+            actor_source="user",
+            actor_user_id=SUBJECT,
+            now=pinned_now + datetime.timedelta(minutes=5),
+            today=today,
+        )
+    finally:
+        await database.close()
+
+    monkeypatch.setattr(routes_admin, "_local_now", lambda: pinned_now)
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            seed=seed,
+            database=runner.app.state.db.database,
+            today=today,
+            yesterday=yesterday,
+            pack_bag_yesterday=pack_bag_yesterday,
+            brush_teeth_yesterday=brush_teeth_yesterday,
+        )
+
+
+# --- authentication: the ONE require_admin dependency guards both routes ----
+
+
+async def test_history_routes_refuse_non_admin_and_absent(
+    history_household: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401, both routes."""
+    client = history_household.client
+    query = f"start={history_household.yesterday.isoformat()}"
+    query += f"&end={history_household.today.isoformat()}"
+    non_admin = make_token(groups=("some-other-group",))
+    for path in ("/api/v1/admin/history", "/api/v1/admin/history.csv"):
+        forbidden = await client.get(
+            f"{path}?{query}",
+            headers={"Authorization": f"Bearer {non_admin}"},
+        )
+        assert forbidden.status_code == 403, path
+        assert forbidden.json()["detail"] == (
+            "Admin access requires membership in the nestquest-admins group"
+        )
+        unauthenticated = await client.get(f"{path}?{query}")
+        assert unauthenticated.status_code == 401, path
+
+
+# --- the JSON history query: the expected rows per filter --------------------
+
+
+async def test_history_all_returns_completed_and_reversal(
+    history_household: SimpleNamespace,
+) -> None:
+    """``all`` returns the completed event AND the reversal, in order.
+
+    Both events belong to Bo's Brush teeth instance due TODAY, so the
+    due-date-scoped range [yesterday, today] reports them under today;
+    the rows are ordered by due date, then instance, then event id —
+    the completed row before its reversal.
+    """
+    client = history_household.client
+    seed = history_household.seed
+    brush = seed.brush_teeth_instance
+    yesterday_iso = history_household.yesterday.isoformat()
+    today_iso = history_household.today.isoformat()
+
+    response = await client.get(
+        "/api/v1/admin/history",
+        params={"filter": "all", "start": yesterday_iso, "end": today_iso},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["filter"] == "all"
+    assert body["start"] == yesterday_iso
+    assert body["end"] == today_iso
+    assert body["count"] == 2
+
+    completed, reversal = body["rows"]
+    assert completed == {
+        "occurred_at": f"{today_iso}T12:00:00+00:00",
+        "event_type": "completed",
+        "child_name": "Bo",
+        "child_id": brush.child_id,
+        "quest_title": "Brush teeth",
+        "instance_id": brush.id,
+        "window": "morning",
+        "due_date": today_iso,
+        "due_time": "17:00",
+        "actor": "panel (Bo)",
+        "was_on_time": True,
+    }
+    assert reversal == {
+        "occurred_at": f"{today_iso}T12:05:00+00:00",
+        "event_type": "uncompleted",
+        "child_name": "Bo",
+        "child_id": brush.child_id,
+        "quest_title": "Brush teeth",
+        "instance_id": brush.id,
+        "window": "morning",
+        "due_date": today_iso,
+        "due_time": "17:00",
+        "actor": SUBJECT,
+        "was_on_time": None,
+    }
+
+
+async def test_history_reversals_returns_only_the_uncompleted_event(
+    history_household: SimpleNamespace,
+) -> None:
+    """``reversals`` returns the uncompleted event only, user-attributed."""
+    client = history_household.client
+    seed = history_household.seed
+    brush = seed.brush_teeth_instance
+    today_iso = history_household.today.isoformat()
+
+    response = await client.get(
+        "/api/v1/admin/history",
+        params={
+            "filter": "reversals",
+            "start": today_iso,
+            "end": today_iso,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["filter"] == "reversals"
+    assert body["count"] == 1
+    (reversal,) = body["rows"]
+    assert reversal["event_type"] == "uncompleted"
+    assert reversal["instance_id"] == brush.id
+    assert reversal["actor"] == SUBJECT
+    assert reversal["occurred_at"] == f"{today_iso}T12:05:00+00:00"
+    assert reversal["was_on_time"] is None
+
+
+async def test_history_missed_returns_derived_missed_instances(
+    history_household: SimpleNamespace,
+) -> None:
+    """``missed`` returns the DERIVED past-due open instances.
+
+    Three seeded instances are open and due yesterday: Ada's Pack bag
+    and Stale chore and Bo's Brush teeth.  Each is a DERIVED row —
+    never a stored event: no ``occurred_at``, ``was_on_time`` None and
+    the spec's ``nightly sweep`` actor.
+    """
+    client = history_household.client
+    seed = history_household.seed
+    yesterday_iso = history_household.yesterday.isoformat()
+
+    response = await client.get(
+        "/api/v1/admin/history",
+        params={
+            "filter": "missed",
+            "start": yesterday_iso,
+            "end": yesterday_iso,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["filter"] == "missed"
+    assert body["count"] == 3
+
+    rows = body["rows"]
+    assert all(row["event_type"] == "missed" for row in rows)
+    assert all(row["occurred_at"] is None for row in rows)
+    assert all(row["was_on_time"] is None for row in rows)
+    assert all(row["actor"] == "nightly sweep" for row in rows)
+    assert all(row["due_date"] == yesterday_iso for row in rows)
+    assert {
+        (row["child_name"], row["quest_title"], row["due_time"])
+        for row in rows
+    } == {
+        ("Ada", "Pack bag", "10:00"),
+        ("Bo", "Brush teeth", "17:00"),
+        ("Ada", "Stale chore", "09:00"),
+    }
+    assert {
+        row["instance_id"]
+        for row in rows
+    } == {
+        history_household.pack_bag_yesterday.id,
+        history_household.brush_teeth_yesterday.id,
+        seed.stale_instance.id,
+    }
+    # Deterministic order: due date, then instance id.
+    keys = [(row["due_date"], row["instance_id"]) for row in rows]
+    assert keys == sorted(keys)
+
+
+async def test_history_range_scopes_by_the_instance_due_date(
+    history_household: SimpleNamespace,
+) -> None:
+    """The range scopes by the INSTANCE's due date, not the wall clock.
+
+    Both events OCCURRED today but belong to the instance due today, so
+    a yesterday-only ``all`` range reports nothing; the missed filter
+    over a today-only range reports nothing either (a due-TODAY open
+    instance is not past due).
+    """
+    client = history_household.client
+    yesterday_iso = history_household.yesterday.isoformat()
+    today_iso = history_household.today.isoformat()
+
+    yesterday_all = await client.get(
+        "/api/v1/admin/history",
+        params={"filter": "all", "start": yesterday_iso, "end": yesterday_iso},
+        headers=_admin_headers(),
+    )
+    assert yesterday_all.status_code == 200
+    assert yesterday_all.json()["count"] == 0
+
+    today_missed = await client.get(
+        "/api/v1/admin/history",
+        params={"filter": "missed", "start": today_iso, "end": today_iso},
+        headers=_admin_headers(),
+    )
+    assert today_missed.status_code == 200
+    assert today_missed.json()["count"] == 0
+
+
+# --- the CSV export: documented columns and the filtered rows ----------------
+
+
+async def test_history_csv_carries_documented_columns_and_all_rows(
+    history_household: SimpleNamespace,
+) -> None:
+    """The CSV export carries the documented header and the filtered rows.
+
+    The body is parsed with csv.reader: the first row is the documented
+    column set, the remaining rows are exactly the ``all`` filter's
+    rows (the completion and its reversal) in the same order the JSON
+    query reports them.
+    """
+    client = history_household.client
+    seed = history_household.seed
+    brush = seed.brush_teeth_instance
+    yesterday_iso = history_household.yesterday.isoformat()
+    today_iso = history_household.today.isoformat()
+
+    response = await client.get(
+        "/api/v1/admin/history.csv",
+        params={"filter": "all", "start": yesterday_iso, "end": today_iso},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.headers["content-type"].startswith("text/csv")
+    disposition = response.headers["content-disposition"]
+    assert disposition.startswith("attachment;")
+    assert (
+        f'filename="nestquest-history-{yesterday_iso}-to-{today_iso}.csv"'
+        in disposition
+    )
+
+    parsed = list(csv.reader(io.StringIO(response.text)))
+    assert parsed[0] == DOCUMENTED_CSV_COLUMNS
+    assert len(parsed) == 3  # header + completed + reversal
+    assert parsed[1] == [
+        f"{today_iso}T12:00:00+00:00",
+        "completed",
+        "Bo",
+        str(brush.child_id),
+        "Brush teeth",
+        str(brush.id),
+        "morning",
+        today_iso,
+        "17:00",
+        "panel (Bo)",
+        "true",
+    ]
+    assert parsed[2] == [
+        f"{today_iso}T12:05:00+00:00",
+        "uncompleted",
+        "Bo",
+        str(brush.child_id),
+        "Brush teeth",
+        str(brush.id),
+        "morning",
+        today_iso,
+        "17:00",
+        SUBJECT,
+        "",
+    ]
+
+
+async def test_history_csv_missed_rows_are_derived_rows(
+    history_household: SimpleNamespace,
+) -> None:
+    """The CSV export of the ``missed`` filter carries the derived rows.
+
+    Same filtered range as the JSON query's missed answer: three rows
+    with an empty ``occurred_at`` (no event exists), the ``missed``
+    event type and the ``nightly sweep`` actor.
+    """
+    client = history_household.client
+    seed = history_household.seed
+    yesterday_iso = history_household.yesterday.isoformat()
+
+    response = await client.get(
+        "/api/v1/admin/history.csv",
+        params={
+            "filter": "missed",
+            "start": yesterday_iso,
+            "end": yesterday_iso,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    parsed = list(csv.reader(io.StringIO(response.text)))
+    assert parsed[0] == DOCUMENTED_CSV_COLUMNS
+    assert len(parsed) == 4  # header + the three derived missed rows
+    missed_rows = parsed[1:]
+    assert all(row[1] == "missed" for row in missed_rows)
+    assert all(row[0] == "" for row in missed_rows)
+    assert all(row[9] == "nightly sweep" for row in missed_rows)
+    assert all(row[10] == "" for row in missed_rows)
+    assert {
+        (row[2], row[4], row[8]) for row in missed_rows
+    } == {
+        ("Ada", "Pack bag", "10:00"),
+        ("Bo", "Brush teeth", "17:00"),
+        ("Ada", "Stale chore", "09:00"),
+    }
+    assert {row[5] for row in missed_rows} == {
+        str(history_household.pack_bag_yesterday.id),
+        str(history_household.brush_teeth_yesterday.id),
+        str(seed.stale_instance.id),
+    }
+
+
+# --- validation: malformed filter and date range are 422, never 500 ----------
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        {"filter": "everything", "start": "2026-03-01", "end": "2026-03-02"},
+        {"filter": "all", "start": "2026/03/01", "end": "2026-03-02"},
+        {"filter": "all", "start": "2026-3-1", "end": "2026-03-02"},
+        {"filter": "all", "start": "2026-03-02", "end": "2026-03-01"},
+        {"filter": "all", "start": "not-a-date", "end": "2026-03-02"},
+    ],
+)
+async def test_history_query_rejects_malformed_params_as_422(
+    history_household: SimpleNamespace, params: dict[str, str]
+) -> None:
+    """An unknown filter or a rejected range is 422 naming the field."""
+    client = history_household.client
+    response = await client.get(
+        "/api/v1/admin/history", params=params, headers=_admin_headers()
+    )
+    assert response.status_code == 422, params
+    detail = response.json()["detail"]
+    assert any(
+        name in detail for name in ("filter", "start", "end")
+    ), detail
+
+
+async def test_history_csv_rejects_malformed_params_as_422(
+    history_household: SimpleNamespace,
+) -> None:
+    """The CSV route maps the same rejections to 422."""
+    client = history_household.client
+    unknown_filter = await client.get(
+        "/api/v1/admin/history.csv",
+        params={"filter": "galaxy", "start": "2026-03-01", "end": "2026-03-02"},
+        headers=_admin_headers(),
+    )
+    assert unknown_filter.status_code == 422
+    assert "filter" in unknown_filter.json()["detail"]
+
+    inverted = await client.get(
+        "/api/v1/admin/history.csv",
+        params={"filter": "all", "start": "2026-03-02", "end": "2026-03-01"},
+        headers=_admin_headers(),
+    )
+    assert inverted.status_code == 422
+    assert "end must be on or after start" in inverted.json()["detail"]
+
+
+async def test_history_query_missing_params_are_422(
+    history_household: SimpleNamespace,
+) -> None:
+    """A missing required query parameter is FastAPI's 422 (the shape)."""
+    client = history_household.client
+    no_params = await client.get(
+        "/api/v1/admin/history", headers=_admin_headers()
+    )
+    assert no_params.status_code == 422
+    missing_start = await client.get(
+        "/api/v1/admin/history",
+        params={"end": "2026-03-02"},
+        headers=_admin_headers(),
+    )
+    assert missing_start.status_code == 422
+    missing_end = await client.get(
+        "/api/v1/admin/history",
+        params={"start": "2026-03-01"},
+        headers=_admin_headers(),
+    )
+    assert missing_end.status_code == 422

tests/test_api_admin_presence.py

@@ -0,0 +1,571 @@
+"""Admin-plane presence route tests (task e4a9d31b).
+
+These tests exercise the admin plane's presence routes against the ASGI
+app with httpx, reusing the shared local-RSA-key/stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.
+
+Every done-condition case is proven against the DATABASE and the DOMAIN,
+read through the SAME core copy the app uses (``nestquest_core.*``; see
+the two-copy caveat in api/nestquest_core.py):
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and nothing is written.
+- ``PUT /api/v1/admin/children/{id}/presence-schedule`` UPSERTS: the
+  first set persists the schedule, setting again REPLACES it — one
+  schedule row per child, never a duplicate.
+- ``POST /api/v1/admin/presence-overrides`` creates the override (its
+  ``id`` comes back) and ``DELETE /api/v1/admin/presence-overrides/{id}``
+  removes it again.
+- The child's RESOLVED presence changes accordingly: the pure
+  :class:`~nestquest_core.presence.PresenceEngine` built from the
+  stored rows answers differently before/after each write — a schedule
+  restricts the child's present weekdays, an override beats the pattern
+  for its range, and deleting the override restores the pattern's
+  answer.
+- The regeneration side effect lives in the core layer the route
+  calls: setting an all-absent schedule through the route deletes the
+  child's future open quest instances and they are NOT rebuilt while
+  the child is absent every day — the same proof the HA service test
+  makes, through the API instead.
+
+Error mapping (documented in api/routes_admin.py): a core ValueError
+for a non-existent child (schedule or override create) or override
+(delete) is 404; every other core ValueError — a rejected schedule
+(cycle length, pattern coverage, anchor date) or override (malformed or
+inverted dates, an overlapping range) — is 422, the same failure class
+FastAPI reports for malformed bodies.
+"""
+from __future__ import annotations
+
+import datetime
+import importlib
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The presence model, DAOs, business layer and materializer through the
+# SAME core copy the app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' database and resolved-
+# presence assertions only; the routes reach them through api.nestquest_core.
+_children = importlib.import_module("nestquest_core.children")
+_dao_presence = importlib.import_module("nestquest_core.dao_presence")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_materialize = importlib.import_module("nestquest_core.materialize")
+_presence = importlib.import_module("nestquest_core.presence")
+
+#: A Monday anchoring every schedule: cycle week 0 starts here.
+ANCHOR_MONDAY = "2026-01-05"
+#: The Tuesday of that week — absent under the test schedule.
+TUESDAY = "2026-01-06"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _schedule_body(
+    cycle_length_weeks: int = 1, pattern: dict[int, list[int]] | None = None
+) -> dict[str, object]:
+    """A valid schedule body anchored to the test Monday."""
+    return {
+        "cycle_length_weeks": cycle_length_weeks,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [0, 2, 4]} if pattern is None else pattern,
+    }
+
+
+def _override_body(
+    child_id: int,
+    *,
+    start_date: str = TUESDAY,
+    end_date: str = TUESDAY,
+    is_present: bool = True,
+) -> dict[str, object]:
+    """A valid single-day override body for the test child."""
+    return {
+        "child_id": child_id,
+        "start_date": start_date,
+        "end_date": end_date,
+        "is_present": is_present,
+        "note": "grandmother's week",
+    }
+
+
+async def _create_child(client: object, name: str = "Ada") -> int:
+    """Create one child through the admin children route; return the id."""
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": name},
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+def _engine(
+    database: object,
+    child_id: int,
+    schedule_record: object | None,
+    override_records: list[object],
+) -> object:
+    """Build the pure PresenceEngine from stored rows (the snapshot shape).
+
+    Mirrors what the core snapshot builder assembles: the child's
+    decoded schedule (absent → present every day) plus its overrides,
+    so :meth:`PresenceEngine.is_present` answers from the SAME rows the
+    routes produced.
+    """
+    schedules: dict[int, object] = {}
+    if schedule_record is not None:
+        schedules[child_id] = _presence.PresenceSchedule.decode(
+            schedule_record.child_id,
+            schedule_record.anchor_date,
+            schedule_record.pattern,
+        )
+    overrides = {
+        child_id: [
+            _presence.PresenceOverride(
+                record.child_id,
+                record.start_date,
+                record.end_date,
+                record.is_present,
+                note=record.note,
+            )
+            for record in override_records
+        ]
+    }
+    return _presence.PresenceEngine(schedules, overrides)
+
+
+async def _child_state(
+    database: object, child_id: int
+) -> tuple[object | None, list[object], object]:
+    """The child's stored schedule record, overrides and PresenceEngine."""
+    schedule_record = await _dao_presence.PresenceSchedulesDao(
+        database
+    ).get_by_child(child_id)
+    override_records = await _dao_presence.PresenceOverridesDao(
+        database
+    ).list_by_child_and_range(child_id, "2020-01-01", "2030-01-01")
+    engine = _engine(database, child_id, schedule_record, override_records)
+    return schedule_record, override_records, engine
+
+
+@pytest.fixture
+async def admin_presence(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth, children and quest-definition tests use
+    (stubbed JWKS, no network); the fixture also exposes the app's ONE
+    database connection so tests assert the routes' effects on the
+    stored rows and the resolved presence built from them.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_presence_routes_refuse_non_admin_and_absent_credentials(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    non_admin = make_token(groups=("some-other-group",))
+    non_admin_headers = {"Authorization": f"Bearer {non_admin}"}
+
+    forbidden_schedule = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=non_admin_headers,
+    )
+    assert forbidden_schedule.status_code == 403
+    assert forbidden_schedule.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+    forbidden_override = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=non_admin_headers,
+    )
+    assert forbidden_override.status_code == 403
+
+    forbidden_delete = await client.delete(
+        "/api/v1/admin/presence-overrides/1", headers=non_admin_headers
+    )
+    assert forbidden_delete.status_code == 403
+
+    unauthenticated_schedule = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+    )
+    assert unauthenticated_schedule.status_code == 401
+
+    unauthenticated_override = await client.post(
+        "/api/v1/admin/presence-overrides", json=_override_body(child_id)
+    )
+    assert unauthenticated_override.status_code == 401
+
+    unauthenticated_delete = await client.delete(
+        "/api/v1/admin/presence-overrides/1"
+    )
+    assert unauthenticated_delete.status_code == 401
+
+    # No route ran its handler: the child still has no schedule and no
+    # overrides (and is present every day, the no-schedule default).
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert _schedule is None
+    assert _overrides == []
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+
+
+# --- schedule upsert --------------------------------------------------------
+
+
+async def test_schedule_upserts_and_replaces_one_row_per_child(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """PUT /presence-schedule persists, and setting again REPLACES it."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+
+    first = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    assert first.json() == {
+        "child_id": child_id,
+        "cycle_length_weeks": 1,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [0, 2, 4]},
+    }
+    # Database proof: exactly ONE schedule row (the schema's
+    # UNIQUE(child_id) makes a second row unrepresentable), stored as
+    # encoded CSV.
+    schedule, _overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert schedule is not None
+    row_id = schedule.id
+    assert schedule.cycle_length_weeks == 1
+    assert schedule.anchor_date == ANCHOR_MONDAY
+    assert schedule.pattern == "0,2,4"
+
+    # Setting again UPSERTS: the SAME row is updated in place — new
+    # values, unchanged id — never a duplicate or a stale first set.
+    second = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(
+            cycle_length_weeks=2, pattern={"0": [], "1": [1, 3]}
+        ),
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json() == {
+        "child_id": child_id,
+        "cycle_length_weeks": 2,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [], "1": [1, 3]},
+    }
+    schedule, _overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert schedule is not None
+    assert schedule.id == row_id
+    assert schedule.cycle_length_weeks == 2
+    assert schedule.pattern == "|1,3"
+
+
+async def test_schedule_changes_the_childs_resolved_presence(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """The resolved presence (PresenceEngine) follows the stored schedule."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+
+    # Before any schedule the child is present every day.
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+    assert engine.is_present(child_id, TUESDAY) is True
+
+    # Mon/Wed/Fri only, week 0 of a one-week cycle.
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+    assert engine.is_present(child_id, TUESDAY) is False
+
+
+async def test_schedule_upsert_regenerates_future_instances(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """Setting an all-absent schedule deletes the child's future instances.
+
+    The regeneration side effect lives in the core layer the route
+    calls: the child's open instances at or after today are deleted
+    across ALL its definitions and NOT rebuilt while the child is
+    absent every day — the same proof the HA service test makes,
+    exercised here through the API route instead.
+    """
+    client = admin_presence.client
+    database = admin_presence.database
+    child_id = await _create_child(client)
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Brush teeth",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [child_id],
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+
+    today = datetime.date.today()
+    start = today.isoformat()
+    end = (today + datetime.timedelta(days=14)).isoformat()
+    await _materialize.materialize(database, start, end, today=today)
+    instances = _dao_instances.QuestInstancesDao(database)
+    assert await instances.list_by_date_range(child_id, start, end)
+
+    # The route's write must regenerate: absent every day removes every
+    # open future instance.
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(pattern={"0": []}),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert await instances.list_by_date_range(child_id, start, end) == []
+
+
+# --- override create + delete -----------------------------------------------
+
+
+async def test_override_is_created_then_deleted_and_presence_resolves(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """POST creates the override, DELETE removes it; presence follows."""
+    client = admin_presence.client
+    database = admin_presence.database
+    child_id = await _create_child(client)
+    schedule_response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert schedule_response.status_code == 200
+
+    # Create: the override comes back with its id and the posted shape.
+    created = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    payload = created.json()
+    assert set(payload) == {
+        "id",
+        "child_id",
+        "start_date",
+        "end_date",
+        "is_present",
+        "note",
+    }
+    assert payload["child_id"] == child_id
+    assert payload["start_date"] == TUESDAY
+    assert payload["end_date"] == TUESDAY
+    assert payload["is_present"] is True
+    assert payload["note"] == "grandmother's week"
+
+    # Database proof: the row exists — and the RESOLVED presence for the
+    # affected date flips: the override beats the Tuesday-absent pattern.
+    _schedule, overrides, engine = await _child_state(database, child_id)
+    assert [record.id for record in overrides] == [payload["id"]]
+    assert engine.is_present(child_id, TUESDAY) is True
+    # A date the override does not cover keeps the pattern's answer.
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+
+    # Delete: the row is gone and the resolved presence follows the
+    # pattern again.
+    deleted = await client.delete(
+        f"/api/v1/admin/presence-overrides/{payload['id']}",
+        headers=_admin_headers(),
+    )
+    assert deleted.status_code == 200
+    assert deleted.json() == {"status": "ok"}
+    assert (
+        await _dao_presence.PresenceOverridesDao(database).get(payload["id"])
+        is None
+    )
+    _schedule, overrides, engine = await _child_state(database, child_id)
+    assert overrides == []
+    assert engine.is_present(child_id, TUESDAY) is False
+
+
+# --- error mapping: 404 vs 422 ----------------------------------------------
+
+
+async def test_schedule_and_override_for_unknown_child_are_404(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """A presence write naming a child that does not exist is 404."""
+    client = admin_presence.client
+
+    schedule = await client.put(
+        "/api/v1/admin/children/999/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert schedule.status_code == 404
+    assert schedule.json()["detail"] == "Child not found"
+
+    override = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(999),
+        headers=_admin_headers(),
+    )
+    assert override.status_code == 404
+    assert override.json()["detail"] == "Child not found"
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        _schedule_body(cycle_length_weeks=0),  # below the 1..4 range
+        _schedule_body(cycle_length_weeks=5),  # above the documented cap
+        # week 1 of the cycle missing:
+        _schedule_body(cycle_length_weeks=2, pattern={"0": [0]}),
+        {
+            "cycle_length_weeks": 1,
+            "anchor_date": "2026-13-40",
+            "pattern": {"0": [0]},
+        },
+        _schedule_body(pattern={"0": [7]}),  # weekday out of 0..6
+        _schedule_body(pattern={"nope": [0]}),  # week index not an integer
+        _schedule_body(pattern={"0": "0,2,4"}),  # weekdays not a list
+    ],
+)
+async def test_rejected_schedule_is_422(
+    admin_presence: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """A rejected schedule is 422 and stores nothing."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=body,
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    assert (
+        await _dao_presence.PresenceSchedulesDao(
+            admin_presence.database
+        ).get_by_child(child_id)
+        is None
+    )
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        _override_body(1, start_date="2026-01-07", end_date="2026-01-06"),
+        _override_body(1, start_date="2026-02-30", end_date="2026-02-30"),
+        dict(_override_body(1), note=5),
+        dict(_override_body(1), child_id="one"),
+    ],
+)
+async def test_rejected_override_is_422(
+    admin_presence: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """A rejected override (inverted/malformed dates, bad note/id) is 422.
+
+    The shape failures (a non-string note, a non-integer child id) are
+    the body model's; the date failures are the PresenceOverride
+    model's.  A string ``is_present`` like ``"yes"`` is deliberately
+    absent: Pydantic's lax bool COERCES it to ``True`` (the same design
+    as the children active route), so it never reaches the model's
+    real-bool check.
+    """
+    response = await admin_presence.client.post(
+        "/api/v1/admin/presence-overrides", json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+
+
+async def test_overlapping_override_is_422_and_first_survives(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """An override overlapping the child's existing one is 422.
+
+    The conflicting write persists nothing: the child's original
+    override survives unchanged.
+    """
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    created = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+
+    response = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id, is_present=False),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    _schedule, overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert [record.id for record in overrides] == [created.json()["id"]]
+
+
+async def test_delete_unknown_override_is_404(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """Deleting an override id that does not exist is 404."""
+    response = await admin_presence.client.delete(
+        "/api/v1/admin/presence-overrides/999", headers=_admin_headers()
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Presence override not found"
+
+    malformed = await admin_presence.client.delete(
+        "/api/v1/admin/presence-overrides/not-an-id",
+        headers=_admin_headers(),
+    )
+    assert malformed.status_code == 422

tests/test_api_admin_questdef.py

@@ -0,0 +1,611 @@
+"""Admin-plane quest-definition route tests (task b586a2f1).
+
+These tests exercise the admin plane's quest-definition routes against
+the ASGI app with httpx, reusing the shared local-RSA-key/stubbed-JWKS
+harness (:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.
+
+Every done-condition case is proven against the DATABASE, read through
+the SAME core copy the app uses (``nestquest_core.*``; see the two-copy
+caveat in api/nestquest_core.py):
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and no definition is written.
+- ``POST /api/v1/admin/quest-definitions``: the created definition
+  EXISTS with its title, description, icon, decoded rule, assignees
+  and windows exactly as posted.
+- ``PATCH /api/v1/admin/quest-definitions/{id}``: an edit PERSISTS the
+  supplied fields and leaves omitted fields untouched (exclude_unset
+  semantics); a supplied ``windows`` list replaces the whole set;
+  assignment is not settable through the edit.
+- ``PATCH /api/v1/admin/quest-definitions/{id}/active`` with
+  ``is_active: false`` sets the flag FALSE and the definition STILL
+  EXISTS with its rule, assignees and windows (never hard-deleted);
+  reactivation restores it.
+
+Error mapping (documented in api/routes_admin.py): a core ValueError
+for a non-existent DEFINITION is 404; every other core ValueError — a
+rejected rule (RuleValidationError), a bad window name or due time, a
+rejected assignee list (an unknown assignee arrives in the BODY, so it
+is 422, never 404) — is 422, the same failure class FastAPI reports
+for malformed bodies.
+"""
+from __future__ import annotations
+
+import importlib
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The quest-definition business layer and its DAO through the SAME core
+# copy the app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' database assertions
+# only; the routes reach them through api.nestquest_core.
+_quest_definitions = importlib.import_module(
+    "nestquest_core.quest_definitions"
+)
+_dao_rules = importlib.import_module("nestquest_core.dao_rules")
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _weekly_rule_body() -> dict[str, object]:
+    """A valid weekly rule body: Mondays and Wednesdays from 2026-03-02."""
+    return {
+        "rule_type": "weekly",
+        "interval": 1,
+        "weekday_set": [0, 2],
+        "start_date": "2026-03-02",
+    }
+
+
+def _weekly_rule_dict() -> dict[str, object]:
+    """The lossless dict form _weekly_rule_body() must decode back to."""
+    return {
+        "rule_type": "weekly",
+        "interval": 1,
+        "weekday_set": [0, 2],
+        "day_of_month": None,
+        "nth_weekday": None,
+        "nth_weekday_weekday": None,
+        "month": None,
+        "start_date": "2026-03-02",
+        "end_date": None,
+    }
+
+
+async def _create_children(
+    client: object, names: tuple[str, ...]
+) -> list[int]:
+    """Create ACTIVE children through the admin children route."""
+    ids = []
+    for name in names:
+        created = await client.post(
+            "/api/v1/admin/children",
+            json={"display_name": name},
+            headers=_admin_headers(),
+        )
+        assert created.status_code == 201
+        ids.append(created.json()["id"])
+    return ids
+
+
+async def _create_definition(
+    client: object, assignee_ids: list[int]
+) -> object:
+    """POST a valid two-window definition for the given assignees."""
+    return await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": assignee_ids,
+            "windows": ["morning", ["evening", "19:30"]],
+            "description": "Put everything back where it belongs.",
+            "icon": "toy-box",
+        },
+        headers=_admin_headers(),
+    )
+
+
+async def _definitions_by_id(database: object) -> dict[int, object]:
+    """Every ACTIVE definition bundle through the app's core copy."""
+    bundles = await _quest_definitions.list_active_definitions(database)
+    return {bundle.definition.id: bundle for bundle in bundles}
+
+
+@pytest.fixture
+async def admin_questdefs(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth and children tests use (stubbed JWKS, no
+    network); the fixture also exposes the app's ONE database connection
+    so tests assert the routes' effects on the stored rows themselves.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_quest_definition_routes_refuse_non_admin_and_absent(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_questdefs.client
+    body = {"title": "Sweep the porch"}
+    non_admin = make_token(groups=("some-other-group",))
+
+    forbidden = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json=body,
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+    unauthenticated_create = await client.post(
+        "/api/v1/admin/quest-definitions", json=body
+    )
+    assert unauthenticated_create.status_code == 401
+
+    unauthenticated_edit = await client.patch(
+        "/api/v1/admin/quest-definitions/1", json={"title": "X"}
+    )
+    assert unauthenticated_edit.status_code == 401
+
+    unauthenticated_active = await client.patch(
+        "/api/v1/admin/quest-definitions/1/active", json={"is_active": False}
+    )
+    assert unauthenticated_active.status_code == 401
+
+    # No route ran its handler: no definition was written.
+    assert await _definitions_by_id(admin_questdefs.database) == {}
+
+
+# --- create -----------------------------------------------------------------
+
+
+async def test_create_persists_definition_with_assignees_and_windows(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """POST /quest-definitions persists rule, assignees and windows."""
+    client = admin_questdefs.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+
+    response = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": [ada, bo],
+            "windows": ["morning", ["evening", "19:30"]],
+            "description": "Put everything back where it belongs.",
+            "icon": "toy-box",
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 201
+    payload = response.json()
+    definition_id = payload["id"]
+    # The serialized shape is the documented definition payload.
+    assert set(payload) == {
+        "id",
+        "title",
+        "description",
+        "icon",
+        "is_active",
+        "rule",
+        "assignees",
+        "windows",
+    }
+    assert payload["title"] == "Tidy the den"
+    assert payload["description"] == "Put everything back where it belongs."
+    assert payload["icon"] == "toy-box"
+    assert payload["is_active"] is True
+    assert payload["rule"] == _weekly_rule_dict()
+    assert {a["id"]: a["display_name"] for a in payload["assignees"]} == {
+        ada: "Ada",
+        bo: "Bo",
+    }
+    assert {w["window"]: w["due_time"] for w in payload["windows"]} == {
+        "morning": None,
+        "evening": "19:30",
+    }
+
+    # Database proof through the app's core copy: the definition, its
+    # decoded rule, its assignees and its windows all exist as posted.
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.title == "Tidy the den"
+    assert stored.definition.description == (
+        "Put everything back where it belongs."
+    )
+    assert stored.definition.icon == "toy-box"
+    assert stored.definition.is_active is True
+    assert stored.rule.to_dict() == _weekly_rule_dict()
+    assert sorted(child.id for child in stored.assignees) == [ada, bo]
+    assert {(w.window, w.due_time) for w in stored.windows} == {
+        ("morning", None),
+        ("evening", "19:30"),
+    }
+
+
+async def test_create_with_required_fields_only_uses_core_defaults(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """Omitted optional fields default to NULL through the core layer."""
+    client = admin_questdefs.client
+    ada = (await _create_children(client, ("Ada",)))[0]
+
+    response = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Water the plants",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": [ada],
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 201
+    payload = response.json()
+    assert payload["description"] is None
+    assert payload["icon"] is None
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        payload["id"]
+    ]
+    assert stored.definition.description is None
+    assert stored.definition.icon is None
+
+
+async def test_create_without_required_fields_is_422(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A body missing any required field is 422 and creates nothing."""
+    client = admin_questdefs.client
+    bodies: list[dict[str, object]] = [
+        {},  # everything missing
+        {"title": "Tidy the den"},  # rule, assignees, windows missing
+        {
+            "title": "Tidy the den",
+            "rule": _weekly_rule_body(),
+        },  # assignees and windows missing
+        {
+            "title": "Tidy the den",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": [1],
+        },  # windows missing
+    ]
+    for body in bodies:
+        response = await client.post(
+            "/api/v1/admin/quest-definitions",
+            json=body,
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 422, body
+    assert await _definitions_by_id(admin_questdefs.database) == {}
+
+
+@pytest.mark.parametrize(
+    "override",
+    [
+        {"title": ""},  # empty: the core rejects it
+        {"title": "   "},  # whitespace-only: empty to a parent
+        {"assignee_child_ids": []},  # empty assignee list
+        {"assignee_child_ids": [999, 1000]},  # unknown children: 422 not 404
+        {"assignee_child_ids": ["Ada"]},  # wrong-typed ids
+        {"windows": []},  # empty window set
+        {"windows": "morning"},  # not a list
+        {"windows": ["attic"]},  # not a const.QUEST_WINDOWS name
+        {"windows": [["morning", "25:00"]]},  # not a 24-hour time
+        {"windows": [["morning", "8:00"]]},  # not a strict HH:MM
+        {"windows": [["morning", None], "morning"]},  # duplicate window
+        {"rule": {"rule_type": "weekly"}},  # weekly without weekday_set
+        {"rule": {"rule_type": "fortnightly"}},  # unknown rule_type
+        {"rule": {"rule_type": "daily", "interval": 0}},  # interval < 1
+        {"rule": {"rule_type": "weekly", "weekday_set": [7]}},  # out of range
+        {"rule": {"rule_type": "monthly_day"}},  # missing day_of_month
+        {"rule": {"rule_type": "daily", "day_of_month": 5}},  # forbidden field
+        {"rule": {"rule_type": "daily", "start_date": "2026-13-01"}},
+        {"rule": None},  # an explicit null rule is a supplied invalid value
+    ],
+)
+async def test_create_rejections_are_422_and_persist_nothing(
+    admin_questdefs: SimpleNamespace, override: dict[str, object]
+) -> None:
+    """Rejected bodies are 422 and never create a definition."""
+    client = admin_questdefs.client
+    ids = await _create_children(client, ("Ada", "Bo"))
+    body: dict[str, object] = {
+        "title": "Tidy the den",
+        "rule": _weekly_rule_body(),
+        "assignee_child_ids": ids,
+        "windows": ["morning", ["evening", "19:30"]],
+    }
+    body.update(override)
+
+    response = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json=body,
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422, override
+    assert await _definitions_by_id(admin_questdefs.database) == {}
+
+
+# --- edit -------------------------------------------------------------------
+
+
+async def test_edit_persists_supplied_fields_and_leaves_omitted_untouched(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """PATCH persists exactly the supplied fields (exclude_unset)."""
+    client = admin_questdefs.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    created = await _create_definition(client, [ada, bo])
+    assert created.status_code == 201
+    definition_id = created.json()["id"]
+    dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
+    original_rule_id = (await dao.get(definition_id)).schedule_rule_id
+
+    # Only title and windows: description, icon, rule and assignees
+    # must survive; the supplied windows REPLACE the whole set.
+    edited = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"title": "Clear the den", "windows": ["afternoon"]},
+        headers=_admin_headers(),
+    )
+    assert edited.status_code == 200
+    assert edited.json()["title"] == "Clear the den"
+    assert {w["window"]: w["due_time"] for w in edited.json()["windows"]} == {
+        "afternoon": None
+    }
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.title == "Clear the den"
+    assert {(w.window, w.due_time) for w in stored.windows} == {
+        ("afternoon", None)
+    }
+    # Omitted fields were never passed to the core: unchanged.
+    assert stored.definition.description == (
+        "Put everything back where it belongs."
+    )
+    assert stored.definition.icon == "toy-box"
+    assert stored.rule.to_dict() == _weekly_rule_dict()
+    # Assignment is not settable through the edit route: untouched.
+    assert sorted(child.id for child in stored.assignees) == [ada, bo]
+
+    # A second edit touching only the rule leaves the metadata and the
+    # windows alone, and updates the rule row IN PLACE (same rule id).
+    rescheduled = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"rule": {"rule_type": "daily", "start_date": "2026-04-01"}},
+        headers=_admin_headers(),
+    )
+    assert rescheduled.status_code == 200
+    assert rescheduled.json()["rule"]["rule_type"] == "daily"
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.title == "Clear the den"
+    assert {(w.window, w.due_time) for w in stored.windows} == {
+        ("afternoon", None)
+    }
+    assert stored.rule.to_dict() == {
+        "rule_type": "daily",
+        "interval": 1,
+        "weekday_set": None,
+        "day_of_month": None,
+        "nth_weekday": None,
+        "nth_weekday_weekday": None,
+        "month": None,
+        "start_date": "2026-04-01",
+        "end_date": None,
+    }
+    record = await dao.get(definition_id)
+    assert record.schedule_rule_id == original_rule_id
+    rule_row = await _dao_rules.ScheduleRulesDao(
+        admin_questdefs.database
+    ).get(record.schedule_rule_id)
+    assert rule_row.rule_type == "daily"
+    assert rule_row.start_date == "2026-04-01"
+
+
+async def test_edit_can_clear_description_and_icon_explicitly(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """An explicit null description/icon clears them (the core's edit path).
+
+    Unlike the children edit, the definition edit supports clearing: a
+    supplied null IS forwarded and writes NULL; everything omitted
+    stays untouched.
+    """
+    client = admin_questdefs.client
+    ada = (await _create_children(client, ("Ada",)))[0]
+    created = await _create_definition(client, [ada])
+    definition_id = created.json()["id"]
+
+    response = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"description": None, "icon": None},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json()["description"] is None
+    assert response.json()["icon"] is None
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.description is None
+    assert stored.definition.icon is None
+    assert stored.definition.title == "Tidy the den"
+    assert stored.rule.to_dict() == _weekly_rule_dict()
+
+
+async def test_edit_unknown_definition_is_404(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """An edit naming a definition that does not exist is 404."""
+    response = await admin_questdefs.client.patch(
+        "/api/v1/admin/quest-definitions/999",
+        json={"title": "Ghost chore"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Quest definition not found"
+
+
+async def test_edit_rejections_are_422_and_change_nothing(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A rejected rule or window set is 422; the stored state survives.
+
+    The core rejects BEFORE any write (and the DAO edit runs in one
+    transaction), so a 422 never leaves a half-applied change behind.
+    """
+    client = admin_questdefs.client
+    ada = (await _create_children(client, ("Ada",)))[0]
+    created = await _create_definition(client, [ada])
+    definition_id = created.json()["id"]
+
+    rejected_bodies: list[dict[str, object]] = [
+        {"rule": {"rule_type": "weekly"}},  # weekly without weekday_set
+        {"rule": None},  # an explicit null rule is rejected by the core
+        {"windows": ["attic"]},  # unknown window name
+        {"windows": None},  # an explicit null window set is rejected
+        {"title": ""},  # the core rejects an empty title
+    ]
+    for body in rejected_bodies:
+        response = await client.patch(
+            f"/api/v1/admin/quest-definitions/{definition_id}",
+            json=body,
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 422, body
+
+    # Database proof: nothing changed for any rejected call.
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.title == "Tidy the den"
+    assert stored.definition.description == (
+        "Put everything back where it belongs."
+    )
+    assert stored.rule.to_dict() == _weekly_rule_dict()
+    assert {(w.window, w.due_time) for w in stored.windows} == {
+        ("morning", None),
+        ("evening", "19:30"),
+    }
+
+
+# --- set_active -------------------------------------------------------------
+
+
+async def test_set_active_toggles_the_flag_and_the_definition_survives(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """is_active=false deactivates; the definition STILL exists intact.
+
+    Deactivation is the only removal path: the row, rule, assignees and
+    windows all survive, and reactivation through the same route
+    restores the flag without anything having been lost.
+    """
+    client = admin_questdefs.client
+    ada = (await _create_children(client, ("Ada",)))[0]
+    created = await _create_definition(client, [ada])
+    assert created.status_code == 201
+    definition_id = created.json()["id"]
+
+    deactivated = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert deactivated.status_code == 200
+    payload = deactivated.json()
+    assert payload["is_active"] is False
+    # Everything else came back unchanged: deactivation only flips a flag.
+    assert payload["title"] == "Tidy the den"
+    assert payload["rule"] == _weekly_rule_dict()
+    assert [a["id"] for a in payload["assignees"]] == [ada]
+    assert {w["window"]: w["due_time"] for w in payload["windows"]} == {
+        "morning": None,
+        "evening": "19:30",
+    }
+
+    # Database proof: the definition left the ACTIVE list but its row,
+    # rule, assignees and windows were NOT hard-deleted.
+    assert definition_id not in await _definitions_by_id(
+        admin_questdefs.database
+    )
+    dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
+    record = await dao.get(definition_id)
+    assert record is not None
+    assert record.is_active is False
+    assert record.title == "Tidy the den"
+    assert await dao.list_assignees(definition_id) != []
+    assert await dao.list_windows(definition_id) != []
+
+    # Reactivation goes through the same route: the flag is set, not
+    # one-way, and the surviving rule is still the one that was stored.
+    reactivated = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}/active",
+        json={"is_active": True},
+        headers=_admin_headers(),
+    )
+    assert reactivated.status_code == 200
+    assert reactivated.json()["is_active"] is True
+    stored = (await _definitions_by_id(admin_questdefs.database))[
+        definition_id
+    ]
+    assert stored.definition.is_active is True
+    assert stored.rule.to_dict() == _weekly_rule_dict()
+    assert [child.id for child in stored.assignees] == [ada]
+
+
+async def test_set_active_unknown_definition_is_404(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A set_active naming a definition that does not exist is 404."""
+    response = await admin_questdefs.client.patch(
+        "/api/v1/admin/quest-definitions/999/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Quest definition not found"
+
+
+async def test_set_active_malformed_body_is_422(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A non-bool is_active is rejected by the body model (422)."""
+    response = await admin_questdefs.client.patch(
+        "/api/v1/admin/quest-definitions/1/active",
+        json={"is_active": "maybe"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422

tests/test_api_admin_uncomplete.py

@@ -0,0 +1,694 @@
+"""Admin-plane uncomplete and regenerate route tests (task da0226b3).
+
+These tests exercise the admin plane's completion-reversal and
+re-materialization routes against the ASGI app with httpx, reusing the
+shared local-RSA-key/stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.  The household is
+seeded with :func:`tests.test_api_panel._seed_household` through the
+SAME core copy the app uses (``nestquest_core.*``; the two-copy caveat
+in api/nestquest_core.py), and the route's ONE clock read
+(:func:`api.routes_admin._local_now`) is pinned the same way the panel
+route's is, so seeded rows and route stamps cannot straddle midnight.
+
+Every done-condition case is proven against the DATABASE, read through
+that same core copy:
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and nothing is written.
+- ``POST /api/v1/admin/instances/{id}/uncomplete``: the completion
+  events log for the instance holds the ORIGINAL ``completed`` row
+  UNTOUCHED plus ONE appended ``uncompleted`` row attributed to the
+  verified token's ``sub`` (actor_source ``user``), and the instance's
+  DERIVED state is ``open`` again; a second uncomplete is a no-op that
+  appends nothing.
+- On an ACTUAL reversal the ``nestquest_quest_uncompleted`` transition
+  arrives on the SSE stream exactly once (subscribed live as
+  :mod:`tests.test_api_sse` does); a no-op uncomplete publishes
+  nothing.
+- ``POST /api/v1/admin/regenerate`` re-materializes without
+  duplicating: running it twice (household, definition and child
+  scopes) leaves the instance count stable, and the scoped variants
+  rebuild only their own scope's rows.
+
+Error mapping (documented in api/routes_admin.py): an unknown
+instance is 404 (the core's ``instance_id:`` prefix, the same key the
+panel complete route uses); unknown definition/child scope ids are 404
+from the route's own existence reads; malformed input — a bad scope,
+a scope/id mismatch, a zero id, a non-integer path id — is 422 before
+any handler code runs.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import importlib
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+
+from api import routes_admin
+from api.app import create_app
+from api.auth import JwksCache
+from api.config import ApiConfig
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    AUDIENCE,
+    ISSUER,
+    JWKS_URL,
+    KID,
+    AdminRunner,
+    StubbedFetch,
+    SUBJECT,
+    jwks_for,
+    make_token,
+)
+from tests.test_api_panel import PANEL_TOKEN, _seed_household
+from tests.test_api_sse import _EventStream, _Server, _stream
+
+# The completion and instance-DAO layers through the SAME core copy the
+# app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' database assertions
+# only; the routes reach them through api.nestquest_core.
+_completion = importlib.import_module("nestquest_core.completion")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+
+#: The documented quest-uncompleted event name (core const, §3).
+EVENT_QUEST_UNCOMPLETED = importlib.import_module(
+    "nestquest_core.const"
+).EVENT_QUEST_UNCOMPLETED
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _pinned_noon() -> tuple[datetime.date, datetime.datetime]:
+    """One host clock read pinned to 12:00 UTC that day.
+
+    The seed's completion and the route's clock both anchor to this
+    instant, so nothing can straddle midnight mid-test.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    return today, datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
+    )
+
+
+async def _instance_count(database: object) -> int:
+    """The whole ``quest_instances`` row count (the sanctioned probe)."""
+    row = await database.fetch_one("SELECT COUNT(*) FROM quest_instances")
+    return row[0]
+
+
+async def _events_for_instance(
+    database: object, instance_id: int
+) -> list:
+    """The instance's completion events, in append order."""
+    dao = _dao_instances.CompletionEventsDao(database)
+    return await dao.list_by_instance(instance_id)
+
+
+async def _derived_state(
+    database: object, instance_id: int, today: datetime.date
+) -> str:
+    """The instance's derived state through the app's core copy."""
+    dao = _dao_instances.QuestInstancesDao(database)
+    instance = await dao.get_by_id(instance_id)
+    assert instance is not None
+    latest = await _dao_instances.CompletionEventsDao(
+        database
+    ).get_latest_for_instance(instance_id)
+    return _completion.derive_state(instance, latest, today)
+
+
+@pytest.fixture
+async def admin_api(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth and children tests use (stubbed JWKS, no
+    network); the fixture exposes the app's ONE database connection so
+    tests assert the routes' effects on the stored rows themselves.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+@pytest.fixture
+async def admin_household(temp_db_path: str, monkeypatch) -> SimpleNamespace:
+    """A seeded household served by the admin-plane app, clock pinned.
+
+    ``_seed_household`` completes Bo's today instance (17:00 due, 12:00
+    pinned — on time), which is the completion the uncomplete tests
+    reverse.  The route's ``_local_now`` is pinned to the same instant.
+    """
+    today, pinned_now = _pinned_noon()
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    monkeypatch.setattr(routes_admin, "_local_now", lambda: pinned_now)
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            seed=seed,
+            database=runner.app.state.db.database,
+            today=today,
+        )
+
+
+class _AdminServer(_Server):
+    """:mod:`tests.test_api_sse`'s live-uvicorn server, admin-enabled.
+
+    Same ephemeral-port uvicorn thread (httpx's ASGI transport buffers
+    SSE bodies, so a live server is the only way to stream), but the
+    app factory swaps the JWKS cache for one built around the harness's
+    stub fetcher, so a real admin JWT verifies with NO network access.
+    """
+
+    def _make_app_factory(self, db_path: str):
+        """The uvicorn app factory with the stubbed JWKS cache."""
+
+        def factory():
+            app = create_app(
+                ApiConfig(
+                    db_path=db_path,
+                    panel_token=PANEL_TOKEN,
+                    oidc_issuer=ISSUER,
+                    oidc_audience=AUDIENCE,
+                    oidc_jwks_url=JWKS_URL,
+                )
+            )
+            app.state.jwks_cache = JwksCache(
+                JWKS_URL, fetcher=StubbedFetch(jwks_for(ADMIN_KEY, KID))
+            )
+            self._server_loop = asyncio.get_running_loop()
+            self.app = app
+            return app
+
+        return factory
+
+
+@pytest.fixture
+async def admin_sse(temp_db_path: str) -> SimpleNamespace:
+    """A seeded household served by a live uvicorn app, clock pinned.
+
+    Same pinning discipline as the panel tests, applied to
+    :func:`api.routes_admin._local_now` BEFORE the server starts (the
+    app is created inside uvicorn's factory thread).
+    """
+    today, pinned_now = _pinned_noon()
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    original = routes_admin._local_now
+    routes_admin._local_now = lambda: pinned_now
+    try:
+        with _AdminServer(temp_db_path) as server:
+            client = httpx.AsyncClient(base_url=server.url(""))
+            async with client:
+                yield SimpleNamespace(
+                    client=client,
+                    seed=seed,
+                    server=server,
+                    database=server.app.state.db.database,
+                    today=today,
+                )
+    finally:
+        routes_admin._local_now = original
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_uncomplete_and_regenerate_refuse_non_admin_and_absent(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    non_admin = make_token(groups=("some-other-group",))
+    events_before = await _events_for_instance(database, brush.id)
+    instances_before = await _instance_count(database)
+
+    uncomplete_url = f"/api/v1/admin/instances/{brush.id}/uncomplete"
+    forbidden = await client.post(
+        uncomplete_url,
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    unauthenticated = await client.post(uncomplete_url)
+    assert unauthenticated.status_code == 401
+
+    regenerate_bodies = (
+        {},
+        {"scope": "household"},
+        {"scope": "child", "child_id": 1},
+    )
+    for body in regenerate_bodies:
+        forbidden = await client.post(
+            "/api/v1/admin/regenerate",
+            json=body,
+            headers={"Authorization": f"Bearer {non_admin}"},
+        )
+        assert forbidden.status_code == 403, body
+        unauthenticated = await client.post(
+            "/api/v1/admin/regenerate", json=body
+        )
+        assert unauthenticated.status_code == 401, body
+
+    # No handler ran: the seeded completion is still the ONLY event for
+    # the instance and the instance table is untouched.
+    assert (
+        await _events_for_instance(database, brush.id) == events_before
+    )
+    assert await _instance_count(database) == instances_before
+
+
+# --- uncomplete: the append-only reversal -----------------------------------
+
+
+async def test_uncomplete_appends_reversal_and_derived_state_is_open(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A reversal APPENDS an uncompleted event; the instance is open again.
+
+    The original completed row is proven byte-identical after the call
+    (frozen-dataclass equality), the new row carries the verified
+    token's ``sub`` as the user actor, and the DERIVED state —
+    recomputed from the latest event — is ``open`` again.
+    """
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    today = admin_household.today
+
+    before = await _events_for_instance(database, brush.id)
+    assert [event.event_type for event in before] == ["completed"]
+    assert await _derived_state(database, brush.id, today) == "done"
+
+    response = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body == {
+        "instance_id": brush.id,
+        "state": "open",
+        "appended": True,
+    }
+
+    events = await _events_for_instance(database, brush.id)
+    assert len(events) == 2
+    # The original completed row is UNTOUCHED — never edited, never
+    # deleted (append-only completion_events).
+    assert events[0] == before[0]
+    assert events[0].event_type == "completed"
+    assert events[0].actor_source == "panel"
+    # The reversal is the NEW row, attributed to the verified token.
+    reversal = events[1]
+    assert reversal.event_type == "uncompleted"
+    assert reversal.instance_id == brush.id
+    assert reversal.child_id == brush.child_id
+    assert reversal.actor_source == "user"
+    assert reversal.actor_user_id == SUBJECT
+    assert reversal.actor_child_id is None
+    assert reversal.was_on_time is None
+    # The derived state is recomputed from the latest event: open again.
+    assert await _derived_state(database, brush.id, today) == "open"
+
+
+async def test_second_uncomplete_is_a_noop_that_appends_nothing(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A repeat uncomplete appends NOTHING and still succeeds."""
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    today = admin_household.today
+
+    first = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    assert first.json()["appended"] is True
+
+    second = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    body = second.json()
+    assert body["appended"] is False
+    assert body["state"] == "open"
+
+    events = await _events_for_instance(database, brush.id)
+    assert len(events) == 2
+    assert [event.event_type for event in events] == [
+        "completed",
+        "uncompleted",
+    ]
+    assert await _derived_state(database, brush.id, today) == "open"
+
+
+async def test_uncomplete_unknown_instance_is_404_and_malformed_is_422(
+    admin_household: SimpleNamespace,
+) -> None:
+    """An unknown instance is 404; a non-integer path id is 422."""
+    client = admin_household.client
+    brush = admin_household.seed.brush_teeth_instance
+    before = await _events_for_instance(
+        admin_household.database, brush.id
+    )
+
+    unknown = await client.post(
+        "/api/v1/admin/instances/999/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert unknown.status_code == 404
+    assert unknown.json()["detail"] == "Quest instance not found"
+
+    malformed = await client.post(
+        "/api/v1/admin/instances/not-an-int/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert malformed.status_code == 422
+
+    # Neither rejection wrote anything.
+    assert (
+        await _events_for_instance(admin_household.database, brush.id)
+        == before
+    )
+
+
+async def test_uncomplete_already_open_past_due_reports_missed(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A no-op uncomplete still reports the honest DERIVED state.
+
+    The seeded stale instance is open and past due: uncompleting it is
+    a no-op (nothing appended) and the derived state is ``missed`` —
+    open-but-past-due, never an event type.
+    """
+    client = admin_household.client
+    database = admin_household.database
+    stale = admin_household.seed.stale_instance
+    today = admin_household.today
+    assert await _derived_state(database, stale.id, today) == "missed"
+
+    response = await client.post(
+        f"/api/v1/admin/instances/{stale.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["appended"] is False
+    assert body["state"] == "missed"
+
+    # The no-op appended nothing: the instance still has NO events.
+    assert await _events_for_instance(database, stale.id) == []
+
+
+# --- the SSE transition on an actual reversal --------------------------------
+
+
+async def test_uncomplete_publishes_exactly_one_uncompleted_transition(
+    admin_sse: SimpleNamespace,
+) -> None:
+    """An actual reversal streams ONE quest-uncompleted event; a no-op none.
+
+    The stream is subscribed FIRST (test_api_sse's live-server reader),
+    the completed instance is reversed through the admin route, and the
+    documented payload fields are asserted; the no-op second call adds
+    no frame.
+    """
+    seed = admin_sse.seed
+    brush = seed.brush_teeth_instance
+    stream = _stream(admin_sse.server.url("/api/v1/panel/events"))
+    async with stream:
+        first = await admin_sse.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=_admin_headers(),
+        )
+        assert first.status_code == 200
+        assert first.json()["appended"] is True
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_UNCOMPLETED
+        assert event_type == "nestquest_quest_uncompleted"
+        assert payload["child_id"] == seed.bo.id
+        assert payload["child_name"] == "Bo"
+        assert payload["instance_id"] == brush.id
+        assert payload["quest_title"] == "Brush teeth"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == brush.due_date
+        assert payload["due_time"] == "17:00"
+        assert payload["occurred_at"]  # strict UTC ISO-8601 stamp
+        assert "was_on_time" not in payload
+
+        # The no-op second call publishes nothing: the stream stays
+        # silent past the window.
+        second = await admin_sse.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=_admin_headers(),
+        )
+        assert second.status_code == 200
+        assert second.json()["appended"] is False
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+
+# --- regenerate: re-materialize without duplicating --------------------------
+
+
+async def _create_child(client: object, name: str) -> int:
+    """Create one child through the admin children route."""
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": name},
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def _create_daily_definition(
+    client: object, assignee_ids: list[int]
+) -> int:
+    """Create one daily one-window definition for the given assignees."""
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": {"rule_type": "daily", "start_date": "2026-03-02"},
+            "assignee_child_ids": assignee_ids,
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def test_regenerate_household_never_duplicates(
+    admin_household: SimpleNamespace,
+) -> None:
+    """Two household regenerations leave the instance count stable.
+
+    The first call re-materializes the rolling horizon
+    ``[today, today + DEFAULT_HORIZON_DAYS]``; the second walks the
+    SAME tuples and upserts them in place — the row count never grows
+    and the core reports the same upsert count both times.
+    """
+    client = admin_household.client
+    database = admin_household.database
+
+    before = await _instance_count(database)
+    assert before > 0  # the seed already materialized a few days
+
+    first = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert first.status_code == 200
+    body = first.json()
+    assert body["scope"] == "household"
+    assert body["definition_id"] is None
+    assert body["child_id"] is None
+    assert body["count"] >= before
+    after_first = await _instance_count(database)
+    assert after_first >= before
+
+    second = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert second.status_code == 200
+    assert second.json()["count"] == body["count"]
+    # The idempotent walk: the second run's row count is EXACTLY the
+    # first's — no duplicated instances.
+    assert await _instance_count(database) == after_first
+
+
+async def test_regenerate_definition_scope_rematerializes(
+    admin_api: SimpleNamespace,
+) -> None:
+    """The definition scope rebuilds that definition's horizon, twice, stably.
+
+    Creating a definition through the admin route materializes NOTHING
+    (the core's create path does not regenerate); the scoped regenerate
+    call is what materializes its instances, and re-running it leaves
+    the row count exactly where the first run left it.
+    """
+    client = admin_api.client
+    database = admin_api.database
+    ada = await _create_child(client, "Ada")
+    definition_id = await _create_daily_definition(client, [ada])
+
+    assert await _instance_count(database) == 0
+
+    first = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": definition_id},
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    body = first.json()
+    assert body["scope"] == "definition"
+    assert body["definition_id"] == definition_id
+    assert body["child_id"] is None
+    assert body["count"] > 0
+    after_first = await _instance_count(database)
+    assert after_first == body["count"]
+
+    second = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": definition_id},
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json()["count"] == body["count"]
+    assert await _instance_count(database) == after_first
+
+
+async def test_regenerate_child_scope_touches_only_that_child(
+    admin_api: SimpleNamespace,
+) -> None:
+    """The child scope rebuilds that child's rows and no others'.
+
+    Two children share one daily definition; the household walk
+    materializes both.  A child-scoped regenerate for Ada re-runs her
+    side of the walk (same count) while Bo's rows keep their identity,
+    and re-running it never grows the table.
+    """
+    client = admin_api.client
+    database = admin_api.database
+    instances = _dao_instances.QuestInstancesDao(database)
+    ada = await _create_child(client, "Ada")
+    bo = await _create_child(client, "Bo")
+    # One daily definition assigned to BOTH children: the household walk
+    # gives each child its own rows.
+    await _create_daily_definition(client, [ada, bo])
+
+    # Materialize the household first so BOTH children have rows.
+    seeded = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert seeded.status_code == 200
+    assert seeded.json()["scope"] == "household"
+
+    ada_rows = await instances.list_by_date_range(
+        ada, "2026-03-02", "2100-01-01"
+    )
+    bo_rows = await instances.list_by_date_range(
+        bo, "2026-03-02", "2100-01-01"
+    )
+    assert ada_rows and bo_rows
+    total = await _instance_count(database)
+    assert total == len(ada_rows) + len(bo_rows)
+
+    for expected_count in (len(ada_rows), len(ada_rows)):
+        response = await client.post(
+            "/api/v1/admin/regenerate",
+            json={"scope": "child", "child_id": ada},
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 200
+        body = response.json()
+        assert body == {
+            "scope": "child",
+            "definition_id": None,
+            "child_id": ada,
+            "count": expected_count,
+        }
+        # Ada's side re-materialized the same number of rows and Bo's
+        # rows kept their identity; the table never grew.
+        assert await instances.list_by_date_range(
+            ada, "2026-03-02", "2100-01-01"
+        )
+        assert await instances.list_by_date_range(
+            bo, "2026-03-02", "2100-01-01"
+        ) == bo_rows
+        assert await _instance_count(database) == total
+
+
+async def test_regenerate_unknown_scope_ids_are_404(
+    admin_api: SimpleNamespace,
+) -> None:
+    """An unknown definition or child scope id is 404, nothing written."""
+    client = admin_api.client
+
+    unknown_definition = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": 999},
+        headers=_admin_headers(),
+    )
+    assert unknown_definition.status_code == 404
+    assert unknown_definition.json()["detail"] == "Quest definition not found"
+
+    unknown_child = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "child", "child_id": 999},
+        headers=_admin_headers(),
+    )
+    assert unknown_child.status_code == 404
+    assert unknown_child.json()["detail"] == "Child not found"
+
+    assert await _instance_count(admin_api.database) == 0
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        {"scope": "galaxy"},  # unknown scope
+        {"scope": "definition"},  # scope without its id
+        {"scope": "child"},  # scope without its id
+        {"scope": "household", "child_id": 1},  # id without its scope
+        {"scope": "household", "definition_id": 1},  # id without its scope
+        {
+            "scope": "definition",
+            "definition_id": 1,
+            "child_id": 2,
+        },  # scope mismatch
+        {"scope": "child", "child_id": 0},  # zero id
+        {"scope": "child", "child_id": "seven"},  # wrong type
+    ],
+)
+async def test_regenerate_malformed_bodies_are_422(
+    admin_api: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """Malformed scope/id combinations are 422 before any core call."""
+    response = await admin_api.client.post(
+        "/api/v1/admin/regenerate", json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422, body
+    assert await _instance_count(admin_api.database) == 0