Admin routes: settings read and update

WHY: the API service had no settings surface of its own — core
NestQuestSettings is built from HA config-entry options, which the API
(no config entry) cannot read. The core now has an HA-free durable
settings store (core/settings_store.py) that keeps a validated
ten-field JSON document in the existing nestquest_meta_state table
under the well-known "settings" key: load_settings resolves it through
the core _FIELDS validators (defaults for absent fields, per-field
fallback over a corrupt document so the service never crashes) and
update_settings validates every provided field BEFORE writing, so a
rejected change — an unknown field or an invalid value, each named in
the error — never leaves a half-applied document behind. The admin
plane's GET/PATCH /settings routes are thin adapters over that store;
the integration's HA options remain a temporary separate source until
Feature 18 wires the client to the API.

.opencode/agents/approver.md

@@ -1,32 +0,0 @@
----
-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

@@ -1,21 +0,0 @@
----
-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

@@ -1,60 +0,0 @@
----
-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

@@ -1,31 +0,0 @@
----
-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

@@ -1,11 +0,0 @@
----
-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

@@ -1,77 +0,0 @@
-# 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,7 +48,6 @@ 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):
@@ -58,11 +57,7 @@ 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, and
-    starts the daily missed-sweep scheduler
-    (:class:`api.scheduler.MissedSweepScheduler`) against the same
-    database and publisher — cancelled before the connection closes on
-    shutdown.
+    every route and SSE subscriber shares that single instance.
     """
 
     @asynccontextmanager
@@ -70,16 +65,10 @@ 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

@@ -198,12 +198,6 @@ core_dao_children = importlib.import_module(
 #: delegating.
 core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules")
 
-#: The HA-free missed sweep (``core.sweep.run_missed_sweep`` +
-#: ``SWEEP_WATERMARK_KEY``) the admin plane's trigger route calls and
-#: the daily scheduler runs — the ONE sweep policy the integration's
-#: bus shim shares.
-core_sweep = importlib.import_module(f"{_CORE_MODULE_NAME}.sweep")
-
 __all__ = [
     "core",
     "core_children",
@@ -224,5 +218,4 @@ __all__ = [
     "core_snapshot",
     "core_settings",
     "core_settings_store",
-    "core_sweep",
 ]

api/routes_admin.py

@@ -139,23 +139,6 @@ are deliberately not unified here:
   module's uniform None-as-absent rule).  The response is the updated
   effective settings, the same shape ``GET`` returns.
 
-The missed-sweep trigger (task 058c7b69) follows the same pattern over
-:mod:`nestquest_core.sweep` — the HA-free sweep policy the
-integration's rollover listener shares:
-
-- ``POST /missed-sweep`` runs the sweep for the API host's local
-  ``today`` (ONE :func:`_local_now` clock read threaded into the core
-  call) and publishes each returned ``(event_type, payload)`` pair
-  DIRECTLY on the app's ONE transition publisher — the sweep-built
-  payload, never re-fetched (the core commits its watermark before
-  returning, so a failed re-fetch after it would lose the missed event
-  forever), the same publish-what-the-core-built shape
-  api/scheduler.py uses.  The watermark rule
-  (a same-day rerun announces nothing, a post-downtime run sweeps the
-  accumulated window once), the no-completion-event rule and the
-  read-only-over-the-domain-tables guarantee all live in the core; the
-  response reports how many transitions were published.
-
 Error mapping (every children, quest-definition and presence route,
 deliberately narrow):
 
@@ -254,7 +237,6 @@ from api.nestquest_core import (
     core_recurrence,
     core_settings,
     core_settings_store,
-    core_sweep,
 )
 
 #: All admin-plane routes share this router; the ONE admin dependency
@@ -672,17 +654,6 @@ class AdminSettingsResponse(BaseModel):
     celebration_enabled: bool
 
 
-class AdminMissedSweepResponse(BaseModel):
-    """The result of ``POST /api/v1/admin/missed-sweep``.
-
-    ``fired`` is the number of ``nestquest_quest_missed`` transitions
-    the run built AND this route published on the SSE stream — zero
-    when the sweep's watermark already covers today (a same-day rerun).
-    """
-
-    fired: int
-
-
 #: 404 detail for a child id the core layer reports as non-existent.
 _CHILD_NOT_FOUND_DETAIL = "Child not found"
 
@@ -832,19 +803,6 @@ def _raise_settings_error(error: ValueError) -> NoReturn:
     raise HTTPException(status_code=422, detail=str(error)) from error
 
 
-def _raise_sweep_error(error: ValueError) -> NoReturn:
-    """Map a core missed-sweep ``ValueError`` onto 422 and raise it.
-
-    The sweep route has no path ids, so it has no 404 case: a core
-    ``ValueError`` — a rejected date the sweep's query refuses before
-    any read, unreachable from this handler because ``today`` comes
-    from the route's own clock read — is 422 with the core's message
-    as the detail.  Only ``ValueError`` is caught, so an unexpected
-    failure re-raises rather than becoming a 4xx.
-    """
-    raise HTTPException(status_code=422, detail=str(error)) from error
-
-
 def _local_now() -> datetime.datetime:
     """Return the API host's local time as ONE timezone-aware clock read.
 
@@ -1677,56 +1635,4 @@ async def admin_update_settings(
     return _settings_response(settings)
 
 
-@router.post(
-    "/missed-sweep",
-    summary="Run the nightly missed-quest sweep now",
-    response_model=AdminMissedSweepResponse,
-)
-async def admin_run_missed_sweep(
-    request: Request,
-) -> AdminMissedSweepResponse:
-    """Run the missed-quest sweep and publish what it built.
-
-    Requires a valid admin JWT (the router's shared
-    :func:`~api.auth.require_admin` dependency — the ONE check; this
-    handler performs NO auth of its own).  Thin adapter over the SAME
-    HA-free sweep policy the integration's rollover listener runs
-    (:mod:`nestquest_core.sweep`): ONE timezone-aware clock read
-    (:func:`_local_now`) pins the API host's local ``today``, and ONE
-    :func:`nestquest_core.sweep.run_missed_sweep` call does everything
-    else — the past-due open-instance query, the documented payload
-    build and the watermark update (a same-day rerun is an empty
-    no-op, and a run after downtime sweeps the accumulated window
-    once).  The route is read-only over the domain tables: it never
-    mutates an instance and never writes a completion event.
-
-    Each returned ``(event_type, payload)`` pair is published DIRECTLY
-    on the app's ONE transition publisher (``request.app.state.publisher``,
-    fan-out fire-and-forget, see api/events.py) — the exact payload the
-    sweep built, never re-fetched and never hand-built here, the same
-    publish-what-the-core-built shape api/scheduler.py uses.  This is a
-    correctness rule, not a style preference: the core commits the
-    idempotency watermark BEFORE returning, so a re-fetch-based publish
-    loop opens a loss window — a concurrent regenerate deleting the
-    instance, or a transient database error, would raise uncaught (a
-    bare 500) AFTER the watermark advanced, and no future sweep would
-    ever re-examine that instance.  Publishing the built payload leaves
-    no such window, and every event of one run keeps the ONE shared
-    ``occurred_at`` stamp the sweep computed for it.  The response
-    reports ``fired`` — the number of transitions published.  Error
-    mapping (:func:`_raise_sweep_error`): every core ``ValueError`` is
-    422; anything else re-raises.
-    """
-    state: DatabaseState = request.app.state.db
-    database = state.database
-    today = _local_now().date()
-    try:
-        events = await core_sweep.run_missed_sweep(database, today=today)
-    except ValueError as error:
-        _raise_sweep_error(error)
-    for event_type, payload in events:
-        request.app.state.publisher.publish(event_type, payload)
-    return AdminMissedSweepResponse(fired=len(events))
-
-
 __all__ = ["router"]

api/scheduler.py

@@ -1,173 +0,0 @@
-"""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/__init__.py

@@ -18,9 +18,7 @@ from homeassistant.helpers.event import async_track_time_change
 
 from .const import (
     CONF_ADMIN_USER_IDS,
-    CONF_SNAPSHOT_STALENESS,
     CONF_UPDATE_INTERVAL,
-    DEFAULT_SNAPSHOT_STALENESS,
     DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
     LOGGER,
@@ -28,12 +26,11 @@ from .const import (
 )
 from .core.settings import NestQuestSettings
 from .frontend import async_register_frontend
-from .coordinator import NestQuestCoordinator, coordinator_client_from_entry
+from .coordinator import NestQuestCoordinator
 from .db import NestQuestDatabase, make_database
 from .materialize import materialize as _materialize_run
 from .migrations import apply_migrations
 from .services import async_deregister_services, async_register_services
-from .sse import NestQuestEventStream
 from .store import async_get_db_path
 from .admin_allowlist import seed_setup_admin
 from .sweep import run_missed_sweep
@@ -245,7 +242,6 @@ class NestQuestRuntimeData:
     remove_update_listener: Callable[[], Any]
     remove_time_change_listener: Callable[[], Any]
     coordinator: Any = None
-    event_stream: NestQuestEventStream | None = None
 
 
 async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]:
@@ -443,7 +439,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     remove_update_listener: Callable[[], Any] | None = None
     remove_time_change_listener: Callable[[], Any] | None = None
     coordinator: Any = None
-    event_stream: NestQuestEventStream | None = None
     try:
         # Seed the admin allowlist on first setup so the owner is never
         # locked out: an empty database takes the persisted admin copy
@@ -495,46 +490,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         _register_services(hass)
         # The shared Feature 10 coordinator: one refresh cycle every
         # entity reads from (CONF_UPDATE_INTERVAL seconds, default
-        # five minutes).  Since Feature 18 it polls the API service's
-        # panel snapshot instead of reading the local database; an
-        # entry with no panel token yet gets the typed-error stub
-        # (unavailable entities, never a setup crash).  Created AFTER
-        # services so a failure below unwinds them through the except
-        # path's listener handling.
-        # ONE API client for the entry, shared by the coordinator's
-        # snapshot polling and the SSE subscription below (the client
-        # is stateless — one session, one connection pool).
-        api_client = coordinator_client_from_entry(hass, entry)
-        # The shared Feature 10 coordinator: one refresh cycle every
-        # entity reads from (CONF_UPDATE_INTERVAL seconds, default
-        # five minutes).  Since Feature 18 it polls the API service's
-        # panel snapshot instead of reading the local database; an
-        # entry with no panel token yet gets the typed-error stub
-        # (unavailable entities, never a setup crash).  Created AFTER
-        # services so a failure below unwinds them through the except
-        # path's listener handling.
+        # five minutes).  Created AFTER services so a failure below
+        # unwinds them through the except path's listener handling.
         coordinator = NestQuestCoordinator(
             hass,
             entry_id=entry.entry_id,
-            api_client=api_client,
+            database=database,
+            settings=settings,
             update_interval_seconds=entry.options.get(
                 CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
             ),
-            staleness_seconds=entry.options.get(
-                CONF_SNAPSHOT_STALENESS, DEFAULT_SNAPSHOT_STALENESS
-            ),
         )
-        # The panel event subscription (the SSE task): a background
-        # task consumes the API service's transition-event stream and
-        # re-fires the four documented HA bus events.  It reuses the
-        # coordinator's client, so it only starts when that client
-        # actually offers a stream — an unconfigured-token stub (or a
-        # test snapshot client without ``stream_events``) leaves it
-        # unstarted, and the typed-error path already covers
-        # availability for those entries.
-        if callable(getattr(api_client, "stream_events", None)):
-            event_stream = NestQuestEventStream(hass, api_client)
-            event_stream.start()
+        await coordinator.async_config_entry_first_refresh()
         runtime_data = NestQuestRuntimeData(
             entry_id=entry.entry_id,
             settings=settings,
@@ -542,32 +509,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
             remove_update_listener=remove_update_listener,
             remove_time_change_listener=remove_time_change_listener,
             coordinator=coordinator,
-            event_stream=event_stream,
         )
-        # The runtime record is registered BEFORE the first refresh:
-        # the refresh path (and anything it calls) resolves the entry's
-        # runtime state through hass.data, and it must find the live
-        # database there — never a half-set-up entry.  The platforms
-        # forward below reads the coordinator off the entry
-        # (the HA runtime-data contract); a forward failure below
+        # runtime_data is assigned BEFORE the platforms forward so each
+        # platform's async_setup_entry reads its coordinator off the
+        # entry (the HA runtime-data contract); a forward failure below
         # unwinds it through the except path.
-        hass.data[DOMAIN][entry.entry_id] = runtime_data
         entry.runtime_data = runtime_data
-        # The first refresh polls the API snapshot; an API failure
-        # (unconfigured token, API service down) does NOT crash setup:
-        # the coordinator reports last_update_success=False, entities
-        # come up unavailable, and HA's standard retry cycle applies
-        # once the API is reachable.  HA's real
-        # ``async_config_entry_first_refresh`` wraps ANY failed update
-        # pass in ConfigEntryNotReady — the original error survives
-        # only as ``__cause__``, never as the raised exception — so the
-        # catch keys on ConfigEntryNotReady, NOT on the typed API
-        # error.  Everything else (a platform-forward failure,
-        # cancellation) still unwinds through the except block below.
-        try:
-            await coordinator.async_config_entry_first_refresh()
-        except ConfigEntryNotReady:
-            pass
         await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     except BaseException:
         # A failure anywhere after the database is open must not leak the
@@ -575,14 +522,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         # unwind EVERY remover we acquired before the failure, keeping the
         # original error as the one raised.  Removal is best-effort — a
         # failing remover must not mask the setup error that triggered it.
-        # The SSE subscription (started just before the runtime record)
-        # is stopped first: like the coordinator it must never outlive
-        # the database being closed below.
-        if event_stream is not None:
-            try:
-                await event_stream.stop()
-            except BaseException:
-                pass
         for remove_listener in (
             remove_time_change_listener,
             remove_update_listener,
@@ -609,19 +548,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         # closes the already-closed database through the stale record.
         if getattr(entry, "runtime_data", None) is not None:
             entry.runtime_data = None
-        # The runtime record was ALSO registered in hass.data before
-        # the first refresh: remove it here on ANY failure, or a later
-        # unload picks up the stale, already-torn-down record
-        # (hass.data wins over entry.runtime_data there) and
-        # re-invokes the already-consumed update-listener remover.
-        hass.data.get(DOMAIN, {}).pop(entry.entry_id, None)
         await database.close()
         raise
     assert remove_update_listener is not None
     assert remove_time_change_listener is not None
-    # The runtime record was already registered before the first
-    # refresh; re-assert the (unchanged) record so the return path
-    # stays readable as one registration site.
     hass.data[DOMAIN][entry.entry_id] = runtime_data
     return True
 
@@ -651,13 +581,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         return False
     unload_error: BaseException | None = None
     if runtime_data is not None:
-        # The SSE subscription is stopped FIRST: its task must be gone
-        # (no frame can fire) before anything else is torn down.
-        event_stream = getattr(runtime_data, "event_stream", None)
-        if event_stream is not None:
-            stop_stream = getattr(event_stream, "stop", None)
-            if stop_stream is not None:
-                await stop_stream()
         remove_update_listener = getattr(runtime_data, "remove_update_listener", None)
         remove_time_change_listener = getattr(
             runtime_data, "remove_time_change_listener", None

custom_components/nestquest/api_client.py

@@ -1,414 +0,0 @@
-"""Panel-plane API client for the NestQuest integration (Feature 18).
-
-The integration talks to the NestQuest API service (Feature 16) over
-HTTP instead of reading the database: this module is the ONE place that
-speaks the API's panel plane — ``GET /api/v1/panel/snapshot`` (today's
-household snapshot) and ``POST /api/v1/panel/instances/{id}/complete``
-(completion as the tapped child profile) — authenticated with the
-static panel service token as a Bearer credential on every request.
-
-Transport-injected by construction: the constructor takes the base URL,
-the panel token, and the HTTP session to use.  The session speaks the
-aiohttp ``ClientSession`` surface (``request(method, url, headers=...,
-json=...)`` returning an async context manager whose response carries
-``status`` and ``await text()``), so tests stub the transport with NO
-network and NO aiohttp installed; production builds the client through
-:func:`client_from_entry`, which hands in Home Assistant's shared
-aiohttp session (``homeassistant.helpers.aiohttp_client
-.async_get_clientsession``).  The module itself imports NEITHER
-homeassistant NOR aiohttp at runtime (aiohttp appears only behind
-``TYPE_CHECKING`` for annotations), so the client is usable anywhere —
-the HA coupling lives in the one production factory.
-
-Every failure is typed: a transport failure (connection refused, DNS
-failure, timeout — the exchange never produced a response) and every
-non-2xx HTTP response raise :class:`NestQuestApiError`.  The error
-carries the HTTP status where there is one (so 401 and 404 are
-distinguishable at the call site) and ``None`` when there is not; the
-original transport error is chained as ``__cause__`` — nothing is
-swallowed.  There are NO retries and the client logs nothing: refresh
-cadence is the coordinator's concern (the next task), and the panel
-token is a secret that must never reach a log line or an error
-message — the error texts name the method, path, and status only.
-
-The panel plane's SSE event stream (``GET /api/v1/panel/events``) is
-exposed by :meth:`NestQuestApiClient.stream_events` — an async
-generator yielding ``(event_type, payload)`` per Server-Sent Events
-frame (``event: <type>`` / ``data: <json>`` lines separated by a blank
-line).  Opening the stream gets the same one-timeout, typed-error
-contract as every other call; reading it has NO body-wide timeout (a
-healthy stream is long-lived — the subscription manager owns
-reconnection and backoff).
-"""
-from __future__ import annotations
-
-import asyncio
-import inspect
-import json
-from collections.abc import AsyncIterator
-from typing import TYPE_CHECKING, Any
-from urllib.parse import urlparse
-
-from .const import (
-    CONF_API_BASE_URL,
-    CONF_PANEL_TOKEN,
-    DEFAULT_API_BASE_URL,
-    DEFAULT_PANEL_TOKEN,
-)
-
-if TYPE_CHECKING:
-    import aiohttp
-
-__all__ = [
-    "COMPLETE_INSTANCE_PATH",
-    "DEFAULT_TIMEOUT_SECONDS",
-    "EVENTS_PATH",
-    "NestQuestApiError",
-    "NestQuestApiClient",
-    "SNAPSHOT_PATH",
-    "client_from_entry",
-    "is_valid_base_url",
-    "resolve_api_config",
-]
-
-#: The panel plane's snapshot route (Feature 16).
-SNAPSHOT_PATH = "/api/v1/panel/snapshot"
-
-#: The panel plane's SSE event-stream route (Feature 16): each
-#: transition arrives as an ``event: <type>`` / ``data: <json>`` frame.
-EVENTS_PATH = "/api/v1/panel/events"
-
-#: The panel plane's completion route template (Feature 16); the
-#: ``{instance_id}`` placeholder is filled per call.
-COMPLETE_INSTANCE_PATH = "/api/v1/panel/instances/{instance_id}/complete"
-
-#: Request timeout, in seconds.  One attempt, one timeout: a hung API
-#: call surfaces as a typed error for the coordinator to report, never
-#: a retry loop.
-DEFAULT_TIMEOUT_SECONDS = 10.0
-
-
-class NestQuestApiError(Exception):
-    """A failed NestQuest API exchange, raised for every failure mode.
-
-    ``status`` carries the HTTP status when a response arrived (a
-    non-2xx response — 401 for a bad token, 404 for an unknown
-    instance) and is ``None`` when the exchange failed at the transport
-    level (connection refused, DNS failure, timeout) — so callers
-    distinguish the two with ``is_transport_error``.  The original
-    transport error, where there is one, is chained as ``__cause__``.
-    Messages carry the method, path, and status — never the request
-    headers, so the panel token can never leak through an error.
-    """
-
-    def __init__(self, message: str, *, status: int | None = None) -> None:
-        super().__init__(message)
-        self.status = status
-
-    @property
-    def is_transport_error(self) -> bool:
-        """True when the exchange failed before any response arrived."""
-        return self.status is None
-
-
-def is_valid_base_url(value: object) -> bool:
-    """Return True when ``value`` is a usable API base URL.
-
-    A full ``http://`` or ``https://`` URL with a host — the exact
-    shape the options flow validates (form ergonomics) and the client
-    constructor enforces (fail fast, never build requests against a
-    scheme-less or host-less string).
-    """
-    if not isinstance(value, str):
-        return False
-    candidate = value.strip()
-    if not candidate:
-        return False
-    parsed = urlparse(candidate)
-    return parsed.scheme in ("http", "https") and bool(parsed.netloc)
-
-
-def resolve_api_config(entry: Any) -> tuple[str, str]:
-    """Resolve ``(base_url, panel_token)`` off a config entry.
-
-    Precedence mirrors the options flow's stored-values resolution:
-    ``entry.options`` first (the live options-flow copy), then
-    ``entry.data`` (the config-flow-era copy), then the const defaults.
-    The token is a secret: callers must never log the returned pair.
-    """
-    options = getattr(entry, "options", None) or {}
-    data = getattr(entry, "data", None) or {}
-
-    def _pick(key: str, default: str) -> str:
-        if key in options:
-            return options[key]
-        if key in data:
-            return data[key]
-        return default
-
-    return (
-        _pick(CONF_API_BASE_URL, DEFAULT_API_BASE_URL),
-        _pick(CONF_PANEL_TOKEN, DEFAULT_PANEL_TOKEN),
-    )
-
-
-def client_from_entry(hass: Any, entry: Any) -> NestQuestApiClient:
-    """Build the production client for a config entry.
-
-    Reads the configured base URL and panel token off the entry
-    (:func:`resolve_api_config`) and hands the client Home Assistant's
-    SHARED aiohttp session (``homeassistant.helpers.aiohttp_client
-    .async_get_clientsession``) so panel calls ride HA's one connection
-    pool.  The helper import is deferred into this function on purpose:
-    the test harness is mock-only and the module must import (and the
-    client must be stub-transport testable) without homeassistant or
-    aiohttp installed.
-    """
-    from homeassistant.helpers.aiohttp_client import async_get_clientsession
-
-    base_url, panel_token = resolve_api_config(entry)
-    return NestQuestApiClient(
-        base_url, panel_token, async_get_clientsession(hass)
-    )
-
-
-class NestQuestApiClient:
-    """Async client for the NestQuest API service's panel plane.
-
-    ONE instance per config entry (the coordinator task wires it into
-    the refresh cycle).  The client is deliberately thin: it adds the
-    Bearer credential and the base URL, enforces one timeout per
-    request, and turns every failure into :class:`NestQuestApiError` —
-    no business logic, no retries, no caching.
-    """
-
-    def __init__(
-        self,
-        base_url: str,
-        panel_token: str,
-        session: aiohttp.ClientSession,
-        *,
-        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
-    ) -> None:
-        """Store the connection settings; the session is injected.
-
-        ``session`` is an aiohttp ``ClientSession`` in production and a
-        stub transport in tests.  ``base_url`` must be a full http(s)
-        URL and ``panel_token`` must be non-empty — a client built
-        without either can only fail later at request time with a
-        less-explicable error, so construction fails fast instead.
-        """
-        if not is_valid_base_url(base_url):
-            raise ValueError(
-                f"base_url must be a full http(s) URL, got {base_url!r}"
-            )
-        if not isinstance(panel_token, str) or not panel_token.strip():
-            raise ValueError("panel_token must be a non-empty string")
-        if (
-            not isinstance(timeout_seconds, (int, float))
-            or isinstance(timeout_seconds, bool)
-            or timeout_seconds <= 0
-        ):
-            raise ValueError(
-                f"timeout_seconds must be a positive number, got {timeout_seconds!r}"
-            )
-        self._base_url = base_url.strip().rstrip("/")
-        self._panel_token = panel_token
-        self._session = session
-        self._timeout_seconds = float(timeout_seconds)
-
-    async def get_snapshot(self) -> dict[str, Any]:
-        """Fetch and parse today's household snapshot.
-
-        Returns the JSON the snapshot route returns (``today_iso``,
-        ``cycle_day``, and the per-child list — see
-        ``api/routes_panel.py`` for the documented shape).  Raises
-        :class:`NestQuestApiError` on transport failure, timeout, or a
-        non-2xx response.
-        """
-        return await self._request_json("GET", SNAPSHOT_PATH)
-
-    async def complete_instance(
-        self, instance_id: int, actor_child_id: int
-    ) -> dict[str, Any]:
-        """Complete one quest instance as the tapped child profile.
-
-        POSTs ``{"actor_child_id": actor_child_id}`` to the complete
-        route and returns the parsed response body (the route answers
-        ``{"status": ...}`` both for a first completion and for an
-        idempotent repeat, so a retried panel tap succeeds).  Raises
-        :class:`NestQuestApiError` on transport failure, timeout, or a
-        non-2xx response — 401 for a wrong token, 404 for an unknown
-        instance.
-        """
-        path = COMPLETE_INSTANCE_PATH.format(instance_id=instance_id)
-        return await self._request_json(
-            "POST", path, json_body={"actor_child_id": actor_child_id}
-        )
-
-    async def stream_events(
-        self,
-    ) -> AsyncIterator[tuple[str, dict[str, Any]]]:
-        """Yield ``(event_type, payload)`` for every panel SSE frame.
-
-        Opens ``GET /api/v1/panel/events`` with the Bearer credential
-        and parses the Server-Sent Events framing the API service
-        emits: ``event: <type>`` and ``data: <json>`` lines closed by a
-        blank line (one frame per transition event; a frame's ``data``
-        may span several ``data:`` lines, joined with newlines — the
-        standard SSE rule).  Line splits across chunk boundaries are
-        buffered, so frames are parsed correctly whatever chunking the
-        transport produces.
-
-        Opening the stream carries the SAME one-timeout, typed-error
-        contract as every other call: a non-2xx response raises
-        :class:`NestQuestApiError` with its status and a transport
-        failure (connection refused, DNS, the open timing out) raises
-        one with ``status=None`` and the original error chained.  The
-        reading loop itself has NO timeout — a healthy stream is
-        long-lived — but any failure raised from the transport
-        mid-stream raises typed as well, so the consumer can reconnect.
-        A frame whose ``data`` is not valid JSON is a protocol failure
-        and raises typed, carrying the response status it arrived
-        with.  Cancellation is never converted: closing the generator
-        (``aclose``, i.e. the consumer's task being cancelled) stops
-        cleanly and closes the underlying response in every path.
-        """
-        url = f"{self._base_url}{EVENTS_PATH}"
-        headers = {"Authorization": f"Bearer {self._panel_token}"}
-        # Opening the response gets ONE request timeout, exactly like
-        # ``_request_json``; the read loop below runs WITHOUT one — a
-        # body-wide timeout would kill a healthy long-lived stream.
-        # The response context manager is entered manually (not via
-        # ``async with``) because the block must span the yields below;
-        # the ``finally`` closes it on EVERY exit path.
-        try:
-            async with asyncio.timeout(self._timeout_seconds):
-                stream_cm = self._session.request(
-                    "GET", url, headers=headers, json=None
-                )
-                response = await stream_cm.__aenter__()
-        except asyncio.CancelledError:
-            # Cancellation is never converted (see ``_request_json``).
-            raise
-        except NestQuestApiError:
-            raise
-        except Exception as err:
-            raise NestQuestApiError(
-                f"NestQuest API GET {EVENTS_PATH} transport failure: {err}",
-                status=None,
-            ) from err
-
-        status = response.status
-        if not 200 <= status < 300:
-            await self._close_stream(stream_cm)
-            raise NestQuestApiError(
-                f"NestQuest API GET {EVENTS_PATH} failed with HTTP {status}",
-                status=status,
-            )
-
-        try:
-            event_type: str | None = None
-            data_lines: list[str] = []
-            buffer = ""
-            async for chunk in response:
-                if isinstance(chunk, (bytes, bytearray)):
-                    chunk = chunk.decode("utf-8")
-                buffer += chunk
-                while "\n" in buffer:
-                    line, buffer = buffer.split("\n", 1)
-                    if line.startswith("event:"):
-                        event_type = line[len("event:"):].strip()
-                    elif line.startswith("data:"):
-                        data_lines.append(line[len("data:"):].strip())
-                    elif not line and event_type is not None:
-                        # A blank line ends the frame.  A trailing
-                        # incomplete frame (stream ends without its
-                        # closing blank line) is discarded, per the
-                        # SSE spec's treatment of incomplete events.
-                        try:
-                            payload = json.loads("\n".join(data_lines))
-                        except ValueError as err:
-                            raise NestQuestApiError(
-                                f"NestQuest API GET {EVENTS_PATH} "
-                                f"delivered a malformed SSE frame: {err}",
-                                status=status,
-                            ) from err
-                        yield event_type, payload
-                        event_type = None
-                        data_lines = []
-        except asyncio.CancelledError:
-            raise
-        except NestQuestApiError:
-            raise
-        except Exception as err:
-            raise NestQuestApiError(
-                f"NestQuest API GET {EVENTS_PATH} transport failure: {err}",
-                status=None,
-            ) from err
-        finally:
-            await self._close_stream(stream_cm)
-
-    @staticmethod
-    async def _close_stream(stream_cm: Any) -> None:
-        """Close an opened streaming response context manager.
-
-        Works for aiohttp's response context manager (whose
-        ``__aexit__`` is a coroutine) and for test stubs that return
-        either a coroutine or a plain value.
-        """
-        result = stream_cm.__aexit__(None, None, None)
-        if inspect.isawaitable(result):
-            await result
-
-    async def _request_json(
-        self,
-        method: str,
-        path: str,
-        *,
-        json_body: dict[str, Any] | None = None,
-    ) -> Any:
-        """Run ONE request and return the parsed JSON body.
-
-        The Bearer token rides the Authorization header only.  A
-        non-2xx response raises :class:`NestQuestApiError` with its
-        status; anything the transport raises (connection, timeout) is
-        wrapped as a transport-level :class:`NestQuestApiError` with
-        the original error chained — never swallowed, never retried.
-        A 2xx body that is not valid JSON is a protocol failure and
-        raises typed too, carrying the status it arrived with.
-        """
-        url = f"{self._base_url}{path}"
-        headers = {"Authorization": f"Bearer {self._panel_token}"}
-        try:
-            async with asyncio.timeout(self._timeout_seconds):
-                async with self._session.request(
-                    method, url, headers=headers, json=json_body
-                ) as response:
-                    status = response.status
-                    if not 200 <= status < 300:
-                        raise NestQuestApiError(
-                            f"NestQuest API {method} {path} failed with "
-                            f"HTTP {status}",
-                            status=status,
-                        )
-                    body_text = await response.text()
-        except asyncio.CancelledError:
-            # Cancellation is never converted: a shutting-down event
-            # loop must stop, not read a synthesized API error.
-            raise
-        except NestQuestApiError:
-            raise
-        except Exception as err:
-            raise NestQuestApiError(
-                f"NestQuest API {method} {path} transport failure: {err}",
-                status=None,
-            ) from err
-        try:
-            return json.loads(body_text)
-        except ValueError as err:
-            raise NestQuestApiError(
-                f"NestQuest API {method} {path} returned malformed "
-                f"JSON: {err}",
-                status=status,
-            ) from err

custom_components/nestquest/coordinator.py

@@ -1,4 +1,4 @@
-"""The shared NestQuest DataUpdateCoordinator (Feature 10, Feature 18).
+"""The shared NestQuest DataUpdateCoordinator (Feature 10).
 
 ONE refresh pass builds the whole entity snapshot: active children,
 today's instances with their derived states (open / done / missed via
@@ -7,81 +7,54 @@ per-child rollup counts.  Every Feature 10 entity reads
 ``coordinator.data`` — no entity queries the database itself, so one
 refresh updates the whole board coherently.
 
-Since Feature 18 the snapshot no longer comes from the local database:
-the coordinator polls the NestQuest API service's panel plane —
-``GET /api/v1/panel/snapshot`` through
-:class:`~.api_client.NestQuestApiClient` on its update interval — and
-RECONSTRUCTS the same core dataclasses the local builder produced
-(:class:`~.core.snapshot.NestQuestSnapshot`,
-:class:`~.core.snapshot.ChildDaySnapshot`,
-:class:`~.core.snapshot.QuestInstanceView`).  ``sensor.py`` and
-``binary_sensor.py`` read ``coordinator.data`` and
-:func:`~.core.snapshot.instance_payload` exactly as before, so every
-entity_id and attribute payload shape is unchanged: the panel cards and
-the four blueprints keep working untouched.
+"Today" and "now" are HA-local (``hass.config.time_zone``), never the
+host clock and never UTC, matching the materialization and completion
+layers.
 
-One loss is deliberate and documented (D-009 revisited): the panel
-payload OMITS missed instances, so a snapshot rebuilt from the API can
-no longer carry them — ``admin_instances`` (``include_missed=True``)
-now equals ``instances`` (they can hold only open and completed rows).
-The admin surface for missed quests is the PWA, which reads the API
-service directly; the integration's entities never needed missed rows
-for their states or counts (the counts arrive precomputed).
+The snapshot assembly itself lives in the Home-Assistant-free
+:mod:`custom_components.nestquest.core.snapshot` module
+(:func:`~core.snapshot.build_snapshot`); this coordinator is now a thin
+HA wrapper that resolves the local timezone and reads the settings handed
+into its constructor, then delegates to that builder, passing the SAME
+HA-local ``now`` it resolved in one clock read.  The API service's panel
+snapshot route shares the same builder, so both consumers render ONE
+snapshot shape.
 
-Failure handling: constructing the client fails fast (ValueError) when
-the entry has no panel token configured, and a fetch failure raises the
-typed :class:`~.api_client.NestQuestApiError`.  Neither crashes
-integration setup: the constructor logs a clear message and hands back
-a client-shaped stub whose refresh raises the same typed error, and
-``_async_update_data`` converts every typed error into
-:class:`~homeassistant.helpers.update_coordinator.UpdateFailed` — the
-exception a HA ``DataUpdateCoordinator`` records as a failed pass
-(``last_update_success = False``, ``data = None``), so the entities
-come up unavailable.  The typed error is never swallowed silently: it
-is logged with its message and chained into the UpdateFailed.  (The
-conversion matters at setup: HA's real
-``async_config_entry_first_refresh`` wraps any failed pass in
-``ConfigEntryNotReady``, so a setup that keys on the original
-exception type would never match it.)
+Presence is resolved through the SAME engine the materializer uses
+(:class:`~.presence.PresenceEngine`): schedules and overrides are read
+with the public DAO methods and the engine's anchor-date arithmetic
+answers ``is this child at this house today`` (D-004 — never ISO week
+parity).
 
-Last-good cache and staleness: every successful pass refreshes a cache
-of the rebuilt snapshot stamped with the :func:`time.monotonic`
-instant of the success.  A typed failure is checked against that stamp
-BEFORE the pass is declared failed: within the configured staleness
-threshold (``CONF_SNAPSHOT_STALENESS``, seconds, default three update
-intervals) the pass SUCCEEDS with the cached snapshot — the panel
-keeps rendering last-good data and the entities stay available — and
-the outage is logged as a warning.  Entities go unavailable only once
-the cache is older than the threshold (or before the first success);
-a later successful fetch refreshes the cache and the entities recover
-on their own, without a config-entry reload.
+``cycle_day`` reports where today falls in the custody cycle of the
+first active child (display order) that HAS a schedule — 1-based day
+``N`` of ``cycle_length_weeks * 7``, computed from that schedule's
+anchor date.  ``0`` means no active child has a schedule at all, so
+the admin header degrades to a harmless zero instead of guessing.
 """
 from __future__ import annotations
 
 import asyncio
 import datetime
