Publish the sweep-built missed events directly in the admin trigger (P1)

The missed-sweep route re-fetched every instance through
publish_quest_missed AFTER core_sweep.run_missed_sweep had already
committed its idempotency watermark.  A concurrent regenerate deleting
an instance — or a transient database error — made that re-fetch raise
ValueError uncaught (a bare 500) and dropped the missed event forever:
the watermark had advanced, so no future sweep re-examines the
instance.  The re-fetch also discarded the ONE shared occurred_at
stamp the sweep computed for the whole run, re-stamping each event
individually.

Publish the already-built (event_type, payload) pairs straight to the
app's ONE publisher instead — the same shape api/scheduler.py uses —
leaving no failure window between the watermark and the announcement.
The deterministic concurrent-deletion test wraps the sweep to delete a
swept instance after it built its events: the trigger must still
answer 200 and stream both sweep-built payloads (it 500'd before the
fix), and all events of one run share the run's occurred_at.

Also add the missing trailing newlines to core/sweep.py and
api/scheduler.py (P3).

api/app.py

@@ -48,6 +48,7 @@ from api.database import DatabaseState, make_database
 from api.events import TransitionPublisher
 from api.routes_admin import router as admin_router
 from api.routes_panel import router as panel_router
+from api.scheduler import MissedSweepScheduler
 
 
 def _build_lifespan(db_path: str):
@@ -57,7 +58,11 @@ def _build_lifespan(db_path: str):
     path in without threading it through FastAPI's lifespan signature.
     The lifespan also installs the app's ONE in-process transition
     publisher on ``app.state.publisher`` (see :mod:`api.events`) so
-    every route and SSE subscriber shares that single instance.
+    every route and SSE subscriber shares that single instance, and
+    starts the daily missed-sweep scheduler
+    (:class:`api.scheduler.MissedSweepScheduler`) against the same
+    database and publisher — cancelled before the connection closes on
+    shutdown.
     """
 
     @asynccontextmanager
@@ -65,10 +70,16 @@ def _build_lifespan(db_path: str):
         state = make_database(db_path)
         app.state.db = state
         app.state.publisher = TransitionPublisher()
+        scheduler = MissedSweepScheduler(
+            state.database, publisher=app.state.publisher
+        )
+        app.state.sweep_scheduler = scheduler
         try:
             await state.open_and_migrate()
+            scheduler.start()
             yield
         finally:
+            await scheduler.stop()
             await state.close()
 
     return lifespan

api/nestquest_core.py

@@ -129,6 +129,17 @@ core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
 #: the snapshot builder's ``settings`` parameter takes.
 core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
 
+#: The durable settings store (``core.settings_store.load_settings`` /
+#: ``update_settings``) the admin plane's settings routes read and
+#: write — the API's OWN settings source: a validated JSON document in
+#: the existing ``nestquest_meta_state`` table, resolved through the
+#: same core settings validators (the integration's HA config-entry
+#: options remain a temporary separate source until Feature 18 wires
+#: the client to the API).
+core_settings_store = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.settings_store"
+)
+
 #: The children business layer (``core.children.create_child`` et al.)
 #: the admin plane's children routes call.
 core_children = importlib.import_module(f"{_CORE_MODULE_NAME}.children")
@@ -187,6 +198,12 @@ core_dao_children = importlib.import_module(
 #: delegating.
 core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules")
 
+#: The HA-free missed sweep (``core.sweep.run_missed_sweep`` +
+#: ``SWEEP_WATERMARK_KEY``) the admin plane's trigger route calls and
+#: the daily scheduler runs — the ONE sweep policy the integration's
+#: bus shim shares.
+core_sweep = importlib.import_module(f"{_CORE_MODULE_NAME}.sweep")
+
 __all__ = [
     "core",
     "core_children",
@@ -206,4 +223,6 @@ __all__ = [
     "core_recurrence",
     "core_snapshot",
     "core_settings",
+    "core_settings_store",
+    "core_sweep",
 ]

api/routes_admin.py

@@ -116,6 +116,46 @@ owns every rule so the two routes cannot drift:
   a ``Content-Disposition`` attachment filename and the documented,
   stable header row (Admin spec §5 event-row fields).
 
+The settings routes (task 2b3de7e5) follow the same pattern over
+:mod:`nestquest_core.settings_store` — the API's OWN durable settings
+store.  The API now OWNS settings in the database (a validated JSON
+document in the existing ``nestquest_meta_state`` key/value table); the
+integration's Home-Assistant config-entry options remain a temporary,
+SEPARATE source until Feature 18 wires the client to the API — the two
+are deliberately not unified here:
+
+- ``GET /settings`` returns the current effective settings: all ten
+  documented fields (the rolling horizon, the day rollover time, the
+  notify target, the three notification times and the four enable
+  toggles) — the defaults on a fresh database.
+- ``PATCH /settings`` updates ONLY the supplied fields: the body is a
+  JSON object of settings field names to new values, passed STRAIGHT
+  to :func:`nestquest_core.settings_store.update_settings` — the
+  unknown-field rejection, the per-field validation (the core settings
+  validators: a positive int, strict ``HH:MM`` strings, real booleans)
+  and the validate-everything-before-write atomicity all live in the
+  core, so a rejected update never changes the stored settings.  A
+  JSON ``null`` value resets its field to the default (the settings
+  module's uniform None-as-absent rule).  The response is the updated
+  effective settings, the same shape ``GET`` returns.
+
+The missed-sweep trigger (task 058c7b69) follows the same pattern over
+:mod:`nestquest_core.sweep` — the HA-free sweep policy the
+integration's rollover listener shares:
+
+- ``POST /missed-sweep`` runs the sweep for the API host's local
+  ``today`` (ONE :func:`_local_now` clock read threaded into the core
+  call) and publishes each returned ``(event_type, payload)`` pair
+  DIRECTLY on the app's ONE transition publisher — the sweep-built
+  payload, never re-fetched (the core commits its watermark before
+  returning, so a failed re-fetch after it would lose the missed event
+  forever), the same publish-what-the-core-built shape
+  api/scheduler.py uses.  The watermark rule
+  (a same-day rerun announces nothing, a post-downtime run sweeps the
+  accumulated window once), the no-completion-event rule and the
+  read-only-over-the-domain-tables guarantee all live in the core; the
+  response reports how many transitions were published.
+
 Error mapping (every children, quest-definition and presence route,
 deliberately narrow):
 
@@ -174,6 +214,13 @@ deliberately narrow):
   invalid, the same failure class Pydantic reports.  Only
   ``ValueError`` is caught, so an unexpected failure re-raises rather
   than becoming a 4xx.
+- The SETTINGS routes have no path ids either, so they have no 404
+  case: a body that is not a JSON object is FastAPI's 422 before any
+  handler code runs, and every core ``ValueError`` — an unknown field
+  or an invalid value, each naming the offending field, raised BEFORE
+  any write so the stored settings are never half-changed — is 422
+  (:func:`_raise_settings_error`).  Only ``ValueError`` is caught, so
+  an unexpected failure re-raises rather than becoming a 4xx.
 
 PATCH edit semantics: the body model's fields are optional and only
 SUPPLIED fields are forwarded (``model_dump(exclude_unset=True)``), so
@@ -205,6 +252,9 @@ from api.nestquest_core import (
     core_presence_management,
     core_quest_definitions,
     core_recurrence,
+    core_settings,
+    core_settings_store,
+    core_sweep,
 )
 
 #: All admin-plane routes share this router; the ONE admin dependency
@@ -601,6 +651,38 @@ class AdminHistoryResponse(BaseModel):
     rows: list[AdminHistoryRow]
 
 
+class AdminSettingsResponse(BaseModel):
+    """The effective settings as the admin plane serializes them.
+
+    Mirrors the core :class:`~nestquest_core.settings.NestQuestSettings`
+    — all TEN documented fields, resolved (every absent stored field
+    already fell back to its default in the core).  The same shape both
+    the GET and the PATCH route answer with.
+    """
+
+    horizon_days: int
+    day_rollover_time: str
+    notify_target: str
+    morning_summary_time: str
+    afternoon_reminder_time: str
+    end_of_day_report_time: str
+    morning_summary_enabled: bool
+    afternoon_reminder_enabled: bool
+    end_of_day_report_enabled: bool
+    celebration_enabled: bool
+
+
+class AdminMissedSweepResponse(BaseModel):
+    """The result of ``POST /api/v1/admin/missed-sweep``.
+
+    ``fired`` is the number of ``nestquest_quest_missed`` transitions
+    the run built AND this route published on the SSE stream — zero
+    when the sweep's watermark already covers today (a same-day rerun).
+    """
+
+    fired: int
+
+
 #: 404 detail for a child id the core layer reports as non-existent.
 _CHILD_NOT_FOUND_DETAIL = "Child not found"
 
@@ -736,6 +818,33 @@ def _raise_history_error(error: ValueError) -> NoReturn:
     raise HTTPException(status_code=422, detail=str(error)) from error
 
 
+def _raise_settings_error(error: ValueError) -> NoReturn:
+    """Map a core settings-store ``ValueError`` onto 422 and raise it.
+
+    The settings routes have no path ids either, so they have no 404
+    case: every ``ValueError`` that survives to here is a rejected
+    change — an unknown field or an invalid value, each raised by the
+    core naming the offending field BEFORE any write, so the stored
+    settings are never half-changed — and 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 _raise_sweep_error(error: ValueError) -> NoReturn:
+    """Map a core missed-sweep ``ValueError`` onto 422 and raise it.
+
+    The sweep route has no path ids, so it has no 404 case: a core
+    ``ValueError`` — a rejected date the sweep's query refuses before
+    any read, unreachable from this handler because ``today`` comes
+    from the route's own clock read — 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 _local_now() -> datetime.datetime:
     """Return the API host's local time as ONE timezone-aware clock read.
 
