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

api/routes_admin.py

@@ -58,6 +58,13 @@ 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
@@ -283,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
 
@@ -290,6 +307,7 @@ import datetime
 from typing import Annotated, Literal, NoReturn
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi.responses import StreamingResponse
 from pydantic import (
     BaseModel,
     StrictBool,
@@ -318,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
@@ -666,6 +685,34 @@ class AdminPresenceOverrideListResponse(BaseModel):
     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``.
 
@@ -1608,6 +1655,45 @@ 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",
@@ -2131,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
+    Returns the shared :func:`api.sse.transition_event_stream` response
-    :func:`_sse_stream` off the app's single in-process publisher
+    off the app's single in-process publisher (``app.state.publisher``,
-    (``app.state.publisher``, installed in the lifespan).  The stream
+    installed in the lifespan).  The stream ends when the client
-    ends when the client disconnects, at which point the subscription
+    disconnects, at which point the subscription is torn down (its
-    is torn down (its queue removed) with nothing leaked.
+    queue removed) with nothing leaked.
     """
-    return StreamingResponse(
+    return transition_event_stream(request)
-        _sse_stream(request),
-        media_type="text/event-stream",
-        headers={"Cache-Control": "no-cache"},
-    )
 
 
 __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/materialize.py

@@ -64,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,
@@ -126,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(
@@ -145,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,
@@ -207,13 +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,
-         snapshot.definition.skip_on_away)
-        for snapshot in snapshots
-    ]
-
     generated_at = _now_stamp()
     instances = QuestInstancesDao(database)
 
@@ -228,32 +299,20 @@ async def materialize(
     start = max(start, today_date)
 
     count = 0
-    cursor = start
+    for generated in _generation_preview(
-    while cursor <= end:
+        snapshots, engine, start, end, child_filter=child_filter
-        iso = cursor.isoformat()
+    ):
-        for (
+        written = await instances.upsert_if_valid(
-            definition_id, rule, assignees, windows, skip_on_away
+            generated.definition_id,
-        ) in definitions:
+            generated.child_id,
-            if not occurs_on(rule, cursor):
+            generated.due_date,
-                continue
+            generated_at,
-            for child in assignees:
+            window=generated.window,
-                if child_filter is not None and child.id not in child_filter:
+            due_time=generated.due_time,
-                    continue
+            today=today,
-                if skip_on_away and not engine.is_present(child.id, cursor):
+        )
-                    continue
+        if written is not None:
-                for window in windows:
+            count += 1
-                    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)
     return count
 
 
@@ -379,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/presence_management.py

@@ -31,13 +31,14 @@ 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
 
 
@@ -134,6 +135,42 @@ 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:

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_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:
+def _stream(url: str, token: str = PANEL_TOKEN) -> _EventStream:
-    """A subscribed SSE reader for ``url``."""
+    """A subscribed SSE reader for ``url``, sending ``token`` as Bearer.
-    return _EventStream()._set_url(url)
+
+    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 ----------------------------------------------