-import time
-from typing import Any
+from zoneinfo import ZoneInfo
 
 from homeassistant.core import HomeAssistant
-from homeassistant.helpers.update_coordinator import (
-    DataUpdateCoordinator,
-    UpdateFailed,
-)
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
-from .api_client import NestQuestApiClient, NestQuestApiError, resolve_api_config
 from .const import (
     CONF_UPDATE_INTERVAL,
-    DEFAULT_SNAPSHOT_STALENESS,
     DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
     LOGGER,
     MIN_UPDATE_INTERVAL,
 )
+from .core.db import NestQuestDatabase
+from .core.settings import NestQuestSettings
 from .core.snapshot import (
     ChildDaySnapshot,
     NestQuestSnapshot,
     QuestInstanceView,
+    build_snapshot,
 )
 
 __all__ = [
@@ -92,131 +65,9 @@ __all__ = [
 ]
 
 
-def snapshot_from_api_payload(payload: dict[str, Any]) -> NestQuestSnapshot:
-    """Reconstruct the core snapshot dataclasses from the API's JSON.
-
-    The panel snapshot route (``api/routes_panel.py``) returns today's
-    household as plain JSON — ``today_iso``, ``cycle_day``, and the
-    per-child list with precomputed counts and the instances in the
-    documented payload shape.  This function maps that shape BACK onto
-    the dataclasses ``sensor.py`` / ``binary_sensor.py`` read, field by
-    field:
-
-    - the API's ``state`` ``"completed"`` maps back to the core view's
-      ``"done"`` (so :func:`~.core.snapshot.instance_payload` renders
-      ``"completed"`` again — the round-trip is identity for open and
-      completed rows);
-    - ``overdue``, ``on_time`` (→ ``was_on_time``), ``completed_at``,
-      ``window``, ``due_time``, ``icon`` and the per-child counts pass
-      through verbatim;
-    - ``present`` and ``next_present`` pass through verbatim (the API
-      resolved them through the same presence engine server-side);
-    - the instance list is rebuilt IN ORDER, so the panel payload the
-      sensors emit matches the route's order byte for byte.
-
-    Because the route omits missed instances (D-009), a snapshot built
-    here cannot contain any ``missed`` view — see the module docstring.
-    """
-    children: list[ChildDaySnapshot] = []
-    for child in payload.get("children", []):
-        views = tuple(
-            QuestInstanceView(
-                instance_id=instance["id"],
-                definition_id=instance["definition_id"],
-                child_id=instance["child_id"],
-                title=instance["title"],
-                icon=instance.get("icon"),
-                window=instance["window"],
-                # The API payload does not carry due_date: every row
-                # is TODAY's instance (the route queries
-                # [today, today]), so the view's due_date is the
-                # snapshot's today_iso — the same value the local
-                # builder stamped on every today row.
-                due_date=payload["today_iso"],
-                due_time=instance.get("due_time"),
-                state=(
-                    "done" if instance["state"] == "completed" else
-                    instance["state"]
-                ),
-                overdue=instance["overdue"],
-                completed_at=instance.get("completed_at"),
-                was_on_time=instance.get("on_time"),
-            )
-            for instance in child.get("instances", [])
-        )
-        children.append(
-            ChildDaySnapshot(
-                child_id=child["child_id"],
-                child_name=child["child_name"],
-                instances=views,
-                present=child["present"],
-                due_today=child["due_today"],
-                completed_today=child["completed_today"],
-                remaining_today=child["remaining_today"],
-                completion_pct=child["completion_pct"],
-                next_present=child.get("next_present"),
-            )
-        )
-    return NestQuestSnapshot(
-        today_iso=payload["today_iso"],
-        children=tuple(children),
-        cycle_day=payload["cycle_day"],
-    )
-
-
-class _UnconfiguredApiClient:
-    """Client-shaped stand-in used when the entry has no panel token.
-
-    Constructing a real :class:`~.api_client.NestQuestApiClient` with an
-    empty token raises ValueError by design (fail fast on a client that
-    could only ever 401).  An entry that simply has not been configured
-    yet must not crash integration setup, so the coordinator swaps in
-    this stub: every refresh raises the SAME typed
-    :class:`~.api_client.NestQuestApiError` a transport failure would,
-    carrying a clear "API not configured" message;
-    :meth:`NestQuestCoordinator._async_update_data` converts it to
-    :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`, so
-    the coordinator marks the entities unavailable and the HA standard
-    retry cycle applies.  The error is logged, never silent.
-    """
-
-    __slots__ = ()
-
-    async def get_snapshot(self) -> dict[str, Any]:
-        """Always raise: there is no configured API to fetch from."""
-        raise NestQuestApiError(
-            "NestQuest API not configured: set the panel token in the "
-            "integration options"
-        )
-
-
-def _monotonic() -> float:
-    """The monotonic clock the staleness math reads.
-
-    A one-line indirection over :func:`time.monotonic` so the staleness
-    tests can drive the clock deterministically (patch THIS function,
-    never the stdlib clock).  Monotonic is deliberate: wall-clock jumps
-    (NTP corrections, suspends) must not shorten or extend the window.
-    """
-    return time.monotonic()
-
-
 class NestQuestCoordinator(DataUpdateCoordinator):
     """Shared refresh cycle for every NestQuest entity.
 
-    One refresh pass = one ``GET /api/v1/panel/snapshot`` through the
-    API client, reconstructed into the core snapshot dataclasses every
-    entity reads.  A failed pass (unconfigured API, transport failure,
-    non-2xx response) raises the typed error out of the client;
-    ``_async_update_data`` converts it to
-    :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`, the
-    exception HA's coordinator machinery records as a failed pass: the
-    coordinator's ``data`` keeps its last good value while
-    ``last_update_success`` goes False — HA marks the entities
-    unavailable until a later pass succeeds.  An unexpected
-    non-typed error still propagates out of ``_async_update_data``
-    and is logged (with its traceback) by HA's machinery, the same way.
-
     Forced refreshes (the service paths pushing an immediate update
     after a completion) are SERIALIZED per entry: overlapping
     complete/uncomplete calls must not interleave their snapshot
@@ -224,15 +75,6 @@ class NestQuestCoordinator(DataUpdateCoordinator):
     newer one and leave entities wrong until the next poll.  Real
     HA's coordinator already serializes internally; the lock keeps
     the same guarantee on the stand-in and costs nothing upstream.
-
-    Last-good cache: a successful pass stamps the rebuilt snapshot
-    with the :func:`~.coordinator._monotonic` instant of the success;
-    a typed failure within the configured staleness threshold
-    (``staleness_seconds``, from ``CONF_SNAPSHOT_STALENESS``) serves
-    that cached snapshot instead of failing the pass, so a short API
-    outage keeps the board up.  Past the threshold — or before the
-    first success — the pass fails as above, and a later successful
-    pass refreshes the cache and recovers the entities on its own.
     """
 
     def __init__(
@@ -240,9 +82,9 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         hass: HomeAssistant,
         *,
         entry_id: str,
-        api_client: NestQuestApiClient,
+        database: NestQuestDatabase,
+        settings: NestQuestSettings,
         update_interval_seconds: int | None = None,
-        staleness_seconds: int | None = None,
     ) -> None:
         interval = update_interval_seconds
         if (
@@ -251,19 +93,6 @@ class NestQuestCoordinator(DataUpdateCoordinator):
             or interval < MIN_UPDATE_INTERVAL
         ):
             interval = DEFAULT_UPDATE_INTERVAL
-        staleness = staleness_seconds
-        if (
-            not isinstance(staleness, int)
-            or isinstance(staleness, bool)
-            or staleness < interval
-        ):
-            # A missing or nonsensical threshold falls back to the
-            # shipped default; a configured-but-too-short one (below
-            # even one poll interval) falls back to whichever of
-            # default/interval spans at least one failed poll, so the
-            # cache is never dead on arrival.  (The options flow
-            # validates the same rule at the form.)
-            staleness = max(DEFAULT_SNAPSHOT_STALENESS, interval)
         super().__init__(
             hass,
             LOGGER,
@@ -271,16 +100,9 @@ class NestQuestCoordinator(DataUpdateCoordinator):
             update_interval=datetime.timedelta(seconds=interval),
         )
         self.entry_id = entry_id
-        self.api_client = api_client
+        self.database = database
+        self.settings = settings
         self._refresh_lock = asyncio.Lock()
-        #: The configured last-good window, in SECONDS (const.py:
-        #: CONF_SNAPSHOT_STALENESS / DEFAULT_SNAPSHOT_STALENESS).
-        self._staleness_seconds = staleness
-        #: The last-good cache: the snapshot from the most recent
-        #: SUCCESSFUL fetch and the :func:`time.monotonic` instant it
-        #: landed.  Both stay None until the first success.
-        self._last_good_snapshot: NestQuestSnapshot | None = None
-        self._last_good_monotonic: float | None = None
 
     async def async_refresh(self) -> None:
         """One refresh at a time; a queued refresh runs AFTER the
@@ -289,98 +111,12 @@ class NestQuestCoordinator(DataUpdateCoordinator):
         async with self._refresh_lock:
             await super().async_refresh()
 
+    def _local_now(self) -> datetime.datetime:
+        time_zone = ZoneInfo(self.hass.config.time_zone)
+        return datetime.datetime.now(time_zone)
+
     async def _async_update_data(self) -> NestQuestSnapshot:
-        """One refresh pass: fetch the panel snapshot and rebuild it.
-
-        A successful pass rebuilds the snapshot and refreshes the
-        last-good cache (the snapshot plus the
-        :func:`time.monotonic` instant of the success).
-
-        A typed API failure (unconfigured token stub, transport
-        failure, non-2xx response) is first checked against that
-        stamp: WITHIN the configured staleness threshold (elapsed time
-        since the last successful fetch, inclusive of the threshold
-        itself) the pass SUCCEEDS with the cached snapshot — the panel
-        keeps rendering last-good data, the entities stay available —
-        and the outage is logged as a clear warning (never silent).
-        With no cache yet, or past the threshold, the typed error is
-        converted to
-        :class:`~homeassistant.helpers.update_coordinator.UpdateFailed`
-        — the exception HA's coordinator machinery records as a failed
-        pass (``last_update_success = False``, ``data`` unchanged) so
-        entities go unavailable and the setup-time first refresh
-        arrives as ``ConfigEntryNotReady`` — with the typed error
-        logged and chained.  Anything else still propagates.
-        """
-        try:
-            payload = await self.api_client.get_snapshot()
-        except NestQuestApiError as err:
-            now = _monotonic()
-            cached = self._last_good_snapshot
-            fetched_at = self._last_good_monotonic
-            if (
-                cached is not None
-                and fetched_at is not None
-                and now - fetched_at <= self._staleness_seconds
-            ):
-                LOGGER.warning(
-                    "NestQuest API refresh failed (%s); serving the "
-                    "last-good snapshot fetched %.0f s ago, within the "
-                    "%d s staleness threshold",
-                    err,
-                    now - fetched_at,
-                    self._staleness_seconds,
-                )
-                return cached
-            LOGGER.error(
-                "NestQuest coordinator refresh failed: %s", err
-            )
-            raise UpdateFailed(str(err)) from err
-        snapshot = snapshot_from_api_payload(payload)
-        self._last_good_snapshot = snapshot
-        self._last_good_monotonic = _monotonic()
-        return snapshot
-
-
-def coordinator_client_from_entry(
-    hass: HomeAssistant, entry: Any, *, session_factory: Any = None
-) -> Any:
-    """Build the refresh client for a config entry, tolerating no token.
-
-    Reads ``(base_url, panel_token)`` off the entry and constructs the
-    real client.  An entry with no panel token configured (the shipped
-    default is the empty string) makes the constructor raise ValueError
-    by design; integration setup must not crash on an unconfigured
-    entry, so this factory swaps in the typed-error stub instead and
-    logs ONE clear message.  The ValueError itself is never re-raised
-    past this point — it is converted, with the reason logged.
-
-    ``session_factory`` overrides where the HTTP session comes from
-    (production: HA's shared aiohttp session via
-    :func:`~.api_client.client_from_entry`); it exists so tests can
-    inject a stub transport without importing aiohttp — the mock-only
-    test harness has no ``homeassistant.helpers.aiohttp_client``.
-    """
-    base_url, panel_token = resolve_api_config(entry)
-    if not isinstance(panel_token, str) or not panel_token.strip():
-        LOGGER.warning(
-            "NestQuest API panel token is not configured for entry %s: "
-            "entities will stay unavailable until a token is set in the "
-            "integration options",
-            entry.entry_id,
+        now = self._local_now()
+        return await build_snapshot(
+            self.database, self.settings, now
         )
-        return _UnconfiguredApiClient()
-    if session_factory is not None:
-        return NestQuestApiClient(
-            base_url, panel_token, session_factory(hass)
-        )
-    from custom_components.nestquest.api_client import client_from_entry
-
-    return client_from_entry(hass, entry)
-
-
-#: Test seam: entry_id → client, consulted by the conftest's client
-#: factory BEFORE the production path.  Always empty in production
-#: (nothing in the integration writes to it); tests register scripted
-#: clients through it and clear it between tests.
-_COORDINATOR_CLIENT_OVERRIDES: dict[str, Any] = {}

custom_components/nestquest/core/const.py

@@ -81,21 +81,6 @@ CONF_UPDATE_INTERVAL = "update_interval"
 DEFAULT_UPDATE_INTERVAL = 300
 MIN_UPDATE_INTERVAL = 30
 
-# Last-good snapshot staleness (Feature 18), in SECONDS: how long the
-# coordinator keeps SERVING the last good snapshot after the API stops
-# answering.  A failed poll whose elapsed time since the last successful
-# fetch is within the threshold returns the cached snapshot (the panel
-# keeps rendering last-good data, entities stay available); past it the
-# poll fails the HA way and the entities go unavailable.  The default is
-# 3x the default update interval (900 s = 15 minutes): three consecutive
-# failed five-minute polls before the board is declared stale — enough
-# to ride out an API service restart, short enough not to render a
-# days-old board all day.  Validation floor: the threshold must be at
-# least the configured update interval, or the cache could never span
-# even one failed poll.
-CONF_SNAPSHOT_STALENESS = "snapshot_staleness"
-DEFAULT_SNAPSHOT_STALENESS = 3 * DEFAULT_UPDATE_INTERVAL
-
 # Notification automations (Feature 11).  The notify target is a free-text
 # Home Assistant service name (e.g. "notify.mobile_app_dads_phone") the
 # household fills in — never a hard-coded personal target; the times are
@@ -116,32 +101,6 @@ CONF_END_OF_DAY_REPORT_ENABLED = "end_of_day_report_enabled"
 CONF_CELEBRATION_ENABLED = "celebration_enabled"
 DEFAULT_AUTOMATION_ENABLED = True
 
-# Panel-plane API client (Feature 18): the integration reaches the
-# NestQuest API service (Feature 16) over HTTP instead of the database.
-# The base URL points at the API service host serving ``/api/v1/panel/*``
-# (uvicorn's default host/port is the sensible starting point); the panel
-# service token is the static Bearer credential the panel plane requires
-# (``NESTQUEST_PANEL_TOKEN`` on the API side).  Both are stored on the
-# entry like every other setting — options first, then the entry-data
-# copy, then defaults.  The TOKEN IS A SECRET: it is never logged, never
-# echoed into error messages, and never committed.  The shipped default
-# token IS the empty string (DEFAULT_PANEL_TOKEN) — an entry with no
-# configured token is therefore NOT a working client: the options flow
-# accepts an empty token as "not configured yet" and only rejects
-# malformed (padded/whitespace-only/non-string) values — the actual
-# non-emptiness guard is enforced at client construction, where
-# NestQuestApiClient fails fast on an empty/whitespace token
-# (api_client.py).  So a caller building the client from a bare
-# entry must either ensure a token is configured first or handle the
-# construction ValueError and surface an "API not configured" state —
-# an unconfigured entry never yields a client that can only 401.
-# This mirrors the API service's fail-closed stance (api/config.py: no
-# default token exists).
-CONF_API_BASE_URL = "api_base_url"
-DEFAULT_API_BASE_URL = "http://127.0.0.1:8000"
-CONF_PANEL_TOKEN = "panel_token"
-DEFAULT_PANEL_TOKEN = ""
-
 # Quest windows (D-008): a definition may declare several day windows;
 # each produces its own instance per day and groups the panel's Quest
 # Log into three columns.  The clock ranges are stored here so they stay

custom_components/nestquest/core/sweep.py

@@ -1,143 +0,0 @@
-"""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/options_flow.py

@@ -12,7 +12,6 @@ from .const import (
     CONF_ADMIN_USER_IDS,
     CONF_AFTERNOON_REMINDER_ENABLED,
     CONF_AFTERNOON_REMINDER_TIME,
-    CONF_API_BASE_URL,
     CONF_CELEBRATION_ENABLED,
     CONF_DAY_ROLLOVER_TIME,
     CONF_END_OF_DAY_REPORT_ENABLED,
@@ -22,10 +21,7 @@ from .const import (
     CONF_MORNING_SUMMARY_TIME,
     CONF_NOTIFY_TARGET,
     CONF_PANEL_IDLE_TIMEOUT,
-    CONF_PANEL_TOKEN,
-    CONF_SNAPSHOT_STALENESS,
     CONF_UPDATE_INTERVAL,
-    DEFAULT_API_BASE_URL,
     DEFAULT_AFTERNOON_REMINDER_TIME,
     DEFAULT_AUTOMATION_ENABLED,
     DEFAULT_DAY_ROLLOVER_TIME,
@@ -33,15 +29,12 @@ from .const import (
     DEFAULT_HORIZON_DAYS,
     DEFAULT_MORNING_SUMMARY_TIME,
     DEFAULT_PANEL_IDLE_TIMEOUT,
-    DEFAULT_PANEL_TOKEN,
-    DEFAULT_SNAPSHOT_STALENESS,
     DEFAULT_UPDATE_INTERVAL,
     DOMAIN,
     MIN_UPDATE_INTERVAL,
     TIME_PATTERN,
 )
 from .admin_allowlist import list_admin_ids, set_admin_ids
-from .api_client import is_valid_base_url
 
 MIN_HORIZON_DAYS = 1
 MIN_PANEL_IDLE_TIMEOUT = 30
@@ -52,18 +45,8 @@ def _parse_time(value: str) -> bool:
     return TIME_PATTERN.fullmatch(value) is not None
 
 
-def _async_validate(
-    user_input: dict[str, Any],
-    current: dict[str, Any] | None = None,
-) -> dict[str, str]:
-    """Validate user input and return field-level errors instead of raising.
-
-    ``current`` carries the stored values (entry options → entry data →
-    defaults) so a raw submit that sets one field but omits a sibling it
-    is validated against (the staleness threshold vs. the update
-    interval) can still check the effective pair; raw submits omitting
-    BOTH keys keep their stored values untouched.
-    """
+def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
+    """Validate user input and return field-level errors instead of raising."""
     errors: dict[str, str] = {}
 
     horizon_days = user_input.get(CONF_HORIZON_DAYS)
@@ -98,33 +81,6 @@ def _async_validate(
     ):
         errors[CONF_UPDATE_INTERVAL] = "invalid"
 
-    # The last-good staleness threshold (Feature 18) is OPTIONAL in raw
-    # submits the same way: a missing key keeps the current value.  It
-    # must be a positive int, and at least the EFFECTIVE update interval
-    # (the one this submit carries, else the stored or default one) — a
-    # threshold shorter than the poll cadence could never serve even one
-    # cached snapshot.
-    staleness = user_input.get(CONF_SNAPSHOT_STALENESS)
-    if staleness is not None and (
-        isinstance(staleness, bool)
-        or not isinstance(staleness, int)
-        or staleness < 1
-    ):
-        errors[CONF_SNAPSHOT_STALENESS] = "invalid"
-    elif staleness is not None and not errors.get(CONF_UPDATE_INTERVAL):
-        stored_interval = (current or {}).get(CONF_UPDATE_INTERVAL)
-        effective_interval = update_interval
-        if effective_interval is None and (
-            isinstance(stored_interval, int)
-            and not isinstance(stored_interval, bool)
-            and stored_interval >= MIN_UPDATE_INTERVAL
-        ):
-            effective_interval = stored_interval
-        if effective_interval is None:
-            effective_interval = DEFAULT_UPDATE_INTERVAL
-        if staleness < effective_interval:
-            errors[CONF_SNAPSHOT_STALENESS] = "invalid_staleness"
-
     # The notification section (Feature 11) is OPTIONAL in raw submits
     # the same way: a missing key keeps the current value.  The three
     # times are strict HH:MM; the notify target is a non-empty HA
@@ -160,25 +116,6 @@ def _async_validate(
         if toggle_value is not None and not isinstance(toggle_value, bool):
             errors[toggle_key] = "invalid"
 
-    # The panel-plane API connection (Feature 18) is OPTIONAL in raw
-    # submits the same way: a missing key keeps the current value.  The
-    # base URL must be a full http(s) URL with a host — the same check
-    # the client's constructor enforces, so a malformed URL fails at
-    # the form, not at the first request.  The token may be EMPTY (the
-    # API connection is simply not configured yet) but a non-empty one
-    # must be pasted unpadded: a padded or whitespace-only value is a
-    # clipboard mistake that would fail the token check at request
-    # time.
-    api_base_url = user_input.get(CONF_API_BASE_URL)
-    if api_base_url is not None and not is_valid_base_url(api_base_url):
-        errors[CONF_API_BASE_URL] = "invalid_url"
-    panel_token = user_input.get(CONF_PANEL_TOKEN)
-    if panel_token is not None and (
-        not isinstance(panel_token, str)
-        or panel_token != panel_token.strip()
-    ):
-        errors[CONF_PANEL_TOKEN] = "invalid"
-
     return errors
 
 
@@ -218,14 +155,6 @@ def _build_schema(
         vol.Required(
             CONF_UPDATE_INTERVAL, default=current[CONF_UPDATE_INTERVAL]
         ): int,
-        # The last-good staleness threshold (Feature 18): how long the
-        # coordinator keeps serving the cached snapshot after the API
-        # stops answering.  Schema-validated submits always carry it
-        # (the default pre-fills from the stored value); raw callers
-        # omitting it keep their stored value (see _async_validate).
-        vol.Required(
-            CONF_SNAPSHOT_STALENESS, default=current[CONF_SNAPSHOT_STALENESS]
-        ): int,
         vol.Required(
             CONF_NOTIFY_TARGET, default=current[CONF_NOTIFY_TARGET]
         ): str,
@@ -257,18 +186,6 @@ def _build_schema(
             CONF_CELEBRATION_ENABLED,
             default=current[CONF_CELEBRATION_ENABLED],
         ): _strict_bool,
-        # Panel-plane API connection (Feature 18): the base URL the API
-        # client talks to and the panel service token it presents as
-        # the Bearer credential.  Schema-validated submits always carry
-        # both (the defaults pre-fill from the stored values); raw
-        # callers omitting them keep their stored values (see
-        # _async_validate).
-        vol.Required(
-            CONF_API_BASE_URL, default=current[CONF_API_BASE_URL]
-        ): str,
-        vol.Required(
-            CONF_PANEL_TOKEN, default=current[CONF_PANEL_TOKEN]
-        ): str,
     }
     if admin_choices is not None:
         # HA's supported multi-select validator: the frontend can
@@ -338,7 +255,7 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
                 ),
             )
 
-        errors = _async_validate(user_input, self._current_values())
+        errors = _async_validate(user_input)
         # The admin ids are validated only when the key is present:
         # schema-validated submits always carry it (vol.Optional's
         # default fills from the current allowlist), while raw callers
@@ -401,10 +318,6 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
                 CONF_UPDATE_INTERVAL,
                 self._current_values()[CONF_UPDATE_INTERVAL],
             ),
-            CONF_SNAPSHOT_STALENESS: user_input.get(
-                CONF_SNAPSHOT_STALENESS,
-                self._current_values()[CONF_SNAPSHOT_STALENESS],
-            ),
         }
         # The notification section (Feature 11) stores the household's
         # choices the same fall-back-to-current way, so raw submits
