Automatically prepare releases for completed features

custom_components/nestquest/core/dao_meta.py

@@ -41,38 +41,4 @@ class MetaStateDao:
                     f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
                     "ON CONFLICT (key) DO UPDATE SET value = excluded.value",
                     (key, value),
-                )
-
-    async def claim(self, key: str, value: str) -> bool:
-        """Atomically set ``key`` to ``value`` unless it already is.
-
-        The single upsert statement carries a ``WHERE`` guard on the
-        conflict path (``t.value IS NOT excluded.value``), so the
-        read-compare-write is ONE SQLite statement — atomic at the
-        DATABASE level, not just within one process: two connections
-        (two processes sharing one SQLite file) cannot both observe
-        "not yet set" and both write, because the second writer's
-        statement re-evaluates the guard against the first writer's
-        committed value and matches zero rows.
-
-        Returns True only when THIS call changed the row: the first
-        claimant wins and every later claimant of the same value
-        returns False.  A DIFFERENT value always wins (an advance, e.g.
-        the sweep watermark moving to a new date) and re-arms the
-        same-value loser check for the new value.
-
-        Runs under the shared connection lock and inside one
-        transaction, like :meth:`set`; ``cursor.rowcount`` (1 for a
-        matched write, 0 for a guarded no-op) is captured by the
-        wrapper on the executor thread, so no sqlite3 object crosses
-        back to the event loop.
-        """
-        async with _connection_lock(self._database):
-            async with self._database.transaction():
-                result = await self._database.execute(
-                    f"INSERT INTO {_TABLE} (key, value) VALUES (?, ?) "
-                    "ON CONFLICT (key) DO UPDATE SET value = excluded.value "
-                    f"WHERE {_TABLE}.value IS NOT excluded.value",
-                    (key, value),
-                )
-        return result.rowcount == 1
+                )

custom_components/nestquest/core/sweep.py

@@ -26,17 +26,12 @@ 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 span runs inside the
+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 of the SAME process queue instead of interleaving.
-The announce decision itself is the DATABASE-level claim
-(:meth:`MetaStateDao.claim`), so two PROCESSES sharing one SQLite file
-(the integration's rollover sweep and the API service's scheduler)
-cannot both announce the same missed batch either: both may read the
-old watermark and build the events, but the single conditional upsert
-lets exactly one of them win the claim — the loser returns [] and
-announces nothing.
+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
@@ -65,11 +60,11 @@ LOGGER = logging.getLogger(__name__)
 SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
 
 #: One asyncio.Lock per (connection wrapper, running loop), mirroring
