Task 92f75680: Admin transition SSE stream (#202)

Co-authored-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
Co-committed-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>

api/routes_admin.py

@@ -30,10 +30,10 @@ hand-rolled here), and the call goes to
 ``set_quest_definition_active``.  The title policy, the window-name
 and due-time policy, the assignee checks, the whole-set window
 replacement and the never-hard-delete policy (deactivation is the only
-removal path) all live in the core.  Assignment is NOT settable
-through the API: the core's edit path takes no assignees.  The edit
-route forwards only SUPPLIED fields (``exclude_unset``), like the
-children edit.
+removal path) all live in the core.  The edit route forwards only
+SUPPLIED fields (``exclude_unset``), like the children edit; a supplied
+``assignee_child_ids`` list REPLACES the whole assignee set inside the
+core edit's one transaction (and its one regeneration).
 
 The presence routes follow the same pattern over
 :mod:`nestquest_core.presence_management` — the ONE presence write
@@ -58,6 +58,24 @@ layer the HA services share, so the two planes cannot drift:
   (children are never hard-deleted; overrides are deletable by
   design) and regenerates that override's child.  It answers
   ``{"status": "ok"}``.
+- ``POST /presence-overrides/consequence`` previews an override
+  BEFORE it is saved: ONE ``preview_presence_override_consequence``
+  call answers ``{removed: N}``, the number of the child's open
+  upcoming instances the materialization walk would stop generating
+  (instances with a completion event are never counted, D-005).
+  Nothing is written; an unknown child is 404, a malformed or inverted
+  range 422.
+- ``GET /children/{child_id}/presence-schedule`` reads the child's
+  schedule through ONE ``get_presence_schedule`` call, wrapped as
+  ``{"schedule": ...}`` in the PUT response's shape — ``null`` when the
+  child has none (present every day), never a 404; an unknown child is
+  404.
+- ``GET /presence-overrides`` lists the household's overrides through
+  ONE ``list_presence_overrides`` call, wrapped as
+  ``{"overrides": [...]}`` in the POST response's shape, ordered by
+  start date, child id, id; the optional ``child_id``, ``start`` and
+  ``end`` query filters (strict dates, ``end >= start``) are the
+  core's to validate.
 
 The uncomplete and regenerate routes (task da0226b3) follow the same
 pattern over :mod:`nestquest_core.completion` and
@@ -172,6 +190,33 @@ integration's rollover listener shares:
   read-only-over-the-domain-tables guarantee all live in the core; the
   response reports how many transitions were published.
 
+The occurrence-preview route (task 5f843564) lets the admin PWA's
+Definitions edit sheet show the next dates a rule fires on WITHOUT
+reimplementing the recurrence rules client-side:
+
+- ``POST /quest-definitions/occurrences-preview`` takes the SAME
+  ``rule`` object create/edit take (:class:`AdminRuleRequest`, built
+  through the SAME :func:`_rule_from_request` / core ``from_dict``
+  validation, so a rejected rule is 422 exactly as on create), an
+  optional ``start_date`` (default: the API host's local ``today``,
+  ONE :func:`_local_now` clock read) and an optional ``count`` (1..50,
+  default 10).  The dates come from ONE
+  :func:`nestquest_core.recurrence.occurrences_between` call — the
+  same engine the materializer walks — so the preview can only agree
+  with what would be generated.  The scan is BOUNDED server-side: the
+  window is ``[start_date, start_date + 366 days]`` inclusive,
+  whatever the caller sends, so every rule shape (a yearly rule
+  included) shows at least its next occurrence, and a caller can never
+  ask for an unbounded walk.  Any strict ``start_date`` is accepted
+  rather than clamped or range-limited — the fixed window already
+  bounds the work — EXCEPT one whose window would reach 9999-12-31,
+  which is 422 (the core's day walk cannot step past the calendar's
+  last date).  ``start_date`` strictness
+  (``YYYY-MM-DD``) is the core's own date validator; a non-strict or
+  unparseable value is 422.  A ``count`` outside 1..50 is the body
+  model's 422 before any handler code runs.  Nothing is read or
+  written: the route touches no database.
+
 Error mapping (every children, quest-definition and presence route,
 deliberately narrow):
 
@@ -182,8 +227,9 @@ deliberately narrow):
   set_active on an unknown id) is mapped to 404.
 - A core ``ValueError`` reporting a NON-EXISTENT DEFINITION (edit or
   set_active on an unknown id — the core prefixes that error with
-  ``definition_id:``) is mapped to 404.  Create's rejected-assignee
-  error names a child from the BODY, not a path id, so it maps to 422.
+  ``definition_id:``) is mapped to 404.  Create's and edit's
+  rejected-assignee error names a child from the BODY, not a path id,
+  so it maps to 422.
 - Every OTHER core ``ValueError`` — an empty display name, an explicit
   ``null`` colour/avatar_ref (clearing is not supported), a reorder
   list that is not a complete permutation of the children (missing,
@@ -244,6 +290,16 @@ an omitted field is never passed to
 :func:`nestquest_core.children.edit_child` and stays unchanged — while
 an explicit ``null`` IS passed and is rejected by the core rather than
 silently ignored.
+
+The EVENTS route (``GET /events``) is the admin plane's live
+transition stream for the Admin PWA: the SAME four transitions
+(quest-completed, quest-uncompleted, quest-missed, child-day-complete)
+off the SAME in-process publisher as the panel plane, framed by the
+shared :func:`api.sse.transition_event_stream` helper.  It inherits the
+router's ONE :func:`~api.auth.require_admin` check (the Authorization
+header — the PWA reads it with a fetch-based reader because
+``EventSource`` cannot set headers), so the panel service token is
+refused here exactly as on every other admin route.
 """
 from __future__ import annotations
 
@@ -251,7 +307,14 @@ import datetime
 from typing import Annotated, Literal, NoReturn
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
-from pydantic import BaseModel, field_validator, model_validator
+from fastapi.responses import StreamingResponse
+from pydantic import (
+    BaseModel,
+    StrictBool,
+    StrictInt,
+    field_validator,
+    model_validator,
+)
 
 from api import transitions as api_transitions
 from api.auth import require_admin
@@ -273,6 +336,7 @@ from api.nestquest_core import (
     core_snapshot,
     core_sweep,
 )
+from api.sse import transition_event_stream
 
 #: All admin-plane routes share this router; the ONE admin dependency
 #: (service-token refusal + JWT verification + nestquest-admins group
@@ -363,11 +427,18 @@ class AdminQuestDefinitionResponse(BaseModel):
     description: str | None
     icon: str | None
     is_active: bool
+    skip_on_away: bool
     rule: AdminRuleResponse
     assignees: list[AdminAssigneeResponse]
     windows: list[AdminDefinitionWindowResponse]
 
 
+class AdminQuestDefinitionListResponse(BaseModel):
+    """The definitions list: every definition, active AND inactive, by id."""
+
+    definitions: list[AdminQuestDefinitionResponse]
+
+
 class AdminChildCreateRequest(BaseModel):
     """The body of ``POST /api/v1/admin/children``.
 
@@ -438,24 +509,67 @@ class AdminRuleRequest(BaseModel):
     end_date: str | None = None
 
 
+#: Upper bound on ``count`` for the occurrence-preview route.
+_PREVIEW_MAX_COUNT = 50
+
+#: Days after ``start_date`` the occurrence-preview window spans
+#: (inclusive), so a yearly rule always shows its next occurrence.
+_PREVIEW_WINDOW_DAYS = 366
+
+
+class AdminOccurrencesPreviewRequest(BaseModel):
+    """The body of ``POST /api/v1/admin/quest-definitions/occurrences-preview``.
+
+    ``rule`` is the SAME object create/edit take.  ``start_date`` is an
+    optional strict ``YYYY-MM-DD`` (validated by the core; default the
+    API host's today).  ``count`` is a strict JSON integer in
+    1..``_PREVIEW_MAX_COUNT`` — anything else is 422 before any handler
+    code runs.
+    """
+
+    rule: AdminRuleRequest
+    start_date: str | None = None
+    count: StrictInt = 10
+
+    @field_validator("count")
+    @classmethod
+    def _count_in_range(cls, value: int) -> int:
+        """Reject a count outside 1.._PREVIEW_MAX_COUNT."""
+        if not 1 <= value <= _PREVIEW_MAX_COUNT:
+            raise ValueError(
+                f"count must be between 1 and {_PREVIEW_MAX_COUNT}"
+            )
+        return value
+
+
+class AdminOccurrencesPreviewResponse(BaseModel):
+    """The next occurrence dates (``YYYY-MM-DD``, ascending)."""
+
+    dates: list[str]
+
+
 class AdminQuestDefinitionCreateRequest(BaseModel):
     """The body of ``POST /api/v1/admin/quest-definitions``.
 
     ``title``, ``rule``, ``assignee_child_ids`` and ``windows`` are
     required (the core rejects an empty title, a non-list or empty
     assignee/window list, and unknown window names); ``description``
-    and ``icon`` are optional and default to NULL through the core.
-    ``windows`` entries are a window name (``morning``) or a two-item
+    and ``icon`` are optional and default to NULL through the core, and
+    ``skip_on_away`` is optional and defaults to true (no instance on a
+    date the child is absent).  ``windows`` entries are a window name (``morning``) or a two-item
     ``[window, due_time]`` pair whose due time is a strict 24-hour
     HH:MM (or ``null`` for none) — both forms the core normalizes.
+    Assignee ids are strict JSON integers: a boolean, float or numeric
+    string is 422, never coerced to a child id.
     """
 
     title: str
     rule: AdminRuleRequest
-    assignee_child_ids: list[int]
+    assignee_child_ids: list[StrictInt]
     windows: list[str | tuple[str, str | None]]
     description: str | None = None
     icon: str | None = None
+    skip_on_away: StrictBool = True
 
 
 class AdminQuestDefinitionEditRequest(BaseModel):
@@ -467,10 +581,13 @@ class AdminQuestDefinitionEditRequest(BaseModel):
     field stays unchanged.  An explicit ``null`` ``description``/
     ``icon`` IS passed and CLEARS the stored value (the core supports
     clearing here, unlike the children edit); an explicit ``null``
-    ``rule`` or ``windows`` is passed and rejected by the core.  A
-    supplied ``windows`` list REPLACES the whole window set, and
-    ``assignees`` are deliberately absent — assignment is not settable
-    through this route.
+    ``rule``, ``windows`` or ``assignee_child_ids`` is passed and
+    rejected by the core.  A supplied ``windows`` list REPLACES the
+    whole window set, and a supplied ``assignee_child_ids`` list
+    REPLACES the whole assignee set (non-empty, every id an existing
+    active child — create's contract; a rejection changes nothing).  An
+    omitted (or ``null``) ``skip_on_away`` keeps the stored value.
+    Assignee ids are strict JSON integers, as on create.
     """
 
     title: str | None = None
@@ -478,6 +595,8 @@ class AdminQuestDefinitionEditRequest(BaseModel):
     icon: str | None = None
     rule: AdminRuleRequest | None = None
     windows: list[str | tuple[str, str | None]] | None = None
+    skip_on_away: StrictBool | None = None
+    assignee_child_ids: list[StrictInt] | None = None
 
 
 class AdminQuestDefinitionActiveRequest(BaseModel):
@@ -550,6 +669,50 @@ class AdminPresenceOverrideResponse(BaseModel):
     note: str | None
 
 
+class AdminPresenceScheduleReadResponse(BaseModel):
+    """The ``GET /children/{child_id}/presence-schedule`` payload.
+
+    ``schedule`` is ``None`` when the child has no schedule — the child
+    is present every day (Feature 05); absence is state, not an error.
+    """
+
+    schedule: AdminPresenceScheduleResponse | None
+
+
+class AdminPresenceOverrideListResponse(BaseModel):
+    """The ``GET /presence-overrides`` payload: the matching overrides."""
+
+    overrides: list[AdminPresenceOverrideResponse]
+
+
+class AdminPresenceOverrideConsequenceRequest(BaseModel):
+    """The body of ``POST /api/v1/admin/presence-overrides/consequence``.
+
+    The create body's shape minus the note (which cannot change what is
+    generated): the strict dates, their order and the real
+    ``is_present`` bool are enforced by the SAME
+    :class:`~nestquest_core.presence.PresenceOverride` constructor the
+    create route's core path builds; this model only guards the SHAPE.
+    Strict fields: a JSON ``true`` is not child id 1, nor ``1`` a bool.
+    """
+
+    child_id: StrictInt
+    start_date: str
+    end_date: str
+    is_present: StrictBool
+
+
+class AdminPresenceOverrideConsequenceResponse(BaseModel):
+    """The consequence preview: ``removed`` upcoming instances.
+
+    The number of open (no completion event) upcoming quest instances
+    of the child that saving the override would remove — the "This
+    removes N upcoming tasks" warning.  Always 0 for a present override.
+    """
+
+    removed: int
+
+
 class AdminUncompleteResponse(BaseModel):
     """The result of ``POST /api/v1/admin/instances/{id}/uncomplete``.
 
@@ -915,6 +1078,40 @@ def _raise_sweep_error(error: ValueError) -> NoReturn:
     raise HTTPException(status_code=422, detail=str(error)) from error
 
 