@@ -423,11 +336,6 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             data[notification_key] = user_input.get(
                 notification_key, current[notification_key]
             )
-        # The API connection section (Feature 18) stores the base URL
-        # and panel token the same fall-back-to-current way, so raw
-        # submits that predate the section keep their stored values.
-        for api_key in (CONF_API_BASE_URL, CONF_PANEL_TOKEN):
-            data[api_key] = user_input.get(api_key, current[api_key])
         if admin_ids_present:
             data[CONF_ADMIN_USER_IDS] = admin_ids
         elif admin_default:
@@ -449,7 +357,6 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             (CONF_DAY_ROLLOVER_TIME, DEFAULT_DAY_ROLLOVER_TIME),
             (CONF_PANEL_IDLE_TIMEOUT, DEFAULT_PANEL_IDLE_TIMEOUT),
             (CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL),
-            (CONF_SNAPSHOT_STALENESS, DEFAULT_SNAPSHOT_STALENESS),
             (CONF_NOTIFY_TARGET, ""),
             (CONF_MORNING_SUMMARY_TIME, DEFAULT_MORNING_SUMMARY_TIME),
             (CONF_AFTERNOON_REMINDER_TIME, DEFAULT_AFTERNOON_REMINDER_TIME),
@@ -458,8 +365,6 @@ class NestQuestOptionsFlow(config_entries.OptionsFlowWithConfigEntry):
             (CONF_AFTERNOON_REMINDER_ENABLED, DEFAULT_AUTOMATION_ENABLED),
             (CONF_END_OF_DAY_REPORT_ENABLED, DEFAULT_AUTOMATION_ENABLED),
             (CONF_CELEBRATION_ENABLED, DEFAULT_AUTOMATION_ENABLED),
-            (CONF_API_BASE_URL, DEFAULT_API_BASE_URL),
-            (CONF_PANEL_TOKEN, DEFAULT_PANEL_TOKEN),
         ):
             if key in options:
                 values[key] = options[key]

custom_components/nestquest/services.py

