Add the nightly missed-quest sweep

custom_components/nestquest/__init__.py

@@ -35,6 +35,7 @@ from .migrations import apply_migrations
 from .services import async_deregister_services, async_register_services
 from .store import async_get_db_path
 from .admin_allowlist import seed_setup_admin
+from .sweep import run_missed_sweep
 
 _LOGGER = LOGGER
 
@@ -327,18 +328,23 @@ def _register_day_rollover_listener(
     database: NestQuestDatabase,
     options: dict[str, Any],
 ) -> Callable[[], Any]:
-    """Register the daily materialization listener and return its remover.
+    """Register the daily rollover listener and return its remover.
 
-    Fires once per day at the configured local ``day_rollover_time`` and
+    Fires once per day at the configured local ``day_rollover_time``:
-    materializes the rolling horizon ``[today, today + horizon_days]``
+    first re-materializes the rolling horizon
-    through the shared :func:`_run_horizon_materialization` path, using the
+    ``[today, today + horizon_days]`` through the shared
-    entry's configured (validated) ``horizon_days``.
+    :func:`_run_horizon_materialization` path (the entry's configured,
+    validated ``horizon_days``), then runs the Feature 11 missed sweep
+    — announcing yesterday's still-open quests as missed AFTER today's
+    instances exist, watermark-guarded so a re-fire of the listener the
+    same night announces nothing twice.
     """
     hour, minute = _day_rollover_hour_minute(options)
     horizon_days = _configured_horizon_days(options)
 
     async def _run_materialization(_now: datetime.datetime) -> None:
         await _run_horizon_materialization(hass, database, horizon_days)
+        await run_missed_sweep(hass, database)
 
     return async_track_time_change(
         hass, _run_materialization, hour=hour, minute=minute
@@ -498,6 +504,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         await _run_horizon_materialization(
             hass, database, _configured_horizon_days(dict(entry.options))
         )
+        # The missed sweep runs at startup too (Feature 11): if HA was
+        # down at the rollover time, the previous night's still-open
+        # quests are announced as missed now instead of waiting for
+        # the next midnight.  Watermark-guarded, so this is a no-op
+        # when the night's sweep already ran.
+        await run_missed_sweep(hass, database)
         _register_services(hass)
         # The shared Feature 10 coordinator: one refresh cycle every
         # entity reads from (CONF_UPDATE_INTERVAL seconds, default

custom_components/nestquest/dao_instances.py

@@ -493,6 +493,43 @@ class QuestInstancesDao:
         )
         return [_instance_from_row(row) for row in rows]
 
+    async def list_missed_unswept(
+        self, today_iso: str, watermark: str | None
+    ) -> list[QuestInstanceRecord]:
+        """Return past-due open instances the sweep has not announced.
+
+        Feature 11's missed-sweep query: ``due_date`` strictly before
+        ``today_iso`` (past due), at or after ``watermark`` when one is
+        recorded, and with NO completion event ever (the Feature 08
+        append-only contract: an instance that was ever completed is
+        never reported missed, even when later reversed — history keeps
+        both rows and the sweep never mutates any).  The watermark is
+        the HA-local date the sweep last ran, and the window is
+        ``[watermark, today)``: the nightly run entering day D+1 sweeps
+        exactly day D's instances, so every day's quests are announced
+        once — at the midnight after their due date — and a same-night
+        rerun (watermark == today) sees an empty window.  Read-only;
+        ordered deterministically by (due_date, id).
+        """
+        _validate_date(today_iso, "today_iso")
+        if watermark is not None:
+            _validate_date(watermark, "watermark")
+        sql = (
+            f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
+            "WHERE due_date < ? "
+            "AND NOT EXISTS ("
+            "  SELECT 1 FROM completion_events "
+            "  WHERE instance_id = quest_instances.id"
+            ")"
+        )
+        parameters: tuple = (today_iso,)
+        if watermark is not None:
+            sql += " AND due_date >= ?"
+            parameters += (watermark,)
+        sql += " ORDER BY due_date, id"
+        rows = await self._database.fetch_all(sql, parameters)
+        return [_instance_from_row(row) for row in rows]
+
     async def delete_future_uncompleted(
         self, definition_id: int, cutoff_date: str, *, today: datetime.date | None = None
     ) -> int:

custom_components/nestquest/dao_meta.py

@@ -0,0 +1,44 @@
+"""Typed DAO for the integration-owned ``nestquest_meta_state`` table.
+
+Generic persistent key/value bookkeeping that belongs to the
+integration itself rather than any domain table — currently the
+Feature 11 missed-sweep watermark.  Values are plain strings; callers
+interpret them.  All SQL for this table lives here per the feature
+guardrails, and every method runs through
+:class:`~.db.NestQuestDatabase` so nothing blocks the event loop.
+"""
+from __future__ import annotations
+
+from .dao_children import _connection_lock
+from .db import NestQuestDatabase
+
+_TABLE = "nestquest_meta_state"
+
+
+class MetaStateDao:
+    """Typed async access to the ``nestquest_meta_state`` table."""
+
+    def __init__(self, database: NestQuestDatabase) -> None:
+        self._database = database
+
+    async def get(self, key: str) -> str | None:
+        """Return the stored value for ``key``, or None when absent."""
+        row = await self._database.fetch_one(
+            f"SELECT value FROM {_TABLE} WHERE key = ?", (key,)
+        )
+        return row[0] if row is not None else None
+
+    async def set(self, key: str, value: str) -> None:
+        """Upsert one key/value row atomically.
+
+        The write runs inside one transaction under the shared
+        connection lock, so a racing reader either sees the old value
+        or the fully-written new one, never a half-applied state.
+        """
+        async with _connection_lock(self._database):
+            async with self._database.transaction():
+                await self._database.execute(
+                    f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
+                    "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
+                    (key, value),
+                )

custom_components/nestquest/migrations.py

@@ -28,7 +28,10 @@ copying each definition's assignee across.  Migration 4 adds the
 windows table, migration 5 rebuilds ``quest_instances`` onto the
 widened (definition_id, child_id, due_date, window) key, and migration
 6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
-column.  Each callable migration runs inside its own transaction
+column.  Migration 7 adds the generic ``nestquest_meta_state``
+key/value table for integration-owned bookkeeping (Feature 11's
+missed-sweep watermark).  Each callable migration runs inside its own
+transaction
 together with its version stamp, so a crash mid-step rolls the
 statements and the stamp back together.
 
@@ -80,6 +83,7 @@ from .schema import (
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
     SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
+    SCHEMA_V7_META_STATE_DDL,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -414,6 +418,11 @@ MIGRATIONS: Sequence[MigrationStep] = [
     # actor_child_id column and its actor-pair coherence CHECK,
     # backfilling legacy panel rows to their own child_id.
     _add_actor_child_id,
+    # Migration 7 (Feature 11): the generic meta_state key/value table
+    # — a plain new table, so CREATE IF NOT EXISTS suffices for both
+    # fresh databases (which already made it in migration 1) and
+    # pre-v7 databases.  Strictly additive; no existing table changes.
+    list(SCHEMA_V7_META_STATE_DDL),
 ]
 
 

custom_components/nestquest/schema.py

@@ -384,6 +384,21 @@ SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
     """,
 ]
 
