NestQuest/tests/test_api_missed_sweep.py

"""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; ``end_date`` keeps it (and create's horizon
        # regeneration) to today only.
        await _quest_definitions.create_quest_definition(
            database,
            "Recoverable chore",
            _recurrence.ScheduleRule.from_dict(
                {
                    "rule_type": "daily",
                    "start_date": today_iso,
                    "end_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()