@@ -6,10 +6,7 @@ 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 (the
 presence writes included, through :mod:`.presence_management` — the
-ONE implementation the API admin plane shares).  Since Feature 18 the
-integration is a client of the NestQuest API service, so
-``complete_quest`` proxies to the API's panel complete route through
-the entry's shared API client instead of writing any database.
+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__``.
@@ -29,7 +26,6 @@ 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 .api_client import NestQuestApiError
 from .const import (
     DOMAIN,
     DOMAIN_SERVICES,
@@ -46,7 +42,7 @@ from .const import (
     SERVICE_UNCOMPLETE_QUEST,
     SERVICE_UPDATE_QUEST_DEFINITION,
 )
-from .events import fire_quest_uncompleted
+from .events import fire_quest_completed, fire_quest_uncompleted
 from .permissions import permission_gate
 from .recurrence import ScheduleRule
 from .service_policy import SERVICE_POLICY
@@ -305,50 +301,22 @@ def _complete_quest(
     async def _handler(call: Any) -> None:
         runtime = _require_runtime(hass, find_runtime)
         instance_id = call.data["instance_id"]
-        # The panel contract (decision 9): a tap completes as the
-        # tapped child profile, the only actor shape the API's panel
-        # route accepts.  The payload schema is unchanged — the panel
-        # cards call it with instance_id, actor=panel, actor_child_id.
-        actor = _actor_kwargs(call)
-        if actor["actor_source"] != ACTOR_PANEL:
-            raise HomeAssistantError(
-                "nestquest.complete_quest completes through the API's "
-                "panel route as the tapped child profile; actor must "
-                "be 'panel' with actor_child_id"
-            )
-        # The entry's ONE shared API client (built at setup): the
-        # coordinator's snapshot poll, the SSE subscription, and this
-        # proxy all ride the same instance.  An entry with no panel
-        # token runs the unconfigured stub, which has no completion
-        # surface — fail fast with a clear, token-free message.
-        complete_instance = getattr(
-            getattr(runtime.coordinator, "api_client", None),
-            "complete_instance",
-            None,
+        # ``appended`` is decided under the same lock as the write, so
+        # a concurrent duplicate call (a double tap) can not produce a
+        # second transition event: the loser appends nothing.
+        result = await completion_layer.complete_instance(
+            runtime.database,
+            instance_id,
+            **_actor_kwargs(call),
         )
-        if complete_instance is None:
-            raise HomeAssistantError(
-                "NestQuest API is not configured; cannot complete a "
-                "quest"
+        if result.appended:
+            await fire_quest_completed(
+                hass,
+                runtime.database,
+                instance_id,
+                was_on_time=result.was_on_time,
             )
-        try:
-            await complete_instance(instance_id, actor["actor_child_id"])
-        except NestQuestApiError as err:
-            # The typed API failure surfaces as a service error the
-            # caller sees — never swallowed, never retried here.  The
-            # typed message names method, path, and status only; the
-            # panel token never appears in it.
-            raise HomeAssistantError(
-                f"NestQuest quest completion failed: {err}"
-            ) from err
-        # NO local completion write and NO local event: the API service
-        # emits the transition on its SSE stream and the integration's
-        # subscription re-fires it on the bus — firing it here too
-        # would double-fire.  The immediate refresh keeps the old
-        # behaviour where the sensors reflect the completion the
-        # moment the service call returns instead of waiting for the
-        # poll interval.
-        await _refresh_entities(runtime)
+            await _refresh_entities(runtime)
 
     return _with_errors(_handler)
 

custom_components/nestquest/sse.py

@@ -1,172 +0,0 @@
-"""HA-bus subscription over the API service's panel SSE stream.
-
-The NestQuest API service (Feature 16) publishes every quest
-transition on ``GET /api/v1/panel/events`` as Server-Sent Events
-frames.  This module is the integration-side subscriber: ONE
-background task consumes :meth:`~.api_client.NestQuestApiClient
-.stream_events` and re-fires each frame on ``hass.bus`` with the SAME
-event type and payload fields the local transition path
-(:mod:`~.events`) emits — ``nestquest_quest_completed``,
-``nestquest_quest_uncompleted``, ``nestquest_quest_missed`` and
-``nestquest_child_day_complete`` — so the four blueprints fire
-identically whether a completion happens through the integration's own
-service or through the API service's panel route.
-
-Failure handling: a failed stream (API down, network blip, non-2xx, a
-dropped connection) is retried FOREVER with a bounded exponential
-backoff — :data:`RECONNECT_MIN_SECONDS` after the first failure,
-doubling up to :data:`RECONNECT_MAX_SECONDS` — so a long API outage
-neither pins the event loop nor spams the log, and a frame received on
-a healthy stream resets the backoff.  Nothing escapes the task: every
-failure is logged with its reason and the loop reconnects; a
-:class:`asyncio.CancelledError` (from :meth:`stop` or integration
-unload) is the only thing that ends it, and it stops cleanly — no
-frame fires after :meth:`stop` returns and the response is closed.
-"""
-from __future__ import annotations
-
-import asyncio
-from typing import Any
-
-from .const import LOGGER
-
-__all__ = [
-    "NestQuestEventStream",
-    "RECONNECT_MAX_SECONDS",
-    "RECONNECT_MIN_SECONDS",
-]
-
-#: Delay before the FIRST reconnect attempt after a stream failure.
-#: One failed frame must not turn into a hot reconnect loop.
-RECONNECT_MIN_SECONDS = 5.0
-
-#: Ceiling for the exponential reconnect backoff: a long API outage
-#: retries at a calm fixed cadence instead of growing without bound.
-RECONNECT_MAX_SECONDS = 60.0
-
-
-class NestQuestEventStream:
-    """Consume the panel SSE stream in the background, fire the HA bus.
-
-    ONE instance per config entry, built in integration setup with the
-    SAME API client the coordinator polls (the client is stateless —
-    one shared session, one connection pool).  ``start()`` schedules
-    the background task; ``stop()`` cancels and awaits it, so unload
-    never leaks the task and no frame fires after teardown.  The task
-    itself is failure-proof: any exception from the stream (typed
-    :class:`~.api_client.NestQuestApiError` or anything else) is
-    logged and retried after the current backoff, never allowed to
-    escape the task.
-    """
-
-    def __init__(
-        self,
-        hass: Any,
-        client: Any,
-        *,
-        reconnect_min_seconds: float = RECONNECT_MIN_SECONDS,
-        reconnect_max_seconds: float = RECONNECT_MAX_SECONDS,
-    ) -> None:
-        """Store the hass bus target, the API client, and the backoff.
-
-        The client is anything exposing ``stream_events()`` (the real
-        :class:`~.api_client.NestQuestApiClient` in production, a stub
-        in tests).  The backoff bounds must be positive and the minimum
-        must not exceed the maximum — a misconfiguration should fail
-        loudly at construction, not quietly reconnect at delay 0.
-        """
-        if (
-            not isinstance(reconnect_min_seconds, (int, float))
-            or isinstance(reconnect_min_seconds, bool)
-            or reconnect_min_seconds <= 0
-            or reconnect_min_seconds > reconnect_max_seconds
-        ):
-            raise ValueError(
-                "reconnect_min_seconds must be a positive number no "
-                f"greater than reconnect_max_seconds, got "
-                f"{reconnect_min_seconds!r}"
-            )
-        if (
-            not isinstance(reconnect_max_seconds, (int, float))
-            or isinstance(reconnect_max_seconds, bool)
-            or reconnect_max_seconds <= 0
-        ):
-            raise ValueError(
-                f"reconnect_max_seconds must be a positive number, got "
-                f"{reconnect_max_seconds!r}"
-            )
-        self._hass = hass
-        self._client = client
-        self._min_backoff = float(reconnect_min_seconds)
-        self._max_backoff = float(reconnect_max_seconds)
-        self._backoff = float(reconnect_min_seconds)
-        self._task: asyncio.Task | None = None
-
-    @property
-    def task(self) -> asyncio.Task | None:
-        """The background task, or ``None`` when never started/stopped."""
-        return self._task
-
-    def start(self) -> None:
-        """Schedule the subscription task (idempotent while it runs)."""
-        if self._task is not None and not self._task.done():
-            return
-        self._backoff = self._min_backoff
-        self._task = asyncio.create_task(
-            self._run(), name="nestquest-sse-subscription"
-        )
-
-    async def stop(self) -> None:
-        """Cancel and await the subscription task.
-
-        Returns only after the task has actually finished, so unload
-        can rely on: no further frame is fired after ``stop()``, and
-        the underlying stream response has been closed.  Safe to call
-        when the task was never started or already stopped.
-        """
-        task, self._task = self._task, None
-        if task is None:
-            return
-        task.cancel()
-        try:
-            await task
-        except asyncio.CancelledError:
-            # The task's own cancellation surfaces here — expected and
-            # swallowed; a cancelling the CALLER would re-raise out of
-            # the surrounding await on its own.
-            pass
-
-    async def _run(self) -> None:
-        """Consume stream frames forever, reconnecting on failure."""
-        while True:
-            try:
-                async for event_type, payload in self._client.stream_events():
-                    # A frame arrived on a healthy stream: the next
-                    # failure reconnects quickly again.
-                    self._backoff = self._min_backoff
-                    try:
-                        self._hass.bus.async_fire(event_type, dict(payload))
-                    except Exception as err:
-                        # Firing one frame must never tear down a
-                        # healthy connection: log and keep consuming.
-                        LOGGER.error(
-                            "NestQuest panel event %s could not be fired "
-                            "on the Home Assistant event bus: %s",
-                            event_type,
-                            err,
-                        )
-            except asyncio.CancelledError:
-                # Unload / shutdown: stop cleanly, never converted.
-                raise
-            except Exception as err:
-                delay = self._backoff
-                self._backoff = min(
-                    self._backoff * 2, self._max_backoff
-                )
-                LOGGER.warning(
-                    "NestQuest panel event stream failed (%s); "
-                    "reconnecting in %.1f s",
-                    err,
-                    delay,
-                )
-                await asyncio.sleep(delay)

custom_components/nestquest/strings.json

@@ -18,14 +18,11 @@
     "step": {
       "init": {
         "title": "NestQuest Settings",
-        "description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
+        "description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
           "panel_idle_timeout": "Panel idle timeout (seconds)",
-          "snapshot_staleness": "Last-good snapshot staleness (seconds)",
-          "api_base_url": "NestQuest API base URL (http://host:port)",
-          "panel_token": "Panel service token",
           "admin_user_ids": "Admin users (parents)"
         }
       }
@@ -33,8 +30,6 @@
     "errors": {
       "invalid": "Invalid value.",
       "invalid_time": "Time must use 24-hour HH:MM format.",
-      "invalid_url": "Enter the API base URL as a full http:// or https:// URL with a host.",
-      "invalid_staleness": "Last-good snapshot staleness must be at least the update interval.",
       "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
       "invalid_admin": "Invalid admin selection."
     }

custom_components/nestquest/sweep.py

@@ -1,40 +1,70 @@
-"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim.
+"""The nightly missed-quest sweep (Feature 11).
 
-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.
+At the configured day-rollover time (the same HA-local listener that
+re-materializes the rolling horizon) this job finds every instance
+whose ``due_date`` is before the HA-local date, that has NO completion
+event, and that falls after the persisted sweep watermark — and fires
+ONE ``nestquest_quest_missed`` event per instance with the documented
+payload.  The sweep is read-only over the domain tables: it never
+mutates an instance and never writes a completion event (``missed``
+is derived state, never an event_type — Feature 08's contract).
 
-Idempotency is the 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).
+Idempotency is the watermark: after a run, ``nestquest_meta_state``
+records the HA-local date the sweep ran, and the next run only
+sweeps instances due in ``[watermark, today)``.  A second run the
+same night therefore fires nothing new, and a run after days of
+downtime sweeps exactly the accumulated window once — no
+per-instance journal, no re-announcing old misses.  An instance swept
+as missed that is later completed simply gains its completion event;
+history keeps both rows (append-only), and the sweep's later runs
+skip it via the no-completion-event rule.
 """
 from __future__ import annotations
 
+import asyncio
 import datetime
 import logging
 from zoneinfo import ZoneInfo
 
-from .core.sweep import (
-    SWEEP_WATERMARK_KEY,
-    run_missed_sweep as _run_core_missed_sweep,
-)
+from .const import EVENT_QUEST_MISSED
+from .dao_children import ChildrenDao
+from .dao_instances import QuestInstancesDao
+from .dao_meta import MetaStateDao
+from .dao_rules import QuestDefinitionsDao
 from .db import NestQuestDatabase
 
 LOGGER = logging.getLogger(__name__)
 
-__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"]
+#: meta_state key holding the HA-local date the sweep last ran.
+SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
+
+#: One asyncio.Lock per (connection wrapper, running loop), mirroring
+#: the migration runner's pattern: the whole watermark read → query →
+#: fire → write span is one critical section, so an overlapping startup
+#: sweep and rollover sweep queue instead of double-announcing.  The
+#: loser re-reads the watermark INSIDE the lock, sees the winner's
+#: date, and fires nothing.
+_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
+
+
+def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
+    """Return the sweep lock bound to ``database``'s running loop."""
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _SWEEP_LOCKS.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _SWEEP_LOCKS[key] = lock
+    return lock
+
+#: UTC ISO-8601 timestamp policy shared with the DAO layers.
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+def _now_stamp() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
 
 
 async def run_missed_sweep(
@@ -49,24 +79,54 @@ 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 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.
+    covers ``today`` (a same-night rerun).
+
+    The whole watermark read → query → fire → write span runs inside
+    the per-database sweep lock: an overlapping startup sweep and a
+    rollover sweep queue instead of interleaving, and the loser
+    re-reads the watermark inside the lock — sees the winner's date —
+    and announces nothing twice.
     """
     today_date = (
         today
         if today is not None
         else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
     )
-    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)
+    today_iso = today_date.isoformat()
+    dao = MetaStateDao(database)
+    async with _sweep_lock(database):
+        watermark = await dao.get(SWEEP_WATERMARK_KEY)
+        if watermark == today_iso:
+            LOGGER.debug(
+                "NestQuest missed sweep already ran for %s", today_iso
+            )
+            return 0
+
+        instances_dao = QuestInstancesDao(database)
+        rows = await instances_dao.list_missed_unswept(today_iso, watermark)
+        children_dao = ChildrenDao(database)
+        definitions_dao = QuestDefinitionsDao(database)
+        now_stamp = _now_stamp()
+        fired = 0
+        for instance in rows:
+            child = await children_dao.get(instance.child_id)
+            definition = await definitions_dao.get(instance.definition_id)
+            hass.bus.async_fire(
+                EVENT_QUEST_MISSED,
+                {
+                    "child_id": instance.child_id,
+                    "child_name": child.display_name if child else None,
+                    "instance_id": instance.id,
+                    "quest_title": definition.title if definition else None,
+                    "window": instance.window,
+                    "due_date": instance.due_date,
+                    "due_time": instance.due_time,
+                    "occurred_at": now_stamp,
+                },
+            )
+            fired += 1
+        await dao.set(SWEEP_WATERMARK_KEY, today_iso)
     LOGGER.info(
-        "NestQuest missed sweep for %s fired %d event(s)",
-        today_date.isoformat(),
-        fired,
+        "NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
     )
     return fired

custom_components/nestquest/translations/en.json

@@ -18,14 +18,11 @@
     "step": {
       "init": {
         "title": "NestQuest Settings",
-        "description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
+        "description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
         "data": {
           "horizon_days": "Generation horizon (days)",
           "day_rollover_time": "Day rollover time (HH:MM)",
           "panel_idle_timeout": "Panel idle timeout (seconds)",
-          "snapshot_staleness": "Last-good snapshot staleness (seconds)",
-          "api_base_url": "NestQuest API base URL (http://host:port)",
-          "panel_token": "Panel service token",
           "admin_user_ids": "Admin users (parents)"
         }
       }
@@ -33,8 +30,6 @@
     "errors": {
       "invalid": "Invalid value.",
       "invalid_time": "Time must use 24-hour HH:MM format.",
-      "invalid_url": "Enter the API base URL as a full http:// or https:// URL with a host.",
-      "invalid_staleness": "Last-good snapshot staleness must be at least the update interval.",
       "no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
       "invalid_admin": "Invalid admin selection."
     }

deploy/nestquest-controller.service

@@ -1,14 +0,0 @@
-[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

@@ -1,11 +0,0 @@
----
-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

@@ -1,56 +0,0 @@
-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

@@ -1,88 +0,0 @@
-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

@@ -1,108 +0,0 @@
-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

@@ -1,79 +0,0 @@
-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

@@ -1,17 +0,0 @@
-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

@@ -1,20 +0,0 @@
-#!/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

@@ -1,131 +0,0 @@
-#!/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

@@ -1,277 +0,0 @@
-#!/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

@@ -1,52 +0,0 @@
-#!/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

@@ -1,111 +0,0 @@
-#!/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

@@ -1,49 +0,0 @@
-#!/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

@@ -1,958 +0,0 @@
-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

@@ -1,32 +0,0 @@
-# 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

@@ -1,35 +0,0 @@
-{
-  "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/conftest.py

@@ -52,7 +52,6 @@ _HA_MODULES = (
     "homeassistant.helpers.entity",
     "homeassistant.helpers.event",
     "homeassistant.helpers.update_coordinator",
-    "homeassistant.helpers.aiohttp_client",
     "homeassistant.components",
     "homeassistant.components.sensor",
     "homeassistant.components.binary_sensor",
@@ -100,7 +99,6 @@ for _parent, _child in (
     ("homeassistant.helpers", "entity"),
     ("homeassistant.helpers", "event"),
     ("homeassistant.helpers", "update_coordinator"),
-    ("homeassistant.helpers", "aiohttp_client"),
     ("homeassistant", "components"),
     ("homeassistant.components", "sensor"),
     ("homeassistant.components", "binary_sensor"),
@@ -367,15 +365,6 @@ class BinarySensorEntity(Entity):
         return self._attr_device_class
 
 
-class UpdateFailed(HomeAssistantError):
-    """Stand-in mirroring homeassistant.helpers.update_coordinator.UpdateFailed.
-
-    The exception an update pass raises to report a failed fetch: the
-    coordinator records ``last_update_success = False`` instead of
-    propagating it out of the refresh.
-    """
-
-
 class DataUpdateCoordinator:
     """Functional stand-in for HA's DataUpdateCoordinator.
 
@@ -384,20 +373,8 @@ class DataUpdateCoordinator:
     ``async_add_listener``/``async_update_listeners`` for entity
     subscription, ``async_refresh``/``async_config_entry_first_refresh``
     driving the subclass's ``_async_update_data``, ``last_update_success``,
-    ``last_exception``, and ``data``.  Interval scheduling is NOT
-    simulated (no real clock ticks in tests) — tests drive refreshes
-    directly.
-
-    The failure contract mirrors real HA exactly (it is the trap this
-    harness once hid): an update failure NEVER propagates out of
-    ``async_refresh`` — the coordinator records
-    ``last_update_success = False``, keeps its last good ``data``, and
-    still notifies listeners so entities re-render unavailable — and
-    ``async_config_entry_first_refresh`` wraps ANY failed pass in
-    :class:`ConfigEntryNotReady`, with the original error only
-    surviving as ``__cause__``.  A stand-in that re-raised the raw
-    update error would let setup code key on the original exception
-    type — a clause real HA can never reach.
+    and ``data``.  Interval scheduling is NOT simulated (no real clock
+    ticks in tests) — tests drive refreshes directly.
     """
 
     def __init__(self, hass, logger, *, name=None, update_interval=None):
@@ -407,7 +384,6 @@ class DataUpdateCoordinator:
         self.update_interval = update_interval
         self.data = None
         self.last_update_success = False
-        self.last_exception = None
         self._listeners = []
 
     def async_add_listener(self, update_callback, context=None):
@@ -427,35 +403,18 @@ class DataUpdateCoordinator:
             callback()
 
     async def async_refresh(self):
-        """Run one update pass and notify listeners.
-
-        Mirrors HA: a failed update pass is RECORDED, never raised —
-        ``last_update_success`` goes False, the exception lands on
-        ``last_exception``, and listeners still fire; only cancellation
-        propagates.
-        """
+        """Run one update pass and notify listeners."""
         try:
             self.data = await self._async_update_data()
             self.last_update_success = True
-        except Exception as err:
+        except BaseException:
             self.last_update_success = False
-            self.last_exception = err
+            raise
         self.async_update_listeners()
 
     async def async_config_entry_first_refresh(self):
-        """First refresh at config-entry setup, per HA's real contract.
-
-        HA's real ``async_config_entry_first_refresh`` wraps ANY failed
-        update pass in ``ConfigEntryNotReady`` — the original error
-        survives only as ``__cause__`` — so setup code that must
-        tolerate a failed first refresh keys on ConfigEntryNotReady,
-        never on the original exception type.
-        """
+        """First refresh at config-entry setup (alias of refresh)."""
         await self.async_refresh()
-        if not self.last_update_success:
-            raise ConfigEntryNotReady(
-                f"error fetching {self.name} data: {self.last_exception}"
-            ) from self.last_exception
 
     async def async_shutdown(self):
         """Release listeners on teardown."""
@@ -465,7 +424,6 @@ class DataUpdateCoordinator:
 _update_coordinator_mock = _ha_mock("homeassistant.helpers.update_coordinator")
 _update_coordinator_mock.DataUpdateCoordinator = DataUpdateCoordinator
 _update_coordinator_mock.CoordinatorEntity = CoordinatorEntity
-_update_coordinator_mock.UpdateFailed = UpdateFailed
 
 _entity_mock = _ha_mock("homeassistant.helpers.entity")
 _entity_mock.DeviceInfo = DeviceInfo
@@ -742,246 +700,6 @@ def wire_entry_to_registry(entry, registry: ListenerRegistry):
     return entry
 
 
-class LazyLocalSnapshotClient:
-    """A test client that serves the snapshot built LOCALLY.
-
-    Feature 18 moved the coordinator's refresh onto the API service's
-    panel snapshot, but most integration tests still seed the local
-    database directly (the DB removal is a later task).  This stub
-    bridges the two: ``get_snapshot`` resolves the entry's runtime
-    record (its database and settings) and runs the SAME pure core
-    builder the API route runs
-    (:func:`~custom_components.nestquest.core.snapshot.build_snapshot`),
-    shaping the payload exactly like ``api/routes_panel.py`` —
-    instances via ``instance_payload(include_missed=False)`` — so the
-    coordinator reconstructs from a byte-identical payload without any
-    network or FastAPI machinery.
-
-    ``complete_instance`` bridges the panel complete route the same
-    way (the ``nestquest.complete_quest`` proxy calls it through the
-    API client): it resolves the entry's database and delegates to the
-    SAME core completion the API route delegates to, with the panel
-    actor shape, and answers the route's ``{"status": ...}`` body.
-    Like the real API service it publishes its transition events on
-    the SSE stream — NOT on the Home Assistant bus and NOT from the
-    service handler; tests that assert bus events drive them through
-    the SSE subscription the way production delivers them.
-
-    The resolution is LAZY on purpose: the coordinator's factory runs
-    mid-setup, before the runtime record exists, so the first refresh
-    (and any refresh while the entry is torn down) serves an EMPTY but
-    valid snapshot; every later refresh reads the seeded database.
-    """
-
-    def __init__(self, hass, entry_id: str) -> None:
-        self._hass = hass
-        self._entry_id = entry_id
-        self.calls = 0
-        #: Every ``(instance_id, CompletionResult)`` this stand-in
-        # produced, in call order — tests read ``appended`` /
-        # ``was_on_time`` off these to check the completion verdicts.
-        self.completions = []
-        #: The SSE frames the stand-in published for its completions
-        # (the API service's stream content), drained by
-        #: :func:`refire_api_transitions`.
-        self.published_frames = []
-
-    def _live_runtime(self):
-        """The entry's runtime record, or ``None`` before/after setup."""
-        from custom_components.nestquest.const import DOMAIN as NQ_DOMAIN
-
-        runtime = self._hass.data.get(NQ_DOMAIN, {}).get(self._entry_id)
-        database = getattr(runtime, "database", None)
-        if database is None or not getattr(database, "connected", True):
-            return None
-        return runtime
-
-    def _live_database(self):
-        """The entry's open database, or ``None`` before/after setup."""
-        runtime = self._live_runtime()
-        return None if runtime is None else runtime.database
-
-    async def get_snapshot(self):
-        import datetime
-        import zoneinfo
-
-        from custom_components.nestquest.const import DOMAIN as NQ_DOMAIN
-        from custom_components.nestquest.core import snapshot as core_snapshot
-
-        self.calls += 1
-        runtime = self._live_runtime()
-        if runtime is None:
-            return {
-                "today_iso": datetime.date.today().isoformat(),
-                "cycle_day": 0,
-                "children": [],
-            }
-        database = runtime.database
-        settings = getattr(runtime, "settings", None)
-        # The HA-LOCAL now, the same single clock read the production
-        # coordinator used to hand the builder: the hass stub's
-        # configured time zone, never the host clock.
-        now = datetime.datetime.now(
-            zoneinfo.ZoneInfo(self._hass.config.time_zone)
-        )
-        snapshot = await core_snapshot.build_snapshot(database, settings, now)
-        return {
-            "today_iso": snapshot.today_iso,
-            "cycle_day": snapshot.cycle_day,
-            "children": [
-                {
-                    "child_id": child.child_id,
-                    "child_name": child.child_name,
-                    "present": child.present,
-                    "next_present": child.next_present,
-                    "due_today": child.due_today,
-                    "completed_today": child.completed_today,
-                    "remaining_today": child.remaining_today,
-                    "completion_pct": child.completion_pct,
-                    "instances": core_snapshot.instance_payload(
-                        child.instances, include_missed=False
-                    ),
-                }
-                for child in snapshot.children
-            ],
-        }
-
-    async def complete_instance(self, instance_id: int, actor_child_id: int):
-        """The panel complete route's stand-in: complete through the
-        SAME core completion the route delegates to, panel actor
-        shape, publish the route's SSE transition frames (built by the
-        same core builders, appended completions only — see
-        ``api/routes_panel.py``), and answer the route's
-        ``{"status": ...}`` body."""
-        import datetime
-
-        from custom_components.nestquest.completion import complete_instance
-        from custom_components.nestquest.core import events as core_events
-
-        result = await complete_instance(
-            self._live_database(),
-            instance_id,
-            actor_source="panel",
-            actor_child_id=actor_child_id,
-        )
-        self.completions.append((instance_id, result))
-        if result.appended:
-            # The route publishes the transition the moment the core
-            # write lands (appended completions only), evaluating the
-            # day-complete rule against the API host's local date —
-            # the frames queue here for the SSE stream to carry.
-            event_type, payload = await core_events.build_quest_completed_event(
-                self._live_database(),
-                instance_id,
-                was_on_time=result.was_on_time,
-            )
-            self.published_frames.append((event_type, payload))
-            now = datetime.datetime.now().astimezone()
-            day_complete = await core_events.build_child_day_complete_event(
-                self._live_database(), payload, today=now.date()
-            )
-            if day_complete is not None:
-                self.published_frames.append(day_complete)
-        return {"status": str(result)}
-
-
-@pytest.fixture(autouse=True)
-def coordinator_serves_local_snapshot():
-    """Default every test entry's coordinator to the local-snapshot client.
-
-    The integration's setup builds its API client through
-    ``coordinator_client_from_entry``; the fixture swaps that factory
-    for a test one handing out :class:`LazyLocalSnapshotClient` (so
-    DB-seeded tests exercise the coordinator's API path against local
-    data).  Tests that script their own client register it with
-    :func:`set_coordinator_client`, which the factory honours first.
-    """
-    import custom_components.nestquest as nq_mod
-    import custom_components.nestquest.coordinator as coordinator_module
-
-    _COORDINATOR_CLIENT_OVERRIDES = (
-        coordinator_module._COORDINATOR_CLIENT_OVERRIDES
-    )
-
-    def _factory(hass, entry, **kwargs):
-        override = _COORDINATOR_CLIENT_OVERRIDES.get(entry.entry_id)
-        if override is not None:
-            return override
-        return LazyLocalSnapshotClient(hass, entry.entry_id)
-
-    import unittest.mock as mock
-
-    _COORDINATOR_CLIENT_OVERRIDES.clear()
-    patcher = mock.patch.object(
-        nq_mod, "coordinator_client_from_entry", _factory
-    )
-    patcher.start()
-    yield
-    patcher.stop()
-    _COORDINATOR_CLIENT_OVERRIDES.clear()
-
-
-def set_coordinator_client(entry, client):
-    """Register a scripted client for ``entry``'s coordinator factory."""
-    import custom_components.nestquest.coordinator as coordinator_module
-
-    coordinator_module._COORDINATOR_CLIENT_OVERRIDES[entry.entry_id] = client
-    return client
-
-
-async def refire_api_transitions(hass, entry, client):
-    """Deliver the SSE frames the stand-in client has published so far
-    through the integration's SSE subscription — the production event
-    path.
-
-    Since Feature 18 the ``complete_quest`` service fires no local
-    bus event: the API service builds and publishes the transition
-    frames on its SSE stream, and the integration's subscription
-    (:mod:`custom_components.nestquest.sse`) re-fires each frame on
-    the bus.  This helper streams the frames the stand-in client
-    published (``client.published_frames``) through a real
-    :class:`~custom_components.nestquest.sse.NestQuestEventStream`
-    and waits (bounded) until the bus has re-fired every one of them.
-    """
-    import asyncio
-
-    from custom_components.nestquest.sse import NestQuestEventStream
-
-    frames, client.published_frames = client.published_frames, []
-
-    class _ScriptedStream:
-        """A healthy stream: the frames, then it stays open."""
-
-        def __init__(self, frames) -> None:
-            self._frames = frames
-
-        def stream_events(self):
-            async def _gen():
-                for frame in self._frames:
-                    yield frame
-                await asyncio.Event().wait()
-
-            return _gen()
-
-    before = {etype: len(hass.bus.fired(etype)) for etype, _ in frames}
-    expected = len(frames)
-    manager = NestQuestEventStream(hass, _ScriptedStream(frames))
-    manager.start()
-    try:
-        async with asyncio.timeout(2):
-            while True:
-                delivered = sum(
-                    len(hass.bus.fired(etype)) - before[etype]
-                    for etype, _ in frames
-                )
-                if delivered >= expected:
-                    break
-                await asyncio.sleep(0.01)
-    finally:
-        await manager.stop()
-    return frames
-
-
 class _HassNamespace(SimpleNamespace):
     """SimpleNamespace subclass so the hass stub can hold weak references."""
 

tests/test_api_client.py

@@ -1,450 +0,0 @@
-"""Tests for the panel-plane API client (Feature 18).
-
-The client is exercised against a STUB transport only — no network, no
-aiohttp, no homeassistant: the injected session is a fake that speaks
-the same ``request(...)`` surface aiohttp's ``ClientSession`` exposes
-and records every call, so the tests assert exactly what the client
-puts on the wire (method, URL, Bearer header, JSON body) and how every
-failure mode (non-2xx, connection failure, timeout, malformed JSON)
-surfaces as the typed :class:`NestQuestApiError`.
-"""
-from __future__ import annotations
-
-import asyncio
-import json
-import sys
-import types
-
-import pytest
-
-from conftest import make_config_entry
-
-from custom_components.nestquest import api_client
-from custom_components.nestquest.api_client import (
-    NestQuestApiClient,
-    NestQuestApiError,
-    is_valid_base_url,
-    resolve_api_config,
-)
-from custom_components.nestquest.const import (
-    CONF_API_BASE_URL,
-    CONF_PANEL_TOKEN,
-    DEFAULT_API_BASE_URL,
-    DEFAULT_PANEL_TOKEN,
-)
-
-BASE_URL = "http://api.test:8000"
-TOKEN = "test-panel-token"
-
-#: A realistic panel snapshot payload (the documented shape of the
-#: Feature 16 snapshot route).
-SNAPSHOT_PAYLOAD = {
-    "today_iso": "2026-09-23",
-    "cycle_day": 2,
-    "children": [
-        {
-            "child_id": 1,
-            "child_name": "Ada",
-            "present": False,
-            "next_present": "2026-09-24",
-            "due_today": 1,
-            "completed_today": 0,
-            "remaining_today": 1,
-            "completion_pct": 0,
-            "instances": [
-                {
-                    "id": 7,
-                    "definition_id": 3,
-                    "child_id": 1,
-                    "title": "Pack bag",
-                    "icon": None,
-                    "window": "morning",
-                    "due_time": "10:00",
-                    "state": "open",
-                    "overdue": True,
-                    "completed_at": None,
-                    "on_time": None,
-                }
-            ],
-        }
-    ],
-}
-
-
-class StubResponse:
-    """A fake aiohttp response: ``status`` plus ``await text()``."""
-
-    def __init__(self, status: int, body: str) -> None:
-        self.status = status
-        self._body = body
-
-    async def text(self) -> str:
-        return self._body
-
-
-class _OutcomeContext:
-    """Async context manager yielding one scripted outcome.
-
-    Mirrors aiohttp's ``session.request(...)`` return value: entering
-    the context performs the request, so a scripted transport error
-    raises THERE — before any response object exists, exactly where
-    aiohttp's connector raises on a refused connection.
-    """
-
-    def __init__(self, response, error) -> None:
-        self._response = response
-        self._error = error
-
-    async def __aenter__(self):
-        if self._error is not None:
-            raise self._error
-        return self._response
-
-    async def __aexit__(self, exc_type, exc, tb) -> bool:
-        return False
-
-
-class StubTransport:
-    """A no-network stand-in for aiohttp's ``ClientSession``.
-
-    Records every ``request(...)`` call as a dict (method, url, headers,
-    json body) and answers from a scripted queue of outcomes — a
-    :class:`StubResponse` or an exception per call, consumed in order —
-    so each test drives exactly one exchange.
-    """
-
-    def __init__(self, *outcomes) -> None:
-        self.calls: list[dict] = []
-        self._outcomes = list(outcomes)
-
-    def request(self, method: str, url: str, **kwargs):
-        self.calls.append(
-            {
-                "method": method,
-                "url": url,
-                "headers": kwargs.get("headers"),
-                "json": kwargs.get("json"),
-            }
-        )
-        if not self._outcomes:
-            raise AssertionError("StubTransport ran out of scripted outcomes")
-        outcome = self._outcomes.pop(0)
-        if isinstance(outcome, BaseException):
-            return _OutcomeContext(None, outcome)
-        return _OutcomeContext(outcome, None)
-
-
-class HangingTransport:
-    """A transport whose request never answers (for the timeout test)."""
-
-    def __init__(self) -> None:
-        self.calls: list[dict] = []
-
-    def request(self, method: str, url: str, **kwargs):
-        self.calls.append({"method": method, "url": url})
-        return _HangingContext()
-
-
-class _HangingContext:
-    async def __aenter__(self):
-        await asyncio.sleep(5)
-        raise AssertionError("hanging transport was never cancelled")
-
-    async def __aexit__(self, exc_type, exc, tb) -> bool:
-        return False
-
-
-def make_client(transport, *, timeout_seconds: float = 5.0) -> NestQuestApiClient:
-    """Build a client over the stub transport with the test credentials."""
-    return NestQuestApiClient(
-        BASE_URL, TOKEN, transport, timeout_seconds=timeout_seconds
-    )
-
-
-# ---------------------------------------------------------------------------
-# Success paths
-# ---------------------------------------------------------------------------
-
-
-async def test_get_snapshot_parses_payload_and_sends_bearer() -> None:
-    transport = StubTransport(
-        StubResponse(200, json.dumps(SNAPSHOT_PAYLOAD))
-    )
-    client = make_client(transport)
-
-    snapshot = await client.get_snapshot()
-
-    assert snapshot == SNAPSHOT_PAYLOAD
-    call = transport.calls[0]
-    assert call["method"] == "GET"
-    assert call["url"] == f"{BASE_URL}/api/v1/panel/snapshot"
-    assert call["headers"] == {"Authorization": f"Bearer {TOKEN}"}
-    assert call["json"] is None
-
-
-async def test_complete_instance_posts_body_and_headers() -> None:
-    transport = StubTransport(StubResponse(200, '{"status": "done"}'))
-    client = make_client(transport)
-
-    result = await client.complete_instance(42, 3)
-
-    assert result == {"status": "done"}
-    call = transport.calls[0]
-    assert call["method"] == "POST"
-    assert call["url"] == f"{BASE_URL}/api/v1/panel/instances/42/complete"
-    assert call["json"] == {"actor_child_id": 3}
-    assert call["headers"]["Authorization"] == f"Bearer {TOKEN}"
-
-
-async def test_base_url_trailing_slash_is_normalized() -> None:
-    """A base URL with a trailing slash must not double the path slash."""
-    transport = StubTransport(StubResponse(200, "{}"))
-    client = NestQuestApiClient(
-        f"{BASE_URL}/", TOKEN, transport, timeout_seconds=5.0
-    )
-    await client.get_snapshot()
-    assert transport.calls[0]["url"] == f"{BASE_URL}/api/v1/panel/snapshot"
-
-
-# ---------------------------------------------------------------------------
-# HTTP failures: the typed error carries the status
-# ---------------------------------------------------------------------------
-
-
-async def test_snapshot_401_raises_typed_error() -> None:
-    """A wrong/missing token answers 401 and surfaces with that status."""
-    transport = StubTransport(
-        StubResponse(401, '{"detail": "Missing or invalid panel service token"}')
-    )
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.get_snapshot()
-
-    error = excinfo.value
-    assert error.status == 401
-    assert not error.is_transport_error
-    assert error.__cause__ is None
-    # The token is a secret: it never reaches the raised message.
-    assert TOKEN not in str(error)
-
-
-async def test_complete_unknown_instance_404_raises_typed_error() -> None:
-    transport = StubTransport(
-        StubResponse(404, '{"detail": "Quest instance not found"}')
-    )
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.complete_instance(999999, 1)
-
-    assert excinfo.value.status == 404
-    assert not excinfo.value.is_transport_error
-
-
-async def test_snapshot_500_raises_typed_error() -> None:
-    """Any non-2xx response is typed, whatever the status."""
-    transport = StubTransport(StubResponse(500, "{}"))
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.get_snapshot()
-
-    assert excinfo.value.status == 500
-
-
-async def test_malformed_2xx_body_raises_typed_with_status() -> None:
-    """A 2xx body that is not JSON is a protocol failure, carrying its status."""
-    transport = StubTransport(StubResponse(200, "<html>not json</html>"))
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.get_snapshot()
-
-    assert excinfo.value.status == 200
-
-
-# ---------------------------------------------------------------------------
-# Transport failures: the typed error carries no status and chains the cause
-# ---------------------------------------------------------------------------
-
-
-async def test_connection_failure_raises_transport_error() -> None:
-    transport = StubTransport(OSError("connection refused"))
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.get_snapshot()
-
-    error = excinfo.value
-    assert error.status is None
-    assert error.is_transport_error
-    assert isinstance(error.__cause__, OSError)
-
-
-async def test_connection_failure_on_complete_is_typed() -> None:
-    transport = StubTransport(OSError("connection refused"))
-    client = make_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.complete_instance(7, 1)
-
-    assert excinfo.value.status is None
-    assert excinfo.value.is_transport_error
-
-
-async def test_request_timeout_raises_transport_error() -> None:
-    """A hung API call surfaces as a typed transport error, once, with
-    no retry: the one request is cancelled by the client's timeout."""
-    transport = HangingTransport()
-    client = make_client(transport, timeout_seconds=0.05)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        await client.get_snapshot()
-
-    assert excinfo.value.status is None
-    assert excinfo.value.is_transport_error
-    assert len(transport.calls) == 1
-
-
-# ---------------------------------------------------------------------------
-# Construction and config resolution
-# ---------------------------------------------------------------------------
-
-
-def test_constructor_requires_full_base_url() -> None:
-    for bad_url in (None, "", "   ", "api.lan:8000", "http://", "ftp://x"):
-        with pytest.raises(ValueError):
-            NestQuestApiClient(bad_url, TOKEN, StubTransport())
-
-
-def test_constructor_requires_non_empty_token() -> None:
-    for bad_token in (None, "", "   "):
-        with pytest.raises(ValueError):
-            NestQuestApiClient(BASE_URL, bad_token, StubTransport())
-
-
-def test_constructor_requires_positive_timeout() -> None:
-    for bad_timeout in (0, -1, "5", True, None):
-        with pytest.raises(ValueError):
-            NestQuestApiClient(
-                BASE_URL, TOKEN, StubTransport(), timeout_seconds=bad_timeout
-            )
-
-
-def test_is_valid_base_url_accepts_full_http_urls() -> None:
-    assert is_valid_base_url("http://127.0.0.1:8000")
-    assert is_valid_base_url("https://nestquest.lan")
-    assert is_valid_base_url(" http://127.0.0.1:8000 ")
-    assert not is_valid_base_url("api.lan:8000")
-    assert not is_valid_base_url("http://")
-    assert not is_valid_base_url("")
-    assert not is_valid_base_url(None)
-    assert not is_valid_base_url(42)
-
-
-def test_resolve_api_config_prefers_options_then_data_then_defaults() -> None:
-    options_only = make_config_entry(
-        options={CONF_API_BASE_URL: "http://from-options:1"}
-    )
-    assert resolve_api_config(options_only) == (
-        "http://from-options:1",
-        DEFAULT_PANEL_TOKEN,
-    )
-
-    data_only = make_config_entry(
-        data={
-            CONF_API_BASE_URL: "http://from-data:2",
-            CONF_PANEL_TOKEN: "from-data",
-        }
-    )
-    assert resolve_api_config(data_only) == ("http://from-data:2", "from-data")
-
-    both = make_config_entry(
-        options={CONF_API_BASE_URL: "http://from-options:3"},
-        data={
-            CONF_API_BASE_URL: "http://from-data:4",
-            CONF_PANEL_TOKEN: "from-data",
-        },
-    )
-    assert resolve_api_config(both) == (
-        "http://from-options:3",
-        "from-data",
-    )
-
-    bare = make_config_entry()
-    assert resolve_api_config(bare) == (DEFAULT_API_BASE_URL, DEFAULT_PANEL_TOKEN)
-
-
-def test_client_from_entry_reads_config_and_shared_session(monkeypatch) -> None:
-    """The production factory resolves the entry config and hands in HA's
-    shared aiohttp session (stubbed module — no real HA, no network)."""
-    fake_module = types.ModuleType("homeassistant.helpers.aiohttp_client")
-    shared_session = object()
-    fake_module.async_get_clientsession = lambda hass: shared_session
-    monkeypatch.setitem(
-        sys.modules, "homeassistant.helpers.aiohttp_client", fake_module
-    )
-    entry = make_config_entry(
-        options={
-            CONF_API_BASE_URL: "http://panel.lan:8443",
-            CONF_PANEL_TOKEN: "entry-token",
-        }
-    )
-
-    client = api_client.client_from_entry(object(), entry)
-
-    assert client._session is shared_session
-    assert client._base_url == "http://panel.lan:8443"
-    assert client._panel_token == "entry-token"
-
-
-def test_client_from_entry_unconfigured_token_raises_at_construction(
-    monkeypatch,
-) -> None:
-    """An entry with NO configured token (the shipped default) raises
-    ValueError AT CONSTRUCTION — the client's fail-fast guard — rather
-    than silently building a client that can only 401 (the documented
-    contract: the options flow's non-empty validation is the user-facing
-    guard; the coordinator must ensure a token is configured first or
-    handle the construction error as "API not configured")."""
-    fake_module = types.ModuleType("homeassistant.helpers.aiohttp_client")
-    fake_module.async_get_clientsession = lambda hass: object()
-    monkeypatch.setitem(
-        sys.modules, "homeassistant.helpers.aiohttp_client", fake_module
-    )
-    entry = make_config_entry(
-        options={CONF_API_BASE_URL: DEFAULT_API_BASE_URL},
-    )
-    assert resolve_api_config(entry)[1] == DEFAULT_PANEL_TOKEN == ""
-
-    with pytest.raises(ValueError, match="panel_token must be a non-empty"):
-        api_client.client_from_entry(object(), entry)
-
-    # Same for a whitespace-only stored token: whitespace is not a token.
-    entry_ws = make_config_entry(options={CONF_PANEL_TOKEN: "   "})
-    with pytest.raises(ValueError, match="panel_token must be a non-empty"):
-        api_client.client_from_entry(object(), entry_ws)
-
-
-async def test_client_logs_nothing() -> None:
-    """The client emits no log records at all (the token must never
-    reach a log line), success or failure."""
-    import logging
-
-    records: list[logging.LogRecord] = []
-
-    class _Capture(logging.Handler):
-        def emit(self, record: logging.LogRecord) -> None:
-            records.append(record)
-
-    handler = _Capture()
-    logging.getLogger().addHandler(handler)
-    try:
-        transport = StubTransport(StubResponse(401, "{}"))
-        client = make_client(transport)
-        with pytest.raises(NestQuestApiError):
-            await client.get_snapshot()
-    finally:
-        logging.getLogger().removeHandler(handler)
-    assert records == []

tests/test_api_missed_sweep.py

@@ -1,671 +0,0 @@
-"""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()

tests/test_complete_quest_proxy.py

@@ -1,283 +0,0 @@
-"""The ``nestquest.complete_quest`` proxy to the API client (Feature 18).
-
-The service is a thin proxy: a panel tap must reach the API's panel
-complete route through the entry's ONE shared API client (the same
-instance the coordinator polls and the SSE subscription reads), with
-exactly the payload contract the panel cards send (``instance_id``,
-``actor=panel``, ``actor_child_id``).  The service performs NO local
-database completion and fires NO local bus event — the API service
-emits the transition on its SSE stream and the integration's
-subscription re-fires it (firing it locally too would double-fire) —
-and it refreshes the coordinator after a successful call so the
-sensors reflect the completion immediately.  Every failure (a typed
-:class:`NestQuestApiError`, an unconfigured client, a non-panel
-actor) surfaces as a :class:`HomeAssistantError`, token-free.
-
-The scripted clients here stand in for the API service: recording and
-failing ``complete_instance`` implementations, no network, no aiohttp.
-"""
-from __future__ import annotations
-
-import datetime
-
-import pytest
-from homeassistant.exceptions import HomeAssistantError
-
-from conftest import (
-    LazyLocalSnapshotClient,
-    set_coordinator_client,
-    wire_entry_to_registry,
-)
-
-from custom_components.nestquest import async_setup_entry
-from custom_components.nestquest.api_client import NestQuestApiError
-from custom_components.nestquest.const import (
-    CONF_ADMIN_USER_IDS,
-    CONF_PANEL_TOKEN,
-    DOMAIN,
-    EVENT_CHILD_DAY_COMPLETE,
-    EVENT_QUEST_COMPLETED,
-    SERVICE_COMPLETE_QUEST,
-)
-from custom_components.nestquest.dao_instances import QuestInstancesDao
-from custom_components.nestquest.materialize import materialize
-from custom_components.nestquest.quest_definitions import (
-    create_quest_definition,
-)
-from custom_components.nestquest.recurrence import ScheduleRule
-
-ADMIN_CTX = {"user_id": "admin-1"}
-
-
-class _RecordingApiClient(LazyLocalSnapshotClient):
-    """Snapshot-only stand-in that RECORDS completion calls.
-
-    ``complete_instance`` deliberately performs NO database write: the
-    API service is remote, and the assertions below must be able to
-    prove the service handler itself completed nothing locally.
-    """
-
-    def __init__(self, hass, entry_id: str) -> None:
-        super().__init__(hass, entry_id)
-        self.complete_calls: list[tuple[int, int]] = []
-
-    async def complete_instance(
-        self, instance_id: int, actor_child_id: int
-    ) -> dict:
-        self.complete_calls.append((instance_id, actor_child_id))
-        return {"status": "done"}
-
-
-class _FailingApiClient(_RecordingApiClient):
-    """A stand-in whose completion fails the way the client types it."""
-
-    def __init__(self, hass, entry_id: str, error: NestQuestApiError) -> None:
-        super().__init__(hass, entry_id)
-        self._error = error
-
-    async def complete_instance(
-        self, instance_id: int, actor_child_id: int
-    ) -> dict:
-        self.complete_calls.append((instance_id, actor_child_id))
-        raise self._error
-
-
-async def _setup_seeded_entry(hass, make_entry, client):
-    """Wire, script, and set up an entry with one child's daily quest."""
-    entry = wire_entry_to_registry(
-        make_entry(
-            data={
-                CONF_ADMIN_USER_IDS: [ADMIN_CTX["user_id"]],
-                CONF_PANEL_TOKEN: "secret-panel-token",
-            }
-        ),
-        hass.registry,
-    )
-    set_coordinator_client(entry, client)
-    assert await async_setup_entry(hass, entry) is True
-    database = entry.runtime_data.database
-    from custom_components.nestquest.children import create_child
-
-    child = await create_child(database, "Ada")
-    today = datetime.date.today()
-    await create_quest_definition(
-        database,
-        "Brush teeth",
-        ScheduleRule.from_dict(
-            {"rule_type": "daily", "start_date": today.isoformat()}
-        ),
-        [child.id],
-        ["morning"],
-    )
-    end = (today + datetime.timedelta(days=3)).isoformat()
-    await materialize(database, today.isoformat(), end, today=today)
-    instances = await QuestInstancesDao(database).list_by_date_range(
-        child.id, today.isoformat(), today.isoformat()
-    )
-    assert instances
-    return entry, client, child, instances[0]
-
-
-def _record_refreshes(entry) -> list[bool]:
-    """Replace the coordinator's refresh with a recording stand-in."""
-    refreshes: list[bool] = []
-
-    async def _fake_refresh():
-        refreshes.append(True)
-
-    entry.runtime_data.coordinator.async_refresh = _fake_refresh
-    return refreshes
-
-
-async def test_complete_quest_reaches_the_shared_api_client(
-    hass, make_entry
-) -> None:
-    """A panel tap calls ``complete_instance`` on the entry's shared
-    client with the tapped child, completes NOTHING locally, fires NO
-    local event, and refreshes the coordinator immediately."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _RecordingApiClient(hass, "test_entry")
-    )
-    assert entry.runtime_data.coordinator.api_client is client, (
-        "the service must proxy through the entry's ONE shared client"
-    )
-    refreshes = _record_refreshes(entry)
-
-    await hass.services.call(
-        DOMAIN,
-        SERVICE_COMPLETE_QUEST,
-        {
-            "instance_id": instance.id,
-            "actor": "panel",
-            "actor_child_id": child.id,
-        },
-    )
-
-    assert client.complete_calls == [(instance.id, child.id)]
-    # No local completion: the instance is still open in the local
-    # database — only the API service writes (through its own client).
-    from custom_components.nestquest.completion import instance_state
-
-    assert (
-        await instance_state(entry.runtime_data.database, instance.id)
-        == "open"
-    )
-    # No local firing: the completed (and day-complete) events arrive
-    # over the SSE stream, re-fired by the subscription — never from
-    # this handler.
-    assert hass.bus.fired(EVENT_QUEST_COMPLETED) == []
-    assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
-    # The old immediate-update behaviour: the sensors' coordinator
-    # refresh runs before the service call returns.
-    assert refreshes == [True]
-
-
-@pytest.mark.parametrize(
-    ("error", "fragment"),
-    [
-        (
-            NestQuestApiError(
-                "NestQuest API POST /api/v1/panel/instances/7/complete "
-                "failed with HTTP 401",
-                status=401,
-            ),
-            "failed with HTTP 401",
-        ),
-        (
-            NestQuestApiError(
-                "NestQuest API POST /api/v1/panel/instances/7/complete "
-                "transport failure: connection refused",
-                status=None,
-            ),
-            "transport failure",
-        ),
-    ],
-)
-async def test_api_failure_surfaces_as_a_service_error(
-    hass, make_entry, error, fragment
-) -> None:
-    """A typed API failure (non-2xx or transport) raises
-    ``HomeAssistantError`` carrying the typed message — which never
-    includes the panel token — with nothing completed or fired
-    locally, and no refresh for a mutation that never happened."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass,
-        make_entry,
-        _FailingApiClient(hass, "test_entry", error),
-    )
-    refreshes = _record_refreshes(entry)
-
-    with pytest.raises(HomeAssistantError, match=fragment) as excinfo:
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "panel",
-                "actor_child_id": child.id,
-            },
-        )
-    # The typed error is chained, and the service error carries the
-    # typed message only — never the panel token.
-    assert excinfo.value.__cause__ is error
-    assert "secret-panel-token" not in str(excinfo.value)
-    assert client.complete_calls == [(instance.id, child.id)]
-    from custom_components.nestquest.completion import instance_state
-
-    assert (
-        await instance_state(entry.runtime_data.database, instance.id)
-        == "open"
-    )
-    assert hass.bus.fired(EVENT_QUEST_COMPLETED) == []
-    assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
-    assert refreshes == []
-
-
-async def test_unconfigured_api_client_surfaces_as_a_service_error(
-    hass, make_entry
-) -> None:
-    """An entry with no completion surface (the unconfigured-token
-    stub the coordinator swaps in) fails the service clearly instead
-    of reaching an attribute error."""
-    from custom_components.nestquest.coordinator import _UnconfiguredApiClient
-
-    entry, _client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _UnconfiguredApiClient()
-    )
-
-    with pytest.raises(
-        HomeAssistantError, match="NestQuest API is not configured"
-    ):
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "panel",
-                "actor_child_id": child.id,
-            },
-        )
-
-
-async def test_non_panel_actor_is_rejected_before_the_client(
-    hass, make_entry
-) -> None:
-    """The API's panel route completes as the tapped child only: a
-    user-actor call fails fast with a clear error and never reaches
-    the client."""
-    entry, client, child, instance = await _setup_seeded_entry(
-        hass, make_entry, _RecordingApiClient(hass, "test_entry")
-    )
-
-    with pytest.raises(HomeAssistantError, match="actor must be 'panel'"):
-        await hass.services.call(
-            DOMAIN,
-            SERVICE_COMPLETE_QUEST,
-            {
-                "instance_id": instance.id,
-                "actor": "user",
-                "actor_user_id": ADMIN_CTX["user_id"],
-            },
-            context=ADMIN_CTX,
-        )
-    assert client.complete_calls == []

