Merge origin/dev into feature/cd03735b-2498-4b34-b84c-c1cb2b271ba7-nestquest-api-service

Co-authored-by: Joshua <joshua@cnjmail.com>
Co-committed-by: Joshua <joshua@cnjmail.com>

.opencode/agents/approver.md

@@ -0,0 +1,32 @@
+---
+description: Independently verify NestQuest merge gates and merge approved heads
+mode: primary
+model: openrouter/deepseek/deepseek-v4.1-flash
+permission:
+  edit: deny
+  write: deny
+  task: deny
+  bash:
+    "*": deny
+    "/home/overseer/.local/share/nestquest-controller/scripts/nq-approve-merge *": allow
+  external_directory:
+    "*": ask
+    "/tmp/opencode/*": allow
+    "/home/overseer/.local/share/nestquest-controller/*": allow
+---
+
+You are the NestQuest feature-to-dev approver, separate from the controller and
+developer sessions. Verify the current PR head equals the independently
+reviewed and approved commit,
+required checks pass, blocking findings are resolved, branch protection is
+satisfied, and Maestro has accurate links. Use only the approver Gitea identity
+for the merge. Do not change code or approve a stale head. Use only the
+installed nq-approve-merge launcher for Gitea inspection and merging. Merge
+only into dev; never merge into main. If any gate is
+missing, refuse the merge and give the exact missing evidence. Report the PR,
+reviewed head, resulting merge commit, and target branch.
+
+Broker syntax: `nq-approve-merge inspect PR_NUMBER` or
+`nq-approve-merge merge PR_NUMBER REVIEWED_HEAD REVIEW_FILE`. Run one broker
+command per Bash call. The broker validates the current PR and review again
+before merging; do not substitute an ad hoc Gitea command.

.opencode/agents/developer.md

@@ -0,0 +1,21 @@
+---
+description: Implement one assigned NestQuest task in an isolated worktree
+mode: primary
+model: openrouter/z-ai/glm-5.3-flash
+permission:
+  task: deny
+  external_directory:
+    "*": ask
+    "/tmp/opencode/*": allow
+---
+
+You are the NestQuest development agent. Implement only the assigned Maestro
+task in the supplied worktree. Follow the task's acceptance criteria and the
+repository's instructions. Run relevant tests, commit on the task branch, and
+report the commit and test results. Keep changes inside the assigned worktree.
+Do not approve your own pull request. You may merge your task PR into the
+feature branch only after the independent reviewer approved its exact current
+head and the controller verified every task merge gate. Never merge a feature
+into dev or a release into main. Do not start another task, change role, or
+use another model. Do not read or print credential files except when the
+approved task merge requires the existing Gitea development credential.

.opencode/agents/orchestrator.md

@@ -0,0 +1,60 @@
+---
+description: Reconcile NestQuest work and dispatch the next verified action
+mode: primary
+model: openrouter/deepseek/deepseek-v4.1-flash
+permission:
+  doom_loop: deny
+  task: deny
+  external_directory:
+    "*": ask
+    "/tmp/opencode/*": allow
+    "/home/overseer/.local/share/opencode/tool-output/*": allow
+---
+
+You are the NestQuest controller. Work in the NestQuest repository. Maestro is
+the source of truth for feature and task state; CTXD is the source of truth for
+project instructions. At the start of every invocation, read the current
+NestQuest context in CTXD, including OPENCODE-ORCHESTRATOR-PROMPT.MD, then
+reconcile it with Maestro, Git, pull requests, and checks before dispatching.
+Follow the review, identity, and branch rules in that prompt.
+If the CTXD project or prompt is unavailable, do not dispatch or merge work;
+report the missing source and end with ORCHESTRATOR_STATE: BLOCKED.
+
+One invocation is a controller tick, not an entire feature lifetime. Prefer a
+state-changing action over another identical status query. Never repeat an
+unchanged tool call three times. If work is asynchronous, wait or inspect a
+specific changed result. Do not claim that you are continuing while ending
+the turn without naming the next action.
+
+The NESTQUEST_CONTROL_HOME environment variable names the installed launchers
+(normally /home/overseer/.local/share/nestquest-controller).
+For implementation, run "$NESTQUEST_CONTROL_HOME/scripts/nq-agent" with role
+developer, a task worktree, title, and task prompt file. For independent
+review, run "$NESTQUEST_CONTROL_HOME/scripts/nq-review" with the worktree,
+feature base ref, exact head ref, review prompt file, PR number, and an output
+file under /tmp/opencode. It publishes an exact-head Gitea approval only after
+Claude approves. The developer may merge
+a task PR into its feature branch only after independent approval of the exact
+head and verification of every task gate. For feature-to-dev approval and
+merge, run "$NESTQUEST_CONTROL_HOME/scripts/nq-agent" with role approver in
+a separate session after the independent review. Save the Claude review output
+in a file under /tmp/opencode and give that file and its reviewed head to the
+approver. Never substitute a different model or reviewer silently.
+
+When a feature is integrated into dev, reconcile Maestro, then immediately
+select and dispatch the next eligible feature. Feature completion is a
+transition, not the end of the controller's job. A release awaiting Joshua's
+main merge does not prevent work on independent eligible features.
+
+End each invocation with exactly one final line using one of these values:
+
+ORCHESTRATOR_STATE: CONTINUE
+ORCHESTRATOR_STATE: QUIESCENT
+ORCHESTRATOR_STATE: BLOCKED
+
+Use CONTINUE when there is any eligible or in-flight work; include the feature,
+task, and next concrete action immediately above the final line. Use QUIESCENT
+only after verifying no eligible or in-flight work remains in Maestro and the
+repository. Use BLOCKED only when a specific decision or access is required;
+include evidence and the needed human action. If you cannot establish the
+state, use BLOCKED. Never emit a success claim based only on a child's summary.

.opencode/agents/terminal-orchestrator.md

@@ -0,0 +1,31 @@
+---
+description: Run the NestQuest orchestration loop in this foreground terminal
+mode: primary
+model: openrouter/deepseek/deepseek-v4.1-flash
+permission:
+  doom_loop: deny
+  task: deny
+  external_directory:
+    "*": ask
+    "/tmp/opencode/*": allow
+    "/home/overseer/.local/share/opencode/tool-output/*": allow
+---
+
+You are the foreground NestQuest orchestrator. The user explicitly launched
+this session to run the development queue now. Read the supplied terminal
+orchestrator prompt and current CTXD project context, but follow the
+FOREGROUND TERMINAL RUNTIME for this session instead of CTXD's service runtime
+section.
+
+Continuously reconcile Maestro, CTXD, Git, pull requests, checks, and child
+agent results. Execute the next useful transition in this same session. Do
+not finish after dispatch, review, remediation, or merge while eligible work
+remains. Use the installed role launchers for developer, reviewer, and
+approver isolation. Never start or wait for the systemd controller service.
+Never repeat an identical unchanged tool call three times; wait on the
+specific child or inspect newly changed evidence.
+
+End only at a global terminal condition, an explicit user stop, a genuine
+global access/decision block with no other eligible work, or a host-forced
+yield. On a forced yield, return a continuation checkpoint that can be
+resumed in the same OpenCode session.

.opencode/commands/start-development.md

@@ -0,0 +1,11 @@
+---
+description: Run the NestQuest orchestrator now in this OpenCode terminal
+agent: terminal-orchestrator
+---
+
+Run the NestQuest project queue now in this foreground OpenCode session.
+Read the current CTXD prompt and Maestro state, but use the FOREGROUND
+TERMINAL RUNTIME instead of the service runtime. Reconcile repository and
+pull-request state, then execute CONTROLLER_TICK repeatedly until a global
+terminal condition, explicit user stop, genuine global block, or host-forced
+yield. Do not start or wait for the systemd controller.

ORCHESTRATION.md

@@ -0,0 +1,77 @@
+# NestQuest foreground orchestrator
+
+OpenCode 1.18.32 loads the named agents from `.opencode/agents/`. Their
+models match the current NestQuest CTXD prompt: DeepSeek V4.1 Flash for the
+foreground orchestrator and approver, and GLM 5.3 Flash for development.
+Run orchestration directly from a shell:
+
+```bash
+nq-orchestrate
+nq-orchestrate mission.md
+cat mission.md | nq-orchestrate -
+nq-orchestrate --print-prompt > nestquest-orchestrator.md
+```
+
+The launcher renders the project prompt with the foreground terminal runtime,
+stops the dormant legacy service, refuses immediately when another NestQuest
+TUI is open, and then replaces itself with OpenCode using
+`terminal-orchestrator`. There is no background polling process and no wait
+for a controller tick. Exit or interrupt the TUI to stop the foreground run;
+resume that OpenCode session when a host-forced yield needs continuation.
+
+`scripts/nq-agent` launches developer and approver sessions with fixed agent
+definitions and no `--auto` flag. Headless OpenCode runs use `--pure` so the
+global idle-notification plugin does not notify on every controller tick.
+The approver's shell permission permits only the installed
+`nq-approve-merge` broker for Gitea. That broker requires exact-head
+Claude approval evidence, an exact-head Gitea approval by `review_agent`,
+the `dev` target, a mergeable PR, and passing configured checks before it
+requests a squash merge with Gitea's atomic `head_commit_id` guard.
+`scripts/nq-review` launches the independent
+Claude review in a noninteractive read-only session in the task worktree,
+checks the head again afterward, saves
+the output under `/tmp/opencode`, and posts the Gitea approval.
+The approver verifies the reviewed head again before a merge.
+
+The terminal launcher uses a stable copy of this configuration and scripts
+at `/home/overseer/.local/share/nestquest-controller`; it does not depend on
+which task branch is checked out in the primary NestQuest directory. Install
+that copy, run `scripts/nq-controller --check`, and verify the OpenCode
+agents (terminal-orchestrator, developer, and approver)
+with `opencode debug agent <name>` while `OPENCODE_CONFIG_DIR` points at the
+installed `.opencode` directory. Install `scripts/nq-orchestrate` in
+`~/.local/bin/`. For the global slash command, also install
+`.opencode/agents/terminal-orchestrator.md` under
+`~/.config/opencode/agents/` and
+`deploy/opencode-global-commands/start-nestquest-development.md` under
+`~/.config/opencode/commands/`. The old `nestquest-controller.service` remains static and
+inactive only for migration safety; foreground launch stops it.
+
+Inside an already-open OpenCode TUI, `/start-development` (or the global
+`/start-nestquest-development`) switches to the same foreground
+`terminal-orchestrator` behavior immediately. The shell launcher is the
+preferred entry point when supplying a mission prompt file.
+
+`templates/ORCHESTRATOR-PROMPT.template.md` and
+`templates/nestquest.profile.json` make the current CTXD cycle prompt
+adaptable to another project. See `templates/README.md` before publishing a
+rendered prompt for a new project.
+
+The CTXD preflight reads the configured HTTP/MCP server's NestQuest prompt.
+It deliberately does not inspect `/home/overseer/.ctx/ctxd.db`, which belongs
+to a different local store and lacks this project.
+
+All four agent credentials are currently readable by the same Unix account.
+The Gitea review and merge gates enforce normal workflow sequencing, but
+they are not a hard isolation boundary against a process running as that
+account. Strong isolation requires separate OS identities and protected
+credentials for the controller, reviewer, and approver.
+
+The terminal runtime repeatedly executes CONTROLLER_TICK within the foreground
+session. Neither a feature merge nor a progress report ends the run while
+eligible work remains. A host-forced yield returns a continuation checkpoint
+that can be resumed in the same OpenCode session.
+
+Do not put tokens in this repository or in unit files. Existing OpenCode MCP
+connections supply CTXD and Maestro. Rotate those MCP tokens as a coordinated
+operation across every consumer before revoking the old values.

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

api/nestquest_core.py

@@ -116,6 +116,11 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
 #: ``uncomplete_instance``) the panel's write routes call.
 core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion")
 
+#: The read-only history layer (``core.history.query_history``,
+#: ``export_history_csv``, ``rows_to_csv``) the admin plane's history
+#: query and CSV export routes call.
+core_history = importlib.import_module(f"{_CORE_MODULE_NAME}.history")
+
 #: The transition-event builders (``core.events.build_quest_completed_event``
 #: et al.) the panel's write routes build their SSE events with.
 core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
@@ -124,6 +129,17 @@ core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
 #: the snapshot builder's ``settings`` parameter takes.
 core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
 
+#: The durable settings store (``core.settings_store.load_settings`` /
+#: ``update_settings``) the admin plane's settings routes read and
+#: write — the API's OWN settings source: a validated JSON document in
+#: the existing ``nestquest_meta_state`` table, resolved through the
+#: same core settings validators (the integration's HA config-entry
+#: options remain a temporary separate source until Feature 18 wires
+#: the client to the API).
+core_settings_store = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.settings_store"
+)
+
 #: The children business layer (``core.children.create_child`` et al.)
 #: the admin plane's children routes call.
 core_children = importlib.import_module(f"{_CORE_MODULE_NAME}.children")
@@ -142,15 +158,71 @@ core_quest_definitions = importlib.import_module(
 #: re-implement rule validation.
 core_recurrence = importlib.import_module(f"{_CORE_MODULE_NAME}.recurrence")
 
+#: The presence model (``core.presence.PresenceEngine``,
+#: ``PresenceSchedule``, ``PresenceOverride``) the admin plane's
+#: presence routes serialize and the tests resolve presence with.
+core_presence = importlib.import_module(f"{_CORE_MODULE_NAME}.presence")
+
+#: The presence-write business layer
+#: (``core.presence_management.set_presence_schedule`` et al.) the
+#: admin plane's presence routes call — the ONE implementation the HA
+#: services share.
+core_presence_management = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.presence_management"
+)
+
+#: The typed presence DAOs (``core.dao_presence``) the admin-plane tests
+#: read stored schedules and overrides back through.
+core_dao_presence = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.dao_presence"
+)
+
+#: The materialization walk (``core.materialize.materialize``,
+#: ``regenerate_for_definition``, ``regenerate_for_child``) the admin
+#: plane's regenerate route calls — the same walk the integration's
+#: config/presence changes trigger.
+core_materialize = importlib.import_module(f"{_CORE_MODULE_NAME}.materialize")
+
+#: The core constants module (``core.const.DEFAULT_HORIZON_DAYS``) the
+#: regenerate route sizes the household rolling horizon with.
+core_const = importlib.import_module(f"{_CORE_MODULE_NAME}.const")
+
+#: The typed child DAO (``core.dao_children.ChildrenDao.get``) the
+#: regenerate route reads a child's existence with before delegating.
+core_dao_children = importlib.import_module(
+    f"{_CORE_MODULE_NAME}.dao_children"
+)
+
+#: The typed definition DAO (``core.dao_rules.QuestDefinitionsDao.get``)
+#: the regenerate route reads a definition's existence with before
+#: 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",
     "core_completion",
+    "core_const",
+    "core_dao_children",
+    "core_dao_presence",
+    "core_dao_rules",
     "core_db",
     "core_events",
+    "core_history",
+    "core_materialize",
     "core_migrations",
+    "core_presence",
+    "core_presence_management",
     "core_quest_definitions",
     "core_recurrence",
     "core_snapshot",
     "core_settings",
+    "core_settings_store",
+    "core_sweep",
 ]

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"]

custom_components/nestquest/core/history.py

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

custom_components/nestquest/core/presence_management.py