-#: the settings store's pattern: the watermark read → query → build
-#: span is one critical section, so an overlapping startup sweep and
-#: rollover sweep of the SAME process queue instead of interleaving.
-#: Same-process callers are fully serialised by it; the cross-process
-#: case is closed by the database-level claim, not by this lock.
+#: 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] = {}
 
 
@@ -104,24 +99,15 @@ async def run_missed_sweep(
     the caller decides how to publish them and announces nothing on
     its own.
 
-    The watermark read → query → build span runs inside the
-    per-database sweep lock, so overlapping callers of ONE process
-    queue instead of interleaving.  The announce decision is the
-    DATABASE-level claim (:meth:`MetaStateDao.claim` — a single
-    conditional upsert on the watermark, atomic across processes):
-    every caller builds the events optimistically, then only the
-    caller whose claim actually advanced the watermark returns them;
-    a loser (another process already announced for ``today``) returns
-    [] and announces nothing twice.  Every event in one run shares one
+    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.
-
-    Failure ordering: the watermark is advanced ONLY by the claim,
-    AFTER the events are fully built — a failure before the claim
-    leaves the watermark unadvanced, so a later run re-sweeps the same
-    window and nothing is lost.
     """
     today_iso = today.isoformat()
     dao = MetaStateDao(database)
@@ -148,23 +134,10 @@ async def run_missed_sweep(
                     ),
                 )
             )
-        # The database-level claim gates the announcement: only the
-        # caller that actually advanced the watermark may publish the
-        # built events, so two processes racing one SQLite file cannot
-        # both announce the same missed batch.  A same-value loser's
-        # claim matches zero rows and this run announces nothing.
-        claimed = await dao.claim(SWEEP_WATERMARK_KEY, today_iso)
-        if not claimed:
-            LOGGER.info(
-                "NestQuest missed sweep for %s lost the watermark claim; "
-                "announcing nothing",
-                today_iso,
-            )
-            return []
+        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
-

orchestration_tests/test_cycle_prompt.py

@@ -51,6 +51,36 @@ class CyclePromptTests(unittest.TestCase):
         self.assertNotIn("The persistent user service runs", output)
         self.assertNotIn("---\n\n---\n\n# 27", output)
 
+    def test_task_complete_feature_triggers_one_guarded_promotion(self):
+        output = RENDER(TEMPLATE, PROFILE)
+        self.assertIn(
+            "feature becomes task-complete and its reviewed merge into `dev` is verified",
+            output,
+        )
+        self.assertIn("Every required feature task is `done`", output)
+        self.assertIn("No task review, remediation, approval, or merge remains pending", output)
+        self.assertIn("feature acceptance criteria and integrated checks pass", output)
+        self.assertIn("Independent approval applies to the exact feature head", output)
+        self.assertIn("approver's merge of that head into `dev` is verified", output)
+        self.assertIn("reflects the completed feature and exact repository references", output)
+        self.assertIn("Task completion inside an unmerged feature branch does not trigger", output)
+        self.assertIn("Do not create a duplicate promotion", output)
+        self.assertIn("Allow only one active promotion pull request to `main`", output)
+        self.assertIn("if it was closed without merge, recalculate", output)
+        self.assertIn("sole active orchestrator for the project", output)
+        self.assertIn("including controller exclusivity, reconciliation", output)
+        self.assertIn("RELEASE_PREPARATION_POLICY = EXCLUDED", output)
+        self.assertIn("Active promotion branch and pull request", output)
+        self.assertIn("An active promotion that has not reached `RELEASE_READY`", output)
+        self.assertIn("RELEASE_PREPARATION_POLICY:\n", output)
+        self.assertIn("A ready promotion is a local checkpoint", output)
+        self.assertIn("continue that work while the ready promotion waits", output)
+        self.assertLess(
+            output.index("Begin an automatically triggered, requested, or scheduled release"),
+            output.index("Advance to the next eligible feature"),
+        )
+        self.assertIn("Only Joshua may merge into `main`", output)
+
 
 if __name__ == "__main__":
     unittest.main()

orchestration_tests/test_nq_orchestrate.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+import fcntl
 import os
 import subprocess
 import tempfile
@@ -47,6 +48,7 @@ class ForegroundLauncherTests(unittest.TestCase):
                 "NQ_PROJECT_ROOT": str(project),
                 "OPENCODE_BIN": str(opencode),
                 "SYSTEMCTL_BIN": str(systemctl),
+                "NQ_STATE_DIR": str(root / "state"),
             })
             printed = subprocess.run([str(SCRIPT), "--print-prompt"], env=env,
                                      text=True, capture_output=True)
@@ -67,6 +69,16 @@ class ForegroundLauncherTests(unittest.TestCase):
                              ["--check-interactive", "--project", str(project)])
 
             output.unlink()
+            lock_path = root / "state/controller.lock"
+            with lock_path.open("w") as held_lock:
+                fcntl.flock(held_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                locked = subprocess.run([str(SCRIPT)], env=env, text=True,
+                                        capture_output=True)
+                self.assertEqual(locked.returncode, 75)
+                self.assertIn("Another NestQuest orchestrator or controller",
+                              locked.stderr)
+                self.assertFalse(output.exists())
+
             open_marker.touch()
             refused = subprocess.run([str(SCRIPT)], env=env, text=True,
                                      capture_output=True)

scripts/nq-orchestrate

@@ -11,6 +11,7 @@ control_home=${NQ_CONTROL_HOME:-/home/overseer/.local/share/nestquest-controller
 project=${NQ_PROJECT_ROOT:-/mnt/ai-storage/projects/NestQuest}
 opencode_bin=${OPENCODE_BIN:-/home/overseer/.opencode/bin/opencode}
 systemctl_bin=${SYSTEMCTL_BIN:-systemctl}
+state_dir=${NQ_STATE_DIR:-/home/overseer/.local/state/nestquest-controller}
 renderer="$control_home/scripts/render-cycle-prompt"
 profile="$control_home/templates/nestquest.profile.json"
 
@@ -41,6 +42,17 @@ fi
 # Foreground mode owns orchestration. A dormant service must never race it.
 "$systemctl_bin" --user stop nestquest-controller.service
 
+# Share the controller's lock and hold it across the final exec. This closes
+# the launch-time race between two foreground sessions as well as the race with
+# the background controller.
+mkdir -p "$state_dir"
+chmod 700 "$state_dir"
+exec 9>"$state_dir/controller.lock"
+if ! flock -n 9; then
+  echo "Another NestQuest orchestrator or controller is running." >&2
+  exit 75
+fi
+
 session_status=$("$control_home/scripts/nq-controller" --check-interactive --project "$project")
 if [[ "$session_status" == "interactive NestQuest session open" ]]; then
   echo "Another interactive NestQuest OpenCode session is open." >&2

templates/ORCHESTRATOR-PROMPT.template.md

@@ -150,6 +150,15 @@ Possible scopes:
 - `PROJECT_QUEUE`
 - `RELEASE_PREPARATION`
 
+Also set `RELEASE_PREPARATION_POLICY` at startup:
+
+- `AUTO` by default.
+- `EXCLUDED` only when the user's launch request explicitly says not to prepare
+  a release or `{{INTEGRATION_BRANCH}}`-to-`{{RELEASE_BRANCH}}` promotion.
+
+Record both `MISSION_SCOPE` and `RELEASE_PREPARATION_POLICY` in durable run
+state. Do not infer exclusion merely from `SINGLE_TASK` or `SINGLE_FEATURE`.
+
 If the user explicitly names one task or feature, respect that scope.
 
 If the user launches the orchestrator for general/autonomous {{PROJECT_NAME}} development without limiting it to one feature, use:
@@ -171,6 +180,7 @@ Maintain a durable orchestration checkpoint using {{WORK_TRACKER}} wherever its
 Track at minimum:
 
 - `MISSION_SCOPE`
+- `RELEASE_PREPARATION_POLICY`
 - Current feature
 - Current controller phase
 - Active development agents
@@ -185,6 +195,9 @@ Track at minimum:
 - Blocked items
 - Tests/checks status
 - Last verified repository state
+- Active promotion branch and pull request
+- Triggering feature and exact `{{INTEGRATION_BRANCH}}` commit for each automatic promotion
+- Completed features pending the next promotion
 - `NEXT_ACTION`
 
 After every material state transition:
@@ -254,8 +267,8 @@ Choose the highest-priority actionable transition in this order:
 8. Remediate feature-level review findings.
 9. Obtain feature approval.
 10. Merge an approved feature into `{{INTEGRATION_BRANCH}}`.
-11. Advance to the next eligible feature when mission scope permits.
-12. Begin a requested/scheduled release when appropriate.
+11. Begin an automatically triggered, requested, or scheduled release when appropriate. The automatic post-feature trigger is in scope for both `SINGLE_FEATURE` and `PROJECT_QUEUE` unless `RELEASE_PREPARATION_POLICY = EXCLUDED`.
+12. Advance to the next eligible feature when mission scope permits.
 13. Continue unrelated work around locally blocked items.
 14. Escalate only when the defined escalation rules require it.
 
@@ -588,9 +601,10 @@ Immediately transition to `FEATURE_INTEGRATION`.
 
 A merged feature is a transition.
 
-It is terminal ONLY when:
-
-`MISSION_SCOPE = SINGLE_FEATURE`
+It is terminal only when `MISSION_SCOPE = SINGLE_FEATURE` and no authorized
+post-feature release transition is pending. Preparing the automatic promotion
+for that completed feature is part of `SINGLE_FEATURE` scope when
+`RELEASE_PREPARATION_POLICY = AUTO`.
 
 Otherwise select the next eligible work.
 
@@ -637,13 +651,19 @@ The orchestrator itself does not perform the merge.
 When `MISSION_SCOPE = PROJECT_QUEUE` and a feature is successfully merged into `{{INTEGRATION_BRANCH}}`:
 
 1. Reconcile {{WORK_TRACKER}}.
-2. Check for a requested/scheduled release.
+2. Evaluate the automatic, requested, and scheduled release triggers.
 3. Identify remaining incomplete features.
 4. Determine which features are eligible based on dependencies and priority.
 5. Select the next eligible feature.
 6. Initialize it.
 7. Continue the development loop.
 
+If an automatic promotion becomes `RELEASE_READY` while another eligible
+feature remains, keep that promotion waiting for {{RELEASE_OWNER_NAME}} and
+continue the project queue. A ready promotion is a local checkpoint in this
+case, not a global terminal condition. Features completed while it remains open
+are recorded as pending for the next promotion under Section 18.
+
 Do not return to the user simply because one feature completed.
 
 If several features are available, choose using:
@@ -658,27 +678,52 @@ If several features are available, choose using:
 
 # 18. RELEASE RULES
 
-Completed features accumulate in `{{INTEGRATION_BRANCH}}`.
+Completed features accumulate in `{{INTEGRATION_BRANCH}}`. With no active
+promotion, each verified feature completion starts release preparation. While
+one promotion is active, later completed features accumulate for the next
+promotion.
 
-Prepare a release only when:
+Prepare a release when any of these conditions is true:
 
 - {{RELEASE_OWNER_NAME}} requests it; OR
-- {{WORK_TRACKER}} schedules it.
+- {{WORK_TRACKER}} schedules it; OR
+- `RELEASE_PREPARATION_POLICY = AUTO`, and a feature becomes task-complete and its reviewed merge into `{{INTEGRATION_BRANCH}}` is verified.
+
+For the automatic trigger, a feature is task-complete only when:
+
+- Every required feature task is `done`, or deliberately `canceled` with recorded evidence.
+- No task review, remediation, approval, or merge remains pending.
+- The feature acceptance criteria and integrated checks pass.
+- Independent approval applies to the exact feature head.
+- The approver's merge of that head into `{{INTEGRATION_BRANCH}}` is verified.
+- {{WORK_TRACKER}} reflects the completed feature and exact repository references.
+
+Task completion inside an unmerged feature branch does not trigger a release.
+The verified feature-to-`{{INTEGRATION_BRANCH}}` merge is the triggering transition.
+
+Before creating a release branch or pull request:
+
+1. Verify this session is the sole active orchestrator for the project. `{{TERMINAL_LAUNCHER}}` enforces that exclusivity by holding the shared controller lock for the full foreground session. If that lock is not held by this session, do not create a release branch or pull request; escalate instead.
+2. Reconcile open release records, release branches, and promotion pull requests.
+3. Do not create a duplicate promotion for an already-covered `{{INTEGRATION_BRANCH}}` commit.
+4. Allow only one active promotion pull request to `{{RELEASE_BRANCH}}`. If an earlier promotion is still open and does not contain the newly completed feature, record the feature as pending for the next automatic promotion. Resume it after the active promotion is merged or closed; if it was closed without merge, recalculate the next promotion from the current validated `{{INTEGRATION_BRANCH}}` head.
+5. Record the triggering feature and exact `{{INTEGRATION_BRANCH}}` commit so the automatic trigger runs once for that merge.
 
 Release process:
 
-1. Reconcile release scope.
-2. Confirm included features are complete.
-3. Create release branch from exact validated `{{INTEGRATION_BRANCH}}`.
-4. Run complete release checks.
-5. Open release PR → `{{RELEASE_BRANCH}}`.
-6. Obtain fresh independent {{REVIEWER_VENDOR}} review.
-7. Remediate findings through development agents.
-8. Incorporate every release correction back into `{{INTEGRATION_BRANCH}}`.
-9. Revalidate.
-10. Assign final release PR to `{{HOST_ACCOUNT}}`.
-11. Update {{WORK_TRACKER}}.
-12. Stop before {{RELEASE_BRANCH}} merge.
+1. Enforce every pre-creation requirement above, including controller exclusivity, reconciliation, duplicate and active-promotion checks, and trigger recording.
+2. Reconcile release scope.
+3. Confirm included features are complete.
+4. Create release branch from exact validated `{{INTEGRATION_BRANCH}}`.
+5. Run complete release checks.
+6. Open release PR → `{{RELEASE_BRANCH}}`.
+7. Obtain fresh independent {{REVIEWER_VENDOR}} review.
+8. Remediate findings through development agents.
+9. Incorporate every release correction back into `{{INTEGRATION_BRANCH}}`.
+10. Revalidate.
+11. Assign final release PR to `{{HOST_ACCOUNT}}`.
+12. Update {{WORK_TRACKER}}.
+13. Stop before {{RELEASE_BRANCH}} merge.
 
 Only {{RELEASE_OWNER_NAME}} may merge into `{{RELEASE_BRANCH}}`.
 
@@ -724,13 +769,23 @@ For `PROJECT_QUEUE`, this additionally requires:
 
 - No eligible incomplete feature remains.
 - No task/review/remediation/merge is pending.
+- No release preparation, release review, or release remediation is pending.
+- Every in-scope automatic trigger is either incorporated into the active
+  `RELEASE_READY` promotion or durably recorded for the next promotion.
 - No development agent is active.
 - No approver is active.
 
+An active promotion that has not reached `RELEASE_READY` makes
+`MISSION_COMPLETE` false.
+
 ## `RELEASE_READY`
 
 Release PR has passed required gates and is assigned to `{{HOST_ACCOUNT}}`, awaiting {{RELEASE_OWNER_NAME}}'s {{RELEASE_BRANCH}} merge.
 
+`RELEASE_READY` is globally terminal only when no other in-scope engineering
+action is executable. In `PROJECT_QUEUE`, if another eligible incomplete
+feature remains, continue that work while the ready promotion waits.
+
 ## `GLOBAL_ESCALATION`
 
 All remaining in-scope work is blocked by decisions/access that require {{RELEASE_OWNER_NAME}}.
@@ -778,9 +833,12 @@ Output a machine-readable continuation checkpoint:
 ```text
 ORCHESTRATOR_STATE: CONTINUE
 MISSION_SCOPE:
+RELEASE_PREPARATION_POLICY:
 CONTROLLER_PHASE:
 CURRENT_FEATURE:
 ACTIVE_TASKS:
+ACTIVE_PROMOTION:
+PENDING_PROMOTION_FEATURES:
 BLOCKED_ITEMS:
 LAST_VERIFIED_STATE:
 NEXT_ACTION:

tests/test_core_sweep.py

@@ -1,295 +0,0 @@
-"""Cross-process safety of the missed-sweep watermark claim.
-
-Feature 11's sweep guards its announcement with the watermark; the
-original guard was a read-then-set inside a per-PROCESS asyncio lock,
-so two PROCESSES sharing one SQLite file (the HA integration's
-rollover sweep and the API service's scheduler) could both read the
-old watermark and both announce the same missed batch.  These tests
-pin the fix:
-
-- :meth:`MetaStateDao.claim` is the atomic, database-level primitive:
-  one conditional upsert whose ``rowcount`` says whether THIS caller
-  changed the row.  The first claimant of a value wins; a second
-  claimant of the same value — even on an INDEPENDENT connection to
-  the same file, the shape two processes produce — loses.
-- :func:`run_missed_sweep` gates the announcement on that claim: two
-  independent ``NestQuestDatabase`` wrappers (two processes' shape)
-  racing one seeded database yields exactly one announcer.
-- The documented sweep semantics are unchanged: a same-night rerun
-  announces nothing, and a run after days of downtime announces the
-  accumulated ``[watermark, today)`` window exactly once.
-
-Everything here runs against the bundled HA-free core copy
-(``custom_components.nestquest.core``) with a plain
-``asyncio.to_thread`` executor — no Home Assistant, no API server.
-"""
-from __future__ import annotations
-
-import asyncio
-import datetime
-from types import SimpleNamespace
-
-import pytest
-
-from custom_components.nestquest.core.dao_instances import (
-    QuestInstancesDao,
-)
-from custom_components.nestquest.core.dao_meta import MetaStateDao
-from custom_components.nestquest.core.db import NestQuestDatabase
-from custom_components.nestquest.core.migrations import apply_migrations
-from custom_components.nestquest.core.children import create_child
-from custom_components.nestquest.core.quest_definitions import (
-    create_quest_definition,
-)
-from custom_components.nestquest.core.materialize import materialize
-from custom_components.nestquest.core.recurrence import ScheduleRule
-from custom_components.nestquest.core.sweep import (
-    SWEEP_WATERMARK_KEY,
-    run_missed_sweep,
-)
-
-#: One missed event per unswept past-due instance — the sweep's only
-#: event type.
-EVENT_QUEST_MISSED = "nestquest_quest_missed"
-
-
-async def _executor(fn, *args, **kwargs):
-    """Run a synchronous callable on a worker thread and await it.
-
-    The :class:`NestQuestDatabase` executor contract (the shape of
-    ``hass.async_add_executor_job``), so sqlite3 calls never block the
-    event loop.
-    """
-    return await asyncio.to_thread(fn, *args, **kwargs)
-
-
-async def _open_database(db_path: str) -> NestQuestDatabase:
-    """Open (without migrating) one independent connection wrapper."""
-    database = NestQuestDatabase(_executor)
-    await database.open(db_path)
-    return database
-
-
-async def _seed_missed_household(
-    db_path: str, today: datetime.date
-) -> SimpleNamespace:
-    """Seed a household with two past-due open instances on ``db_path``.
-
-    One child, two quests whose ONLY instances are due YESTERDAY
-    (materialized with ``today`` pinned to yesterday, ``end_date``
-    closing the rules so the later today-walk cannot duplicate them).
-    Seeding uses its OWN connection, opened and closed here, so the
-    connections the tests race with start fresh.
-    """
-    database = await _open_database(db_path)
-    try:
-        await apply_migrations(database)
-        today_iso = today.isoformat()
-        yesterday = today - datetime.timedelta(days=1)
-        yesterday_iso = yesterday.isoformat()
-        rule = ScheduleRule.from_dict(
-            {
-                "rule_type": "daily",
-                "start_date": yesterday_iso,
-                "end_date": yesterday_iso,
-            }
-        )
-        ada = await create_child(database, "Ada", sort_order=0)
-        await create_quest_definition(
-            database, "Stale chore", rule, [ada.id], [("morning", "09:00")]
-        )
-        await create_quest_definition(
-            database, "Old chore", rule, [ada.id], [("evening", "18:30")]
-        )
-        await materialize(database, yesterday_iso, yesterday_iso, today=yesterday)
-        # A quest due TODAY (created after the yesterday walk so it has
-        # no past-due instance): the post-downtime catch-up run's
-        # target — inside [watermark=today, later_today) but not
-        # inside the day-one sweep's window.
-        await create_quest_definition(
-            database,
-            "Recoverable chore",
-            ScheduleRule.from_dict(
-                {"rule_type": "daily", "start_date": today_iso}
-            ),
-            [ada.id],
-            [("afternoon", "17:00")],
-        )
-        await materialize(database, today_iso, today_iso, today=today)
-        instances = QuestInstancesDao(database)
-        rows = await instances.list_by_child_and_date(ada.id, yesterday_iso)
-        assert len(rows) == 2
-        today_rows = await instances.list_by_child_and_date(
-            ada.id, today_iso
-        )
-        assert len(today_rows) == 1
-        return SimpleNamespace(
-            child=ada,
-            instances=rows,
-            recoverable_instance=today_rows[0],
-            yesterday_iso=yesterday_iso,
-            today_iso=today_iso,
-        )
-    finally:
-        await database.close()
-
-
-# --- the claim primitive: atomic at the database level ------------------
-
-
-async def test_claim_first_caller_wins_and_second_same_value_loses(
-    temp_db_path: str,
-) -> None:
-    """claim() is a conditional upsert: first True, same-value second False.
-
-    A different value always wins (an advance re-arms the check), and
-    the stored value follows the winner.
-    """
-    database = await _open_database(temp_db_path)
-    try:
-        await apply_migrations(database)
-        dao = MetaStateDao(database)
-        assert await dao.get(SWEEP_WATERMARK_KEY) is None
-
-        assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is True
-        # Same value again: the row is already there — nothing changed.
-        assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is False
-        assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-01"
-        # A DIFFERENT value is an advance, not a re-claim: it wins.
-        assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is True
-        assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-02"
-        assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is False
-    finally:
-        await database.close()
-
-
-async def test_claim_is_atomic_across_independent_connections(
-    temp_db_path: str,
-) -> None:
-    """Two connections to one file cannot both claim the same value.
-
-    This is the two-process shape: independent sqlite3 connections (one
-    per ``NestQuestDatabase`` wrapper), racing the SAME key on the SAME
-    event loop with no shared lock.  The single conditional upsert is
-    serialized by SQLite itself, so exactly one caller's rowcount is 1.
-    """
-    first = await _open_database(temp_db_path)
-    second = await _open_database(temp_db_path)
-    try:
-        await apply_migrations(first)
-        won = await asyncio.gather(
-            MetaStateDao(first).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
-            MetaStateDao(second).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
-        )
-        assert sorted(won) == [False, True]
-        # One stored row, holding the winner's value.
-        assert await MetaStateDao(first).get(SWEEP_WATERMARK_KEY) == (
-            "2026-09-01"
-        )
-        assert await MetaStateDao(second).get(SWEEP_WATERMARK_KEY) == (
-            "2026-09-01"
-        )
-    finally:
-        await first.close()
-        await second.close()
-
-
-# --- the sweep: the announcement is gated by the atomic claim -----------
-
-
-async def test_two_process_shape_race_exactly_one_announces(
-    temp_db_path: str,
-) -> None:
-    """Two independent connections racing the sweep: ONE announcer.
-
-    Each wrapper carries its own connection AND its own per-process
-    sweep lock, so nothing in-process serialises the two runs — the
-    only guard left is the database-level claim.  Both read the unset
-    watermark and both build the events, but exactly one claim wins:
-    one run returns the missed events, the loser returns [].
-    """
-    today = datetime.datetime.now().astimezone().date()
-    seed = await _seed_missed_household(temp_db_path, today)
-    first = await _open_database(temp_db_path)
-    second = await _open_database(temp_db_path)
-    try:
-        results = await asyncio.gather(
-            run_missed_sweep(first, today=today),
-            run_missed_sweep(second, today=today),
-        )
-        announcers = [r for r in results if r]
-        assert len(announcers) == 1
-        event_type, payload = announcers[0][0]
-        assert event_type == EVENT_QUEST_MISSED
-        assert payload["due_date"] == seed.yesterday_iso
-        # Both instances of the seeded window, one event each.
-        assert len(announcers[0]) == 2
-        losers = [r for r in results if not r]
-        assert losers == [[]]
-        # The watermark holds the run's date for whoever looks next.
-        assert (
-            await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
-            == seed.today_iso
-        )
-    finally:
-        await first.close()
-        await second.close()
-
-
-async def test_same_night_rerun_announces_nothing(temp_db_path: str) -> None:
-    """A second sweep for the same date is an empty no-op (the watermark)."""
-    today = datetime.datetime.now().astimezone().date()
-    await _seed_missed_household(temp_db_path, today)
-    database = await _open_database(temp_db_path)
-    try:
-        first = await run_missed_sweep(database, today=today)
-        assert len(first) == 2
-        second = await run_missed_sweep(database, today=today)
-        assert second == []
-    finally:
-        await database.close()
-
-
-async def test_post_downtime_run_announces_accumulated_window_once(
-    temp_db_path: str,
-) -> None:
-    """After days of downtime the catch-up run sweeps the window once.
-
-    The first run sweeps yesterday's instances (watermark := today);
-    ``today`` then jumps two days and the catch-up run — on a
-    DIFFERENT connection, the two-process shape — announces the
-    accumulated ``[watermark, later_today)`` window exactly once, and
-    a rerun on the original connection is silent again.
-    """
-    today = datetime.datetime.now().astimezone().date()
-    later_today = today + datetime.timedelta(days=2)
-    seed = await _seed_missed_household(temp_db_path, today)
-    first = await _open_database(temp_db_path)
-    second = await _open_database(temp_db_path)
-    try:
-        # Day one: yesterday's two instances fire once.
-        day_one = await run_missed_sweep(first, today=today)
-        assert len(day_one) == 2
-        assert all(
-            payload["due_date"] == seed.yesterday_iso
-            for _, payload in day_one
-        )
-
-        # The downtime catch-up: the window is [watermark, later_today).
-        catchup = await run_missed_sweep(second, today=later_today)
-        assert len(catchup) == 1
-        event_type, payload = catchup[0]
-        assert event_type == EVENT_QUEST_MISSED
-        assert payload["due_date"] == seed.today_iso
-        assert payload["instance_id"] == seed.recoverable_instance.id
-        assert payload["child_id"] == seed.child.id
-
-        # A further rerun announces nothing.
-        assert await run_missed_sweep(first, today=later_today) == []
-        assert (
-            await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
-            == later_today.isoformat()
-        )
-    finally:
-        await first.close()
-        await second.close()