Admin routes: uncomplete and regenerate

The admin plane could reverse nothing and re-materialize nothing: a
mis-tapped completion could only be fixed by hand in the database, and
a config or presence change left the household's rolling horizon
stale until the integration itself regenerated it.

- POST /api/v1/admin/instances/{id}/uncomplete appends the reversal
  through core uncomplete_instance (actor_source 'user' with the
  verified JWT's sub — D-008), so the original completed row is never
  edited or deleted and the derived state returns to open.  An actual
  reversal publishes the quest-uncompleted transition via the existing
  api.transitions helper; an already-open no-op publishes nothing and
  still answers with the honest derived state (open, or missed when
  past due).
- POST /api/v1/admin/regenerate re-materializes the rolling horizon
  through the same core path the integration uses: household (default)
  = materialize over [today, today + const.DEFAULT_HORIZON_DAYS] (the
  API config supplies no horizon knob), definition/child =
  regenerate_for_definition / regenerate_for_child after ONE DAO
  existence read each (unknown id -> 404, which the core's regenerate
  path itself does not check).  The walk is idempotent, so re-running
  never duplicates an instance.

Both handlers are thin adapters: Pydantic models carry the request
shape (scope/id pairing is the model's job), the business rules stay
in core, and the error mapping stays narrow — unknown
instance/definition/child 404, rejected argument 422, anything else
re-raises.

api/app.py

@@ -48,7 +48,6 @@ 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):
@@ -58,11 +57,7 @@ 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, and
-    starts the daily missed-sweep scheduler
-    (:class:`api.scheduler.MissedSweepScheduler`) against the same
-    database and publisher — cancelled before the connection closes on
-    shutdown.
+    every route and SSE subscriber shares that single instance.
     """
 
     @asynccontextmanager
@@ -70,16 +65,10 @@ 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

@@ -116,11 +116,6 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
 #: ``uncomplete_instance``) the panel's write routes call.
 core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion")
 
-#: The read-only history layer (``core.history.query_history``,
-#: ``export_history_csv``, ``rows_to_csv``) the admin plane's history
-#: query and CSV export routes call.
-core_history = importlib.import_module(f"{_CORE_MODULE_NAME}.history")
-
 #: The transition-event builders (``core.events.build_quest_completed_event``
 #: et al.) the panel's write routes build their SSE events with.
 core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
@@ -129,17 +124,6 @@ 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")
@@ -198,12 +182,6 @@ 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",
@@ -214,7 +192,6 @@ __all__ = [
     "core_dao_rules",
     "core_db",
     "core_events",
-    "core_history",
     "core_materialize",
     "core_migrations",
     "core_presence",
@@ -223,6 +200,4 @@ __all__ = [
     "core_recurrence",
     "core_snapshot",
     "core_settings",
-    "core_settings_store",
-    "core_sweep",
 ]

api/routes_admin.py

@@ -92,70 +92,6 @@ pattern over :mod:`nestquest_core.completion` and
   instance.  The response carries the scope, the scoped id (if any)
   and the core's upsert ``count``.
 
-The history routes (task 2d2dda53) follow the same pattern over
-:mod:`nestquest_core.history` — the ONE read-only history layer, which
-owns every rule so the two routes cannot drift:
-
-- ``GET /history?filter=all|reversals|missed&start=YYYY-MM-DD&end=YYYY-MM-DD``
-  returns the history rows for ONE filter over the CLOSED due-date
-  range ``[start, end]``: ``all`` (every completion event — completed
-  AND uncompleted — whose instance is due in the range), ``reversals``
-  (the uncompleted events only) and ``missed`` (the DERIVED past-due
-  open instances, computed by the core's own derive-state rule —
-  ``missed`` is never a stored event_type and nothing is appended).
-  The range scopes by the instance's ``due_date`` (the Feature 13
-  rule: history asks what happened to the tasks due that day), not the
-  event's wall-clock stamp.  The filter and date-range policies live
-  in the core (strict ``YYYY-MM-DD``, ``end >= start``, a
-  ``ValueError`` naming the offending field); the handler is a thin
-  adapter that threads ONE clock read (:func:`_local_now`, the missed
-  derivation's ``today``) and serializes the core's rows.
-- ``GET /history.csv?...`` exports the SAME filtered range as CSV:
-  ONE :func:`nestquest_core.history.export_history_csv` call — the
-  same query path the JSON route uses — answered as ``text/csv`` with
-  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):
 
@@ -205,22 +141,6 @@ deliberately narrow):
   stance.  Every core ``ValueError`` (a rejected horizon, a malformed
   date — unreachable from this body model, but mapped anyway) is 422;
   anything else re-raises.
-- The HISTORY routes have no path ids, so they have no 404 case: a
-  missing query parameter (the request's shape) is FastAPI's 422
-  before any handler code runs, and every core ``ValueError`` — an
-  unknown filter, a non-strict or inverted date range, each naming the
-  offending field (``filter`` / ``start`` / ``end``) — is 422
-  (:func:`_raise_history_error`): well-formed but semantically
-  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
@@ -234,7 +154,7 @@ from __future__ import annotations
 import datetime
 from typing import Annotated, Literal, NoReturn
 
-from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, field_validator, model_validator
 
 from api import transitions as api_transitions
@@ -246,15 +166,11 @@ from api.nestquest_core import (
     core_const,
     core_dao_children,
     core_dao_rules,
-    core_history,
     core_materialize,
     core_presence,
     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
@@ -610,79 +526,6 @@ class AdminRegenerateResponse(BaseModel):
     count: int
 
 
-class AdminHistoryRow(BaseModel):
-    """One history row as the admin plane serializes it.
-
-    Mirrors :class:`nestquest_core.history.HistoryRow` — the Admin spec
-    §5 event-row fields, the same set the CSV export's header carries.
-    A stored event carries its UTC ``occurred_at`` stamp and the
-    ``completed``/``uncompleted`` ``event_type``; a derived missed row
-    (``missed`` is NEVER a stored event_type) carries ``event_type``
-    ``'missed'``, ``occurred_at`` ``None`` (there is no event) and the
-    ``nightly sweep`` actor.
-    """
-
-    occurred_at: str | None
-    event_type: str
-    child_name: str
-    child_id: int
-    quest_title: str
-    instance_id: int
-    window: str
-    due_date: str
-    due_time: str | None
-    actor: str
-    was_on_time: bool | None
-
-
-class AdminHistoryResponse(BaseModel):
-    """The history query's answer.
-
-    ``filter``/``start``/``end`` echo the request, ``count`` is the row
-    count, and ``rows`` are the core's rows in its deterministic order
-    (``due_date``, then instance, then event id — the per-instance
-    append order, so a completion and its reversal keep their order).
-    """
-
-    filter: str
-    start: str
-    end: str
-    count: int
-    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"
 
@@ -803,48 +646,6 @@ def _raise_regenerate_error(error: ValueError) -> NoReturn:
     raise HTTPException(status_code=422, detail=str(error)) from error
 
 
-def _raise_history_error(error: ValueError) -> NoReturn:
-    """Map a core history ``ValueError`` onto 422 and raise it.
-
-    The history routes have no path ids, so they have no 404 case:
-    every ``ValueError`` that survives to here is a rejected argument —
-    an unknown filter, a non-strict date, or an inverted range, each
-    raised by the core naming the offending field (``filter`` /
-    ``start`` / ``end``) BEFORE any read — and is 422 with the core's
-    message as the detail.  Only ``ValueError`` is caught by the
-    handlers, so an unexpected failure re-raises rather than becoming a
-    4xx.
-    """
-    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.
 
@@ -1493,240 +1294,4 @@ async def admin_regenerate(
     )
 
 
-def _history_row_response(row: core_history.HistoryRow) -> AdminHistoryRow:
-    """Serialize one core HistoryRow into the documented payload."""
-    return AdminHistoryRow(
-        occurred_at=row.occurred_at,
-        event_type=row.event_type,
-        child_name=row.child_name,
-        child_id=row.child_id,
-        quest_title=row.quest_title,
-        instance_id=row.instance_id,
-        window=row.window,
-        due_date=row.due_date,
-        due_time=row.due_time,
-        actor=row.actor,
-        was_on_time=row.was_on_time,
-    )
-
-
-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",
-    response_model=AdminHistoryResponse,
-)
-async def admin_history_query(
-    request: Request,
-    start: str,
-    end: str,
-    filter: str = "all",
-) -> AdminHistoryResponse:
-    """Return the history rows for one filter over the closed range.
-
-    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: ``filter``
-    (default ``all``), ``start`` and ``end`` go straight to
-    :func:`nestquest_core.history.query_history` as STRINGS — the
-    filter policy (``all`` / ``reversals`` / ``missed``) and the
-    date-range policy (strict ``YYYY-MM-DD``, ``end >= start``) live in
-    the core, which raises a ``ValueError`` naming the offending field;
-    the handler only threads ONE clock read (:func:`_local_now`) as the
-    missed derivation's ``today`` and serializes the returned rows.
-    ``start``/``end`` are required query parameters (missing is
-    FastAPI's 422 before any handler code runs); every core
-    ``ValueError`` is 422 (:func:`_raise_history_error`).
-    """
-    state: DatabaseState = request.app.state.db
-    today = _local_now().date()
-    try:
-        rows = await core_history.query_history(
-            state.database, filter, start, end, today=today
-        )
-    except ValueError as error:
-        _raise_history_error(error)
-    return AdminHistoryResponse(
-        filter=filter,
-        start=start,
-        end=end,
-        count=len(rows),
-        rows=[_history_row_response(row) for row in rows],
-    )
-
-
-@router.get(
-    "/history.csv",
-    summary="Export the completion history as CSV",
-    response_class=Response,
-)
-async def admin_history_csv(
-    request: Request,
-    start: str,
-    end: str,
-    filter: str = "all",
-) -> Response:
-    """Export the SAME filtered history range as a CSV attachment.
-
-    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 ONE
-    :func:`nestquest_core.history.export_history_csv` call — the same
-    query path the JSON route uses, so the CSV can never drift from the
-    query's filtered range — with the core's validation policy and the
-    :mod:`csv`-based serialization (stable documented header row, RFC
-    4180 quoting) living in the core.  The answer is ``text/csv`` with
-    a ``Content-Disposition`` attachment filename naming the range;
-    the range is core-validated strict ``YYYY-MM-DD`` by the time the
-    filename is built, so it is injection-safe.  Error mapping is the
-    JSON route's (:func:`_raise_history_error`).
-    """
-    state: DatabaseState = request.app.state.db
-    today = _local_now().date()
-    try:
-        csv_text = await core_history.export_history_csv(
-            state.database, filter, start, end, today=today
-        )
-    except ValueError as error:
-        _raise_history_error(error)
-    filename = f"nestquest-history-{start}-to-{end}.csv"
-    return Response(
-        content=csv_text,
-        media_type="text/csv",
-        headers={
-            "Content-Disposition": f'attachment; filename="{filename}"'
-        },
-    )
-
-
-@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

@@ -1,173 +0,0 @@
-"""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/history.py

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

custom_components/nestquest/core/settings_store.py

@@ -1,210 +0,0 @@
-"""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

@@ -1,143 +0,0 @@
-"""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,40 +1,70 @@
-"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim.
+"""The nightly missed-quest sweep (Feature 11).
 
-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.
+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).
 
-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).
+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.
 """
 from __future__ import annotations
 
+import asyncio
 import datetime
 import logging
 from zoneinfo import ZoneInfo
 
-from .core.sweep import (
-    SWEEP_WATERMARK_KEY,
-    run_missed_sweep as _run_core_missed_sweep,
-)
+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 .db import NestQuestDatabase
 
 LOGGER = logging.getLogger(__name__)
 
-__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"]
+#: 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
+    )
 
 
 async def run_missed_sweep(
@@ -49,24 +79,54 @@ 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 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.
+    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.
     """
     today_date = (
         today
         if today is not None
         else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
     )
-    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)
+    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)
     LOGGER.info(
-        "NestQuest missed sweep for %s fired %d event(s)",
-        today_date.isoformat(),
-        fired,
+        "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
     )
     return fired

tests/test_api_admin_history.py

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

tests/test_api_admin_settings.py

@@ -1,398 +0,0 @@
-"""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

@@ -1,671 +0,0 @@
-"""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()