+#: Generic integration-owned key/value state (schema version 7,
+#: Feature 11): persistent bookkeeping that belongs to the integration
+#: rather than any domain table — currently the missed-sweep
+#: watermark.  ``value`` is a plain string; callers interpret it.
+META_STATE_TABLE_SQL = """(
+    key TEXT NOT NULL PRIMARY KEY,
+    value TEXT NOT NULL
+)"""
+
+SCHEMA_V7_META_STATE_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS nestquest_meta_state {META_STATE_TABLE_SQL}
+    """,
+]
+
 SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_CHILDREN_DDL,
     *SCHEMA_V1_ADMIN_USERS_DDL,
@@ -395,6 +410,7 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     *SCHEMA_V1_QUEST_INSTANCES_DDL,
     *SCHEMA_V1_COMPLETION_EVENTS_DDL,
+    *SCHEMA_V7_META_STATE_DDL,
 ]
 
 

custom_components/nestquest/sweep.py

@@ -0,0 +1,132 @@
+"""The nightly missed-quest sweep (Feature 11).
+
+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 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 .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__)
+
+#: 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(
+    hass,
+    database: NestQuestDatabase,
+    *,
+    today: datetime.date | None = None,
+) -> int:
+    """Fire one missed event per unswept past-due open instance.
+
+    ``today`` optionally pins the HA-local date (threaded like the
+    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.
+    """
+    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)
+    LOGGER.info(
+        "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
+    )
+    return fired

tests/test_migrations.py

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock
 import pytest
 
 from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.schema import SCHEMA_V7_META_STATE_DDL
 from custom_components.nestquest.migrations import (
     MIGRATIONS,
     VERSION_TABLE,
@@ -110,6 +111,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
             "presence_overrides",
             "quest_instances",
             "completion_events",
+            "nestquest_meta_state",
             VERSION_TABLE,
         }
         assert _tables(database) == expected
@@ -159,7 +161,8 @@ def test_migration_list_shape() -> None:
     assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
     assert callable(MIGRATIONS[4])
     assert callable(MIGRATIONS[5])