@@ -1401,6 +1510,29 @@ def _history_row_response(row: core_history.HistoryRow) -> AdminHistoryRow:
     )
 
 
+def _settings_response(
+    settings: core_settings.NestQuestSettings,
+) -> AdminSettingsResponse:
+    """Serialize one core NestQuestSettings into the documented payload.
+
+    Explicit field-by-field (not ``dataclasses.asdict``) so the payload
+    stays pinned to the ten documented fields even if the core
+    dataclass later grows another one.
+    """
+    return AdminSettingsResponse(
+        horizon_days=settings.horizon_days,
+        day_rollover_time=settings.day_rollover_time,
+        notify_target=settings.notify_target,
+        morning_summary_time=settings.morning_summary_time,
+        afternoon_reminder_time=settings.afternoon_reminder_time,
+        end_of_day_report_time=settings.end_of_day_report_time,
+        morning_summary_enabled=settings.morning_summary_enabled,
+        afternoon_reminder_enabled=settings.afternoon_reminder_enabled,
+        end_of_day_report_enabled=settings.end_of_day_report_enabled,
+        celebration_enabled=settings.celebration_enabled,
+    )
+
+
 @router.get(
     "/history",
     summary="Query the completion history for one filter and range",
@@ -1489,4 +1621,112 @@ async def admin_history_csv(
     )
 
 
+@router.get(
+    "/settings",
+    summary="Read the effective settings",
+    response_model=AdminSettingsResponse,
+)
+async def admin_get_settings(request: Request) -> AdminSettingsResponse:
+    """Return the current effective settings (all ten documented fields).
+
+    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.settings_store.load_settings` call — the API
+    owns settings in the database (the integration's HA options remain
+    a temporary separate source until Feature 18 wires the client to
+    the API), and the stored-document decode, the per-field validation
+    and the defaults-for-absent-fields fallback all live in the core.
+    A corrupt stored document never fails the read: the core falls
+    back to the defaults (per field where possible).
+    """
+    state: DatabaseState = request.app.state.db
+    settings = await core_settings_store.load_settings(state.database)
+    return _settings_response(settings)
+
+
+@router.patch(
+    "/settings",
+    summary="Update the supplied settings fields",
+    response_model=AdminSettingsResponse,
+)
+async def admin_update_settings(
+    changes: dict[str, object], request: Request
+) -> AdminSettingsResponse:
+    """Update the supplied settings fields and return the new state.
+
+    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 body — a
+    JSON object of settings field names to new values — goes STRAIGHT
+    to :func:`nestquest_core.settings_store.update_settings`, which
+    rejects an unknown field, resolves each supplied field through the
+    core settings validators (a JSON ``null`` resets its field to the
+    default) and merges onto the current effective settings only after
+    every value has validated — so a rejected update never changes the
+    stored settings.  Every core ``ValueError`` is 422
+    (:func:`_raise_settings_error`); anything else re-raises.
+    """
+    state: DatabaseState = request.app.state.db
+    try:
+        settings = await core_settings_store.update_settings(
+            state.database, changes
+        )
+    except ValueError as error:
+        _raise_settings_error(error)
+    return _settings_response(settings)
+
+
+@router.post(
+    "/missed-sweep",
+    summary="Run the nightly missed-quest sweep now",
+    response_model=AdminMissedSweepResponse,
+)
+async def admin_run_missed_sweep(
+    request: Request,
+) -> AdminMissedSweepResponse:
+    """Run the missed-quest sweep and publish what it built.
+
+    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 over the SAME
+    HA-free sweep policy the integration's rollover listener runs
+    (:mod:`nestquest_core.sweep`): ONE timezone-aware clock read
+    (:func:`_local_now`) pins the API host's local ``today``, and ONE
+    :func:`nestquest_core.sweep.run_missed_sweep` call does everything
+    else — the past-due open-instance query, the documented payload
+    build and the watermark update (a same-day rerun is an empty
+    no-op, and a run after downtime sweeps the accumulated window
+    once).  The route is read-only over the domain tables: it never
+    mutates an instance and never writes a completion event.
+
+    Each returned ``(event_type, payload)`` pair is published DIRECTLY
+    on the app's ONE transition publisher (``request.app.state.publisher``,
+    fan-out fire-and-forget, see api/events.py) — the exact payload the
+    sweep built, never re-fetched and never hand-built here, the same
+    publish-what-the-core-built shape api/scheduler.py uses.  This is a
+    correctness rule, not a style preference: the core commits the
+    idempotency watermark BEFORE returning, so a re-fetch-based publish
+    loop opens a loss window — a concurrent regenerate deleting the
+    instance, or a transient database error, would raise uncaught (a
+    bare 500) AFTER the watermark advanced, and no future sweep would
+    ever re-examine that instance.  Publishing the built payload leaves
+    no such window, and every event of one run keeps the ONE shared
+    ``occurred_at`` stamp the sweep computed for it.  The response
+    reports ``fired`` — the number of transitions published.  Error
+    mapping (:func:`_raise_sweep_error`): every core ``ValueError`` is
+    422; anything else re-raises.
+    """
+    state: DatabaseState = request.app.state.db
+    database = state.database
+    today = _local_now().date()
+    try:
+        events = await core_sweep.run_missed_sweep(database, today=today)
+    except ValueError as error:
+        _raise_sweep_error(error)
+    for event_type, payload in events:
+        request.app.state.publisher.publish(event_type, payload)
+    return AdminMissedSweepResponse(fired=len(events))
+
+
 __all__ = ["router"]

api/scheduler.py

@@ -0,0 +1,173 @@
+"""Daily missed-sweep background scheduler for the API service.
+
+The API service has no Home Assistant, so nobody re-fires the
+integration's day-rollover listener for it: this module owns the API
+plane's own background job.  One :class:`MissedSweepScheduler` per app
+(installed on ``app.state`` by the lifespan) sleeps until the
+configured ``day_rollover_time`` — read fresh from the settings store
+at the top of EVERY cycle, so an admin's settings change applies on
+the next scheduling decision without a restart — then runs the
+HA-free core's :func:`nestquest_core.sweep.run_missed_sweep` for the
+API host's local date and publishes each returned
+``nestquest_quest_missed`` event on the app's ONE transition publisher
+(the same SSE stream every route publishes through).
+
+All sweep policy lives in the core (the watermark, the
+``[watermark, today)`` window, the no-completion-event rule): this
+scheduler only decides WHEN.  A same-night rerun — e.g. two workers,
+or a settings change moving the rollover past itself — fires nothing:
+the core's watermark makes the run an empty no-op.
+
+Testability: ``clock`` (an aware-now callable) and ``sleep`` (an
+awaitable delay) are injectable, so tests pin the time, fast-forward
+through the wait, and observe the computed delays without waiting a
+real day.  Cancellation: :meth:`stop` cancels the loop task and waits
+for it, so the lifespan's shutdown is clean and a mid-run sweep is
+interrupted before the connection closes.  A failed iteration (a
+database hiccup, a rejected setting) is logged and retried after a
+fixed backoff rather than silently killing the job.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import logging
+
+from api.nestquest_core import core_settings_store, core_sweep
+
+LOGGER = logging.getLogger(__name__)
+
+#: Backoff after a failed scheduling cycle (settings read or sweep
+#: error), so a persistent failure retries every minute instead of
+#: spinning or dying.
+_RETRY_BACKOFF_SECONDS = 60.0
+
+
+def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float:
+    """Seconds from ``now`` until the next occurrence of ``target``.
+
+    ``target`` is a plain ``HH:MM`` time of day: the next wall-clock
+    occurrence strictly AFTER ``now`` (an equal time schedules for
+    tomorrow, so the just-run sweep is never instantly re-run).
+    """
+    scheduled = now.replace(
+        hour=target.hour, minute=target.minute, second=0, microsecond=0
+    )
+    if scheduled <= now:
+        scheduled += datetime.timedelta(days=1)
+    return (scheduled - now).total_seconds()
+
+
+class MissedSweepScheduler:
+    """Run the missed sweep once per day at ``day_rollover_time``."""
+
+    def __init__(
+        self,
+        database: object,
+        *,
+        publisher: object | None = None,
+        clock: object | None = None,
+        sleep: object | None = None,
+    ) -> None:
+        """Create the scheduler (not yet running).
+
+        ``database`` is the app's live :class:`~api.database.NestQuestDatabase`.
+        ``publisher`` is the app's ONE
+        :class:`~api.events.TransitionPublisher` (publishing is skipped
+        when ``None``).  ``clock`` defaults to the API host's local
+        aware now (``datetime.datetime.now().astimezone()``) — the same
+        single-clock discipline the routes' ``_local_now`` uses;
+        ``sleep`` defaults to :func:`asyncio.sleep`.
+        """
+        self._database = database
+        self._publisher = publisher
+        self._clock = clock if clock is not None else (
+            lambda: datetime.datetime.now().astimezone()
+        )
+        self._sleep = sleep if sleep is not None else asyncio.sleep
+        self._task: asyncio.Task | None = None
+
+    def start(self) -> None:
+        """Start the background loop (idempotent: a running one stays)."""
+        if self._task is None or self._task.done():
+            self._task = asyncio.create_task(
+                self._run_forever(), name="nestquest-missed-sweep"
+            )
+
+    async def stop(self) -> None:
+        """Cancel the loop and wait for it to finish (safe if never started)."""
+        if self._task is not None:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+
+    @property
+    def running(self) -> bool:
+        """Whether the background loop task is alive."""
+        return self._task is not None and not self._task.done()
+
+    async def _run_forever(self) -> None:
+        """Schedule → sweep → repeat, forever (until cancelled)."""
+        while True:
+            try:
+                settings = await core_settings_store.load_settings(
+                    self._database
+                )
+                rollover = datetime.time.fromisoformat(
+                    settings.day_rollover_time
+                )
+            except Exception:
+                # Not silent: the failure is logged and retried after
+                # the backoff, so one bad settings document or a
+                # transient database error never kills the scheduler.
+                LOGGER.exception(
+                    "NestQuest missed-sweep scheduler failed to resolve "
+                    "the day rollover time; retrying in %gs",
+                    _RETRY_BACKOFF_SECONDS,
+                )
+                await self._sleep(_RETRY_BACKOFF_SECONDS)
+                continue
+            delay = _delay_seconds(rollover, self._clock())
+            LOGGER.info(
+                "NestQuest missed sweep scheduled for %s (%.0fs from now)",
+                settings.day_rollover_time,
+                delay,
+            )
+            await self._sleep(delay)
+            try:
+                await self._run_once(self._clock().date())
+            except Exception:
+                # Not silent either: the sweep error is logged and the
+                # loop backs off before the next cycle, whose watermark
+                # re-read makes an already-swept day a no-op.
+                LOGGER.exception(
+                    "NestQuest scheduled missed sweep failed; retrying in "
+                    "%gs",
+                    _RETRY_BACKOFF_SECONDS,
+                )
+                await self._sleep(_RETRY_BACKOFF_SECONDS)
+
+    async def _run_once(self, today: datetime.date) -> None:
+        """Run ONE sweep for ``today`` and publish what it built.
+
+        A failure propagates to :meth:`_run_forever`, which logs it and
+        backs off before the next cycle; the watermark makes a re-run
+        of an already-swept day announce nothing twice.
+        """
+        events = await core_sweep.run_missed_sweep(
+            self._database, today=today
+        )
+        for event_type, payload in events:
+            if self._publisher is not None:
+                self._publisher.publish(event_type, payload)
+        LOGGER.info(
+            "NestQuest scheduled missed sweep for %s published %d event(s)",
+            today.isoformat(),
+            len(events),
+        )
+
+
+__all__ = ["MissedSweepScheduler"]

custom_components/nestquest/core/settings_store.py

@@ -0,0 +1,210 @@
+"""Durable settings store for the NestQuest API service.
+
+The API service has NO Home Assistant config entry, so it cannot build
+:class:`~.settings.NestQuestSettings` from ``entry.options`` the way the
+integration does.  This module gives the API its own durable settings
+source: a JSON document kept in the EXISTING ``nestquest_meta_state``
+key/value table (:class:`~.dao_meta.MetaStateDao`) under the well-known
+key :data:`SETTINGS_META_KEY` — no schema change, no new table —
+validated through the SAME field table the settings object is built
+from (:data:`~.settings._FIELDS`, the table driving
+``from_options``/``from_options_resilient``).  No validation rule is
+re-implemented here.
+
+OWNERSHIP: the API now owns settings in the database.  The
+integration's Home-Assistant config-entry options remain a temporary,
+SEPARATE source until Feature 18 converts the integration into a thin
+API client; the two are deliberately NOT unified here, and neither
+side reads the other's values.
+
+STORED SERIALIZATION — the document at ``SETTINGS_META_KEY`` is a JSON
+object holding exactly the TEN settings fields, keyed by field name
+(which equals the ``CONF_*`` option key for every field), complete
+after the first update:
+
+- ``horizon_days`` — int >= 1,
+- ``day_rollover_time``, ``morning_summary_time``,
+  ``afternoon_reminder_time``, ``end_of_day_report_time`` — strict
+  24-hour ``HH:MM`` strings,
+- ``notify_target`` — a string (a padded value is stripped by the
+  core's normalizer),
+- ``morning_summary_enabled``, ``afternoon_reminder_enabled``,
+  ``end_of_day_report_enabled``, ``celebration_enabled`` — real
+  booleans.
+
+Only values whose core validators ACCEPTED are ever written, so a
+stored document always round-trips through the core's option builders.
+An absent key (a fresh database) resolves to the all-defaults
+settings, and a corrupt document degrades to defaults or per-field
+defaults rather than crashing the service (see :func:`load_settings`).
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from .const import LOGGER
+from .dao_meta import MetaStateDao
+from .db import NestQuestDatabase
+from .settings import NestQuestSettings, _FIELDS
+
+#: The well-known ``nestquest_meta_state`` key holding the settings
+#: document.  One key, one JSON object — the same table the missed
+#: sweep's watermark already uses, so no migration is needed.
+SETTINGS_META_KEY = "settings"
+
+#: The per-field resolution table keyed by field name: each row is the
+#: ``(default, validator)`` pair the settings module's ``_FIELDS`` table
+#: carries for that field.  ``update_settings`` resolves each PROVIDED
+#: field through its row's validator — the SAME core validators
+#: ``from_options`` applies — and never re-implements one here.
+#: (Field names equal the ``CONF_*`` option keys, so one dict serves
+#: both the API's field names and the stored document's keys.)
+_RESOLUTIONS: dict[str, tuple[Any, Any]] = {
+    key: (default, validator) for _field, key, default, validator in _FIELDS
+}
+
+#: The known field names, in the settings module's field order — the
+#: allowed keys of an update body and the stored document's key set.
+_FIELD_NAMES: tuple[str, ...] = tuple(
+    key for _field, key, _default, _validator in _FIELDS
+)
+
+#: One asyncio.Lock per (connection wrapper, running loop), the missed
+#: sweep's pattern: the whole stored-document read → merge → write span
+#: is one critical section, so two concurrent updates queue instead of
+#: one losing the other's change — the loser re-reads the document
+#: INSIDE the lock and merges onto the winner's settings.
+_SETTINGS_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
+
+
+def _settings_lock(database: NestQuestDatabase) -> asyncio.Lock:
+    """Return the settings lock bound to ``database``'s running loop."""
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _SETTINGS_LOCKS.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _SETTINGS_LOCKS[key] = lock
+    return lock
+
+
+async def _load_document(database: NestQuestDatabase) -> dict[str, Any]:
+    """Read and decode the stored settings document (empty when unusable).
+
+    An absent key — a fresh database that never updated settings — and
+    an unparseable value (invalid JSON, or JSON that is not an object)
+    return an empty mapping, which the option builders resolve to the
+    all-defaults settings; the corrupt cases log a warning first.  A
+    parseable object is returned as stored (its per-field validity is
+    the builder's concern, not this reader's).
+    """
+    raw = await MetaStateDao(database).get(SETTINGS_META_KEY)
+    if raw is None:
+        return {}
+    try:
+        document = json.loads(raw)
+    except ValueError:
+        LOGGER.warning(
+            "NestQuest stored settings (meta_state key %r) are not valid "
+            "JSON; falling back to the default settings",
+            SETTINGS_META_KEY,
+        )
+        return {}
+    if not isinstance(document, dict):
+        LOGGER.warning(
+            "NestQuest stored settings (meta_state key %r) are not a JSON "
+            "object; falling back to the default settings",
+            SETTINGS_META_KEY,
+        )
+        return {}
+    return document
+
+
+def _named(field: str, error: ValueError) -> str:
+    """Return a field-named message for a rejected validator value.
+
+    The horizon, time and notify-target validators already name their
+    field in the message; the boolean validator's message is generic
+    (``expected a boolean, got ...``), so it gets the field prefixed.
+    Every rejected update's error therefore names its field either way.
+    """
+    message = str(error)
+    if message.startswith(f"{field} "):
+        return message
+    return f"{field}: {message}"
+
+
+async def load_settings(database: NestQuestDatabase) -> NestQuestSettings:
+    """Return the current effective settings stored for ``database``.
+
+    Reads :data:`SETTINGS_META_KEY` and resolves the stored document
+    through the core settings validators, with each absent field
+    falling back to its default.  An absent key and an unparseable
+    document fall back to the all-defaults settings (via
+    :func:`_load_document`), and a parseable document with a single
+    invalid FIELD falls back PER FIELD — through
+    :meth:`NestQuestSettings.from_options_resilient`, the core's own
+    resilient builder — so a corrupt sibling never resets the stored
+    values around it.  This never raises: the service must keep
+    serving reads even over a corrupt store.
+    """
+    document = await _load_document(database)
+    return NestQuestSettings.from_options_resilient(document)
+
+
+async def update_settings(
+    database: NestQuestDatabase, changes: Mapping[str, Any]
+) -> NestQuestSettings:
+    """Validate ``changes`` and persist them as the effective settings.
+
+    ``changes`` maps settings field names to new values.  EVERY name is
+    checked against the known field table BEFORE anything is validated
+    or written: an unknown name raises a ``ValueError`` naming it (and
+    the known names), leaving the stored settings untouched.  Each
+    PROVIDED field is then resolved from_options-style — its
+    :data:`~.settings._FIELDS` validator applied, with a JSON ``null``
+    mapping to the field's default (the settings module's uniform
+    None-as-absent rule) — and an invalid value raises a ``ValueError``
+    naming the field, also BEFORE anything is written, so a rejected
+    update can never leave a half-applied document behind.
+
+    The validated changes are merged onto the current effective
+    settings and the merged (complete, ten-field) document is persisted
+    in one atomic upsert; the updated settings are returned.  The whole
+    read → merge → write span runs inside the per-database settings
+    lock, so concurrent updates queue instead of interleaving.
+    """
+    if not isinstance(changes, Mapping):
+        raise ValueError(
+            "settings changes must be a JSON object mapping field names "
+            "to values"
+        )
+    unknown = sorted(
+        str(name) for name in changes if name not in _RESOLUTIONS
+    )
+    if unknown:
+        listed = ", ".join(repr(name) for name in unknown)
+        known = ", ".join(_FIELD_NAMES)
+        raise ValueError(
+            f"unknown settings field(s): {listed}; known fields: {known}"
+        )
+    async with _settings_lock(database):
+        current = await load_settings(database)
+        document: dict[str, Any] = {
+            field: getattr(current, field) for field in _FIELD_NAMES
+        }
+        for name, raw_value in changes.items():
+            default, validator = _RESOLUTIONS[name]
+            try:
+                document[name] = validator(
+                    default if raw_value is None else raw_value
+                )
+            except ValueError as error:
+                raise ValueError(_named(name, error)) from error
+        await MetaStateDao(database).set(
+            SETTINGS_META_KEY, json.dumps(document, sort_keys=True)
+        )
+    return NestQuestSettings(**document)

custom_components/nestquest/core/sweep.py

@@ -0,0 +1,143 @@
+"""The nightly missed-quest sweep, Home-Assistant-free (Feature 11).
+
+This is the HA-free heart of the missed sweep, shared by the two
+planes: the Home Assistant integration's :mod:`~..sweep` shim (which
+fires the returned events on ``hass.bus``) and the API service (which
+publishes them on the SSE stream and can also trigger the run on
+demand).  It owns the ONE sweep policy:
+
+At the configured day-rollover time (or on an explicit trigger) this
+job finds every instance whose ``due_date`` is before the
+household-local date, that has NO completion event, and that falls
+after the persisted sweep watermark — and RETURNS one
+``nestquest_quest_missed`` event per instance with the documented
+payload.  Like every core module, it never touches the Home Assistant
+event bus and never mutates an instance: ``missed`` is derived state,
+never an event_type (Feature 08's contract), so the sweep is read-only
+over the domain tables.
+
+Idempotency is the watermark: after a run, ``nestquest_meta_state``
+records the household-local date the sweep ran, and the next run only
+sweeps instances due in ``[watermark, today)``.  A second run the same
+night therefore fires nothing new, and a run after days of downtime
+sweeps exactly the accumulated window once — no per-instance journal,
+no re-announcing old misses.  An instance swept as missed that is
+later completed simply gains its completion event; history keeps both
+rows (append-only), and the sweep's later runs skip it via the
+no-completion-event rule.
+
+The whole watermark read → query → build → write span runs inside the
+per-database sweep lock (one lock per connection wrapper and running
+loop, the settings store's pattern): an overlapping startup sweep and
+a rollover sweep queue instead of interleaving, and the loser re-reads
+the watermark INSIDE the lock — sees the winner's date — and announces
+nothing twice.
+
+The caller supplies ``today`` as a plain :class:`datetime.date`: the
+household-local calendar date, read from ``hass.config.time_zone`` by
+the integration or from the API host's local clock by the API.  This
+module never reads a timezone database or config of its own.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import logging
+
+from .const import EVENT_QUEST_MISSED
+from .dao_meta import MetaStateDao
+from .dao_instances import QuestInstancesDao
+from .db import NestQuestDatabase
+# Intentional intra-package reuse of the shared payload builder: the
+# missed event's payload is the documented §3 instance payload, and
+# building it through the ONE builder keeps this sweep's payloads from
+# drifting from the completion/uncompleted builders' shape.
+from .events import _instance_payload
+
+LOGGER = logging.getLogger(__name__)
+
+#: meta_state key holding the household-local date the sweep last ran.
+SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
+
+#: One asyncio.Lock per (connection wrapper, running loop), mirroring
+#: the settings store's pattern: the whole watermark read → query →
+#: build → write span is one critical section, so an overlapping
+#: startup sweep and rollover sweep queue instead of double-announcing.
+#: The loser re-reads the watermark INSIDE the lock, sees the winner's
+#: date, and fires nothing.
+_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
+
+
+def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
+    """Return the sweep lock bound to ``database``'s running loop."""
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _SWEEP_LOCKS.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _SWEEP_LOCKS[key] = lock
+    return lock
+
+
+#: Strict UTC ISO-8601 timestamp (the DAO layers' policy): one shape,
+#: explicit offset, second precision.
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+async def run_missed_sweep(
+    database: NestQuestDatabase,
+    *,
+    today: datetime.date,
+) -> list[tuple[str, dict]]:
+    """Build one missed event per unswept past-due open instance.
+
+    ``today`` is the household-local calendar date the sweep runs for
+    (the caller owns the clock: the integration reads it from
+    ``hass.config.time_zone``, the API from its host's local time).
+    Returns the ``(event_type, payload)`` tuples to announce, EMPTY
+    when the watermark already covers ``today`` (a same-night rerun) —
+    the caller decides how to publish them and announces nothing on
+    its own.
+
+    The whole watermark read → query → build → write span runs inside
+    the per-database sweep lock: an overlapping startup sweep and a
+    rollover sweep queue instead of interleaving, and the loser
+    re-reads the watermark inside the lock — sees the winner's date —
+    and announces nothing twice.  Every event in one run shares one
+    ``occurred_at`` stamp (one run, one announcement moment), and the
+    payload comes from the shared
+    :func:`~.events._instance_payload` builder so the shape can never
+    drift from the documented §3 contract.
+    """
+    today_iso = today.isoformat()
+    dao = MetaStateDao(database)
+    async with _sweep_lock(database):
+        watermark = await dao.get(SWEEP_WATERMARK_KEY)
+        if watermark == today_iso:
+            LOGGER.debug(
+                "NestQuest missed sweep already ran for %s", today_iso
+            )
+            return []
+
+        instances_dao = QuestInstancesDao(database)
+        rows = await instances_dao.list_missed_unswept(today_iso, watermark)
+        now_stamp = datetime.datetime.now(datetime.timezone.utc).strftime(
+            _UTC_TIMESTAMP_FORMAT
+        )
+        events: list[tuple[str, dict]] = []
+        for instance in rows:
+            events.append(
+                (
+                    EVENT_QUEST_MISSED,
+                    await _instance_payload(
+                        database, instance.id, now_stamp
+                    ),
+                )
+            )
+        await dao.set(SWEEP_WATERMARK_KEY, today_iso)
+    LOGGER.info(
+        "NestQuest missed sweep for %s built %d event(s)",
+        today_iso,
+        len(events),
+    )
+    return events

custom_components/nestquest/sweep.py

@@ -1,70 +1,40 @@
-"""The nightly missed-quest sweep (Feature 11).
+"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim.
 
-At the configured day-rollover time (the same HA-local listener that
-re-materializes the rolling horizon) this job finds every instance
-whose ``due_date`` is before the HA-local date, that has NO completion
-event, and that falls after the persisted sweep watermark — and fires
-ONE ``nestquest_quest_missed`` event per instance with the documented
-payload.  The sweep is read-only over the domain tables: it never
-mutates an instance and never writes a completion event (``missed``
-is derived state, never an event_type — Feature 08's contract).
+The sweep's policy is the HA-free core's :mod:`~.core.sweep`: the
+watermark read → ``list_missed_unswept`` query → payload build →
+watermark span, all inside the per-database sweep lock, shared with
+the API plane.  This module is the integration's thin adapter over it:
+at the configured day-rollover time (the same HA-local listener that
+re-materializes the rolling horizon) and once at setup it calls the
+core's :func:`~.core.sweep.run_missed_sweep` and fires each returned
+event on ``hass.bus`` — ONE ``nestquest_quest_missed`` per past-due
+open instance with the documented payload.  The core never touches
+Home Assistant; the firing lives here.
 
-Idempotency is the watermark: after a run, ``nestquest_meta_state``
-records the HA-local date the sweep ran, and the next run only
-sweeps instances due in ``[watermark, today)``.  A second run the
-same night therefore fires nothing new, and a run after days of
-downtime sweeps exactly the accumulated window once — no
-per-instance journal, no re-announcing old misses.  An instance swept
-as missed that is later completed simply gains its completion event;
-history keeps both rows (append-only), and the sweep's later runs
-skip it via the no-completion-event rule.
+Idempotency is the core's watermark: after a run, ``nestquest_meta_state``
+records the HA-local date the sweep ran, and the next run only sweeps
+instances due in ``[watermark, today)`` — a second run the same night
+fires nothing new, and a run after days of downtime sweeps exactly the
+accumulated window once.  An instance swept as missed that is later
+completed simply gains its completion event; the sweep is read-only
+over the domain tables (``missed`` is derived state, never an
+event_type — Feature 08's contract).
 """
 from __future__ import annotations
 
-import asyncio
 import datetime
 import logging
 from zoneinfo import ZoneInfo
 
-from .const import EVENT_QUEST_MISSED
-from .dao_children import ChildrenDao
-from .dao_instances import QuestInstancesDao
-from .dao_meta import MetaStateDao
-from .dao_rules import QuestDefinitionsDao
+from .core.sweep import (
+    SWEEP_WATERMARK_KEY,
+    run_missed_sweep as _run_core_missed_sweep,
+)
 from .db import NestQuestDatabase
 
 LOGGER = logging.getLogger(__name__)
 
-#: meta_state key holding the HA-local date the sweep last ran.
-SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
-
-#: One asyncio.Lock per (connection wrapper, running loop), mirroring
-#: the migration runner's pattern: the whole watermark read → query →
-#: fire → write span is one critical section, so an overlapping startup
-#: sweep and rollover sweep queue instead of double-announcing.  The
-#: loser re-reads the watermark INSIDE the lock, sees the winner's
-#: date, and fires nothing.
-_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
-
-
-def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
-    """Return the sweep lock bound to ``database``'s running loop."""
-    loop = asyncio.get_running_loop()
-    key = (id(database), id(loop))
-    lock = _SWEEP_LOCKS.get(key)
-    if lock is None:
-        lock = asyncio.Lock()
-        _SWEEP_LOCKS[key] = lock
-    return lock
-
-#: UTC ISO-8601 timestamp policy shared with the DAO layers.
-_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
-
-
-def _now_stamp() -> str:
-    return datetime.datetime.now(datetime.timezone.utc).strftime(
-        _UTC_TIMESTAMP_FORMAT
-    )
+__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"]
 
 
 async def run_missed_sweep(
@@ -79,54 +49,24 @@ async def run_missed_sweep(
     materialization and completion layers); when omitted it is read
     from ``hass.config.time_zone``.  Returns the number of events
     fired.  Fires nothing and returns 0 when the watermark already
-    covers ``today`` (a same-night rerun).
-
-    The whole watermark read → query → fire → write span runs inside
-    the per-database sweep lock: an overlapping startup sweep and a
-    rollover sweep queue instead of interleaving, and the loser
-    re-reads the watermark inside the lock — sees the winner's date —
-    and announces nothing twice.
+    covers ``today`` (a same-night rerun) — the watermark read, the
+    query, the payload build and the watermark write all live in the
+    core's locked
+    :func:`~.core.sweep.run_missed_sweep`; this shim only fires each
+    returned event on the bus and counts them.
     """
     today_date = (
         today
         if today is not None
         else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
     )
-    today_iso = today_date.isoformat()
-    dao = MetaStateDao(database)
-    async with _sweep_lock(database):
-        watermark = await dao.get(SWEEP_WATERMARK_KEY)
-        if watermark == today_iso:
-            LOGGER.debug(
-                "NestQuest missed sweep already ran for %s", today_iso
-            )
-            return 0
-
-        instances_dao = QuestInstancesDao(database)
-        rows = await instances_dao.list_missed_unswept(today_iso, watermark)
-        children_dao = ChildrenDao(database)
-        definitions_dao = QuestDefinitionsDao(database)
-        now_stamp = _now_stamp()
-        fired = 0
-        for instance in rows:
-            child = await children_dao.get(instance.child_id)
-            definition = await definitions_dao.get(instance.definition_id)
-            hass.bus.async_fire(
-                EVENT_QUEST_MISSED,
-                {
-                    "child_id": instance.child_id,
-                    "child_name": child.display_name if child else None,
-                    "instance_id": instance.id,
-                    "quest_title": definition.title if definition else None,
-                    "window": instance.window,
-                    "due_date": instance.due_date,
-                    "due_time": instance.due_time,
-                    "occurred_at": now_stamp,
-                },
-            )
-            fired += 1
-        await dao.set(SWEEP_WATERMARK_KEY, today_iso)
+    events = await _run_core_missed_sweep(database, today=today_date)
+    for event_type, payload in events:
+        hass.bus.async_fire(event_type, payload)
+    fired = len(events)
     LOGGER.info(
-        "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
+        "NestQuest missed sweep for %s fired %d event(s)",
+        today_date.isoformat(),
+        fired,
     )
     return fired

tests/test_api_admin_settings.py

@@ -0,0 +1,398 @@
+"""Admin-plane settings route tests (task 2b3de7e5).
+
+These tests exercise the admin plane's settings read and update routes
+against the ASGI app with httpx, reusing the shared local-RSA-key/
+stubbed-JWKS harness (:mod:`tests.admin_jwt_harness`) so an admin JWT
+genuinely authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.  The stored settings
+document is read back through the SAME core copy the app uses
+(``nestquest_core.*``; see the two-copy caveat in api/nestquest_core.py)
+via its :class:`nestquest_core.dao_meta.MetaStateDao`, so persistence is
+proven against the DATABASE the routes wrote, not a parallel copy.
+
+Proven per the done-condition, everything read back through the API:
+
+- ``GET /api/v1/admin/settings`` returns the current effective settings
+  — the ten documented fields at their defaults on a fresh database.
+- ``PATCH /api/v1/admin/settings`` PERSISTS the supplied fields and the
+  NEXT read reflects them; a later partial update merges onto the
+  current effective settings (omitted fields keep their stored value),
+  and the document is stored under the well-known ``nestquest_meta_state``
+  key.
+- An invalid value (a non-positive or boolean horizon, a non-strict
+  time, a non-bool toggle, a non-string target) is rejected with 422
+  naming the field, and the stored settings are UNCHANGED — the next
+  read still shows the previous value.  An unknown field is rejected
+  the same way, also when mixed with a valid field (validate
+  everything before writing).
+- A corrupt stored document (invalid JSON, a non-object, an invalid
+  field value) does not crash the service: the read falls back to the
+  defaults, per field where possible.
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: a non-admin JWT is 403 and an absent credential is 401
+  on both routes — and no refused patch writes anything.
+"""
+from __future__ import annotations
+
+import importlib
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The meta-state DAO through the SAME core copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy) —
+# used for the tests' database assertions only; the routes reach it
+# through api.nestquest_core.
+_dao_meta = importlib.import_module("nestquest_core.dao_meta")
+
+#: The well-known ``nestquest_meta_state`` key the API stores its
+#: settings under — asserted literally so the contract is pinned here,
+#: independent of the core constant.
+SETTINGS_META_KEY = "settings"
+
+#: The all-defaults effective settings a fresh database reports — the
+#: documented ten fields, asserted literally so the payload contract is
+#: pinned here, independent of the core constants.
+DEFAULT_SETTINGS = {
+    "horizon_days": 14,
+    "day_rollover_time": "00:00",
+    "notify_target": "",
+    "morning_summary_time": "08:00",
+    "afternoon_reminder_time": "15:00",
+    "end_of_day_report_time": "20:00",
+    "morning_summary_enabled": True,
+    "afternoon_reminder_enabled": True,
+    "end_of_day_report_enabled": True,
+    "celebration_enabled": True,
+}
+
+
+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 settings_client(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the children tests use (stubbed JWKS, no network);
+    the fixture also exposes the app's ONE database connection so tests
+    can read (or corrupt) the stored settings document itself.
+    """
+    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,
+        )
+
+
+# --- read: the effective settings, defaults on a fresh database --------
+
+
+async def test_get_settings_returns_defaults_on_a_fresh_database(
+    settings_client: SimpleNamespace,
+) -> None:
+    """GET returns the ten documented fields at their defaults."""
+    response = await settings_client.client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+
+# --- update: persists, and the NEXT read reflects it --------------------
+
+
+async def test_update_persists_and_the_next_read_reflects_it(
+    settings_client: SimpleNamespace,
+) -> None:
+    """PATCH persists the supplied fields; GET reads them back."""
+    client = settings_client.client
+    changes = {
+        "horizon_days": 21,
+        "day_rollover_time": "03:30",
+        "notify_target": "notify.mobile_app_dads_phone",
+        "celebration_enabled": False,
+    }
+    patched = await client.patch(
+        "/api/v1/admin/settings", json=changes, headers=_admin_headers()
+    )
+    assert patched.status_code == 200
+    expected = {**DEFAULT_SETTINGS, **changes}
+    assert patched.json() == expected
+
+    # The NEXT read reflects the persisted settings...
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.status_code == 200
+    assert reread.json() == expected
+
+    # ...and the DATABASE holds the merged document: the complete ten
+    # fields, keyed by field name, under the well-known meta_state key.
+    raw = await _dao_meta.MetaStateDao(settings_client.database).get(
+        SETTINGS_META_KEY
+    )
+    assert raw is not None
+    assert json.loads(raw) == expected
+
+
+async def test_partial_update_merges_onto_the_current_settings(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A later PATCH changes only its own fields; the rest survive."""
+    client = settings_client.client
+    first = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7, "notify_target": " notify.x "},
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    # The notify target went through the core's own normalizer: stripped.
+    assert first.json()["horizon_days"] == 7
+    assert first.json()["notify_target"] == "notify.x"
+
+    second = await client.patch(
+        "/api/v1/admin/settings",
+        json={"morning_summary_time": "07:45"},
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json() == {
+        **DEFAULT_SETTINGS,
+        "horizon_days": 7,
+        "notify_target": "notify.x",
+        "morning_summary_time": "07:45",
+    }
+
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.json() == second.json()
+
+
+async def test_empty_update_is_a_noop(
+    settings_client: SimpleNamespace,
+) -> None:
+    """PATCHing no fields leaves the effective settings as they are."""
+    client = settings_client.client
+    response = await client.patch(
+        "/api/v1/admin/settings", json={}, headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+
+async def test_null_resets_a_field_to_its_default(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A JSON ``null`` maps to the field's default (None-as-absent)."""
+    client = settings_client.client
+    set_horizon = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7},
+        headers=_admin_headers(),
+    )
+    assert set_horizon.status_code == 200
+    assert set_horizon.json()["horizon_days"] == 7
+
+    reset = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": None},
+        headers=_admin_headers(),
+    )
+    assert reset.status_code == 200
+    assert reset.json()["horizon_days"] == 14
+
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.json()["horizon_days"] == 14
+
+
+# --- validation: rejected values never change the stored settings -------
+
+
+@pytest.mark.parametrize(
+    ("changes", "field"),
+    [
+        ({"horizon_days": 0}, "horizon_days"),
+        ({"horizon_days": -3}, "horizon_days"),
+        # A bool is rejected — the core's guardrail survives the route.
+        ({"horizon_days": True}, "horizon_days"),
+        ({"horizon_days": "soon"}, "horizon_days"),
+        ({"day_rollover_time": "25:00"}, "day_rollover_time"),
+        ({"day_rollover_time": "7:30"}, "day_rollover_time"),
+        ({"morning_summary_enabled": "yes"}, "morning_summary_enabled"),
+        ({"celebration_enabled": 1}, "celebration_enabled"),
+        ({"notify_target": 5}, "notify_target"),
+    ],
+)
+async def test_invalid_value_is_rejected_and_stored_settings_unchanged(
+    settings_client: SimpleNamespace, changes: dict[str, object], field: str
+) -> None:
+    """A rejected value is 422 naming the field; the next read is same."""
+    client = settings_client.client
+    seeded = await client.patch(
+        "/api/v1/admin/settings",
+        json={
+            "horizon_days": 21,
+            "day_rollover_time": "01:30",
+            "morning_summary_enabled": False,
+        },
+        headers=_admin_headers(),
+    )
+    assert seeded.status_code == 200
+    before = seeded.json()
+
+    response = await client.patch(
+        "/api/v1/admin/settings", json=changes, headers=_admin_headers()
+    )
+    assert response.status_code == 422, changes
+    assert field in str(response.json()["detail"]), changes
+
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.status_code == 200
+    assert after.json() == before
+
+
+async def test_unknown_field_is_rejected_and_stored_settings_unchanged(
+    settings_client: SimpleNamespace,
+) -> None:
+    """An unknown field is 422 naming it — alone or beside a valid one."""
+    client = settings_client.client
+    unknown = await client.patch(
+        "/api/v1/admin/settings",
+        json={"bogus_setting": 1},
+        headers=_admin_headers(),
+    )
+    assert unknown.status_code == 422
+    assert "bogus_setting" in str(unknown.json()["detail"])
+
+    mixed = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7, "bogus_setting": 1},
+        headers=_admin_headers(),
+    )
+    assert mixed.status_code == 422
+    assert "bogus_setting" in str(mixed.json()["detail"])
+
+    # Neither rejection wrote anything, not even the valid sibling.
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS
+
+
+async def test_non_object_body_is_422(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A body that is not a JSON object is FastAPI's 422 (the shape)."""
+    client = settings_client.client
+    for body in ([1, 2], "nope", 5):
+        response = await client.patch(
+            "/api/v1/admin/settings", json=body, headers=_admin_headers()
+        )
+        assert response.status_code == 422, body
+    # A bare JSON null is likewise not an object.
+    response = await client.patch(
+        "/api/v1/admin/settings",
+        content=b"null",
+        headers={**_admin_headers(), "content-type": "application/json"},
+    )
+    assert response.status_code == 422
+
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS
+
+
+# --- resilience: a corrupt stored document does not crash the service ---
+
+
+async def test_corrupt_stored_document_falls_back_to_defaults(
+    settings_client: SimpleNamespace,
+) -> None:
+    """GET degrades gracefully over an unusable settings document."""
+    client = settings_client.client
+    dao = _dao_meta.MetaStateDao(settings_client.database)
+
+    # Not JSON at all: all defaults, no crash.
+    await dao.set(SETTINGS_META_KEY, "{not-json")
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+    # JSON but not an object: all defaults, no crash.
+    await dao.set(SETTINGS_META_KEY, json.dumps([1, 2]))
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+    # An object with one invalid field: PER-FIELD fallback — the valid
+    # sibling survives, the invalid one falls back to its default.
+    await dao.set(
+        SETTINGS_META_KEY,
+        json.dumps({"horizon_days": 7, "day_rollover_time": "soon"}),
+    )
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == {**DEFAULT_SETTINGS, "horizon_days": 7}
+
+
+# --- authentication: the ONE require_admin dependency guards both routes
+
+
+async def test_settings_routes_refuse_non_admin_and_absent_credentials(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401; none write."""
+    client = settings_client.client
+    non_admin = make_token(groups=("some-other-group",))
+    forbidden_get = await client.get(
+        "/api/v1/admin/settings",
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden_get.status_code == 403
+    assert forbidden_get.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    forbidden_patch = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7},
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden_patch.status_code == 403
+
+    absent_get = await client.get("/api/v1/admin/settings")
+    assert absent_get.status_code == 401
+    absent_patch = await client.patch(
+        "/api/v1/admin/settings", json={"horizon_days": 7}
+    )
+    assert absent_patch.status_code == 401
+
+    # No refused request changed anything: the settings are untouched.
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS

tests/test_api_missed_sweep.py

@@ -0,0 +1,671 @@
+"""Admin-plane missed-sweep trigger tests (task 058c7b69).
+
+These tests exercise ``POST /api/v1/admin/missed-sweep`` — the admin
+plane's trigger for the nightly missed-quest sweep — against a REAL
+server (uvicorn on an ephemeral port, the SSE tests' documented
+approach: httpx's ASGI transport buffers whole bodies, so a live
+stream needs a live server), plus the API service's daily scheduler:
+
+- The trigger requires the router's ONE ``require_admin`` check: a
+  non-admin JWT is 403 and an absent credential is 401, and neither
+  refusal runs the sweep.
+- Seeded past-due open instances: ONE ``nestquest_quest_missed``
+  transition per instance arrives on the LIVE SSE stream with the
+  documented payload fields (child_id, child_name, instance_id,
+  quest_title, window, due_date, due_time, occurred_at — §3), while
+  the seeded TODAY-due instance publishes nothing (the sweep's rule
+  is ``due_date`` strictly before ``today``).
+- A same-day rerun publishes nothing (the core watermark).
+- A run after downtime (``today`` advanced two days) publishes the
+  accumulated window once — the instance due on day one fires on the
+  catch-up run — and a further rerun is silent again.
+- Publishing is the api/scheduler.py shape — the sweep-BUILT
+  ``(event_type, payload)`` pairs go straight to the publisher, no
+  per-event re-fetch: all events of one run share the run's ONE
+  ``occurred_at`` stamp, and a deterministic concurrent-deletion case
+  (the sweep is wrapped to delete one swept instance after it built
+  its events and committed its watermark) still answers 200 and
+  streams every built event — a re-fetch there would 500 and lose the
+  missed event forever.
+- The daily scheduler (:class:`api.scheduler.MissedSweepScheduler`)
+  computes the delay to the configured ``day_rollover_time``, runs the
+  sweep when the (injectable) clock reaches it, publishes the built
+  events on its publisher, advances to the NEXT day's rollover, and
+  cancels cleanly.
+
+The household is seeded through the SAME core copy the app uses (the
+``nestquest_core.*`` package registered by ``api/nestquest_core.py``;
+see the two-copy caveat there) and the sweep's watermark is read back
+through the same copy's :class:`~nestquest_core.dao_meta.MetaStateDao`
+— on the SERVER's event loop for the uvicorn-hosted app (the database
+belongs to that thread's loop, so assertions schedule there), on the
+test's own loop for the scheduler test's standalone database.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import importlib
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest
+import uvicorn
+
+from api import routes_admin
+from api.app import create_app
+from api.auth import JwksCache
+from api.config import ApiConfig
+from api.database import _executor
+from api.nestquest_core import core_db as _core_db
+from api.nestquest_core import core_migrations as _core_migrations
+from api.scheduler import MissedSweepScheduler
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    AUDIENCE,
+    ISSUER,
+    JWKS_URL,
+    KID,
+    StubbedFetch,
+    jwks_for,
+    make_token,
+)
+from tests.test_api_panel import PANEL_TOKEN
+from tests.test_api_sse import EVENT_QUEST_MISSED, READ_TIMEOUT, _free_port, _stream
+
+# The bundled core submodules, through the SAME copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
+_children = importlib.import_module("nestquest_core.children")
+_quest_definitions = importlib.import_module(
+    "nestquest_core.quest_definitions"
+)
+_materialize = importlib.import_module("nestquest_core.materialize")
+_recurrence = importlib.import_module("nestquest_core.recurrence")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_dao_meta = importlib.import_module("nestquest_core.dao_meta")
+_settings_store = importlib.import_module("nestquest_core.settings_store")
+_core_sweep = importlib.import_module("nestquest_core.sweep")
+
+#: The well-known ``nestquest_meta_state`` key holding the sweep
+#: watermark — asserted literally so the contract is pinned here.
+SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+async def _seed_missed_household(
+    db_path: str, today: datetime.date
+) -> SimpleNamespace:
+    """Seed the missed-sweep household on ``db_path`` and return records.
+
+    Two children; two quests whose ONLY instances are due YESTERDAY
+    (created via ``materialize`` with ``today`` pinned to yesterday —
+    the no-past guard accepts the rows, and they land past-due with no
+    completion events: the sweep's targets), and one quest due TODAY
+    (not past-due: the sweep must NOT announce it).  The yesterday
+    rules carry ``end_date`` so the later today-walk cannot give them
+    a second instance.  Opens, migrates, seeds and CLOSES its own
+    connection so the app's lifespan connection never overlaps it.
+    """
+    database = _core_db.NestQuestDatabase(_executor)
+    await database.open(db_path)
+    try:
+        await _core_migrations.apply_migrations(database)
+        today_iso = today.isoformat()
+        yesterday = today - datetime.timedelta(days=1)
+        yesterday_iso = yesterday.isoformat()
+
+        ada = await _children.create_child(database, "Ada", sort_order=0)
+        bo = await _children.create_child(database, "Bo", sort_order=1)
+
+        yesterday_rule = _recurrence.ScheduleRule.from_dict(
+            {
+                "rule_type": "daily",
+                "start_date": yesterday_iso,
+                "end_date": yesterday_iso,
+            }
+        )
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Stale chore",
+            yesterday_rule,
+            [ada.id],
+            [("morning", "09:00")],
+        )
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Old chore",
+            yesterday_rule,
+            [bo.id],
+            [("evening", "18:30")],
+        )
+        # Past-due open instances for the two yesterday quests (the
+        # walk pins ``today`` to yesterday so the rows are accepted).
+        await _materialize.materialize(
+            database, yesterday_iso, yesterday_iso, today=yesterday
+        )
+        # Today's quest: created after the yesterday walk so it has no
+        # past-due instance, then materialized for today only.
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Recoverable chore",
+            _recurrence.ScheduleRule.from_dict(
+                {"rule_type": "daily", "start_date": today_iso}
+            ),
+            [ada.id],
+            [("afternoon", "17:00")],
+        )
+        await _materialize.materialize(
+            database, today_iso, today_iso, today=today
+        )
+
+        instances = _dao_instances.QuestInstancesDao(database)
+        ada_yesterday = await instances.list_by_child_and_date(
+            ada.id, yesterday_iso
+        )
+        assert len(ada_yesterday) == 1
+        stale = ada_yesterday[0]
+        bo_yesterday = await instances.list_by_child_and_date(
+            bo.id, yesterday_iso
+        )
+        assert len(bo_yesterday) == 1
+        old = bo_yesterday[0]
+        ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
+        assert len(ada_today) == 1
+        recoverable = ada_today[0]
+        return SimpleNamespace(
+            ada=ada,
+            bo=bo,
+            stale_instance=stale,
+            old_instance=old,
+            recoverable_instance=recoverable,
+            yesterday_iso=yesterday_iso,
+            today_iso=today_iso,
+        )
+    finally:
+        await database.close()
+
+
+class _AdminServer:
+    """Run the app with uvicorn on an ephemeral port, in a thread.
+
+    The SSE tests' documented approach (see tests/test_api_sse.py):
+    uvicorn's factory mode runs ``create_app`` INSIDE the server
+    thread, so the lifespan — database, publisher, the daily sweep
+    scheduler — belongs to that thread's event loop.  The app's JWKS
+    cache is swapped for one built around the stubbed fetcher BEFORE
+    the server starts, so admin JWTs genuinely verify with no network.
+    """
+
+    def __init__(self, db_path: str) -> None:
+        self.port = _free_port()
+        self._server_loop: asyncio.AbstractEventLoop | None = None
+        self._server = uvicorn.Server(
+            uvicorn.Config(
+                self._make_app_factory(db_path),
+                host="127.0.0.1",
+                port=self.port,
+                log_level="warning",
+            )
+        )
+        self._thread = threading.Thread(target=self._server.run, daemon=True)
+
+    def _make_app_factory(self, db_path: str):
+        """The uvicorn app factory, closing over ``db_path``."""
+
+        def factory():
+            app = create_app(
+                ApiConfig(
+                    db_path=db_path,
+                    panel_token=PANEL_TOKEN,
+                    oidc_issuer=ISSUER,
+                    oidc_audience=AUDIENCE,
+                    oidc_jwks_url=JWKS_URL,
+                )
+            )
+            app.state.jwks_cache = JwksCache(
+                JWKS_URL, fetcher=StubbedFetch(jwks_for(ADMIN_KEY, KID))
+            )
+            self._server_loop = asyncio.get_running_loop()
+            self.app = app
+            return app
+
+        return factory
+
+    def __enter__(self) -> "_AdminServer":
+        self._thread.start()
+        deadline = time.monotonic() + READ_TIMEOUT
+        while not self._server.started:
+            if time.monotonic() > deadline or not self._thread.is_alive():
+                raise RuntimeError("uvicorn did not start")
+            time.sleep(0.01)
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self._server.should_exit = True
+        self._thread.join(timeout=READ_TIMEOUT)
+
+    def url(self, path: str) -> str:
+        """The full URL for ``path`` on this server."""
+        return f"http://127.0.0.1:{self.port}{path}"
+
+    def run_on_loop(self, coro) -> object:
+        """Run ``coro`` on the SERVER's event loop and wait for it.
+
+        The app (and its database) lives on the uvicorn thread's loop,
+        so database assertions schedule there.  Bounded by
+        READ_TIMEOUT so a wedged schedule fails instead of hanging.
+        """
+        return asyncio.run_coroutine_threadsafe(
+            coro, self._server_loop
+        ).result(timeout=READ_TIMEOUT)
+
+
+def _watermark(server: _AdminServer) -> str | None:
+    """The stored sweep watermark, read on the server's own loop."""
+    database = server.app.state.db.database
+    return server.run_on_loop(
+        _dao_meta.MetaStateDao(database).get(SWEEP_WATERMARK_KEY)
+    )
+
+
+@pytest.fixture
+def temp_db_path(tmp_path: Path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def missed_app(temp_db_path: str) -> SimpleNamespace:
+    """The seeded household served by a live uvicorn app, clock pinned.
+
+    ``routes_admin._local_now`` is pinned to today 12:00 (the same
+    pinning discipline the panel/SSE tests use) BEFORE the server
+    starts, so the trigger route's ``today`` matches the seed; the
+    downtime test moves the pin forward mid-test.  Restored in
+    ``finally`` even on failure.
+    """
+    now_local = datetime.datetime.now().astimezone()
+    pinned_now = now_local.replace(hour=12, minute=0, second=0, microsecond=0)
+    today = pinned_now.date()
+    seed = await _seed_missed_household(temp_db_path, today)
+    original = routes_admin._local_now
+    routes_admin._local_now = lambda: pinned_now
+    try:
+        with _AdminServer(temp_db_path) as server:
+            client = httpx.AsyncClient(base_url=server.url(""))
+            async with client:
+                yield SimpleNamespace(
+                    client=client,
+                    seed=seed,
+                    server=server,
+                    today=today,
+                    pinned_now=pinned_now,
+                )
+    finally:
+        routes_admin._local_now = original
+
+
+# --- the trigger: one missed transition per past-due open instance -----
+
+
+async def test_trigger_publishes_one_missed_event_per_instance(
+    missed_app,
+) -> None:
+    """The trigger streams one documented missed transition per instance.
+
+    The stream is subscribed FIRST, the sweep is triggered through the
+    route as an admin, and each emitted frame's type and every
+    documented payload field are asserted; the TODAY-due instance must
+    NOT appear (the sweep's rule is strictly past-due).
+    """
+    ns = missed_app
+    seed = ns.seed
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        response = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=_admin_headers()
+        )
+        assert response.status_code == 200
+        assert response.json() == {"fired": 2}
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert event_type == "nestquest_quest_missed"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["instance_id"] == seed.stale_instance.id
+        assert payload["quest_title"] == "Stale chore"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == seed.yesterday_iso
+        assert payload["due_time"] == "09:00"
+        assert payload["occurred_at"]  # strict UTC ISO-8601 stamp
+        run_stamp = payload["occurred_at"]
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert payload["child_id"] == seed.bo.id
+        assert payload["child_name"] == "Bo"
+        assert payload["instance_id"] == seed.old_instance.id
+        assert payload["quest_title"] == "Old chore"
+        assert payload["window"] == "evening"
+        assert payload["due_date"] == seed.yesterday_iso
+        assert payload["due_time"] == "18:30"
+        # One run, ONE stamp: the route publishes the sweep-built
+        # payloads directly, so every event of this run carries the
+        # same ``occurred_at`` the sweep computed for the whole run.
+        assert payload["occurred_at"] == run_stamp
+
+        # Exactly one transition per instance: today's not-past-due
+        # quest never arrives.
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+    # The run recorded the watermark: the household-local date the
+    # sweep ran for, under the well-known meta_state key.
+    assert _watermark(ns.server) == seed.today_iso
+
+
+async def test_trigger_publishes_sweep_built_payloads_without_refetch(
+    missed_app,
+) -> None:
+    """Publishing never re-fetches: a concurrently deleted instance
+    cannot fail the trigger or lose its already-built missed event.
+
+    The core sweep commits the idempotency watermark BEFORE returning,
+    so a publish loop that re-fetches each instance opens a loss
+    window: an instance deleted inside that window (e.g. a concurrent
+    regenerate) would raise ``ValueError`` uncaught — a bare 500 — and
+    the missed event would be gone forever (the watermark already
+    advanced, so no future sweep re-examines it).  The route publishes
+    the sweep-built ``(event_type, payload)`` pairs DIRECTLY (the
+    api/scheduler.py shape), which this test exercises deterministically:
+    the sweep is wrapped to delete one swept instance AFTER the sweep
+    built its events and committed its watermark but BEFORE the route
+    publishes, and the trigger must still answer 200 with ``fired: 2``
+    and stream BOTH sweep-built payloads — the deleted instance's
+    included, stamped with the run's ONE shared ``occurred_at``.
+    """
+    ns = missed_app
+    seed = ns.seed
+    database = ns.server.app.state.db.database
+    original_sweep = _core_sweep.run_missed_sweep
+
+    async def sweep_then_delete_stale(sweep_database, *, today):
+        events = await original_sweep(sweep_database, today=today)
+        # The concurrent deletion lands after the sweep built its
+        # events and committed its watermark, but before the route
+        # publishes: the row is gone while the built event survives.
+        await sweep_database.execute(
+            "DELETE FROM quest_instances WHERE id = ?",
+            (seed.stale_instance.id,),
+        )
+        return events
+
+    _core_sweep.run_missed_sweep = sweep_then_delete_stale
+    try:
+        stream = _stream(ns.server.url("/api/v1/panel/events"))
+        async with stream:
+            response = await ns.client.post(
+                "/api/v1/admin/missed-sweep", headers=_admin_headers()
+            )
+            assert response.status_code == 200
+            assert response.json() == {"fired": 2}
+
+            # The FIRST event is the deleted instance's — published
+            # from the payload built before the deletion, never
+            # re-fetched after it (a re-fetch would 500 here).
+            event_type, payload = await stream.next_event()
+            assert event_type == EVENT_QUEST_MISSED
+            assert payload["instance_id"] == seed.stale_instance.id
+            assert payload["child_name"] == "Ada"
+            assert payload["quest_title"] == "Stale chore"
+            assert payload["window"] == "morning"
+            assert payload["due_date"] == seed.yesterday_iso
+            assert payload["occurred_at"]
+            run_stamp = payload["occurred_at"]
+
+            event_type, payload = await stream.next_event()
+            assert event_type == EVENT_QUEST_MISSED
+            assert payload["instance_id"] == seed.old_instance.id
+            assert payload["occurred_at"] == run_stamp
+
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+    finally:
+        _core_sweep.run_missed_sweep = original_sweep
+
+    # The watermark advanced even though one instance was deleted
+    # mid-flight — nothing was lost to the publish-time failure window.
+    assert _watermark(ns.server) == seed.today_iso
+
+
+async def test_same_day_rerun_publishes_nothing(missed_app) -> None:
+    """A second trigger the same day publishes nothing (the watermark)."""
+    ns = missed_app
+    headers = _admin_headers()
+    first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert first.status_code == 200
+    assert first.json() == {"fired": 2}
+
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        second = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=headers
+        )
+        assert second.status_code == 200
+        assert second.json() == {"fired": 0}
+        with pytest.raises(asyncio.TimeoutError):
+            # The watermark makes the rerun an empty no-op: the stream
+            # stays silent.
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+
+async def test_run_after_downtime_publishes_accumulated_window_once(
+    missed_app,
+) -> None:
+    """After days of downtime the catch-up run sweeps the window once.
+
+    The first trigger sweeps yesterday's instances (watermark := today).
+    ``today`` then jumps two days: the accumulated window
+    ``[watermark, today)`` holds exactly the seeded TODAY-due instance
+    (no completion events), which fires once — and a same-day rerun is
+    silent again.
+    """
+    ns = missed_app
+    headers = _admin_headers()
+    first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert first.status_code == 200
+    assert first.json() == {"fired": 2}
+
+    # Advance the pinned clock two days: the app is "down" over the
+    # rollovers in between.
+    routes_admin._local_now = lambda: ns.pinned_now + datetime.timedelta(
+        days=2
+    )
+    later_today = (ns.today + datetime.timedelta(days=2)).isoformat()
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        catchup = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=headers
+        )
+        assert catchup.status_code == 200
+        assert catchup.json() == {"fired": 1}
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert payload["instance_id"] == ns.seed.recoverable_instance.id
+        assert payload["child_id"] == ns.seed.ada.id
+        assert payload["quest_title"] == "Recoverable chore"
+        assert payload["window"] == "afternoon"
+        assert payload["due_date"] == ns.seed.today_iso
+        assert payload["due_time"] == "17:00"
+        assert payload["occurred_at"]
+
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+    # The watermark moved to the new today...
+    assert _watermark(ns.server) == later_today
+    # ...so a further rerun publishes nothing.
+    again = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert again.status_code == 200
+    assert again.json() == {"fired": 0}
+
+
+# --- authentication: the router's ONE require_admin dependency ----------
+
+
+async def test_trigger_refuses_non_admin_and_absent_credentials(
+    missed_app,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401; none sweep."""
+    ns = missed_app
+    non_admin = make_token(groups=("some-other-group",))
+    forbidden = await ns.client.post(
+        "/api/v1/admin/missed-sweep",
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    absent = await ns.client.post("/api/v1/admin/missed-sweep")
+    assert absent.status_code == 401
+
+    # Neither refusal ran the sweep: the watermark is still unset, and
+    # the next admin trigger fires the full seeded set.
+    assert _watermark(ns.server) is None
+    after = await ns.client.post(
+        "/api/v1/admin/missed-sweep", headers=_admin_headers()
+    )
+    assert after.status_code == 200
+    assert after.json() == {"fired": 2}
+
+
+# --- the daily scheduler -------------------------------------------------
+
+
+class _RecordingPublisher:
+    """A transition-publisher stub recording every published event."""
+
+    def __init__(self) -> None:
+        self.published: list[tuple[str, dict]] = []
+
+    def publish(self, event_type: str, payload: dict) -> None:
+        self.published.append((event_type, payload))
+
+
+async def test_scheduler_runs_sweep_at_rollover_and_cancels_cleanly(
+    temp_db_path: str,
+) -> None:
+    """The scheduler sweeps at the configured rollover and stops clean.
+
+    The injectable clock starts at 10:00 and the settings pin the
+    rollover to 10:05, so the first computed delay is five minutes;
+    the injectable sleep fast-forwards the clock, the sweep runs for
+    the (now advanced) clock's date, and the seeded past-due open
+    instance is published on the recording publisher exactly once.
+    The next cycle schedules for the NEXT day's rollover (23h55m).
+    ``stop`` cancels the loop task and the scheduler reports stopped.
+    """
+    database = _core_db.NestQuestDatabase(_executor)
+    await database.open(temp_db_path)
+    try:
+        await _core_migrations.apply_migrations(database)
+        now_local = datetime.datetime.now().astimezone()
+        start = now_local.replace(
+            hour=10, minute=0, second=0, microsecond=0
+        )
+        ada = await _children.create_child(database, "Ada", sort_order=0)
+        yesterday = (start.date() - datetime.timedelta(days=1)).isoformat()
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Stale chore",
+            _recurrence.ScheduleRule.from_dict(
+                {
+                    "rule_type": "daily",
+                    "start_date": yesterday,
+                    "end_date": yesterday,
+                }
+            ),
+            [ada.id],
+            [("morning", "09:00")],
+        )
+        await _materialize.materialize(
+            database, yesterday, yesterday, today=start.date() - datetime.timedelta(days=1)
+        )
+        await _settings_store.update_settings(
+            database, {"day_rollover_time": "10:05"}
+        )
+
+        clock_state = {"now": start}
+        delays: list[float] = []
+
+        def clock() -> datetime.datetime:
+            return clock_state["now"]
+
+        async def sleep_stub(seconds: float) -> None:
+            delays.append(seconds)
+            if len(delays) == 1:
+                # Fast-forward the clock across the rollover wait ONCE,
+                # so the first sweep runs at the rollover instant; the
+                # clock then stays frozen at that instant, so the
+                # follow-up cycles sweep the same (already-watermarked)
+                # date and the assertions read a stable state.
+                clock_state["now"] = clock_state["now"] + datetime.timedelta(
+                    seconds=seconds
+                )
+            await asyncio.sleep(0)
+
+        publisher = _RecordingPublisher()
+        scheduler = MissedSweepScheduler(
+            database, publisher=publisher, clock=clock, sleep=sleep_stub
+        )
+        scheduler.start()
+        try:
+            deadline = time.monotonic() + READ_TIMEOUT
+            while not publisher.published:
+                if time.monotonic() > deadline:
+                    pytest.fail("scheduler never published a missed event")
+                await asyncio.sleep(0.01)
+            # The second cycle's scheduling decision may lag the first
+            # publication by a few awaits; wait for it the same way.
+            while len(delays) < 2:
+                if time.monotonic() > deadline:
+                    pytest.fail("scheduler never scheduled a second cycle")
+                await asyncio.sleep(0.01)
+
+            event_type, payload = publisher.published[0]
+            assert event_type == "nestquest_quest_missed"
+            assert payload["child_id"] == ada.id
+            assert payload["quest_title"] == "Stale chore"
+            assert payload["window"] == "morning"
+            assert payload["due_date"] == yesterday
+            assert payload["due_time"] == "09:00"
+            assert payload["occurred_at"]
+
+            # The rollover delay was honored exactly...
+            assert delays[0] == pytest.approx(300.0)
+            # ...and the next cycle targets the NEXT day's rollover
+            # (the just-passed instant schedules for tomorrow).
+            assert delays[1] == pytest.approx(86400.0)
+            # The watermark pins the run to the sweep-day's date.
+            watermark = await _dao_meta.MetaStateDao(database).get(
+                SWEEP_WATERMARK_KEY
+            )
+            assert watermark == start.date().isoformat()
+        finally:
+            await scheduler.stop()
+        assert not scheduler.running
+        # Exactly one announcement: the watermark makes every further
+        # cycle an empty no-op.
+        assert len(publisher.published) == 1
+    finally:
+        await database.close()