+def _raise_preview_error(error: ValueError) -> NoReturn:
+    """Map a core recurrence ``ValueError`` from the preview route to 422.
+
+    The preview route has no path ids, so it has no 404 case: a
+    rejected rule (:class:`~nestquest_core.recurrence.RuleValidationError`)
+    or a non-strict ``start_date`` is 422 with the core's message as
+    the detail.  Only ``ValueError`` is caught, so an unexpected
+    failure re-raises rather than becoming a 4xx.
+    """
+    raise HTTPException(status_code=422, detail=str(error)) from error
+
+
+def _preview_window_end(start_date: str) -> str:
+    """Return the preview window's inclusive end for ``start_date``.
+
+    ``start_date`` + :data:`_PREVIEW_WINDOW_DAYS`.  Parsing here only
+    locates the window; the STRICT ``YYYY-MM-DD`` check stays the
+    core's — :func:`~nestquest_core.recurrence.occurrences_between`
+    re-validates the same string and rejects any lenient form this
+    parse accepted.  Raises ``ValueError`` for an unparseable value and
+    for a start whose window would reach the calendar's last date (the
+    core's day walk steps one day PAST its end, which overflows there).
+    """
+    start = datetime.date.fromisoformat(start_date)
+    if datetime.date.max - start <= datetime.timedelta(
+        days=_PREVIEW_WINDOW_DAYS
+    ):
+        raise ValueError(
+            f"start_date {start_date!r} is too close to the end of the "
+            "calendar for a preview window"
+        )
+    return (start + datetime.timedelta(days=_PREVIEW_WINDOW_DAYS)).isoformat()
+
+
 def _local_now() -> datetime.datetime:
     """Return the API host's local time as ONE timezone-aware clock read.
 
@@ -957,6 +1154,7 @@ def _quest_definition_response(
         description=bundle.definition.description,
         icon=bundle.definition.icon,
         is_active=bundle.definition.is_active,
+        skip_on_away=bundle.definition.skip_on_away,
         rule=_rule_response(bundle.rule),
         assignees=[
             AdminAssigneeResponse(id=child.id, display_name=child.display_name)
@@ -1168,6 +1366,32 @@ async def admin_reorder_children(
     return {"status": "ok"}
 
 
+@router.get(
+    "/quest-definitions",
+    summary="List quest definitions",
+    response_model=AdminQuestDefinitionListResponse,
+)
+async def admin_list_quest_definitions(
+    request: Request,
+) -> AdminQuestDefinitionListResponse:
+    """Return every quest definition in rising id order.
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own).  Thin adapter: one
+    :func:`nestquest_core.quest_definitions.list_all_definitions` call,
+    active AND inactive definitions included (the admin screen manages
+    both), each serialized exactly as the create/edit responses are.
+    """
+    state: DatabaseState = request.app.state.db
+    bundles = await core_quest_definitions.list_all_definitions(
+        state.database
+    )
+    return AdminQuestDefinitionListResponse(
+        definitions=[_quest_definition_response(bundle) for bundle in bundles]
+    )
+
+
 @router.post(
     "/quest-definitions",
     summary="Create a quest definition",
@@ -1202,12 +1426,49 @@ async def admin_create_quest_definition(
             body.windows,
             description=body.description,
             icon=body.icon,
+            skip_on_away=body.skip_on_away,
         )
     except ValueError as error:
         _raise_quest_definition_error(error)
     return _quest_definition_response(bundle)
 
 
+@router.post(
+    "/quest-definitions/occurrences-preview",
+    summary="Preview the next occurrence dates of a schedule rule",
+    response_model=AdminOccurrencesPreviewResponse,
+)
+async def admin_preview_occurrences(
+    body: AdminOccurrencesPreviewRequest,
+) -> AdminOccurrencesPreviewResponse:
+    """Return the first ``count`` dates the supplied rule fires on.
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own).  Thin adapter: the ``rule``
+    becomes ONE :class:`~nestquest_core.recurrence.ScheduleRule`
+    (:func:`_rule_from_request`, the create/edit path), and the dates
+    come from ONE :func:`nestquest_core.recurrence.occurrences_between`
+    call over the bounded window ``[start_date, start_date + 366
+    days]`` — ``start_date`` defaulting to ONE :func:`_local_now` clock
+    read — truncated to ``count``.  Every core ``ValueError`` is 422
+    (:func:`_raise_preview_error`).  No database access.
+    """
+    start_date = (
+        body.start_date
+        if body.start_date is not None
+        else _local_now().date().isoformat()
+    )
+    try:
+        rule = _rule_from_request(body.rule)
+        dates = core_recurrence.occurrences_between(
+            rule, start_date, _preview_window_end(start_date)
+        )
+    except ValueError as error:
+        _raise_preview_error(error)
+    return AdminOccurrencesPreviewResponse(dates=dates[: body.count])
+
+
 @router.patch(
     "/quest-definitions/{definition_id}",
     summary="Edit a quest definition",
@@ -1227,10 +1488,12 @@ async def admin_edit_quest_definition(
     field is never passed to
     :func:`nestquest_core.quest_definitions.edit_quest_definition` and
     stays unchanged; a supplied ``windows`` list replaces the WHOLE
-    window set in the core, and a supplied ``rule`` is rebuilt into a
-    ScheduleRule (:func:`_rule_from_request`, validated by the model).
-    An explicit ``null`` ``rule``/``windows`` reaches the core as None
-    and is rejected there (422), while an explicit ``null``
+    window set in the core, a supplied ``assignee_child_ids`` list
+    replaces the WHOLE assignee set in the same core transaction, and a
+    supplied ``rule`` is rebuilt into a ScheduleRule
+    (:func:`_rule_from_request`, validated by the model).  An explicit
+    ``null`` ``rule``/``windows``/``assignee_child_ids`` reaches the
+    core as None and is rejected there (422), while an explicit ``null``
     ``description``/``icon`` clears the stored value — the core edit
     path supports clearing, unlike the children edit.  An unknown
     definition id is mapped to 404 and any other core ``ValueError``
@@ -1392,6 +1655,114 @@ async def admin_delete_presence_override(
     return {"status": "ok"}
 
 
+@router.post(
+    "/presence-overrides/consequence",
+    summary="Count the upcoming instances a presence override would remove",
+    response_model=AdminPresenceOverrideConsequenceResponse,
+)
+async def admin_presence_override_consequence(
+    body: AdminPresenceOverrideConsequenceRequest, request: Request
+) -> AdminPresenceOverrideConsequenceResponse:
+    """Return how many upcoming instances saving the override removes.
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own).  Thin adapter: ONE
+    :func:`nestquest_core.presence_management.preview_presence_override_consequence`
+    call, anchored on ONE :func:`_local_now` clock read, which compares
+    the materialization walk's output for the child with and without
+    the proposed override and excludes instances that already have a
+    completion event (D-005).  Nothing is written.
+
+    Errors are mapped by :func:`_raise_child_error`: an unknown child
+    is 404; a malformed or inverted range is 422.
+    """
+    state: DatabaseState = request.app.state.db
+    try:
+        removed = await (
+            core_presence_management.preview_presence_override_consequence(
+                state.database,
+                body.child_id,
+                body.start_date,
+                body.end_date,
+                body.is_present,
+                today=_local_now().date(),
+            )
+        )
+    except ValueError as error:
+        _raise_child_error(error)
+    return AdminPresenceOverrideConsequenceResponse(removed=removed)
+
+
+@router.get(
+    "/children/{child_id}/presence-schedule",
+    summary="Read a child's repeating presence schedule",
+    response_model=AdminPresenceScheduleReadResponse,
+)
+async def admin_get_presence_schedule(
+    child_id: int, request: Request
+) -> AdminPresenceScheduleReadResponse:
+    """Return the child's presence schedule, or ``schedule: null``.
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own).  Thin adapter: one
+    :func:`nestquest_core.presence_management.get_presence_schedule`
+    call; a stored schedule is serialized exactly as the PUT response
+    is, and a child without one answers ``{"schedule": null}`` (present
+    every day).  Errors are mapped by :func:`_raise_child_error`: an
+    unknown child is 404.
+    """
+    state: DatabaseState = request.app.state.db
+    try:
+        schedule = await core_presence_management.get_presence_schedule(
+            state.database, child_id
+        )
+    except ValueError as error:
+        _raise_child_error(error)
+    return AdminPresenceScheduleReadResponse(
+        schedule=(
+            None if schedule is None else _presence_schedule_response(schedule)
+        )
+    )
+
+
+@router.get(
+    "/presence-overrides",
+    summary="List presence overrides",
+    response_model=AdminPresenceOverrideListResponse,
+)
+async def admin_list_presence_overrides(
+    request: Request,
+    child_id: int | None = None,
+    start: str | None = None,
+    end: str | None = None,
+) -> AdminPresenceOverrideListResponse:
+    """Return the household's presence overrides, optionally filtered.
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own).  Thin adapter: the optional
+    ``child_id``, ``start`` and ``end`` query parameters go straight to
+    :func:`nestquest_core.presence_management.list_presence_overrides`
+    — the date policy (strict ``YYYY-MM-DD``, ``end >= start``) and the
+    overlap rule live in the core — and each record is serialized
+    exactly as the POST response is, ordered by start date, child id,
+    id.  Errors are mapped by :func:`_raise_child_error`: an unknown
+    ``child_id`` is 404, a rejected date or range is 422.
+    """
+    state: DatabaseState = request.app.state.db
+    try:
+        records = await core_presence_management.list_presence_overrides(
+            state.database, child_id=child_id, start=start, end=end
+        )
+    except ValueError as error:
+        _raise_child_error(error)
+    return AdminPresenceOverrideListResponse(
+        overrides=[_presence_override_response(record) for record in records]
+    )
+
+
 @router.post(
     "/instances/{instance_id}/uncomplete",
     summary="Reverse one completion (append a reversal event)",
@@ -1846,4 +2217,25 @@ async def admin_snapshot(request: Request) -> AdminSnapshotResponse:
     )
 
 
+
+@router.get(
+    "/events",
+    summary="Stream admin transition events (SSE)",
+)
+async def admin_events(request: Request) -> StreamingResponse:
+    """Stream the app's transition events to an admin client
+    (Server-Sent Events).
+
+    Requires a valid admin JWT (the router's shared
+    :func:`~api.auth.require_admin` dependency — the ONE check; this
+    handler performs NO auth of its own), so an unauthenticated or
+    panel-token request is refused before any stream is established.
+
+    Returns the shared :func:`api.sse.transition_event_stream` response
+    off the app's single in-process publisher (``app.state.publisher``)
+    — the same publisher, framing and teardown as the panel stream.
+    """
+    return transition_event_stream(request)
+
+
 __all__ = ["router"]

api/routes_panel.py

@@ -53,9 +53,6 @@ from __future__ import annotations
 
 import datetime
 import importlib
-import json
-
-from collections.abc import AsyncIterator
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
@@ -70,6 +67,7 @@ from api.nestquest_core import (
     core_snapshot,
 )
 from api import transitions as api_transitions
+from api.sse import transition_event_stream
 
 #: The instance DAO, imported on the app's core copy for the
 #: child-mismatch guard on the complete route.
@@ -328,23 +326,6 @@ async def panel_complete_instance(
     return {"status": str(result)}
 
 
-async def _sse_stream(request: Request) -> AsyncIterator[str]:
-    """Yield the SSE frames for every published transition event.
-
-    One subscriber of the app's single publisher (resolved off
-    ``app.state.publisher``); each event becomes one
-    ``event: <type>\\ndata: <json>\\n\\n`` frame.  Breaking out of the
-    publisher's subscription (on client disconnect) tears the
-    subscription down cleanly — no queue is leaked.
-    """
-    publisher = request.app.state.publisher
-    async for event_type, payload in publisher.subscribe():
-        yield (
-            f"event: {event_type}\n"
-            f"data: {json.dumps(payload)}\n\n"
-        )
-
-
 @router.get(
     "/events",
     summary="Stream panel transition events (SSE)",
@@ -357,17 +338,13 @@ async def panel_events(request: Request) -> StreamingResponse:
     :func:`~api.dependencies.require_panel_token` dependency — the ONE
     token check; this handler performs NO auth of its own).
 
-    Returns a ``text/event-stream`` response whose body is produced by
-    :func:`_sse_stream` off the app's single in-process publisher
-    (``app.state.publisher``, installed in the lifespan).  The stream
-    ends when the client disconnects, at which point the subscription
-    is torn down (its queue removed) with nothing leaked.
+    Returns the shared :func:`api.sse.transition_event_stream` response
+    off the app's single in-process publisher (``app.state.publisher``,
+    installed in the lifespan).  The stream ends when the client
+    disconnects, at which point the subscription is torn down (its
+    queue removed) with nothing leaked.
     """
-    return StreamingResponse(
-        _sse_stream(request),
-        media_type="text/event-stream",
-        headers={"Cache-Control": "no-cache"},
-    )
+    return transition_event_stream(request)
 
 
 __all__ = ["router"]

api/sse.py

@@ -0,0 +1,50 @@
+"""The ONE Server-Sent Events response helper for the API service.
+
+Both SSE planes — the panel plane (``/api/v1/panel/events``, panel
+service token, LAN-only) and the admin plane (``/api/v1/admin/events``,
+Authentik admin JWT) — stream the SAME transitions off the SAME
+in-process publisher (``app.state.publisher``, see :mod:`api.events`)
+with the SAME framing and teardown, so the stream lives here once and
+each plane's route only supplies its own router-level auth.
+"""
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+
+from fastapi import Request
+from fastapi.responses import StreamingResponse
+
+__all__ = ["transition_event_stream"]
+
+
+async def _sse_stream(request: Request) -> AsyncIterator[str]:
+    """Yield the SSE frames for every published transition event.
+
+    One subscriber of the app's single publisher (resolved off
+    ``app.state.publisher``); each event becomes one
+    ``event: <type>\\ndata: <json>\\n\\n`` frame.  Breaking out of the
+    publisher's subscription (on client disconnect) tears the
+    subscription down cleanly — no queue is leaked.
+    """
+    publisher = request.app.state.publisher
+    async for event_type, payload in publisher.subscribe():
+        yield (
+            f"event: {event_type}\n"
+            f"data: {json.dumps(payload)}\n\n"
+        )
+
+
+def transition_event_stream(request: Request) -> StreamingResponse:
+    """A ``text/event-stream`` response over the app's transition events.
+
+    Performs NO auth of its own: the calling route's router-level
+    dependency is the ONE check for its plane.  The stream ends when
+    the client disconnects, at which point the subscription is torn
+    down (its queue removed) with nothing leaked.
+    """
+    return StreamingResponse(
+        _sse_stream(request),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache"},
+    )

custom_components/nestquest/core/dao_presence.py

@@ -270,6 +270,57 @@ class PresenceOverridesDao:
         )
         return [_override_from_row(row) for row in rows]
 
+    async def list_filtered(
+        self,
+        *,
+        child_id: int | None = None,
+        start: str | None = None,
+        end: str | None = None,
+    ) -> list[PresenceOverrideRecord]:
+        """Return the household's overrides, optionally narrowed.
+
+        Every filter is optional and they combine: ``child_id`` keeps
+        that child's overrides only (an unknown child raises ValueError
+        rather than reading as "no overrides"); ``start`` keeps
+        overrides ending on or after it and ``end`` keeps overrides
+        starting on or before it, so both together select the overrides
+        overlapping the closed range ``[start, end]`` (the
+        :meth:`list_by_child_and_range` boundary rule).  Dates are
+        strict YYYY-MM-DD and ``end`` must not precede ``start``.
+        Ordered stably by start date, then child id, then id.
+        """
+        if start is not None:
+            _validate_date(start, "start")
+        if end is not None:
+            _validate_date(end, "end")
+        if start is not None and end is not None and end < start:
+            raise ValueError(
+                f"end must be on or after start, got {end!r} < {start!r}"
+            )
+        clauses: list[str] = []
+        params: list[object] = []
+        if child_id is not None:
+            child = await self._database.fetch_one(
+                "SELECT 1 FROM children WHERE id = ?", (child_id,)
+            )
+            if child is None:
+                raise ValueError(f"child {child_id} does not exist")
+            clauses.append("child_id = ?")
+            params.append(child_id)
+        if start is not None:
+            clauses.append("end_date >= ?")
+            params.append(start)
+        if end is not None:
+            clauses.append("start_date <= ?")
+            params.append(end)
+        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
+        rows = await self._database.fetch_all(
+            f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides{where} "
+            "ORDER BY start_date, child_id, id",
+            tuple(params),
+        )
+        return [_override_from_row(row) for row in rows]
+
     async def get(self, override_id: int) -> PresenceOverrideRecord | None:
         """Return one override by id, or None when it does not exist."""
         row = await self._database.fetch_one(

custom_components/nestquest/core/dao_rules.py

@@ -105,6 +105,19 @@ def _validate_window(window: str) -> None:
         )
 
 
+def _validate_skip_on_away(value: object) -> int:
+    """Return ``value`` as 0/1, raising ValueError for anything else.
+
+    The fresh-schema CHECK constraint enforces this in SQLite, but a
+    migrated database gained the column through ``ADD COLUMN`` without
+    it, so every DAO write path validates here instead.  Accepts a bool
+    or the ints 0/1; floats, strings and other ints are rejected.
+    """
+    if not isinstance(value, int) or value not in (0, 1):
+        raise ValueError(f"skip_on_away must be 0 or 1, got {value!r}")
+    return int(value)
+
+
 @dataclass(frozen=True)
 class ScheduleRuleRecord:
     """One row of ``schedule_rules``."""
@@ -152,6 +165,7 @@ class QuestDefinitionRecord:
     due_time: str | None
     is_active: bool
     created_at: str
+    skip_on_away: bool
 
 
 @dataclass(frozen=True)
@@ -187,7 +201,7 @@ _RULE_COLUMNS = (
 )
 _DEFINITION_COLUMNS = (
     "id, title, description, icon, schedule_rule_id, due_time, "
-    "is_active, created_at"
+    "is_active, created_at, skip_on_away"
 )
 _WINDOW_COLUMNS = "definition_id, window, due_time"
 
@@ -216,6 +230,7 @@ def _definition_from_row(row: tuple) -> QuestDefinitionRecord:
         due_time=row[5],
         is_active=bool(row[6]),
         created_at=row[7],
+        skip_on_away=bool(row[8]),
     )
 
 
@@ -632,6 +647,7 @@ class QuestDefinitionsDao:
         due_time: str | None = None,
         is_active: bool = True,
         assignee_child_ids: list[int] | None = None,
+        skip_on_away: bool = True,
     ) -> QuestDefinitionRecord:
         """Insert one definition and return the record as stored.
 
@@ -644,6 +660,7 @@ class QuestDefinitionsDao:
         empty list or None creates an unassigned definition, which
         callers may fill via :meth:`add_assignee`.
         """
+        skip_on_away_value = _validate_skip_on_away(skip_on_away)
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 await self._validate_rule_exists(schedule_rule_id)
@@ -652,7 +669,8 @@ class QuestDefinitionsDao:
                 result = await self._database.execute(
                     "INSERT INTO quest_definitions (title, description, "
                     "icon, schedule_rule_id, due_time, is_active, "
-                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    "created_at, skip_on_away) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         title,
                         description,
@@ -661,6 +679,7 @@ class QuestDefinitionsDao:
                         due_time,
                         int(is_active),
                         created_at,
+                        skip_on_away_value,
                     ),
                 )
                 definition_id = result.lastrowid
@@ -684,6 +703,7 @@ class QuestDefinitionsDao:
         *,
         description: str | None = None,
         icon: str | None = None,
+        skip_on_away: bool = True,
     ) -> QuestDefinitionSnapshot:
         """Insert a rule, definition, assignees and windows atomically.
 
@@ -713,6 +733,7 @@ class QuestDefinitionsDao:
         _validate_date(schedule_rule.start_date, "start_date")
         if schedule_rule.end_date is not None:
             _validate_date(schedule_rule.end_date, "end_date")
+        skip_on_away_value = _validate_skip_on_away(skip_on_away)
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 for child_id in assignee_child_ids:
@@ -721,8 +742,18 @@ class QuestDefinitionsDao:
                 result = await self._database.execute(
                     "INSERT INTO quest_definitions (title, description, "
                     "icon, schedule_rule_id, due_time, is_active, "
-                    "created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
-                    (title, description, icon, rule_id, None, 1, created_at),
+                    "created_at, skip_on_away) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        title,
+                        description,
+                        icon,
+                        rule_id,
+                        None,
+                        1,
+                        created_at,
+                        skip_on_away_value,
+                    ),
                 )
                 definition_id = result.lastrowid
                 for child_id in assignee_child_ids:
@@ -764,8 +795,10 @@ class QuestDefinitionsDao:
         icon: str | None | object = _UNSET,
         rule: ScheduleRuleStorage | object = _UNSET,
         windows: list[tuple[str, str | None]] | object = _UNSET,
+        skip_on_away: bool | object = _UNSET,
+        assignee_child_ids: list[int] | object = _UNSET,
     ) -> QuestDefinitionSnapshot:
-        """Edit a definition's metadata, rule and windows atomically.
+        """Edit a definition's metadata, rule, windows and assignees atomically.
 
         Mirrors :meth:`create_with_rule_and_windows`: everything runs
         inside ONE transaction under the connection lock, and the
@@ -774,9 +807,10 @@ class QuestDefinitionsDao:
         ``_UNSET`` meaning "leave this field alone"; ``None`` writes SQL
         NULL so optional metadata (description, icon) can be cleared.
 
-        - ``title``/``description``/``icon`` update the definition row
-          in place; the definition-level ``due_time`` column is never
-          written (per-window due times supersede it, D-008).
+        - ``title``/``description``/``icon``/``skip_on_away`` update the
+          definition row in place; the definition-level ``due_time``
+          column is never written (per-window due times supersede it,
+          D-008).
         - ``rule`` (a pre-validated :class:`ScheduleRuleStorage`)
           overwrites the referenced ``schedule_rules`` row in place: the
           definition keeps its ``schedule_rule_id``, so no new or
@@ -785,13 +819,19 @@ class QuestDefinitionsDao:
           are deleted and the given set re-inserted, so missing windows
           are added, changed due times updated and removed windows
           dropped.
+        - ``assignee_child_ids`` REPLACES the whole assignee set: every
+          id is validated as existing-and-active BEFORE any write (the
+          same check :meth:`add_assignee` applies), then links for
+          children no longer listed are removed and links for newly
+          listed children inserted; unchanged links are left alone.
 
-        Assignment is deliberately not settable here (multi-assignee is
-        managed by :meth:`add_assignee`/:meth:`remove_assignee`), and
-        activation is untouched.  Raises ValueError when the definition
-        does not exist.  Edits change future instances only: no
+        Activation is untouched.  Raises ValueError when the definition
+        does not exist or an assignee is rejected; either way the
+        transaction rolls back and nothing is written.  Edits change future instances only: no
         ``quest_instances`` or ``completion_events`` row is touched.
         """
+        if skip_on_away is not _UNSET:
+            skip_on_away = _validate_skip_on_away(skip_on_away)
         async with _connection_lock(self._database):
             async with self._database.transaction():
                 definition = await self.get(definition_id)
@@ -799,6 +839,9 @@ class QuestDefinitionsDao:
                     raise ValueError(
                         f"quest definition {definition_id} does not exist"
                     )