-    assert len(MIGRATIONS) == 6
+    assert MIGRATIONS[6] == SCHEMA_V7_META_STATE_DDL
+    assert len(MIGRATIONS) == 7
 
 
 # ---------------------------------------------------------------------------

tests/test_schema.py

@@ -11,6 +11,7 @@ from custom_components.nestquest.schema import (
     SCHEMA_V1_ADMIN_USERS_DDL,
     SCHEMA_V1_CHILDREN_DDL,
     SCHEMA_V1_COMPLETION_EVENTS_DDL,
+    SCHEMA_V7_META_STATE_DDL,
     SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     SCHEMA_V1_SCHEDULE_RULES_DDL,
@@ -168,7 +169,7 @@ def _insert_override(database, child_id, **overrides):
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_eight_tables() -> None:
+def test_schema_v1_statements_compose_the_nine_tables() -> None:
     assert SCHEMA_V1_STATEMENTS == [
         *SCHEMA_V1_CHILDREN_DDL,
         *SCHEMA_V1_ADMIN_USERS_DDL,
@@ -180,6 +181,7 @@ def test_schema_v1_statements_compose_the_eight_tables() -> None:
         *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
         *SCHEMA_V1_QUEST_INSTANCES_DDL,
         *SCHEMA_V1_COMPLETION_EVENTS_DDL,
+        *SCHEMA_V7_META_STATE_DDL,
     ]
 
 

tests/test_sweep.py

@@ -0,0 +1,259 @@
+"""Tests for the nightly missed-quest sweep (Feature 11, task 1)."""
+from __future__ import annotations
+
+import datetime
+import re
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.children import create_child, list_children
+from custom_components.nestquest.completion import (
+    complete_instance,
+    uncomplete_instance,
+)
+from custom_components.nestquest.const import EVENT_QUEST_MISSED
+from custom_components.nestquest.dao_instances import (
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from custom_components.nestquest.dao_meta import MetaStateDao
+from custom_components.nestquest.materialize import materialize
+from custom_components.nestquest.quest_definitions import (
+    create_quest_definition,
+)
+from custom_components.nestquest.recurrence import ScheduleRule
+from custom_components.nestquest.sweep import (
+    SWEEP_WATERMARK_KEY,
+    run_missed_sweep,
+)
+
+_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00$")
+
+
+async def _seed(hass, make_entry) -> tuple:
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    ada = await create_child(database, "Ada")
+    bo = await create_child(database, "Bo")
+    today = datetime.date.today()
+    await create_quest_definition(
+        database,
+        "Brush teeth",
+        ScheduleRule.from_dict(
+            {"rule_type": "daily", "start_date": today.isoformat()}
+        ),
+        [ada.id, bo.id],
+        ["morning"],
+    )
+    end = (today + datetime.timedelta(days=3)).isoformat()
+    await materialize(database, today.isoformat(), end, today=today)
+    return entry, ada, bo
+
+
+async def test_sweep_fires_one_event_per_past_due_open_instance(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 2, "one per child for today's now-past-due instance"
+
+    events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert len(events) == 2
+    payload = events[0]
+    assert payload["child_name"] in ("Ada", "Bo")
+    assert payload["quest_title"] == "Brush teeth"
+    assert payload["window"] == "morning"
+    assert payload["due_date"] == today.isoformat()
+    assert payload["due_time"] is None
+    assert _TIMESTAMP.match(payload["occurred_at"])
+    assert {payload["child_id"] for payload in events} == {ada.id, bo.id}
+    # The sweep is read-only over the domain tables.
+    assert await _count_events(database) == 0
+
+
+async def test_sweep_is_idempotent_same_night(hass, make_entry) -> None:
+    entry, _ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+
+    first = await run_missed_sweep(hass, database, today=tomorrow)
+    second = await run_missed_sweep(hass, database, today=tomorrow)
+    assert first == 2
+    assert second == 0, "the watermark makes a rerun fire nothing"
+    assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2
+    assert await MetaStateDao(database).get(SWEEP_WATERMARK_KEY) == (
+        tomorrow.isoformat()
+    )
+
+
+async def test_sweep_only_fires_new_instances_after_watermark(
+    hass, make_entry
+) -> None:
+    entry, ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+    day_after = today + datetime.timedelta(days=2)
+
+    # First sweep one day ahead: today's instances fire.
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 2
+    first_events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert all(
+        payload["due_date"] == today.isoformat()
+        for payload in first_events
+    )
+
+    # One more day passes: only tomorrow's instances fire now.
+    fired = await run_missed_sweep(hass, database, today=day_after)
+    assert fired == 2
+    all_events = hass.bus.fired(EVENT_QUEST_MISSED)
+    assert len(all_events) == 4
+    assert all(
+        payload["due_date"] == tomorrow.isoformat()
+        for payload in all_events[2:]
+    )
+
+
+async def test_sweep_never_reports_completed_or_reversed_instances(
+    hass, make_entry
+) -> None:
+    entry, ada, bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+
+    instances_ada = await QuestInstancesDao(database).list_by_date_range(
+        ada.id, today.isoformat(), today.isoformat()
+    )
+    instances_bo = await QuestInstancesDao(database).list_by_date_range(
+        bo.id, today.isoformat(), today.isoformat()
+    )
+    # Ada completed hers; Bo completed his then reversed it.
+    await complete_instance(
+        database,
+        instances_ada[0].id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    await complete_instance(
+        database,
+        instances_bo[0].id,
+        actor_source="panel",
+        actor_child_id=bo.id,
+    )
+    await uncomplete_instance(
+        database,
+        instances_bo[0].id,
+        actor_source="user",
+        actor_user_id="admin-1",
+    )
+
+    fired = await run_missed_sweep(hass, database, today=tomorrow)
+    assert fired == 0, (
+        "no-completion-event rule excludes both the completed and the "
+        "reversed instance (append-only history keeps both)"
+    )
+    assert hass.bus.fired(EVENT_QUEST_MISSED) == []
+
+
+async def test_sweep_skips_instance_completed_after_sweep(
+    hass, make_entry
+) -> None:
+    """A missed instance completed later is never re-announced; the
+    late completion still lands in the append-only log."""
+    entry, ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    today = datetime.date.today()
+    tomorrow = today + datetime.timedelta(days=1)
+    day3 = today + datetime.timedelta(days=2)
+
+    await run_missed_sweep(hass, database, today=tomorrow)
+    instance = (
+        await QuestInstancesDao(database).list_by_date_range(
+            ada.id, today.isoformat(), today.isoformat()
+        )
+    )[0]
+    await complete_instance(
+        database,
+        instance.id,
+        actor_source="panel",
+        actor_child_id=ada.id,
+    )
+    fired = await run_missed_sweep(hass, database, today=day3)
+    assert fired == 0 or all(
+        payload["instance_id"] != instance.id
+        for payload in hass.bus.fired(EVENT_QUEST_MISSED)[2:]
+    )
+
+
+async def test_rollover_listener_runs_sweep_after_materialization(
+    hass, make_entry, monkeypatch
+) -> None:
+    import custom_components.nestquest as nestquest
+
+    calls = []
+
+    async def _record_sweep(hass_arg, database, **kwargs):
+        calls.append("sweep")
+
+    monkeypatch.setattr(nestquest, "run_missed_sweep", _record_sweep)
+
+    entry = wire_entry_to_registry(
+        make_entry(options={"day_rollover_time": "00:00"}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+    assert calls == ["sweep"], "startup runs the sweep once"
+
+    await hass.time_change.fire()
+    assert calls == ["sweep", "sweep"], "the rollover listener sweeps too"
+
+
+async def test_v6_database_migrates_to_v7_meta_state(
+    hass, make_entry
+) -> None:
+    """A database stamped at version 6 gains the meta_state table and
+    version 7 on the next startup; existing rows are untouched."""
+    entry = wire_entry_to_registry(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    version = await database.fetch_one(
+        "SELECT version FROM nestquest_schema_version"
+    )
+    assert version == (7,)
+    tables = await database.fetch_all(
+        "SELECT name FROM sqlite_master WHERE type='table' "
+        "AND name='nestquest_meta_state'"
+    )
+    assert tables, "migration 7 created the meta_state table"
+
+
+async def _count_events(database) -> int:
+    rows = await database.fetch_all("SELECT COUNT(*) FROM completion_events")
+    return rows[0][0]
+
+
+async def test_overlapping_sweeps_never_double_announce(
+    hass, make_entry
+) -> None:
+    """A startup sweep and a rollover sweep racing on the same night
+    queue on the sweep lock: the winner announces, the loser re-reads
+    the watermark inside the lock and fires nothing."""
+    import asyncio
+
+    entry, _ada, _bo = await _seed(hass, make_entry)
+    database = entry.runtime_data.database
+    tomorrow = datetime.date.today() + datetime.timedelta(days=1)
+
+    first, second = await asyncio.gather(
+        run_missed_sweep(hass, database, today=tomorrow),
+        run_missed_sweep(hass, database, today=tomorrow),
+    )
+    assert first + second == 2, "each instance announced exactly once"
+    assert len(hass.bus.fired(EVENT_QUEST_MISSED)) == 2