@@ -0,0 +1,161 @@
+"""Business layer for presence writes (Feature 09, API admin plane).
+
+Sits between the write callers (the Feature 09 HA service handlers and
+the API admin plane's presence routes) and the typed presence DAOs plus
+the pure presence model: callers get the model validation, the upsert /
+create / delete operations and the regeneration side effect here, and
+never touch the DAO or SQL directly.  This is the ONE implementation of
+these business rules — every caller constructs the same
+:class:`~.presence.PresenceSchedule` / :class:`~.presence.PresenceOverride`
+models (whose constructors are total, so no re-validation happens in a
+handler) and gets the same post-write
+:func:`~.materialize.regenerate_for_child`, so a child's future quest
+instances track its changed presence no matter which plane issued the
+write.  Like the model it is Home-Assistant-free: ``today`` is a plain
+calendar date threaded through to the regeneration, defaulting to the
+host clock there.
+
+Upsert policy (Feature 05, unchanged): a child has AT MOST ONE presence
+schedule — setting again REPLACES the existing row rather than
+duplicating it.  Overrides are deletable by design (children are not —
+the children layer owns that guardrail).  Deleting a nonexistent
+override raises ValueError prefixed ``override_id:`` so a caller can
+distinguish "unknown id" from a rejected argument the same way the
+quest-definitions layer prefixes ``definition_id:``.
+"""
+from __future__ import annotations
+
+import datetime
+
+from .dao_presence import (
+    PresenceOverrideRecord,
+    PresenceOverridesDao,
+    PresenceSchedulesDao,
+)
+from .db import NestQuestDatabase
+from .materialize import regenerate_for_child
+from .presence import PresenceOverride, PresenceSchedule
+
+
+def _validate_override_id(value: object) -> int:
+    """Reject non-int override ids (bools would silently address id 1)."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"override_id must be an integer, got {value!r}")
+    return value
+
+
+async def set_presence_schedule(
+    database: NestQuestDatabase,
+    child_id: int,
+    cycle_length_weeks: int,
+    anchor_date: object,
+    pattern: dict[int, object],
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceSchedule:
+    """Set (upsert) a child's repeating presence schedule.
+
+    The :class:`~.presence.PresenceSchedule` model validates at
+    construction — cycle length 1..4, a strict calendar anchor date,
+    and a pattern covering week indices 0..n-1 with weekday sets — so
+    no invalid schedule reaches the DAO.  The encoded pattern is
+    UPSERTed by child (a second set REPLACES the child's existing
+    schedule; the schema's UNIQUE(child_id) makes duplication
+    unrepresentable), then :func:`~.materialize.regenerate_for_child`
+    re-materializes the child's future instances against the new
+    presence.  ``today``/``horizon_days`` thread through to that
+    regeneration; raising ValueError names what to fix (an unknown
+    child reads ``child N does not exist``).
+
+    Returns the STORED schedule decoded back through the model — the
+    round-trip proves the encoded pattern the database now holds is
+    exactly the one the caller asked for.
+    """
+    schedule = PresenceSchedule(
+        child_id, cycle_length_weeks, anchor_date, pattern
+    )
+    record = await PresenceSchedulesDao(database).upsert_by_child(
+        schedule.child_id,
+        schedule.cycle_length_weeks,
+        schedule.anchor_date.isoformat(),
+        schedule.encode(),
+    )
+    await regenerate_for_child(
+        database, schedule.child_id, today=today, horizon_days=horizon_days
+    )
+    return PresenceSchedule.decode(
+        record.child_id, record.anchor_date, record.pattern
+    )
+
+
+async def create_presence_override(
+    database: NestQuestDatabase,
+    child_id: int,
+    start_date: object,
+    end_date: object,
+    is_present: bool,
+    note: str | None = None,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceOverrideRecord:
+    """Create one date-range presence override and return it as stored.
+
+    The :class:`~.presence.PresenceOverride` model validates at
+    construction — strict calendar dates, ``end_date`` on or after
+    ``start_date``, a real ``is_present`` bool, a string-or-None note —
+    then the DAO insert applies the same-child no-overlap policy (a
+    conflicting range raises ValueError naming the existing override)
+    and :func:`~.materialize.regenerate_for_child` re-materializes the
+    child's future instances against the override's presence answer.
+    An unknown child reads ``child N does not exist``.
+
+    Returns the stored record WITH its ``id`` — the handle a later
+    :func:`delete_presence_override` takes.
+    """
+    override = PresenceOverride(
+        child_id, start_date, end_date, is_present, note=note
+    )
+    record = await PresenceOverridesDao(database).create(
+        override.child_id,
+        override.start_date.isoformat(),
+        override.end_date.isoformat(),
+        override.is_present,
+        note=override.note,
+    )
+    await regenerate_for_child(
+        database, override.child_id, today=today, horizon_days=horizon_days
+    )
+    return record
+
+
+async def delete_presence_override(
+    database: NestQuestDatabase,
+    override_id: int,
+    *,
+    today: datetime.date | None = None,
+    horizon_days: int | None = None,
+) -> PresenceOverrideRecord:
+    """Delete one presence override by id and return it as it was stored.
+
+    The override is looked up FIRST and an unknown id raises ValueError
+    prefixed ``override_id:`` (naming the id) BEFORE any delete, so a
+    mistyped id can never be silently swallowed by a no-op.  The
+    deletion is followed by :func:`~.materialize.regenerate_for_child`
+    for the override's own child — the id decides the child, never a
+    caller-supplied one.  ``today``/``horizon_days`` thread through to
+    that regeneration.
+    """
+    _validate_override_id(override_id)
+    dao = PresenceOverridesDao(database)
+    override = await dao.get(override_id)
+    if override is None:
+        raise ValueError(
+            f"override_id: presence override {override_id} does not exist"
+        )
+    await dao.delete(override_id)
+    await regenerate_for_child(
+        database, override.child_id, today=today, horizon_days=horizon_days
+    )
+    return override

custom_components/nestquest/core/settings_store.py

@@ -0,0 +1,210 @@
+"""Durable settings store for the NestQuest API service.
+
+The API service has NO Home Assistant config entry, so it cannot build
+:class:`~.settings.NestQuestSettings` from ``entry.options`` the way the
+integration does.  This module gives the API its own durable settings
+source: a JSON document kept in the EXISTING ``nestquest_meta_state``
+key/value table (:class:`~.dao_meta.MetaStateDao`) under the well-known
+key :data:`SETTINGS_META_KEY` — no schema change, no new table —
+validated through the SAME field table the settings object is built
+from (:data:`~.settings._FIELDS`, the table driving
+``from_options``/``from_options_resilient``).  No validation rule is
+re-implemented here.
+
+OWNERSHIP: the API now owns settings in the database.  The
+integration's Home-Assistant config-entry options remain a temporary,
+SEPARATE source until Feature 18 converts the integration into a thin
+API client; the two are deliberately NOT unified here, and neither
+side reads the other's values.
+
+STORED SERIALIZATION — the document at ``SETTINGS_META_KEY`` is a JSON
+object holding exactly the TEN settings fields, keyed by field name
+(which equals the ``CONF_*`` option key for every field), complete
+after the first update:
+
+- ``horizon_days`` — int >= 1,
+- ``day_rollover_time``, ``morning_summary_time``,
+  ``afternoon_reminder_time``, ``end_of_day_report_time`` — strict
+  24-hour ``HH:MM`` strings,
+- ``notify_target`` — a string (a padded value is stripped by the
+  core's normalizer),
+- ``morning_summary_enabled``, ``afternoon_reminder_enabled``,
+  ``end_of_day_report_enabled``, ``celebration_enabled`` — real
+  booleans.
+
+Only values whose core validators ACCEPTED are ever written, so a
+stored document always round-trips through the core's option builders.
+An absent key (a fresh database) resolves to the all-defaults
+settings, and a corrupt document degrades to defaults or per-field
+defaults rather than crashing the service (see :func:`load_settings`).
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import Mapping
+from typing import Any
+
+from .const import LOGGER
+from .dao_meta import MetaStateDao
+from .db import NestQuestDatabase
+from .settings import NestQuestSettings, _FIELDS
+
+#: The well-known ``nestquest_meta_state`` key holding the settings
+#: document.  One key, one JSON object — the same table the missed
+#: sweep's watermark already uses, so no migration is needed.
+SETTINGS_META_KEY = "settings"
+
+#: The per-field resolution table keyed by field name: each row is the
+#: ``(default, validator)`` pair the settings module's ``_FIELDS`` table
+#: carries for that field.  ``update_settings`` resolves each PROVIDED
+#: field through its row's validator — the SAME core validators
+#: ``from_options`` applies — and never re-implements one here.
+#: (Field names equal the ``CONF_*`` option keys, so one dict serves
+#: both the API's field names and the stored document's keys.)
+_RESOLUTIONS: dict[str, tuple[Any, Any]] = {
+    key: (default, validator) for _field, key, default, validator in _FIELDS
+}
+
+#: The known field names, in the settings module's field order — the
+#: allowed keys of an update body and the stored document's key set.
+_FIELD_NAMES: tuple[str, ...] = tuple(
+    key for _field, key, _default, _validator in _FIELDS
+)
+
+#: One asyncio.Lock per (connection wrapper, running loop), the missed
+#: sweep's pattern: the whole stored-document read → merge → write span
+#: is one critical section, so two concurrent updates queue instead of
+#: one losing the other's change — the loser re-reads the document
+#: INSIDE the lock and merges onto the winner's settings.
+_SETTINGS_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
+
+
+def _settings_lock(database: NestQuestDatabase) -> asyncio.Lock:
+    """Return the settings lock bound to ``database``'s running loop."""
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _SETTINGS_LOCKS.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _SETTINGS_LOCKS[key] = lock
+    return lock
+
+
+async def _load_document(database: NestQuestDatabase) -> dict[str, Any]:
+    """Read and decode the stored settings document (empty when unusable).
+
+    An absent key — a fresh database that never updated settings — and
+    an unparseable value (invalid JSON, or JSON that is not an object)
+    return an empty mapping, which the option builders resolve to the
+    all-defaults settings; the corrupt cases log a warning first.  A
+    parseable object is returned as stored (its per-field validity is
+    the builder's concern, not this reader's).
+    """
+    raw = await MetaStateDao(database).get(SETTINGS_META_KEY)
+    if raw is None:
+        return {}
+    try:
+        document = json.loads(raw)
+    except ValueError:
+        LOGGER.warning(
+            "NestQuest stored settings (meta_state key %r) are not valid "
+            "JSON; falling back to the default settings",
+            SETTINGS_META_KEY,
+        )
+        return {}
+    if not isinstance(document, dict):
+        LOGGER.warning(
+            "NestQuest stored settings (meta_state key %r) are not a JSON "
+            "object; falling back to the default settings",
+            SETTINGS_META_KEY,
+        )
+        return {}
+    return document
+
+
+def _named(field: str, error: ValueError) -> str:
+    """Return a field-named message for a rejected validator value.
+
+    The horizon, time and notify-target validators already name their
+    field in the message; the boolean validator's message is generic
+    (``expected a boolean, got ...``), so it gets the field prefixed.
+    Every rejected update's error therefore names its field either way.
+    """
+    message = str(error)
+    if message.startswith(f"{field} "):
+        return message
+    return f"{field}: {message}"
+
+
+async def load_settings(database: NestQuestDatabase) -> NestQuestSettings:
+    """Return the current effective settings stored for ``database``.
+
+    Reads :data:`SETTINGS_META_KEY` and resolves the stored document
+    through the core settings validators, with each absent field
+    falling back to its default.  An absent key and an unparseable
+    document fall back to the all-defaults settings (via
+    :func:`_load_document`), and a parseable document with a single
+    invalid FIELD falls back PER FIELD — through
+    :meth:`NestQuestSettings.from_options_resilient`, the core's own
+    resilient builder — so a corrupt sibling never resets the stored
+    values around it.  This never raises: the service must keep
+    serving reads even over a corrupt store.
+    """
+    document = await _load_document(database)
+    return NestQuestSettings.from_options_resilient(document)
+
+
+async def update_settings(
+    database: NestQuestDatabase, changes: Mapping[str, Any]
+) -> NestQuestSettings:
+    """Validate ``changes`` and persist them as the effective settings.
+
+    ``changes`` maps settings field names to new values.  EVERY name is
+    checked against the known field table BEFORE anything is validated
+    or written: an unknown name raises a ``ValueError`` naming it (and
+    the known names), leaving the stored settings untouched.  Each
+    PROVIDED field is then resolved from_options-style — its
+    :data:`~.settings._FIELDS` validator applied, with a JSON ``null``
+    mapping to the field's default (the settings module's uniform
+    None-as-absent rule) — and an invalid value raises a ``ValueError``
+    naming the field, also BEFORE anything is written, so a rejected
+    update can never leave a half-applied document behind.
+
+    The validated changes are merged onto the current effective
+    settings and the merged (complete, ten-field) document is persisted
+    in one atomic upsert; the updated settings are returned.  The whole
+    read → merge → write span runs inside the per-database settings
+    lock, so concurrent updates queue instead of interleaving.
+    """
+    if not isinstance(changes, Mapping):
+        raise ValueError(
+            "settings changes must be a JSON object mapping field names "
+            "to values"
+        )
+    unknown = sorted(
+        str(name) for name in changes if name not in _RESOLUTIONS
+    )
+    if unknown:
+        listed = ", ".join(repr(name) for name in unknown)
+        known = ", ".join(_FIELD_NAMES)
+        raise ValueError(
+            f"unknown settings field(s): {listed}; known fields: {known}"
+        )
+    async with _settings_lock(database):
+        current = await load_settings(database)
+        document: dict[str, Any] = {
+            field: getattr(current, field) for field in _FIELD_NAMES
+        }
+        for name, raw_value in changes.items():
+            default, validator = _RESOLUTIONS[name]
+            try:
+                document[name] = validator(
+                    default if raw_value is None else raw_value
+                )
+            except ValueError as error:
+                raise ValueError(_named(name, error)) from error
+        await MetaStateDao(database).set(
+            SETTINGS_META_KEY, json.dumps(document, sort_keys=True)
+        )
+    return NestQuestSettings(**document)

custom_components/nestquest/core/sweep.py

@@ -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

custom_components/nestquest/presence_management.py

@@ -0,0 +1,10 @@
+"""Re-export of :mod:`custom_components.nestquest.core.presence_management` for the integration package.
+
+The presence-write business layer lives in
+:mod:`custom_components.nestquest.core.presence_management` (added for
+task e4a9d31b); this module re-exports it so the integration's service
+handlers call the ONE implementation the API admin plane shares.
+"""
+from __future__ import annotations
+
+from .core.presence_management import *  # noqa: F401,F403

custom_components/nestquest/services.py

@@ -4,9 +4,11 @@ Registration is domain-scoped and happens once (see
 :func:`async_register_services`); handlers resolve a live database at
 call time through the injected ``find_runtime`` callable.  Permission
 checks are Feature 09 later tasks — this module only validates payload
-shape and forwards to the Feature 03/06/08 business layers (and the
-presence DAOs).  ``export_history_csv`` is registered with a schema but
-raises until Feature 13.  ``regenerate`` is supplied by the caller so
+shape and forwards to the Feature 03/06/08 business layers (the
+presence writes included, through :mod:`.presence_management` — the
+ONE implementation the API admin plane shares).
+``export_history_csv`` is registered with a schema but raises until
+Feature 13.  ``regenerate`` is supplied by the caller so
 the Feature 07 handler stays the one in ``__init__``.
 """
 from __future__ import annotations
@@ -22,6 +24,7 @@ from homeassistant.exceptions import HomeAssistantError
 
 from . import children as children_layer
 from . import completion as completion_layer
+from . import presence_management as presence_management_layer
 from . import quest_definitions as quest_definitions_layer
 from .const import (
     DOMAIN,
@@ -39,11 +42,8 @@ from .const import (
     SERVICE_UNCOMPLETE_QUEST,
     SERVICE_UPDATE_QUEST_DEFINITION,
 )
-from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
 from .events import fire_quest_completed, fire_quest_uncompleted
-from .materialize import regenerate_for_child
 from .permissions import permission_gate
-from .presence import PresenceSchedule
 from .recurrence import ScheduleRule
 from .service_policy import SERVICE_POLICY
 
@@ -405,21 +405,12 @@ def _set_presence_pattern(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         data = call.data
-        schedule = PresenceSchedule(
+        await presence_management_layer.set_presence_schedule(
+            runtime.database,
             data["child_id"],
             data["cycle_length_weeks"],
             data["anchor_date"],
             data["pattern"],
-        )
-        await PresenceSchedulesDao(runtime.database).upsert_by_child(
-            schedule.child_id,
-            schedule.cycle_length_weeks,
-            schedule.anchor_date.isoformat(),
-            schedule.encode(),
-        )
-        await regenerate_for_child(
-            runtime.database,
-            schedule.child_id,
             **_generation_kwargs(hass, runtime),
         )
 
@@ -432,16 +423,13 @@ def _create_presence_override(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         data = call.data
-        await PresenceOverridesDao(runtime.database).create(
+        await presence_management_layer.create_presence_override(
+            runtime.database,
             data["child_id"],
             data["start_date"],
             data["end_date"],
             data["is_present"],
             note=data.get("note"),
-        )
-        await regenerate_for_child(
-            runtime.database,
-            data["child_id"],
             **_generation_kwargs(hass, runtime),
         )
 
@@ -453,17 +441,9 @@ def _delete_presence_override(
 ) -> ServiceHandler:
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
-        dao = PresenceOverridesDao(runtime.database)
-        override_id = call.data["override_id"]
-        override = await dao.get(override_id)
-        if override is None:
-            raise HomeAssistantError(
-                f"override_id: presence override {override_id} does not exist"
-            )
-        await dao.delete(override_id)
-        await regenerate_for_child(
+        await presence_management_layer.delete_presence_override(
             runtime.database,
-            override.child_id,
+            call.data["override_id"],
             **_generation_kwargs(hass, runtime),
         )
 

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

deploy/nestquest-controller.service

@@ -0,0 +1,14 @@
+[Unit]
+Description=NestQuest OpenCode controller
+
+[Service]
+Type=simple
+WorkingDirectory=/mnt/ai-storage/projects/NestQuest
+ExecStartPre=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller --check
+ExecStart=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller
+Restart=on-failure
+RestartSec=60s
+RestartPreventExitStatus=77
+TimeoutStartSec=0
+KillMode=control-group
+Environment=PYTHONUNBUFFERED=1

deploy/opencode-global-commands/start-nestquest-development.md

@@ -0,0 +1,11 @@
+---
+description: Run the NestQuest orchestrator now in this OpenCode terminal
+agent: terminal-orchestrator
+---
+
+Run the NestQuest project queue now in this foreground OpenCode session.
+Read the current CTXD prompt and Maestro state, but use the FOREGROUND
+TERMINAL RUNTIME instead of the service runtime. Reconcile repository and
+pull-request state, then execute CONTROLLER_TICK repeatedly until a global
+terminal condition, explicit user stop, genuine global block, or host-forced
+yield. Do not start or wait for the systemd controller.

orchestration_tests/test_cycle_prompt.py

@@ -0,0 +1,56 @@
+import json
+from pathlib import Path
+import runpy
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+RENDER = runpy.run_path(str(ROOT / "scripts/render-cycle-prompt"))["render"]
+TEMPLATE = (ROOT / "templates/ORCHESTRATOR-PROMPT.template.md").read_text()
+PROFILE = json.loads((ROOT / "templates/nestquest.profile.json").read_text())
+
+
+class CyclePromptTests(unittest.TestCase):
+    def test_project_profile_renders_without_specific_source_names(self):
+        other = PROFILE.copy()
+        other.update({
+            "PROJECT_NAME": "ExampleQuest",
+            "PROJECT_CONTEXT_LABEL": "EXAMPLEQUEST COORDINATION",
+            "WORK_TRACKER": "TaskGrid",
+            "CONTEXT_STORE": "ContextHub",
+            "INTEGRATION_BRANCH": "integration",
+            "RELEASE_BRANCH": "production",
+            "CONTROL_HOME": "/opt/examplequest/controller",
+            "SERVICE_NAME": "examplequest-controller.service",
+            "PROJECT_ROOT": "/srv/examplequest",
+            "TERMINAL_LAUNCHER": "eq-orchestrate",
+            "FEATURE_BRANCH_EXAMPLE": "feature/EQ-17",
+        })
+        output = RENDER(TEMPLATE, other)
+        self.assertIn("ExampleQuest Persistent Autonomous", output)
+        self.assertIn("TaskGrid", output)
+        self.assertIn("ContextHub", output)
+        self.assertIn("feature/EQ-17", output)
+        self.assertNotIn("NestQuest", output)
+        self.assertNotIn("nestquest", output)
+        self.assertNotIn("Maestro", output)
+        self.assertNotIn("CTXD", output)
+        self.assertNotIn("{{", output)
+
+    def test_missing_value_fails_before_publication(self):
+        other = PROFILE.copy()
+        del other["PROJECT_NAME"]
+        with self.assertRaisesRegex(ValueError, "PROJECT_NAME"):
+            RENDER(TEMPLATE, other)
+
+    def test_template_contains_only_terminal_runtime(self):
+        output = RENDER(TEMPLATE, PROFILE)
+        self.assertIn("# 27. FOREGROUND TERMINAL RUNTIME", output)
+        self.assertIn("nq-orchestrate", output)
+        self.assertNotIn("# 27. HOST CONTROLLER RUNTIME", output)
+        self.assertNotIn("The persistent user service runs", output)
+        self.assertNotIn("---\n\n---\n\n# 27", output)
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_approve_merge.py

@@ -0,0 +1,88 @@
+from pathlib import Path
+import contextlib
+import io
+import runpy
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+APPROVE = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-approve-merge"))
+HEAD = "a" * 40
+
+
+class MergeGateTests(unittest.TestCase):
+    def test_main_rejects_moved_head_unmergeable_and_pending_checks(self):
+        root = Path("/tmp/opencode")
+        root.mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=root) as directory:
+            review_file = Path(directory) / "review.txt"
+            review_file.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
+            for problem in ("moved", "unmergeable", "pending", "none"):
+                with self.subTest(problem=problem):
+                    merged = False
+                    calls = []
+                    def fake_request(path, token, payload=None):
+                        nonlocal merged
+                        calls.append((path, payload))
+                        if path.endswith("/reviews"):
+                            return [{"id": 1, "state": "APPROVED", "commit_id": HEAD,
+                                     "user": {"login": "review_agent"}}]
+                        if path.endswith("/status"):
+                            return {"total_count": 1, "state": "pending" if problem == "pending" else "success"}
+                        if path.endswith("/merge"):
+                            merged = True
+                            return {}
+                        if path.endswith("/pulls/149"):
+                            return {"number": 149, "base": {"ref": "dev"},
+                                    "head": {"sha": "b" * 40 if problem == "moved" else HEAD},
+                                    "state": "open", "merged": merged,
+                                    "mergeable": problem != "unmergeable"}
+                        raise AssertionError(path)
+                    globals_ = APPROVE["main"].__globals__
+                    with patch.dict(globals_, {"request": fake_request,
+                                               "token_from_file": lambda: "test"}), \
+                         patch.object(sys, "argv", ["nq-approve-merge", "merge", "149", HEAD,
+                                                    str(review_file)]), \
+                         contextlib.redirect_stdout(io.StringIO()):
+                        if problem == "none":
+                            self.assertEqual(APPROVE["main"](), 0)
+                        else:
+                            with self.assertRaises(RuntimeError):
+                                APPROVE["main"]()
+                    self.assertEqual(merged, problem == "none")
+                    if problem == "none":
+                        merge_calls = [payload for path, payload in calls if path.endswith("/merge")]
+                        self.assertEqual(merge_calls,
+                                         [{"do": "squash", "head_commit_id": HEAD}])
+
+    def test_review_evidence_must_match_exact_head_and_approval(self):
+        root = Path("/tmp/opencode")
+        root.mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=root) as directory:
+            review = Path(directory) / "review.txt"
+            review.write_text(f"Findings: none\nREVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
+            APPROVE["validate_review"](review, HEAD)
+            with self.assertRaisesRegex(RuntimeError, "exact head"):
+                APPROVE["validate_review"](review, "b" * 40)
+            review.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: CHANGES_REQUIRED\n")
+            with self.assertRaisesRegex(RuntimeError, "exact head"):
+                APPROVE["validate_review"](review, HEAD)
+
+    def test_requires_distinct_gitea_reviewer_on_exact_commit(self):
+        records = [{"state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}}]
+        with patch.dict(APPROVE["has_platform_review"].__globals__, {"request": lambda *_: records}):
+            self.assertTrue(APPROVE["has_platform_review"](149, HEAD, "unused"))
+            self.assertFalse(APPROVE["has_platform_review"](149, "b" * 40, "unused"))
+            records[0]["user"]["login"] = "orchestrator-agent"
+            self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
+            records[:] = [
+                {"id": 1, "state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}},
+                {"id": 2, "state": "REQUEST_CHANGES", "commit_id": HEAD, "user": {"login": "review_agent"}},
+            ]
+            self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_controller.py

@@ -0,0 +1,108 @@
+import json
+import io
+from pathlib import Path
+import runpy
+import sqlite3
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+CONTROL = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-controller"))
+
+
+class ControllerProtocolTests(unittest.TestCase):
+    def test_ctxd_preflight_reads_configured_endpoint_and_checks_identity(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = root / ".config/opencode/opencode.json"
+            config.parent.mkdir(parents=True)
+            config.write_text(json.dumps({"mcp": {"ctxd": {
+                "url": "http://localhost:9091/mcp",
+                "headers": {"Authorization": "Bearer test"},
+            }}}))
+            record = {"project_id": CONTROL["CTXD_PROJECT"],
+                      "file_path": CONTROL["CTXD_FILE"],
+                      "content": "# CONTROLLER_TICK", "version": 6}
+            with patch.object(Path, "home", return_value=root), \
+                 patch.dict(CONTROL["check_ctxd"].__globals__["urllib"].request.__dict__,
+                            {"urlopen": lambda request, timeout: io.BytesIO(json.dumps(record).encode())}):
+                self.assertEqual(CONTROL["check_ctxd"](), 6)
+                record["project_id"] = "wrong"
+                with self.assertRaisesRegex(RuntimeError, "wrong NestQuest prompt"):
+                    CONTROL["check_ctxd"]()
+                record["project_id"] = CONTROL["CTXD_PROJECT"]
+                config.write_text(json.dumps({"mcp": {"ctxd": {
+                    "url": "http://localhost:9091/mcp/",
+                    "headers": {"Authorization": "Bearer test"},
+                }}}))
+                self.assertEqual(CONTROL["check_ctxd"](), 6)
+                config.write_text(json.dumps({"mcp": {"ctxd": {
+                    "url": "http://localhost:9091/invalid",
+                    "headers": {"Authorization": "Bearer test"},
+                }}}))
+                with self.assertRaisesRegex(RuntimeError, "must end in /mcp"):
+                    CONTROL["check_ctxd"]()
+
+    def test_installed_opencode_text_event(self):
+        event = {"type": "text", "sessionID": "ses_test", "part": {"type": "text", "text": "Next task queued.\nORCHESTRATOR_STATE: CONTINUE\n"}}
+        self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (event["part"]["text"], None))
+        self.assertEqual(CONTROL["STATE_RE"].findall(event["part"]["text"]), ["CONTINUE"])
+
+    def test_error_event_does_not_publish_provider_message(self):
+        event = {"type": "error", "error": {"name": "APIError", "data": {"message": "secret provider response"}}}
+        self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (None, "APIError"))
+        self.assertEqual(CONTROL["parse_event"]("null"), (None, None))
+        self.assertEqual(CONTROL["parse_event"]("[]"), (None, None))
+
+    def test_tick_uses_final_text(self):
+        event = json.dumps({"type": "text", "part": {"type": "text", "text": "ORCHESTRATOR_STATE: QUIESCENT\n"}})
+        command = [sys.executable, "-c", f"import json,sys; sys.stdin.read(); print({event!r})"]
+        status, answer, error = CONTROL["run_tick"](command, {}, "tick", 10)
+        self.assertEqual((status, error), (0, None))
+        self.assertEqual(CONTROL["STATE_RE"].findall(answer), ["QUIESCENT"])
+
+    def test_timeout_kills_child_group_and_marks_tick_failed(self):
+        command = [sys.executable, "-c", "import sys,time; sys.stdin.read(); time.sleep(10)"]
+        status, answer, error = CONTROL["run_tick"](command, {}, "tick", 1)
+        self.assertEqual(status, 124)
+        self.assertEqual(answer, "")
+        self.assertIn("timeout", error)
+
+    def test_interactive_session_detects_dir_and_resumed_session(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            project = root / "project"
+            project.mkdir()
+            proc = root / "proc"
+            entry = proc / "123"
+            entry.mkdir(parents=True)
+            (entry / "cwd").symlink_to(root)
+            database = root / "opencode.db"
+            with sqlite3.connect(database) as db:
+                db.execute("CREATE TABLE session (id TEXT, directory TEXT)")
+                db.execute("INSERT INTO session VALUES (?, ?)", ("ses_test", str(project)))
+            (entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(project).encode() + b"\0")
+            self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
+            (entry / "cmdline").write_bytes(b"opencode\0web\0--dir\0" + str(project).encode() + b"\0")
+            self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
+            (entry / "cmdline").write_bytes(b"opencode\0serve\0")
+            (entry / "cwd").unlink()
+            (entry / "cwd").symlink_to(project)
+            self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
+            (entry / "cwd").unlink()
+            (entry / "cwd").symlink_to(root)
+            (entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
+            self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
+            (entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(root).encode() + b"\0")
+            self.assertFalse(CONTROL["interactive_nestquest_session"](project, proc, database))
+            with sqlite3.connect(database) as db:
+                db.execute("DROP TABLE session")
+            (entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
+            with self.assertRaisesRegex(RuntimeError, "lookup failed"):
+                CONTROL["interactive_nestquest_session"](project, proc, database)
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_orchestrate.py

@@ -0,0 +1,79 @@
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts/nq-orchestrate"
+
+
+class ForegroundLauncherTests(unittest.TestCase):
+    def test_launches_terminal_agent_with_rendered_and_user_prompt(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            control = root / "control"
+            scripts = control / "scripts"
+            agents = control / ".opencode/agents"
+            templates = control / "templates"
+            project = root / "project"
+            for path in (scripts, agents, templates, project):
+                path.mkdir(parents=True, exist_ok=True)
+            (templates / "nestquest.profile.json").write_text("{}")
+            (agents / "terminal-orchestrator.md").write_text("agent")
+            renderer = scripts / "render-cycle-prompt"
+            renderer.write_text("#!/bin/sh\nprintf 'BASE TERMINAL PROMPT'")
+            controller = scripts / "nq-controller"
+            controller_args = root / "controller-args"
+            open_marker = root / "open-session"
+            controller.write_text(
+                f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {controller_args}\n"
+                f"if test -f {open_marker}; then printf 'interactive NestQuest session open\\n'; "
+                "else printf 'no interactive NestQuest session\\n'; fi\n"
+            )
+            opencode = root / "opencode"
+            output = root / "args"
+            opencode.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {output}\n")
+            systemctl = root / "systemctl"
+            stopped = root / "stopped"
+            systemctl.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {stopped}\n")
+            for executable in (renderer, controller, opencode, systemctl):
+                executable.chmod(0o700)
+            mission = root / "mission.md"
+            mission.write_text("Complete feature EQ-17.")
+            env = os.environ.copy()
+            env.update({
+                "NQ_CONTROL_HOME": str(control),
+                "NQ_PROJECT_ROOT": str(project),
+                "OPENCODE_BIN": str(opencode),
+                "SYSTEMCTL_BIN": str(systemctl),
+            })
+            printed = subprocess.run([str(SCRIPT), "--print-prompt"], env=env,
+                                     text=True, capture_output=True)
+            self.assertEqual(printed.returncode, 0, printed.stderr)
+            self.assertEqual(printed.stdout, "BASE TERMINAL PROMPT\n")
+            self.assertFalse(stopped.exists())
+            result = subprocess.run([str(SCRIPT), str(mission)], env=env, text=True,
+                                    capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            args = output.read_text()
+            self.assertIn("--agent\nterminal-orchestrator", args)
+            self.assertIn("BASE TERMINAL PROMPT", args)
+            self.assertIn("# USER MISSION", args)
+            self.assertIn("Complete feature EQ-17.", args)
+            self.assertEqual(stopped.read_text().splitlines(),
+                             ["--user", "stop", "nestquest-controller.service"])
+            self.assertEqual(controller_args.read_text().splitlines(),
+                             ["--check-interactive", "--project", str(project)])
+
+            output.unlink()
+            open_marker.touch()
+            refused = subprocess.run([str(SCRIPT)], env=env, text=True,
+                                     capture_output=True)
+            self.assertEqual(refused.returncode, 75)
+            self.assertIn("Another interactive NestQuest", refused.stderr)
+            self.assertFalse(output.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

orchestration_tests/test_nq_review.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+import runpy
+import unittest
+
+REVIEW = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-review"))
+HEAD = "a" * 40
+
+class ReviewLauncherTests(unittest.TestCase):
+    def test_exact_head_verdict(self):
+        self.assertEqual(REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n", HEAD), "APPROVED")
+        with self.assertRaisesRegex(RuntimeError, "one verdict"):
+            REVIEW["verdict"](f"REVIEWED_HEAD: {'b' * 40}\nREVIEW_VERDICT: APPROVED\n", HEAD)
+        with self.assertRaisesRegex(RuntimeError, "one verdict"):
+            REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\nREVIEW_VERDICT: CHANGES_REQUIRED\n", HEAD)
+
+if __name__ == "__main__":
+    unittest.main()

scripts/nq-agent

@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [[ $# -ne 4 ]]; then
+  echo "usage: nq-agent developer|approver WORKTREE TITLE PROMPT_FILE" >&2
+  exit 64
+fi
+
+role=$1
+worktree=$2
+title=$3
+prompt_file=$4
+case "$role" in developer|approver) ;; *) echo "invalid role: $role" >&2; exit 64 ;; esac
+[[ -d "$worktree" && -f "$prompt_file" ]] || { echo "missing worktree or prompt" >&2; exit 66; }
+
+repo_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd -P)
+export OPENCODE_CONFIG_DIR="$repo_dir/.opencode"
+opencode_bin=${OPENCODE_BIN:-/home/overseer/.opencode/bin/opencode}
+
+exec "$opencode_bin" run --pure --dir "$worktree" --agent "$role" --title "$title" --format json < "$prompt_file"

scripts/nq-approve-merge

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""Inspect or merge a reviewed NestQuest feature PR into dev only."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import sys
+import urllib.error
+import urllib.request
+
+
+API = "https://code.cubecraftlabs.com/api/v1/repos/Maniacs_Mansion/NestQuest"
+CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
+FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
+
+
+def token_from_file() -> str:
+    for line in CREDENTIALS.read_text().splitlines():
+        if line.startswith("GITEA_APPROVER_AGENT_TOKEN="):
+            return line.split("=", 1)[1].strip().strip("\"'")
+    raise RuntimeError("GITEA_APPROVER_AGENT_TOKEN is unavailable")
+
+
+def request(path: str, token: str, payload: dict | None = None) -> dict | list:
+    body = None if payload is None else json.dumps(payload).encode()
+    headers = {"Authorization": "token " + token, "Accept": "application/json"}
+    if body is not None:
+        headers["Content-Type"] = "application/json"
+    call = urllib.request.Request(API + path, data=body, headers=headers)
+    try:
+        with urllib.request.urlopen(call, timeout=20) as response:
+            raw = response.read()
+            return json.loads(raw) if raw else {}
+    except urllib.error.HTTPError as error:
+        raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
+
+
+def inspect(pr_number: int, token: str) -> dict:
+    pr = request(f"/pulls/{pr_number}", token)
+    head = (pr.get("head") or {}).get("sha")
+    base = (pr.get("base") or {}).get("ref")
+    if not isinstance(head, str) or not FULL_SHA.fullmatch(head):
+        raise RuntimeError("Gitea returned an invalid PR head")
+    status = request(f"/commits/{head}/status", token)
+    return {
+        "number": pr.get("number"),
+        "base": base,
+        "head": head,
+        "state": pr.get("state"),
+        "merged": pr.get("merged"),
+        "mergeable": pr.get("mergeable"),
+        "check_count": status.get("total_count") or 0,
+        "check_state": status.get("state"),
+    }
+
+
+def validate_review(path: Path, head: str) -> None:
+    if not path.is_file() or not path.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
+        raise RuntimeError("review evidence must be a file under /tmp/opencode")
+    text = path.read_text(errors="replace")
+    reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", text)
+    verdicts = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", text)
+    if len(reviewed) != 1 or len(verdicts) != 1 or reviewed[0] != head or verdicts[0] != "APPROVED":
+        raise RuntimeError("independent review does not approve this exact head")
+
+
+def has_platform_review(pr_number: int, head: str, token: str) -> bool:
+    reviews = request(f"/pulls/{pr_number}/reviews", token)
+    if not isinstance(reviews, list):
+        raise RuntimeError("Gitea returned invalid review records")
+    by_reviewer = [
+        review for review in reviews
+        if isinstance(review, dict)
+        and (review.get("user") or {}).get("login") == "review_agent"
+        and review.get("commit_id") == head
+    ]
+    if not by_reviewer:
+        return False
+    latest = max(by_reviewer, key=lambda review: int(review.get("id") or 0))
+    return latest.get("state") == "APPROVED"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("operation", choices=("inspect", "merge"))
+    parser.add_argument("pr", type=int)
+    parser.add_argument("reviewed_head", nargs="?")
+    parser.add_argument("review_file", nargs="?", type=Path)
+    args = parser.parse_args()
+    if args.pr < 1:
+        parser.error("PR number must be positive")
+    if args.operation == "merge" and (not args.reviewed_head or not args.review_file):
+        parser.error("merge requires reviewed_head and review_file")
+    token = token_from_file()
+    before = inspect(args.pr, token)
+    if args.operation == "inspect":
+        print(json.dumps(before, sort_keys=True))
+        return 0
+    if not FULL_SHA.fullmatch(args.reviewed_head):
+        parser.error("reviewed head must be a full SHA")
+    validate_review(args.review_file, args.reviewed_head)
+    if not has_platform_review(args.pr, args.reviewed_head, token):
+        raise RuntimeError("Gitea lacks an exact-head approval by review_agent")
+    if before["base"] != "dev" or before["head"] != args.reviewed_head:
+        raise RuntimeError("PR target or current head differs from the approved review")
+    if before["state"] != "open" or before["merged"] or before["mergeable"] is not True:
+        raise RuntimeError("PR is closed, merged, or not mergeable")
+    if before["check_count"] and before["check_state"] != "success":
+        raise RuntimeError("a configured commit check has not passed")
+    # Gitea rechecks branch protection at the merge endpoint. The caller must
+    # independently verify Maestro and task-level evidence before invoking us.
+    # Gitea checks head_commit_id atomically while processing the merge.
+    request(f"/pulls/{args.pr}/merge", token,
+            {"do": "squash", "head_commit_id": args.reviewed_head})
+    after = inspect(args.pr, token)
+    if after["merged"] is not True or after["head"] != args.reviewed_head:
+        raise RuntimeError("merge result is missing or differs from the reviewed head")
+    print(json.dumps(after, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (OSError, ValueError, RuntimeError) as error:
+        print(f"approval gate failed: {error}", file=sys.stderr)
+        raise SystemExit(1)

scripts/nq-controller

@@ -0,0 +1,277 @@
+#!/usr/bin/env python3
+"""Durable, single-owner controller for one NestQuest OpenCode tick at a time."""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import selectors
+import signal
+import sqlite3
+import subprocess
+import sys
+import time
+import urllib.request
+
+
+DEFAULT_PROJECT = Path("/mnt/ai-storage/projects/NestQuest")
+DEFAULT_STATE = Path.home() / ".local/state/nestquest-controller"
+STATE_RE = re.compile(r"(?m)^ORCHESTRATOR_STATE: (CONTINUE|QUIESCENT|BLOCKED)\s*$")
+MAX_TEXT = 64_000
+MAX_LINE = 4_000_000
+CTXD_PROJECT = "nestquest-house-chaos-coordination"
+CTXD_FILE = "OPENCODE-ORCHESTRATOR-PROMPT.MD"
+
+
+def check_ctxd() -> int:
+    """Use OpenCode's configured endpoint, never an unrelated host DB."""
+    config = json.loads((Path.home() / ".config/opencode/opencode.json").read_text())
+    mcp = config["mcp"]["ctxd"]
+    configured_url = mcp["url"].rstrip("/")
+    if not configured_url.endswith("/mcp"):
+        raise RuntimeError("CTXD MCP URL must end in /mcp")
+    endpoint = configured_url.removesuffix("/mcp")
+    url = f"{endpoint}/projects/{CTXD_PROJECT}/files/{CTXD_FILE}"
+    with urllib.request.urlopen(
+        urllib.request.Request(url, headers=mcp["headers"]), timeout=8
+    ) as response:
+        record = json.load(response)
+    if (
+        record.get("project_id") != CTXD_PROJECT
+        or record.get("file_path") != CTXD_FILE
+        or "CONTROLLER_TICK" not in record.get("content", "")
+    ):
+        raise RuntimeError("CTXD returned the wrong NestQuest prompt")
+    return int(record["version"])
+
+
+def interactive_nestquest_session(
+    project: Path,
+    proc_root: Path = Path("/proc"),
+    database: Path = Path.home() / ".local/share/opencode/opencode.db",
+) -> bool:
+    """Avoid racing a user-owned OpenCode TUI for the same repository."""
+    try:
+        db = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
+    except sqlite3.Error as error:
+        raise RuntimeError("OpenCode session database is unavailable") from error
+    try:
+        for entry in proc_root.iterdir():
+            if not entry.name.isdecimal() or int(entry.name) == os.getpid():
+                continue
+            try:
+                args = (entry / "cmdline").read_bytes().split(b"\0")
+                args = [a.decode(errors="replace") for a in args if a]
+                if not args or Path(args[0]).name != "opencode":
+                    continue
+                if len(args) > 1 and args[1] in ("run", "debug"):
+                    continue
+                session = None
+                for flag in ("-s", "--session"):
+                    if flag in args and args.index(flag) + 1 < len(args):
+                        session = args[args.index(flag) + 1]
+                        break
+                if session:
+                    row = db.execute("SELECT directory FROM session WHERE id=?", (session,)).fetchone()
+                    if row and Path(row[0]).resolve() == project.resolve():
+                        return True
+                elif "--dir" in args and args.index("--dir") + 1 < len(args):
+                    if Path(args[args.index("--dir") + 1]).resolve() == project.resolve():
+                        return True
+                elif (entry / "cwd").resolve() == project.resolve():
+                    return True
+            except sqlite3.Error as error:
+                raise RuntimeError("OpenCode session lookup failed") from error
+            except TypeError as error:
+                raise RuntimeError("OpenCode session directory is invalid") from error
+            except (OSError, PermissionError):
+                continue
+    finally:
+        db.close()
+    return False
+
+
+def parse_event(raw: str) -> tuple[str | None, str | None]:
+    try:
+        event = json.loads(raw)
+    except json.JSONDecodeError:
+        return None, None
+    if not isinstance(event, dict):
+        return None, None
+    kind = event.get("type")
+    part = event.get("part") or {}
+    if kind == "text" and isinstance(part, dict):
+        return part.get("text"), None
+    if kind == "error":
+        error = event.get("error") or {}
+        known = {"ProviderAuthError", "UnknownError", "MessageOutputLengthError", "MessageAbortedError", "StructuredOutputError", "ContextOverflowError", "APIError"}
+        name = error.get("name") if isinstance(error, dict) else None
+        return None, name if name in known else "OpenCode error"
+    return None, None
+
+
+def run_tick(command: list[str], env: dict[str, str], prompt: str, timeout: int) -> tuple[int, str, str | None]:
+    process = subprocess.Popen(
+        command,
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        env=env,
+        start_new_session=True,
+    )
+    assert process.stdin is not None and process.stdout is not None
+    try:
+        process.stdin.write(prompt.encode())
+    except BrokenPipeError:
+        pass
+    finally:
+        process.stdin.close()
+    selector = selectors.DefaultSelector()
+    selector.register(process.stdout, selectors.EVENT_READ)
+    deadline = time.monotonic() + timeout
+    buffer = b""
+    texts: list[str] = []
+    last_error: str | None = None
+    diagnostic_lines = 0
+    try:
+        while selector.get_map():
+            if time.monotonic() >= deadline:
+                try:
+                    os.killpg(process.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+                process.wait()
+                return 124, "".join(texts)[-MAX_TEXT:], "controller tick exceeded documented timeout"
+            for key, _ in selector.select(timeout=1):
+                chunk = os.read(key.fd, 65536)
+                if not chunk:
+                    selector.unregister(key.fileobj)
+                    continue
+                buffer += chunk
+                while b"\n" in buffer:
+                    line, buffer = buffer.split(b"\n", 1)
+                    text, error = parse_event(line.decode(errors="replace"))
+                    if not line.lstrip().startswith(b"{"):
+                        diagnostic_lines += 1
+                    if text:
+                        texts.append(text)
+                        if sum(map(len, texts)) > MAX_TEXT:
+                            texts = ["".join(texts)[-MAX_TEXT:]]
+                    if error:
+                        last_error = error
+                if len(buffer) > MAX_LINE:
+                    buffer = b""  # A tool-output event is not a controller verdict.
+        if buffer:
+            text, error = parse_event(buffer.decode(errors="replace"))
+            if text:
+                texts.append(text)
+            if error:
+                last_error = error
+        code = process.wait()
+        if code and last_error is None and diagnostic_lines:
+            last_error = f"OpenCode emitted {diagnostic_lines} non-JSON diagnostic lines"
+        return code, "".join(texts)[-MAX_TEXT:], last_error
+    finally:
+        selector.close()
+        process.stdout.close()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--project", type=Path, default=DEFAULT_PROJECT)
+    parser.add_argument("--state-dir", type=Path, default=DEFAULT_STATE)
+    parser.add_argument("--once", action="store_true", help="run only one model turn")
+    parser.add_argument("--check", action="store_true", help="validate paths without calling a model")
+    parser.add_argument("--check-interactive", action="store_true", help="report an open NestQuest TUI without calling a model")
+    parser.add_argument("--tick-timeout", type=int, default=6 * 3600)
+    args = parser.parse_args()
+
+    project = args.project.resolve()
+    control_home = Path(__file__).resolve().parent.parent
+    config_dir = control_home / ".opencode"
+    binary = Path(os.environ.get("OPENCODE_BIN", "/home/overseer/.opencode/bin/opencode"))
+    if not (project.is_dir() and config_dir.is_dir() and binary.is_file()):
+        parser.error("project, agent configuration, or OpenCode binary is missing")
+    if args.check:
+        try:
+            ctxd_version = check_ctxd()
+        except (OSError, KeyError, ValueError, RuntimeError) as error:
+            print(f"CTXD preflight failed: {type(error).__name__}", file=sys.stderr)
+            return 1
+        print(f"ready: project={project} configuration={config_dir} binary={binary} ctxd_prompt_version={ctxd_version}")
+        return 0
+    if args.check_interactive:
+        try:
+            active = interactive_nestquest_session(project)
+        except RuntimeError:
+            print("OpenCode session state unavailable", file=sys.stderr)
+            return 77
+        print("interactive NestQuest session open" if active else "no interactive NestQuest session")
+        return 0
+
+    args.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
+    os.chmod(args.state_dir, 0o700)
+    with (args.state_dir / "controller.lock").open("w") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            print("another NestQuest controller is running", file=sys.stderr)
+            return 75
+        command = [str(binary), "run", "--pure", "--dir", str(project), "--agent", "orchestrator", "--format", "json", "--title", "NestQuest controller tick"]
+        env = os.environ.copy()
+        env["OPENCODE_CONFIG_DIR"] = str(config_dir)
+        env["NESTQUEST_CONTROL_HOME"] = str(control_home)
+        failures = 0
+        while True:
+            try:
+                interactive = interactive_nestquest_session(project)
+            except RuntimeError:
+                print("OpenCode session state unavailable; waiting", file=sys.stderr, flush=True)
+                if args.once:
+                    return 77
+                time.sleep(60)
+                continue
+            if interactive:
+                print("interactive NestQuest session still open; waiting", flush=True)
+                if args.once:
+                    return 75
+                time.sleep(60)
+                continue
+            prompt = (
+                "Run one NestQuest controller tick. Read CTXD and Maestro afresh, "
+                "reconcile Git and PR state, then execute the next eligible action. "
+                "Use NESTQUEST_CONTROL_HOME for role launch scripts. "
+                "Return a final ORCHESTRATOR_STATE line as your agent instructions require."
+            )
+            status, answer, error = run_tick(command, env, prompt, args.tick_timeout)
+            states = STATE_RE.findall(answer)
+            state = states[-1] if status == 0 and error is None and states else "ERROR"
+            if state == "CONTINUE":
+                failures = 0
+            elif state == "QUIESCENT":
+                failures = 0
+            elif state == "BLOCKED":
+                failures = 0
+            else:
+                failures += 1
+            summary = {"time": int(time.time()), "state": state, "exit_code": status, "error": error, "failure_count": failures, "answer_tail": answer[-4000:]}
+            target = args.state_dir / "last-run.json"
+            temporary = args.state_dir / "last-run.json.tmp"
+            temporary.write_text(json.dumps(summary, indent=2) + "\n")
+            os.chmod(temporary, 0o600)
+            temporary.replace(target)
+            print(f"tick state={state} exit={status} failures={failures}", flush=True)
+            if args.once:
+                return 0 if state in ("CONTINUE", "QUIESCENT") else 77
+            if state == "BLOCKED" or failures >= 3:
+                print("controller blocked; inspect last-run.json", file=sys.stderr)
+                return 77
+            time.sleep(300 if state == "QUIESCENT" else min(120, 10 * 2**failures))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/nq-orchestrate

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+# Launch the NestQuest orchestrator immediately in the foreground terminal.
+set -euo pipefail
+
+if [[ $# -gt 1 ]]; then
+  echo "usage: nq-orchestrate [MISSION_PROMPT_FILE|-|--print-prompt]" >&2
+  exit 64
+fi
+
+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}
+renderer="$control_home/scripts/render-cycle-prompt"
+profile="$control_home/templates/nestquest.profile.json"
+
+for required in "$renderer" "$profile"; do
+  [[ -e "$required" ]] || { echo "missing required path: $required" >&2; exit 66; }
+done
+
+prompt=$("$renderer" "$profile")
+if [[ ${1:-} == "--print-prompt" ]]; then
+  printf '%s\n' "$prompt"
+  exit 0
+fi
+for required in "$project" "$opencode_bin" "$control_home/.opencode/agents/terminal-orchestrator.md"; do
+  [[ -e "$required" ]] || { echo "missing required path: $required" >&2; exit 66; }
+done
+if [[ $# -eq 1 ]]; then
+  if [[ "$1" == "-" ]]; then
+    mission=$(cat)
+  else
+    [[ -f "$1" ]] || { echo "mission prompt not found: $1" >&2; exit 66; }
+    mission=$(cat "$1")
+  fi
+  [[ -n "${mission//[[:space:]]/}" ]] || { echo "mission prompt is empty" >&2; exit 65; }
+  prompt+=$'\n\n---\n\n# USER MISSION\n\n'
+  prompt+="$mission"
+fi
+
+# Foreground mode owns orchestration. A dormant service must never race it.
+"$systemctl_bin" --user stop nestquest-controller.service
+
+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
+  echo "Close it, then run nq-orchestrate again." >&2
+  exit 75
+fi
+
+export OPENCODE_CONFIG_DIR="$control_home/.opencode"
+exec "$opencode_bin" "$project" --pure --agent terminal-orchestrator --prompt "$prompt"

scripts/nq-review

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Run Claude on an exact PR head and publish an independent Gitea approval."""
+from __future__ import annotations
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+
+API = "https://code.cubecraftlabs.com/api/v1"
+REPO = "/repos/Maniacs_Mansion/NestQuest"
+CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
+CLAUDE = "/home/overseer/.local/bin/claude"
+SHA = re.compile(r"^[0-9a-f]{40}$")
+
+def git(worktree: Path, *args: str) -> str:
+    return subprocess.check_output(["git", "-C", str(worktree), *args], text=True).strip()
+
+def token() -> str:
+    for line in CREDENTIALS.read_text().splitlines():
+        if line.startswith("GITEA_REVIEW_AGENT_TOKEN="):
+            return line.split("=", 1)[1].strip().strip("\"'")
+    raise RuntimeError("review_agent credential is unavailable")
+
+def api(path: str, credential: str, payload: dict | None = None) -> dict:
+    body = None if payload is None else json.dumps(payload).encode()
+    headers = {"Authorization": "token " + credential, "Accept": "application/json"}
+    if body is not None:
+        headers["Content-Type"] = "application/json"
+    call = urllib.request.Request(API + path, data=body, headers=headers)
+    try:
+        with urllib.request.urlopen(call, timeout=20) as response:
+            return json.loads(response.read())
+    except urllib.error.HTTPError as error:
+        raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
+
+def verdict(output: str, head: str) -> str:
+    reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", output)
+    decisions = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", output)
+    if reviewed != [head] or len(decisions) != 1:
+        raise RuntimeError("Claude did not return one verdict for the exact reviewed head")
+    return decisions[0]
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("worktree", type=Path)
+    parser.add_argument("base_ref")
+    parser.add_argument("head_ref")
+    parser.add_argument("prompt_file", type=Path)
+    parser.add_argument("pr", type=int)
+    parser.add_argument("output_file", type=Path)
+    args = parser.parse_args()
+    if not args.worktree.is_dir() or not args.prompt_file.is_file() or args.pr < 1:
+        parser.error("valid worktree, prompt file, and PR number required")
+    if not args.output_file.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
+        parser.error("review output must be under /tmp/opencode")
+    base = git(args.worktree, "rev-parse", "--verify", args.base_ref + "^{commit}")
+    head = git(args.worktree, "rev-parse", "--verify", args.head_ref + "^{commit}")
+    if not SHA.fullmatch(head) or git(args.worktree, "rev-parse", "HEAD") != head:
+        raise RuntimeError("worktree HEAD differs from requested head")
+    subprocess.run(["git", "-C", str(args.worktree), "merge-base", "--is-ancestor", base, head], check=True)
+    prompt = args.prompt_file.read_text() + (
+        f"\n\nReview only the delta from base {base} to exact head {head}. "
+        "Verify current HEAD before issuing a verdict. Read only; do not modify files. "
+        "End with two plain lines: REVIEWED_HEAD: <full 40-character SHA> "
+        "and REVIEW_VERDICT: APPROVED or REVIEW_VERDICT: CHANGES_REQUIRED."
+    )
+    result = subprocess.run(
+        [CLAUDE, "-p", prompt, "--model", "claude-sonnet-5", "--effort", "high",
+         "--permission-mode", "dontAsk", "--disallowedTools", "Edit", "Write",
+         "NotebookEdit", "--allowedTools", "Read", "Grep", "Glob",
+         "Bash(git diff:*)", "Bash(git log:*)", "Bash(git show:*)", "Bash(git status:*)"],
+        cwd=args.worktree, text=True, capture_output=True, timeout=1200,
+    )
+    args.output_file.parent.mkdir(parents=True, exist_ok=True)
+    args.output_file.write_text(result.stdout)
+    os.chmod(args.output_file, 0o600)
+    if result.returncode:
+        raise RuntimeError(f"Claude exited {result.returncode}: {result.stderr[-300:]}")
+    decision = verdict(result.stdout, head)
+    if git(args.worktree, "rev-parse", "HEAD") != head:
+        raise RuntimeError("worktree HEAD changed during review")
+    print(result.stdout, end="")
+    if decision != "APPROVED":
+        return 2
+    credential = token()
+    if api("/user", credential).get("login") != "review_agent":
+        raise RuntimeError("review credential belongs to wrong account")
+    pr = api(f"{REPO}/pulls/{args.pr}", credential)
+    if pr.get("state") != "open" or (pr.get("head") or {}).get("sha") != head:
+        raise RuntimeError("PR head changed before review publication")
+    published = api(
+        f"{REPO}/pulls/{args.pr}/reviews", credential,
+        {"body": "Independent Claude Sonnet 5 review completed for this exact commit.",
+         "commit_id": head, "event": "APPROVED"},
+    )
+    if published.get("state") != "APPROVED" or published.get("commit_id") != head:
+        raise RuntimeError("Gitea did not confirm exact-head approval")
+    return 0
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError,
+            subprocess.TimeoutExpired) as error:
+        print(f"review gate failed: {error}", file=sys.stderr)
+        raise SystemExit(1)

scripts/render-cycle-prompt

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Render a project-specific controller prompt from the reusable template."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import re
+import sys
+
+
+TEMPLATE = Path(__file__).resolve().parent.parent / "templates/ORCHESTRATOR-PROMPT.template.md"
+PLACEHOLDER = re.compile(r"\{\{([A-Z][A-Z0-9_]*)\}\}")
+
+
+def render(template: str, profile: dict[str, str]) -> str:
+    keys = set(PLACEHOLDER.findall(template))
+    missing = sorted(keys - profile.keys())
+    if missing:
+        raise ValueError("missing profile values: " + ", ".join(missing))
+    for key in keys:
+        if not isinstance(profile[key], str) or not profile[key].strip():
+            raise ValueError(f"{key} must be a nonempty string")
+    result = PLACEHOLDER.sub(lambda match: profile[match.group(1)], template)
+    if PLACEHOLDER.search(result):
+        raise ValueError("profile values must not contain template placeholders")
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("profile", type=Path)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    profile = json.loads(args.profile.read_text())
+    if not isinstance(profile, dict):
+        parser.error("profile must be a JSON object")
+    content = render(TEMPLATE.read_text(), profile)
+    if args.output:
+        args.output.write_text(content)
+        print(args.output)
+    else:
+        sys.stdout.write(content)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

templates/ORCHESTRATOR-PROMPT.template.md

@@ -0,0 +1,958 @@
+TYPE: PROJECT CONTEXT
+PROJECT: {{PROJECT_CONTEXT_LABEL}}
+STATUS: ACTIVE
+LAST-UPDATED: {{LAST_UPDATED}}
+TAGS:
+  - {{PROJECT_CONTEXT_LABEL}}
+  - CONTEXT
+--
+___
+
+# {{PROJECT_NAME}} Persistent Autonomous Development Orchestrator
+
+You are the primary autonomous engineering controller for the {{PROJECT_NAME}} {{WORK_TRACKER}} project.
+
+You are not a one-shot planner, status reporter, or implementation agent.
+
+You are a **persistent development-loop controller** responsible for continuously advancing the authorized project scope through:
+
+**plan → implement → validate → review → remediate → re-review → merge → verify → advance**
+
+until a legitimate GLOBAL TERMINAL CONDITION is reached.
+
+---
+
+# 1. PRIME DIRECTIVE
+
+Once execution begins, continue performing useful engineering actions without requiring additional user prompts.
+
+## Non-negotiable continuation rules
+
+1. **Completing one action is a state transition, not permission to stop.**
+2. After EVERY tool result, subagent result, review result, merge result, CI result, or repository-changing event, immediately execute the `CONTROLLER_TICK` defined below.
+3. If an executable `NEXT_ACTION` exists, perform it before producing a final user-facing response.
+4. A nonterminal state MUST lead to another tool call, agent dispatch, review, merge operation, verification action, or other executable action.
+5. Do not end execution merely because:
+   - A developer finished.
+   - A pull request was opened.
+   - A review finished.
+   - Corrections were pushed.
+   - A task was merged.
+   - A feature was merged.
+   - CI was started.
+   - Another agent was dispatched.
+   - One work item became blocked.
+6. Statements such as:
+   - "The agent has been dispatched."
+   - "The review is complete."
+   - "The next step is..."
+   - "I will now..."
+   - "Waiting for..."
+   
+   are NOT terminal outcomes.
+
+If the stated next step can be performed with available tools, PERFORM IT.
+
+7. A progress message never substitutes for the next required action.
+8. Never wait for the user merely to authorize routine continuation already permitted by this prompt.
+
+---
+
+# 2. EXECUTION CONFIGURATION
+
+Use these exact roles.
+
+## Orchestrator
+
+- Harness: OpenCode
+- Provider: {{MODEL_PROVIDER}}
+- Model: `{{CONTROLLER_MODEL}}`
+- Role: Persistent controller and state-machine owner
+
+The orchestrator does NOT implement application code.
+
+## Development Agent
+
+- Harness: OpenCode
+- Provider: {{MODEL_PROVIDER}}
+- Model: `{{DEVELOPER_MODEL}}`
+- Maximum concurrent development agents: {{MAX_DEVELOPERS}}
+
+Development agents implement and remediate code.
+
+## Independent Reviewer
+
+- Harness: {{REVIEWER_HARNESS}}
+- Model: `{{REVIEWER_MODEL}}`
+- Reasoning effort: `high`
+- Fresh independent session for every required review
+- Read-only
+
+Invocation:
+
+```bash
+{{REVIEWER_CLI}} -p "<review prompt>" \
+  --model {{REVIEWER_MODEL}} \
+  --effort high \
+  --permission-mode dontAsk \
+  --disallowedTools Edit Write NotebookEdit \
+  --allowedTools Read Grep Glob \
+    "Bash(git diff:*)" \
+    "Bash(git log:*)" \
+    "Bash(git show:*)" \
+    "Bash(git status:*)"
+```
+
+The reviewer reviews code and test evidence but does not execute arbitrary test commands.
+
+## Approver
+
+- Harness: OpenCode
+- Provider: {{MODEL_PROVIDER}}
+- Model: `{{CONTROLLER_MODEL}}`
+- MUST use a separate agent session from both orchestrator and developer
+- Merge-control role only
+
+Do not silently substitute models, harnesses, or roles.
+
+---
+
+# 3. SOURCES OF TRUTH
+
+Use, in order:
+
+1. {{WORK_TRACKER}}:
+   - Feature scope
+   - Tasks
+   - Dependencies
+   - Acceptance criteria
+   - Status
+   - Findings
+   - Branches
+   - Pull requests
+2. {{CONTEXT_STORE}} critical project context.
+3. Repository instructions and architecture.
+4. CI configuration and branch protections.
+5. This orchestration policy.
+
+Repository security and branch protections always take precedence.
+
+---
+
+# 4. RUN SCOPE
+
+Determine `MISSION_SCOPE` from the user's launch request.
+
+Possible scopes:
+
+- `SINGLE_TASK`
+- `SINGLE_FEATURE`
+- `PROJECT_QUEUE`
+- `RELEASE_PREPARATION`
+
+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:
+
+`MISSION_SCOPE = PROJECT_QUEUE`
+
+IMPORTANT:
+
+**Feature completion is not inherently a terminal condition.**
+
+When `MISSION_SCOPE = PROJECT_QUEUE`, completing a feature causes an immediate transition to selecting the next eligible project feature.
+
+---
+
+# 5. DURABLE CONTROLLER STATE
+
+Maintain a durable orchestration checkpoint using {{WORK_TRACKER}} wherever its schema permits.
+
+Track at minimum:
+
+- `MISSION_SCOPE`
+- Current feature
+- Current controller phase
+- Active development agents
+- Active task IDs
+- Task branches
+- Pull requests
+- Current head SHAs
+- Pending reviews
+- Reviewed SHAs
+- Pending remediation
+- Merge-ready items
+- Blocked items
+- Tests/checks status
+- Last verified repository state
+- `NEXT_ACTION`
+
+After every material state transition:
+
+1. Update {{WORK_TRACKER}}.
+2. Set `NEXT_ACTION`.
+3. Continue execution.
+
+`NEXT_ACTION` must be concrete and executable.
+
+Good:
+
+`Dispatch developer for {{TASK_ID_EXAMPLE}} from {{FEATURE_BRANCH_EXAMPLE}}.`
+
+Bad:
+
+`Continue development.`
+
+---
+
+# 6. CONTROLLER_TICK
+
+Run this algorithm after initialization and after EVERY completed action.
+
+## CONTROLLER_TICK
+
+### A. RECONCILE
+
+Inspect the actual current state of:
+
+- {{WORK_TRACKER}}
+- Git repository
+- Relevant branches
+- Pull requests
+- Current head commits
+- CI/check status
+- Active child agents
+- Review results
+- Merge status
+
+Never assume a previous action succeeded.
+
+### B. INGEST RESULTS
+
+Process every newly completed:
+
+- Development result
+- Review
+- Remediation
+- CI/check result
+- Merge
+- Approver result
+
+Update the durable state.
+
+### C. IDENTIFY NEXT TRANSITION
+
+Choose the highest-priority actionable transition in this order:
+
+1. Resolve a failed mandatory engineering gate.
+2. Process `CHANGES_REQUIRED`.
+3. Re-review remediated current heads.
+4. Merge task PRs whose gates are fully satisfied.
+5. Dispatch reviews for review-ready tasks.
+6. Fill available development slots with eligible tasks.
+7. Begin feature integration when all required tasks are merged.
+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.
+13. Continue unrelated work around locally blocked items.
+14. Escalate only when the defined escalation rules require it.
+
+### D. EXECUTE
+
+Execute the selected action NOW.
+
+Do not merely describe it.
+
+### E. REPEAT
+
+When the action returns, immediately begin another `CONTROLLER_TICK`.
+
+---
+
+# 7. DEVELOPMENT SLOT SCHEDULER
+
+Maximum concurrent development agents: **{{MAX_DEVELOPERS}}**.
+
+Keep development capacity utilized when safe.
+
+Do NOT use batch barriers.
+
+Example:
+
+If Developer A finishes while Developer B is still working:
+
+1. Process Developer A's result immediately.
+2. Start review/remediation/merge work for A as appropriate.
+3. If another eligible task exists and a developer slot is free, fill it.
+4. Do not wait unnecessarily for Developer B.
+
+Reviewers and approvers do not consume developer slots.
+
+Avoid parallel development of tightly coupled tasks when overlapping modifications would create excessive merge conflict risk.
+
+---
+
+# 8. FEATURE INITIALIZATION
+
+For a feature:
+
+1. Fetch the latest validated `{{INTEGRATION_BRANCH}}`.
+2. Inspect the feature and existing {{WORK_TRACKER}} tasks.
+3. Determine acceptance criteria.
+4. Determine dependencies.
+5. Identify missing implementation, testing, migration, documentation, observability, rollout, or rollback work.
+6. Create genuinely missing {{WORK_TRACKER}} tasks.
+7. Construct the dependency graph.
+8. Create exactly one feature branch from validated `{{INTEGRATION_BRANCH}}`.
+9. Publish it.
+10. Record the source commit and feature branch.
+11. Populate the ready-task queue.
+12. Continue immediately into the task loop.
+
+Do not repeatedly reconstruct the entire dependency graph when nothing affecting dependencies has changed.
+
+Update it incrementally after:
+
+- Task completion
+- Newly discovered work
+- Dependency changes
+- Blockers
+- Scope changes
+
+---
+
+# 9. TASK STATE MACHINE
+
+Each task follows:
+
+`PLANNED`
+→ `READY`
+→ `IN_PROGRESS`
+→ `PR_OPEN`
+→ `PRE_REVIEW_GATE`
+→ `REVIEW_PENDING`
+→ `APPROVED`
+→ `MERGE_READY`
+→ `MERGED`
+
+Remediation path:
+
+`REVIEW_PENDING`
+→ `CHANGES_REQUIRED`
+→ `REMEDIATION`
+→ `PRE_REVIEW_GATE`
+→ `REVIEW_PENDING`
+
+Blocked path:
+
+`ANY_STATE`
+→ `LOCALLY_BLOCKED`
+
+`LOCALLY_BLOCKED` is NOT automatically a global terminal state.
+
+---
+
+# 10. DEVELOPMENT AGENT CONTRACT
+
+Every development-agent assignment must include:
+
+- {{WORK_TRACKER}} task ID
+- Purpose
+- Exact scope
+- Acceptance criteria
+- Base branch
+- Assigned task branch
+- Required tests/checks
+- Relevant repository instructions
+- Known review findings when performing remediation
+- Explicit instruction not to expand scope unnecessarily
+
+The agent must:
+
+1. Work only in its isolated task worktree.
+2. Implement the assigned task.
+3. Add/update appropriate tests.
+4. Run relevant local checks.
+5. Self-review the diff.
+6. Remove debug artifacts and unrelated changes.
+7. Commit the work.
+8. Push its task branch.
+9. Open/update the task PR against the feature branch.
+10. Return structured results.
+
+Required return format:
+
+```text
+STATUS: READY_FOR_REVIEW | BLOCKED | FAILED
+TASK_ID:
+BRANCH:
+HEAD_SHA:
+PR:
+FILES_CHANGED:
+ACCEPTANCE_CRITERIA:
+  - criterion: PASS | FAIL
+TESTS:
+  - command:
+    result:
+CHECKS:
+BLOCKER:
+NOTES:
+```
+
+The orchestrator must independently verify important repository/PR state.
+
+---
+
+# 11. PRE-REVIEW GATE
+
+Do NOT spend an independent {{REVIEWER_VENDOR}} review on obviously unready code.
+
+Before dispatching {{REVIEWER_VENDOR}}, verify:
+
+- PR exists.
+- Expected head SHA exists.
+- Required local checks passed.
+- Required tests passed.
+- Build/lint/type/security checks required at this stage passed.
+- No unresolved merge conflict exists.
+- Acceptance criteria appear implemented.
+- No obvious debug/generated/unrelated files exist.
+
+If a mandatory check fails:
+
+Return the task to the development agent BEFORE independent review.
+
+---
+
+# 12. INDEPENDENT REVIEW CONTRACT
+
+For task review, compare:
+
+`task head → current feature-branch base`
+
+For feature review, compare:
+
+`feature head → current {{INTEGRATION_BRANCH}} base`
+
+For release review, compare:
+
+`release head → {{RELEASE_BRANCH}}`
+
+Every review must state the exact reviewed head SHA.
+
+Review:
+
+- Correctness
+- Acceptance criteria
+- Test adequacy
+- Security/privacy
+- Data integrity
+- Error handling
+- Concurrency
+- Compatibility/regressions
+- Performance risk
+- Maintainability
+- Migration correctness
+- Documentation
+- Rollout/rollback
+
+The reviewer is read-only.
+
+The reviewer may inspect tests and supplied check evidence but is NOT expected to execute non-Git test/build commands with the restricted tool configuration.
+
+## Reviewer result semantics
+
+Return exactly one:
+
+### `APPROVED`
+
+Use when:
+
+- No High findings exist.
+- No mandatory engineering gate is demonstrably violated.
+- Medium/Low findings may still exist.
+
+### `CHANGES_REQUIRED`
+
+Use only when:
+
+- At least one High finding exists; OR
+- Acceptance criteria are not satisfied; OR
+- A mandatory engineering gate is violated.
+
+### `BLOCKED`
+
+Use only when the reviewer cannot reach a reliable conclusion because necessary code, repository state, evidence, or access is unavailable.
+
+Every finding must include:
+
+```text
+ID:
+SEVERITY: HIGH | MEDIUM | LOW
+FILE:
+LOCATION:
+EXPLANATION:
+IMPACT:
+EVIDENCE:
+RECOMMENDED_CORRECTION:
+```
+
+Any new implementation commit invalidates the previous approval.
+
+---
+
+# 13. FINDING POLICY
+
+## High
+
+Must be corrected and independently re-reviewed.
+
+Examples:
+
+- Incorrect required behavior
+- Security/privacy exposure
+- Data corruption/loss
+- Major regression
+- Broken compatibility
+- Missing acceptance criterion
+
+## Medium
+
+Does not automatically block.
+
+Orchestrator must either:
+
+- Have it fixed; OR
+- Record a technical disposition explaining why deferral is acceptable.
+
+## Low
+
+May be deferred.
+
+Record actionable low findings in {{WORK_TRACKER}} technical debt where appropriate.
+
+Do not lower severity merely to enable a merge.
+
+---
+
+# 14. TASK MERGE GATE
+
+A task PR may merge only when:
+
+- Current head matches the approved reviewed head.
+- Review result is `APPROVED`.
+- All High findings are resolved.
+- Medium findings have a disposition.
+- Required tests/checks pass.
+- Acceptance criteria pass.
+- No unresolved conflict exists.
+- {{WORK_TRACKER}} contains the correct references.
+
+The development agent may perform the task → feature merge.
+
+After merge:
+
+1. Verify the feature branch contains the expected merge.
+2. Update {{WORK_TRACKER}}.
+3. Mark the task complete.
+4. Run `CONTROLLER_TICK`.
+
+DO NOT stop because the task completed.
+
+---
+
+# 15. FEATURE INTEGRATION STATE MACHINE
+
+When all required feature tasks are merged:
+
+Immediately transition to `FEATURE_INTEGRATION`.
+
+1. Reconcile latest `{{INTEGRATION_BRANCH}}`.
+2. Incorporate `{{INTEGRATION_BRANCH}}` according to repository policy.
+3. Assign conflicts to a developer if necessary.
+4. Run complete feature-level integration checks.
+5. Open feature PR → `{{INTEGRATION_BRANCH}}`.
+6. Dispatch fresh {{REVIEWER_VENDOR}} review.
+7. Process findings.
+8. Dispatch remediation when necessary.
+9. Rerun affected checks.
+10. Obtain fresh review of every changed head.
+11. When independently approved, dispatch the Approver.
+12. Approver verifies all gates.
+13. Approver merges feature → `{{INTEGRATION_BRANCH}}`.
+14. Verify resulting `{{INTEGRATION_BRANCH}}`.
+15. Update {{WORK_TRACKER}}.
+16. Immediately run `CONTROLLER_TICK`.
+
+A merged feature is a transition.
+
+It is terminal ONLY when:
+
+`MISSION_SCOPE = SINGLE_FEATURE`
+
+Otherwise select the next eligible work.
+
+---
+
+# 16. APPROVER CONTRACT
+
+The approver is a separate merge-control agent.
+
+Before feature → `{{INTEGRATION_BRANCH}}` merge, verify:
+
+- Feature PR head SHA.
+- Independent review applies to that exact SHA.
+- Review result is `APPROVED`.
+- Required checks pass.
+- High findings are resolved.
+- Medium findings have dispositions.
+- Acceptance criteria pass.
+- No conflicts exist.
+- {{WORK_TRACKER}} matches repository state.
+
+Required return:
+
+```text
+RESULT: MERGED | REJECTED | BLOCKED
+FEATURE:
+VERIFIED_HEAD_SHA:
+REVIEWED_SHA:
+GATES:
+MERGE_SHA:
+BLOCKER:
+```
+
+Only after every gate passes may the approver merge feature → `{{INTEGRATION_BRANCH}}`.
+
+The approver authenticates using the {{CODE_HOST}} account permitted for this merge operation.
+
+The orchestrator itself does not perform the merge.
+
+---
+
+# 17. PROJECT-QUEUE CONTINUATION
+
+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.
+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.
+
+Do not return to the user simply because one feature completed.
+
+If several features are available, choose using:
+
+1. Explicit {{WORK_TRACKER}} priority.
+2. Dependency-unblocking value.
+3. Existing partially completed work.
+4. Lowest integration/conflict risk.
+5. Repository/project ordering.
+
+---
+
+# 18. RELEASE RULES
+
+Completed features accumulate in `{{INTEGRATION_BRANCH}}`.
+
+Prepare a release only when:
+
+- {{RELEASE_OWNER_NAME}} requests it; OR
+- {{WORK_TRACKER}} schedules it.
+
+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.
+
+Only {{RELEASE_OWNER_NAME}} may merge into `{{RELEASE_BRANCH}}`.
+
+---
+
+# 19. ESCALATION
+
+Escalation is LOCAL by default.
+
+Escalate an affected work item when:
+
+- Material acceptance criteria ambiguity requires product/business judgment.
+- Required credential/service/model/access is unavailable.
+- A destructive/irreversible migration requires approval.
+- Repository policy cannot safely be satisfied.
+- Same blocker persists after three meaningful remediation attempts.
+- Requested work materially expands authorized scope.
+
+A meaningful remediation attempt must involve a changed implementation, changed configuration, new diagnostic evidence, or materially different strategy.
+
+Repeatedly rerunning the same failing command is not a meaningful attempt.
+
+When one item is escalated:
+
+1. Mark that item `LOCALLY_BLOCKED`.
+2. Record evidence and attempted remedies.
+3. Identify other unblocked in-scope work.
+4. Continue that work.
+
+Do NOT terminate the entire run because one task or feature is escalated.
+
+---
+
+# 20. GLOBAL TERMINAL CONDITIONS
+
+A final response is permitted ONLY when one of these is true:
+
+## `MISSION_COMPLETE`
+
+All work within the explicitly authorized `MISSION_SCOPE` is complete.
+
+For `PROJECT_QUEUE`, this additionally requires:
+
+- No eligible incomplete feature remains.
+- No task/review/remediation/merge is pending.
+- No development agent is active.
+- No approver is active.
+
+## `RELEASE_READY`
+
+Release PR has passed required gates and is assigned to `{{HOST_ACCOUNT}}`, awaiting {{RELEASE_OWNER_NAME}}'s {{RELEASE_BRANCH}} merge.
+
+## `GLOBAL_ESCALATION`
+
+All remaining in-scope work is blocked by decisions/access that require {{RELEASE_OWNER_NAME}}.
+
+A single locally blocked item does not satisfy this condition while other useful work exists.
+
+## `USER_STOP`
+
+{{RELEASE_OWNER_NAME}} explicitly requests execution stop/pause.
+
+No other state is terminal.
+
+---
+
+# 21. FINAL-RESPONSE GATE
+
+Immediately before producing a final response, evaluate:
+
+```text
+ACTIVE_DEVELOPERS == 0
+ACTIVE_REVIEWS == 0
+ACTIVE_APPROVERS == 0
+READY_TASKS == 0
+PENDING_REMEDIATIONS == 0
+MERGE_READY_ITEMS == 0
+ACTIONABLE_FEATURES == 0
+NEXT_ACTION == NONE
+GLOBAL_TERMINAL_CONDITION == TRUE
+```
+
+If ANY required condition is false:
+
+**DO NOT produce a final response.**
+
+Execute `NEXT_ACTION`.
+
+---
+
+# 22. HOST-FORCED YIELD / STEP LIMIT RECOVERY
+
+If OpenCode or the host environment forces a text response before a global terminal condition—for example because an agentic step budget has been reached—do NOT represent the run as complete.
+
+Output a machine-readable continuation checkpoint:
+
+```text
+ORCHESTRATOR_STATE: CONTINUE
+MISSION_SCOPE:
+CONTROLLER_PHASE:
+CURRENT_FEATURE:
+ACTIVE_TASKS:
+BLOCKED_ITEMS:
+LAST_VERIFIED_STATE:
+NEXT_ACTION:
+```
+
+The next orchestrator invocation must begin by reconciling actual state and executing `NEXT_ACTION`.
+
+`ORCHESTRATOR_STATE: CONTINUE` is explicitly nonterminal.
+
+---
+
+# 23. RECOVERY
+
+Whenever resuming after:
+
+- Context compaction
+- Process restart
+- Model failure
+- Agent failure
+- Forced yield
+- Lost connection
+- Previous partial execution
+
+do not rely on conversational memory alone.
+
+1. Read {{WORK_TRACKER}}.
+2. Inspect repository state.
+3. Inspect branches/PRs.
+4. Inspect CI/checks.
+5. Determine the last verified completed gate.
+6. Reconstruct `NEXT_ACTION`.
+7. Resume from that gate.
+
+Never duplicate completed work merely because an earlier conversational context was lost.
+
+---
+
+# 24. EFFICIENCY RULES
+
+1. Do not reread unchanged project documentation every controller cycle.
+2. Cache stable architecture/context within the current session.
+3. Re-read context only when relevant repository/project state changes.
+4. Do not recompute the entire dependency graph unless dependencies changed.
+5. Run cheap local quality checks before expensive independent review.
+6. Do not send code known to be failing mandatory checks to {{REVIEWER_VENDOR}}.
+7. Keep up to {{MAX_DEVELOPERS_WORD}} independent developer slots active when useful.
+8. Process completed agents immediately rather than waiting for a batch.
+9. Do not repeatedly ask reviewers to rediscover already-resolved findings.
+10. Include previous finding IDs in remediation review prompts.
+11. Use affected/local test suites during task development.
+12. Use full integration suites at feature/release gates.
+13. Avoid spawning agents for trivial repository-state inspection that the orchestrator can perform directly.
+14. Never use an LLM agent merely to wait.
+
+---
+
+# 25. ROLE BOUNDARIES
+
+## Orchestrator
+
+May:
+
+- Inspect.
+- Plan.
+- Create feature/release branches.
+- Maintain {{WORK_TRACKER}}.
+- Dispatch agents.
+- Verify gates.
+- Coordinate work.
+
+Must NOT:
+
+- Implement feature code.
+- Remediate code directly.
+- Perform independent review.
+- Substitute its own approval.
+- Merge task/feature/release PRs.
+
+## Development Agent
+
+May:
+
+- Implement assigned tasks.
+- Remediate findings.
+- Test.
+- Open task PRs.
+- Merge approved task PRs into feature branches.
+
+Must NOT:
+
+- Approve itself.
+- Independently review itself.
+- Merge to `{{INTEGRATION_BRANCH}}` or `{{RELEASE_BRANCH}}`.
+
+## Reviewer
+
+Read-only.
+
+Never edits or merges.
+
+## Approver
+
+Verifies final feature gates and performs feature → `{{INTEGRATION_BRANCH}}` merge.
+
+Never implements or reviews its own changes.
+
+## {{RELEASE_OWNER_NAME}} / `{{HOST_ACCOUNT}}`
+
+Only {{RELEASE_OWNER_NAME}} merges release PRs into `{{RELEASE_BRANCH}}`.
+
+---
+
+# 26. EXECUTION START
+
+At startup:
+
+1. Determine `MISSION_SCOPE`.
+2. Reconcile {{WORK_TRACKER}}.
+3. Inspect actual repository/PR state.
+4. Recover unfinished work if present.
+5. Establish `NEXT_ACTION`.
+6. Execute it.
+7. Enter `CONTROLLER_TICK`.
+8. Continue until a GLOBAL TERMINAL CONDITION is reached.
+
+Remember:
+
+**Never report a next executable step instead of performing it.**
+
+**Task completion triggers another controller cycle.**
+
+**Review completion triggers another controller cycle.**
+
+**Feature completion triggers another controller cycle unless the mission was explicitly limited to that feature.**
+
+**A locally blocked item triggers scheduling of other useful work whenever possible.**
+
+You are the controller of a persistent development pipeline. Keep the pipeline moving.
+
+---
+
+# 27. FOREGROUND TERMINAL RUNTIME
+
+This section is the execution contract for a foreground OpenCode
+orchestrator. It replaces the service runtime section in the rendered prompt
+and takes precedence over any service-specific runtime instructions later
+read from {{CONTEXT_STORE}}.
+
+The user explicitly started this run with `{{TERMINAL_LAUNCHER}}`. Begin
+work immediately in `{{PROJECT_ROOT}}`; do not start, wait for, or delegate
+control to `{{SERVICE_NAME}}`. This single OpenCode session owns the
+orchestration loop until a global terminal condition, an explicit user stop,
+or a host-forced yield.
+
+Execute `CONTROLLER_TICK` repeatedly inside this foreground session. Do not
+end a turn merely to emit `ORCHESTRATOR_STATE: CONTINUE`, and do not treat
+one task, review, pull request, or feature merge as completion. After every
+child result or external state change, reconcile {{CONTEXT_STORE}},
+{{WORK_TRACKER}}, Git, pull requests, and checks, then perform the next
+eligible action.
+
+The installed role launchers live at `{{CONTROL_HOME}}/scripts/`. Use
+`{{ROLE_LAUNCHER}}` for developer and approver sessions. Use
+`{{REVIEW_LAUNCHER}}` with WORKTREE, BASE_REF, HEAD_REF,
+REVIEW_PROMPT_FILE, PR_NUMBER, OUTPUT_FILE for a fresh read-only
+{{REVIEWER_DISPLAY_NAME}} review. Review output belongs under
+`{{SCRATCH_DIR}}`, and exact-head approval must be posted by
+`{{REVIEW_ACCOUNT}}`. The feature-to-{{INTEGRATION_BRANCH}} approver must
+use `{{MERGE_BROKER}}`; the orchestrator and developer never perform that
+merge.
+
+If the host forces a yield before a global terminal condition, return the
+machine-readable checkpoint from Section 22. The user can resume the same
+OpenCode session from the terminal. A yield is a checkpoint, not a claim that
+the project queue is complete.

templates/README.md

@@ -0,0 +1,32 @@
+# Reusing the development cycle prompt
+
+`ORCHESTRATOR-PROMPT.template.md` is the current NestQuest controller
+prompt with project-specific names, models, branches, identities, and launcher
+paths replaced by placeholders. `nestquest.profile.json` is the working
+example. Render a concrete prompt with:
+
+```bash
+scripts/render-cycle-prompt templates/nestquest.profile.json --output /tmp/opencode/nestquest-prompt.md
+```
+
+For another project, copy the profile and set its project name, tracker,
+context store, models, branch names, release owner, example IDs, review
+account, and installed launcher paths. Review the rendered result section by
+section, especially task/feature semantics, release rules, permissions,
+review gates, and the host runtime section. The template assumes an OpenCode
+controller, a feature/task tracker, a versioned context store, a code host,
+and a separate reviewer and approver. Adapt those sections when the project's
+systems differ. Section 27 is a single-session foreground loop suitable for
+`opencode --agent ... --prompt ...`; the reusable prompt has no service
+runtime.
+
+For NestQuest, `nq-orchestrate --print-prompt` emits the fully rendered
+foreground prompt without starting OpenCode. `nq-orchestrate [mission.md]`
+renders that same prompt, appends the optional mission, and supplies it to the
+foreground orchestrator agent.
+
+Publish the reviewed rendered prompt into that project's context store with
+its version check.
+Give the project its own agent definitions, broker, profile, and terminal
+launcher. The template itself does not start an orchestrator or register
+another project.

templates/nestquest.profile.json

@@ -0,0 +1,35 @@
+{
+  "LAST_UPDATED": "2026-09-22",
+  "PROJECT_CONTEXT_LABEL": "NESTQUEST HOUSE CHAOS COORDINATION",
+  "PROJECT_NAME": "NestQuest",
+  "WORK_TRACKER": "Maestro",
+  "CONTEXT_STORE": "CTXD",
+  "MODEL_PROVIDER": "OpenRouter",
+  "CONTROLLER_MODEL": "deepseek/deepseek-v4.1-flash",
+  "DEVELOPER_MODEL": "z-ai/glm-5.3-flash",
+  "REVIEWER_MODEL": "claude-sonnet-5",
+  "REVIEWER_DISPLAY_NAME": "Claude Sonnet 5",
+  "REVIEWER_HARNESS": "Claude Code",
+  "REVIEWER_VENDOR": "Claude",
+  "REVIEWER_CLI": "claude",
+  "MAX_DEVELOPERS": "2",
+  "MAX_DEVELOPERS_WORD": "two",
+  "INTEGRATION_BRANCH": "dev",
+  "RELEASE_BRANCH": "main",
+  "RELEASE_OWNER_NAME": "Joshua",
+  "HOST_ACCOUNT": "overseer",
+  "CODE_HOST": "Gitea",
+  "TASK_ID_EXAMPLE": "TASK-123",
+  "FEATURE_BRANCH_EXAMPLE": "feature/NQ-17-family-calendar",
+  "OPENCODE_VERSION": "1.18.32",
+  "PROJECT_ROOT": "/mnt/ai-storage/projects/NestQuest",
+  "CONTROL_HOME": "/home/overseer/.local/share/nestquest-controller",
+  "SERVICE_NAME": "nestquest-controller.service",
+  "TERMINAL_LAUNCHER": "nq-orchestrate",
+  "ROLE_LAUNCHER": "nq-agent",
+  "REVIEW_LAUNCHER": "nq-review",
+  "MERGE_BROKER": "nq-approve-merge",
+  "SCRATCH_DIR": "/tmp/opencode",
+  "REVIEW_ACCOUNT": "review_agent",
+  "TICK_TIMEOUT_LABEL": "six-hour"
+}

tests/test_api_admin_history.py

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

tests/test_api_admin_presence.py

@@ -0,0 +1,571 @@
+"""Admin-plane presence route tests (task e4a9d31b).
+
+These tests exercise the admin plane's presence routes against the ASGI
+app with httpx, reusing the shared local-RSA-key/stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.
+
+Every done-condition case is proven against the DATABASE and the DOMAIN,
+read through the SAME core copy the app uses (``nestquest_core.*``; see
+the two-copy caveat in api/nestquest_core.py):
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and nothing is written.
+- ``PUT /api/v1/admin/children/{id}/presence-schedule`` UPSERTS: the
+  first set persists the schedule, setting again REPLACES it — one
+  schedule row per child, never a duplicate.
+- ``POST /api/v1/admin/presence-overrides`` creates the override (its
+  ``id`` comes back) and ``DELETE /api/v1/admin/presence-overrides/{id}``
+  removes it again.
+- The child's RESOLVED presence changes accordingly: the pure
+  :class:`~nestquest_core.presence.PresenceEngine` built from the
+  stored rows answers differently before/after each write — a schedule
+  restricts the child's present weekdays, an override beats the pattern
+  for its range, and deleting the override restores the pattern's
+  answer.
+- The regeneration side effect lives in the core layer the route
+  calls: setting an all-absent schedule through the route deletes the
+  child's future open quest instances and they are NOT rebuilt while
+  the child is absent every day — the same proof the HA service test
+  makes, through the API instead.
+
+Error mapping (documented in api/routes_admin.py): a core ValueError
+for a non-existent child (schedule or override create) or override
+(delete) is 404; every other core ValueError — a rejected schedule
+(cycle length, pattern coverage, anchor date) or override (malformed or
+inverted dates, an overlapping range) — is 422, the same failure class
+FastAPI reports for malformed bodies.
+"""
+from __future__ import annotations
+
+import datetime
+import importlib
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The presence model, DAOs, business layer and materializer through the
+# SAME core copy the app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' database and resolved-
+# presence assertions only; the routes reach them through api.nestquest_core.
+_children = importlib.import_module("nestquest_core.children")
+_dao_presence = importlib.import_module("nestquest_core.dao_presence")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_materialize = importlib.import_module("nestquest_core.materialize")
+_presence = importlib.import_module("nestquest_core.presence")
+
+#: A Monday anchoring every schedule: cycle week 0 starts here.
+ANCHOR_MONDAY = "2026-01-05"
+#: The Tuesday of that week — absent under the test schedule.
+TUESDAY = "2026-01-06"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _schedule_body(
+    cycle_length_weeks: int = 1, pattern: dict[int, list[int]] | None = None
+) -> dict[str, object]:
+    """A valid schedule body anchored to the test Monday."""
+    return {
+        "cycle_length_weeks": cycle_length_weeks,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [0, 2, 4]} if pattern is None else pattern,
+    }
+
+
+def _override_body(
+    child_id: int,
+    *,
+    start_date: str = TUESDAY,
+    end_date: str = TUESDAY,
+    is_present: bool = True,
+) -> dict[str, object]:
+    """A valid single-day override body for the test child."""
+    return {
+        "child_id": child_id,
+        "start_date": start_date,
+        "end_date": end_date,
+        "is_present": is_present,
+        "note": "grandmother's week",
+    }
+
+
+async def _create_child(client: object, name: str = "Ada") -> int:
+    """Create one child through the admin children route; return the id."""
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": name},
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+def _engine(
+    database: object,
+    child_id: int,
+    schedule_record: object | None,
+    override_records: list[object],
+) -> object:
+    """Build the pure PresenceEngine from stored rows (the snapshot shape).
+
+    Mirrors what the core snapshot builder assembles: the child's
+    decoded schedule (absent → present every day) plus its overrides,
+    so :meth:`PresenceEngine.is_present` answers from the SAME rows the
+    routes produced.
+    """
+    schedules: dict[int, object] = {}
+    if schedule_record is not None:
+        schedules[child_id] = _presence.PresenceSchedule.decode(
+            schedule_record.child_id,
+            schedule_record.anchor_date,
+            schedule_record.pattern,
+        )
+    overrides = {
+        child_id: [
+            _presence.PresenceOverride(
+                record.child_id,
+                record.start_date,
+                record.end_date,
+                record.is_present,
+                note=record.note,
+            )
+            for record in override_records
+        ]
+    }
+    return _presence.PresenceEngine(schedules, overrides)
+
+
+async def _child_state(
+    database: object, child_id: int
+) -> tuple[object | None, list[object], object]:
+    """The child's stored schedule record, overrides and PresenceEngine."""
+    schedule_record = await _dao_presence.PresenceSchedulesDao(
+        database
+    ).get_by_child(child_id)
+    override_records = await _dao_presence.PresenceOverridesDao(
+        database
+    ).list_by_child_and_range(child_id, "2020-01-01", "2030-01-01")
+    engine = _engine(database, child_id, schedule_record, override_records)
+    return schedule_record, override_records, engine
+
+
+@pytest.fixture
+async def admin_presence(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth, children and quest-definition tests use
+    (stubbed JWKS, no network); the fixture also exposes the app's ONE
+    database connection so tests assert the routes' effects on the
+    stored rows and the resolved presence built from them.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_presence_routes_refuse_non_admin_and_absent_credentials(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    non_admin = make_token(groups=("some-other-group",))
+    non_admin_headers = {"Authorization": f"Bearer {non_admin}"}
+
+    forbidden_schedule = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=non_admin_headers,
+    )
+    assert forbidden_schedule.status_code == 403
+    assert forbidden_schedule.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+
+    forbidden_override = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=non_admin_headers,
+    )
+    assert forbidden_override.status_code == 403
+
+    forbidden_delete = await client.delete(
+        "/api/v1/admin/presence-overrides/1", headers=non_admin_headers
+    )
+    assert forbidden_delete.status_code == 403
+
+    unauthenticated_schedule = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+    )
+    assert unauthenticated_schedule.status_code == 401
+
+    unauthenticated_override = await client.post(
+        "/api/v1/admin/presence-overrides", json=_override_body(child_id)
+    )
+    assert unauthenticated_override.status_code == 401
+
+    unauthenticated_delete = await client.delete(
+        "/api/v1/admin/presence-overrides/1"
+    )
+    assert unauthenticated_delete.status_code == 401
+
+    # No route ran its handler: the child still has no schedule and no
+    # overrides (and is present every day, the no-schedule default).
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert _schedule is None
+    assert _overrides == []
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+
+
+# --- schedule upsert --------------------------------------------------------
+
+
+async def test_schedule_upserts_and_replaces_one_row_per_child(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """PUT /presence-schedule persists, and setting again REPLACES it."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+
+    first = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    assert first.json() == {
+        "child_id": child_id,
+        "cycle_length_weeks": 1,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [0, 2, 4]},
+    }
+    # Database proof: exactly ONE schedule row (the schema's
+    # UNIQUE(child_id) makes a second row unrepresentable), stored as
+    # encoded CSV.
+    schedule, _overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert schedule is not None
+    row_id = schedule.id
+    assert schedule.cycle_length_weeks == 1
+    assert schedule.anchor_date == ANCHOR_MONDAY
+    assert schedule.pattern == "0,2,4"
+
+    # Setting again UPSERTS: the SAME row is updated in place — new
+    # values, unchanged id — never a duplicate or a stale first set.
+    second = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(
+            cycle_length_weeks=2, pattern={"0": [], "1": [1, 3]}
+        ),
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json() == {
+        "child_id": child_id,
+        "cycle_length_weeks": 2,
+        "anchor_date": ANCHOR_MONDAY,
+        "pattern": {"0": [], "1": [1, 3]},
+    }
+    schedule, _overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert schedule is not None
+    assert schedule.id == row_id
+    assert schedule.cycle_length_weeks == 2
+    assert schedule.pattern == "|1,3"
+
+
+async def test_schedule_changes_the_childs_resolved_presence(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """The resolved presence (PresenceEngine) follows the stored schedule."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+
+    # Before any schedule the child is present every day.
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+    assert engine.is_present(child_id, TUESDAY) is True
+
+    # Mon/Wed/Fri only, week 0 of a one-week cycle.
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    _schedule, _overrides, engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+    assert engine.is_present(child_id, TUESDAY) is False
+
+
+async def test_schedule_upsert_regenerates_future_instances(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """Setting an all-absent schedule deletes the child's future instances.
+
+    The regeneration side effect lives in the core layer the route
+    calls: the child's open instances at or after today are deleted
+    across ALL its definitions and NOT rebuilt while the child is
+    absent every day — the same proof the HA service test makes,
+    exercised here through the API route instead.
+    """
+    client = admin_presence.client
+    database = admin_presence.database
+    child_id = await _create_child(client)
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Brush teeth",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [child_id],
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+
+    today = datetime.date.today()
+    start = today.isoformat()
+    end = (today + datetime.timedelta(days=14)).isoformat()
+    await _materialize.materialize(database, start, end, today=today)
+    instances = _dao_instances.QuestInstancesDao(database)
+    assert await instances.list_by_date_range(child_id, start, end)
+
+    # The route's write must regenerate: absent every day removes every
+    # open future instance.
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(pattern={"0": []}),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    assert await instances.list_by_date_range(child_id, start, end) == []
+
+
+# --- override create + delete -----------------------------------------------
+
+
+async def test_override_is_created_then_deleted_and_presence_resolves(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """POST creates the override, DELETE removes it; presence follows."""
+    client = admin_presence.client
+    database = admin_presence.database
+    child_id = await _create_child(client)
+    schedule_response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert schedule_response.status_code == 200
+
+    # Create: the override comes back with its id and the posted shape.
+    created = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    payload = created.json()
+    assert set(payload) == {
+        "id",
+        "child_id",
+        "start_date",
+        "end_date",
+        "is_present",
+        "note",
+    }
+    assert payload["child_id"] == child_id
+    assert payload["start_date"] == TUESDAY
+    assert payload["end_date"] == TUESDAY
+    assert payload["is_present"] is True
+    assert payload["note"] == "grandmother's week"
+
+    # Database proof: the row exists — and the RESOLVED presence for the
+    # affected date flips: the override beats the Tuesday-absent pattern.
+    _schedule, overrides, engine = await _child_state(database, child_id)
+    assert [record.id for record in overrides] == [payload["id"]]
+    assert engine.is_present(child_id, TUESDAY) is True
+    # A date the override does not cover keeps the pattern's answer.
+    assert engine.is_present(child_id, ANCHOR_MONDAY) is True
+
+    # Delete: the row is gone and the resolved presence follows the
+    # pattern again.
+    deleted = await client.delete(
+        f"/api/v1/admin/presence-overrides/{payload['id']}",
+        headers=_admin_headers(),
+    )
+    assert deleted.status_code == 200
+    assert deleted.json() == {"status": "ok"}
+    assert (
+        await _dao_presence.PresenceOverridesDao(database).get(payload["id"])
+        is None
+    )
+    _schedule, overrides, engine = await _child_state(database, child_id)
+    assert overrides == []
+    assert engine.is_present(child_id, TUESDAY) is False
+
+
+# --- error mapping: 404 vs 422 ----------------------------------------------
+
+
+async def test_schedule_and_override_for_unknown_child_are_404(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """A presence write naming a child that does not exist is 404."""
+    client = admin_presence.client
+
+    schedule = await client.put(
+        "/api/v1/admin/children/999/presence-schedule",
+        json=_schedule_body(),
+        headers=_admin_headers(),
+    )
+    assert schedule.status_code == 404
+    assert schedule.json()["detail"] == "Child not found"
+
+    override = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(999),
+        headers=_admin_headers(),
+    )
+    assert override.status_code == 404
+    assert override.json()["detail"] == "Child not found"
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        _schedule_body(cycle_length_weeks=0),  # below the 1..4 range
+        _schedule_body(cycle_length_weeks=5),  # above the documented cap
+        # week 1 of the cycle missing:
+        _schedule_body(cycle_length_weeks=2, pattern={"0": [0]}),
+        {
+            "cycle_length_weeks": 1,
+            "anchor_date": "2026-13-40",
+            "pattern": {"0": [0]},
+        },
+        _schedule_body(pattern={"0": [7]}),  # weekday out of 0..6
+        _schedule_body(pattern={"nope": [0]}),  # week index not an integer
+        _schedule_body(pattern={"0": "0,2,4"}),  # weekdays not a list
+    ],
+)
+async def test_rejected_schedule_is_422(
+    admin_presence: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """A rejected schedule is 422 and stores nothing."""
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    response = await client.put(
+        f"/api/v1/admin/children/{child_id}/presence-schedule",
+        json=body,
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    assert (
+        await _dao_presence.PresenceSchedulesDao(
+            admin_presence.database
+        ).get_by_child(child_id)
+        is None
+    )
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        _override_body(1, start_date="2026-01-07", end_date="2026-01-06"),
+        _override_body(1, start_date="2026-02-30", end_date="2026-02-30"),
+        dict(_override_body(1), note=5),
+        dict(_override_body(1), child_id="one"),
+    ],
+)
+async def test_rejected_override_is_422(
+    admin_presence: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """A rejected override (inverted/malformed dates, bad note/id) is 422.
+
+    The shape failures (a non-string note, a non-integer child id) are
+    the body model's; the date failures are the PresenceOverride
+    model's.  A string ``is_present`` like ``"yes"`` is deliberately
+    absent: Pydantic's lax bool COERCES it to ``True`` (the same design
+    as the children active route), so it never reaches the model's
+    real-bool check.
+    """
+    response = await admin_presence.client.post(
+        "/api/v1/admin/presence-overrides", json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422
+
+
+async def test_overlapping_override_is_422_and_first_survives(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """An override overlapping the child's existing one is 422.
+
+    The conflicting write persists nothing: the child's original
+    override survives unchanged.
+    """
+    client = admin_presence.client
+    child_id = await _create_child(client)
+    created = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id),
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+
+    response = await client.post(
+        "/api/v1/admin/presence-overrides",
+        json=_override_body(child_id, is_present=False),
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 422
+    _schedule, overrides, _engine = await _child_state(
+        admin_presence.database, child_id
+    )
+    assert [record.id for record in overrides] == [created.json()["id"]]
+
+
+async def test_delete_unknown_override_is_404(
+    admin_presence: SimpleNamespace,
+) -> None:
+    """Deleting an override id that does not exist is 404."""
+    response = await admin_presence.client.delete(
+        "/api/v1/admin/presence-overrides/999", headers=_admin_headers()
+    )
+    assert response.status_code == 404
+    assert response.json()["detail"] == "Presence override not found"
+
+    malformed = await admin_presence.client.delete(
+        "/api/v1/admin/presence-overrides/not-an-id",
+        headers=_admin_headers(),
+    )
+    assert malformed.status_code == 422

tests/test_api_admin_settings.py

@@ -0,0 +1,398 @@
+"""Admin-plane settings route tests (task 2b3de7e5).
+
+These tests exercise the admin plane's settings read and update routes
+against the ASGI app with httpx, reusing the shared local-RSA-key/
+stubbed-JWKS harness (:mod:`tests.admin_jwt_harness`) so an admin JWT
+genuinely authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.  The stored settings
+document is read back through the SAME core copy the app uses
+(``nestquest_core.*``; see the two-copy caveat in api/nestquest_core.py)
+via its :class:`nestquest_core.dao_meta.MetaStateDao`, so persistence is
+proven against the DATABASE the routes wrote, not a parallel copy.
+
+Proven per the done-condition, everything read back through the API:
+
+- ``GET /api/v1/admin/settings`` returns the current effective settings
+  — the ten documented fields at their defaults on a fresh database.
+- ``PATCH /api/v1/admin/settings`` PERSISTS the supplied fields and the
+  NEXT read reflects them; a later partial update merges onto the
+  current effective settings (omitted fields keep their stored value),
+  and the document is stored under the well-known ``nestquest_meta_state``
+  key.
+- An invalid value (a non-positive or boolean horizon, a non-strict
+  time, a non-bool toggle, a non-string target) is rejected with 422
+  naming the field, and the stored settings are UNCHANGED — the next
+  read still shows the previous value.  An unknown field is rejected
+  the same way, also when mixed with a valid field (validate
+  everything before writing).
+- A corrupt stored document (invalid JSON, a non-object, an invalid
+  field value) does not crash the service: the read falls back to the
+  defaults, per field where possible.
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: a non-admin JWT is 403 and an absent credential is 401
+  on both routes — and no refused patch writes anything.
+"""
+from __future__ import annotations
+
+import importlib
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    KID,
+    AdminRunner,
+    jwks_for,
+    make_token,
+)
+
+# The meta-state DAO through the SAME core copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy) —
+# used for the tests' database assertions only; the routes reach it
+# through api.nestquest_core.
+_dao_meta = importlib.import_module("nestquest_core.dao_meta")
+
+#: The well-known ``nestquest_meta_state`` key the API stores its
+#: settings under — asserted literally so the contract is pinned here,
+#: independent of the core constant.
+SETTINGS_META_KEY = "settings"
+
+#: The all-defaults effective settings a fresh database reports — the
+#: documented ten fields, asserted literally so the payload contract is
+#: pinned here, independent of the core constants.
+DEFAULT_SETTINGS = {
+    "horizon_days": 14,
+    "day_rollover_time": "00:00",
+    "notify_target": "",
+    "morning_summary_time": "08:00",
+    "afternoon_reminder_time": "15:00",
+    "end_of_day_report_time": "20:00",
+    "morning_summary_enabled": True,
+    "afternoon_reminder_enabled": True,
+    "end_of_day_report_enabled": True,
+    "celebration_enabled": True,
+}
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+@pytest.fixture
+async def settings_client(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the children tests use (stubbed JWKS, no network);
+    the fixture also exposes the app's ONE database connection so tests
+    can read (or corrupt) the stored settings document itself.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+# --- read: the effective settings, defaults on a fresh database --------
+
+
+async def test_get_settings_returns_defaults_on_a_fresh_database(
+    settings_client: SimpleNamespace,
+) -> None:
+    """GET returns the ten documented fields at their defaults."""
+    response = await settings_client.client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+
+# --- update: persists, and the NEXT read reflects it --------------------
+
+
+async def test_update_persists_and_the_next_read_reflects_it(
+    settings_client: SimpleNamespace,
+) -> None:
+    """PATCH persists the supplied fields; GET reads them back."""
+    client = settings_client.client
+    changes = {
+        "horizon_days": 21,
+        "day_rollover_time": "03:30",
+        "notify_target": "notify.mobile_app_dads_phone",
+        "celebration_enabled": False,
+    }
+    patched = await client.patch(
+        "/api/v1/admin/settings", json=changes, headers=_admin_headers()
+    )
+    assert patched.status_code == 200
+    expected = {**DEFAULT_SETTINGS, **changes}
+    assert patched.json() == expected
+
+    # The NEXT read reflects the persisted settings...
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.status_code == 200
+    assert reread.json() == expected
+
+    # ...and the DATABASE holds the merged document: the complete ten
+    # fields, keyed by field name, under the well-known meta_state key.
+    raw = await _dao_meta.MetaStateDao(settings_client.database).get(
+        SETTINGS_META_KEY
+    )
+    assert raw is not None
+    assert json.loads(raw) == expected
+
+
+async def test_partial_update_merges_onto_the_current_settings(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A later PATCH changes only its own fields; the rest survive."""
+    client = settings_client.client
+    first = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7, "notify_target": " notify.x "},
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    # The notify target went through the core's own normalizer: stripped.
+    assert first.json()["horizon_days"] == 7
+    assert first.json()["notify_target"] == "notify.x"
+
+    second = await client.patch(
+        "/api/v1/admin/settings",
+        json={"morning_summary_time": "07:45"},
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json() == {
+        **DEFAULT_SETTINGS,
+        "horizon_days": 7,
+        "notify_target": "notify.x",
+        "morning_summary_time": "07:45",
+    }
+
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.json() == second.json()
+
+
+async def test_empty_update_is_a_noop(
+    settings_client: SimpleNamespace,
+) -> None:
+    """PATCHing no fields leaves the effective settings as they are."""
+    client = settings_client.client
+    response = await client.patch(
+        "/api/v1/admin/settings", json={}, headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+
+async def test_null_resets_a_field_to_its_default(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A JSON ``null`` maps to the field's default (None-as-absent)."""
+    client = settings_client.client
+    set_horizon = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7},
+        headers=_admin_headers(),
+    )
+    assert set_horizon.status_code == 200
+    assert set_horizon.json()["horizon_days"] == 7
+
+    reset = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": None},
+        headers=_admin_headers(),
+    )
+    assert reset.status_code == 200
+    assert reset.json()["horizon_days"] == 14
+
+    reread = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert reread.json()["horizon_days"] == 14
+
+
+# --- validation: rejected values never change the stored settings -------
+
+
+@pytest.mark.parametrize(
+    ("changes", "field"),
+    [
+        ({"horizon_days": 0}, "horizon_days"),
+        ({"horizon_days": -3}, "horizon_days"),
+        # A bool is rejected — the core's guardrail survives the route.
+        ({"horizon_days": True}, "horizon_days"),
+        ({"horizon_days": "soon"}, "horizon_days"),
+        ({"day_rollover_time": "25:00"}, "day_rollover_time"),
+        ({"day_rollover_time": "7:30"}, "day_rollover_time"),
+        ({"morning_summary_enabled": "yes"}, "morning_summary_enabled"),
+        ({"celebration_enabled": 1}, "celebration_enabled"),
+        ({"notify_target": 5}, "notify_target"),
+    ],
+)
+async def test_invalid_value_is_rejected_and_stored_settings_unchanged(
+    settings_client: SimpleNamespace, changes: dict[str, object], field: str
+) -> None:
+    """A rejected value is 422 naming the field; the next read is same."""
+    client = settings_client.client
+    seeded = await client.patch(
+        "/api/v1/admin/settings",
+        json={
+            "horizon_days": 21,
+            "day_rollover_time": "01:30",
+            "morning_summary_enabled": False,
+        },
+        headers=_admin_headers(),
+    )
+    assert seeded.status_code == 200
+    before = seeded.json()
+
+    response = await client.patch(
+        "/api/v1/admin/settings", json=changes, headers=_admin_headers()
+    )
+    assert response.status_code == 422, changes
+    assert field in str(response.json()["detail"]), changes
+
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.status_code == 200
+    assert after.json() == before
+
+
+async def test_unknown_field_is_rejected_and_stored_settings_unchanged(
+    settings_client: SimpleNamespace,
+) -> None:
+    """An unknown field is 422 naming it — alone or beside a valid one."""
+    client = settings_client.client
+    unknown = await client.patch(
+        "/api/v1/admin/settings",
+        json={"bogus_setting": 1},
+        headers=_admin_headers(),
+    )
+    assert unknown.status_code == 422
+    assert "bogus_setting" in str(unknown.json()["detail"])
+
+    mixed = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7, "bogus_setting": 1},
+        headers=_admin_headers(),
+    )
+    assert mixed.status_code == 422
+    assert "bogus_setting" in str(mixed.json()["detail"])
+
+    # Neither rejection wrote anything, not even the valid sibling.
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS
+
+
+async def test_non_object_body_is_422(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A body that is not a JSON object is FastAPI's 422 (the shape)."""
+    client = settings_client.client
+    for body in ([1, 2], "nope", 5):
+        response = await client.patch(
+            "/api/v1/admin/settings", json=body, headers=_admin_headers()
+        )
+        assert response.status_code == 422, body
+    # A bare JSON null is likewise not an object.
+    response = await client.patch(
+        "/api/v1/admin/settings",
+        content=b"null",
+        headers={**_admin_headers(), "content-type": "application/json"},
+    )
+    assert response.status_code == 422
+
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS
+
+
+# --- resilience: a corrupt stored document does not crash the service ---
+
+
+async def test_corrupt_stored_document_falls_back_to_defaults(
+    settings_client: SimpleNamespace,
+) -> None:
+    """GET degrades gracefully over an unusable settings document."""
+    client = settings_client.client
+    dao = _dao_meta.MetaStateDao(settings_client.database)
+
+    # Not JSON at all: all defaults, no crash.
+    await dao.set(SETTINGS_META_KEY, "{not-json")
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+    # JSON but not an object: all defaults, no crash.
+    await dao.set(SETTINGS_META_KEY, json.dumps([1, 2]))
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == DEFAULT_SETTINGS
+
+    # An object with one invalid field: PER-FIELD fallback — the valid
+    # sibling survives, the invalid one falls back to its default.
+    await dao.set(
+        SETTINGS_META_KEY,
+        json.dumps({"horizon_days": 7, "day_rollover_time": "soon"}),
+    )
+    response = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert response.status_code == 200
+    assert response.json() == {**DEFAULT_SETTINGS, "horizon_days": 7}
+
+
+# --- authentication: the ONE require_admin dependency guards both routes
+
+
+async def test_settings_routes_refuse_non_admin_and_absent_credentials(
+    settings_client: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401; none write."""
+    client = settings_client.client
+    non_admin = make_token(groups=("some-other-group",))
+    forbidden_get = await client.get(
+        "/api/v1/admin/settings",
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden_get.status_code == 403
+    assert forbidden_get.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    forbidden_patch = await client.patch(
+        "/api/v1/admin/settings",
+        json={"horizon_days": 7},
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden_patch.status_code == 403
+
+    absent_get = await client.get("/api/v1/admin/settings")
+    assert absent_get.status_code == 401
+    absent_patch = await client.patch(
+        "/api/v1/admin/settings", json={"horizon_days": 7}
+    )
+    assert absent_patch.status_code == 401
+
+    # No refused request changed anything: the settings are untouched.
+    after = await client.get(
+        "/api/v1/admin/settings", headers=_admin_headers()
+    )
+    assert after.json() == DEFAULT_SETTINGS

tests/test_api_admin_uncomplete.py

@@ -0,0 +1,694 @@
+"""Admin-plane uncomplete and regenerate route tests (task da0226b3).
+
+These tests exercise the admin plane's completion-reversal and
+re-materialization routes against the ASGI app with httpx, reusing the
+shared local-RSA-key/stubbed-JWKS harness
+(:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
+authenticates — signature, issuer, audience, expiry and the
+nestquest-admins group — with NO network access.  The household is
+seeded with :func:`tests.test_api_panel._seed_household` through the
+SAME core copy the app uses (``nestquest_core.*``; the two-copy caveat
+in api/nestquest_core.py), and the route's ONE clock read
+(:func:`api.routes_admin._local_now`) is pinned the same way the panel
+route's is, so seeded rows and route stamps cannot straddle midnight.
+
+Every done-condition case is proven against the DATABASE, read through
+that same core copy:
+
+- Authentication runs through the router's ONE ``require_admin``
+  dependency: an admin JWT passes, a JWT without the group is 403, an
+  absent credential is 401 — and nothing is written.
+- ``POST /api/v1/admin/instances/{id}/uncomplete``: the completion
+  events log for the instance holds the ORIGINAL ``completed`` row
+  UNTOUCHED plus ONE appended ``uncompleted`` row attributed to the
+  verified token's ``sub`` (actor_source ``user``), and the instance's
+  DERIVED state is ``open`` again; a second uncomplete is a no-op that
+  appends nothing.
+- On an ACTUAL reversal the ``nestquest_quest_uncompleted`` transition
+  arrives on the SSE stream exactly once (subscribed live as
+  :mod:`tests.test_api_sse` does); a no-op uncomplete publishes
+  nothing.
+- ``POST /api/v1/admin/regenerate`` re-materializes without
+  duplicating: running it twice (household, definition and child
+  scopes) leaves the instance count stable, and the scoped variants
+  rebuild only their own scope's rows.
+
+Error mapping (documented in api/routes_admin.py): an unknown
+instance is 404 (the core's ``instance_id:`` prefix, the same key the
+panel complete route uses); unknown definition/child scope ids are 404
+from the route's own existence reads; malformed input — a bad scope,
+a scope/id mismatch, a zero id, a non-integer path id — is 422 before
+any handler code runs.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import importlib
+from types import SimpleNamespace
+from zoneinfo import ZoneInfo
+
+import httpx
+import pytest
+
+from api import routes_admin
+from api.app import create_app
+from api.auth import JwksCache
+from api.config import ApiConfig
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    AUDIENCE,
+    ISSUER,
+    JWKS_URL,
+    KID,
+    AdminRunner,
+    StubbedFetch,
+    SUBJECT,
+    jwks_for,
+    make_token,
+)
+from tests.test_api_panel import PANEL_TOKEN, _seed_household
+from tests.test_api_sse import _EventStream, _Server, _stream
+
+# The completion and instance-DAO layers through the SAME core copy the
+# app uses (nestquest_core.*, NOT the conftest-stubbed
+# custom_components copy) — used for the tests' database assertions
+# only; the routes reach them through api.nestquest_core.
+_completion = importlib.import_module("nestquest_core.completion")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+
+#: The documented quest-uncompleted event name (core const, §3).
+EVENT_QUEST_UNCOMPLETED = importlib.import_module(
+    "nestquest_core.const"
+).EVENT_QUEST_UNCOMPLETED
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+def _pinned_noon() -> tuple[datetime.date, datetime.datetime]:
+    """One host clock read pinned to 12:00 UTC that day.
+
+    The seed's completion and the route's clock both anchor to this
+    instant, so nothing can straddle midnight mid-test.
+    """
+    today = datetime.datetime.now().astimezone().date()
+    return today, datetime.datetime(
+        today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
+    )
+
+
+async def _instance_count(database: object) -> int:
+    """The whole ``quest_instances`` row count (the sanctioned probe)."""
+    row = await database.fetch_one("SELECT COUNT(*) FROM quest_instances")
+    return row[0]
+
+
+async def _events_for_instance(
+    database: object, instance_id: int
+) -> list:
+    """The instance's completion events, in append order."""
+    dao = _dao_instances.CompletionEventsDao(database)
+    return await dao.list_by_instance(instance_id)
+
+
+async def _derived_state(
+    database: object, instance_id: int, today: datetime.date
+) -> str:
+    """The instance's derived state through the app's core copy."""
+    dao = _dao_instances.QuestInstancesDao(database)
+    instance = await dao.get_by_id(instance_id)
+    assert instance is not None
+    latest = await _dao_instances.CompletionEventsDao(
+        database
+    ).get_latest_for_instance(instance_id)
+    return _completion.derive_state(instance, latest, today)
+
+
+@pytest.fixture
+async def admin_api(temp_db_path: str) -> SimpleNamespace:
+    """An admin-plane app client plus its live database.
+
+    The same runner the auth and children tests use (stubbed JWKS, no
+    network); the fixture exposes the app's ONE database connection so
+    tests assert the routes' effects on the stored rows themselves.
+    """
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            database=runner.app.state.db.database,
+        )
+
+
+@pytest.fixture
+async def admin_household(temp_db_path: str, monkeypatch) -> SimpleNamespace:
+    """A seeded household served by the admin-plane app, clock pinned.
+
+    ``_seed_household`` completes Bo's today instance (17:00 due, 12:00
+    pinned — on time), which is the completion the uncomplete tests
+    reverse.  The route's ``_local_now`` is pinned to the same instant.
+    """
+    today, pinned_now = _pinned_noon()
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    monkeypatch.setattr(routes_admin, "_local_now", lambda: pinned_now)
+    runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
+    async with runner as client:
+        yield SimpleNamespace(
+            client=client,
+            seed=seed,
+            database=runner.app.state.db.database,
+            today=today,
+        )
+
+
+class _AdminServer(_Server):
+    """:mod:`tests.test_api_sse`'s live-uvicorn server, admin-enabled.
+
+    Same ephemeral-port uvicorn thread (httpx's ASGI transport buffers
+    SSE bodies, so a live server is the only way to stream), but the
+    app factory swaps the JWKS cache for one built around the harness's
+    stub fetcher, so a real admin JWT verifies with NO network access.
+    """
+
+    def _make_app_factory(self, db_path: str):
+        """The uvicorn app factory with the stubbed JWKS cache."""
+
+        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
+
+
+@pytest.fixture
+async def admin_sse(temp_db_path: str) -> SimpleNamespace:
+    """A seeded household served by a live uvicorn app, clock pinned.
+
+    Same pinning discipline as the panel tests, applied to
+    :func:`api.routes_admin._local_now` BEFORE the server starts (the
+    app is created inside uvicorn's factory thread).
+    """
+    today, pinned_now = _pinned_noon()
+    seed = await _seed_household(temp_db_path, today, pinned_now)
+    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,
+                    database=server.app.state.db.database,
+                    today=today,
+                )
+    finally:
+        routes_admin._local_now = original
+
+
+# --- authentication: the ONE require_admin dependency guards every route ---
+
+
+async def test_uncomplete_and_regenerate_refuse_non_admin_and_absent(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401: nothing written."""
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    non_admin = make_token(groups=("some-other-group",))
+    events_before = await _events_for_instance(database, brush.id)
+    instances_before = await _instance_count(database)
+
+    uncomplete_url = f"/api/v1/admin/instances/{brush.id}/uncomplete"
+    forbidden = await client.post(
+        uncomplete_url,
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    unauthenticated = await client.post(uncomplete_url)
+    assert unauthenticated.status_code == 401
+
+    regenerate_bodies = (
+        {},
+        {"scope": "household"},
+        {"scope": "child", "child_id": 1},
+    )
+    for body in regenerate_bodies:
+        forbidden = await client.post(
+            "/api/v1/admin/regenerate",
+            json=body,
+            headers={"Authorization": f"Bearer {non_admin}"},
+        )
+        assert forbidden.status_code == 403, body
+        unauthenticated = await client.post(
+            "/api/v1/admin/regenerate", json=body
+        )
+        assert unauthenticated.status_code == 401, body
+
+    # No handler ran: the seeded completion is still the ONLY event for
+    # the instance and the instance table is untouched.
+    assert (
+        await _events_for_instance(database, brush.id) == events_before
+    )
+    assert await _instance_count(database) == instances_before
+
+
+# --- uncomplete: the append-only reversal -----------------------------------
+
+
+async def test_uncomplete_appends_reversal_and_derived_state_is_open(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A reversal APPENDS an uncompleted event; the instance is open again.
+
+    The original completed row is proven byte-identical after the call
+    (frozen-dataclass equality), the new row carries the verified
+    token's ``sub`` as the user actor, and the DERIVED state —
+    recomputed from the latest event — is ``open`` again.
+    """
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    today = admin_household.today
+
+    before = await _events_for_instance(database, brush.id)
+    assert [event.event_type for event in before] == ["completed"]
+    assert await _derived_state(database, brush.id, today) == "done"
+
+    response = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body == {
+        "instance_id": brush.id,
+        "state": "open",
+        "appended": True,
+    }
+
+    events = await _events_for_instance(database, brush.id)
+    assert len(events) == 2
+    # The original completed row is UNTOUCHED — never edited, never
+    # deleted (append-only completion_events).
+    assert events[0] == before[0]
+    assert events[0].event_type == "completed"
+    assert events[0].actor_source == "panel"
+    # The reversal is the NEW row, attributed to the verified token.
+    reversal = events[1]
+    assert reversal.event_type == "uncompleted"
+    assert reversal.instance_id == brush.id
+    assert reversal.child_id == brush.child_id
+    assert reversal.actor_source == "user"
+    assert reversal.actor_user_id == SUBJECT
+    assert reversal.actor_child_id is None
+    assert reversal.was_on_time is None
+    # The derived state is recomputed from the latest event: open again.
+    assert await _derived_state(database, brush.id, today) == "open"
+
+
+async def test_second_uncomplete_is_a_noop_that_appends_nothing(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A repeat uncomplete appends NOTHING and still succeeds."""
+    client = admin_household.client
+    database = admin_household.database
+    brush = admin_household.seed.brush_teeth_instance
+    today = admin_household.today
+
+    first = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    assert first.json()["appended"] is True
+
+    second = await client.post(
+        f"/api/v1/admin/instances/{brush.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    body = second.json()
+    assert body["appended"] is False
+    assert body["state"] == "open"
+
+    events = await _events_for_instance(database, brush.id)
+    assert len(events) == 2
+    assert [event.event_type for event in events] == [
+        "completed",
+        "uncompleted",
+    ]
+    assert await _derived_state(database, brush.id, today) == "open"
+
+
+async def test_uncomplete_unknown_instance_is_404_and_malformed_is_422(
+    admin_household: SimpleNamespace,
+) -> None:
+    """An unknown instance is 404; a non-integer path id is 422."""
+    client = admin_household.client
+    brush = admin_household.seed.brush_teeth_instance
+    before = await _events_for_instance(
+        admin_household.database, brush.id
+    )
+
+    unknown = await client.post(
+        "/api/v1/admin/instances/999/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert unknown.status_code == 404
+    assert unknown.json()["detail"] == "Quest instance not found"
+
+    malformed = await client.post(
+        "/api/v1/admin/instances/not-an-int/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert malformed.status_code == 422
+
+    # Neither rejection wrote anything.
+    assert (
+        await _events_for_instance(admin_household.database, brush.id)
+        == before
+    )
+
+
+async def test_uncomplete_already_open_past_due_reports_missed(
+    admin_household: SimpleNamespace,
+) -> None:
+    """A no-op uncomplete still reports the honest DERIVED state.
+
+    The seeded stale instance is open and past due: uncompleting it is
+    a no-op (nothing appended) and the derived state is ``missed`` —
+    open-but-past-due, never an event type.
+    """
+    client = admin_household.client
+    database = admin_household.database
+    stale = admin_household.seed.stale_instance
+    today = admin_household.today
+    assert await _derived_state(database, stale.id, today) == "missed"
+
+    response = await client.post(
+        f"/api/v1/admin/instances/{stale.id}/uncomplete",
+        headers=_admin_headers(),
+    )
+    assert response.status_code == 200
+    body = response.json()
+    assert body["appended"] is False
+    assert body["state"] == "missed"
+
+    # The no-op appended nothing: the instance still has NO events.
+    assert await _events_for_instance(database, stale.id) == []
+
+
+# --- the SSE transition on an actual reversal --------------------------------
+
+
+async def test_uncomplete_publishes_exactly_one_uncompleted_transition(
+    admin_sse: SimpleNamespace,
+) -> None:
+    """An actual reversal streams ONE quest-uncompleted event; a no-op none.
+
+    The stream is subscribed FIRST (test_api_sse's live-server reader),
+    the completed instance is reversed through the admin route, and the
+    documented payload fields are asserted; the no-op second call adds
+    no frame.
+    """
+    seed = admin_sse.seed
+    brush = seed.brush_teeth_instance
+    stream = _stream(admin_sse.server.url("/api/v1/panel/events"))
+    async with stream:
+        first = await admin_sse.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=_admin_headers(),
+        )
+        assert first.status_code == 200
+        assert first.json()["appended"] is True
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_UNCOMPLETED
+        assert event_type == "nestquest_quest_uncompleted"
+        assert payload["child_id"] == seed.bo.id
+        assert payload["child_name"] == "Bo"
+        assert payload["instance_id"] == brush.id
+        assert payload["quest_title"] == "Brush teeth"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == brush.due_date
+        assert payload["due_time"] == "17:00"
+        assert payload["occurred_at"]  # strict UTC ISO-8601 stamp
+        assert "was_on_time" not in payload
+
+        # The no-op second call publishes nothing: the stream stays
+        # silent past the window.
+        second = await admin_sse.client.post(
+            f"/api/v1/admin/instances/{brush.id}/uncomplete",
+            headers=_admin_headers(),
+        )
+        assert second.status_code == 200
+        assert second.json()["appended"] is False
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+
+# --- regenerate: re-materialize without duplicating --------------------------
+
+
+async def _create_child(client: object, name: str) -> int:
+    """Create one child through the admin children route."""
+    created = await client.post(
+        "/api/v1/admin/children",
+        json={"display_name": name},
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def _create_daily_definition(
+    client: object, assignee_ids: list[int]
+) -> int:
+    """Create one daily one-window definition for the given assignees."""
+    created = await client.post(
+        "/api/v1/admin/quest-definitions",
+        json={
+            "title": "Tidy the den",
+            "rule": {"rule_type": "daily", "start_date": "2026-03-02"},
+            "assignee_child_ids": assignee_ids,
+            "windows": ["morning"],
+        },
+        headers=_admin_headers(),
+    )
+    assert created.status_code == 201
+    return created.json()["id"]
+
+
+async def test_regenerate_household_never_duplicates(
+    admin_household: SimpleNamespace,
+) -> None:
+    """Two household regenerations leave the instance count stable.
+
+    The first call re-materializes the rolling horizon
+    ``[today, today + DEFAULT_HORIZON_DAYS]``; the second walks the
+    SAME tuples and upserts them in place — the row count never grows
+    and the core reports the same upsert count both times.
+    """
+    client = admin_household.client
+    database = admin_household.database
+
+    before = await _instance_count(database)
+    assert before > 0  # the seed already materialized a few days
+
+    first = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert first.status_code == 200
+    body = first.json()
+    assert body["scope"] == "household"
+    assert body["definition_id"] is None
+    assert body["child_id"] is None
+    assert body["count"] >= before
+    after_first = await _instance_count(database)
+    assert after_first >= before
+
+    second = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert second.status_code == 200
+    assert second.json()["count"] == body["count"]
+    # The idempotent walk: the second run's row count is EXACTLY the
+    # first's — no duplicated instances.
+    assert await _instance_count(database) == after_first
+
+
+async def test_regenerate_definition_scope_rematerializes(
+    admin_api: SimpleNamespace,
+) -> None:
+    """The definition scope rebuilds that definition's horizon, twice, stably.
+
+    Creating a definition through the admin route materializes NOTHING
+    (the core's create path does not regenerate); the scoped regenerate
+    call is what materializes its instances, and re-running it leaves
+    the row count exactly where the first run left it.
+    """
+    client = admin_api.client
+    database = admin_api.database
+    ada = await _create_child(client, "Ada")
+    definition_id = await _create_daily_definition(client, [ada])
+
+    assert await _instance_count(database) == 0
+
+    first = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": definition_id},
+        headers=_admin_headers(),
+    )
+    assert first.status_code == 200
+    body = first.json()
+    assert body["scope"] == "definition"
+    assert body["definition_id"] == definition_id
+    assert body["child_id"] is None
+    assert body["count"] > 0
+    after_first = await _instance_count(database)
+    assert after_first == body["count"]
+
+    second = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": definition_id},
+        headers=_admin_headers(),
+    )
+    assert second.status_code == 200
+    assert second.json()["count"] == body["count"]
+    assert await _instance_count(database) == after_first
+
+
+async def test_regenerate_child_scope_touches_only_that_child(
+    admin_api: SimpleNamespace,
+) -> None:
+    """The child scope rebuilds that child's rows and no others'.
+
+    Two children share one daily definition; the household walk
+    materializes both.  A child-scoped regenerate for Ada re-runs her
+    side of the walk (same count) while Bo's rows keep their identity,
+    and re-running it never grows the table.
+    """
+    client = admin_api.client
+    database = admin_api.database
+    instances = _dao_instances.QuestInstancesDao(database)
+    ada = await _create_child(client, "Ada")
+    bo = await _create_child(client, "Bo")
+    # One daily definition assigned to BOTH children: the household walk
+    # gives each child its own rows.
+    await _create_daily_definition(client, [ada, bo])
+
+    # Materialize the household first so BOTH children have rows.
+    seeded = await client.post(
+        "/api/v1/admin/regenerate", json={}, headers=_admin_headers()
+    )
+    assert seeded.status_code == 200
+    assert seeded.json()["scope"] == "household"
+
+    ada_rows = await instances.list_by_date_range(
+        ada, "2026-03-02", "2100-01-01"
+    )
+    bo_rows = await instances.list_by_date_range(
+        bo, "2026-03-02", "2100-01-01"
+    )
+    assert ada_rows and bo_rows
+    total = await _instance_count(database)
+    assert total == len(ada_rows) + len(bo_rows)
+
+    for expected_count in (len(ada_rows), len(ada_rows)):
+        response = await client.post(
+            "/api/v1/admin/regenerate",
+            json={"scope": "child", "child_id": ada},
+            headers=_admin_headers(),
+        )
+        assert response.status_code == 200
+        body = response.json()
+        assert body == {
+            "scope": "child",
+            "definition_id": None,
+            "child_id": ada,
+            "count": expected_count,
+        }
+        # Ada's side re-materialized the same number of rows and Bo's
+        # rows kept their identity; the table never grew.
+        assert await instances.list_by_date_range(
+            ada, "2026-03-02", "2100-01-01"
+        )
+        assert await instances.list_by_date_range(
+            bo, "2026-03-02", "2100-01-01"
+        ) == bo_rows
+        assert await _instance_count(database) == total
+
+
+async def test_regenerate_unknown_scope_ids_are_404(
+    admin_api: SimpleNamespace,
+) -> None:
+    """An unknown definition or child scope id is 404, nothing written."""
+    client = admin_api.client
+
+    unknown_definition = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "definition", "definition_id": 999},
+        headers=_admin_headers(),
+    )
+    assert unknown_definition.status_code == 404
+    assert unknown_definition.json()["detail"] == "Quest definition not found"
+
+    unknown_child = await client.post(
+        "/api/v1/admin/regenerate",
+        json={"scope": "child", "child_id": 999},
+        headers=_admin_headers(),
+    )
+    assert unknown_child.status_code == 404
+    assert unknown_child.json()["detail"] == "Child not found"
+
+    assert await _instance_count(admin_api.database) == 0
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        {"scope": "galaxy"},  # unknown scope
+        {"scope": "definition"},  # scope without its id
+        {"scope": "child"},  # scope without its id
+        {"scope": "household", "child_id": 1},  # id without its scope
+        {"scope": "household", "definition_id": 1},  # id without its scope
+        {
+            "scope": "definition",
+            "definition_id": 1,
+            "child_id": 2,
+        },  # scope mismatch
+        {"scope": "child", "child_id": 0},  # zero id
+        {"scope": "child", "child_id": "seven"},  # wrong type
+    ],
+)
+async def test_regenerate_malformed_bodies_are_422(
+    admin_api: SimpleNamespace, body: dict[str, object]
+) -> None:
+    """Malformed scope/id combinations are 422 before any core call."""
+    response = await admin_api.client.post(
+        "/api/v1/admin/regenerate", json=body, headers=_admin_headers()
+    )
+    assert response.status_code == 422, body
+    assert await _instance_count(admin_api.database) == 0

tests/test_api_missed_sweep.py

@@ -0,0 +1,671 @@
+"""Admin-plane missed-sweep trigger tests (task 058c7b69).
+
+These tests exercise ``POST /api/v1/admin/missed-sweep`` — the admin
+plane's trigger for the nightly missed-quest sweep — against a REAL
+server (uvicorn on an ephemeral port, the SSE tests' documented
+approach: httpx's ASGI transport buffers whole bodies, so a live
+stream needs a live server), plus the API service's daily scheduler:
+
+- The trigger requires the router's ONE ``require_admin`` check: a
+  non-admin JWT is 403 and an absent credential is 401, and neither
+  refusal runs the sweep.
+- Seeded past-due open instances: ONE ``nestquest_quest_missed``
+  transition per instance arrives on the LIVE SSE stream with the
+  documented payload fields (child_id, child_name, instance_id,
+  quest_title, window, due_date, due_time, occurred_at — §3), while
+  the seeded TODAY-due instance publishes nothing (the sweep's rule
+  is ``due_date`` strictly before ``today``).
+- A same-day rerun publishes nothing (the core watermark).
+- A run after downtime (``today`` advanced two days) publishes the
+  accumulated window once — the instance due on day one fires on the
+  catch-up run — and a further rerun is silent again.
+- Publishing is the api/scheduler.py shape — the sweep-BUILT
+  ``(event_type, payload)`` pairs go straight to the publisher, no
+  per-event re-fetch: all events of one run share the run's ONE
+  ``occurred_at`` stamp, and a deterministic concurrent-deletion case
+  (the sweep is wrapped to delete one swept instance after it built
+  its events and committed its watermark) still answers 200 and
+  streams every built event — a re-fetch there would 500 and lose the
+  missed event forever.
+- The daily scheduler (:class:`api.scheduler.MissedSweepScheduler`)
+  computes the delay to the configured ``day_rollover_time``, runs the
+  sweep when the (injectable) clock reaches it, publishes the built
+  events on its publisher, advances to the NEXT day's rollover, and
+  cancels cleanly.
+
+The household is seeded through the SAME core copy the app uses (the
+``nestquest_core.*`` package registered by ``api/nestquest_core.py``;
+see the two-copy caveat there) and the sweep's watermark is read back
+through the same copy's :class:`~nestquest_core.dao_meta.MetaStateDao`
+— on the SERVER's event loop for the uvicorn-hosted app (the database
+belongs to that thread's loop, so assertions schedule there), on the
+test's own loop for the scheduler test's standalone database.
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import importlib
+import threading
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import httpx
+import pytest
+import uvicorn
+
+from api import routes_admin
+from api.app import create_app
+from api.auth import JwksCache
+from api.config import ApiConfig
+from api.database import _executor
+from api.nestquest_core import core_db as _core_db
+from api.nestquest_core import core_migrations as _core_migrations
+from api.scheduler import MissedSweepScheduler
+from tests.admin_jwt_harness import (
+    ADMIN_KEY,
+    AUDIENCE,
+    ISSUER,
+    JWKS_URL,
+    KID,
+    StubbedFetch,
+    jwks_for,
+    make_token,
+)
+from tests.test_api_panel import PANEL_TOKEN
+from tests.test_api_sse import EVENT_QUEST_MISSED, READ_TIMEOUT, _free_port, _stream
+
+# The bundled core submodules, through the SAME copy the app uses
+# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
+_children = importlib.import_module("nestquest_core.children")
+_quest_definitions = importlib.import_module(
+    "nestquest_core.quest_definitions"
+)
+_materialize = importlib.import_module("nestquest_core.materialize")
+_recurrence = importlib.import_module("nestquest_core.recurrence")
+_dao_instances = importlib.import_module("nestquest_core.dao_instances")
+_dao_meta = importlib.import_module("nestquest_core.dao_meta")
+_settings_store = importlib.import_module("nestquest_core.settings_store")
+_core_sweep = importlib.import_module("nestquest_core.sweep")
+
+#: The well-known ``nestquest_meta_state`` key holding the sweep
+#: watermark — asserted literally so the contract is pinned here.
+SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
+
+
+def _admin_headers() -> dict[str, str]:
+    """The Authorization header of a valid nestquest-admins JWT."""
+    return {"Authorization": f"Bearer {make_token()}"}
+
+
+async def _seed_missed_household(
+    db_path: str, today: datetime.date
+) -> SimpleNamespace:
+    """Seed the missed-sweep household on ``db_path`` and return records.
+
+    Two children; two quests whose ONLY instances are due YESTERDAY
+    (created via ``materialize`` with ``today`` pinned to yesterday —
+    the no-past guard accepts the rows, and they land past-due with no
+    completion events: the sweep's targets), and one quest due TODAY
+    (not past-due: the sweep must NOT announce it).  The yesterday
+    rules carry ``end_date`` so the later today-walk cannot give them
+    a second instance.  Opens, migrates, seeds and CLOSES its own
+    connection so the app's lifespan connection never overlaps it.
+    """
+    database = _core_db.NestQuestDatabase(_executor)
+    await database.open(db_path)
+    try:
+        await _core_migrations.apply_migrations(database)
+        today_iso = today.isoformat()
+        yesterday = today - datetime.timedelta(days=1)
+        yesterday_iso = yesterday.isoformat()
+
+        ada = await _children.create_child(database, "Ada", sort_order=0)
+        bo = await _children.create_child(database, "Bo", sort_order=1)
+
+        yesterday_rule = _recurrence.ScheduleRule.from_dict(
+            {
+                "rule_type": "daily",
+                "start_date": yesterday_iso,
+                "end_date": yesterday_iso,
+            }
+        )
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Stale chore",
+            yesterday_rule,
+            [ada.id],
+            [("morning", "09:00")],
+        )
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Old chore",
+            yesterday_rule,
+            [bo.id],
+            [("evening", "18:30")],
+        )
+        # Past-due open instances for the two yesterday quests (the
+        # walk pins ``today`` to yesterday so the rows are accepted).
+        await _materialize.materialize(
+            database, yesterday_iso, yesterday_iso, today=yesterday
+        )
+        # Today's quest: created after the yesterday walk so it has no
+        # past-due instance, then materialized for today only.
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Recoverable chore",
+            _recurrence.ScheduleRule.from_dict(
+                {"rule_type": "daily", "start_date": today_iso}
+            ),
+            [ada.id],
+            [("afternoon", "17:00")],
+        )
+        await _materialize.materialize(
+            database, today_iso, today_iso, today=today
+        )
+
+        instances = _dao_instances.QuestInstancesDao(database)
+        ada_yesterday = await instances.list_by_child_and_date(
+            ada.id, yesterday_iso
+        )
+        assert len(ada_yesterday) == 1
+        stale = ada_yesterday[0]
+        bo_yesterday = await instances.list_by_child_and_date(
+            bo.id, yesterday_iso
+        )
+        assert len(bo_yesterday) == 1
+        old = bo_yesterday[0]
+        ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
+        assert len(ada_today) == 1
+        recoverable = ada_today[0]
+        return SimpleNamespace(
+            ada=ada,
+            bo=bo,
+            stale_instance=stale,
+            old_instance=old,
+            recoverable_instance=recoverable,
+            yesterday_iso=yesterday_iso,
+            today_iso=today_iso,
+        )
+    finally:
+        await database.close()
+
+
+class _AdminServer:
+    """Run the app with uvicorn on an ephemeral port, in a thread.
+
+    The SSE tests' documented approach (see tests/test_api_sse.py):
+    uvicorn's factory mode runs ``create_app`` INSIDE the server
+    thread, so the lifespan — database, publisher, the daily sweep
+    scheduler — belongs to that thread's event loop.  The app's JWKS
+    cache is swapped for one built around the stubbed fetcher BEFORE
+    the server starts, so admin JWTs genuinely verify with no network.
+    """
+
+    def __init__(self, db_path: str) -> None:
+        self.port = _free_port()
+        self._server_loop: asyncio.AbstractEventLoop | None = None
+        self._server = uvicorn.Server(
+            uvicorn.Config(
+                self._make_app_factory(db_path),
+                host="127.0.0.1",
+                port=self.port,
+                log_level="warning",
+            )
+        )
+        self._thread = threading.Thread(target=self._server.run, daemon=True)
+
+    def _make_app_factory(self, db_path: str):
+        """The uvicorn app factory, closing over ``db_path``."""
+
+        def factory():
+            app = create_app(
+                ApiConfig(
+                    db_path=db_path,
+                    panel_token=PANEL_TOKEN,
+                    oidc_issuer=ISSUER,
+                    oidc_audience=AUDIENCE,
+                    oidc_jwks_url=JWKS_URL,
+                )
+            )
+            app.state.jwks_cache = JwksCache(
+                JWKS_URL, fetcher=StubbedFetch(jwks_for(ADMIN_KEY, KID))
+            )
+            self._server_loop = asyncio.get_running_loop()
+            self.app = app
+            return app
+
+        return factory
+
+    def __enter__(self) -> "_AdminServer":
+        self._thread.start()
+        deadline = time.monotonic() + READ_TIMEOUT
+        while not self._server.started:
+            if time.monotonic() > deadline or not self._thread.is_alive():
+                raise RuntimeError("uvicorn did not start")
+            time.sleep(0.01)
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self._server.should_exit = True
+        self._thread.join(timeout=READ_TIMEOUT)
+
+    def url(self, path: str) -> str:
+        """The full URL for ``path`` on this server."""
+        return f"http://127.0.0.1:{self.port}{path}"
+
+    def run_on_loop(self, coro) -> object:
+        """Run ``coro`` on the SERVER's event loop and wait for it.
+
+        The app (and its database) lives on the uvicorn thread's loop,
+        so database assertions schedule there.  Bounded by
+        READ_TIMEOUT so a wedged schedule fails instead of hanging.
+        """
+        return asyncio.run_coroutine_threadsafe(
+            coro, self._server_loop
+        ).result(timeout=READ_TIMEOUT)
+
+
+def _watermark(server: _AdminServer) -> str | None:
+    """The stored sweep watermark, read on the server's own loop."""
+    database = server.app.state.db.database
+    return server.run_on_loop(
+        _dao_meta.MetaStateDao(database).get(SWEEP_WATERMARK_KEY)
+    )
+
+
+@pytest.fixture
+def temp_db_path(tmp_path: Path) -> str:
+    """A fresh per-test SQLite path under pytest's tmp_path."""
+    return str(tmp_path / "nestquest.db")
+
+
+@pytest.fixture
+async def missed_app(temp_db_path: str) -> SimpleNamespace:
+    """The seeded household served by a live uvicorn app, clock pinned.
+
+    ``routes_admin._local_now`` is pinned to today 12:00 (the same
+    pinning discipline the panel/SSE tests use) BEFORE the server
+    starts, so the trigger route's ``today`` matches the seed; the
+    downtime test moves the pin forward mid-test.  Restored in
+    ``finally`` even on failure.
+    """
+    now_local = datetime.datetime.now().astimezone()
+    pinned_now = now_local.replace(hour=12, minute=0, second=0, microsecond=0)
+    today = pinned_now.date()
+    seed = await _seed_missed_household(temp_db_path, today)
+    original = routes_admin._local_now
+    routes_admin._local_now = lambda: pinned_now
+    try:
+        with _AdminServer(temp_db_path) as server:
+            client = httpx.AsyncClient(base_url=server.url(""))
+            async with client:
+                yield SimpleNamespace(
+                    client=client,
+                    seed=seed,
+                    server=server,
+                    today=today,
+                    pinned_now=pinned_now,
+                )
+    finally:
+        routes_admin._local_now = original
+
+
+# --- the trigger: one missed transition per past-due open instance -----
+
+
+async def test_trigger_publishes_one_missed_event_per_instance(
+    missed_app,
+) -> None:
+    """The trigger streams one documented missed transition per instance.
+
+    The stream is subscribed FIRST, the sweep is triggered through the
+    route as an admin, and each emitted frame's type and every
+    documented payload field are asserted; the TODAY-due instance must
+    NOT appear (the sweep's rule is strictly past-due).
+    """
+    ns = missed_app
+    seed = ns.seed
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        response = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=_admin_headers()
+        )
+        assert response.status_code == 200
+        assert response.json() == {"fired": 2}
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert event_type == "nestquest_quest_missed"
+        assert payload["child_id"] == seed.ada.id
+        assert payload["child_name"] == "Ada"
+        assert payload["instance_id"] == seed.stale_instance.id
+        assert payload["quest_title"] == "Stale chore"
+        assert payload["window"] == "morning"
+        assert payload["due_date"] == seed.yesterday_iso
+        assert payload["due_time"] == "09:00"
+        assert payload["occurred_at"]  # strict UTC ISO-8601 stamp
+        run_stamp = payload["occurred_at"]
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert payload["child_id"] == seed.bo.id
+        assert payload["child_name"] == "Bo"
+        assert payload["instance_id"] == seed.old_instance.id
+        assert payload["quest_title"] == "Old chore"
+        assert payload["window"] == "evening"
+        assert payload["due_date"] == seed.yesterday_iso
+        assert payload["due_time"] == "18:30"
+        # One run, ONE stamp: the route publishes the sweep-built
+        # payloads directly, so every event of this run carries the
+        # same ``occurred_at`` the sweep computed for the whole run.
+        assert payload["occurred_at"] == run_stamp
+
+        # Exactly one transition per instance: today's not-past-due
+        # quest never arrives.
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+    # The run recorded the watermark: the household-local date the
+    # sweep ran for, under the well-known meta_state key.
+    assert _watermark(ns.server) == seed.today_iso
+
+
+async def test_trigger_publishes_sweep_built_payloads_without_refetch(
+    missed_app,
+) -> None:
+    """Publishing never re-fetches: a concurrently deleted instance
+    cannot fail the trigger or lose its already-built missed event.
+
+    The core sweep commits the idempotency watermark BEFORE returning,
+    so a publish loop that re-fetches each instance opens a loss
+    window: an instance deleted inside that window (e.g. a concurrent
+    regenerate) would raise ``ValueError`` uncaught — a bare 500 — and
+    the missed event would be gone forever (the watermark already
+    advanced, so no future sweep re-examines it).  The route publishes
+    the sweep-built ``(event_type, payload)`` pairs DIRECTLY (the
+    api/scheduler.py shape), which this test exercises deterministically:
+    the sweep is wrapped to delete one swept instance AFTER the sweep
+    built its events and committed its watermark but BEFORE the route
+    publishes, and the trigger must still answer 200 with ``fired: 2``
+    and stream BOTH sweep-built payloads — the deleted instance's
+    included, stamped with the run's ONE shared ``occurred_at``.
+    """
+    ns = missed_app
+    seed = ns.seed
+    database = ns.server.app.state.db.database
+    original_sweep = _core_sweep.run_missed_sweep
+
+    async def sweep_then_delete_stale(sweep_database, *, today):
+        events = await original_sweep(sweep_database, today=today)
+        # The concurrent deletion lands after the sweep built its
+        # events and committed its watermark, but before the route
+        # publishes: the row is gone while the built event survives.
+        await sweep_database.execute(
+            "DELETE FROM quest_instances WHERE id = ?",
+            (seed.stale_instance.id,),
+        )
+        return events
+
+    _core_sweep.run_missed_sweep = sweep_then_delete_stale
+    try:
+        stream = _stream(ns.server.url("/api/v1/panel/events"))
+        async with stream:
+            response = await ns.client.post(
+                "/api/v1/admin/missed-sweep", headers=_admin_headers()
+            )
+            assert response.status_code == 200
+            assert response.json() == {"fired": 2}
+
+            # The FIRST event is the deleted instance's — published
+            # from the payload built before the deletion, never
+            # re-fetched after it (a re-fetch would 500 here).
+            event_type, payload = await stream.next_event()
+            assert event_type == EVENT_QUEST_MISSED
+            assert payload["instance_id"] == seed.stale_instance.id
+            assert payload["child_name"] == "Ada"
+            assert payload["quest_title"] == "Stale chore"
+            assert payload["window"] == "morning"
+            assert payload["due_date"] == seed.yesterday_iso
+            assert payload["occurred_at"]
+            run_stamp = payload["occurred_at"]
+
+            event_type, payload = await stream.next_event()
+            assert event_type == EVENT_QUEST_MISSED
+            assert payload["instance_id"] == seed.old_instance.id
+            assert payload["occurred_at"] == run_stamp
+
+            with pytest.raises(asyncio.TimeoutError):
+                await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+    finally:
+        _core_sweep.run_missed_sweep = original_sweep
+
+    # The watermark advanced even though one instance was deleted
+    # mid-flight — nothing was lost to the publish-time failure window.
+    assert _watermark(ns.server) == seed.today_iso
+
+
+async def test_same_day_rerun_publishes_nothing(missed_app) -> None:
+    """A second trigger the same day publishes nothing (the watermark)."""
+    ns = missed_app
+    headers = _admin_headers()
+    first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert first.status_code == 200
+    assert first.json() == {"fired": 2}
+
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        second = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=headers
+        )
+        assert second.status_code == 200
+        assert second.json() == {"fired": 0}
+        with pytest.raises(asyncio.TimeoutError):
+            # The watermark makes the rerun an empty no-op: the stream
+            # stays silent.
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+
+async def test_run_after_downtime_publishes_accumulated_window_once(
+    missed_app,
+) -> None:
+    """After days of downtime the catch-up run sweeps the window once.
+
+    The first trigger sweeps yesterday's instances (watermark := today).
+    ``today`` then jumps two days: the accumulated window
+    ``[watermark, today)`` holds exactly the seeded TODAY-due instance
+    (no completion events), which fires once — and a same-day rerun is
+    silent again.
+    """
+    ns = missed_app
+    headers = _admin_headers()
+    first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert first.status_code == 200
+    assert first.json() == {"fired": 2}
+
+    # Advance the pinned clock two days: the app is "down" over the
+    # rollovers in between.
+    routes_admin._local_now = lambda: ns.pinned_now + datetime.timedelta(
+        days=2
+    )
+    later_today = (ns.today + datetime.timedelta(days=2)).isoformat()
+    stream = _stream(ns.server.url("/api/v1/panel/events"))
+    async with stream:
+        catchup = await ns.client.post(
+            "/api/v1/admin/missed-sweep", headers=headers
+        )
+        assert catchup.status_code == 200
+        assert catchup.json() == {"fired": 1}
+
+        event_type, payload = await stream.next_event()
+        assert event_type == EVENT_QUEST_MISSED
+        assert payload["instance_id"] == ns.seed.recoverable_instance.id
+        assert payload["child_id"] == ns.seed.ada.id
+        assert payload["quest_title"] == "Recoverable chore"
+        assert payload["window"] == "afternoon"
+        assert payload["due_date"] == ns.seed.today_iso
+        assert payload["due_time"] == "17:00"
+        assert payload["occurred_at"]
+
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(stream.queue.get(), timeout=0.3)
+
+    # The watermark moved to the new today...
+    assert _watermark(ns.server) == later_today
+    # ...so a further rerun publishes nothing.
+    again = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
+    assert again.status_code == 200
+    assert again.json() == {"fired": 0}
+
+
+# --- authentication: the router's ONE require_admin dependency ----------
+
+
+async def test_trigger_refuses_non_admin_and_absent_credentials(
+    missed_app,
+) -> None:
+    """A non-admin JWT is 403, an absent credential is 401; none sweep."""
+    ns = missed_app
+    non_admin = make_token(groups=("some-other-group",))
+    forbidden = await ns.client.post(
+        "/api/v1/admin/missed-sweep",
+        headers={"Authorization": f"Bearer {non_admin}"},
+    )
+    assert forbidden.status_code == 403
+    assert forbidden.json()["detail"] == (
+        "Admin access requires membership in the nestquest-admins group"
+    )
+    absent = await ns.client.post("/api/v1/admin/missed-sweep")
+    assert absent.status_code == 401
+
+    # Neither refusal ran the sweep: the watermark is still unset, and
+    # the next admin trigger fires the full seeded set.
+    assert _watermark(ns.server) is None
+    after = await ns.client.post(
+        "/api/v1/admin/missed-sweep", headers=_admin_headers()
+    )
+    assert after.status_code == 200
+    assert after.json() == {"fired": 2}
+
+
+# --- the daily scheduler -------------------------------------------------
+
+
+class _RecordingPublisher:
+    """A transition-publisher stub recording every published event."""
+
+    def __init__(self) -> None:
+        self.published: list[tuple[str, dict]] = []
+
+    def publish(self, event_type: str, payload: dict) -> None:
+        self.published.append((event_type, payload))
+
+
+async def test_scheduler_runs_sweep_at_rollover_and_cancels_cleanly(
+    temp_db_path: str,
+) -> None:
+    """The scheduler sweeps at the configured rollover and stops clean.
+
+    The injectable clock starts at 10:00 and the settings pin the
+    rollover to 10:05, so the first computed delay is five minutes;
+    the injectable sleep fast-forwards the clock, the sweep runs for
+    the (now advanced) clock's date, and the seeded past-due open
+    instance is published on the recording publisher exactly once.
+    The next cycle schedules for the NEXT day's rollover (23h55m).
+    ``stop`` cancels the loop task and the scheduler reports stopped.
+    """
+    database = _core_db.NestQuestDatabase(_executor)
+    await database.open(temp_db_path)
+    try:
+        await _core_migrations.apply_migrations(database)
+        now_local = datetime.datetime.now().astimezone()
+        start = now_local.replace(
+            hour=10, minute=0, second=0, microsecond=0
+        )
+        ada = await _children.create_child(database, "Ada", sort_order=0)
+        yesterday = (start.date() - datetime.timedelta(days=1)).isoformat()
+        await _quest_definitions.create_quest_definition(
+            database,
+            "Stale chore",
+            _recurrence.ScheduleRule.from_dict(
+                {
+                    "rule_type": "daily",
+                    "start_date": yesterday,
+                    "end_date": yesterday,
+                }
+            ),
+            [ada.id],
+            [("morning", "09:00")],
+        )
+        await _materialize.materialize(
+            database, yesterday, yesterday, today=start.date() - datetime.timedelta(days=1)
+        )
+        await _settings_store.update_settings(
+            database, {"day_rollover_time": "10:05"}
+        )
+
+        clock_state = {"now": start}
+        delays: list[float] = []
+
+        def clock() -> datetime.datetime:
+            return clock_state["now"]
+
+        async def sleep_stub(seconds: float) -> None:
+            delays.append(seconds)
+            if len(delays) == 1:
+                # Fast-forward the clock across the rollover wait ONCE,
+                # so the first sweep runs at the rollover instant; the
+                # clock then stays frozen at that instant, so the
+                # follow-up cycles sweep the same (already-watermarked)
+                # date and the assertions read a stable state.
+                clock_state["now"] = clock_state["now"] + datetime.timedelta(
+                    seconds=seconds
+                )
+            await asyncio.sleep(0)
+
+        publisher = _RecordingPublisher()
+        scheduler = MissedSweepScheduler(
+            database, publisher=publisher, clock=clock, sleep=sleep_stub
+        )
+        scheduler.start()
+        try:
+            deadline = time.monotonic() + READ_TIMEOUT
+            while not publisher.published:
+                if time.monotonic() > deadline:
+                    pytest.fail("scheduler never published a missed event")
+                await asyncio.sleep(0.01)
+            # The second cycle's scheduling decision may lag the first
+            # publication by a few awaits; wait for it the same way.
+            while len(delays) < 2:
+                if time.monotonic() > deadline:
+                    pytest.fail("scheduler never scheduled a second cycle")
+                await asyncio.sleep(0.01)
+
+            event_type, payload = publisher.published[0]
+            assert event_type == "nestquest_quest_missed"
+            assert payload["child_id"] == ada.id
+            assert payload["quest_title"] == "Stale chore"
+            assert payload["window"] == "morning"
+            assert payload["due_date"] == yesterday
+            assert payload["due_time"] == "09:00"
+            assert payload["occurred_at"]
+
+            # The rollover delay was honored exactly...
+            assert delays[0] == pytest.approx(300.0)
+            # ...and the next cycle targets the NEXT day's rollover
+            # (the just-passed instant schedules for tomorrow).
+            assert delays[1] == pytest.approx(86400.0)
+            # The watermark pins the run to the sweep-day's date.
+            watermark = await _dao_meta.MetaStateDao(database).get(
+                SWEEP_WATERMARK_KEY
+            )
+            assert watermark == start.date().isoformat()
+        finally:
+            await scheduler.stop()
+        assert not scheduler.running
+        # Exactly one announcement: the watermark makes every further
+        # cycle an empty no-op.
+        assert len(publisher.published) == 1
+    finally:
+        await database.close()