+                if assignee_child_ids is not _UNSET:
+                    for child_id in assignee_child_ids:
+                        await self._validate_assignable_child(child_id)
                 assignments: list[str] = []
                 parameters: list[object] = []
                 for column, value in (
@@ -809,6 +852,9 @@ class QuestDefinitionsDao:
                     if value is not _UNSET:
                         assignments.append(f"{column} = ?")
                         parameters.append(value)
+                if skip_on_away is not _UNSET:
+                    assignments.append("skip_on_away = ?")
+                    parameters.append(skip_on_away)
                 if assignments:
                     parameters.append(definition_id)
                     await self._database.execute(
@@ -844,6 +890,10 @@ class QuestDefinitionsDao:
                             "VALUES (?, ?, ?)",
                             (definition_id, window, due_time),
                         )
+                if assignee_child_ids is not _UNSET:
+                    await self._replace_assignees(
+                        definition_id, assignee_child_ids
+                    )
                 updated = await self.get(definition_id)
                 assignees = await self.list_assignees(definition_id)
                 window_records = await self.list_windows(definition_id)
@@ -859,6 +909,36 @@ class QuestDefinitionsDao:
             windows=window_records,
         )
 
+    async def _replace_assignees(
+        self, definition_id: int, child_ids: list[int]
+    ) -> None:
+        """Make the definition's assignee links exactly ``child_ids``.
+
+        Removes the links for children not in ``child_ids`` and inserts
+        the missing ones.  MUST be called inside the connection lock and
+        an open transaction, after the ids were validated as assignable.
+        """
+        rows = await self._database.fetch_all(
+            "SELECT child_id FROM quest_definition_assignees "
+            "WHERE definition_id = ?",
+            (definition_id,),
+        )
+        current = {row[0] for row in rows}
+        wanted = set(child_ids)
+        for child_id in sorted(current - wanted):
+            await self._database.execute(
+                "DELETE FROM quest_definition_assignees "
+                "WHERE definition_id = ? AND child_id = ?",
+                (definition_id, child_id),
+            )
+        for child_id in child_ids:
+            if child_id not in current:
+                await self._database.execute(
+                    "INSERT INTO quest_definition_assignees "
+                    "(definition_id, child_id) VALUES (?, ?)",
+                    (definition_id, child_id),
+                )
+
     async def _validate_rule_exists(self, schedule_rule_id: int) -> None:
         """Raise ValueError unless the rule exists.
 
@@ -907,6 +987,13 @@ class QuestDefinitionsDao:
         )
         return [_definition_from_row(row) for row in rows]
 
+    async def list_all(self) -> list[QuestDefinitionRecord]:
+        """Return every definition, active AND inactive, oldest first."""
+        rows = await self._database.fetch_all(
+            f"SELECT {_DEFINITION_COLUMNS} FROM quest_definitions ORDER BY id"
+        )
+        return [_definition_from_row(row) for row in rows]
+
     async def _snapshots_for(
         self, definitions: list[QuestDefinitionRecord]
     ) -> list[QuestDefinitionSnapshot]:
@@ -951,6 +1038,18 @@ class QuestDefinitionsDao:
                 definitions = await self.list_active()
                 return await self._snapshots_for(definitions)
 
+    async def list_snapshots_all(self) -> list[QuestDefinitionSnapshot]:
+        """Return every definition (active AND inactive) with its rule,
+        assignees and windows, all read inside one locked transaction.
+
+        Mirrors :meth:`list_snapshots_active` over :meth:`list_all`
+        (oldest first), with the same coherence guarantee.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                definitions = await self.list_all()
+                return await self._snapshots_for(definitions)
+
     async def read_snapshots_active(self) -> list[QuestDefinitionSnapshot]:
         """Return every active definition's snapshot WITHOUT locking.
 
@@ -990,13 +1089,16 @@ class QuestDefinitionsDao:
         description: str | None | object = _UNSET,
         icon: str | None | object = _UNSET,
         due_time: str | None | object = _UNSET,
+        skip_on_away: bool | None = None,
     ) -> int:
         """Update the given fields; returns rows updated (0 if absent).
 
         Arguments default to the module sentinel ``_UNSET`` meaning
         "leave this column alone"; passing ``None`` explicitly writes
         SQL NULL, so callers can remove optional metadata
-        (description, icon, due_time).  Assignment and activation are
+        (description, icon, due_time).  ``skip_on_away`` is NOT NULL,
+        so its ``None`` default leaves it unchanged; a supplied value
+        must be 0/1 (a bool).  Assignment and activation are
         deliberately NOT settable here: they have their own explicit
         operations (:meth:`add_assignee`, :meth:`remove_assignee`,
         :meth:`set_active`) so they stay separately loggable and
@@ -1013,6 +1115,9 @@ class QuestDefinitionsDao:
             if value is not _UNSET:
                 assignments.append(f"{column} = ?")
                 parameters.append(value)
+        if skip_on_away is not None:
+            assignments.append("skip_on_away = ?")
+            parameters.append(_validate_skip_on_away(skip_on_away))
         if not assignments:
             return 0
         parameters.append(definition_id)

custom_components/nestquest/core/materialize.py

@@ -18,7 +18,8 @@ The walk is deterministic and idempotent:
   :class:`~.presence.PresenceEngine` from that same coherent read.
 - For each date in the range, each definition whose decoded rule
   :func:`~.recurrence.occurs_on` on that date, each assignee who is
-  ``is_present`` on that date and each declared window, one instance is
+  ``is_present`` on that date (or every assignee, when the definition's
+  ``skip_on_away`` is off) and each declared window, one instance is
   written through :class:`~.dao_instances.QuestInstancesDao.upsert_if_valid`
   with a ``generated_at`` stamp shared by the whole batch (one coherent
   time per run).
@@ -63,9 +64,14 @@ the presence DAO (``dao_presence.py``) must stay free of business rules
 from __future__ import annotations
 
 import datetime
+from typing import NamedTuple
 
 from .const import DEFAULT_HORIZON_DAYS
