diff --git a/api/app.py b/api/app.py index d14c09f..be6a525 100644 --- a/api/app.py +++ b/api/app.py @@ -48,6 +48,7 @@ from api.database import DatabaseState, make_database from api.events import TransitionPublisher from api.routes_admin import router as admin_router from api.routes_panel import router as panel_router +from api.scheduler import MissedSweepScheduler def _build_lifespan(db_path: str): @@ -57,7 +58,11 @@ def _build_lifespan(db_path: str): path in without threading it through FastAPI's lifespan signature. The lifespan also installs the app's ONE in-process transition publisher on ``app.state.publisher`` (see :mod:`api.events`) so - every route and SSE subscriber shares that single instance. + every route and SSE subscriber shares that single instance, and + starts the daily missed-sweep scheduler + (:class:`api.scheduler.MissedSweepScheduler`) against the same + database and publisher — cancelled before the connection closes on + shutdown. """ @asynccontextmanager @@ -65,10 +70,16 @@ def _build_lifespan(db_path: str): state = make_database(db_path) app.state.db = state app.state.publisher = TransitionPublisher() + scheduler = MissedSweepScheduler( + state.database, publisher=app.state.publisher + ) + app.state.sweep_scheduler = scheduler try: await state.open_and_migrate() + scheduler.start() yield finally: + await scheduler.stop() await state.close() return lifespan diff --git a/api/nestquest_core.py b/api/nestquest_core.py index 5cfb60d..1fe8408 100644 --- a/api/nestquest_core.py +++ b/api/nestquest_core.py @@ -198,6 +198,12 @@ core_dao_children = importlib.import_module( #: delegating. core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules") +#: The HA-free missed sweep (``core.sweep.run_missed_sweep`` + +#: ``SWEEP_WATERMARK_KEY``) the admin plane's trigger route calls and +#: the daily scheduler runs — the ONE sweep policy the integration's +#: bus shim shares. +core_sweep = importlib.import_module(f"{_CORE_MODULE_NAME}.sweep") + __all__ = [ "core", "core_children", @@ -218,4 +224,5 @@ __all__ = [ "core_snapshot", "core_settings", "core_settings_store", + "core_sweep", ] diff --git a/api/routes_admin.py b/api/routes_admin.py index 6c8ff9a..4a35aa2 100644 --- a/api/routes_admin.py +++ b/api/routes_admin.py @@ -139,6 +139,20 @@ are deliberately not unified here: 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 ``nestquest_quest_missed`` event + on the SSE stream through + :func:`api.transitions.publish_quest_missed`. 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): @@ -237,6 +251,7 @@ from api.nestquest_core import ( core_recurrence, core_settings, core_settings_store, + core_sweep, ) #: All admin-plane routes share this router; the ONE admin dependency @@ -654,6 +669,17 @@ class AdminSettingsResponse(BaseModel): 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,6 +829,19 @@ def _raise_settings_error(error: ValueError) -> NoReturn: 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. @@ -1635,4 +1674,51 @@ async def admin_update_settings( 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 is published on the SSE stream through + :func:`api.transitions.publish_quest_missed` (the payload built by + the core builder, never hand-built here; fan-out is + fire-and-forget, see api/events.py). 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: + await api_transitions.publish_quest_missed( + database, + int(payload["instance_id"]), + request.app.state.publisher, + ) + return AdminMissedSweepResponse(fired=len(events)) + + __all__ = ["router"] diff --git a/api/scheduler.py b/api/scheduler.py new file mode 100644 index 0000000..503360d --- /dev/null +++ b/api/scheduler.py @@ -0,0 +1,173 @@ +"""Daily missed-sweep background scheduler for the API service. + +The API service has no Home Assistant, so nobody re-fires the +integration's day-rollover listener for it: this module owns the API +plane's own background job. One :class:`MissedSweepScheduler` per app +(installed on ``app.state`` by the lifespan) sleeps until the +configured ``day_rollover_time`` — read fresh from the settings store +at the top of EVERY cycle, so an admin's settings change applies on +the next scheduling decision without a restart — then runs the +HA-free core's :func:`nestquest_core.sweep.run_missed_sweep` for the +API host's local date and publishes each returned +``nestquest_quest_missed`` event on the app's ONE transition publisher +(the same SSE stream every route publishes through). + +All sweep policy lives in the core (the watermark, the +``[watermark, today)`` window, the no-completion-event rule): this +scheduler only decides WHEN. A same-night rerun — e.g. two workers, +or a settings change moving the rollover past itself — fires nothing: +the core's watermark makes the run an empty no-op. + +Testability: ``clock`` (an aware-now callable) and ``sleep`` (an +awaitable delay) are injectable, so tests pin the time, fast-forward +through the wait, and observe the computed delays without waiting a +real day. Cancellation: :meth:`stop` cancels the loop task and waits +for it, so the lifespan's shutdown is clean and a mid-run sweep is +interrupted before the connection closes. A failed iteration (a +database hiccup, a rejected setting) is logged and retried after a +fixed backoff rather than silently killing the job. +""" +from __future__ import annotations + +import asyncio +import datetime +import logging + +from api.nestquest_core import core_settings_store, core_sweep + +LOGGER = logging.getLogger(__name__) + +#: Backoff after a failed scheduling cycle (settings read or sweep +#: error), so a persistent failure retries every minute instead of +#: spinning or dying. +_RETRY_BACKOFF_SECONDS = 60.0 + + +def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float: + """Seconds from ``now`` until the next occurrence of ``target``. + + ``target`` is a plain ``HH:MM`` time of day: the next wall-clock + occurrence strictly AFTER ``now`` (an equal time schedules for + tomorrow, so the just-run sweep is never instantly re-run). + """ + scheduled = now.replace( + hour=target.hour, minute=target.minute, second=0, microsecond=0 + ) + if scheduled <= now: + scheduled += datetime.timedelta(days=1) + return (scheduled - now).total_seconds() + + +class MissedSweepScheduler: + """Run the missed sweep once per day at ``day_rollover_time``.""" + + def __init__( + self, + database: object, + *, + publisher: object | None = None, + clock: object | None = None, + sleep: object | None = None, + ) -> None: + """Create the scheduler (not yet running). + + ``database`` is the app's live :class:`~api.database.NestQuestDatabase`. + ``publisher`` is the app's ONE + :class:`~api.events.TransitionPublisher` (publishing is skipped + when ``None``). ``clock`` defaults to the API host's local + aware now (``datetime.datetime.now().astimezone()``) — the same + single-clock discipline the routes' ``_local_now`` uses; + ``sleep`` defaults to :func:`asyncio.sleep`. + """ + self._database = database + self._publisher = publisher + self._clock = clock if clock is not None else ( + lambda: datetime.datetime.now().astimezone() + ) + self._sleep = sleep if sleep is not None else asyncio.sleep + self._task: asyncio.Task | None = None + + def start(self) -> None: + """Start the background loop (idempotent: a running one stays).""" + if self._task is None or self._task.done(): + self._task = asyncio.create_task( + self._run_forever(), name="nestquest-missed-sweep" + ) + + async def stop(self) -> None: + """Cancel the loop and wait for it to finish (safe if never started).""" + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + + @property + def running(self) -> bool: + """Whether the background loop task is alive.""" + return self._task is not None and not self._task.done() + + async def _run_forever(self) -> None: + """Schedule → sweep → repeat, forever (until cancelled).""" + while True: + try: + settings = await core_settings_store.load_settings( + self._database + ) + rollover = datetime.time.fromisoformat( + settings.day_rollover_time + ) + except Exception: + # Not silent: the failure is logged and retried after + # the backoff, so one bad settings document or a + # transient database error never kills the scheduler. + LOGGER.exception( + "NestQuest missed-sweep scheduler failed to resolve " + "the day rollover time; retrying in %gs", + _RETRY_BACKOFF_SECONDS, + ) + await self._sleep(_RETRY_BACKOFF_SECONDS) + continue + delay = _delay_seconds(rollover, self._clock()) + LOGGER.info( + "NestQuest missed sweep scheduled for %s (%.0fs from now)", + settings.day_rollover_time, + delay, + ) + await self._sleep(delay) + try: + await self._run_once(self._clock().date()) + except Exception: + # Not silent either: the sweep error is logged and the + # loop backs off before the next cycle, whose watermark + # re-read makes an already-swept day a no-op. + LOGGER.exception( + "NestQuest scheduled missed sweep failed; retrying in " + "%gs", + _RETRY_BACKOFF_SECONDS, + ) + await self._sleep(_RETRY_BACKOFF_SECONDS) + + async def _run_once(self, today: datetime.date) -> None: + """Run ONE sweep for ``today`` and publish what it built. + + A failure propagates to :meth:`_run_forever`, which logs it and + backs off before the next cycle; the watermark makes a re-run + of an already-swept day announce nothing twice. + """ + events = await core_sweep.run_missed_sweep( + self._database, today=today + ) + for event_type, payload in events: + if self._publisher is not None: + self._publisher.publish(event_type, payload) + LOGGER.info( + "NestQuest scheduled missed sweep for %s published %d event(s)", + today.isoformat(), + len(events), + ) + + +__all__ = ["MissedSweepScheduler"] \ No newline at end of file diff --git a/custom_components/nestquest/core/sweep.py b/custom_components/nestquest/core/sweep.py new file mode 100644 index 0000000..a23bc15 --- /dev/null +++ b/custom_components/nestquest/core/sweep.py @@ -0,0 +1,143 @@ +"""The nightly missed-quest sweep, Home-Assistant-free (Feature 11). + +This is the HA-free heart of the missed sweep, shared by the two +planes: the Home Assistant integration's :mod:`~..sweep` shim (which +fires the returned events on ``hass.bus``) and the API service (which +publishes them on the SSE stream and can also trigger the run on +demand). It owns the ONE sweep policy: + +At the configured day-rollover time (or on an explicit trigger) this +job finds every instance whose ``due_date`` is before the +household-local date, that has NO completion event, and that falls +after the persisted sweep watermark — and RETURNS one +``nestquest_quest_missed`` event per instance with the documented +payload. Like every core module, it never touches the Home Assistant +event bus and never mutates an instance: ``missed`` is derived state, +never an event_type (Feature 08's contract), so the sweep is read-only +over the domain tables. + +Idempotency is the watermark: after a run, ``nestquest_meta_state`` +records the household-local date the sweep ran, and the next run only +sweeps instances due in ``[watermark, today)``. A second run the same +night therefore fires nothing new, and a run after days of downtime +sweeps exactly the accumulated window once — no per-instance journal, +no re-announcing old misses. An instance swept as missed that is +later completed simply gains its completion event; history keeps both +rows (append-only), and the sweep's later runs skip it via the +no-completion-event rule. + +The whole watermark read → query → build → write span runs inside the +per-database sweep lock (one lock per connection wrapper and running +loop, the settings store's pattern): an overlapping startup sweep and +a rollover sweep queue instead of interleaving, and the loser re-reads +the watermark INSIDE the lock — sees the winner's date — and announces +nothing twice. + +The caller supplies ``today`` as a plain :class:`datetime.date`: the +household-local calendar date, read from ``hass.config.time_zone`` by +the integration or from the API host's local clock by the API. This +module never reads a timezone database or config of its own. +""" +from __future__ import annotations + +import asyncio +import datetime +import logging + +from .const import EVENT_QUEST_MISSED +from .dao_meta import MetaStateDao +from .dao_instances import QuestInstancesDao +from .db import NestQuestDatabase +# Intentional intra-package reuse of the shared payload builder: the +# missed event's payload is the documented §3 instance payload, and +# building it through the ONE builder keeps this sweep's payloads from +# drifting from the completion/uncompleted builders' shape. +from .events import _instance_payload + +LOGGER = logging.getLogger(__name__) + +#: meta_state key holding the household-local date the sweep last ran. +SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date" + +#: One asyncio.Lock per (connection wrapper, running loop), mirroring +#: the settings store's pattern: the whole watermark read → query → +#: build → write span is one critical section, so an overlapping +#: startup sweep and rollover sweep queue instead of double-announcing. +#: The loser re-reads the watermark INSIDE the lock, sees the winner's +#: date, and fires nothing. +_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {} + + +def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock: + """Return the sweep lock bound to ``database``'s running loop.""" + loop = asyncio.get_running_loop() + key = (id(database), id(loop)) + lock = _SWEEP_LOCKS.get(key) + if lock is None: + lock = asyncio.Lock() + _SWEEP_LOCKS[key] = lock + return lock + + +#: Strict UTC ISO-8601 timestamp (the DAO layers' policy): one shape, +#: explicit offset, second precision. +_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00" + + +async def run_missed_sweep( + database: NestQuestDatabase, + *, + today: datetime.date, +) -> list[tuple[str, dict]]: + """Build one missed event per unswept past-due open instance. + + ``today`` is the household-local calendar date the sweep runs for + (the caller owns the clock: the integration reads it from + ``hass.config.time_zone``, the API from its host's local time). + Returns the ``(event_type, payload)`` tuples to announce, EMPTY + when the watermark already covers ``today`` (a same-night rerun) — + the caller decides how to publish them and announces nothing on + its own. + + The whole watermark read → query → build → write span runs inside + the per-database sweep lock: an overlapping startup sweep and a + rollover sweep queue instead of interleaving, and the loser + re-reads the watermark inside the lock — sees the winner's date — + and announces nothing twice. Every event in one run shares one + ``occurred_at`` stamp (one run, one announcement moment), and the + payload comes from the shared + :func:`~.events._instance_payload` builder so the shape can never + drift from the documented §3 contract. + """ + today_iso = today.isoformat() + dao = MetaStateDao(database) + async with _sweep_lock(database): + watermark = await dao.get(SWEEP_WATERMARK_KEY) + if watermark == today_iso: + LOGGER.debug( + "NestQuest missed sweep already ran for %s", today_iso + ) + return [] + + instances_dao = QuestInstancesDao(database) + rows = await instances_dao.list_missed_unswept(today_iso, watermark) + now_stamp = datetime.datetime.now(datetime.timezone.utc).strftime( + _UTC_TIMESTAMP_FORMAT + ) + events: list[tuple[str, dict]] = [] + for instance in rows: + events.append( + ( + EVENT_QUEST_MISSED, + await _instance_payload( + database, instance.id, now_stamp + ), + ) + ) + await dao.set(SWEEP_WATERMARK_KEY, today_iso) + LOGGER.info( + "NestQuest missed sweep for %s built %d event(s)", + today_iso, + len(events), + ) + return events \ No newline at end of file diff --git a/custom_components/nestquest/sweep.py b/custom_components/nestquest/sweep.py index e60ab72..937f969 100644 --- a/custom_components/nestquest/sweep.py +++ b/custom_components/nestquest/sweep.py @@ -1,70 +1,40 @@ -"""The nightly missed-quest sweep (Feature 11). +"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim. -At the configured day-rollover time (the same HA-local listener that -re-materializes the rolling horizon) this job finds every instance -whose ``due_date`` is before the HA-local date, that has NO completion -event, and that falls after the persisted sweep watermark — and fires -ONE ``nestquest_quest_missed`` event per instance with the documented -payload. The sweep is read-only over the domain tables: it never -mutates an instance and never writes a completion event (``missed`` -is derived state, never an event_type — Feature 08's contract). +The sweep's policy is the HA-free core's :mod:`~.core.sweep`: the +watermark read → ``list_missed_unswept`` query → payload build → +watermark span, all inside the per-database sweep lock, shared with +the API plane. This module is the integration's thin adapter over it: +at the configured day-rollover time (the same HA-local listener that +re-materializes the rolling horizon) and once at setup it calls the +core's :func:`~.core.sweep.run_missed_sweep` and fires each returned +event on ``hass.bus`` — ONE ``nestquest_quest_missed`` per past-due +open instance with the documented payload. The core never touches +Home Assistant; the firing lives here. -Idempotency is the watermark: after a run, ``nestquest_meta_state`` -records the HA-local date the sweep ran, and the next run only -sweeps instances due in ``[watermark, today)``. A second run the -same night therefore fires nothing new, and a run after days of -downtime sweeps exactly the accumulated window once — no -per-instance journal, no re-announcing old misses. An instance swept -as missed that is later completed simply gains its completion event; -history keeps both rows (append-only), and the sweep's later runs -skip it via the no-completion-event rule. +Idempotency is the core's watermark: after a run, ``nestquest_meta_state`` +records the HA-local date the sweep ran, and the next run only sweeps +instances due in ``[watermark, today)`` — a second run the same night +fires nothing new, and a run after days of downtime sweeps exactly the +accumulated window once. An instance swept as missed that is later +completed simply gains its completion event; the sweep is read-only +over the domain tables (``missed`` is derived state, never an +event_type — Feature 08's contract). """ from __future__ import annotations -import asyncio import datetime import logging from zoneinfo import ZoneInfo -from .const import EVENT_QUEST_MISSED -from .dao_children import ChildrenDao -from .dao_instances import QuestInstancesDao -from .dao_meta import MetaStateDao -from .dao_rules import QuestDefinitionsDao +from .core.sweep import ( + SWEEP_WATERMARK_KEY, + run_missed_sweep as _run_core_missed_sweep, +) from .db import NestQuestDatabase LOGGER = logging.getLogger(__name__) -#: meta_state key holding the HA-local date the sweep last ran. -SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date" - -#: One asyncio.Lock per (connection wrapper, running loop), mirroring -#: the migration runner's pattern: the whole watermark read → query → -#: fire → write span is one critical section, so an overlapping startup -#: sweep and rollover sweep queue instead of double-announcing. The -#: loser re-reads the watermark INSIDE the lock, sees the winner's -#: date, and fires nothing. -_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {} - - -def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock: - """Return the sweep lock bound to ``database``'s running loop.""" - loop = asyncio.get_running_loop() - key = (id(database), id(loop)) - lock = _SWEEP_LOCKS.get(key) - if lock is None: - lock = asyncio.Lock() - _SWEEP_LOCKS[key] = lock - return lock - -#: UTC ISO-8601 timestamp policy shared with the DAO layers. -_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00" - - -def _now_stamp() -> str: - return datetime.datetime.now(datetime.timezone.utc).strftime( - _UTC_TIMESTAMP_FORMAT - ) +__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"] async def run_missed_sweep( @@ -79,54 +49,24 @@ async def run_missed_sweep( materialization and completion layers); when omitted it is read from ``hass.config.time_zone``. Returns the number of events fired. Fires nothing and returns 0 when the watermark already - covers ``today`` (a same-night rerun). - - The whole watermark read → query → fire → write span runs inside - the per-database sweep lock: an overlapping startup sweep and a - rollover sweep queue instead of interleaving, and the loser - re-reads the watermark inside the lock — sees the winner's date — - and announces nothing twice. + covers ``today`` (a same-night rerun) — the watermark read, the + query, the payload build and the watermark write all live in the + core's locked + :func:`~.core.sweep.run_missed_sweep`; this shim only fires each + returned event on the bus and counts them. """ today_date = ( today if today is not None else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date() ) - today_iso = today_date.isoformat() - dao = MetaStateDao(database) - async with _sweep_lock(database): - watermark = await dao.get(SWEEP_WATERMARK_KEY) - if watermark == today_iso: - LOGGER.debug( - "NestQuest missed sweep already ran for %s", today_iso - ) - return 0 - - instances_dao = QuestInstancesDao(database) - rows = await instances_dao.list_missed_unswept(today_iso, watermark) - children_dao = ChildrenDao(database) - definitions_dao = QuestDefinitionsDao(database) - now_stamp = _now_stamp() - fired = 0 - for instance in rows: - child = await children_dao.get(instance.child_id) - definition = await definitions_dao.get(instance.definition_id) - hass.bus.async_fire( - EVENT_QUEST_MISSED, - { - "child_id": instance.child_id, - "child_name": child.display_name if child else None, - "instance_id": instance.id, - "quest_title": definition.title if definition else None, - "window": instance.window, - "due_date": instance.due_date, - "due_time": instance.due_time, - "occurred_at": now_stamp, - }, - ) - fired += 1 - await dao.set(SWEEP_WATERMARK_KEY, today_iso) + events = await _run_core_missed_sweep(database, today=today_date) + for event_type, payload in events: + hass.bus.async_fire(event_type, payload) + fired = len(events) LOGGER.info( - "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired + "NestQuest missed sweep for %s fired %d event(s)", + today_date.isoformat(), + fired, ) return fired \ No newline at end of file diff --git a/tests/test_api_missed_sweep.py b/tests/test_api_missed_sweep.py new file mode 100644 index 0000000..8ba0708 --- /dev/null +++ b/tests/test_api_missed_sweep.py @@ -0,0 +1,584 @@ +"""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. +- 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") + +#: 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 + + 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" + assert payload["occurred_at"] + + # 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_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() \ No newline at end of file