tests/test_core_snapshot.py

@@ -123,11 +123,9 @@ async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) ->
     database = entry.runtime_data.database
     settings = entry.runtime_data.settings
 
-    # Single clock read for the seed/snapshot anchor (no midnight skew);
-    # the coordinator no longer owns a clock read (the API serves the
-    # snapshot), so the test resolves the HA-local zone directly.
+    # Single clock read for the seed/snapshot anchor (no midnight skew).
+    today = coordinator._local_now().date()
     time_zone = ZoneInfo(hass.config.time_zone)
-    today = datetime.datetime.now(time_zone).date()
     # Pin the build instant at 12:00 local: past Ada's 10:00 due time
     # (overdue) and before Bo's 17:00, and the completion moment for Bo.
     pinned_now = datetime.datetime(
@@ -296,38 +294,13 @@ async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) ->
         == "missed"
     )
 
-    # REQUIRED wrapper tie-in: since Feature 18 the coordinator's
-    # refresh polls the snapshot through its API client, so pinning the
-    # client to serve the payload of the SAME snapshot built with
-    # pinned_now must reproduce the builder's snapshot byte-for-byte.
-    # The dataclasses are frozen, so == is a full deep compare — a
-    # reconstruction that dropped or reshaped a field would diverge
-    # here even though the literal assertions above still pass.
-    from conftest import set_coordinator_client
-
-    class _PinnedPayloadClient:
-        async def get_snapshot(self):
-            return {
-                "today_iso": built.today_iso,
-                "cycle_day": built.cycle_day,
-                "children": [
-                    {
-                        "child_id": child.child_id,
-                        "child_name": child.child_name,
-                        "present": child.present,
-                        "next_present": child.next_present,
-                        "due_today": child.due_today,
-                        "completed_today": child.completed_today,
-                        "remaining_today": child.remaining_today,
-                        "completion_pct": child.completion_pct,
-                        "instances": instance_payload(
-                            child.instances, include_missed=False
-                        ),
-                    }
-                    for child in built.children
-                ],
-            }
-
-    set_coordinator_client(entry, _PinnedPayloadClient())
+    # REQUIRED wrapper tie-in: the coordinator's _async_update_data is a
+    # thin WRAPPER over build_snapshot, so pinning its clock to the SAME
+    # pinned_now the builder was called with must reproduce the builder's
+    # snapshot byte-for-byte.  The dataclasses are frozen, so == is a full
+    # deep compare — a wrapper that mis-passed today vs now, or stopped
+    # calling the builder, would diverge here even though the literal
+    # assertions above still pass.
+    monkeypatch.setattr(coordinator, "_local_now", lambda: pinned_now)
     await coordinator.async_refresh()
     assert coordinator.data == built

tests/test_entity_matrix.py

@@ -14,7 +14,7 @@ from __future__ import annotations
 import datetime
 import re
 
-from conftest import refire_api_transitions, wire_entry_to_registry
+from conftest import wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry
 from custom_components.nestquest.children import (
@@ -154,11 +154,6 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
                 "actor_child_id": ada.id,
             },
         )
-    # The transition events arrive over the API's SSE stream (the
-    # subscription re-fires them), never from the service handler.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 2
     for payload in hass.bus.fired(EVENT_QUEST_COMPLETED):
         assert _TIMESTAMP.match(payload["occurred_at"])
@@ -185,12 +180,9 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
     assert _sensor(hass, ada.id, "all_done").is_on is False
     assert _sensor(hass, ada.id, "quests_remaining_today").native_value == 1
 
-    # --- The instances payload omits missed instances (D-009) — and
-    # since Feature 18 the snapshot comes from the API panel route,
-    # which omits missed rows ENTIRELY, so admin_instances cannot
-    # carry them either (the admin surface for missed quests is the
-    # PWA, reading the API directly): pin the derive clock forward in
-    # the TEST ONLY so today's open instances read missed.
+    # --- The instances payload omits missed instances while
+    # admin_instances includes them (D-009): pin the derive clock
+    # forward in the TEST ONLY so today's open instances read missed.
     original_refresh = coordinator._async_update_data
 
     async def _future_dated_refresh():
@@ -211,9 +203,9 @@ async def test_entity_and_event_matrix(hass, make_entry) -> None:
     await coordinator.async_refresh()
     attributes = _sensor(hass, bo.id, "quests_due_today").extra_state_attributes
     assert attributes["instances"] == [], "panel payload omits missed"
-    assert attributes["admin_instances"] == [], (
-        "the API payload omits missed rows, so the admin payload "
-        "carries none either (the PWA is the admin surface now)"
+    assert len(attributes["admin_instances"]) == 2
+    assert all(
+        entry["state"] == "missed" for entry in attributes["admin_instances"]
     )
     coordinator._async_update_data = original_refresh
     await coordinator.async_refresh()
@@ -239,10 +231,6 @@ async def test_payload_schema_matches_the_documented_contract(
             "actor_child_id": ada.id,
         },
     )
-    # The completed event arrives over the API's SSE stream.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     payload = hass.bus.fired(EVENT_QUEST_COMPLETED)[0]
     for key, check in (
         ("child_id", lambda v: isinstance(v, int)),

tests/test_events.py

@@ -1,21 +1,9 @@
-"""Tests for the NestQuest bus events (Feature 10, task 6).
-
-Since Feature 18 the integration is a client of the NestQuest API
-service: ``nestquest.complete_quest`` proxies to the API's panel
-complete route and fires NO local bus event — the API service builds
-and publishes the transition frames on its SSE stream, and the
-integration's subscription (:mod:`custom_components.nestquest.sse`)
-re-fires them.  The service-driven tests below therefore deliver the
-frames through :func:`conftest.refire_api_transitions` (the
-production event path, driven synchronously) before asserting the
-bus; the shim-level tests drive :mod:`custom_components.nestquest
-.events` directly, the way its remaining callers do.
-"""
+"""Tests for the NestQuest bus events (Feature 10, task 6)."""
 from __future__ import annotations
 
 import re
 
-from conftest import refire_api_transitions, wire_entry_to_registry
+from conftest import wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry
 from custom_components.nestquest.children import create_child
@@ -98,11 +86,6 @@ async def test_quest_completed_fires_with_documented_payload(
         SERVICE_COMPLETE_QUEST,
         {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
     )
-    # The completed event arrives over the API's SSE stream (the
-    # subscription re-fires it), never from the service handler.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     fired = hass.bus.fired(EVENT_QUEST_COMPLETED)
     assert len(fired) == 1
     payload = fired[0]
@@ -128,15 +111,6 @@ async def test_recomplete_no_op_fires_nothing(hass, make_entry) -> None:
         SERVICE_COMPLETE_QUEST,
         {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
     )
-    # The no-op verdict is decided in the API's completion layer: the
-    # second call appended nothing, so the API published no second
-    # frame and the bus stays at exactly one event.
-    client = entry.runtime_data.coordinator.api_client
-    assert [result.appended for _id, result in client.completions] == [
-        True,
-        False,
-    ]
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
 
 
@@ -154,9 +128,6 @@ async def test_child_day_complete_fires_when_day_clears(
                 "actor_child_id": child.id,
             },
         )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
     assert len(day_complete) == 1, "exactly one day-complete per cleared day"
     payload = day_complete[0]
@@ -182,12 +153,6 @@ async def test_child_day_complete_does_not_fire_partway(
             "actor_child_id": child.id,
         },
     )
-    # The API publishes the completion but not a day-complete (the
-    # child's day is not clear yet); the subscription re-fires only
-    # what was published.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert hass.bus.fired(EVENT_CHILD_DAY_COMPLETE) == []
     await hass.services.call(
         DOMAIN,
@@ -198,9 +163,6 @@ async def test_child_day_complete_does_not_fire_partway(
             "actor_child_id": child.id,
         },
     )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
 
 
@@ -261,15 +223,6 @@ async def test_appended_verdict_is_atomic_under_duplicate_calls(
         for _ in range(2)
     ]
     await asyncio.gather(*calls)
-    # The API's completion layer (the same core lock) decides the
-    # verdict: one call appended, the duplicate did not — so the API
-    # published exactly one frame and the bus carries one event.
-    client = entry.runtime_data.coordinator.api_client
-    assert len(client.completions) == 2
-    assert [result.appended for _id, result in client.completions].count(
-        True
-    ) == 1
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == 1
 
 
@@ -310,9 +263,6 @@ async def test_day_complete_ignored_for_backdated_completion(
                 "actor_child_id": child.id,
             },
         )
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1
 
     await _complete(
@@ -330,19 +280,12 @@ async def test_day_complete_ignored_for_backdated_completion(
             "actor_child_id": child.id,
         },
     )
-    # The service call on the already-done future instance is a no-op
-    # inside the API's completion layer: nothing appended, nothing
-    # published, nothing re-fired — today's single announcement stands.
-    client = entry.runtime_data.coordinator.api_client
-    assert client.completions[-1][1].appended is False
-    await refire_api_transitions(hass, entry, client)
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == 1, (
         "a back-dated/future completion must not re-announce today"
     )
-    # The direct business-layer call published nothing (only the API's
-    # completion path publishes), and the follow-up service call on
-    # the now-done future instance was a no-op: today's single
-    # completion stands.
+    # The direct business-layer call fired nothing (only the service
+    # path fires), and the follow-up service call on the now-done
+    # future instance was a no-op: today's single completion stands.
     assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == len(instances)
 
 
@@ -381,10 +324,6 @@ async def test_completion_updates_entities_immediately(
     assert remaining_after == len(instances) - 1, (
         "sensor must reflect the completion without a manual refresh"
     )
-    # The transition events arrive over the API's SSE stream.
-    await refire_api_transitions(
-        hass, entry, entry.runtime_data.coordinator.api_client
-    )
     assert len(hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)) == (
         1 if len(instances) == 1 else 0
     )
@@ -599,32 +538,17 @@ async def test_shim_fires_completed_before_day_complete(hass, make_entry) -> Non
     """The shim fires ``nestquest_quest_completed`` BEFORE
     ``nestquest_child_day_complete`` (the pre-extraction ordering), so
     a failure in the day-complete DB reads cannot suppress the
-    completed event.  Asserted by bus event ORDER, not just counts.
-
-    Driven directly: since Feature 18 the completion service proxies
-    to the API (whose event ordering is tested in test_api_sse.py);
-    this targets the shim module itself, the way its remaining
-    callers invoke it.
-    """
-    from custom_components.nestquest.completion import (
-        complete_instance as _complete,
-    )
-    from custom_components.nestquest.events import fire_quest_completed
-
+    completed event.  Asserted by bus event ORDER, not just counts."""
     entry, child, instances = await _setup_and_seed(hass, make_entry)
-    database = entry.runtime_data.database
     for instance in instances:
-        result = await _complete(
-            database,
-            instance.id,
-            actor_source="panel",
-            actor_child_id=child.id,
-        )
-        await fire_quest_completed(
-            hass,
-            database,
-            instance.id,
-            was_on_time=result.was_on_time,
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
         )
     ordered = [
         etype for etype, _payload in hass.bus.events
@@ -808,31 +732,21 @@ async def test_day_complete_fully_cleared_today_yields_event(
 async def test_shim_day_complete_uses_ha_local_timezone(
     hass, make_entry
 ) -> None:
-    """The shim's day-complete rule reads ``hass.config.time_zone``
-    and derives ``today`` in HA-local time, not UTC.
+    """The day-complete rule reads ``hass.config.time_zone`` and
+    derives ``today`` in HA-local time, not UTC.
 
     Picks a zone whose current local date differs from UTC, seeds an
-    instance for THAT zone's today (not UTC's today), fires the shim
-    directly for a completion of that instance, and asserts
+    instance for THAT zone's today (not UTC's today), completes it
+    through the service path, and asserts
     ``nestquest_child_day_complete`` fires — proving the local date
     reached the day-complete rule.  If no candidate zone has a
     different date right now (only possible inside the ~1h window
     around 12:00 UTC when all zones share one date), the test skips
     rather than run vacuously.
-
-    Driven directly: since Feature 18 the completion service proxies
-    to the API (whose day-complete ``today`` is the API host's local
-    date, tested in test_api_sse.py); this targets the shim module
-    itself, the way its remaining callers invoke it.
     """
     import datetime as _dt
     from zoneinfo import ZoneInfo
 
-    from custom_components.nestquest.completion import (
-        complete_instance as _complete,
-    )
-    from custom_components.nestquest.events import fire_quest_completed
-
     utc_today = _dt.datetime.now(_dt.timezone.utc).date()
     far_zone = None
     far_today = None
@@ -861,19 +775,15 @@ async def test_shim_day_complete_uses_ha_local_timezone(
     entry, child, instances = await _setup_and_seed(
         hass, make_entry, seed_date=far_today
     )
-    database = entry.runtime_data.database
     for instance in instances:
-        result = await _complete(
-            database,
-            instance.id,
-            actor_source="panel",
-            actor_child_id=child.id,
-        )
-        await fire_quest_completed(
-            hass,
-            database,
-            instance.id,
-            was_on_time=result.was_on_time,
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_COMPLETE_QUEST,
+            {
+                "instance_id": instance.id,
+                "actor": "panel",
+                "actor_child_id": child.id,
+            },
         )
     day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
     assert len(day_complete) == 1, (

tests/test_options_flow.py

@@ -15,28 +15,22 @@ from custom_components.nestquest import config_flow
 from custom_components.nestquest import options_flow
 from custom_components.nestquest.const import (
     CONF_ADMIN_USER_IDS,
+    CONF_DAY_ROLLOVER_TIME,
+    CONF_HORIZON_DAYS,
+    CONF_PANEL_IDLE_TIMEOUT,
+    CONF_UPDATE_INTERVAL,
     CONF_AFTERNOON_REMINDER_ENABLED,
     CONF_AFTERNOON_REMINDER_TIME,
-    CONF_API_BASE_URL,
     CONF_CELEBRATION_ENABLED,
-    CONF_DAY_ROLLOVER_TIME,
     CONF_END_OF_DAY_REPORT_ENABLED,
     CONF_END_OF_DAY_REPORT_TIME,
-    CONF_HORIZON_DAYS,
     CONF_MORNING_SUMMARY_ENABLED,
     CONF_MORNING_SUMMARY_TIME,
     CONF_NOTIFY_TARGET,
-    CONF_PANEL_IDLE_TIMEOUT,
-    CONF_PANEL_TOKEN,
-    CONF_SNAPSHOT_STALENESS,
-    CONF_UPDATE_INTERVAL,
-    DEFAULT_API_BASE_URL,
     DEFAULT_AFTERNOON_REMINDER_TIME,
     DEFAULT_AUTOMATION_ENABLED,
     DEFAULT_END_OF_DAY_REPORT_TIME,
     DEFAULT_MORNING_SUMMARY_TIME,
-    DEFAULT_PANEL_TOKEN,
-    DEFAULT_SNAPSHOT_STALENESS,
     DEFAULT_UPDATE_INTERVAL,
     DEFAULT_DAY_ROLLOVER_TIME,
     DEFAULT_HORIZON_DAYS,
@@ -55,13 +49,11 @@ VALID_INPUT = {
 }
 
 #: What a schema-validated submit produces from a bare VALID_INPUT on a
-#: fresh entry: the Feature 10 refresh interval, the Feature 11
-#: notification section, and the Feature 18 API connection fill from
-#: their defaults.
+#: fresh entry: the Feature 10 refresh interval and the Feature 11
+#: notification section fill from their defaults.
 FULL_INPUT = {
     **VALID_INPUT,
     CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
-    CONF_SNAPSHOT_STALENESS: DEFAULT_SNAPSHOT_STALENESS,
     CONF_NOTIFY_TARGET: "",
     CONF_MORNING_SUMMARY_TIME: DEFAULT_MORNING_SUMMARY_TIME,
     CONF_AFTERNOON_REMINDER_TIME: DEFAULT_AFTERNOON_REMINDER_TIME,
@@ -70,8 +62,6 @@ FULL_INPUT = {
     CONF_AFTERNOON_REMINDER_ENABLED: DEFAULT_AUTOMATION_ENABLED,
     CONF_END_OF_DAY_REPORT_ENABLED: DEFAULT_AUTOMATION_ENABLED,
     CONF_CELEBRATION_ENABLED: DEFAULT_AUTOMATION_ENABLED,
-    CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
-    CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
 }
 
 
@@ -249,7 +239,6 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
             CONF_DAY_ROLLOVER_TIME: "05:45",
             CONF_PANEL_IDLE_TIMEOUT: 120,
             CONF_UPDATE_INTERVAL: 90,
-            CONF_SNAPSHOT_STALENESS: 450,
             CONF_NOTIFY_TARGET: "notify.mobile_app_test",
             CONF_MORNING_SUMMARY_TIME: "07:00",
             CONF_AFTERNOON_REMINDER_TIME: "14:00",
@@ -262,7 +251,6 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
         CONF_DAY_ROLLOVER_TIME: "05:45",
         CONF_PANEL_IDLE_TIMEOUT: 120,
         CONF_UPDATE_INTERVAL: 90,
-        CONF_SNAPSHOT_STALENESS: 450,
         CONF_NOTIFY_TARGET: "notify.mobile_app_test",
         CONF_MORNING_SUMMARY_TIME: "07:00",
         CONF_AFTERNOON_REMINDER_TIME: "14:00",
@@ -271,8 +259,6 @@ def test_options_flow_schema_applies_stored_defaults() -> None:
         CONF_AFTERNOON_REMINDER_ENABLED: True,
         CONF_END_OF_DAY_REPORT_ENABLED: True,
         CONF_CELEBRATION_ENABLED: True,
-        CONF_API_BASE_URL: DEFAULT_API_BASE_URL,
-        CONF_PANEL_TOKEN: DEFAULT_PANEL_TOKEN,
         CONF_ADMIN_USER_IDS: [],
     }
 
@@ -286,9 +272,6 @@ def test_options_flow_schema_coerces_field_types() -> None:
     assert validators[CONF_HORIZON_DAYS] is int
     assert validators[CONF_DAY_ROLLOVER_TIME] is str
     assert validators[CONF_PANEL_IDLE_TIMEOUT] is int
-    # The Feature 18 API connection fields are free-text strings.
-    assert validators[CONF_API_BASE_URL] is str
-    assert validators[CONF_PANEL_TOKEN] is str
 
 
 @pytest.mark.parametrize("bad_horizon", [0, -5])
@@ -343,10 +326,8 @@ def test_options_flow_rejects_invalid_idle_timeout(bad_timeout) -> None:
     assert result["errors"] == {CONF_PANEL_IDLE_TIMEOUT: "invalid"}
 
 
-def test_options_flow_creates_entry_with_validated_keys() -> None:
-    """A valid submit creates an entry with exactly the validated keys
-    (the legacy settings plus every later section's keys, filled from
-    defaults where the raw submit omitted them)."""
+def test_options_flow_creates_entry_with_four_keys() -> None:
+    """A valid submit creates an entry with exactly the four validated keys."""
     flow = _make_flow(_make_entry())
     result = _run(flow.async_step_init(dict(VALID_INPUT)))
     assert result["type"] == "create_entry"
@@ -818,200 +799,3 @@ def test_options_flow_no_hard_coded_personal_target_in_source() -> None:
     assert const.__dict__.get("DEFAULT_NOTIFY_TARGET", "") == ""
     form = _run(_make_flow(_make_entry()).async_step_init(None))
     assert form["data_schema"]({})[CONF_NOTIFY_TARGET] == ""
-
-
-# ---------------------------------------------------------------------------
-# Panel-plane API connection section (Feature 18)
-# ---------------------------------------------------------------------------
-
-
-def test_options_flow_api_section_defaults_saved() -> None:
-    """A bare submit saves the API connection defaults: the const base
-    URL and an EMPTY panel token (no token is ever invented)."""
-    flow = _make_flow(_make_entry())
-    result = _run(flow.async_step_init(dict(VALID_INPUT)))
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_API_BASE_URL] == DEFAULT_API_BASE_URL
-    assert result["data"][CONF_PANEL_TOKEN] == ""
-
-
-def test_options_flow_api_base_url_must_be_full_http_url() -> None:
-    """Scheme-less, host-less, empty, and non-string base URLs reject
-    with a field error — the same rule the client's constructor
-    enforces, so a malformed URL fails at the form, not at the first
-    request."""
-    flow = _make_flow(_make_entry())
-    for bad_url in ("api.lan:8000", "http://", "", "   ", 42):
-        result = _run(
-            flow.async_step_init(
-                {**VALID_INPUT, CONF_API_BASE_URL: bad_url}
-            )
-        )
-        assert result["type"] == "form"
-        assert result["errors"] == {CONF_API_BASE_URL: "invalid_url"}
-
-
-def test_options_flow_panel_token_validation() -> None:
-    """The EMPTY token is valid (the API connection is simply not
-    configured yet); padded and non-string values reject — a padded
-    paste would fail the token check at request time instead."""
-    flow = _make_flow(_make_entry())
-    result = _run(
-        flow.async_step_init({**VALID_INPUT, CONF_PANEL_TOKEN: ""})
-    )
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_PANEL_TOKEN] == ""
-
-    for bad_token in ("  secret  ", "   ", 42):
-        flow = _make_flow(_make_entry())
-        result = _run(
-            flow.async_step_init(
-                {**VALID_INPUT, CONF_PANEL_TOKEN: bad_token}
-            )
-        )
-        assert result["type"] == "form"
-        assert result["errors"] == {CONF_PANEL_TOKEN: "invalid"}
-
-
-def test_options_flow_api_section_stored_values_round_trip() -> None:
-    """Stored API settings are the form's defaults and a raw legacy
-    submit without them keeps the stored values."""
-    entry = _make_entry(
-        options={
-            **VALID_INPUT,
-            CONF_API_BASE_URL: "http://panel.lan:8443",
-            CONF_PANEL_TOKEN: "stored-token",
-        }
-    )
-    flow = _make_flow(entry)
-    form = _run(flow.async_step_init(None))
-    schema_defaults = form["data_schema"]({})
-    assert schema_defaults[CONF_API_BASE_URL] == "http://panel.lan:8443"
-    assert schema_defaults[CONF_PANEL_TOKEN] == "stored-token"
-
-    # A raw submit carrying only the legacy keys keeps the stored
-    # API connection settings.
-    result = _run(flow.async_step_init(dict(VALID_INPUT)))
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_API_BASE_URL] == "http://panel.lan:8443"
-    assert result["data"][CONF_PANEL_TOKEN] == "stored-token"
-
-
-def test_options_flow_strings_cover_api_section(
-    strings: dict, translations_en: dict
-) -> None:
-    """Both string files name the API fields and the URL error."""
-    data = strings["options"]["step"]["init"]["data"]
-    for key in (CONF_API_BASE_URL, CONF_PANEL_TOKEN):
-        assert data.get(key), f"Missing field name for '{key}'"
-    assert strings["options"]["errors"]["invalid_url"]
-    assert translations_en["options"] == strings["options"]
-
-
-# ---------------------------------------------------------------------------
-# Last-good snapshot staleness section (Feature 18)
-# ---------------------------------------------------------------------------
-
-
-def test_options_flow_staleness_below_update_interval_refused() -> None:
-    """A staleness threshold shorter than the submit's update interval
-    is refused with the dedicated error: such a threshold could never
-    serve even one cached snapshot."""
-    flow = _make_flow(_make_entry())
-    result = _run(
-        flow.async_step_init(
-            {
-                **VALID_INPUT,
-                CONF_UPDATE_INTERVAL: 300,
-                CONF_SNAPSHOT_STALENESS: 60,
-            }
-        )
-    )
-    assert result["type"] == "form"
-    assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid_staleness"}
-
-
-def test_options_flow_staleness_below_stored_interval_refused() -> None:
-    """A raw submit setting the staleness but omitting the update
-    interval is checked against the STORED interval, not the default."""
-    entry = _make_entry(options={CONF_UPDATE_INTERVAL: 300})
-    flow = _make_flow(entry)
-    result = _run(
-        flow.async_step_init({**VALID_INPUT, CONF_SNAPSHOT_STALENESS: 60})
-    )
-    assert result["type"] == "form"
-    assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid_staleness"}
-
-
-def test_options_flow_staleness_must_be_positive_int() -> None:
-    """Non-int, boolean, and non-positive staleness values reject with
-    the plain invalid error."""
-    flow = _make_flow(_make_entry())
-    for bad_staleness in ("900", True, 0, -10):
-        result = _run(
-            flow.async_step_init(
-                {**VALID_INPUT, CONF_SNAPSHOT_STALENESS: bad_staleness}
-            )
-        )
-        assert result["type"] == "form"
-        assert result["errors"] == {CONF_SNAPSHOT_STALENESS: "invalid"}
-
-
-def test_options_flow_staleness_equal_to_interval_saves() -> None:
-    """A threshold exactly as long as the update interval is valid (the
-    boundary is inclusive)."""
-    flow = _make_flow(_make_entry())
-    result = _run(
-        flow.async_step_init(
-            {
-                **VALID_INPUT,
-                CONF_UPDATE_INTERVAL: 300,
-                CONF_SNAPSHOT_STALENESS: 300,
-            }
-        )
-    )
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_SNAPSHOT_STALENESS] == 300
-
-
-def test_options_flow_staleness_defaults_saved() -> None:
-    """A bare submit saves the staleness default (three update
-    intervals) alongside the other timing defaults."""
-    flow = _make_flow(_make_entry())
-    result = _run(flow.async_step_init(dict(VALID_INPUT)))
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_SNAPSHOT_STALENESS] == DEFAULT_SNAPSHOT_STALENESS
-
-
-def test_options_flow_staleness_stored_values_round_trip() -> None:
-    """A stored staleness is the form's default and a raw legacy submit
-    without it keeps the stored value."""
-    entry = _make_entry(
-        options={
-            **VALID_INPUT,
-            CONF_UPDATE_INTERVAL: 300,
-            CONF_SNAPSHOT_STALENESS: 1200,
-        }
-    )
-    flow = _make_flow(entry)
-    form = _run(flow.async_step_init(None))
-    schema_defaults = form["data_schema"]({})
-    assert schema_defaults[CONF_SNAPSHOT_STALENESS] == 1200
-
-    # A raw submit carrying only the legacy keys keeps the stored
-    # staleness (and never trips the interval cross-check).
-    result = _run(flow.async_step_init(dict(VALID_INPUT)))
-    assert result["type"] == "create_entry"
-    assert result["data"][CONF_SNAPSHOT_STALENESS] == 1200
-
-
-def test_options_flow_strings_cover_staleness_section(
-    strings: dict, translations_en: dict
-) -> None:
-    """Both string files name the staleness field and its error."""
-    data = strings["options"]["step"]["init"]["data"]
-    assert data.get(CONF_SNAPSHOT_STALENESS), (
-        f"Missing field name for '{CONF_SNAPSHOT_STALENESS}'"
-    )
-    assert strings["options"]["errors"]["invalid_staleness"]
-    assert translations_en["options"] == strings["options"]

tests/test_sensors.py

@@ -221,23 +221,19 @@ async def test_due_today_attributes_carry_the_panel_payload(
     assert bo_instance["on_time"] is None
 
 
-async def test_due_today_payload_omits_missed_admin_equals_panel(
+async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
     hass, make_entry, monkeypatch
 ) -> None:
-    """D-009: the panel payload omits missed instances.
+    """D-009: the panel payload omits missed instances, the admin one keeps them.
 
-    Since Feature 18 the snapshot comes from the API panel route,
-    which omits missed rows entirely (the admin surface for missed
-    quests is the PWA, reading the API directly).  The entity payload
-    therefore cannot carry missed rows any more: ``admin_instances``
-    equals ``instances`` — open and completed rows only.  A missed
-    instance is still derived locally (``due_date`` before the
-    snapshot's "today"); the pinned-FUTURE clock at the
-    ``derive_state`` boundary makes today's open instances read
-    missed through the same derivation rule, and this test pins the
-    change: neither payload carries them, and the count rollups
-    (precomputed server-side from the full day) are unaffected here
-    because the API counts arrive per the route's own build.
+    A missed instance is derived (``due_date`` before the snapshot's
+    "today"), never stored.  The coordinator's day listing is keyed on
+    ``_local_now``'s date, so pinning the clock forward would move the
+    listing window off today's rows entirely; the pinned FUTURE clock
+    is therefore applied at the ``derive_state`` boundary instead —
+    today's open instances then read as missed through the same
+    derivation rule, without back-dating instance rows past the DAO's
+    no-past guard or touching append-only completion events.
     """
     entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
         hass, make_entry
@@ -260,13 +256,10 @@ async def test_due_today_payload_omits_missed_admin_equals_panel(
     assert attributes["instances"] == [], (
         "missed instances are omitted from the panel payload"
     )
-    assert attributes["admin_instances"] == [], (
-        "the API payload omits missed rows, so the admin payload "
-        "carries none either (the PWA is the admin surface now)"
-    )
-    # Counts still roll the missed instance up as not-completed: the
-    # counts arrive precomputed from the route's own build, which rolls
-    # the FULL day (the missed row included) into the numbers.
+    [missed] = attributes["admin_instances"]
+    assert missed["state"] == "missed"
+    assert missed["completed_at"] is None
+    # Counts still roll the missed instance up as not-completed.
     assert _sensor(hass, bo.id, "quests_completed_today").native_value == 0
     assert _sensor(hass, bo.id, "quests_remaining_today").native_value == 1
 

tests/test_sse_subscription.py

@@ -1,619 +0,0 @@
-"""Tests for the panel SSE subscription (HA-bus re-firing task).
-
-Three layers are exercised, all STUB-driven — no network, no aiohttp,
-no real Home Assistant:
-
-- The API client's ``stream_events`` (transport stub): what goes on
-  the wire (GET, the events URL, the Bearer header), SSE frame
-  parsing (``event:``/``data:`` lines, blank-line frame ends, chunk
-  boundaries, multi-line data), and every failure mode (non-2xx,
-  connection failure, mid-stream drop, malformed frame) surfacing as
-  the typed :class:`NestQuestApiError`; cancellation closes the
-  stream cleanly.
-- The :class:`~custom_components.nestquest.sse.NestQuestEventStream`
-  subscription manager (duck-typed stream client): the four documented
-  transition events (design/ENTITIES-AND-SERVICES.md §3, exactly as
-  ``tests/test_api_sse.py`` observes them on the API side) fire on
-  ``hass.bus`` with the documented payload fields; ``stop()`` cancels
-  the task so no frame fires afterwards; a stream failure reconnects
-  with a bounded, reset-on-success backoff and never crashes the
-  task.
-- The integration wiring: setup starts the stream alongside the
-  runtime record, unload tears it down (no firing afterwards, task
-  cancelled), and an entry whose client has no stream surface (the
-  unconfigured-token stub, the test snapshot client) starts none.
-"""
-from __future__ import annotations
-
-import asyncio
-import json
-
-import pytest
-
-from conftest import set_coordinator_client, wire_entry_to_registry
-
-import custom_components.nestquest as nq_integration
-from custom_components.nestquest import (
-    async_setup_entry,
-    async_unload_entry,
-)
-from custom_components.nestquest import sse as sse_module
-from custom_components.nestquest.api_client import (
-    NestQuestApiClient,
-    NestQuestApiError,
-)
-from custom_components.nestquest.sse import NestQuestEventStream
-
-BASE_URL = "http://api.test:8000"
-TOKEN = "test-panel-token"
-
-# Captured BEFORE any test patches asyncio.sleep (the reconnect test
-# swaps in a recording fake; this reference keeps helpers sleeping for
-# real regardless).
-_REAL_SLEEP = asyncio.sleep
-
-#: The four documented transition events with their documented payload
-#: fields — the exact shapes the API side streams (test_api_sse.py).
-DOCUMENTED_FRAMES = (
-    (
-        "nestquest_quest_completed",
-        {
-            "child_id": 1,
-            "child_name": "Ada",
-            "instance_id": 7,
-            "quest_title": "Pack bag",
-            "window": "morning",
-            "due_date": "2026-09-23",
-            "due_time": "10:00",
-            "occurred_at": "2026-09-23T12:00:00+00:00",
-            "was_on_time": False,
-        },
-    ),
-    (
-        "nestquest_quest_uncompleted",
-        {
-            "child_id": 1,
-            "child_name": "Ada",
-            "instance_id": 7,
-            "quest_title": "Pack bag",
-            "window": "morning",
-            "due_date": "2026-09-23",
-            "due_time": "10:00",
-            "occurred_at": "2026-09-23T12:00:00+00:00",
-        },
-    ),
-    (
-        "nestquest_quest_missed",
-        {
-            "child_id": 1,
-            "child_name": "Ada",
-            "instance_id": 9,
-            "quest_title": "Stale chore",
-            "window": "morning",
-            "due_date": "2026-09-23",
-            "due_time": "09:00",
-            "occurred_at": "2026-09-23T23:30:00+00:00",
-        },
-    ),
-    (
-        "nestquest_child_day_complete",
-        {
-            "child_id": 1,
-            "child_name": "Ada",
-            "quests_due": 1,
-            "quests_completed": 1,
-            "occurred_at": "2026-09-23T12:00:00+00:00",
-        },
-    ),
-)
-
-#: The documented payload field sets per event type.
-DOCUMENTED_FIELDS = {
-    "nestquest_quest_completed": {
-        "child_id",
-        "child_name",
-        "instance_id",
-        "quest_title",
-        "window",
-        "due_date",
-        "due_time",
-        "occurred_at",
-        "was_on_time",
-    },
-    "nestquest_quest_uncompleted": {
-        "child_id",
-        "child_name",
-        "instance_id",
-        "quest_title",
-        "window",
-        "due_date",
-        "due_time",
-        "occurred_at",
-    },
-    "nestquest_quest_missed": {
-        "child_id",
-        "child_name",
-        "instance_id",
-        "quest_title",
-        "window",
-        "due_date",
-        "due_time",
-        "occurred_at",
-    },
-    "nestquest_child_day_complete": {
-        "child_id",
-        "child_name",
-        "quests_due",
-        "quests_completed",
-        "occurred_at",
-    },
-}
-
-#: A minimal panel snapshot payload for the full-setup wiring tests
-#: (the shape ``snapshot_from_api_payload`` consumes; no children).
-MINIMAL_SNAPSHOT = {"today_iso": "2026-09-23", "cycle_day": 1, "children": []}
-
-
-# ---------------------------------------------------------------------------
-# Stub transports and stream clients (no network anywhere)
-# ---------------------------------------------------------------------------
-
-
-class _StreamingResponse:
-    """A fake aiohttp streaming response: ``status`` + chunk iteration.
-
-    ``chunks`` are the body pieces the stream yields in order; a chunk
-    may be an exception, raised when the iteration reaches it (a
-    mid-stream transport drop).  ``hold_open`` keeps the stream open
-    forever after the chunks run out, for cancellation tests.
-    """
-
-    def __init__(self, status: int, chunks, *, hold_open: bool = False) -> None:
-        self.status = status
-        self._chunks = list(chunks)
-        self._hold_open = hold_open
-        self.closed = False
-
-    def __aiter__(self):
-        return self._iterate()
-
-    async def _iterate(self):
-        for chunk in self._chunks:
-            if isinstance(chunk, BaseException):
-                raise chunk
-            yield chunk
-        if self._hold_open:
-            await asyncio.Event().wait()  # cancelled, never ends
-
-
-class _RequestContext:
-    """Async context manager mirroring ``session.request(...)``'s return.
-
-    Entering performs the request (a scripted transport error raises
-    THERE, where aiohttp's connector raises on a refused connection);
-    exiting marks the response closed, so tests can assert the stream
-    was released.
-    """
-
-    def __init__(self, response, error) -> None:
-        self._response = response
-        self._error = error
-
-    async def __aenter__(self):
-        if self._error is not None:
-            raise self._error
-        return self._response
-
-    async def __aexit__(self, exc_type, exc, tb) -> bool:
-        if self._response is not None:
-            self._response.closed = True
-        return False
-
-
-class _StreamTransport:
-    """A no-network stand-in for aiohttp's ``ClientSession``.
-
-    Records every ``request(...)`` call and answers from a scripted
-    queue of outcomes (a :class:`_StreamingResponse` or an exception
-    per attempt, consumed in order).
-    """
-
-    def __init__(self, *outcomes) -> None:
-        self.calls: list[dict] = []
-        self._outcomes = list(outcomes)
-
-    def request(self, method: str, url: str, **kwargs):
-        self.calls.append(
-            {
-                "method": method,
-                "url": url,
-                "headers": kwargs.get("headers"),
-                "json": kwargs.get("json"),
-            }
-        )
-        outcome = self._outcomes.pop(0)
-        if isinstance(outcome, BaseException):
-            return _RequestContext(None, outcome)
-        return _RequestContext(outcome, None)
-
-
-def make_stream_client(transport) -> NestQuestApiClient:
-    """Build a client over the streaming stub with the test credentials."""
-    return NestQuestApiClient(BASE_URL, TOKEN, transport)
-
-
-def _frame_text(event_type: str, payload: dict) -> str:
-    """One documented SSE frame: event line, data line, blank line."""
-    return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n"
-
-
-class _ScriptedStreamClient:
-    """Duck-typed API client: one scripted stream per connection.
-
-    ``script`` holds generator FACTORIES — ``stream_events()`` pops the
-    next one and returns its async generator, so a test scripts exactly
-    what each connection attempt yields (or how it fails).
-    """
-
-    def __init__(self, script) -> None:
-        self._script = list(script)
-        self.calls = 0
-
-    def stream_events(self):
-        self.calls += 1
-        return self._script.pop(0)()
-
-
-def _frames(*frames):
-    """A live stream factory: yields ``frames``, then stays open."""
-
-    def _factory():
-        async def _gen():
-            for frame in frames:
-                yield frame
-            await asyncio.Event().wait()  # cancelled, never ends
-
-        return _gen()
-
-    return _factory
-
-
-def _fail(error: BaseException):
-    """A stream factory that fails before yielding anything."""
-
-    def _factory():
-        async def _gen():
-            raise error
-            yield  # noqa: unreachable — makes _gen an async generator
-
-        return _gen()
-
-    return _factory
-
-
-def _drop_after(error: BaseException, *frames):
-    """A stream factory that yields ``frames`` then drops mid-stream."""
-
-    def _factory():
-        async def _gen():
-            for frame in frames:
-                yield frame
-            raise error
-
-        return _gen()
-
-    return _factory
-
-
-async def _wait_until(predicate, *, timeout: float = 2.0) -> None:
-    """Poll ``predicate`` until true or the deadline passes."""
-    loop = asyncio.get_running_loop()
-    deadline = loop.time() + timeout
-    while not predicate():
-        if loop.time() > deadline:
-            raise AssertionError("condition not met within the deadline")
-        await _REAL_SLEEP(0.01)
-
-
-def make_manager(
-    hass, client, *, min_backoff: float = 0.01, max_backoff: float = 0.05
-) -> NestQuestEventStream:
-    """A manager with fast (test-sized) backoff bounds."""
-    return NestQuestEventStream(
-        hass,
-        client,
-        reconnect_min_seconds=min_backoff,
-        reconnect_max_seconds=max_backoff,
-    )
-
-
-# ---------------------------------------------------------------------------
-# The client's stream_events: parsing and typed failures
-# ---------------------------------------------------------------------------
-
-
-async def test_stream_events_yields_documented_frames_and_sends_bearer() -> None:
-    """All four documented frames parse, with the Bearer GET on the wire."""
-    body = "".join(
-        _frame_text(event_type, payload)
-        for event_type, payload in DOCUMENTED_FRAMES
-    )
-    transport = _StreamTransport(_StreamingResponse(200, [body]))
-    client = make_stream_client(transport)
-
-    frames = [frame async for frame in client.stream_events()]
-
-    assert frames == list(DOCUMENTED_FRAMES)
-    call = transport.calls[0]
-    assert call["method"] == "GET"
-    assert call["url"] == f"{BASE_URL}/api/v1/panel/events"
-    assert call["headers"] == {"Authorization": f"Bearer {TOKEN}"}
-    assert call["json"] is None
-
-
-async def test_stream_events_buffers_lines_across_chunks_and_joins_data() -> None:
-    """Chunk-split lines parse, and multi-line data joins per SSE rules."""
-    first_type, first_payload = DOCUMENTED_FRAMES[0]
-    # The first frame is cut MID-LINE across two chunks; the second
-    # frame spreads its JSON over two ``data:`` lines.
-    chunks = [
-        f"event: {first_type}\nda",
-        f"ta: {json.dumps(first_payload)}\n\nevent: "
-        f"{DOCUMENTED_FRAMES[1][0]}\n",
-        f'data: {{"extra":\ndata: true}}\n\n',
-    ]
-    transport = _StreamTransport(_StreamingResponse(200, chunks))
-    client = make_stream_client(transport)
-
-    frames = [frame async for frame in client.stream_events()]
-
-    assert len(frames) == 2
-    assert frames[0] == DOCUMENTED_FRAMES[0]
-    # The split-out frame's data spread over two ``data:`` lines joins
-    # with a newline (the SSE standard) and parses as one JSON body.
-    assert frames[1] == (DOCUMENTED_FRAMES[1][0], {"extra": True})
-
-
-async def test_stream_events_non_2xx_raises_typed_and_closes() -> None:
-    """A wrong token (401) is typed with its status; no stream is read."""
-    response = _StreamingResponse(401, ["event: x\ndata: {}\n\n"])
-    transport = _StreamTransport(response)
-    client = make_stream_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        async for _ in client.stream_events():
-            pass
-
-    assert excinfo.value.status == 401
-    assert not excinfo.value.is_transport_error
-    assert TOKEN not in str(excinfo.value)
-    assert response.closed
-
-
-async def test_stream_events_connection_failure_raises_typed_transport_error() -> None:
-    """A failure before any response is a typed transport error."""
-    transport = _StreamTransport(OSError("connection refused"))
-    client = make_stream_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        async for _ in client.stream_events():
-            pass
-
-    assert excinfo.value.status is None
-    assert excinfo.value.is_transport_error
-    assert isinstance(excinfo.value.__cause__, OSError)
-
-
-async def test_stream_events_midstream_drop_raises_typed() -> None:
-    """A transport error mid-stream is typed, so the consumer reconnects."""
-    response = _StreamingResponse(
-        200,
-        [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1]), OSError("reset")],
-    )
-    transport = _StreamTransport(response)
-    client = make_stream_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        async for _ in client.stream_events():
-            pass
-
-    assert excinfo.value.status is None
-    assert excinfo.value.is_transport_error
-    assert isinstance(excinfo.value.__cause__, OSError)
-    assert response.closed
-
-
-async def test_stream_events_stops_cleanly_on_cancellation() -> None:
-    """Closing the generator (the consumer task's cancellation) stops
-    cleanly: no error, the underlying response is closed, and nothing
-    further is yielded."""
-    response = _StreamingResponse(
-        200,
-        [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1])],
-        hold_open=True,
-    )
-    transport = _StreamTransport(response)
-    client = make_stream_client(transport)
-
-    generator = client.stream_events()
-    first = await asyncio.wait_for(generator.__anext__(), timeout=2.0)
-    assert first == DOCUMENTED_FRAMES[0]
-
-    await asyncio.wait_for(generator.aclose(), timeout=2.0)
-    assert response.closed
-
-
-async def test_stream_events_malformed_frame_raises_typed() -> None:
-    """A frame whose data is not JSON is a typed protocol failure."""
-    response = _StreamingResponse(
-        200, ["event: nestquest_quest_completed\ndata: not-json\n\n"]
-    )
-    transport = _StreamTransport(response)
-    client = make_stream_client(transport)
-
-    with pytest.raises(NestQuestApiError) as excinfo:
-        async for _ in client.stream_events():
-            pass
-
-    assert excinfo.value.status == 200
-    assert response.closed
-
-
-# ---------------------------------------------------------------------------
-# The subscription manager: firing, teardown, reconnect
-# ---------------------------------------------------------------------------
-
-
-async def test_manager_fires_documented_frames_on_ha_bus(hass) -> None:
-    """Every documented transition fires on hass.bus with its payload."""
-    client = _ScriptedStreamClient([_frames(*DOCUMENTED_FRAMES)])
-    manager = make_manager(hass, client)
-    manager.start()
-    try:
-        await _wait_until(lambda: len(hass.bus.events) >= 4)
-        assert hass.bus.events == [tuple(f) for f in DOCUMENTED_FRAMES]
-        for event_type, payload in DOCUMENTED_FRAMES:
-            fired = hass.bus.fired(event_type)
-            assert fired, f"{event_type} never fired"
-            assert set(fired[-1]) == DOCUMENTED_FIELDS[event_type]
-    finally:
-        await manager.stop()
-
-
-async def test_manager_stop_cancels_task_and_blocks_later_frames(hass) -> None:
-    """stop() cancels and awaits the task: no frame fires afterwards,
-    even when the (dead) stream is later offered more data."""
-    stream_release = asyncio.Event()
-    first_type, first_payload = DOCUMENTED_FRAMES[0]
-    later_type, later_payload = DOCUMENTED_FRAMES[1]
-
-    def _factory():
-        async def _gen():
-            yield (first_type, dict(first_payload))
-            await stream_release.wait()
-            yield (later_type, dict(later_payload))
-
-        return _gen()
-
-    client = _ScriptedStreamClient([_factory])
-    manager = make_manager(hass, client)
-    manager.start()
-    await _wait_until(lambda: len(hass.bus.events) == 1)
-
-    task = manager.task
-    await manager.stop()
-    assert task.cancelled()
-    assert manager.task is None
-
-    # Offer the stream more data after teardown: nothing fires.
-    stream_release.set()
-    await _REAL_SLEEP(0.05)
-    assert hass.bus.events == [(first_type, dict(first_payload))]
-
-
-async def test_manager_reconnects_with_bounded_backoff(hass, monkeypatch) -> None:
-    """A failed stream reconnects with a bounded, reset-on-success
-    backoff and never crashes the task."""
-    delays: list[float] = []
-
-    async def fake_sleep(delay, *args, **kwargs):
-        delays.append(delay)
-        await _REAL_SLEEP(0)
-
-    monkeypatch.setattr(sse_module.asyncio, "sleep", fake_sleep)
-
-    stream_error = ConnectionError("stream reset")
-    client = _ScriptedStreamClient(
-        [
-            _fail(RuntimeError("api down")),
-            _fail(RuntimeError("api down")),
-            _fail(RuntimeError("api down")),
-            _fail(RuntimeError("api down")),
-            _drop_after(stream_error, DOCUMENTED_FRAMES[0]),
-            _frames(*DOCUMENTED_FRAMES),
-        ]
-    )
-    manager = make_manager(hass, client, min_backoff=0.05, max_backoff=0.2)
-    manager.start()
-    try:
-        await _wait_until(lambda: len(hass.bus.events) >= 5)
-        # 5s → 10s → 20s (cap) → 20s (cap), then a frame on a healthy
-        # stream reset it to the 5s minimum for the next failure.
-        assert delays == [0.05, 0.1, 0.2, 0.2, 0.05]
-        assert hass.bus.events == [
-            tuple(DOCUMENTED_FRAMES[0]),
-            *[tuple(f) for f in DOCUMENTED_FRAMES],
-        ]
-        # The task is still alive (subscribed and idle), not crashed.
-        assert manager.task is not None and not manager.task.done()
-    finally:
-        await manager.stop()
-
-
-def test_manager_rejects_invalid_backoff_bounds(hass) -> None:
-    """Non-positive or inverted backoff bounds fail at construction."""
-    with pytest.raises(ValueError):
-        NestQuestEventStream(hass, object(), reconnect_min_seconds=0)
-    with pytest.raises(ValueError):
-        NestQuestEventStream(hass, object(), reconnect_min_seconds=-1)
-    with pytest.raises(ValueError):
-        NestQuestEventStream(hass, object(), reconnect_max_seconds=0)
-    with pytest.raises(ValueError):
-        NestQuestEventStream(hass, object(), reconnect_min_seconds=10.0, reconnect_max_seconds=5.0)
-
-
-# ---------------------------------------------------------------------------
-# Integration wiring: setup starts the stream, unload stops it
-# ---------------------------------------------------------------------------
-
-
-class _SetupClient:
-    """Client for full-setup wiring tests: scripted snapshot + stream."""
-
-    def __init__(self, script) -> None:
-        self._script = list(script)
-        self.stream_calls = 0
-
-    async def get_snapshot(self) -> dict:
-        return dict(MINIMAL_SNAPSHOT)
-
-    def stream_events(self):
-        self.stream_calls += 1
-        return self._script.pop(0)()
-
-
-async def test_setup_starts_stream_and_unload_stops_it(hass, make_entry) -> None:
-    """Setup starts the subscription alongside the runtime record; unload
-    tears the task down so no frame fires afterwards."""
-    client = _SetupClient([_frames(*DOCUMENTED_FRAMES)])
-    set_coordinator_client(make_entry(), client)
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
-    assert await nq_integration.async_setup_entry(hass, entry) is True
-
-    runtime = entry.runtime_data
-    assert runtime.event_stream is not None
-    await _wait_until(lambda: len(hass.bus.events) >= 4)
-    assert [event for event, _ in hass.bus.events] == [
-        frame[0] for frame in DOCUMENTED_FRAMES
-    ]
-
-    task = runtime.event_stream.task
-    assert await nq_integration.async_unload_entry(hass, entry) is True
-    assert task.cancelled()
-    assert runtime.event_stream.task is None
-
-    events_after_unload = len(hass.bus.events)
-    await _REAL_SLEEP(0.05)
-    assert len(hass.bus.events) == events_after_unload
-
-
-async def test_setup_without_stream_surface_starts_no_subscription(
-    hass, make_entry
-) -> None:
-    """An entry whose API client has no stream surface (the local-snapshot
-    stub the harness installs, like the unconfigured-token stub in
-    production) starts NO subscription — and setup still succeeds."""
-    entry = wire_entry_to_registry(make_entry(), hass.registry)
-    assert await nq_integration.async_setup_entry(hass, entry) is True
-    assert entry.runtime_data.event_stream is None
-    assert await nq_integration.async_unload_entry(hass, entry) is True