-from .dao_instances import QuestInstancesDao, _resolve_today
+from .dao_instances import (
+    CompletionEventsDao,
+    QuestInstancesDao,
+    _resolve_today,
+)
 from .dao_rules import (
     _validate_date,
     load_materialization_input,
@@ -125,8 +131,15 @@ def _decode_rule(snapshot) -> ScheduleRule:
 def _build_engine(
     schedules_records,
     overrides_records,
+    *,
+    extra_overrides: tuple[PresenceOverride, ...] = (),
 ) -> PresenceEngine:
-    """Convert raw presence records into an immutable PresenceEngine."""
+    """Convert raw presence records into an immutable PresenceEngine.
+
+    ``extra_overrides`` layers already-validated, NOT-yet-stored
+    overrides on top of the stored ones — the consequence preview's
+    proposed override (:func:`count_removed_by_override`).
+    """
     schedules: dict[int, PresenceSchedule] = {}
     for child_id, record in schedules_records.items():
         schedules[child_id] = PresenceSchedule.decode(
@@ -144,9 +157,75 @@ def _build_engine(
             )
             for record in records
         ]
+    for override in extra_overrides:
+        overrides.setdefault(override.child_id, []).append(override)
     return PresenceEngine(schedules, overrides)
 
 
+class GeneratedTuple(NamedTuple):
+    """One instance the walk WOULD write: its upsert key plus due time."""
+
+    definition_id: int
+    child_id: int
+    due_date: str
+    window: str
+    due_time: str | None
+
+
+def _generation_preview(
+    snapshots,
+    engine: PresenceEngine,
+    start: datetime.date,
+    end: datetime.date,
+    *,
+    child_filter: set[int] | None = None,
+) -> list[GeneratedTuple]:
+    """Return the tuples the walk would write over ``[start, end]``.
+
+    The pure, read-only half of :func:`materialize`: for each date, each
+    definition whose decoded rule :func:`~.recurrence.occurs_on` it,
+    each assignee in ``child_filter`` (every assignee when None) who is
+    ``is_present`` on it — or every assignee when the definition's
+    ``skip_on_away`` is off — and each declared window, one tuple, in
+    the walk's write order.  :func:`materialize` writes exactly these
+    tuples and :func:`count_removed_by_override` compares two of these
+    previews, so the write and the consequence preview can never
+    diverge.  No database access.
+    """
+    definitions = [
+        (snapshot.definition.id, _decode_rule(snapshot),
+         snapshot.assignees, snapshot.windows,
+         snapshot.definition.skip_on_away)
+        for snapshot in snapshots
+    ]
+    generated: list[GeneratedTuple] = []
+    cursor = start
+    while cursor <= end:
+        iso = cursor.isoformat()
+        for (
+            definition_id, rule, assignees, windows, skip_on_away
+        ) in definitions:
+            if not occurs_on(rule, cursor):
+                continue
+            for child in assignees:
+                if child_filter is not None and child.id not in child_filter:
+                    continue
+                if skip_on_away and not engine.is_present(child.id, cursor):
+                    continue
+                for window in windows:
+                    generated.append(
+                        GeneratedTuple(
+                            definition_id,
+                            child.id,
+                            iso,
+                            window.window,
+                            window.due_time,
+                        )
+                    )
+        cursor += datetime.timedelta(days=1)
+    return generated
+
+
 async def materialize(
     database: NestQuestDatabase,
     start_date: str,
@@ -206,12 +285,6 @@ async def materialize(
     engine = _build_engine(schedules_records, overrides_records)
     child_filter = None if child_ids is None else set(child_ids)
 
-    definitions = [
-        (snapshot.definition.id, _decode_rule(snapshot),
-         snapshot.assignees, snapshot.windows)
-        for snapshot in snapshots
-    ]
-
     generated_at = _now_stamp()
     instances = QuestInstancesDao(database)
 
@@ -226,30 +299,20 @@ async def materialize(
     start = max(start, today_date)
 
     count = 0
-    cursor = start
-    while cursor <= end:
-        iso = cursor.isoformat()
-        for definition_id, rule, assignees, windows in definitions:
-            if not occurs_on(rule, cursor):
-                continue
-            for child in assignees:
-                if child_filter is not None and child.id not in child_filter:
-                    continue
-                if not engine.is_present(child.id, cursor):
-                    continue
-                for window in windows:
-                    written = await instances.upsert_if_valid(
-                        definition_id,
-                        child.id,
-                        iso,
-                        generated_at,
-                        window=window.window,
-                        due_time=window.due_time,
-                        today=today,
-                    )
-                    if written is not None:
-                        count += 1
-        cursor += datetime.timedelta(days=1)
+    for generated in _generation_preview(
+        snapshots, engine, start, end, child_filter=child_filter
+    ):
+        written = await instances.upsert_if_valid(
+            generated.definition_id,
+            generated.child_id,
+            generated.due_date,
+            generated_at,
+            window=generated.window,
+            due_time=generated.due_time,
+            today=today,
+        )
+        if written is not None:
+            count += 1
     return count
 
 
@@ -375,3 +438,93 @@ async def regenerate_for_child(
     return await materialize(
         database, start.isoformat(), end_date, child_ids=[child_id], today=today
     )
+
+
+async def count_removed_by_override(
+    database: NestQuestDatabase,
+    override: PresenceOverride,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> int:
+    """Count the upcoming instances saving ``override`` would remove.
+
+    Read-only: nothing is written.  The window is the part of the
+    override's range inside the rolling horizon ``[today, today +
+    horizon_days]`` that :func:`regenerate_for_child` rebuilds after the
+    override is saved (outside the range the two presence answers agree,
+    so nothing there can change).  The materialization input is read
+    ONCE (:func:`~.dao_rules.load_materialization_input`) and
+    :func:`_generation_preview` runs twice over it for the override's
+    child — with the stored presence, and with ``override`` layered on
+    top (:func:`_build_engine`'s ``extra_overrides``).  The count is the
+    tuples the first preview generates and the second does not, MINUS
+    every (definition, date, window) whose stored instance already has a
+    completion event (D-005: regeneration never deletes those).  A
+    definition with ``skip_on_away`` off generates regardless of
+    presence, so it contributes nothing; a present override only ever
+    ADDS tuples, so it counts 0.
+
+    An override overlapping a stored one (which the create path refuses)
+    resolves through the engine's latest-start-wins rule.  ``today`` and
+    ``horizon_days`` resolve exactly as :func:`regenerate_for_child`
+    resolves them; a malformed horizon raises ValueError.
+    """
+    horizon = _resolve_horizon_days(horizon_days)
+    today_date = _resolve_today(today)
+    start = max(today_date, override.start_date)
+    end = min(
+        today_date + datetime.timedelta(days=horizon), override.end_date
+    )
+    if end < start:
+        return 0
+    start_date, end_date = start.isoformat(), end.isoformat()
+    child_id = override.child_id
+
+    snapshots, schedules_records, overrides_records = await (
+        load_materialization_input(database, start_date, end_date)
+    )
+    current = _generation_preview(
+        snapshots,
+        _build_engine(schedules_records, overrides_records),
+        start,
+        end,
+        child_filter={child_id},
+    )
+    proposed = set(
+        _generation_preview(
+            snapshots,
+            _build_engine(
+                schedules_records,
+                overrides_records,
+                extra_overrides=(override,),
+            ),
+            start,
+            end,
+            child_filter={child_id},
+        )
+    )
+    removed = [generated for generated in current if generated not in proposed]
+    if not removed:
+        return 0
+
+    stored = await QuestInstancesDao(database).list_by_date_range(
+        child_id, start_date, end_date
+    )
+    touched = {
+        event.instance_id
+        for event in await CompletionEventsDao(
+            database
+        ).list_by_child_and_date_range(child_id, start_date, end_date)
+    }
+    completed = {
+        (instance.definition_id, instance.due_date, instance.window)
+        for instance in stored
+        if instance.id in touched
+    }
+    return sum(
+        1
+        for generated in removed
+        if (generated.definition_id, generated.due_date, generated.window)
+        not in completed
+    )

custom_components/nestquest/core/migrations.py

@@ -30,8 +30,9 @@ widened (definition_id, child_id, due_date, window) key, and migration
 6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
 column.  Migration 7 adds the generic ``nestquest_meta_state``
 key/value table for integration-owned bookkeeping (Feature 11's
-missed-sweep watermark).  Each callable migration runs inside its own
-transaction
+missed-sweep watermark).  Migration 8 adds the ``skip_on_away``
+column to ``quest_definitions`` (Feature 22), defaulting to 1.  Each
+callable migration runs inside its own transaction
 together with its version stamp, so a crash mid-step rolls the
 statements and the stamp back together.
 
@@ -400,6 +401,28 @@ async def _add_actor_child_id(
         await database.execute("PRAGMA foreign_keys = ON")
 
 
+async def _add_skip_on_away(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Add the ``skip_on_away`` column to ``quest_definitions``.
+
+    Strictly additive: ``ALTER TABLE ... ADD COLUMN`` with a NOT NULL
+    DEFAULT 1, so existing rows read as 1 (skip absent dates — the
+    behaviour before the flag existed).  Fresh databases already carry
+    the column from the canonical DDL, so the ALTER is guarded and the
+    step only stamps there.
+    """
+    async with database.transaction():
+        if not await _table_has_column(
+            database, "quest_definitions", "skip_on_away"
+        ):
+            await database.execute(
+                "ALTER TABLE quest_definitions ADD COLUMN "
+                "skip_on_away INTEGER NOT NULL DEFAULT 1"
+            )
+        await stamp_schema_version(database, target_version)
+
+
 #: The ordered migration list.  Append-only: never edit an applied
 #: entry, add the next one instead.  (Migration 1's content was
 #: rewritten pre-release per D-007 — no version 1 database shipped.)
@@ -423,6 +446,10 @@ MIGRATIONS: Sequence[MigrationStep] = [
     # fresh databases (which already made it in migration 1) and
     # pre-v7 databases.  Strictly additive; no existing table changes.
     list(SCHEMA_V7_META_STATE_DDL),
+    # Migration 8 (Feature 22): the skip_on_away flag on
+    # quest_definitions — an additive column defaulting to 1, so every
+    # pre-existing definition keeps today's skip-when-absent behaviour.
+    _add_skip_on_away,
 ]
 
 

custom_components/nestquest/core/presence_management.py

@@ -22,18 +22,23 @@ 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:``.
+
+The admin plane's presence READS go through here too
+(:func:`get_presence_schedule`, :func:`list_presence_overrides`), so a
+reader never touches the DAO directly either.
 """
 from __future__ import annotations
 
 import datetime
 
+from .dao_children import ChildrenDao
 from .dao_presence import (
     PresenceOverrideRecord,
     PresenceOverridesDao,
     PresenceSchedulesDao,
 )
 from .db import NestQuestDatabase
-from .materialize import regenerate_for_child
+from .materialize import count_removed_by_override, regenerate_for_child
 from .presence import PresenceOverride, PresenceSchedule
 
 
@@ -130,6 +135,78 @@ async def create_presence_override(
     return record
 
 
+async def preview_presence_override_consequence(
+    database: NestQuestDatabase,
+    child_id: int,
+    start_date: object,
+    end_date: object,
+    is_present: bool,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> int:
+    """Return how many upcoming instances saving this override removes.
+
+    The admin override editor's live consequence warning, computed
+    BEFORE anything is saved: the arguments build the SAME
+    :class:`~.presence.PresenceOverride` :func:`create_presence_override`
+    builds (so a malformed or inverted range, or a non-bool
+    ``is_present``, raises the same ValueError), then
+    :func:`~.materialize.count_removed_by_override` compares what the
+    materialization walk generates for the child with and without it.
+    Nothing is written.  An unknown child reads ``child N does not
+    exist``; an INACTIVE child has no instances to remove (the walk's
+    write path never materializes one), so it counts 0.
+    ``today``/``horizon_days`` resolve as the post-save regeneration
+    resolves them.
+    """
+    override = PresenceOverride(child_id, start_date, end_date, is_present)
+    child = await ChildrenDao(database).get(override.child_id)
+    if child is None:
+        raise ValueError(f"child {override.child_id} does not exist")
+    if not child.is_active:
+        return 0
+    return await count_removed_by_override(
+        database, override, today=today, horizon_days=horizon_days
+    )
+
+
+async def get_presence_schedule(
+    database: NestQuestDatabase, child_id: int
+) -> PresenceSchedule | None:
+    """Return the child's presence schedule decoded, or None when unset.
+
+    None means the child exists and has no schedule row — present
+    every day (Feature 05).  An unknown child raises ValueError
+    (``child N does not exist``) rather than reading as present.
+    """
+    record = await PresenceSchedulesDao(database).get_by_child(child_id)
+    if record is None:
+        return None
+    return PresenceSchedule.decode(
+        record.child_id, record.anchor_date, record.pattern
+    )
+
+
+async def list_presence_overrides(
+    database: NestQuestDatabase,
+    *,
+    child_id: int | None = None,
+    start: str | None = None,
+    end: str | None = None,
+) -> list[PresenceOverrideRecord]:
+    """Return the household's presence overrides, optionally filtered.
+
+    ``child_id`` narrows to one child (unknown: ``child N does not
+    exist``); ``start``/``end`` keep the overrides overlapping that
+    closed range (strict YYYY-MM-DD, ``end >= start``).  Ordered by
+    start date, then child id, then id.
+    """
+    return await PresenceOverridesDao(database).list_filtered(
+        child_id=child_id, start=start, end=end
+    )
+
+
 async def delete_presence_override(
     database: NestQuestDatabase,
     override_id: int,

custom_components/nestquest/core/quest_definitions.py

@@ -4,8 +4,9 @@ Sits between the Feature 09 service gate and the typed
 :class:`~.dao_rules.QuestDefinitionsDao`: callers get validation and the
 all-or-nothing create/edit here, and never touch the DAO SQL directly.
 :func:`create_quest_definition` is the create path,
-:func:`edit_quest_definition` the edit path, :func:`assign_child` and
-:func:`unassign_child` the assignment paths, and
+:func:`edit_quest_definition` the edit path (which may also replace the
+whole assignee set), :func:`assign_child` and
+:func:`unassign_child` the single-child assignment paths, and
 :func:`set_quest_definition_active` the deactivate/reactivate path.
 There is no permission check here (that is Feature 09).
 
@@ -191,6 +192,17 @@ def _validate_definition_id(value: object) -> int:
     return value
 
 
+def _validate_skip_on_away(value: object) -> bool:
+    """Reject a non-bool ``skip_on_away`` before any write.
+
+    The DAO's ``int()`` would otherwise silently coerce strings and
+    numerics ("0", 1) into a state the caller never asked for.
+    """
+    if not isinstance(value, bool):
+        raise ValueError(f"skip_on_away must be a real bool, got {value!r}")
+    return value
+
+
 def _validate_child_id(value: object) -> int:
     """Reject non-int child ids (bools included) before any lookup.
 
@@ -212,6 +224,7 @@ async def create_quest_definition(
     *,
     description: str | None = None,
     icon: str | None = None,
+    skip_on_away: bool = True,
 ) -> CreatedQuestDefinition:
     """Create a quest definition (rule + assignees + windows) atomically.
 
@@ -231,6 +244,7 @@ async def create_quest_definition(
     window_specs = _normalize_windows(windows)
     description_value = _validate_text(description, "description")
     icon_value = _validate_text(icon, "icon")
+    skip_on_away_value = _validate_skip_on_away(skip_on_away)
 
     storage = schedule_rule_to_storage(rule)
     dao = QuestDefinitionsDao(database)
@@ -243,6 +257,7 @@ async def create_quest_definition(
             window_specs,
             description=description_value,
             icon=icon_value,
+            skip_on_away=skip_on_away_value,
         )
     except ValueError as error:
         # The DAO's assignee check names only the child; re-raise so the
@@ -271,10 +286,12 @@ async def edit_quest_definition(
     icon: str | None | object = _UNSET,
     rule: ScheduleRule | object = _UNSET,
     windows: list[str | tuple[str, str | None]] | object = _UNSET,
+    skip_on_away: bool | None = None,
+    assignee_child_ids: list[int] | object = _UNSET,
     today: datetime.date | None = None,
     horizon_days: int | None = None,
 ) -> CreatedQuestDefinition:
-    """Edit a quest definition's metadata, rule and windows atomically.
+    """Edit a quest definition's metadata, rule, windows and assignees atomically.
 
     Arguments default to the sentinel ``_UNSET`` meaning "leave this
     field alone".  A provided ``title`` is validated like create's
@@ -285,9 +302,16 @@ async def edit_quest_definition(
     ``schedule_rules`` row is updated, never orphaned or duplicated.  A
     provided ``windows`` list REPLACES the whole window set (each entry
     a ``const.QUEST_WINDOWS`` name, optionally with a strict HH:MM due
-    time), matching create's non-empty contract.  Assignment is NOT
-    settable here.  Raises ValueError when the definition does not
-    exist or any argument is rejected.
+    time), matching create's non-empty contract.  A provided
+    ``skip_on_away`` (a real bool) replaces the stored flag; ``None``
+    leaves it unchanged.  A provided ``assignee_child_ids`` REPLACES the
+    whole assignee set under create's contract (a non-empty list of
+    distinct plain ints, each an existing ACTIVE child — the DAO checks
+    that inside the edit's transaction): newly listed children are
+    assigned and unlisted ones unassigned.  Raises ValueError when the
+    definition does not exist or any argument is rejected; a rejection
+    writes nothing, so the stored assignees (and every other field)
+    stay unchanged.
 
     Returns the updated definition together with its decoded rule,
     assignees and windows — all read back inside the DAO's transaction,
@@ -331,6 +355,14 @@ async def edit_quest_definition(
     if windows is not _UNSET:
         window_specs = _normalize_windows(windows)
 
+    assignee_ids: list[int] | object = _UNSET
+    if assignee_child_ids is not _UNSET:
+        assignee_ids = _validate_assignee_ids(assignee_child_ids)
+
+    skip_on_away_value: bool | object = _UNSET
+    if skip_on_away is not None:
+        skip_on_away_value = _validate_skip_on_away(skip_on_away)
+
     dao = QuestDefinitionsDao(database)
     try:
         snapshot = await dao.edit_definition(
@@ -340,14 +372,18 @@ async def edit_quest_definition(
             icon=icon_value,
             rule=rule_storage,
             windows=window_specs,
+            skip_on_away=skip_on_away_value,
+            assignee_child_ids=assignee_ids,
         )
     except ValueError as error:
-        # The DAO's missing-definition error names only the id; re-raise
-        # so the public contract names the field the caller passed (matching
-        # assign/unassign/deactivate).
+        # The DAO's missing-definition and assignee errors name only the
+        # id; re-raise so the public contract names the field the caller
+        # passed (matching create/assign/unassign/deactivate).
         message = str(error)
         if message.startswith("quest definition "):
             raise ValueError(f"definition_id: {message}") from error
+        if message.startswith("child "):
+            raise ValueError(f"assignee_child_ids: {message}") from error
         raise
 
     decoded_rule = schedule_rule_from_storage(
@@ -558,6 +594,21 @@ async def list_active_definitions(
     return [_bundle_snapshot(snapshot) for snapshot in snapshots]
 
 
+async def list_all_definitions(
+    database: NestQuestDatabase,
+) -> list[CreatedQuestDefinition]:
+    """Return EVERY definition, active AND inactive (rich view).
+
+    Mirrors :func:`list_active_definitions` without the activity
+    filter: ordered by rising definition id, each entry the same
+    :class:`CreatedQuestDefinition` snapshot, read by the DAO inside one
+    locked transaction.
+    """
+    dao = QuestDefinitionsDao(database)
+    snapshots = await dao.list_snapshots_all()
+    return [_bundle_snapshot(snapshot) for snapshot in snapshots]
+
+
 async def list_definitions_for_child(
     database: NestQuestDatabase,
     child_id: int,

custom_components/nestquest/core/schema.py

@@ -257,7 +257,8 @@ QUEST_DEFINITIONS_TABLE_SQL = """(
     schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
     due_time TEXT,
     is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
-    created_at TEXT NOT NULL
+    created_at TEXT NOT NULL,
+    skip_on_away INTEGER NOT NULL DEFAULT 1 CHECK (skip_on_away IN (0, 1))
 )"""
 
 SCHEMA_V1_QUEST_DEFINITIONS_DDL: list[str] = [

tests/test_api_admin_occurrences.py

@@ -0,0 +1,431 @@
+"""Admin-plane occurrence-preview route tests (task 5f843564).
+
+These tests exercise ``POST /api/v1/admin/quest-definitions/occurrences-preview``
+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 with NO network access.
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: no credential is 401, the panel service token is 403
+  (refused as a credential type), a JWT without the admin group is
+  403, an admin JWT is 200.
+- The returned dates are asserted EXACTLY for a weekly rule with an
+  interval and a start_date, a daily rule with an interval, a rule
+  with an end_date (the list stops at it), a month-end-clamped monthly
+  rule, and the default start (the pinned host clock's today); one
+  case also asserts parity with the core's own ``occurrences_between``
+  over the documented window.
+- A rejected rule (the same core ``from_dict`` validation create/edit
+  use), a non-strict ``start_date`` and a ``count`` outside 1..50 are
+  422.
+"""
+from __future__ import annotations
+
+import datetime
+import importlib
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from api import routes_admin
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The recurrence engine through the SAME core copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
+_recurrence = importlib.import_module("nestquest_core.recurrence")
+
+PREVIEW_URL = "/api/v1/admin/quest-definitions/occurrences-preview"
+
+#: The panel service token the harness app is configured with.
+HARNESS_PANEL_TOKEN = "admin-test-panel-token"
+
+#: The 403 detail for the panel service token on an admin route.
+PANEL_TOKEN_REJECTION_DETAIL = (
+    "This credential type is not accepted for admin access"
+)
+
+#: The 403 detail for a valid JWT without the admin group.
+FORBIDDEN_DETAIL = (
+    "Admin access requires membership in the nestquest-admins group"
+)
+
+#: The pinned host instant: Wednesday 2026-09-23, noon UTC.
+PINNED_NOW = datetime.datetime(2026, 9, 23, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _weekly_rule() -> dict[str, object]:
+    """Every OTHER week's Tuesday and Friday, anchored on Tue 2026-03-03."""
+    return {
+        "rule_type": "weekly",
+        "interval": 2,
+        "weekday_set": [1, 4],
+        "start_date": "2026-03-03",
+    }
+
+
+@pytest.fixture
+async def preview(temp_db_path: str, monkeypatch) -> SimpleNamespace:
+    """An admin-plane app client with the host clock pinned (and counted)."""
+    clock_reads: list[datetime.datetime] = []
+
+    def _admin_now() -> datetime.datetime:
+        clock_reads.append(PINNED_NOW)
+        return PINNED_NOW
+
+    monkeypatch.setattr(routes_admin, "_local_now", _admin_now)
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(client=client, clock_reads=clock_reads)
+
+
+async def _post(client: object, body: dict[str, object]) -> object:
+    """POST a preview body with an admin JWT."""
+    return await client.post(PREVIEW_URL, json=body, headers=_admin_headers())
+
+
+# --- authentication: the ONE require_admin dependency ----------------------
+
+
+async def test_preview_without_credential_is_401(
+    preview: SimpleNamespace,
+) -> None:
+    """No Authorization header is the uniform 401."""
+    response = await preview.client.post(
+        PREVIEW_URL, json={"rule": _weekly_rule()}
+    )
+    assert response.status_code == 401
+
+
+async def test_preview_refuses_panel_service_token(
+    preview: SimpleNamespace,
+) -> None:
+    """The panel service token is refused as a credential type (403)."""
+    response = await preview.client.post(
+        PREVIEW_URL,
+        json={"rule": _weekly_rule()},
+        headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
+
+
+async def test_preview_refuses_non_admin_jwt(
+    preview: SimpleNamespace,
+) -> None:
+    """A valid JWT without the nestquest-admins group is 403."""
+    non_admin = make_token(groups=("some-other-group",))
+    response = await preview.client.post(
+        PREVIEW_URL,
+        json={"rule": _weekly_rule()},
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == FORBIDDEN_DETAIL
+
+
+async def test_preview_admin_jwt_is_200(preview: SimpleNamespace) -> None:
+    """An admin JWT is 200 with the documented ``dates`` shape."""
+    response = await _post(
+        preview.client, {"rule": _weekly_rule(), "start_date": "2026-03-01"}
+    )
+    assert response.status_code == 200
+    assert set(response.json()) == {"dates"}
+
+
+# --- the returned dates ----------------------------------------------------
+
+
+async def test_weekly_interval_rule_from_before_its_start(
+    preview: SimpleNamespace,
+) -> None:
+    """Every other week's Tue/Fri, starting at the rule's own start_date.
+
+    The preview starts on Sun 2026-03-01, before the rule's anchor, so
+    the first date is the anchor itself; the off weeks (Mar 10-16,
+    Mar 24-30) are skipped.
+    """
+    response = await _post(
+        preview.client,
+        {"rule": _weekly_rule(), "start_date": "2026-03-01", "count": 6},
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == [
+        "2026-03-03",
+        "2026-03-06",
+        "2026-03-17",
+        "2026-03-20",
+        "2026-03-31",
+        "2026-04-03",
+    ]
+
+
+async def test_weekly_interval_rule_from_mid_cadence(
+    preview: SimpleNamespace,
+) -> None:
+    """A start inside an on-week keeps the rule's anchor cadence."""
+    response = await _post(
+        preview.client,
+        {"rule": _weekly_rule(), "start_date": "2026-03-18", "count": 4},
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == [
+        "2026-03-20",
+        "2026-03-31",
+        "2026-04-03",
+        "2026-04-14",
+    ]
+
+
+async def test_daily_interval_rule(preview: SimpleNamespace) -> None:
+    """Every third day from 2026-01-01, previewed from 2026-01-05."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {
+                "rule_type": "daily",
+                "interval": 3,
+                "start_date": "2026-01-01",
+            },
+            "start_date": "2026-01-05",
+            "count": 4,
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == [
+        "2026-01-07",
+        "2026-01-10",
+        "2026-01-13",
+        "2026-01-16",
+    ]
+
+
+async def test_rule_with_end_date_stops_at_it(
+    preview: SimpleNamespace,
+) -> None:
+    """A daily rule ending 2026-05-04 yields only its four dates."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {
+                "rule_type": "daily",
+                "start_date": "2026-05-01",
+                "end_date": "2026-05-04",
+            },
+            "start_date": "2026-04-28",
+            "count": 10,
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == [
+        "2026-05-01",
+        "2026-05-02",
+        "2026-05-03",
+        "2026-05-04",
+    ]
+
+
+async def test_monthly_day_rule_clamps_to_month_end(
+    preview: SimpleNamespace,
+) -> None:
+    """The core's month-end policy: the 31st fires on Feb 28th."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {
+                "rule_type": "monthly_day",
+                "day_of_month": 31,
+                "start_date": "2026-01-31",
+            },
+            "start_date": "2026-01-01",
+            "count": 3,
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == [
+        "2026-01-31",
+        "2026-02-28",
+        "2026-03-31",
+    ]
+
+
+async def test_default_start_is_host_today_with_default_count(
+    preview: SimpleNamespace,
+) -> None:
+    """No start_date: ONE clock read's date; no count: ten dates."""
+    response = await _post(preview.client, {"rule": {"rule_type": "daily"}})
+    assert response.status_code == 200
+    expected = [
+        (datetime.date(2026, 9, 23) + datetime.timedelta(days=n)).isoformat()
+        for n in range(10)
+    ]
+    assert response.json()["dates"] == expected
+    assert preview.clock_reads == [PINNED_NOW]
+
+
+async def test_supplied_start_date_reads_no_clock(
+    preview: SimpleNamespace,
+) -> None:
+    """A supplied start_date is used as-is: the host clock is not read."""
+    response = await _post(
+        preview.client, {"rule": _weekly_rule(), "start_date": "2026-03-01"}
+    )
+    assert response.status_code == 200
+    assert preview.clock_reads == []
+
+
+async def test_dates_match_core_occurrences_between_over_window(
+    preview: SimpleNamespace,
+) -> None:
+    """The route's dates ARE the core's walk over the 366-day window."""
+    rule_body = {
+        "rule_type": "monthly_weekday",
+        "nth_weekday": -1,
+        "nth_weekday_weekday": 4,
+        "start_date": "2026-01-01",
+    }
+    response = await _post(
+        preview.client,
+        {"rule": rule_body, "start_date": "2026-02-10", "count": 50},
+    )
+    assert response.status_code == 200
+    core_dates = _recurrence.occurrences_between(
+        _recurrence.ScheduleRule.from_dict(rule_body),
+        "2026-02-10",
+        "2027-02-11",
+    )
+    assert response.json()["dates"] == core_dates
+    assert len(core_dates) == 12
+
+
+async def test_yearly_rule_shows_next_occurrence_within_window(
+    preview: SimpleNamespace,
+) -> None:
+    """The 366-day window always reaches a yearly rule's next date."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {
+                "rule_type": "yearly",
+                "month": 9,
+                "day_of_month": 22,
+                "start_date": "2020-09-22",
+            },
+            "start_date": "2026-09-23",
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == ["2027-09-22"]
+
+
+async def test_start_near_calendar_end_is_422(
+    preview: SimpleNamespace,
+) -> None:
+    """A window that would reach 9999-12-31 is rejected, not clamped."""
+    response = await _post(
+        preview.client,
+        {"rule": {"rule_type": "daily"}, "start_date": "9999-12-30"},
+    )
+    assert response.status_code == 422
+    assert "too close to the end of the calendar" in response.json()["detail"]
+
+
+async def test_latest_accepted_start_date(preview: SimpleNamespace) -> None:
+    """The last start whose window ends before 9999-12-31 is accepted."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {"rule_type": "daily"},
+            "start_date": "9998-12-29",
+            "count": 1,
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["dates"] == ["9998-12-29"]
+
+
+# --- rejections: 422 -------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "rule",
+    [
+        {"rule_type": "weekly"},  # weekly without weekday_set
+        {"rule_type": "fortnightly"},  # unknown rule_type
+        {"rule_type": "daily", "interval": 0},
+        {"rule_type": "daily", "weekday_set": [1]},  # forbidden field
+        {"rule_type": "daily", "start_date": "2026-3-1"},
+        {
+            "rule_type": "daily",
+            "start_date": "2026-05-04",
+            "end_date": "2026-05-01",
+        },
+    ],
+)
+async def test_malformed_rule_is_422(
+    preview: SimpleNamespace, rule: dict[str, object]
+) -> None:
+    """A rule the core's from_dict rejects is 422, as on create/edit."""
+    response = await _post(
+        preview.client, {"rule": rule, "start_date": "2026-03-01"}
+    )
+    assert response.status_code == 422
+
+
+async def test_missing_rule_is_422(preview: SimpleNamespace) -> None:
+    """The rule is required: the body model's 422."""
+    response = await _post(preview.client, {"start_date": "2026-03-01"})
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize(
+    "start_date", ["2026-3-1", "20260301", "tomorrow", "2026-02-30", ""]
+)
+async def test_non_strict_start_date_is_422(
+    preview: SimpleNamespace, start_date: str
+) -> None:
+    """start_date must be a strict YYYY-MM-DD date (the core's check)."""
+    response = await _post(
+        preview.client, {"rule": _weekly_rule(), "start_date": start_date}
+    )
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize("count", [0, -1, 51, 1000, True, "5", 2.5])
+async def test_count_out_of_range_is_422(
+    preview: SimpleNamespace, count: object
+) -> None:
+    """count must be a strict JSON integer in 1..50."""
+    response = await _post(
+        preview.client,
+        {"rule": _weekly_rule(), "start_date": "2026-03-01", "count": count},
+    )
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize("count", [1, 50])
+async def test_count_bounds_are_inclusive(
+    preview: SimpleNamespace, count: int
+) -> None:
+    """count 1 and 50 are accepted and truncate the list exactly."""
+    response = await _post(
+        preview.client,
+        {
+            "rule": {"rule_type": "daily"},
+            "start_date": "2026-01-01",
+            "count": count,
+        },
+    )
+    assert response.status_code == 200
+    assert len(response.json()["dates"]) == count

tests/test_api_admin_presence_consequence.py

@@ -0,0 +1,448 @@
+"""Admin override consequence-count route tests (task 7616e3eb).
+
+``POST /api/v1/admin/presence-overrides/consequence`` answers, BEFORE an
+override is saved, how many of the child's upcoming open quest
+instances saving it would remove (``{"removed": N}``) — the ADMIN-SPEC
+§6 "This removes N upcoming tasks" warning.  These tests exercise it
+against the ASGI app with the shared stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`):
+
+- a two-week custody pattern (present the whole first week of the
+  cycle, absent the whole second) where an absent override over a
+  known span removes an EXACT number of instances;
+- ranges that remove nothing (already-absent days, a present
+  override) count 0;
+- an instance with a completion event inside the range is NOT counted
+  (D-005 — regeneration never deletes it);
+- a ``skip_on_away = false`` definition contributes nothing;
+- the count MATCHES a direct materialise comparison: the override is
+  then actually created (which regenerates the child) and the removed
+  open future instances are counted from the database;
+- the preview writes nothing;
+- the four auth outcomes, an unknown child 404 and a malformed range
+  422.
+
+The cycle is anchored on the REAL host "today" (the create route's
+regeneration reads the host clock, so the comparison test cannot pin
+it): days today..today+6 are cycle week 0 (present every day), days
+today+7..today+13 week 1 (absent every day), today+14 week 0 again.
+"""
+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 DAOs and materializer through the SAME core copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_dao_presence = importlib.import_module("nestquest_core.dao_presence")
+_materialize = importlib.import_module("nestquest_core.materialize")
+
+CONSEQUENCE_URL = "/api/v1/admin/presence-overrides/consequence"
+OVERRIDES_URL = "/api/v1/admin/presence-overrides"
+
+#: The panel service token the harness app is configured with.
+HARNESS_PANEL_TOKEN = "admin-test-panel-token"
+
+#: The 403 detail for the panel service token on an admin route.
+PANEL_TOKEN_REJECTION_DETAIL = (
+    "This credential type is not accepted for admin access"
+)
+
+#: The 403 detail for a valid JWT without the admin group.
+FORBIDDEN_DETAIL = (
+    "Admin access requires membership in the nestquest-admins group"
+)
+
+#: The two-week custody pattern: week 0 every weekday, week 1 none.
+CUSTODY_PATTERN = {"0": [0, 1, 2, 3, 4, 5, 6], "1": []}
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _day(offset: int) -> str:
+    """The ISO date ``offset`` days after the host's today."""
+    return (datetime.date.today() + datetime.timedelta(days=offset)).isoformat()
+
+
+def _body(
+    child_id: int, start: int, end: int, *, is_present: bool = False
+) -> dict[str, object]:
+    """A consequence body over ``[today+start, today+end]``."""
+    return {
+        "child_id": child_id,
+        "start_date": _day(start),
+        "end_date": _day(end),
+        "is_present": is_present,
+    }
+
+
+async def _create_child(client: object, name: str = "Ada") -> int:
+    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,
+    child_id: int,
+    title: str,
+    windows: list[str],
+    *,
+    skip_on_away: bool = True,
+) -> int:
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": title,
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [child_id],
+            "windows": windows,
+            "skip_on_away": skip_on_away,
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def _set_custody_schedule(client: object, child_id: int) -> None:
+    """Anchor the two-week custody cycle on today (week 0 starts today)."""
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json={
+            "cycle_length_weeks": 2,
+            "anchor_date": _day(0),
+            "pattern": CUSTODY_PATTERN,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+
+
+async def _materialize_horizon(database: object) -> None:
+    """Materialize the household horizon so the stored state is current."""
+    today = datetime.date.today()
+    await _materialize.materialize(
+        database, _day(0), _day(14), today=today
+    )
+
+
+async def _open_future_keys(database: object, child_id: int) -> set[tuple]:
+    """The child's stored open (no completion event) future instance keys."""
+    instances = await _dao_instances.QuestInstancesDao(
+        database
+    ).list_by_date_range(child_id, _day(0), _day(60))
+    events = _dao_instances.CompletionEventsDao(database)
+    keys = set()
+    for instance in instances:
+        if not await events.list_by_instance(instance.id):
+            keys.add(
+                (instance.definition_id, instance.due_date, instance.window)
+            )
+    return keys
+
+
+async def _complete(
+    database: object, child_id: int, definition_id: int, due_date: str,
+    window: str,
+) -> None:
+    """Append a completion event to the stored instance."""
+    instance = await _dao_instances.QuestInstancesDao(database).get(
+        definition_id, child_id, due_date, window
+    )
+    assert instance is not None
+    await _dao_instances.CompletionEventsDao(database).append(
+        instance.id,
+        child_id,
+        "completed",
+        "user",
+        datetime.datetime.now(datetime.timezone.utc).isoformat(
+            timespec="seconds"
+        ),
+        True,
+        actor_user_id="admin-user-1",
+    )
+
+
+async def _consequence(client: object, body: dict[str, object]) -> int:
+    response = await client.post(
+        CONSEQUENCE_URL, json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 200, response.text
+    return response.json()["removed"]
+
+
+@pytest.fixture
+async def custody(temp_db_path: str) -> SimpleNamespace:
+    """A child on the two-week custody cycle with a two-window chore.
+
+    Also assigns a ``skip_on_away = false`` daily chore (generated
+    regardless of presence) and materializes the horizon, so the stored
+    instances are the walk's current output.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        database = runner.app.state.db.database
+        child_id = await _create_child(client)
+        chore_id = await _create_daily_definition(
+            client, child_id, "Brush teeth", ["morning", "evening"]
+        )
+        always_id = await _create_daily_definition(
+            client, child_id, "Feed the fish", ["afternoon"],
+            skip_on_away=False,
+        )
+        await _set_custody_schedule(client, child_id)
+        await _materialize_horizon(database)
+        yield SimpleNamespace(
+            client=client,
+            database=database,
+            child_id=child_id,
+            chore_id=chore_id,
+            always_id=always_id,
+        )
+
+
+# --- the count ---------------------------------------------------------------
+
+
+async def test_custody_override_removes_a_known_number(
+    custody: SimpleNamespace,
+) -> None:
+    """Absent over today+2..today+8 removes days +2..+6 x two windows.
+
+    Days +7 and +8 are already absent (cycle week 1), and the
+    skip_on_away=false chore is generated regardless of presence, so
+    exactly 5 present days x 2 windows = 10 instances would go.
+    """
+    removed = await _consequence(
+        custody.client, _body(custody.child_id, 2, 8)
+    )
+    assert removed == 10
+
+
+async def test_already_absent_range_removes_nothing(
+    custody: SimpleNamespace,
+) -> None:
+    """An absent override over cycle week 1 changes nothing: 0."""
+    assert await _consequence(
+        custody.client, _body(custody.child_id, 7, 13)
+    ) == 0
+
+
+async def test_present_override_removes_nothing(
+    custody: SimpleNamespace,
+) -> None:
+    """A present override only ever adds instances: 0."""
+    assert await _consequence(
+        custody.client, _body(custody.child_id, 2, 8, is_present=True)
+    ) == 0
+
+
+async def test_past_range_removes_nothing(custody: SimpleNamespace) -> None:
+    """A range entirely before today touches no upcoming instance: 0."""
+    assert await _consequence(
+        custody.client, _body(custody.child_id, -20, -1)
+    ) == 0
+
+
+async def test_completed_instance_in_range_is_not_counted(
+    custody: SimpleNamespace,
+) -> None:
+    """D-005: an instance with a completion event is never removed."""
+    await _complete(
+        custody.database, custody.child_id, custody.chore_id, _day(3),
+        "morning",
+    )
+    removed = await _consequence(
+        custody.client, _body(custody.child_id, 2, 8)
+    )
+    assert removed == 9
+
+
+async def test_skip_on_away_false_definition_contributes_nothing(
+    temp_db_path: str,
+) -> None:
+    """A child whose only chore ignores presence: absent removes 0."""
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        database = runner.app.state.db.database
+        child_id = await _create_child(client)
+        await _create_daily_definition(
+            client, child_id, "Feed the fish", ["morning"],
+            skip_on_away=False,
+        )
+        await _set_custody_schedule(client, child_id)
+        await _materialize_horizon(database)
+        assert await _consequence(client, _body(child_id, 0, 14)) == 0
+
+
+async def test_count_matches_a_direct_materialise_comparison(
+    custody: SimpleNamespace,
+) -> None:
+    """Saving the override removes EXACTLY the previewed instances.
+
+    With a completed instance inside the range, the preview is taken,
+    then the override is really created (its core path regenerates the
+    child) and the open future instances that disappeared are counted
+    from the database.  The completed instance survives the regeneration.
+    """
+    client, database = custody.client, custody.database
+    await _complete(
+        database, custody.child_id, custody.chore_id, _day(4), "evening"
+    )
+    body = _body(custody.child_id, 1, 9)
+    removed = await _consequence(client, body)
+
+    before = await _open_future_keys(database, custody.child_id)
+    created = await client.post(
+        OVERRIDES_URL, json=body, headers=_admin_headers()
+    )
+    assert created.status_code == 201
+    after = await _open_future_keys(database, custody.child_id)
+
+    assert after <= before
+    assert len(before - after) == removed
+    # Days +1..+6 present x two windows, minus the completed one.
+    assert removed == 11
+    completed = await _dao_instances.QuestInstancesDao(database).get(
+        custody.chore_id, custody.child_id, _day(4), "evening"
+    )
+    assert completed is not None
+
+
+async def test_preview_writes_nothing(custody: SimpleNamespace) -> None:
+    """No override is stored and no instance is removed by the preview."""
+    client, database = custody.client, custody.database
+    before = await _open_future_keys(database, custody.child_id)
+    assert await _consequence(client, _body(custody.child_id, 0, 14)) > 0
+    assert await _open_future_keys(database, custody.child_id) == before
+    overrides = await _dao_presence.PresenceOverridesDao(
+        database
+    ).list_by_child_and_range(custody.child_id, "2000-01-01", "2100-01-01")
+    assert overrides == []
+
+
+# --- errors ------------------------------------------------------------------
+
+
+async def test_unknown_child_is_404(custody: SimpleNamespace) -> None:
+    response = await custody.client.post(
+        CONSEQUENCE_URL, json=_body(9999, 1, 2), headers=_admin_headers()
+    )
+    assert response.status_code == 404
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"start_date": "2026-02-30"},
+        {"end_date": "2026-1-5"},
+        {"start_date": "2026-03-10", "end_date": "2026-03-01"},
+        {"is_present": "sometimes"},
+        {"child_id": "one"},
+    ],
+)
+async def test_malformed_range_or_status_is_422(
+    custody: SimpleNamespace, overrides: dict[str, object]
+) -> None:
+    body = {**_body(custody.child_id, 1, 2), **overrides}
+    response = await custody.client.post(
+        CONSEQUENCE_URL, json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+
+
+@pytest.mark.parametrize(
+    "overrides",
+    [
+        {"child_id": True},
+        {"child_id": False},
+        {"child_id": "1"},
+        {"child_id": 1.0},
+        {"is_present": 1},
+        {"is_present": 0},
+        {"is_present": "true"},
+        {"is_present": "false"},
+    ],
+)
+async def test_coercible_child_id_or_status_is_422(
+    custody: SimpleNamespace, overrides: dict[str, object]
+) -> None:
+    """Strict fields: a bool id or a numeric/string status is rejected.
+
+    The fixture child's id is 1, so a coerced ``true`` would otherwise
+    target it and return 200.  Nothing is stored and nothing removed.
+    """
+    assert custody.child_id == 1
+    database = custody.database
+    before = await _open_future_keys(database, custody.child_id)
+    body = {**_body(custody.child_id, 0, 14), **overrides}
+    response = await custody.client.post(
+        CONSEQUENCE_URL, json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+    assert await _open_future_keys(database, custody.child_id) == before
+    overrides_stored = await _dao_presence.PresenceOverridesDao(
+        database
+    ).list_by_child_and_range(custody.child_id, "2000-01-01", "2100-01-01")
+    assert overrides_stored == []
+
+
+# --- authentication: the router's ONE require_admin --------------------------
+
+
+async def test_no_credential_is_401(custody: SimpleNamespace) -> None:
+    response = await custody.client.post(
+        CONSEQUENCE_URL, json=_body(custody.child_id, 1, 2)
+    )
+    assert response.status_code == 401
+
+
+async def test_panel_service_token_is_403(custody: SimpleNamespace) -> None:
+    response = await custody.client.post(
+        CONSEQUENCE_URL,
+        json=_body(custody.child_id, 1, 2),
+        headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
+
+
+async def test_non_admin_jwt_is_403(custody: SimpleNamespace) -> None:
+    non_admin = make_token(groups=("some-other-group",))
+    response = await custody.client.post(
+        CONSEQUENCE_URL,
+        json=_body(custody.child_id, 1, 2),
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == FORBIDDEN_DETAIL
+
+
+async def test_admin_jwt_is_200(custody: SimpleNamespace) -> None:
+    response = await custody.client.post(
+        CONSEQUENCE_URL,
+        json=_body(custody.child_id, 1, 2),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json() == {"removed": 4}

tests/test_api_admin_presence_reads.py

@@ -0,0 +1,345 @@
+"""Admin-plane presence READ route tests (task 07fde892).
+
+``GET /api/v1/admin/children/{child_id}/presence-schedule`` and
+``GET /api/v1/admin/presence-overrides`` against the ASGI app with
+httpx, through the shared stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`) — no network access.
+
+- The schedule read answers ``{"schedule": {...}}`` in the PUT
+  response's shape for a seeded schedule, ``{"schedule": null}`` for a
+  child with none (present every day — NOT a 404), and 404 for an
+  unknown child.
+- The overrides read answers ``{"overrides": [...]}`` in the POST
+  response's shape, ordered by start date, child id, id; ``child_id``
+  and the ``start``/``end`` range each narrow the list; a malformed or
+  inverted range is 422 and an unknown ``child_id`` is 404.
+- Both routes inherit the router's ONE ``require_admin``: no
+  credential 401, the panel service token 403, a non-admin JWT 403, an
+  admin JWT 200.
+
+Seeding goes through the admin write routes, so the reads are proven
+against rows the real write path stored.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import ADMIN_KEY, KID, AdminRunner, jwks_for, make_token
+
+OVERRIDES_URL = "/api/v1/admin/presence-overrides"
+
+#: The panel service token the harness app is configured with.
+HARNESS_PANEL_TOKEN = "admin-test-panel-token"
+
+#: The 403 detail for the panel service token on an admin route.
+PANEL_TOKEN_REJECTION_DETAIL = (
+    "This credential type is not accepted for admin access"
+)
+
+#: The 403 detail for a valid JWT without the admin group.
+FORBIDDEN_DETAIL = (
+    "Admin access requires membership in the nestquest-admins group"
+)
+
+
+def _schedule_url(child_id: int) -> str:
+    return f"/api/v1/admin/children/{child_id}/presence-schedule"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+async def _create_child(client: object, name: str) -> 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"]
+
+
+async def _create_override(
+    client: object,
+    child_id: int,
+    start_date: str,
+    end_date: str,
+    *,
+    is_present: bool = False,
+    note: str | None = None,
+) -> dict[str, object]:
+    """Create one override through the admin write route; return it."""
+    created = await client.post(
+        OVERRIDES_URL,
+        json={
+            "child_id": child_id,
+            "start_date": start_date,
+            "end_date": end_date,
+            "is_present": is_present,
+            "note": note,
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201, created.text
+    return created.json()
+
+
+@pytest.fixture
+async def reads(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane client over a fresh database."""
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(client=client)
+
+
+@pytest.fixture
+async def seeded(reads: SimpleNamespace) -> SimpleNamespace:
+    """Two children with interleaved overrides (Ada has a schedule)."""
+    client = reads.client
+    ada = await _create_child(client, "Ada")
+    ben = await _create_child(client, "Ben")
+    put = await client.put(
+        _schedule_url(ada),
+        json={
+            "cycle_length_weeks": 2,
+            "anchor_date": "2026-01-05",
+            "pattern": {"0": [4, 0, 2], "1": []},
+        },
+        headers=_admin_headers(),
+    )
+    assert put.status_code == 200
+    ada_march = await _create_override(
+        client, ada, "2026-03-01", "2026-03-07", note="grandparents"
+    )
+    ben_january = await _create_override(
+        client, ben, "2026-01-10", "2026-01-12", is_present=True
+    )
+    ben_march = await _create_override(client, ben, "2026-03-01", "2026-03-01")
+    ada_may = await _create_override(client, ada, "2026-05-20", "2026-05-20")
+    return SimpleNamespace(
+        client=client,
+        ada=ada,
+        ben=ben,
+        ada_march=ada_march,
+        ben_january=ben_january,
+        ben_march=ben_march,
+        ada_may=ada_may,
+    )
+
+
+# --- presence schedule read ------------------------------------------------
+
+
+async def test_schedule_read_returns_the_stored_schedule(
+    seeded: SimpleNamespace,
+) -> None:
+    """A seeded schedule reads back in the PUT response's shape."""
+    response = await seeded.client.get(
+        _schedule_url(seeded.ada), headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "schedule": {
+            "child_id": seeded.ada,
+            "cycle_length_weeks": 2,
+            "anchor_date": "2026-01-05",
+            "pattern": {"0": [0, 2, 4], "1": []},
+        }
+    }
+
+
+async def test_schedule_read_without_schedule_is_null_not_404(
+    seeded: SimpleNamespace,
+) -> None:
+    """A child with no schedule (present every day) is schedule: null."""
+    response = await seeded.client.get(
+        _schedule_url(seeded.ben), headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == {"schedule": None}
+
+
+async def test_schedule_read_unknown_child_is_404(
+    reads: SimpleNamespace,
+) -> None:
+    """An unknown child is 404 with the children routes' detail."""
+    response = await reads.client.get(
+        _schedule_url(999), headers=_admin_headers()
+    )
+    assert response.status_code == 404
+    assert response.json() == {"detail": "Child not found"}
+
+
+# --- presence overrides read -----------------------------------------------
+
+
+async def test_overrides_read_lists_all_in_stable_order(
+    seeded: SimpleNamespace,
+) -> None:
+    """Unfiltered: every override, by start date, then child id, then id."""
+    response = await seeded.client.get(
+        OVERRIDES_URL, headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "overrides": [
+            seeded.ben_january,
+            seeded.ada_march,
+            seeded.ben_march,
+            seeded.ada_may,
+        ]
+    }
+    assert seeded.ada_march == {
+        "id": seeded.ada_march["id"],
+        "child_id": seeded.ada,
+        "start_date": "2026-03-01",
+        "end_date": "2026-03-07",
+        "is_present": False,
+        "note": "grandparents",
+    }
+
+
+async def test_overrides_read_empty_household(reads: SimpleNamespace) -> None:
+    """No overrides at all is an empty list."""
+    response = await reads.client.get(OVERRIDES_URL, headers=_admin_headers())
+    assert response.status_code == 200
+    assert response.json() == {"overrides": []}
+
+
+async def test_overrides_read_filters_by_child(
+    seeded: SimpleNamespace,
+) -> None:
+    """child_id keeps only that child's overrides."""
+    response = await seeded.client.get(
+        OVERRIDES_URL,
+        params={"child_id": seeded.ada},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "overrides": [seeded.ada_march, seeded.ada_may]
+    }
+
+
+async def test_overrides_read_filters_by_overlapping_range(
+    seeded: SimpleNamespace,
+) -> None:
+    """start/end keep overrides overlapping the closed range (inclusive)."""
+    # 2026-03-07 is Ada's March override's LAST day: the boundary counts.
+    response = await seeded.client.get(
+        OVERRIDES_URL,
+        params={"start": "2026-03-07", "end": "2026-05-20"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json() == {
+        "overrides": [seeded.ada_march, seeded.ada_may]
+    }
+
+
+async def test_overrides_read_open_ended_range_filters(
+    seeded: SimpleNamespace,
+) -> None:
+    """start alone keeps later-ending, end alone keeps earlier-starting."""
+    from_april = await seeded.client.get(
+        OVERRIDES_URL, params={"start": "2026-04-01"}, headers=_admin_headers()
+    )
+    assert from_april.json() == {"overrides": [seeded.ada_may]}
+    until_january = await seeded.client.get(
+        OVERRIDES_URL, params={"end": "2026-01-31"}, headers=_admin_headers()
+    )
+    assert until_january.json() == {"overrides": [seeded.ben_january]}
+
+
+async def test_overrides_read_combines_child_and_range(
+    seeded: SimpleNamespace,
+) -> None:
+    """child_id and the range narrow together."""
+    response = await seeded.client.get(
+        OVERRIDES_URL,
+        params={"child_id": seeded.ben, "start": "2026-02-01", "end": "2026-12-31"},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert response.json() == {"overrides": [seeded.ben_march]}
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        {"start": "2026-03-10", "end": "2026-03-01"},
+        {"start": "2026-3-1"},
+        {"end": "not-a-date"},
+        {"start": "2026-02-30"},
+    ],
+)
+async def test_overrides_read_invalid_range_is_422(
+    seeded: SimpleNamespace, params: dict[str, str]
+) -> None:
+    """A malformed date or an inverted range is the core's 422."""
+    response = await seeded.client.get(
+        OVERRIDES_URL, params=params, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+
+
+async def test_overrides_read_unknown_child_is_404(
+    reads: SimpleNamespace,
+) -> None:
+    """An unknown child_id filter is 404, never an empty list."""
+    response = await reads.client.get(
+        OVERRIDES_URL, params={"child_id": 999}, headers=_admin_headers()
+    )
+    assert response.status_code == 404
+    assert response.json() == {"detail": "Child not found"}
+
+
+# --- authentication: the router's ONE require_admin --------------------------
+
+
+@pytest.fixture(params=["schedule", "overrides"])
+async def read_url(request: pytest.FixtureRequest, reads: SimpleNamespace) -> str:
+    """Each read route's URL (the schedule route for a real child)."""
+    if request.param == "overrides":
+        return OVERRIDES_URL
+    return _schedule_url(await _create_child(reads.client, "Ada"))
+
+
+async def test_read_without_credential_is_401(
+    reads: SimpleNamespace, read_url: str
+) -> None:
+    response = await reads.client.get(read_url)
+    assert response.status_code == 401
+
+
+async def test_read_refuses_panel_service_token(
+    reads: SimpleNamespace, read_url: str
+) -> None:
+    response = await reads.client.get(
+        read_url, headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"}
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
+
+
+async def test_read_refuses_non_admin_jwt(
+    reads: SimpleNamespace, read_url: str
+) -> None:
+    non_admin = make_token(groups=("some-other-group",))
+    response = await reads.client.get(
+        read_url, headers={"Authorization": f"Bearer {non_admin}"}
+    )
+    assert response.status_code == 403
+    assert response.json()["detail"] == FORBIDDEN_DETAIL
+
+
+async def test_read_admin_jwt_is_200(
+    reads: SimpleNamespace, read_url: str
+) -> None:
+    response = await reads.client.get(read_url, headers=_admin_headers())
+    assert response.status_code == 200

tests/test_api_admin_questdef.py

@@ -18,8 +18,9 @@ caveat in api/nestquest_core.py):
   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.
+  semantics); a supplied ``windows`` list replaces the whole set; an
+  omitted ``assignee_child_ids`` leaves the assignees untouched (the
+  supplied-list cases live in test_api_admin_questdef_assignees.py).
 - ``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);
@@ -215,6 +216,7 @@ async def test_create_persists_definition_with_assignees_and_windows(
         "description",
         "icon",
         "is_active",
+        "skip_on_away",
         "rule",
         "assignees",
         "windows",
@@ -223,6 +225,7 @@ async def test_create_persists_definition_with_assignees_and_windows(
     assert payload["description"] == "Put everything back where it belongs."
     assert payload["icon"] == "toy-box"
     assert payload["is_active"] is True
+    assert payload["skip_on_away"] is True
     assert payload["rule"] == _weekly_rule_dict()
     assert {a["id"]: a["display_name"] for a in payload["assignees"]} == {
         ada: "Ada",
@@ -395,7 +398,7 @@ async def test_edit_persists_supplied_fields_and_leaves_omitted_untouched(
     )
     assert stored.definition.icon == "toy-box"
     assert stored.rule.to_dict() == _weekly_rule_dict()
-    # Assignment is not settable through the edit route: untouched.
+    # assignee_child_ids was omitted: the assignees are untouched.
     assert sorted(child.id for child in stored.assignees) == [ada, bo]
 
     # A second edit touching only the rule leaves the metadata and the
@@ -521,9 +524,130 @@ async def test_edit_rejections_are_422_and_change_nothing(
     }
 
 
+@pytest.mark.parametrize("coercible", [0, 1, "true", "false"])
+async def test_create_rejects_coercible_skip_on_away_and_persists_nothing(
+    admin_questdefs: SimpleNamespace, coercible: object
+) -> None:
+    """A numeric or string skip_on_away is 422, not coerced to a bool."""
+    client = admin_questdefs.client
+    dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
+    ada = (await _create_children(client, ("Ada",)))[0]
+    existing = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Feed the cat",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": [ada],
+            "windows": ["morning"],
+            "skip_on_away": False,
+        },
+        headers=_admin_headers(),
+    )
+    assert existing.status_code == 201
+    existing_id = existing.json()["id"]
+
+    response = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": _weekly_rule_body(),
+            "assignee_child_ids": [ada],
+            "windows": ["morning"],
+            "skip_on_away": coercible,
+        },
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    assert set(await _definitions_by_id(admin_questdefs.database)) == {
+        existing_id
+    }
+    assert (await dao.get(existing_id)).skip_on_away is False
+
+
+@pytest.mark.parametrize("coercible", [0, 1, "true", "false"])
+async def test_edit_rejects_coercible_skip_on_away_and_changes_nothing(
+    admin_questdefs: SimpleNamespace, coercible: object
+) -> None:
+    """A numeric or string skip_on_away edit is 422; the flag is kept."""
+    client = admin_questdefs.client
+    dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
+    ada = (await _create_children(client, ("Ada",)))[0]
+    definition_id = (await _create_definition(client, [ada])).json()["id"]
+    # Start from each stored value so a coerced write would be visible.
+    for stored in (True, False):
+        set_flag = await client.patch(
+            f"/api/v1/admin/quest-definitions/{definition_id}",
+            json={"skip_on_away": stored},
+            headers=_admin_headers(),
+        )
+        assert set_flag.status_code == 200
+
+        response = await client.patch(
+            f"/api/v1/admin/quest-definitions/{definition_id}",
+            json={"skip_on_away": coercible, "title": "Renamed"},
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 422
+        stored_definition = await dao.get(definition_id)
+        assert stored_definition.skip_on_away is stored
+        assert stored_definition.title == "Tidy the den"
+
+
 # --- set_active -------------------------------------------------------------
 
 
+async def test_edit_toggles_skip_on_away_and_omission_keeps_it(
+    admin_questdefs: SimpleNamespace,
+) -> None:
+    """A supplied skip_on_away persists; an edit omitting it keeps it."""
+    client = admin_questdefs.client
+    dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
+    ada = (await _create_children(client, ("Ada",)))[0]
+    created = await _create_definition(client, [ada])
+    assert created.status_code == 201
+    definition_id = created.json()["id"]
+    assert created.json()["skip_on_away"] is True
+
+    toggled_off = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"skip_on_away": False},
+        headers=_admin_headers(),
+    )
+    assert toggled_off.status_code == 200
+    assert toggled_off.json()["skip_on_away"] is False
+    assert (await dao.get(definition_id)).skip_on_away is False
+
+    # An edit that omits the flag leaves the stored False untouched.
+    renamed = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"title": "Tidy the den properly"},
+        headers=_admin_headers(),
+    )
+    assert renamed.status_code == 200
+    assert renamed.json()["skip_on_away"] is False
+    stored = await dao.get(definition_id)
+    assert stored.title == "Tidy the den properly"
+    assert stored.skip_on_away is False
+
+    toggled_on = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"skip_on_away": True},
+        headers=_admin_headers(),
+    )
+    assert toggled_on.status_code == 200
+    assert toggled_on.json()["skip_on_away"] is True
+    assert (await dao.get(definition_id)).skip_on_away is True
+
+    # A non-boolean value is a malformed body: 422, nothing changes.
+    rejected = await client.patch(
+        f"/api/v1/admin/quest-definitions/{definition_id}",
+        json={"skip_on_away": "sometimes"},
+        headers=_admin_headers(),
+    )
+    assert rejected.status_code == 422
+    assert (await dao.get(definition_id)).skip_on_away is True
+
+
 async def test_set_active_toggles_the_flag_and_the_definition_survives(
     admin_questdefs: SimpleNamespace,
 ) -> None:

tests/test_api_admin_questdef_assignees.py

@@ -0,0 +1,431 @@
+"""Admin-plane editable assignees on the definition edit route (task 1c8096f6).
+
+``PATCH /api/v1/admin/quest-definitions/{definition_id}`` accepts an
+optional ``assignee_child_ids`` list that REPLACES the definition's
+assignee set inside the core edit's ONE transaction
+(:func:`nestquest_core.quest_definitions.edit_quest_definition`):
+
+- newly listed children are assigned and unlisted ones unassigned; the
+  response and the database both carry the resulting set;
+- an empty list, a duplicate, an unknown child or an inactive child is
+  422 and NOTHING is written — the stored assignees (and any other
+  field supplied in the same body) are unchanged;
+- a JSON boolean, float or numeric string id is 422 on create and edit
+  (never coerced to an int) and nothing is written;
+- an omitted ``assignee_child_ids`` leaves the assignees unchanged;
+- an assignment change regenerates the definition's future instances
+  through the existing core path;
+- the route stays behind the router's ONE ``require_admin`` dependency
+  (absent credential 401, panel service token 403, non-admin JWT 403,
+  admin JWT 200).
+
+Every case is proven against the DATABASE, read through the SAME core
+copy the app uses (``nestquest_core.*``; see api/nestquest_core.py).
+"""
+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,
+)
+
+_dao_rules = importlib.import_module("nestquest_core.dao_rules")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_materialize = importlib.import_module("nestquest_core.materialize")
+_quest_definitions = importlib.import_module(
+    "nestquest_core.quest_definitions"
+)
+
+#: The panel service token the harness app is configured with.
+HARNESS_PANEL_TOKEN = "admin-test-panel-token"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _edit_path(definition_id: int) -> str:
+    """The edit route for one definition."""
+    return f"/api/v1/admin/quest-definitions/{definition_id}"
+
+
+@pytest.fixture
+async def admin_assignees(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database."""
+    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,
+        )
+
+
+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]) -> int:
+    """POST a daily morning definition for the given assignees."""
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Brush teeth",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": assignee_ids,
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def _stored_assignee_ids(
+    database: object, definition_id: int
+) -> set[int]:
+    """The definition's assignee ids as stored."""
+    dao = _dao_rules.QuestDefinitionsDao(database)
+    return {child.id for child in await dao.list_assignees(definition_id)}
+
+
+def _response_assignee_ids(response: object) -> set[int]:
+    """The assignee ids a definition payload reports."""
+    return {assignee["id"] for assignee in response.json()["assignees"]}
+
+
+# --- add / remove / replace -------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("initial", "requested"),
+    [
+        (("Ada",), ("Ada", "Bo")),  # add a child
+        (("Ada", "Bo"), ("Ada",)),  # remove a child
+        (("Ada", "Bo"), ("Cy",)),  # replace the whole set
+    ],
+)
+async def test_edit_replaces_the_assignee_set(
+    admin_assignees: SimpleNamespace,
+    initial: tuple[str, ...],
+    requested: tuple[str, ...],
+) -> None:
+    """A supplied list becomes exactly the stored and returned set."""
+    client = admin_assignees.client
+    names = ("Ada", "Bo", "Cy")
+    ids = dict(zip(names, await _create_children(client, names)))
+    definition_id = await _create_definition(
+        client, [ids[name] for name in initial]
+    )
+    wanted = {ids[name] for name in requested}
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"assignee_child_ids": [ids[name] for name in requested]},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 200
+    assert _response_assignee_ids(response) == wanted
+    assert {a["display_name"] for a in response.json()["assignees"]} == set(
+        requested
+    )
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == wanted
+    # The rest of the definition is untouched by an assignee-only edit.
+    assert response.json()["title"] == "Brush teeth"
+    assert [w["window"] for w in response.json()["windows"]] == ["morning"]
+
+
+async def test_edit_with_assignees_and_other_fields_applies_both(
+    admin_assignees: SimpleNamespace,
+) -> None:
+    """Assignees and metadata supplied together land in one edit."""
+    client = admin_assignees.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    definition_id = await _create_definition(client, [ada])
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"title": "Floss", "assignee_child_ids": [bo]},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 200
+    assert response.json()["title"] == "Floss"
+    assert _response_assignee_ids(response) == {bo}
+    dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
+    assert (await dao.get(definition_id)).title == "Floss"
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {bo}
+
+
+# --- rejections leave the stored state unchanged ----------------------------
+
+
+@pytest.mark.parametrize(
+    ("case", "expected_detail"),
+    [
+        ("empty", "assignee_child_ids must not be empty"),
+        ("unknown", "assignee_child_ids: child 999 does not exist"),
+        ("inactive", "assignee_child_ids: child {cy} is inactive"),
+        ("duplicate", "assignee_child_ids must not contain duplicate"),
+        ("null", "assignee_child_ids must be a list"),
+        ("mixed_unknown", "assignee_child_ids: child 999 does not exist"),
+        ("wrong_type", None),  # the body model's own 422
+    ],
+)
+async def test_rejected_assignee_list_is_422_and_changes_nothing(
+    admin_assignees: SimpleNamespace,
+    case: str,
+    expected_detail: str | None,
+) -> None:
+    """Empty/unknown/inactive/malformed lists are 422 with no mutation.
+
+    The same body also renames the definition, proving the whole edit
+    is all-or-nothing: neither the assignees nor the title change.
+    """
+    client = admin_assignees.client
+    ada, bo, cy, di = await _create_children(
+        client, ("Ada", "Bo", "Cy", "Di")
+    )
+    definition_id = await _create_definition(client, [ada, bo])
+    deactivated = await client.patch(
+        f"/api/v1/admin/children/{cy}/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert deactivated.status_code == 200
+
+    requested: object = {
+        "empty": [],
+        "unknown": [999],
+        "inactive": [ada, cy],
+        "duplicate": [ada, ada],
+        "null": None,
+        # A valid new child alongside an unknown one: di must NOT be
+        # assigned (all-or-nothing), and ada/bo must NOT be removed.
+        "mixed_unknown": [di, 999],
+        "wrong_type": ["Ada"],
+    }[case]
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"title": "Renamed", "assignee_child_ids": requested},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 422, case
+    if expected_detail is not None:
+        assert response.json()["detail"].startswith(
+            expected_detail.format(cy=cy)
+        ), response.json()
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {ada, bo}
+    dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
+    assert (await dao.get(definition_id)).title == "Brush teeth"
+
+
+#: JSON ids the lax ``int`` parser would coerce (``true`` -> 1).
+_COERCIBLE_IDS = [True, 1.0, "1"]
+
+
+@pytest.mark.parametrize("coercible", _COERCIBLE_IDS)
+async def test_edit_rejects_coercible_assignee_ids_and_changes_nothing(
+    admin_assignees: SimpleNamespace, coercible: object
+) -> None:
+    """A bool, float or numeric-string id edit is 422; nothing changes."""
+    client = admin_assignees.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    # Assign bo only, so a coerced id 1 (ada) would be a visible change.
+    definition_id = await _create_definition(client, [bo])
+    assert ada == 1
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"title": "Renamed", "assignee_child_ids": [coercible]},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 422
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {bo}
+    dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
+    assert (await dao.get(definition_id)).title == "Brush teeth"
+
+
+@pytest.mark.parametrize("coercible", _COERCIBLE_IDS)
+async def test_create_rejects_coercible_assignee_ids_and_persists_nothing(
+    admin_assignees: SimpleNamespace, coercible: object
+) -> None:
+    """A bool, float or numeric-string id create is 422; nothing is written."""
+    client = admin_assignees.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    assert ada == 1
+    existing_id = await _create_definition(client, [bo])
+
+    response = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [coercible],
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 422
+    bundles = await _quest_definitions.list_active_definitions(
+        admin_assignees.database
+    )
+    assert [bundle.definition.id for bundle in bundles] == [existing_id]
+    assert await _stored_assignee_ids(
+        admin_assignees.database, existing_id
+    ) == {bo}
+
+
+async def test_rejected_assignees_on_unknown_definition_is_404(
+    admin_assignees: SimpleNamespace,
+) -> None:
+    """An unknown definition id is still 404 when assignees are supplied."""
+    client = admin_assignees.client
+    (ada,) = await _create_children(client, ("Ada",))
+    response = await client.patch(
+        _edit_path(999),
+        json={"assignee_child_ids": [ada]},
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Quest definition not found"
+
+
+# --- omitted keeps ----------------------------------------------------------
+
+
+async def test_omitted_assignees_are_unchanged(
+    admin_assignees: SimpleNamespace,
+) -> None:
+    """An edit without assignee_child_ids never touches the assignees."""
+    client = admin_assignees.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    definition_id = await _create_definition(client, [ada, bo])
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"title": "Floss"},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 200
+    assert _response_assignee_ids(response) == {ada, bo}
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {ada, bo}
+
+
+# --- regeneration -----------------------------------------------------------
+
+
+async def test_assignment_change_regenerates_future_instances(
+    admin_assignees: SimpleNamespace,
+) -> None:
+    """Swapping the assignee moves the future open instances with it.
+
+    The regeneration lives in the core edit path the route calls: the
+    definition's open instances at or after today are deleted and the
+    horizon re-materialized for the NEW assignee set.
+    """
+    client = admin_assignees.client
+    database = admin_assignees.database
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    definition_id = await _create_definition(client, [ada])
+
+    today = datetime.date.today()
+    start = today.isoformat()
+    end = (today + datetime.timedelta(days=7)).isoformat()
+    await _materialize.materialize(database, start, end, today=today)
+    instances = _dao_instances.QuestInstancesDao(database)
+    assert await instances.list_by_date_range(ada, start, end)
+    assert await instances.list_by_date_range(bo, start, end) == []
+
+    response = await client.patch(
+        _edit_path(definition_id),
+        json={"assignee_child_ids": [bo]},
+        headers=_admin_headers(),
+    )
+
+    assert response.status_code == 200
+    assert await instances.list_by_date_range(ada, start, end) == []
+    regenerated = await instances.list_by_date_range(bo, start, end)
+    assert regenerated
+    assert {i.definition_id for i in regenerated} == {definition_id}
+
+
+# --- authentication ---------------------------------------------------------
+
+
+async def test_assignee_edit_auth_outcomes(
+    admin_assignees: SimpleNamespace,
+) -> None:
+    """401 absent, 403 panel token, 403 non-admin, 200 admin."""
+    client = admin_assignees.client
+    ada, bo = await _create_children(client, ("Ada", "Bo"))
+    definition_id = await _create_definition(client, [ada])
+    body = {"assignee_child_ids": [bo]}
+
+    absent = await client.patch(_edit_path(definition_id), json=body)
+    assert absent.status_code == 401
+
+    panel = await client.patch(
+        _edit_path(definition_id),
+        json=body,
+        headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
+    )
+    assert panel.status_code == 403
+
+    non_admin_token = make_token(groups=("some-other-group",))
+    non_admin = await client.patch(
+        _edit_path(definition_id),
+        json=body,
+        headers={"Authorization": f"Bearer {non_admin_token}"},
+    )
+    assert non_admin.status_code == 403
+
+    # None of the refused calls reached the handler.
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {ada}
+
+    admin = await client.patch(
+        _edit_path(definition_id), json=body, headers=_admin_headers()
+    )
+    assert admin.status_code == 200
+    assert await _stored_assignee_ids(
+        admin_assignees.database, definition_id
+    ) == {bo}

tests/test_api_admin_questdef_list.py

@@ -0,0 +1,299 @@
+"""Admin-plane quest-definition LIST route tests (task d4667573).
+
+``GET /api/v1/admin/quest-definitions`` 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
+with NO network access.
+
+- An admin JWT lists EVERY definition — active AND inactive — in rising
+  id order, each in the same documented shape as the create/edit
+  responses (id, title, description, icon, is_active, rule, assignees,
+  windows).
+- An empty household lists nothing: 200 with an empty list.
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an absent credential is 401, the panel service token is
+  refused outright (403, the credential-type detail), and a JWT without
+  the nestquest-admins group is 403.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+#: The admin list route under test.
+LIST_PATH = "/api/v1/admin/quest-definitions"
+
+#: The panel service token the harness app is configured with.
+HARNESS_PANEL_TOKEN = "admin-test-panel-token"
+
+#: The 403 detail for the panel service token on an admin route.
+PANEL_TOKEN_REJECTION_DETAIL = (
+    "This credential type is not accepted for admin access"
+)
+
+#: The 403 detail for a valid JWT without the admin group.
+FORBIDDEN_DETAIL = (
+    "Admin access requires membership in the nestquest-admins group"
+)
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+@pytest.fixture
+async def admin_client(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client over a fresh (empty) household."""
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield client
+
+
+async def _seed_active_and_inactive(client: object) -> tuple[int, int, int]:
+    """Seed one child, an ACTIVE definition and an INACTIVE one.
+
+    Returns ``(child_id, active_id, inactive_id)``.  Seeded through the
+    admin routes so the rows are exactly what the admin plane writes.
+    """
+    child = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Ada"},
+        headers=_admin_headers(),
+    )
+    assert child.status_code == 201
+    child_id = child.json()["id"]
+
+    active = await client.post(
+        LIST_PATH,
+        json={
+            "title": "Tidy the den",
+            "rule": {
+                "rule_type": "weekly",
+                "interval": 1,
+                "weekday_set": [0, 2],
+                "start_date": "2026-03-02",
+            },
+            "assignee_child_ids": [child_id],
+            "windows": ["morning", ["evening", "19:30"]],
+            "description": "Put everything back where it belongs.",
+            "icon": "toy-box",
+        },
+        headers=_admin_headers(),
+    )
+    assert active.status_code == 201
+
+    inactive = await client.post(
+        LIST_PATH,
+        json={
+            "title": "Water the plants",
+            "rule": {
+                "rule_type": "daily",
+                "interval": 1,
+                "start_date": "2026-03-01",
+            },
+            "assignee_child_ids": [child_id],
+            "windows": ["afternoon"],
+        },
+        headers=_admin_headers(),
+    )
+    assert inactive.status_code == 201
+    inactive_id = inactive.json()["id"]
+    deactivated = await client.patch(
+        f"{LIST_PATH}/{inactive_id}/active",
+        json={"is_active": False},
+        headers=_admin_headers(),
+    )
+    assert deactivated.status_code == 200
+
+    return child_id, active.json()["id"], inactive_id
+
+
+async def test_admin_lists_active_and_inactive_definitions_with_full_shape(
+    admin_client: object,
+) -> None:
+    """Both definitions come back, by rising id, in the documented shape."""
+    child_id, active_id, inactive_id = await _seed_active_and_inactive(
+        admin_client
+    )
+
+    response = await admin_client.get(LIST_PATH, headers=_admin_headers())
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "definitions": [
+            {
+                "id": active_id,
+                "title": "Tidy the den",
+                "description": "Put everything back where it belongs.",
+                "icon": "toy-box",
+                "is_active": True,
+                "skip_on_away": True,
+                "rule": {
+                    "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,
+                },
+                "assignees": [{"id": child_id, "display_name": "Ada"}],
+                "windows": [
+                    {"window": "morning", "due_time": None},
+                    {"window": "evening", "due_time": "19:30"},
+                ],
+            },
+            {
+                "id": inactive_id,
+                "title": "Water the plants",
+                "description": None,
+                "icon": None,
+                "is_active": False,
+                "skip_on_away": True,
+                "rule": {
+                    "rule_type": "daily",
+                    "interval": 1,
+                    "weekday_set": None,
+                    "day_of_month": None,
+                    "nth_weekday": None,
+                    "nth_weekday_weekday": None,
+                    "month": None,
+                    "start_date": "2026-03-01",
+                    "end_date": None,
+                },
+                "assignees": [{"id": child_id, "display_name": "Ada"}],
+                "windows": [{"window": "afternoon", "due_time": None}],
+            },
+        ]
+    }
+
+
+async def test_list_items_match_the_create_response_shape(
+    admin_client: object,
+) -> None:
+    """Each listed item is exactly what the create route returned."""
+    child = await admin_client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Ben"},
+        headers=_admin_headers(),
+    )
+    created = await admin_client.post(
+        LIST_PATH,
+        json={
+            "title": "Feed the cat",
+            "rule": {
+                "rule_type": "daily",
+                "interval": 1,
+                "start_date": "2026-03-01",
+            },
+            "assignee_child_ids": [child.json()["id"]],
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+
+    response = await admin_client.get(LIST_PATH, headers=_admin_headers())
+
+    assert response.status_code == 200
+    assert response.json() == {"definitions": [created.json()]}
+
+
+async def test_skip_on_away_round_trips_true_and_false_through_list(
+    admin_client: object,
+) -> None:
+    """skip_on_away posted true, false, or omitted lists back as stored."""
+    child = await admin_client.post(
+        "/api/v1/admin/children",
+        json={"display_name": "Cleo"},
+        headers=_admin_headers(),
+    )
+    child_id = child.json()["id"]
+
+    async def create(title: str, extra: dict[str, object]) -> dict:
+        response = await admin_client.post(
+            LIST_PATH,
+            json={
+                "title": title,
+                "rule": {
+                    "rule_type": "daily",
+                    "interval": 1,
+                    "start_date": "2026-03-01",
+                },
+                "assignee_child_ids": [child_id],
+                "windows": ["morning"],
+                **extra,
+            },
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 201
+        return response.json()
+
+    skipping = await create("Skip when away", {"skip_on_away": True})
+    keeping = await create("Keep when away", {"skip_on_away": False})
+    defaulted = await create("Default", {})
+    assert skipping["skip_on_away"] is True
+    assert keeping["skip_on_away"] is False
+    assert defaulted["skip_on_away"] is True
+
+    response = await admin_client.get(LIST_PATH, headers=_admin_headers())
+
+    assert response.status_code == 200
+    listed = {item["id"]: item for item in response.json()["definitions"]}
+    assert listed[skipping["id"]]["skip_on_away"] is True
+    assert listed[keeping["id"]]["skip_on_away"] is False
+    assert listed[defaulted["id"]]["skip_on_away"] is True
+    assert response.json() == {"definitions": [skipping, keeping, defaulted]}
+
+
+async def test_empty_household_lists_no_definitions(
+    admin_client: object,
+) -> None:
+    """With no definitions stored the list is 200 and empty."""
+    response = await admin_client.get(LIST_PATH, headers=_admin_headers())
+
+    assert response.status_code == 200
+    assert response.json() == {"definitions": []}
+
+
+async def test_list_without_credentials_is_401(admin_client: object) -> None:
+    """An absent credential is the uniform 401."""
+    response = await admin_client.get(LIST_PATH)
+
+    assert response.status_code == 401
+
+
+async def test_list_with_panel_service_token_is_403(
+    admin_client: object,
+) -> None:
+    """The panel service token is refused outright, never a JWT."""
+    response = await admin_client.get(
+        LIST_PATH,
+        headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
+    )
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
+
+
+async def test_list_with_non_admin_jwt_is_403(admin_client: object) -> None:
+    """A valid JWT without the nestquest-admins group is 403."""
+    non_admin = make_token(groups=("some-other-group",))
+    response = await admin_client.get(
+        LIST_PATH, headers={"Authorization": f"Bearer {non_admin}"}
+    )
+
+    assert response.status_code == 403
+    assert response.json()["detail"] == FORBIDDEN_DETAIL

tests/test_api_admin_sse.py

@@ -0,0 +1,283 @@
+"""Admin-plane SSE tests: the admin transition stream (task 92f75680).
+
+These tests exercise ``GET /api/v1/admin/events`` — the admin plane's
+Server-Sent Events stream for the Admin PWA — against a REAL server:
+
+- The stream inherits the admin router's ONE ``require_admin``
+  dependency: an absent credential is 401, the panel service token and
+  a non-admin JWT are 403, all before any stream is established.
+- An admin JWT subscribes; each of the four documented transitions
+  (quest-completed, quest-uncompleted, quest-missed, child-day-complete)
+  arrives as an ``event: <type>`` / ``data: <json>`` frame carrying its
+  documented payload fields (design/ENTITIES-AND-SERVICES.md §3).
+- The admin and panel streams are fed by the SAME in-process publisher:
+  one transition arrives, identically, on both.
+
+Streaming approach: :mod:`tests.test_api_sse`'s live-uvicorn reader
+(httpx's ASGITransport buffers SSE bodies), on the admin-enabled
+:class:`tests.test_api_admin_uncomplete._AdminServer` whose JWKS cache
+is stubbed, so a real admin JWT verifies with NO network access.
+"""
+from __future__ import annotations
+
+import datetime
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+
+from api import routes_admin, routes_panel
+from tests.admin_jwt_harness import make_token
+from tests.test_api_admin_uncomplete import _AdminServer
+from tests.test_api_panel import PANEL_TOKEN, _seed_household
+from tests.test_api_sse import (
+    EVENT_CHILD_DAY_COMPLETE,
+    EVENT_QUEST_COMPLETED,
+    EVENT_QUEST_MISSED,
+    EVENT_QUEST_UNCOMPLETED,
+    READ_TIMEOUT,
+    _stream,
+)
+
+EVENTS_PATH = "/api/v1/admin/events"
+
+
+@pytest.fixture
+def temp_db_path(tmp_path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def admin_live(temp_db_path: str) -> SimpleNamespace:
+    """A seeded household served by a live admin-enabled uvicorn app.
+
+    Both planes' clocks are pinned to 12:00 UTC (past Ada's 10:00 due,
+    before Bo's 17:00) BEFORE the server starts — the app is created
+    inside uvicorn's factory thread.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    pinned_now = datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
+    )
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    original_panel = routes_panel._local_now
+    original_admin = routes_admin._local_now
+    routes_panel._local_now = lambda: pinned_now
+    routes_admin._local_now = lambda: pinned_now
+    try:
+        with _AdminServer(temp_db_path) as server:
+            client = httpx.AsyncClient(
+                base_url=server.url(""), timeout=READ_TIMEOUT
+            )
+            async with client:
+                yield SimpleNamespace(client=client, seed=seed, server=server)
+    finally:
+        routes_panel._local_now = original_panel
+        routes_admin._local_now = original_admin
+
+
+async def _panel_complete(live: SimpleNamespace, instance_id: int, child_id: int):
+    """Complete ``instance_id`` through the panel route (a real transition)."""
+    response = await live.client.post(
+        f"/api/v1/panel/instances/{instance_id}/complete",
+        headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+        json={"actor_child_id": child_id},
+    )
+    assert response.status_code == 200
+
+
+# --- auth: the ONE require_admin dependency, no stream on refusal ----------
+
+
+async def test_admin_events_refuses_absent_panel_and_non_admin(
+    admin_live: SimpleNamespace,
+) -> None:
+    """No credential is 401; the panel token and a non-admin JWT are 403.
+
+    Each refusal is a complete (non-streaming) JSON error response, so
+    no stream is established for any of them.
+    """
+    client = admin_live.client
+
+    absent = await client.get(EVENTS_PATH)
+    assert absent.status_code == 401
+    assert not absent.headers["content-type"].startswith("text/event-stream")
+
+    panel = await client.get(
+        EVENTS_PATH, headers={"Authorization": f"Bearer {PANEL_TOKEN}"}
+    )
+    assert panel.status_code == 403
+    assert panel.json()["detail"] == (
+        "This credential type is not accepted for admin access"
+    )
+
+    non_admin = await client.get(
+        EVENTS_PATH,
+        headers={
+            "Authorization": f"Bearer {make_token(groups=('other-group',))}"
+        },
+    )
+    assert non_admin.status_code == 403
+    assert non_admin.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+
+async def test_admin_jwt_is_not_accepted_on_panel_events(
+    admin_live: SimpleNamespace,
+) -> None:
+    """The panel plane is unchanged: an admin JWT does not open it."""
+    response = await admin_live.client.get(
+        "/api/v1/panel/events",
+        headers={"Authorization": f"Bearer {make_token()}"},
+    )
+    assert response.status_code == 401
+
+
+async def test_admin_events_sets_no_cache(admin_live: SimpleNamespace) -> None:
+    """The admin stream is ``text/event-stream`` with ``no-cache``."""
+    async with admin_live.client.stream(
+        "GET",
+        EVENTS_PATH,
+        headers={"Authorization": f"Bearer {make_token()}"},
+    ) as response:
+        assert response.status_code == 200
+        assert response.headers["content-type"].startswith("text/event-stream")
+        assert response.headers["cache-control"] == "no-cache"
+
+
+# --- the four documented transitions on the admin stream -------------------
+
+
+async def test_admin_stream_carries_completed_and_day_complete(
+    admin_live: SimpleNamespace,
+) -> None:
+    """A panel completion streams quest-completed then child-day-complete.
+
+    Ada's only quest today is Pack bag, so completing it also clears her
+    day; both frames arrive on the ADMIN stream with their documented
+    payload fields.
+    """
+    seed = admin_live.seed
+    pack_bag = seed.pack_bag_instance
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_COMPLETED == "nestquest_quest_completed"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["instance_id"] == pack_bag.id
+        assert payload["quest_title"] == "Pack bag"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == pack_bag.due_date
+        assert payload["due_time"] == "10:00"
+        assert payload["occurred_at"]
+        assert payload["was_on_time"] is False  # 10:00 due, 12:00 now
+        completed_at = payload["occurred_at"]
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_CHILD_DAY_COMPLETE
+        assert event_type == "nestquest_child_day_complete"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["quests_due"] == 1
+        assert payload["quests_completed"] == 1
+        assert payload["occurred_at"] == completed_at
+
+
+async def test_admin_stream_carries_uncompleted(
+    admin_live: SimpleNamespace,
+) -> None:
+    """The admin uncomplete route's reversal streams quest-uncompleted."""
+    seed = admin_live.seed
+    brush = seed.brush_teeth_instance
+    admin_headers = {"Authorization": f"Bearer {make_token()}"}
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        response = await admin_live.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=admin_headers,
+        )
+        assert response.status_code == 200
+        assert response.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"]
+    assert "was_on_time" not in payload
+
+
+async def test_admin_stream_carries_missed(admin_live: SimpleNamespace) -> None:
+    """The missed-sweep publisher's transition streams to the admin client.
+
+    Drives the API-side helper the sweep calls, on the SERVER's loop,
+    against the app's own database and its ONE publisher.
+    """
+    from api import transitions
+
+    seed = admin_live.seed
+    stale = seed.stale_instance
+    app = admin_live.server.app
+    stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    async with stream:
+        admin_live.server.run_on_loop(
+            transitions.publish_quest_missed(
+                app.state.db.database, stale.id, app.state.publisher
+            )
+        )
+        event_type, payload = await stream.next_event()
+    assert event_type == EVENT_QUEST_MISSED == "nestquest_quest_missed"
+    assert payload["child_id"] == seed.ada.id
+    assert payload["child_name"] == "Ada"
+    assert payload["instance_id"] == stale.id
+    assert payload["quest_title"] == "Stale chore"
+    assert payload["window"] == "morning"
+    assert payload["due_date"] == stale.due_date
+    assert payload["due_time"] == "09:00"
+    assert payload["occurred_at"]
+
+
+# --- one publisher feeds both planes ---------------------------------------
+
+
+async def test_admin_and_panel_streams_share_one_publisher(
+    admin_live: SimpleNamespace,
+) -> None:
+    """One transition arrives, identically, on the admin AND panel streams.
+
+    The route publishes once to ``app.state.publisher``; both planes'
+    subscribers receive the same frames (same ``occurred_at``), which a
+    second, admin-only publisher could never produce.
+    """
+    seed = admin_live.seed
+    pack_bag = seed.pack_bag_instance
+    admin_stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
+    panel_stream = _stream(admin_live.server.url("/api/v1/panel/events"))
+    async with admin_stream, panel_stream:
+        await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
+
+        admin_events = [
+            await admin_stream.next_event(),
+            await admin_stream.next_event(),
+        ]
+        panel_events = [
+            await panel_stream.next_event(),
+            await panel_stream.next_event(),
+        ]
+    assert [event_type for event_type, _ in admin_events] == [
+        EVENT_QUEST_COMPLETED,
+        EVENT_CHILD_DAY_COMPLETE,
+    ]
+    assert admin_events == panel_events

tests/test_api_sse.py

@@ -158,7 +158,8 @@ class _EventStream:
     parsed frame with a timeout.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, token: str = PANEL_TOKEN) -> None:
+        self.token = token
         self.queue: asyncio.Queue = asyncio.Queue()
         self._task: asyncio.Task | None = None
         self._client: httpx.AsyncClient | None = None
@@ -170,7 +171,7 @@ class _EventStream:
         self._stream_cm = self._client.stream(
             "GET",
             self.url,
-            headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
+            headers={"Authorization": f"Bearer {self.token}"},
         )
         self._response = await self._stream_cm.__aenter__()
         assert self._response.status_code == 200
@@ -263,9 +264,13 @@ async def seeded_panel(temp_db_path: str) -> SimpleNamespace:
         routes_panel._local_now = original
 
 
-def _stream(url: str) -> _EventStream:
-    """A subscribed SSE reader for ``url``."""
-    return _EventStream()._set_url(url)
+def _stream(url: str, token: str = PANEL_TOKEN) -> _EventStream:
+    """A subscribed SSE reader for ``url``, sending ``token`` as Bearer.
+
+    Defaults to the panel service token; the admin-plane stream tests
+    pass an admin JWT instead.
+    """
+    return _EventStream(token)._set_url(url)
 
 
 # --- auth and content type ----------------------------------------------

tests/test_dao_rules.py

@@ -847,6 +847,7 @@ def test_definition_update_cannot_change_assignee_or_active(
             "description",
             "icon",
             "due_time",
+            "skip_on_away",
         }
         assert await definitions.update(
             definition.id, title="A2"
@@ -862,6 +863,72 @@ def test_definition_update_cannot_change_assignee_or_active(
     _with_db(tmp_path, "definition-update-shape.db")(_body)
 
 
+def test_definition_update_toggles_skip_on_away_and_omission_keeps_it(
+    tmp_path,
+) -> None:
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        assert definition.skip_on_away is True
+        assert await definitions.update(
+            definition.id, skip_on_away=False
+        ) == 1
+        assert (await definitions.get(definition.id)).skip_on_away is False
+        # Omitting the flag (None) leaves the stored False untouched.
+        assert await definitions.update(definition.id, title="A2") == 1
+        kept = await definitions.get(definition.id)
+        assert kept.title == "A2"
+        assert kept.skip_on_away is False
+        assert await definitions.update(definition.id) == 0
+        assert await definitions.update(
+            definition.id, skip_on_away=True
+        ) == 1
+        assert (await definitions.get(definition.id)).skip_on_away is True
+
+    _with_db(tmp_path, "definition-update-skip-on-away.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [2, -1, 0.0, 1.5, "1", "true", b"\x01"])
+def test_skip_on_away_rejected_unless_zero_or_one_at_every_write(
+    tmp_path, bad
+) -> None:
+    """Every DAO entry point taking the flag rejects non-0/1 values.
+
+    This is the only guard on migrated databases, whose column was
+    added without the fresh-schema CHECK constraint.
+    """
+    async def _body(database, rules, definitions, children, child):
+        rule = await rules.create("daily", "2026-09-14")
+        definition = await definitions.create("A", rule.id, NOW)
+        with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
+            await definitions.create("B", rule.id, NOW, skip_on_away=bad)
+        with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
+            await definitions.create_with_rule_and_windows(
+                "C",
+                schedule_rule_to_storage(
+                    ScheduleRule(
+                        rule_type=RuleType.DAILY, start_date="2026-09-14"
+                    )
+                ),
+                NOW,
+                [child.id],
+                [("morning", None)],
+                skip_on_away=bad,
+            )
+        with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
+            await definitions.edit_definition(
+                definition.id, title="A2", skip_on_away=bad
+            )
+        with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
+            await definitions.update(
+                definition.id, title="A2", skip_on_away=bad
+            )
+        stored = await definitions.list_all()
+        assert [(d.title, d.skip_on_away) for d in stored] == [("A", True)]
+
+    _with_db(tmp_path, "definition-skip-on-away-invalid.db")(_body)
+
+
 def test_definition_update_no_fields_returns_zero(tmp_path) -> None:
     async def _body(database, rules, definitions, children, child):
         rule = await rules.create("daily", "2026-09-14")

tests/test_materialize.py

@@ -657,6 +657,133 @@ def test_materialize_honours_overrides(tmp_path) -> None:
     _with_db(tmp_path, "materialize-override.db")(_body)
 
 
+async def _absent_midweek_setup(database, skip_on_away: bool):
+    """One child absent on days 2..4 of a future week, one daily quest.
+
+    Returns ``(child_id, definition_id, start, end_iso, absent_dates)``.
+    """
+    from custom_components.nestquest.core.dao_presence import (
+        PresenceOverridesDao,
+    )
+
+    child = await ChildrenDao(database).create("Ada", NOW)
+    start = _future_monday()
+    end_iso = (start + datetime.timedelta(days=6)).isoformat()
+    absent = {
+        (start + datetime.timedelta(days=offset)).isoformat()
+        for offset in (2, 3, 4)
+    }
+    await PresenceOverridesDao(database).create(
+        child.id, min(absent), max(absent), False, note="away"
+    )
+    created = await create_quest_definition(
+        database,
+        "Daily chore",
+        ScheduleRule(rule_type=RuleType.DAILY, start_date=start.isoformat()),
+        [child.id],
+        ["morning"],
+        skip_on_away=skip_on_away,
+    )
+    assert created.definition.skip_on_away is skip_on_away
+    return child.id, created.definition.id, start, end_iso, absent
+
+
+def test_materialize_skip_on_away_true_skips_absent_dates(tmp_path) -> None:
+    async def _body(database):
+        child_id, _, start, end_iso, absent = await _absent_midweek_setup(
+            database, skip_on_away=True
+        )
+        count = await materialize(database, start.isoformat(), end_iso)
+        records = await QuestInstancesDao(database).list_by_date_range(
+            child_id, start.isoformat(), end_iso
+        )
+        due_dates = {record.due_date for record in records}
+        assert count == 4
+        assert due_dates.isdisjoint(absent)
+        assert len(due_dates) == 4
+
+    _with_db(tmp_path, "skip-on-away-true.db")(_body)
+
+
+def test_materialize_skip_on_away_false_generates_absent_dates(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        child_id, _, start, end_iso, absent = await _absent_midweek_setup(
+            database, skip_on_away=False
+        )
+        count = await materialize(database, start.isoformat(), end_iso)
+        records = await QuestInstancesDao(database).list_by_date_range(
+            child_id, start.isoformat(), end_iso
+        )
+        due_dates = {record.due_date for record in records}
+        assert count == 7
+        assert absent <= due_dates
+        assert len(due_dates) == 7
+
+    _with_db(tmp_path, "skip-on-away-false.db")(_body)
+
+
+def test_edit_skip_on_away_regenerates_absent_dates(tmp_path) -> None:
+    """Toggling the flag through the edit path rebuilds future instances."""
+    async def _body(database):
+        child_id, definition_id, start, end_iso, absent = (
+            await _absent_midweek_setup(database, skip_on_away=True)
+        )
+        await materialize(database, start.isoformat(), end_iso)
+        dao = QuestInstancesDao(database)
+
+        async def due_dates() -> set[str]:
+            records = await dao.list_by_date_range(
+                child_id, start.isoformat(), end_iso
+            )
+            return {record.due_date for record in records}
+
+        assert (await due_dates()).isdisjoint(absent)
+
+        edited = await edit_quest_definition(
+            database, definition_id, skip_on_away=False
+        )
+        assert edited.definition.skip_on_away is False
+        assert absent <= await due_dates()
+
+        # None leaves the stored flag unchanged.
+        unchanged = await edit_quest_definition(
+            database, definition_id, title="Daily chore (renamed)"
+        )
+        assert unchanged.definition.skip_on_away is False
+
+        edited = await edit_quest_definition(
+            database, definition_id, skip_on_away=True
+        )
+        assert edited.definition.skip_on_away is True
+        assert (await due_dates()).isdisjoint(absent)
+
+    _with_db(tmp_path, "skip-on-away-edit.db")(_body)
+
+
+def test_skip_on_away_rejects_non_bool(tmp_path) -> None:
+    async def _body(database):
+        child = await ChildrenDao(database).create("Ada", NOW)
+        rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-01-01")
+        with pytest.raises(ValueError, match="skip_on_away must be a real bool"):
+            await create_quest_definition(
+                database, "Chore", rule, [child.id], ["morning"],
+                skip_on_away=1,
+            )
+        created = await create_quest_definition(
+            database, "Chore", rule, [child.id], ["morning"]
+        )
+        with pytest.raises(ValueError, match="skip_on_away must be a real bool"):
+            await edit_quest_definition(
+                database, created.definition.id, skip_on_away="false"
+            )
+        stored = await QuestDefinitionsDao(database).get(created.definition.id)
+        assert stored.skip_on_away is True
+
+    _with_db(tmp_path, "skip-on-away-validation.db")(_body)
+
+
 def test_materialize_rejects_malformed_or_inverted_bounds(tmp_path) -> None:
     async def _body(database):
         start = (_future_monday()).isoformat()

tests/test_migrations.py

@@ -153,7 +153,9 @@ def test_migration_list_shape() -> None:
     migration 3 brings pre-D-008 definitions to the multi-assignee
     model, migration 4 adds the windows table, migration 5 rebuilds
     quest_instances onto the widened (definition, child, date, window)
-    key, and migration 6 adds actor_child_id to completion_events.
+    key, migration 6 adds actor_child_id to completion_events,
+    migration 7 adds the meta_state table and migration 8 adds
+    skip_on_away to quest_definitions.
     """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
     assert callable(MIGRATIONS[1])
@@ -162,7 +164,8 @@ def test_migration_list_shape() -> None:
     assert callable(MIGRATIONS[4])
     assert callable(MIGRATIONS[5])
     assert MIGRATIONS[6] == SCHEMA_V7_META_STATE_DDL
-    assert len(MIGRATIONS) == 7
+    assert callable(MIGRATIONS[7])
+    assert len(MIGRATIONS) == 8
 
 
 # ---------------------------------------------------------------------------
@@ -564,6 +567,13 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
             database.fetch_all("PRAGMA table_info(quest_definitions)")
         )
         assert "child_id" not in {c[1] for c in columns}
+        # The rebuilt row reads the pre-flag behaviour: skip on away.
+        skip = _run(
+            database.fetch_one(
+                "SELECT skip_on_away FROM quest_definitions WHERE id = 1"
+            )
+        )
+        assert skip == (1,)
         # The single assignee moved into the assignees table.
         assignees = _run(
             database.fetch_all(
@@ -637,6 +647,87 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
         _run(database.close())
 
 
+# ---------------------------------------------------------------------------
+# Migration 8: skip_on_away on quest_definitions (Feature 22)
+# ---------------------------------------------------------------------------
+
+
+def _seed_v7_database(database) -> None:
+    """Build a genuine version-7 database holding one definition.
+
+    Migrations 1..7 run first; the fresh v1 DDL already carries
+    ``skip_on_away`` (canonical shape), so the column is dropped to
+    reproduce the exact pre-migration-8 table a shipped v7 file holds,
+    and a definition row is written through that old column list.
+    """
+    assert _migrate(database, MIGRATIONS[:7]) == 7
+    _run(
+        database.execute(
+            "ALTER TABLE quest_definitions DROP COLUMN skip_on_away"
+        )
+    )
+    columns = _run(database.fetch_all("PRAGMA table_info(quest_definitions)"))
+    assert "skip_on_away" not in {c[1] for c in columns}
+    _run(
+        database.execute(
+            "INSERT INTO schedule_rules (id, rule_type, interval, "
+            "start_date) VALUES (1, 'daily', 1, '2026-09-01')"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO quest_definitions (id, title, schedule_rule_id, "
+            "is_active, created_at) VALUES (1, 'Dishes', 1, 1, "
+            "'2026-09-01T00:00:00+00:00')"
+        )
+    )
+
+
+def test_v7_definitions_gain_skip_on_away_defaulting_true(tmp_path) -> None:
+    from custom_components.nestquest.core.dao_rules import (
+        QuestDefinitionsDao,
+    )
+
+    database = _open_db(tmp_path / "v7-upgrade.db")
+    try:
+        _seed_v7_database(database)
+        before = _version(database)
+        assert before == 7
+
+        final = _migrate(database)
+
+        assert final == before + 1
+        assert _version(database) == before + 1
+        columns = {
+            c[1]: c
+            for c in _run(
+                database.fetch_all("PRAGMA table_info(quest_definitions)")
+            )
+        }
+        assert "skip_on_away" in columns
+        # (cid, name, type, notnull, dflt_value, pk)
+        assert columns["skip_on_away"][2:5] == ("INTEGER", 1, "1")
+        record = _run(QuestDefinitionsDao(database).get(1))
+        assert record is not None
+        assert record.title == "Dishes"
+        assert record.skip_on_away is True
+    finally:
+        _run(database.close())
+
+
+def test_skip_on_away_migration_is_noop_on_fresh_schema(tmp_path) -> None:
+    """A fresh file already has the column; step 8 only stamps."""
+    database = _open_db(tmp_path / "fresh-v8.db")
+    try:
+        assert _migrate(database) == len(MIGRATIONS)
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(quest_definitions)")
+        )
+        assert [c[1] for c in columns].count("skip_on_away") == 1
+    finally:
+        _run(database.close())
+
+
 # ---------------------------------------------------------------------------
 # Idempotence: second start is a no-op
 # ---------------------------------------------------------------------------

tests/test_schema.py

@@ -322,6 +322,7 @@ def test_quest_definitions_columns_types_and_constraints(tmp_path) -> None:
             ("due_time", "TEXT", 0, None, 0),
             ("is_active", "INTEGER", 1, "1", 0),
             ("created_at", "TEXT", 1, None, 0),
+            ("skip_on_away", "INTEGER", 1, "1", 0),
         ]
     finally:
         _run(database.close())
@@ -633,6 +634,30 @@ def test_schedule_rules_rule_type_check_enforced(tmp_path) -> None:
         _run(database.close())
 
 
+@pytest.mark.parametrize("bad_flag", [2, -1])
+def test_quest_definitions_skip_on_away_check_enforced(
+    tmp_path, bad_flag
+) -> None:
+    database = _open_db(tmp_path / f"skip-on-away-{bad_flag}.db")
+    try:
+        _apply(database)
+        _child(database)
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_definition(database, skip_on_away=bad_flag)
+        definition_id = _insert_definition(database, skip_on_away=0)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _run(
+                database.execute(
+                    "UPDATE quest_definitions SET skip_on_away = ? "
+                    "WHERE id = ?",
+                    (bad_flag, definition_id),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
 def test_schedule_rules_interval_zero_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "interval-zero.db")
     try: