Compare commits

..
Author SHA1 Message Date
overseer ffda0a2ed7 complete_quest proxies to the API so the panel tap is one write
WHY: Feature 18 makes the integration a client of the NestQuest API
service — every database write must go through the API.  The
complete_quest service still wrote the local database and fired the
completed event itself, which would double-fire once the SSE
subscription re-fires the API's transition (and stops being true the
moment the local database is dropped).  The service now proxies to the
entry's ONE shared API client (complete_instance: the panel complete
route) and keeps only the old immediate-update behaviour: a successful
call refreshes the coordinator so the sensors reflect the completion
at once; the bus event comes solely from the SSE stream.  Typed API
failures (and an unconfigured client, and a non-panel actor) surface
as HomeAssistantError with token-free messages instead of being
swallowed; the payload contract (instance_id, actor=panel,
actor_child_id) is unchanged for the panel cards.

Tests: new tests/test_complete_quest_proxy.py asserts the call reaches
the shared client with the right arguments, that no local write or
local event happens on the API path, that a non-2xx/transport failure
raises a service error chained to the typed error (never containing
the token), and that the unconfigured stub fails clearly.  Existing
tests that drove complete_quest through the local database move to the
API path via the harness stand-in client (which performs the same core
completion and publishes the same SSE frames the API route does) and
assert their bus events after the frames are delivered through the
real SSE subscription; the shim-level tests drive the events shim
directly, the way its remaining callers invoke it.

Task 47c7c90e (complete_quest proxies to the API, Feature 18).
2026-09-23 07:44:46 -04:00
dev_agent 84d179336b Task b1ed9c45: SSE subscription re-fires the four HA bus events (#168) 2026-09-23 07:04:39 -04:00
dev_agent fafabeabbe Task 6e70d166: Coordinator polls the snapshot and builds the sensors (#167) 2026-09-23 06:33:44 -04:00
dev_agent 78fd75ec45 Task 37d5e806: API client module for the integration (Feature 18) (#166) 2026-09-23 05:12:27 -04:00
dev_agent d78ef196d0 Feature 16: NestQuest API Service (#165)
Co-authored-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
Co-committed-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
2026-09-23 04:37:26 -04:00
orchestrator-agent b5c2b55d9b Run NestQuest orchestration directly in the terminal (#151)
Co-authored-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
Co-committed-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
2026-09-22 21:35:30 -04:00
orchestrator-agent 8ddae78551 Start NestQuest development manually from OpenCode (#150)
Co-authored-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
Co-committed-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
2026-09-22 11:46:40 -04:00
orchestrator-agent 4ac4d3f1ec Add durable NestQuest OpenCode controller and role launchers (#149)
Co-authored-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
Co-committed-by: orchestrator-agent <orchestrator@cubecraftlabs.com>
2026-09-22 11:18:23 -04:00
41 changed files with 5723 additions and 374 deletions
+32
View File
@@ -0,0 +1,32 @@
---
description: Independently verify NestQuest merge gates and merge approved heads
mode: primary
model: openrouter/deepseek/deepseek-v4.1-flash
permission:
edit: deny
write: deny
task: deny
bash:
"*": deny
"/home/overseer/.local/share/nestquest-controller/scripts/nq-approve-merge *": allow
external_directory:
"*": ask
"/tmp/opencode/*": allow
"/home/overseer/.local/share/nestquest-controller/*": allow
---
You are the NestQuest feature-to-dev approver, separate from the controller and
developer sessions. Verify the current PR head equals the independently
reviewed and approved commit,
required checks pass, blocking findings are resolved, branch protection is
satisfied, and Maestro has accurate links. Use only the approver Gitea identity
for the merge. Do not change code or approve a stale head. Use only the
installed nq-approve-merge launcher for Gitea inspection and merging. Merge
only into dev; never merge into main. If any gate is
missing, refuse the merge and give the exact missing evidence. Report the PR,
reviewed head, resulting merge commit, and target branch.
Broker syntax: `nq-approve-merge inspect PR_NUMBER` or
`nq-approve-merge merge PR_NUMBER REVIEWED_HEAD REVIEW_FILE`. Run one broker
command per Bash call. The broker validates the current PR and review again
before merging; do not substitute an ad hoc Gitea command.
+21
View File
@@ -0,0 +1,21 @@
---
description: Implement one assigned NestQuest task in an isolated worktree
mode: primary
model: openrouter/z-ai/glm-5.3-flash
permission:
task: deny
external_directory:
"*": ask
"/tmp/opencode/*": allow
---
You are the NestQuest development agent. Implement only the assigned Maestro
task in the supplied worktree. Follow the task's acceptance criteria and the
repository's instructions. Run relevant tests, commit on the task branch, and
report the commit and test results. Keep changes inside the assigned worktree.
Do not approve your own pull request. You may merge your task PR into the
feature branch only after the independent reviewer approved its exact current
head and the controller verified every task merge gate. Never merge a feature
into dev or a release into main. Do not start another task, change role, or
use another model. Do not read or print credential files except when the
approved task merge requires the existing Gitea development credential.
+60
View File
@@ -0,0 +1,60 @@
---
description: Reconcile NestQuest work and dispatch the next verified action
mode: primary
model: openrouter/deepseek/deepseek-v4.1-flash
permission:
doom_loop: deny
task: deny
external_directory:
"*": ask
"/tmp/opencode/*": allow
"/home/overseer/.local/share/opencode/tool-output/*": allow
---
You are the NestQuest controller. Work in the NestQuest repository. Maestro is
the source of truth for feature and task state; CTXD is the source of truth for
project instructions. At the start of every invocation, read the current
NestQuest context in CTXD, including OPENCODE-ORCHESTRATOR-PROMPT.MD, then
reconcile it with Maestro, Git, pull requests, and checks before dispatching.
Follow the review, identity, and branch rules in that prompt.
If the CTXD project or prompt is unavailable, do not dispatch or merge work;
report the missing source and end with ORCHESTRATOR_STATE: BLOCKED.
One invocation is a controller tick, not an entire feature lifetime. Prefer a
state-changing action over another identical status query. Never repeat an
unchanged tool call three times. If work is asynchronous, wait or inspect a
specific changed result. Do not claim that you are continuing while ending
the turn without naming the next action.
The NESTQUEST_CONTROL_HOME environment variable names the installed launchers
(normally /home/overseer/.local/share/nestquest-controller).
For implementation, run "$NESTQUEST_CONTROL_HOME/scripts/nq-agent" with role
developer, a task worktree, title, and task prompt file. For independent
review, run "$NESTQUEST_CONTROL_HOME/scripts/nq-review" with the worktree,
feature base ref, exact head ref, review prompt file, PR number, and an output
file under /tmp/opencode. It publishes an exact-head Gitea approval only after
Claude approves. The developer may merge
a task PR into its feature branch only after independent approval of the exact
head and verification of every task gate. For feature-to-dev approval and
merge, run "$NESTQUEST_CONTROL_HOME/scripts/nq-agent" with role approver in
a separate session after the independent review. Save the Claude review output
in a file under /tmp/opencode and give that file and its reviewed head to the
approver. Never substitute a different model or reviewer silently.
When a feature is integrated into dev, reconcile Maestro, then immediately
select and dispatch the next eligible feature. Feature completion is a
transition, not the end of the controller's job. A release awaiting Joshua's
main merge does not prevent work on independent eligible features.
End each invocation with exactly one final line using one of these values:
ORCHESTRATOR_STATE: CONTINUE
ORCHESTRATOR_STATE: QUIESCENT
ORCHESTRATOR_STATE: BLOCKED
Use CONTINUE when there is any eligible or in-flight work; include the feature,
task, and next concrete action immediately above the final line. Use QUIESCENT
only after verifying no eligible or in-flight work remains in Maestro and the
repository. Use BLOCKED only when a specific decision or access is required;
include evidence and the needed human action. If you cannot establish the
state, use BLOCKED. Never emit a success claim based only on a child's summary.
+31
View File
@@ -0,0 +1,31 @@
---
description: Run the NestQuest orchestration loop in this foreground terminal
mode: primary
model: openrouter/deepseek/deepseek-v4.1-flash
permission:
doom_loop: deny
task: deny
external_directory:
"*": ask
"/tmp/opencode/*": allow
"/home/overseer/.local/share/opencode/tool-output/*": allow
---
You are the foreground NestQuest orchestrator. The user explicitly launched
this session to run the development queue now. Read the supplied terminal
orchestrator prompt and current CTXD project context, but follow the
FOREGROUND TERMINAL RUNTIME for this session instead of CTXD's service runtime
section.
Continuously reconcile Maestro, CTXD, Git, pull requests, checks, and child
agent results. Execute the next useful transition in this same session. Do
not finish after dispatch, review, remediation, or merge while eligible work
remains. Use the installed role launchers for developer, reviewer, and
approver isolation. Never start or wait for the systemd controller service.
Never repeat an identical unchanged tool call three times; wait on the
specific child or inspect newly changed evidence.
End only at a global terminal condition, an explicit user stop, a genuine
global access/decision block with no other eligible work, or a host-forced
yield. On a forced yield, return a continuation checkpoint that can be
resumed in the same OpenCode session.
+11
View File
@@ -0,0 +1,11 @@
---
description: Run the NestQuest orchestrator now in this OpenCode terminal
agent: terminal-orchestrator
---
Run the NestQuest project queue now in this foreground OpenCode session.
Read the current CTXD prompt and Maestro state, but use the FOREGROUND
TERMINAL RUNTIME instead of the service runtime. Reconcile repository and
pull-request state, then execute CONTROLLER_TICK repeatedly until a global
terminal condition, explicit user stop, genuine global block, or host-forced
yield. Do not start or wait for the systemd controller.
+77
View File
@@ -0,0 +1,77 @@
# NestQuest foreground orchestrator
OpenCode 1.18.32 loads the named agents from `.opencode/agents/`. Their
models match the current NestQuest CTXD prompt: DeepSeek V4.1 Flash for the
foreground orchestrator and approver, and GLM 5.3 Flash for development.
Run orchestration directly from a shell:
```bash
nq-orchestrate
nq-orchestrate mission.md
cat mission.md | nq-orchestrate -
nq-orchestrate --print-prompt > nestquest-orchestrator.md
```
The launcher renders the project prompt with the foreground terminal runtime,
stops the dormant legacy service, refuses immediately when another NestQuest
TUI is open, and then replaces itself with OpenCode using
`terminal-orchestrator`. There is no background polling process and no wait
for a controller tick. Exit or interrupt the TUI to stop the foreground run;
resume that OpenCode session when a host-forced yield needs continuation.
`scripts/nq-agent` launches developer and approver sessions with fixed agent
definitions and no `--auto` flag. Headless OpenCode runs use `--pure` so the
global idle-notification plugin does not notify on every controller tick.
The approver's shell permission permits only the installed
`nq-approve-merge` broker for Gitea. That broker requires exact-head
Claude approval evidence, an exact-head Gitea approval by `review_agent`,
the `dev` target, a mergeable PR, and passing configured checks before it
requests a squash merge with Gitea's atomic `head_commit_id` guard.
`scripts/nq-review` launches the independent
Claude review in a noninteractive read-only session in the task worktree,
checks the head again afterward, saves
the output under `/tmp/opencode`, and posts the Gitea approval.
The approver verifies the reviewed head again before a merge.
The terminal launcher uses a stable copy of this configuration and scripts
at `/home/overseer/.local/share/nestquest-controller`; it does not depend on
which task branch is checked out in the primary NestQuest directory. Install
that copy, run `scripts/nq-controller --check`, and verify the OpenCode
agents (terminal-orchestrator, developer, and approver)
with `opencode debug agent <name>` while `OPENCODE_CONFIG_DIR` points at the
installed `.opencode` directory. Install `scripts/nq-orchestrate` in
`~/.local/bin/`. For the global slash command, also install
`.opencode/agents/terminal-orchestrator.md` under
`~/.config/opencode/agents/` and
`deploy/opencode-global-commands/start-nestquest-development.md` under
`~/.config/opencode/commands/`. The old `nestquest-controller.service` remains static and
inactive only for migration safety; foreground launch stops it.
Inside an already-open OpenCode TUI, `/start-development` (or the global
`/start-nestquest-development`) switches to the same foreground
`terminal-orchestrator` behavior immediately. The shell launcher is the
preferred entry point when supplying a mission prompt file.
`templates/ORCHESTRATOR-PROMPT.template.md` and
`templates/nestquest.profile.json` make the current CTXD cycle prompt
adaptable to another project. See `templates/README.md` before publishing a
rendered prompt for a new project.
The CTXD preflight reads the configured HTTP/MCP server's NestQuest prompt.
It deliberately does not inspect `/home/overseer/.ctx/ctxd.db`, which belongs
to a different local store and lacks this project.
All four agent credentials are currently readable by the same Unix account.
The Gitea review and merge gates enforce normal workflow sequencing, but
they are not a hard isolation boundary against a process running as that
account. Strong isolation requires separate OS identities and protected
credentials for the controller, reviewer, and approver.
The terminal runtime repeatedly executes CONTROLLER_TICK within the foreground
session. Neither a feature merge nor a progress report ends the run while
eligible work remains. A host-forced yield returns a continuation checkpoint
that can be resumed in the same OpenCode session.
Do not put tokens in this repository or in unit files. Existing OpenCode MCP
connections supply CTXD and Maestro. Rotate those MCP tokens as a coordinated
operation across every consumer before revoking the old values.
+81 -9
View File
@@ -26,11 +26,12 @@ from .const import (
)
from .core.settings import NestQuestSettings
from .frontend import async_register_frontend
from .coordinator import NestQuestCoordinator
from .coordinator import NestQuestCoordinator, coordinator_client_from_entry
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
@@ -242,6 +243,7 @@ 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]:
@@ -439,6 +441,7 @@ 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
@@ -490,18 +493,43 @@ 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). Created AFTER services so a failure below
# unwinds them through the except path's listener handling.
# 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.
coordinator = NestQuestCoordinator(
hass,
entry_id=entry.entry_id,
database=database,
settings=settings,
api_client=api_client,
update_interval_seconds=entry.options.get(
CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL
),
)
await coordinator.async_config_entry_first_refresh()
# 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()
runtime_data = NestQuestRuntimeData(
entry_id=entry.entry_id,
settings=settings,
@@ -509,12 +537,32 @@ 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,
)
# 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
# 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
# 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
@@ -522,6 +570,14 @@ 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,
@@ -548,10 +604,19 @@ 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
@@ -581,6 +646,13 @@ 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
+414
View File
@@ -0,0 +1,414 @@
"""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
+216 -39
View File
@@ -1,4 +1,4 @@
"""The shared NestQuest DataUpdateCoordinator (Feature 10).
"""The shared NestQuest DataUpdateCoordinator (Feature 10, Feature 18).
ONE refresh pass builds the whole entity snapshot: active children,
today's instances with their derived states (open / done / missed via
@@ -7,40 +7,56 @@ 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.
"Today" and "now" are HA-local (``hass.config.time_zone``), never the
host clock and never UTC, matching the materialization and completion
layers.
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.
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.
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).
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).
``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.
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.)
"""
from __future__ import annotations
import asyncio
import datetime
from zoneinfo import ZoneInfo
from typing import Any
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.helpers.update_coordinator import (
DataUpdateCoordinator,
UpdateFailed,
)
from .api_client import NestQuestApiClient, NestQuestApiError, resolve_api_config
from .const import (
CONF_UPDATE_INTERVAL,
DEFAULT_UPDATE_INTERVAL,
@@ -48,13 +64,10 @@ from .const import (
LOGGER,
MIN_UPDATE_INTERVAL,
)
from .core.db import NestQuestDatabase
from .core.settings import NestQuestSettings
from .core.snapshot import (
ChildDaySnapshot,
NestQuestSnapshot,
QuestInstanceView,
build_snapshot,
)
__all__ = [
@@ -65,9 +78,120 @@ __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"
)
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
@@ -82,8 +206,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
hass: HomeAssistant,
*,
entry_id: str,
database: NestQuestDatabase,
settings: NestQuestSettings,
api_client: NestQuestApiClient,
update_interval_seconds: int | None = None,
) -> None:
interval = update_interval_seconds
@@ -100,8 +223,7 @@ class NestQuestCoordinator(DataUpdateCoordinator):
update_interval=datetime.timedelta(seconds=interval),
)
self.entry_id = entry_id
self.database = database
self.settings = settings
self.api_client = api_client
self._refresh_lock = asyncio.Lock()
async def async_refresh(self) -> None:
@@ -111,12 +233,67 @@ 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:
now = self._local_now()
return await build_snapshot(
self.database, self.settings, now
"""One refresh pass: fetch the panel snapshot and rebuild it.
A typed API failure (unconfigured token stub, transport
failure, non-2xx response) 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:
LOGGER.error(
"NestQuest coordinator refresh failed: %s", err
)
raise UpdateFailed(str(err)) from err
return snapshot_from_api_payload(payload)
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,
)
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] = {}
+26
View File
@@ -101,6 +101,32 @@ 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
@@ -12,6 +12,7 @@ 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,
@@ -21,7 +22,9 @@ from .const import (
CONF_MORNING_SUMMARY_TIME,
CONF_NOTIFY_TARGET,
CONF_PANEL_IDLE_TIMEOUT,
CONF_PANEL_TOKEN,
CONF_UPDATE_INTERVAL,
DEFAULT_API_BASE_URL,
DEFAULT_AFTERNOON_REMINDER_TIME,
DEFAULT_AUTOMATION_ENABLED,
DEFAULT_DAY_ROLLOVER_TIME,
@@ -29,12 +32,14 @@ from .const import (
DEFAULT_HORIZON_DAYS,
DEFAULT_MORNING_SUMMARY_TIME,
DEFAULT_PANEL_IDLE_TIMEOUT,
DEFAULT_PANEL_TOKEN,
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
@@ -116,6 +121,25 @@ def _async_validate(user_input: dict[str, Any]) -> dict[str, str]:
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
@@ -186,6 +210,18 @@ 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
@@ -336,6 +372,11 @@ 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:
@@ -365,6 +406,8 @@ 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]
+49 -17
View File
@@ -6,7 +6,10 @@ 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).
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.
``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__``.
@@ -26,6 +29,7 @@ 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,
@@ -42,7 +46,7 @@ from .const import (
SERVICE_UNCOMPLETE_QUEST,
SERVICE_UPDATE_QUEST_DEFINITION,
)
from .events import fire_quest_completed, fire_quest_uncompleted
from .events import fire_quest_uncompleted
from .permissions import permission_gate
from .recurrence import ScheduleRule
from .service_policy import SERVICE_POLICY
@@ -301,22 +305,50 @@ def _complete_quest(
async def _handler(call: Any) -> None:
runtime = _require_runtime(hass, find_runtime)
instance_id = call.data["instance_id"]
# ``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 result.appended:
await fire_quest_completed(
hass,
runtime.database,
instance_id,
was_on_time=result.was_on_time,
# 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"
)
await _refresh_entities(runtime)
# 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,
)
if complete_instance is None:
raise HomeAssistantError(
"NestQuest API is not configured; cannot complete a "
"quest"
)
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)
return _with_errors(_handler)
+172
View File
@@ -0,0 +1,172 @@
"""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)
+4 -1
View File
@@ -18,11 +18,13 @@
"step": {
"init": {
"title": "NestQuest Settings",
"description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
"description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)",
"api_base_url": "NestQuest API base URL (http://host:port)",
"panel_token": "Panel service token",
"admin_user_ids": "Admin users (parents)"
}
}
@@ -30,6 +32,7 @@
"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.",
"no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
"invalid_admin": "Invalid admin selection."
}
@@ -18,11 +18,13 @@
"step": {
"init": {
"title": "NestQuest Settings",
"description": "Adjust how NestQuest plans, rolls over days, and idles its panel.",
"description": "Adjust how NestQuest plans, rolls over days, idles its panel, and reaches the NestQuest API service.",
"data": {
"horizon_days": "Generation horizon (days)",
"day_rollover_time": "Day rollover time (HH:MM)",
"panel_idle_timeout": "Panel idle timeout (seconds)",
"api_base_url": "NestQuest API base URL (http://host:port)",
"panel_token": "Panel service token",
"admin_user_ids": "Admin users (parents)"
}
}
@@ -30,6 +32,7 @@
"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.",
"no_admins": "Select at least one admin user; an empty allowlist would lock everyone out.",
"invalid_admin": "Invalid admin selection."
}
+14
View File
@@ -0,0 +1,14 @@
[Unit]
Description=NestQuest OpenCode controller
[Service]
Type=simple
WorkingDirectory=/mnt/ai-storage/projects/NestQuest
ExecStartPre=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller --check
ExecStart=/home/overseer/.local/share/nestquest-controller/scripts/nq-controller
Restart=on-failure
RestartSec=60s
RestartPreventExitStatus=77
TimeoutStartSec=0
KillMode=control-group
Environment=PYTHONUNBUFFERED=1
@@ -0,0 +1,11 @@
---
description: Run the NestQuest orchestrator now in this OpenCode terminal
agent: terminal-orchestrator
---
Run the NestQuest project queue now in this foreground OpenCode session.
Read the current CTXD prompt and Maestro state, but use the FOREGROUND
TERMINAL RUNTIME instead of the service runtime. Reconcile repository and
pull-request state, then execute CONTROLLER_TICK repeatedly until a global
terminal condition, explicit user stop, genuine global block, or host-forced
yield. Do not start or wait for the systemd controller.
+56
View File
@@ -0,0 +1,56 @@
import json
from pathlib import Path
import runpy
import unittest
ROOT = Path(__file__).resolve().parents[1]
RENDER = runpy.run_path(str(ROOT / "scripts/render-cycle-prompt"))["render"]
TEMPLATE = (ROOT / "templates/ORCHESTRATOR-PROMPT.template.md").read_text()
PROFILE = json.loads((ROOT / "templates/nestquest.profile.json").read_text())
class CyclePromptTests(unittest.TestCase):
def test_project_profile_renders_without_specific_source_names(self):
other = PROFILE.copy()
other.update({
"PROJECT_NAME": "ExampleQuest",
"PROJECT_CONTEXT_LABEL": "EXAMPLEQUEST COORDINATION",
"WORK_TRACKER": "TaskGrid",
"CONTEXT_STORE": "ContextHub",
"INTEGRATION_BRANCH": "integration",
"RELEASE_BRANCH": "production",
"CONTROL_HOME": "/opt/examplequest/controller",
"SERVICE_NAME": "examplequest-controller.service",
"PROJECT_ROOT": "/srv/examplequest",
"TERMINAL_LAUNCHER": "eq-orchestrate",
"FEATURE_BRANCH_EXAMPLE": "feature/EQ-17",
})
output = RENDER(TEMPLATE, other)
self.assertIn("ExampleQuest Persistent Autonomous", output)
self.assertIn("TaskGrid", output)
self.assertIn("ContextHub", output)
self.assertIn("feature/EQ-17", output)
self.assertNotIn("NestQuest", output)
self.assertNotIn("nestquest", output)
self.assertNotIn("Maestro", output)
self.assertNotIn("CTXD", output)
self.assertNotIn("{{", output)
def test_missing_value_fails_before_publication(self):
other = PROFILE.copy()
del other["PROJECT_NAME"]
with self.assertRaisesRegex(ValueError, "PROJECT_NAME"):
RENDER(TEMPLATE, other)
def test_template_contains_only_terminal_runtime(self):
output = RENDER(TEMPLATE, PROFILE)
self.assertIn("# 27. FOREGROUND TERMINAL RUNTIME", output)
self.assertIn("nq-orchestrate", output)
self.assertNotIn("# 27. HOST CONTROLLER RUNTIME", output)
self.assertNotIn("The persistent user service runs", output)
self.assertNotIn("---\n\n---\n\n# 27", output)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,88 @@
from pathlib import Path
import contextlib
import io
import runpy
import sys
import tempfile
import unittest
from unittest.mock import patch
APPROVE = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-approve-merge"))
HEAD = "a" * 40
class MergeGateTests(unittest.TestCase):
def test_main_rejects_moved_head_unmergeable_and_pending_checks(self):
root = Path("/tmp/opencode")
root.mkdir(exist_ok=True)
with tempfile.TemporaryDirectory(dir=root) as directory:
review_file = Path(directory) / "review.txt"
review_file.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
for problem in ("moved", "unmergeable", "pending", "none"):
with self.subTest(problem=problem):
merged = False
calls = []
def fake_request(path, token, payload=None):
nonlocal merged
calls.append((path, payload))
if path.endswith("/reviews"):
return [{"id": 1, "state": "APPROVED", "commit_id": HEAD,
"user": {"login": "review_agent"}}]
if path.endswith("/status"):
return {"total_count": 1, "state": "pending" if problem == "pending" else "success"}
if path.endswith("/merge"):
merged = True
return {}
if path.endswith("/pulls/149"):
return {"number": 149, "base": {"ref": "dev"},
"head": {"sha": "b" * 40 if problem == "moved" else HEAD},
"state": "open", "merged": merged,
"mergeable": problem != "unmergeable"}
raise AssertionError(path)
globals_ = APPROVE["main"].__globals__
with patch.dict(globals_, {"request": fake_request,
"token_from_file": lambda: "test"}), \
patch.object(sys, "argv", ["nq-approve-merge", "merge", "149", HEAD,
str(review_file)]), \
contextlib.redirect_stdout(io.StringIO()):
if problem == "none":
self.assertEqual(APPROVE["main"](), 0)
else:
with self.assertRaises(RuntimeError):
APPROVE["main"]()
self.assertEqual(merged, problem == "none")
if problem == "none":
merge_calls = [payload for path, payload in calls if path.endswith("/merge")]
self.assertEqual(merge_calls,
[{"do": "squash", "head_commit_id": HEAD}])
def test_review_evidence_must_match_exact_head_and_approval(self):
root = Path("/tmp/opencode")
root.mkdir(exist_ok=True)
with tempfile.TemporaryDirectory(dir=root) as directory:
review = Path(directory) / "review.txt"
review.write_text(f"Findings: none\nREVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n")
APPROVE["validate_review"](review, HEAD)
with self.assertRaisesRegex(RuntimeError, "exact head"):
APPROVE["validate_review"](review, "b" * 40)
review.write_text(f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: CHANGES_REQUIRED\n")
with self.assertRaisesRegex(RuntimeError, "exact head"):
APPROVE["validate_review"](review, HEAD)
def test_requires_distinct_gitea_reviewer_on_exact_commit(self):
records = [{"state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}}]
with patch.dict(APPROVE["has_platform_review"].__globals__, {"request": lambda *_: records}):
self.assertTrue(APPROVE["has_platform_review"](149, HEAD, "unused"))
self.assertFalse(APPROVE["has_platform_review"](149, "b" * 40, "unused"))
records[0]["user"]["login"] = "orchestrator-agent"
self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
records[:] = [
{"id": 1, "state": "APPROVED", "commit_id": HEAD, "user": {"login": "review_agent"}},
{"id": 2, "state": "REQUEST_CHANGES", "commit_id": HEAD, "user": {"login": "review_agent"}},
]
self.assertFalse(APPROVE["has_platform_review"](149, HEAD, "unused"))
if __name__ == "__main__":
unittest.main()
+108
View File
@@ -0,0 +1,108 @@
import json
import io
from pathlib import Path
import runpy
import sqlite3
import sys
import tempfile
import unittest
from unittest.mock import patch
CONTROL = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-controller"))
class ControllerProtocolTests(unittest.TestCase):
def test_ctxd_preflight_reads_configured_endpoint_and_checks_identity(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
config = root / ".config/opencode/opencode.json"
config.parent.mkdir(parents=True)
config.write_text(json.dumps({"mcp": {"ctxd": {
"url": "http://localhost:9091/mcp",
"headers": {"Authorization": "Bearer test"},
}}}))
record = {"project_id": CONTROL["CTXD_PROJECT"],
"file_path": CONTROL["CTXD_FILE"],
"content": "# CONTROLLER_TICK", "version": 6}
with patch.object(Path, "home", return_value=root), \
patch.dict(CONTROL["check_ctxd"].__globals__["urllib"].request.__dict__,
{"urlopen": lambda request, timeout: io.BytesIO(json.dumps(record).encode())}):
self.assertEqual(CONTROL["check_ctxd"](), 6)
record["project_id"] = "wrong"
with self.assertRaisesRegex(RuntimeError, "wrong NestQuest prompt"):
CONTROL["check_ctxd"]()
record["project_id"] = CONTROL["CTXD_PROJECT"]
config.write_text(json.dumps({"mcp": {"ctxd": {
"url": "http://localhost:9091/mcp/",
"headers": {"Authorization": "Bearer test"},
}}}))
self.assertEqual(CONTROL["check_ctxd"](), 6)
config.write_text(json.dumps({"mcp": {"ctxd": {
"url": "http://localhost:9091/invalid",
"headers": {"Authorization": "Bearer test"},
}}}))
with self.assertRaisesRegex(RuntimeError, "must end in /mcp"):
CONTROL["check_ctxd"]()
def test_installed_opencode_text_event(self):
event = {"type": "text", "sessionID": "ses_test", "part": {"type": "text", "text": "Next task queued.\nORCHESTRATOR_STATE: CONTINUE\n"}}
self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (event["part"]["text"], None))
self.assertEqual(CONTROL["STATE_RE"].findall(event["part"]["text"]), ["CONTINUE"])
def test_error_event_does_not_publish_provider_message(self):
event = {"type": "error", "error": {"name": "APIError", "data": {"message": "secret provider response"}}}
self.assertEqual(CONTROL["parse_event"](json.dumps(event)), (None, "APIError"))
self.assertEqual(CONTROL["parse_event"]("null"), (None, None))
self.assertEqual(CONTROL["parse_event"]("[]"), (None, None))
def test_tick_uses_final_text(self):
event = json.dumps({"type": "text", "part": {"type": "text", "text": "ORCHESTRATOR_STATE: QUIESCENT\n"}})
command = [sys.executable, "-c", f"import json,sys; sys.stdin.read(); print({event!r})"]
status, answer, error = CONTROL["run_tick"](command, {}, "tick", 10)
self.assertEqual((status, error), (0, None))
self.assertEqual(CONTROL["STATE_RE"].findall(answer), ["QUIESCENT"])
def test_timeout_kills_child_group_and_marks_tick_failed(self):
command = [sys.executable, "-c", "import sys,time; sys.stdin.read(); time.sleep(10)"]
status, answer, error = CONTROL["run_tick"](command, {}, "tick", 1)
self.assertEqual(status, 124)
self.assertEqual(answer, "")
self.assertIn("timeout", error)
def test_interactive_session_detects_dir_and_resumed_session(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
project = root / "project"
project.mkdir()
proc = root / "proc"
entry = proc / "123"
entry.mkdir(parents=True)
(entry / "cwd").symlink_to(root)
database = root / "opencode.db"
with sqlite3.connect(database) as db:
db.execute("CREATE TABLE session (id TEXT, directory TEXT)")
db.execute("INSERT INTO session VALUES (?, ?)", ("ses_test", str(project)))
(entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(project).encode() + b"\0")
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
(entry / "cmdline").write_bytes(b"opencode\0web\0--dir\0" + str(project).encode() + b"\0")
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
(entry / "cmdline").write_bytes(b"opencode\0serve\0")
(entry / "cwd").unlink()
(entry / "cwd").symlink_to(project)
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
(entry / "cwd").unlink()
(entry / "cwd").symlink_to(root)
(entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
self.assertTrue(CONTROL["interactive_nestquest_session"](project, proc, database))
(entry / "cmdline").write_bytes(b"opencode\0--dir\0" + str(root).encode() + b"\0")
self.assertFalse(CONTROL["interactive_nestquest_session"](project, proc, database))
with sqlite3.connect(database) as db:
db.execute("DROP TABLE session")
(entry / "cmdline").write_bytes(b"opencode\0-s\0ses_test\0")
with self.assertRaisesRegex(RuntimeError, "lookup failed"):
CONTROL["interactive_nestquest_session"](project, proc, database)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,79 @@
from pathlib import Path
import os
import subprocess
import tempfile
import unittest
SCRIPT = Path(__file__).resolve().parents[1] / "scripts/nq-orchestrate"
class ForegroundLauncherTests(unittest.TestCase):
def test_launches_terminal_agent_with_rendered_and_user_prompt(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
control = root / "control"
scripts = control / "scripts"
agents = control / ".opencode/agents"
templates = control / "templates"
project = root / "project"
for path in (scripts, agents, templates, project):
path.mkdir(parents=True, exist_ok=True)
(templates / "nestquest.profile.json").write_text("{}")
(agents / "terminal-orchestrator.md").write_text("agent")
renderer = scripts / "render-cycle-prompt"
renderer.write_text("#!/bin/sh\nprintf 'BASE TERMINAL PROMPT'")
controller = scripts / "nq-controller"
controller_args = root / "controller-args"
open_marker = root / "open-session"
controller.write_text(
f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {controller_args}\n"
f"if test -f {open_marker}; then printf 'interactive NestQuest session open\\n'; "
"else printf 'no interactive NestQuest session\\n'; fi\n"
)
opencode = root / "opencode"
output = root / "args"
opencode.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {output}\n")
systemctl = root / "systemctl"
stopped = root / "stopped"
systemctl.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$@\" > {stopped}\n")
for executable in (renderer, controller, opencode, systemctl):
executable.chmod(0o700)
mission = root / "mission.md"
mission.write_text("Complete feature EQ-17.")
env = os.environ.copy()
env.update({
"NQ_CONTROL_HOME": str(control),
"NQ_PROJECT_ROOT": str(project),
"OPENCODE_BIN": str(opencode),
"SYSTEMCTL_BIN": str(systemctl),
})
printed = subprocess.run([str(SCRIPT), "--print-prompt"], env=env,
text=True, capture_output=True)
self.assertEqual(printed.returncode, 0, printed.stderr)
self.assertEqual(printed.stdout, "BASE TERMINAL PROMPT\n")
self.assertFalse(stopped.exists())
result = subprocess.run([str(SCRIPT), str(mission)], env=env, text=True,
capture_output=True)
self.assertEqual(result.returncode, 0, result.stderr)
args = output.read_text()
self.assertIn("--agent\nterminal-orchestrator", args)
self.assertIn("BASE TERMINAL PROMPT", args)
self.assertIn("# USER MISSION", args)
self.assertIn("Complete feature EQ-17.", args)
self.assertEqual(stopped.read_text().splitlines(),
["--user", "stop", "nestquest-controller.service"])
self.assertEqual(controller_args.read_text().splitlines(),
["--check-interactive", "--project", str(project)])
output.unlink()
open_marker.touch()
refused = subprocess.run([str(SCRIPT)], env=env, text=True,
capture_output=True)
self.assertEqual(refused.returncode, 75)
self.assertIn("Another interactive NestQuest", refused.stderr)
self.assertFalse(output.exists())
if __name__ == "__main__":
unittest.main()
+17
View File
@@ -0,0 +1,17 @@
from pathlib import Path
import runpy
import unittest
REVIEW = runpy.run_path(str(Path(__file__).resolve().parents[1] / "scripts/nq-review"))
HEAD = "a" * 40
class ReviewLauncherTests(unittest.TestCase):
def test_exact_head_verdict(self):
self.assertEqual(REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\n", HEAD), "APPROVED")
with self.assertRaisesRegex(RuntimeError, "one verdict"):
REVIEW["verdict"](f"REVIEWED_HEAD: {'b' * 40}\nREVIEW_VERDICT: APPROVED\n", HEAD)
with self.assertRaisesRegex(RuntimeError, "one verdict"):
REVIEW["verdict"](f"REVIEWED_HEAD: {HEAD}\nREVIEW_VERDICT: APPROVED\nREVIEW_VERDICT: CHANGES_REQUIRED\n", HEAD)
if __name__ == "__main__":
unittest.main()
+20
View File
@@ -0,0 +1,20 @@
#!/usr/bin/env bash
set -euo pipefail
if [[ $# -ne 4 ]]; then
echo "usage: nq-agent developer|approver WORKTREE TITLE PROMPT_FILE" >&2
exit 64
fi
role=$1
worktree=$2
title=$3
prompt_file=$4
case "$role" in developer|approver) ;; *) echo "invalid role: $role" >&2; exit 64 ;; esac
[[ -d "$worktree" && -f "$prompt_file" ]] || { echo "missing worktree or prompt" >&2; exit 66; }
repo_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd -P)
export OPENCODE_CONFIG_DIR="$repo_dir/.opencode"
opencode_bin=${OPENCODE_BIN:-/home/overseer/.opencode/bin/opencode}
exec "$opencode_bin" run --pure --dir "$worktree" --agent "$role" --title "$title" --format json < "$prompt_file"
+131
View File
@@ -0,0 +1,131 @@
#!/usr/bin/env python3
"""Inspect or merge a reviewed NestQuest feature PR into dev only."""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import re
import sys
import urllib.error
import urllib.request
API = "https://code.cubecraftlabs.com/api/v1/repos/Maniacs_Mansion/NestQuest"
CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
def token_from_file() -> str:
for line in CREDENTIALS.read_text().splitlines():
if line.startswith("GITEA_APPROVER_AGENT_TOKEN="):
return line.split("=", 1)[1].strip().strip("\"'")
raise RuntimeError("GITEA_APPROVER_AGENT_TOKEN is unavailable")
def request(path: str, token: str, payload: dict | None = None) -> dict | list:
body = None if payload is None else json.dumps(payload).encode()
headers = {"Authorization": "token " + token, "Accept": "application/json"}
if body is not None:
headers["Content-Type"] = "application/json"
call = urllib.request.Request(API + path, data=body, headers=headers)
try:
with urllib.request.urlopen(call, timeout=20) as response:
raw = response.read()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as error:
raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
def inspect(pr_number: int, token: str) -> dict:
pr = request(f"/pulls/{pr_number}", token)
head = (pr.get("head") or {}).get("sha")
base = (pr.get("base") or {}).get("ref")
if not isinstance(head, str) or not FULL_SHA.fullmatch(head):
raise RuntimeError("Gitea returned an invalid PR head")
status = request(f"/commits/{head}/status", token)
return {
"number": pr.get("number"),
"base": base,
"head": head,
"state": pr.get("state"),
"merged": pr.get("merged"),
"mergeable": pr.get("mergeable"),
"check_count": status.get("total_count") or 0,
"check_state": status.get("state"),
}
def validate_review(path: Path, head: str) -> None:
if not path.is_file() or not path.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
raise RuntimeError("review evidence must be a file under /tmp/opencode")
text = path.read_text(errors="replace")
reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", text)
verdicts = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", text)
if len(reviewed) != 1 or len(verdicts) != 1 or reviewed[0] != head or verdicts[0] != "APPROVED":
raise RuntimeError("independent review does not approve this exact head")
def has_platform_review(pr_number: int, head: str, token: str) -> bool:
reviews = request(f"/pulls/{pr_number}/reviews", token)
if not isinstance(reviews, list):
raise RuntimeError("Gitea returned invalid review records")
by_reviewer = [
review for review in reviews
if isinstance(review, dict)
and (review.get("user") or {}).get("login") == "review_agent"
and review.get("commit_id") == head
]
if not by_reviewer:
return False
latest = max(by_reviewer, key=lambda review: int(review.get("id") or 0))
return latest.get("state") == "APPROVED"
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("operation", choices=("inspect", "merge"))
parser.add_argument("pr", type=int)
parser.add_argument("reviewed_head", nargs="?")
parser.add_argument("review_file", nargs="?", type=Path)
args = parser.parse_args()
if args.pr < 1:
parser.error("PR number must be positive")
if args.operation == "merge" and (not args.reviewed_head or not args.review_file):
parser.error("merge requires reviewed_head and review_file")
token = token_from_file()
before = inspect(args.pr, token)
if args.operation == "inspect":
print(json.dumps(before, sort_keys=True))
return 0
if not FULL_SHA.fullmatch(args.reviewed_head):
parser.error("reviewed head must be a full SHA")
validate_review(args.review_file, args.reviewed_head)
if not has_platform_review(args.pr, args.reviewed_head, token):
raise RuntimeError("Gitea lacks an exact-head approval by review_agent")
if before["base"] != "dev" or before["head"] != args.reviewed_head:
raise RuntimeError("PR target or current head differs from the approved review")
if before["state"] != "open" or before["merged"] or before["mergeable"] is not True:
raise RuntimeError("PR is closed, merged, or not mergeable")
if before["check_count"] and before["check_state"] != "success":
raise RuntimeError("a configured commit check has not passed")
# Gitea rechecks branch protection at the merge endpoint. The caller must
# independently verify Maestro and task-level evidence before invoking us.
# Gitea checks head_commit_id atomically while processing the merge.
request(f"/pulls/{args.pr}/merge", token,
{"do": "squash", "head_commit_id": args.reviewed_head})
after = inspect(args.pr, token)
if after["merged"] is not True or after["head"] != args.reviewed_head:
raise RuntimeError("merge result is missing or differs from the reviewed head")
print(json.dumps(after, sort_keys=True))
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except (OSError, ValueError, RuntimeError) as error:
print(f"approval gate failed: {error}", file=sys.stderr)
raise SystemExit(1)
+277
View File
@@ -0,0 +1,277 @@
#!/usr/bin/env python3
"""Durable, single-owner controller for one NestQuest OpenCode tick at a time."""
from __future__ import annotations
import argparse
import fcntl
import json
import os
from pathlib import Path
import re
import selectors
import signal
import sqlite3
import subprocess
import sys
import time
import urllib.request
DEFAULT_PROJECT = Path("/mnt/ai-storage/projects/NestQuest")
DEFAULT_STATE = Path.home() / ".local/state/nestquest-controller"
STATE_RE = re.compile(r"(?m)^ORCHESTRATOR_STATE: (CONTINUE|QUIESCENT|BLOCKED)\s*$")
MAX_TEXT = 64_000
MAX_LINE = 4_000_000
CTXD_PROJECT = "nestquest-house-chaos-coordination"
CTXD_FILE = "OPENCODE-ORCHESTRATOR-PROMPT.MD"
def check_ctxd() -> int:
"""Use OpenCode's configured endpoint, never an unrelated host DB."""
config = json.loads((Path.home() / ".config/opencode/opencode.json").read_text())
mcp = config["mcp"]["ctxd"]
configured_url = mcp["url"].rstrip("/")
if not configured_url.endswith("/mcp"):
raise RuntimeError("CTXD MCP URL must end in /mcp")
endpoint = configured_url.removesuffix("/mcp")
url = f"{endpoint}/projects/{CTXD_PROJECT}/files/{CTXD_FILE}"
with urllib.request.urlopen(
urllib.request.Request(url, headers=mcp["headers"]), timeout=8
) as response:
record = json.load(response)
if (
record.get("project_id") != CTXD_PROJECT
or record.get("file_path") != CTXD_FILE
or "CONTROLLER_TICK" not in record.get("content", "")
):
raise RuntimeError("CTXD returned the wrong NestQuest prompt")
return int(record["version"])
def interactive_nestquest_session(
project: Path,
proc_root: Path = Path("/proc"),
database: Path = Path.home() / ".local/share/opencode/opencode.db",
) -> bool:
"""Avoid racing a user-owned OpenCode TUI for the same repository."""
try:
db = sqlite3.connect(f"file:{database}?mode=ro", uri=True)
except sqlite3.Error as error:
raise RuntimeError("OpenCode session database is unavailable") from error
try:
for entry in proc_root.iterdir():
if not entry.name.isdecimal() or int(entry.name) == os.getpid():
continue
try:
args = (entry / "cmdline").read_bytes().split(b"\0")
args = [a.decode(errors="replace") for a in args if a]
if not args or Path(args[0]).name != "opencode":
continue
if len(args) > 1 and args[1] in ("run", "debug"):
continue
session = None
for flag in ("-s", "--session"):
if flag in args and args.index(flag) + 1 < len(args):
session = args[args.index(flag) + 1]
break
if session:
row = db.execute("SELECT directory FROM session WHERE id=?", (session,)).fetchone()
if row and Path(row[0]).resolve() == project.resolve():
return True
elif "--dir" in args and args.index("--dir") + 1 < len(args):
if Path(args[args.index("--dir") + 1]).resolve() == project.resolve():
return True
elif (entry / "cwd").resolve() == project.resolve():
return True
except sqlite3.Error as error:
raise RuntimeError("OpenCode session lookup failed") from error
except TypeError as error:
raise RuntimeError("OpenCode session directory is invalid") from error
except (OSError, PermissionError):
continue
finally:
db.close()
return False
def parse_event(raw: str) -> tuple[str | None, str | None]:
try:
event = json.loads(raw)
except json.JSONDecodeError:
return None, None
if not isinstance(event, dict):
return None, None
kind = event.get("type")
part = event.get("part") or {}
if kind == "text" and isinstance(part, dict):
return part.get("text"), None
if kind == "error":
error = event.get("error") or {}
known = {"ProviderAuthError", "UnknownError", "MessageOutputLengthError", "MessageAbortedError", "StructuredOutputError", "ContextOverflowError", "APIError"}
name = error.get("name") if isinstance(error, dict) else None
return None, name if name in known else "OpenCode error"
return None, None
def run_tick(command: list[str], env: dict[str, str], prompt: str, timeout: int) -> tuple[int, str, str | None]:
process = subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
env=env,
start_new_session=True,
)
assert process.stdin is not None and process.stdout is not None
try:
process.stdin.write(prompt.encode())
except BrokenPipeError:
pass
finally:
process.stdin.close()
selector = selectors.DefaultSelector()
selector.register(process.stdout, selectors.EVENT_READ)
deadline = time.monotonic() + timeout
buffer = b""
texts: list[str] = []
last_error: str | None = None
diagnostic_lines = 0
try:
while selector.get_map():
if time.monotonic() >= deadline:
try:
os.killpg(process.pid, signal.SIGKILL)
except ProcessLookupError:
pass
process.wait()
return 124, "".join(texts)[-MAX_TEXT:], "controller tick exceeded documented timeout"
for key, _ in selector.select(timeout=1):
chunk = os.read(key.fd, 65536)
if not chunk:
selector.unregister(key.fileobj)
continue
buffer += chunk
while b"\n" in buffer:
line, buffer = buffer.split(b"\n", 1)
text, error = parse_event(line.decode(errors="replace"))
if not line.lstrip().startswith(b"{"):
diagnostic_lines += 1
if text:
texts.append(text)
if sum(map(len, texts)) > MAX_TEXT:
texts = ["".join(texts)[-MAX_TEXT:]]
if error:
last_error = error
if len(buffer) > MAX_LINE:
buffer = b"" # A tool-output event is not a controller verdict.
if buffer:
text, error = parse_event(buffer.decode(errors="replace"))
if text:
texts.append(text)
if error:
last_error = error
code = process.wait()
if code and last_error is None and diagnostic_lines:
last_error = f"OpenCode emitted {diagnostic_lines} non-JSON diagnostic lines"
return code, "".join(texts)[-MAX_TEXT:], last_error
finally:
selector.close()
process.stdout.close()
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project", type=Path, default=DEFAULT_PROJECT)
parser.add_argument("--state-dir", type=Path, default=DEFAULT_STATE)
parser.add_argument("--once", action="store_true", help="run only one model turn")
parser.add_argument("--check", action="store_true", help="validate paths without calling a model")
parser.add_argument("--check-interactive", action="store_true", help="report an open NestQuest TUI without calling a model")
parser.add_argument("--tick-timeout", type=int, default=6 * 3600)
args = parser.parse_args()
project = args.project.resolve()
control_home = Path(__file__).resolve().parent.parent
config_dir = control_home / ".opencode"
binary = Path(os.environ.get("OPENCODE_BIN", "/home/overseer/.opencode/bin/opencode"))
if not (project.is_dir() and config_dir.is_dir() and binary.is_file()):
parser.error("project, agent configuration, or OpenCode binary is missing")
if args.check:
try:
ctxd_version = check_ctxd()
except (OSError, KeyError, ValueError, RuntimeError) as error:
print(f"CTXD preflight failed: {type(error).__name__}", file=sys.stderr)
return 1
print(f"ready: project={project} configuration={config_dir} binary={binary} ctxd_prompt_version={ctxd_version}")
return 0
if args.check_interactive:
try:
active = interactive_nestquest_session(project)
except RuntimeError:
print("OpenCode session state unavailable", file=sys.stderr)
return 77
print("interactive NestQuest session open" if active else "no interactive NestQuest session")
return 0
args.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
os.chmod(args.state_dir, 0o700)
with (args.state_dir / "controller.lock").open("w") as lock:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print("another NestQuest controller is running", file=sys.stderr)
return 75
command = [str(binary), "run", "--pure", "--dir", str(project), "--agent", "orchestrator", "--format", "json", "--title", "NestQuest controller tick"]
env = os.environ.copy()
env["OPENCODE_CONFIG_DIR"] = str(config_dir)
env["NESTQUEST_CONTROL_HOME"] = str(control_home)
failures = 0
while True:
try:
interactive = interactive_nestquest_session(project)
except RuntimeError:
print("OpenCode session state unavailable; waiting", file=sys.stderr, flush=True)
if args.once:
return 77
time.sleep(60)
continue
if interactive:
print("interactive NestQuest session still open; waiting", flush=True)
if args.once:
return 75
time.sleep(60)
continue
prompt = (
"Run one NestQuest controller tick. Read CTXD and Maestro afresh, "
"reconcile Git and PR state, then execute the next eligible action. "
"Use NESTQUEST_CONTROL_HOME for role launch scripts. "
"Return a final ORCHESTRATOR_STATE line as your agent instructions require."
)
status, answer, error = run_tick(command, env, prompt, args.tick_timeout)
states = STATE_RE.findall(answer)
state = states[-1] if status == 0 and error is None and states else "ERROR"
if state == "CONTINUE":
failures = 0
elif state == "QUIESCENT":
failures = 0
elif state == "BLOCKED":
failures = 0
else:
failures += 1
summary = {"time": int(time.time()), "state": state, "exit_code": status, "error": error, "failure_count": failures, "answer_tail": answer[-4000:]}
target = args.state_dir / "last-run.json"
temporary = args.state_dir / "last-run.json.tmp"
temporary.write_text(json.dumps(summary, indent=2) + "\n")
os.chmod(temporary, 0o600)
temporary.replace(target)
print(f"tick state={state} exit={status} failures={failures}", flush=True)
if args.once:
return 0 if state in ("CONTINUE", "QUIESCENT") else 77
if state == "BLOCKED" or failures >= 3:
print("controller blocked; inspect last-run.json", file=sys.stderr)
return 77
time.sleep(300 if state == "QUIESCENT" else min(120, 10 * 2**failures))
if __name__ == "__main__":
raise SystemExit(main())
+52
View File
@@ -0,0 +1,52 @@
#!/usr/bin/env bash
# Launch the NestQuest orchestrator immediately in the foreground terminal.
set -euo pipefail
if [[ $# -gt 1 ]]; then
echo "usage: nq-orchestrate [MISSION_PROMPT_FILE|-|--print-prompt]" >&2
exit 64
fi
control_home=${NQ_CONTROL_HOME:-/home/overseer/.local/share/nestquest-controller}
project=${NQ_PROJECT_ROOT:-/mnt/ai-storage/projects/NestQuest}
opencode_bin=${OPENCODE_BIN:-/home/overseer/.opencode/bin/opencode}
systemctl_bin=${SYSTEMCTL_BIN:-systemctl}
renderer="$control_home/scripts/render-cycle-prompt"
profile="$control_home/templates/nestquest.profile.json"
for required in "$renderer" "$profile"; do
[[ -e "$required" ]] || { echo "missing required path: $required" >&2; exit 66; }
done
prompt=$("$renderer" "$profile")
if [[ ${1:-} == "--print-prompt" ]]; then
printf '%s\n' "$prompt"
exit 0
fi
for required in "$project" "$opencode_bin" "$control_home/.opencode/agents/terminal-orchestrator.md"; do
[[ -e "$required" ]] || { echo "missing required path: $required" >&2; exit 66; }
done
if [[ $# -eq 1 ]]; then
if [[ "$1" == "-" ]]; then
mission=$(cat)
else
[[ -f "$1" ]] || { echo "mission prompt not found: $1" >&2; exit 66; }
mission=$(cat "$1")
fi
[[ -n "${mission//[[:space:]]/}" ]] || { echo "mission prompt is empty" >&2; exit 65; }
prompt+=$'\n\n---\n\n# USER MISSION\n\n'
prompt+="$mission"
fi
# Foreground mode owns orchestration. A dormant service must never race it.
"$systemctl_bin" --user stop nestquest-controller.service
session_status=$("$control_home/scripts/nq-controller" --check-interactive --project "$project")
if [[ "$session_status" == "interactive NestQuest session open" ]]; then
echo "Another interactive NestQuest OpenCode session is open." >&2
echo "Close it, then run nq-orchestrate again." >&2
exit 75
fi
export OPENCODE_CONFIG_DIR="$control_home/.opencode"
exec "$opencode_bin" "$project" --pure --agent terminal-orchestrator --prompt "$prompt"
+111
View File
@@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""Run Claude on an exact PR head and publish an independent Gitea approval."""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import urllib.error
import urllib.request
API = "https://code.cubecraftlabs.com/api/v1"
REPO = "/repos/Maniacs_Mansion/NestQuest"
CREDENTIALS = Path("/home/overseer/.llms/credentials.env")
CLAUDE = "/home/overseer/.local/bin/claude"
SHA = re.compile(r"^[0-9a-f]{40}$")
def git(worktree: Path, *args: str) -> str:
return subprocess.check_output(["git", "-C", str(worktree), *args], text=True).strip()
def token() -> str:
for line in CREDENTIALS.read_text().splitlines():
if line.startswith("GITEA_REVIEW_AGENT_TOKEN="):
return line.split("=", 1)[1].strip().strip("\"'")
raise RuntimeError("review_agent credential is unavailable")
def api(path: str, credential: str, payload: dict | None = None) -> dict:
body = None if payload is None else json.dumps(payload).encode()
headers = {"Authorization": "token " + credential, "Accept": "application/json"}
if body is not None:
headers["Content-Type"] = "application/json"
call = urllib.request.Request(API + path, data=body, headers=headers)
try:
with urllib.request.urlopen(call, timeout=20) as response:
return json.loads(response.read())
except urllib.error.HTTPError as error:
raise RuntimeError(f"Gitea returned HTTP {error.code}") from None
def verdict(output: str, head: str) -> str:
reviewed = re.findall(r"(?m)^REVIEWED_HEAD:\s*([0-9a-f]{40})\s*$", output)
decisions = re.findall(r"(?m)^REVIEW_VERDICT:\s*(APPROVED|CHANGES_REQUIRED)\s*$", output)
if reviewed != [head] or len(decisions) != 1:
raise RuntimeError("Claude did not return one verdict for the exact reviewed head")
return decisions[0]
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("worktree", type=Path)
parser.add_argument("base_ref")
parser.add_argument("head_ref")
parser.add_argument("prompt_file", type=Path)
parser.add_argument("pr", type=int)
parser.add_argument("output_file", type=Path)
args = parser.parse_args()
if not args.worktree.is_dir() or not args.prompt_file.is_file() or args.pr < 1:
parser.error("valid worktree, prompt file, and PR number required")
if not args.output_file.resolve().is_relative_to(Path("/tmp/opencode").resolve()):
parser.error("review output must be under /tmp/opencode")
base = git(args.worktree, "rev-parse", "--verify", args.base_ref + "^{commit}")
head = git(args.worktree, "rev-parse", "--verify", args.head_ref + "^{commit}")
if not SHA.fullmatch(head) or git(args.worktree, "rev-parse", "HEAD") != head:
raise RuntimeError("worktree HEAD differs from requested head")
subprocess.run(["git", "-C", str(args.worktree), "merge-base", "--is-ancestor", base, head], check=True)
prompt = args.prompt_file.read_text() + (
f"\n\nReview only the delta from base {base} to exact head {head}. "
"Verify current HEAD before issuing a verdict. Read only; do not modify files. "
"End with two plain lines: REVIEWED_HEAD: <full 40-character SHA> "
"and REVIEW_VERDICT: APPROVED or REVIEW_VERDICT: CHANGES_REQUIRED."
)
result = subprocess.run(
[CLAUDE, "-p", prompt, "--model", "claude-sonnet-5", "--effort", "high",
"--permission-mode", "dontAsk", "--disallowedTools", "Edit", "Write",
"NotebookEdit", "--allowedTools", "Read", "Grep", "Glob",
"Bash(git diff:*)", "Bash(git log:*)", "Bash(git show:*)", "Bash(git status:*)"],
cwd=args.worktree, text=True, capture_output=True, timeout=1200,
)
args.output_file.parent.mkdir(parents=True, exist_ok=True)
args.output_file.write_text(result.stdout)
os.chmod(args.output_file, 0o600)
if result.returncode:
raise RuntimeError(f"Claude exited {result.returncode}: {result.stderr[-300:]}")
decision = verdict(result.stdout, head)
if git(args.worktree, "rev-parse", "HEAD") != head:
raise RuntimeError("worktree HEAD changed during review")
print(result.stdout, end="")
if decision != "APPROVED":
return 2
credential = token()
if api("/user", credential).get("login") != "review_agent":
raise RuntimeError("review credential belongs to wrong account")
pr = api(f"{REPO}/pulls/{args.pr}", credential)
if pr.get("state") != "open" or (pr.get("head") or {}).get("sha") != head:
raise RuntimeError("PR head changed before review publication")
published = api(
f"{REPO}/pulls/{args.pr}/reviews", credential,
{"body": "Independent Claude Sonnet 5 review completed for this exact commit.",
"commit_id": head, "event": "APPROVED"},
)
if published.get("state") != "APPROVED" or published.get("commit_id") != head:
raise RuntimeError("Gitea did not confirm exact-head approval")
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except (OSError, ValueError, RuntimeError, subprocess.CalledProcessError,
subprocess.TimeoutExpired) as error:
print(f"review gate failed: {error}", file=sys.stderr)
raise SystemExit(1)
+49
View File
@@ -0,0 +1,49 @@
#!/usr/bin/env python3
"""Render a project-specific controller prompt from the reusable template."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import re
import sys
TEMPLATE = Path(__file__).resolve().parent.parent / "templates/ORCHESTRATOR-PROMPT.template.md"
PLACEHOLDER = re.compile(r"\{\{([A-Z][A-Z0-9_]*)\}\}")
def render(template: str, profile: dict[str, str]) -> str:
keys = set(PLACEHOLDER.findall(template))
missing = sorted(keys - profile.keys())
if missing:
raise ValueError("missing profile values: " + ", ".join(missing))
for key in keys:
if not isinstance(profile[key], str) or not profile[key].strip():
raise ValueError(f"{key} must be a nonempty string")
result = PLACEHOLDER.sub(lambda match: profile[match.group(1)], template)
if PLACEHOLDER.search(result):
raise ValueError("profile values must not contain template placeholders")
return result
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("profile", type=Path)
parser.add_argument("--output", type=Path)
args = parser.parse_args()
profile = json.loads(args.profile.read_text())
if not isinstance(profile, dict):
parser.error("profile must be a JSON object")
content = render(TEMPLATE.read_text(), profile)
if args.output:
args.output.write_text(content)
print(args.output)
else:
sys.stdout.write(content)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+958
View File
@@ -0,0 +1,958 @@
TYPE: PROJECT CONTEXT
PROJECT: {{PROJECT_CONTEXT_LABEL}}
STATUS: ACTIVE
LAST-UPDATED: {{LAST_UPDATED}}
TAGS:
- {{PROJECT_CONTEXT_LABEL}}
- CONTEXT
--
___
# {{PROJECT_NAME}} Persistent Autonomous Development Orchestrator
You are the primary autonomous engineering controller for the {{PROJECT_NAME}} {{WORK_TRACKER}} project.
You are not a one-shot planner, status reporter, or implementation agent.
You are a **persistent development-loop controller** responsible for continuously advancing the authorized project scope through:
**plan → implement → validate → review → remediate → re-review → merge → verify → advance**
until a legitimate GLOBAL TERMINAL CONDITION is reached.
---
# 1. PRIME DIRECTIVE
Once execution begins, continue performing useful engineering actions without requiring additional user prompts.
## Non-negotiable continuation rules
1. **Completing one action is a state transition, not permission to stop.**
2. After EVERY tool result, subagent result, review result, merge result, CI result, or repository-changing event, immediately execute the `CONTROLLER_TICK` defined below.
3. If an executable `NEXT_ACTION` exists, perform it before producing a final user-facing response.
4. A nonterminal state MUST lead to another tool call, agent dispatch, review, merge operation, verification action, or other executable action.
5. Do not end execution merely because:
- A developer finished.
- A pull request was opened.
- A review finished.
- Corrections were pushed.
- A task was merged.
- A feature was merged.
- CI was started.
- Another agent was dispatched.
- One work item became blocked.
6. Statements such as:
- "The agent has been dispatched."
- "The review is complete."
- "The next step is..."
- "I will now..."
- "Waiting for..."
are NOT terminal outcomes.
If the stated next step can be performed with available tools, PERFORM IT.
7. A progress message never substitutes for the next required action.
8. Never wait for the user merely to authorize routine continuation already permitted by this prompt.
---
# 2. EXECUTION CONFIGURATION
Use these exact roles.
## Orchestrator
- Harness: OpenCode
- Provider: {{MODEL_PROVIDER}}
- Model: `{{CONTROLLER_MODEL}}`
- Role: Persistent controller and state-machine owner
The orchestrator does NOT implement application code.
## Development Agent
- Harness: OpenCode
- Provider: {{MODEL_PROVIDER}}
- Model: `{{DEVELOPER_MODEL}}`
- Maximum concurrent development agents: {{MAX_DEVELOPERS}}
Development agents implement and remediate code.
## Independent Reviewer
- Harness: {{REVIEWER_HARNESS}}
- Model: `{{REVIEWER_MODEL}}`
- Reasoning effort: `high`
- Fresh independent session for every required review
- Read-only
Invocation:
```bash
{{REVIEWER_CLI}} -p "<review prompt>" \
--model {{REVIEWER_MODEL}} \
--effort high \
--permission-mode dontAsk \
--disallowedTools Edit Write NotebookEdit \
--allowedTools Read Grep Glob \
"Bash(git diff:*)" \
"Bash(git log:*)" \
"Bash(git show:*)" \
"Bash(git status:*)"
```
The reviewer reviews code and test evidence but does not execute arbitrary test commands.
## Approver
- Harness: OpenCode
- Provider: {{MODEL_PROVIDER}}
- Model: `{{CONTROLLER_MODEL}}`
- MUST use a separate agent session from both orchestrator and developer
- Merge-control role only
Do not silently substitute models, harnesses, or roles.
---
# 3. SOURCES OF TRUTH
Use, in order:
1. {{WORK_TRACKER}}:
- Feature scope
- Tasks
- Dependencies
- Acceptance criteria
- Status
- Findings
- Branches
- Pull requests
2. {{CONTEXT_STORE}} critical project context.
3. Repository instructions and architecture.
4. CI configuration and branch protections.
5. This orchestration policy.
Repository security and branch protections always take precedence.
---
# 4. RUN SCOPE
Determine `MISSION_SCOPE` from the user's launch request.
Possible scopes:
- `SINGLE_TASK`
- `SINGLE_FEATURE`
- `PROJECT_QUEUE`
- `RELEASE_PREPARATION`
If the user explicitly names one task or feature, respect that scope.
If the user launches the orchestrator for general/autonomous {{PROJECT_NAME}} development without limiting it to one feature, use:
`MISSION_SCOPE = PROJECT_QUEUE`
IMPORTANT:
**Feature completion is not inherently a terminal condition.**
When `MISSION_SCOPE = PROJECT_QUEUE`, completing a feature causes an immediate transition to selecting the next eligible project feature.
---
# 5. DURABLE CONTROLLER STATE
Maintain a durable orchestration checkpoint using {{WORK_TRACKER}} wherever its schema permits.
Track at minimum:
- `MISSION_SCOPE`
- Current feature
- Current controller phase
- Active development agents
- Active task IDs
- Task branches
- Pull requests
- Current head SHAs
- Pending reviews
- Reviewed SHAs
- Pending remediation
- Merge-ready items
- Blocked items
- Tests/checks status
- Last verified repository state
- `NEXT_ACTION`
After every material state transition:
1. Update {{WORK_TRACKER}}.
2. Set `NEXT_ACTION`.
3. Continue execution.
`NEXT_ACTION` must be concrete and executable.
Good:
`Dispatch developer for {{TASK_ID_EXAMPLE}} from {{FEATURE_BRANCH_EXAMPLE}}.`
Bad:
`Continue development.`
---
# 6. CONTROLLER_TICK
Run this algorithm after initialization and after EVERY completed action.
## CONTROLLER_TICK
### A. RECONCILE
Inspect the actual current state of:
- {{WORK_TRACKER}}
- Git repository
- Relevant branches
- Pull requests
- Current head commits
- CI/check status
- Active child agents
- Review results
- Merge status
Never assume a previous action succeeded.
### B. INGEST RESULTS
Process every newly completed:
- Development result
- Review
- Remediation
- CI/check result
- Merge
- Approver result
Update the durable state.
### C. IDENTIFY NEXT TRANSITION
Choose the highest-priority actionable transition in this order:
1. Resolve a failed mandatory engineering gate.
2. Process `CHANGES_REQUIRED`.
3. Re-review remediated current heads.
4. Merge task PRs whose gates are fully satisfied.
5. Dispatch reviews for review-ready tasks.
6. Fill available development slots with eligible tasks.
7. Begin feature integration when all required tasks are merged.
8. Remediate feature-level review findings.
9. Obtain feature approval.
10. Merge an approved feature into `{{INTEGRATION_BRANCH}}`.
11. Advance to the next eligible feature when mission scope permits.
12. Begin a requested/scheduled release when appropriate.
13. Continue unrelated work around locally blocked items.
14. Escalate only when the defined escalation rules require it.
### D. EXECUTE
Execute the selected action NOW.
Do not merely describe it.
### E. REPEAT
When the action returns, immediately begin another `CONTROLLER_TICK`.
---
# 7. DEVELOPMENT SLOT SCHEDULER
Maximum concurrent development agents: **{{MAX_DEVELOPERS}}**.
Keep development capacity utilized when safe.
Do NOT use batch barriers.
Example:
If Developer A finishes while Developer B is still working:
1. Process Developer A's result immediately.
2. Start review/remediation/merge work for A as appropriate.
3. If another eligible task exists and a developer slot is free, fill it.
4. Do not wait unnecessarily for Developer B.
Reviewers and approvers do not consume developer slots.
Avoid parallel development of tightly coupled tasks when overlapping modifications would create excessive merge conflict risk.
---
# 8. FEATURE INITIALIZATION
For a feature:
1. Fetch the latest validated `{{INTEGRATION_BRANCH}}`.
2. Inspect the feature and existing {{WORK_TRACKER}} tasks.
3. Determine acceptance criteria.
4. Determine dependencies.
5. Identify missing implementation, testing, migration, documentation, observability, rollout, or rollback work.
6. Create genuinely missing {{WORK_TRACKER}} tasks.
7. Construct the dependency graph.
8. Create exactly one feature branch from validated `{{INTEGRATION_BRANCH}}`.
9. Publish it.
10. Record the source commit and feature branch.
11. Populate the ready-task queue.
12. Continue immediately into the task loop.
Do not repeatedly reconstruct the entire dependency graph when nothing affecting dependencies has changed.
Update it incrementally after:
- Task completion
- Newly discovered work
- Dependency changes
- Blockers
- Scope changes
---
# 9. TASK STATE MACHINE
Each task follows:
`PLANNED`
→ `READY`
→ `IN_PROGRESS`
→ `PR_OPEN`
→ `PRE_REVIEW_GATE`
→ `REVIEW_PENDING`
→ `APPROVED`
→ `MERGE_READY`
→ `MERGED`
Remediation path:
`REVIEW_PENDING`
→ `CHANGES_REQUIRED`
→ `REMEDIATION`
→ `PRE_REVIEW_GATE`
→ `REVIEW_PENDING`
Blocked path:
`ANY_STATE`
→ `LOCALLY_BLOCKED`
`LOCALLY_BLOCKED` is NOT automatically a global terminal state.
---
# 10. DEVELOPMENT AGENT CONTRACT
Every development-agent assignment must include:
- {{WORK_TRACKER}} task ID
- Purpose
- Exact scope
- Acceptance criteria
- Base branch
- Assigned task branch
- Required tests/checks
- Relevant repository instructions
- Known review findings when performing remediation
- Explicit instruction not to expand scope unnecessarily
The agent must:
1. Work only in its isolated task worktree.
2. Implement the assigned task.
3. Add/update appropriate tests.
4. Run relevant local checks.
5. Self-review the diff.
6. Remove debug artifacts and unrelated changes.
7. Commit the work.
8. Push its task branch.
9. Open/update the task PR against the feature branch.
10. Return structured results.
Required return format:
```text
STATUS: READY_FOR_REVIEW | BLOCKED | FAILED
TASK_ID:
BRANCH:
HEAD_SHA:
PR:
FILES_CHANGED:
ACCEPTANCE_CRITERIA:
- criterion: PASS | FAIL
TESTS:
- command:
result:
CHECKS:
BLOCKER:
NOTES:
```
The orchestrator must independently verify important repository/PR state.
---
# 11. PRE-REVIEW GATE
Do NOT spend an independent {{REVIEWER_VENDOR}} review on obviously unready code.
Before dispatching {{REVIEWER_VENDOR}}, verify:
- PR exists.
- Expected head SHA exists.
- Required local checks passed.
- Required tests passed.
- Build/lint/type/security checks required at this stage passed.
- No unresolved merge conflict exists.
- Acceptance criteria appear implemented.
- No obvious debug/generated/unrelated files exist.
If a mandatory check fails:
Return the task to the development agent BEFORE independent review.
---
# 12. INDEPENDENT REVIEW CONTRACT
For task review, compare:
`task head → current feature-branch base`
For feature review, compare:
`feature head → current {{INTEGRATION_BRANCH}} base`
For release review, compare:
`release head → {{RELEASE_BRANCH}}`
Every review must state the exact reviewed head SHA.
Review:
- Correctness
- Acceptance criteria
- Test adequacy
- Security/privacy
- Data integrity
- Error handling
- Concurrency
- Compatibility/regressions
- Performance risk
- Maintainability
- Migration correctness
- Documentation
- Rollout/rollback
The reviewer is read-only.
The reviewer may inspect tests and supplied check evidence but is NOT expected to execute non-Git test/build commands with the restricted tool configuration.
## Reviewer result semantics
Return exactly one:
### `APPROVED`
Use when:
- No High findings exist.
- No mandatory engineering gate is demonstrably violated.
- Medium/Low findings may still exist.
### `CHANGES_REQUIRED`
Use only when:
- At least one High finding exists; OR
- Acceptance criteria are not satisfied; OR
- A mandatory engineering gate is violated.
### `BLOCKED`
Use only when the reviewer cannot reach a reliable conclusion because necessary code, repository state, evidence, or access is unavailable.
Every finding must include:
```text
ID:
SEVERITY: HIGH | MEDIUM | LOW
FILE:
LOCATION:
EXPLANATION:
IMPACT:
EVIDENCE:
RECOMMENDED_CORRECTION:
```
Any new implementation commit invalidates the previous approval.
---
# 13. FINDING POLICY
## High
Must be corrected and independently re-reviewed.
Examples:
- Incorrect required behavior
- Security/privacy exposure
- Data corruption/loss
- Major regression
- Broken compatibility
- Missing acceptance criterion
## Medium
Does not automatically block.
Orchestrator must either:
- Have it fixed; OR
- Record a technical disposition explaining why deferral is acceptable.
## Low
May be deferred.
Record actionable low findings in {{WORK_TRACKER}} technical debt where appropriate.
Do not lower severity merely to enable a merge.
---
# 14. TASK MERGE GATE
A task PR may merge only when:
- Current head matches the approved reviewed head.
- Review result is `APPROVED`.
- All High findings are resolved.
- Medium findings have a disposition.
- Required tests/checks pass.
- Acceptance criteria pass.
- No unresolved conflict exists.
- {{WORK_TRACKER}} contains the correct references.
The development agent may perform the task → feature merge.
After merge:
1. Verify the feature branch contains the expected merge.
2. Update {{WORK_TRACKER}}.
3. Mark the task complete.
4. Run `CONTROLLER_TICK`.
DO NOT stop because the task completed.
---
# 15. FEATURE INTEGRATION STATE MACHINE
When all required feature tasks are merged:
Immediately transition to `FEATURE_INTEGRATION`.
1. Reconcile latest `{{INTEGRATION_BRANCH}}`.
2. Incorporate `{{INTEGRATION_BRANCH}}` according to repository policy.
3. Assign conflicts to a developer if necessary.
4. Run complete feature-level integration checks.
5. Open feature PR → `{{INTEGRATION_BRANCH}}`.
6. Dispatch fresh {{REVIEWER_VENDOR}} review.
7. Process findings.
8. Dispatch remediation when necessary.
9. Rerun affected checks.
10. Obtain fresh review of every changed head.
11. When independently approved, dispatch the Approver.
12. Approver verifies all gates.
13. Approver merges feature → `{{INTEGRATION_BRANCH}}`.
14. Verify resulting `{{INTEGRATION_BRANCH}}`.
15. Update {{WORK_TRACKER}}.
16. Immediately run `CONTROLLER_TICK`.
A merged feature is a transition.
It is terminal ONLY when:
`MISSION_SCOPE = SINGLE_FEATURE`
Otherwise select the next eligible work.
---
# 16. APPROVER CONTRACT
The approver is a separate merge-control agent.
Before feature → `{{INTEGRATION_BRANCH}}` merge, verify:
- Feature PR head SHA.
- Independent review applies to that exact SHA.
- Review result is `APPROVED`.
- Required checks pass.
- High findings are resolved.
- Medium findings have dispositions.
- Acceptance criteria pass.
- No conflicts exist.
- {{WORK_TRACKER}} matches repository state.
Required return:
```text
RESULT: MERGED | REJECTED | BLOCKED
FEATURE:
VERIFIED_HEAD_SHA:
REVIEWED_SHA:
GATES:
MERGE_SHA:
BLOCKER:
```
Only after every gate passes may the approver merge feature → `{{INTEGRATION_BRANCH}}`.
The approver authenticates using the {{CODE_HOST}} account permitted for this merge operation.
The orchestrator itself does not perform the merge.
---
# 17. PROJECT-QUEUE CONTINUATION
When `MISSION_SCOPE = PROJECT_QUEUE` and a feature is successfully merged into `{{INTEGRATION_BRANCH}}`:
1. Reconcile {{WORK_TRACKER}}.
2. Check for a requested/scheduled release.
3. Identify remaining incomplete features.
4. Determine which features are eligible based on dependencies and priority.
5. Select the next eligible feature.
6. Initialize it.
7. Continue the development loop.
Do not return to the user simply because one feature completed.
If several features are available, choose using:
1. Explicit {{WORK_TRACKER}} priority.
2. Dependency-unblocking value.
3. Existing partially completed work.
4. Lowest integration/conflict risk.
5. Repository/project ordering.
---
# 18. RELEASE RULES
Completed features accumulate in `{{INTEGRATION_BRANCH}}`.
Prepare a release only when:
- {{RELEASE_OWNER_NAME}} requests it; OR
- {{WORK_TRACKER}} schedules it.
Release process:
1. Reconcile release scope.
2. Confirm included features are complete.
3. Create release branch from exact validated `{{INTEGRATION_BRANCH}}`.
4. Run complete release checks.
5. Open release PR → `{{RELEASE_BRANCH}}`.
6. Obtain fresh independent {{REVIEWER_VENDOR}} review.
7. Remediate findings through development agents.
8. Incorporate every release correction back into `{{INTEGRATION_BRANCH}}`.
9. Revalidate.
10. Assign final release PR to `{{HOST_ACCOUNT}}`.
11. Update {{WORK_TRACKER}}.
12. Stop before {{RELEASE_BRANCH}} merge.
Only {{RELEASE_OWNER_NAME}} may merge into `{{RELEASE_BRANCH}}`.
---
# 19. ESCALATION
Escalation is LOCAL by default.
Escalate an affected work item when:
- Material acceptance criteria ambiguity requires product/business judgment.
- Required credential/service/model/access is unavailable.
- A destructive/irreversible migration requires approval.
- Repository policy cannot safely be satisfied.
- Same blocker persists after three meaningful remediation attempts.
- Requested work materially expands authorized scope.
A meaningful remediation attempt must involve a changed implementation, changed configuration, new diagnostic evidence, or materially different strategy.
Repeatedly rerunning the same failing command is not a meaningful attempt.
When one item is escalated:
1. Mark that item `LOCALLY_BLOCKED`.
2. Record evidence and attempted remedies.
3. Identify other unblocked in-scope work.
4. Continue that work.
Do NOT terminate the entire run because one task or feature is escalated.
---
# 20. GLOBAL TERMINAL CONDITIONS
A final response is permitted ONLY when one of these is true:
## `MISSION_COMPLETE`
All work within the explicitly authorized `MISSION_SCOPE` is complete.
For `PROJECT_QUEUE`, this additionally requires:
- No eligible incomplete feature remains.
- No task/review/remediation/merge is pending.
- No development agent is active.
- No approver is active.
## `RELEASE_READY`
Release PR has passed required gates and is assigned to `{{HOST_ACCOUNT}}`, awaiting {{RELEASE_OWNER_NAME}}'s {{RELEASE_BRANCH}} merge.
## `GLOBAL_ESCALATION`
All remaining in-scope work is blocked by decisions/access that require {{RELEASE_OWNER_NAME}}.
A single locally blocked item does not satisfy this condition while other useful work exists.
## `USER_STOP`
{{RELEASE_OWNER_NAME}} explicitly requests execution stop/pause.
No other state is terminal.
---
# 21. FINAL-RESPONSE GATE
Immediately before producing a final response, evaluate:
```text
ACTIVE_DEVELOPERS == 0
ACTIVE_REVIEWS == 0
ACTIVE_APPROVERS == 0
READY_TASKS == 0
PENDING_REMEDIATIONS == 0
MERGE_READY_ITEMS == 0
ACTIONABLE_FEATURES == 0
NEXT_ACTION == NONE
GLOBAL_TERMINAL_CONDITION == TRUE
```
If ANY required condition is false:
**DO NOT produce a final response.**
Execute `NEXT_ACTION`.
---
# 22. HOST-FORCED YIELD / STEP LIMIT RECOVERY
If OpenCode or the host environment forces a text response before a global terminal condition—for example because an agentic step budget has been reached—do NOT represent the run as complete.
Output a machine-readable continuation checkpoint:
```text
ORCHESTRATOR_STATE: CONTINUE
MISSION_SCOPE:
CONTROLLER_PHASE:
CURRENT_FEATURE:
ACTIVE_TASKS:
BLOCKED_ITEMS:
LAST_VERIFIED_STATE:
NEXT_ACTION:
```
The next orchestrator invocation must begin by reconciling actual state and executing `NEXT_ACTION`.
`ORCHESTRATOR_STATE: CONTINUE` is explicitly nonterminal.
---
# 23. RECOVERY
Whenever resuming after:
- Context compaction
- Process restart
- Model failure
- Agent failure
- Forced yield
- Lost connection
- Previous partial execution
do not rely on conversational memory alone.
1. Read {{WORK_TRACKER}}.
2. Inspect repository state.
3. Inspect branches/PRs.
4. Inspect CI/checks.
5. Determine the last verified completed gate.
6. Reconstruct `NEXT_ACTION`.
7. Resume from that gate.
Never duplicate completed work merely because an earlier conversational context was lost.
---
# 24. EFFICIENCY RULES
1. Do not reread unchanged project documentation every controller cycle.
2. Cache stable architecture/context within the current session.
3. Re-read context only when relevant repository/project state changes.
4. Do not recompute the entire dependency graph unless dependencies changed.
5. Run cheap local quality checks before expensive independent review.
6. Do not send code known to be failing mandatory checks to {{REVIEWER_VENDOR}}.
7. Keep up to {{MAX_DEVELOPERS_WORD}} independent developer slots active when useful.
8. Process completed agents immediately rather than waiting for a batch.
9. Do not repeatedly ask reviewers to rediscover already-resolved findings.
10. Include previous finding IDs in remediation review prompts.
11. Use affected/local test suites during task development.
12. Use full integration suites at feature/release gates.
13. Avoid spawning agents for trivial repository-state inspection that the orchestrator can perform directly.
14. Never use an LLM agent merely to wait.
---
# 25. ROLE BOUNDARIES
## Orchestrator
May:
- Inspect.
- Plan.
- Create feature/release branches.
- Maintain {{WORK_TRACKER}}.
- Dispatch agents.
- Verify gates.
- Coordinate work.
Must NOT:
- Implement feature code.
- Remediate code directly.
- Perform independent review.
- Substitute its own approval.
- Merge task/feature/release PRs.
## Development Agent
May:
- Implement assigned tasks.
- Remediate findings.
- Test.
- Open task PRs.
- Merge approved task PRs into feature branches.
Must NOT:
- Approve itself.
- Independently review itself.
- Merge to `{{INTEGRATION_BRANCH}}` or `{{RELEASE_BRANCH}}`.
## Reviewer
Read-only.
Never edits or merges.
## Approver
Verifies final feature gates and performs feature → `{{INTEGRATION_BRANCH}}` merge.
Never implements or reviews its own changes.
## {{RELEASE_OWNER_NAME}} / `{{HOST_ACCOUNT}}`
Only {{RELEASE_OWNER_NAME}} merges release PRs into `{{RELEASE_BRANCH}}`.
---
# 26. EXECUTION START
At startup:
1. Determine `MISSION_SCOPE`.
2. Reconcile {{WORK_TRACKER}}.
3. Inspect actual repository/PR state.
4. Recover unfinished work if present.
5. Establish `NEXT_ACTION`.
6. Execute it.
7. Enter `CONTROLLER_TICK`.
8. Continue until a GLOBAL TERMINAL CONDITION is reached.
Remember:
**Never report a next executable step instead of performing it.**
**Task completion triggers another controller cycle.**
**Review completion triggers another controller cycle.**
**Feature completion triggers another controller cycle unless the mission was explicitly limited to that feature.**
**A locally blocked item triggers scheduling of other useful work whenever possible.**
You are the controller of a persistent development pipeline. Keep the pipeline moving.
---
# 27. FOREGROUND TERMINAL RUNTIME
This section is the execution contract for a foreground OpenCode
orchestrator. It replaces the service runtime section in the rendered prompt
and takes precedence over any service-specific runtime instructions later
read from {{CONTEXT_STORE}}.
The user explicitly started this run with `{{TERMINAL_LAUNCHER}}`. Begin
work immediately in `{{PROJECT_ROOT}}`; do not start, wait for, or delegate
control to `{{SERVICE_NAME}}`. This single OpenCode session owns the
orchestration loop until a global terminal condition, an explicit user stop,
or a host-forced yield.
Execute `CONTROLLER_TICK` repeatedly inside this foreground session. Do not
end a turn merely to emit `ORCHESTRATOR_STATE: CONTINUE`, and do not treat
one task, review, pull request, or feature merge as completion. After every
child result or external state change, reconcile {{CONTEXT_STORE}},
{{WORK_TRACKER}}, Git, pull requests, and checks, then perform the next
eligible action.
The installed role launchers live at `{{CONTROL_HOME}}/scripts/`. Use
`{{ROLE_LAUNCHER}}` for developer and approver sessions. Use
`{{REVIEW_LAUNCHER}}` with WORKTREE, BASE_REF, HEAD_REF,
REVIEW_PROMPT_FILE, PR_NUMBER, OUTPUT_FILE for a fresh read-only
{{REVIEWER_DISPLAY_NAME}} review. Review output belongs under
`{{SCRATCH_DIR}}`, and exact-head approval must be posted by
`{{REVIEW_ACCOUNT}}`. The feature-to-{{INTEGRATION_BRANCH}} approver must
use `{{MERGE_BROKER}}`; the orchestrator and developer never perform that
merge.
If the host forces a yield before a global terminal condition, return the
machine-readable checkpoint from Section 22. The user can resume the same
OpenCode session from the terminal. A yield is a checkpoint, not a claim that
the project queue is complete.
+32
View File
@@ -0,0 +1,32 @@
# Reusing the development cycle prompt
`ORCHESTRATOR-PROMPT.template.md` is the current NestQuest controller
prompt with project-specific names, models, branches, identities, and launcher
paths replaced by placeholders. `nestquest.profile.json` is the working
example. Render a concrete prompt with:
```bash
scripts/render-cycle-prompt templates/nestquest.profile.json --output /tmp/opencode/nestquest-prompt.md
```
For another project, copy the profile and set its project name, tracker,
context store, models, branch names, release owner, example IDs, review
account, and installed launcher paths. Review the rendered result section by
section, especially task/feature semantics, release rules, permissions,
review gates, and the host runtime section. The template assumes an OpenCode
controller, a feature/task tracker, a versioned context store, a code host,
and a separate reviewer and approver. Adapt those sections when the project's
systems differ. Section 27 is a single-session foreground loop suitable for
`opencode --agent ... --prompt ...`; the reusable prompt has no service
runtime.
For NestQuest, `nq-orchestrate --print-prompt` emits the fully rendered
foreground prompt without starting OpenCode. `nq-orchestrate [mission.md]`
renders that same prompt, appends the optional mission, and supplies it to the
foreground orchestrator agent.
Publish the reviewed rendered prompt into that project's context store with
its version check.
Give the project its own agent definitions, broker, profile, and terminal
launcher. The template itself does not start an orchestrator or register
another project.
+35
View File
@@ -0,0 +1,35 @@
{
"LAST_UPDATED": "2026-09-22",
"PROJECT_CONTEXT_LABEL": "NESTQUEST HOUSE CHAOS COORDINATION",
"PROJECT_NAME": "NestQuest",
"WORK_TRACKER": "Maestro",
"CONTEXT_STORE": "CTXD",
"MODEL_PROVIDER": "OpenRouter",
"CONTROLLER_MODEL": "deepseek/deepseek-v4.1-flash",
"DEVELOPER_MODEL": "z-ai/glm-5.3-flash",
"REVIEWER_MODEL": "claude-sonnet-5",
"REVIEWER_DISPLAY_NAME": "Claude Sonnet 5",
"REVIEWER_HARNESS": "Claude Code",
"REVIEWER_VENDOR": "Claude",
"REVIEWER_CLI": "claude",
"MAX_DEVELOPERS": "2",
"MAX_DEVELOPERS_WORD": "two",
"INTEGRATION_BRANCH": "dev",
"RELEASE_BRANCH": "main",
"RELEASE_OWNER_NAME": "Joshua",
"HOST_ACCOUNT": "overseer",
"CODE_HOST": "Gitea",
"TASK_ID_EXAMPLE": "TASK-123",
"FEATURE_BRANCH_EXAMPLE": "feature/NQ-17-family-calendar",
"OPENCODE_VERSION": "1.18.32",
"PROJECT_ROOT": "/mnt/ai-storage/projects/NestQuest",
"CONTROL_HOME": "/home/overseer/.local/share/nestquest-controller",
"SERVICE_NAME": "nestquest-controller.service",
"TERMINAL_LAUNCHER": "nq-orchestrate",
"ROLE_LAUNCHER": "nq-agent",
"REVIEW_LAUNCHER": "nq-review",
"MERGE_BROKER": "nq-approve-merge",
"SCRATCH_DIR": "/tmp/opencode",
"REVIEW_ACCOUNT": "review_agent",
"TICK_TIMEOUT_LABEL": "six-hour"
}
+288 -6
View File
@@ -52,6 +52,7 @@ _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",
@@ -99,6 +100,7 @@ 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"),
@@ -365,6 +367,15 @@ 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.
@@ -373,8 +384,20 @@ 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``,
and ``data``. Interval scheduling is NOT simulated (no real clock
ticks in tests) — tests drive refreshes directly.
``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.
"""
def __init__(self, hass, logger, *, name=None, update_interval=None):
@@ -384,6 +407,7 @@ 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):
@@ -403,18 +427,35 @@ class DataUpdateCoordinator:
callback()
async def async_refresh(self):
"""Run one update pass and notify listeners."""
"""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.
"""
try:
self.data = await self._async_update_data()
self.last_update_success = True
except BaseException:
except Exception as err:
self.last_update_success = False
raise
self.last_exception = err
self.async_update_listeners()
async def async_config_entry_first_refresh(self):
"""First refresh at config-entry setup (alias of refresh)."""
"""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.
"""
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."""
@@ -424,6 +465,7 @@ 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
@@ -700,6 +742,246 @@ 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."""
+450
View File
@@ -0,0 +1,450 @@
"""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 == []
+283
View File
@@ -0,0 +1,283 @@
"""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 == []
+501 -236
View File
@@ -1,263 +1,512 @@
"""Tests for the shared NestQuest coordinator (Feature 10, task 1)."""
"""Tests for the shared NestQuest coordinator (Feature 10, Feature 18).
The coordinator now polls the API service's panel snapshot through the
API client (Feature 18) and reconstructs the core snapshot dataclasses
every entity reads. The tests stub the client — NO network, NO aiohttp
— with a FIXTURE snapshot payload (the documented shape of the Feature
16 snapshot route, ``api/routes_panel.py``) and assert the
reconstruction: per-child counts and presence, the instance views
(``completed`` mapped back to ``done``), the household rollups, the
cycle day, and the unconfigured / failed-fetch behaviour.
"""
from __future__ import annotations
import datetime
import pytest
from conftest import wire_entry_to_registry
from conftest import set_coordinator_client, wire_entry_to_registry
from custom_components.nestquest import async_setup_entry, async_unload_entry
from custom_components.nestquest.children import (
create_child,
list_children,
set_child_active,
from custom_components.nestquest.api_client import (
NestQuestApiClient,
NestQuestApiError,
)
from custom_components.nestquest.completion import complete_instance
from custom_components.nestquest.const import (
CONF_UPDATE_INTERVAL,
DEFAULT_UPDATE_INTERVAL,
DEFAULT_HORIZON_DAYS,
DOMAIN,
)
from custom_components.nestquest.coordinator import NestQuestCoordinator
from custom_components.nestquest.dao_presence import PresenceSchedulesDao
from custom_components.nestquest.materialize import materialize
from custom_components.nestquest.quest_definitions import (
create_quest_definition,
from custom_components.nestquest.coordinator import (
NestQuestCoordinator,
coordinator_client_from_entry,
snapshot_from_api_payload,
)
from custom_components.nestquest.recurrence import ScheduleRule
ADMIN_CTX = {"user_id": "admin-1"}
from custom_components.nestquest.core.snapshot import instance_payload
from homeassistant.helpers.update_coordinator import UpdateFailed
def _today_iso() -> str:
return datetime.date.today().isoformat()
async def _seed_day(database, *, completed: bool = False) -> None:
"""Two active children + one inactive, one daily definition each."""
await create_child(database, "Ada")
await create_child(database, "Bo")
inactive = await create_child(database, "Ghost")
await set_child_active(database, inactive.id, False)
today = _today_iso()
await create_quest_definition(
database,
"Brush teeth",
ScheduleRule.from_dict({"rule_type": "daily", "start_date": today}),
[child.id for child in await list_children(database, active_only=True)],
["morning"],
class StubSnapshotClient:
"""A no-network stand-in for the API client's snapshot surface.
Answers ``get_snapshot()`` from a scripted queue of outcomes — a
payload dict or an exception per call, consumed in order — and
records every call, so each test drives exactly one refresh.
"""
def __init__(self, *outcomes) -> None:
self.calls = 0
self._outcomes = list(outcomes)
self._last = outcomes[-1] if outcomes else None
async def get_snapshot(self):
self.calls += 1
if not self._outcomes:
# Re-serve the last outcome: setup's first refresh and the
# test's explicit pass both poll the same scripted state.
outcome = self._last
else:
outcome = self._outcomes.pop(0)
self._last = outcome
if isinstance(outcome, BaseException):
raise outcome
return outcome
#: A fixture panel snapshot (the documented Feature 16 route shape):
#: two children — Ada owes one, owes none of it yet, is away today with
#: a returns date; Bo completed his only quest and is present.
FIXTURE_SNAPSHOT = {
"today_iso": _today_iso(),
"cycle_day": 3,
"children": [
{
"child_id": 1,
"child_name": "Ada",
"present": False,
"next_present": "2026-09-24",
"due_today": 2,
"completed_today": 1,
"remaining_today": 1,
"completion_pct": 50,
"instances": [
{
"id": 7,
"definition_id": 3,
"child_id": 1,
"title": "Brush teeth",
"icon": "🦷",
"window": "morning",
"due_time": "08:00",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:55:00+00:00",
"on_time": True,
},
{
"id": 8,
"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,
},
],
},
{
"child_id": 2,
"child_name": "Bo",
"present": True,
"next_present": None,
"due_today": 1,
"completed_today": 1,
"remaining_today": 0,
"completion_pct": 100,
"instances": [
{
"id": 9,
"definition_id": 4,
"child_id": 2,
"title": "Make bed",
"icon": "🛏",
"window": "morning",
"due_time": "07:30",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:31:00+00:00",
"on_time": True,
}
],
},
],
}
async def _wire_entry(hass, make_entry, client, options=None):
"""Build an entry whose coordinator uses the scripted API client.
The client is registered with the conftest override BEFORE setup:
the first refresh at setup already polls the snapshot, so the
platform setup sees real data and creates the whole entity set.
"""
from conftest import set_coordinator_client
entry = wire_entry_to_registry(
make_entry(
options=options,
data={"api_base_url": "http://api.test:8000", "panel_token": "tok"},
),
hass.registry,
)
end = (
datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
).isoformat()
await materialize(database, today, end, today=datetime.date.today())
async def test_coordinator_snapshot_counts_and_states(hass, make_entry) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
set_coordinator_client(entry, client)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert coordinator is not None
database = entry.runtime_data.database
await _seed_day(database)
assert coordinator.api_client is client
return entry, coordinator
async def test_coordinator_builds_snapshot_from_fixture(hass, make_entry) -> None:
"""One refresh = one snapshot GET; the fixture's values land on the
snapshot the entities read, per child and household-wide."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
calls_after_setup = client.calls
snapshot = await coordinator._async_update_data()
assert client.calls == calls_after_setup + 1
assert snapshot.today_iso == _today_iso()
assert len(snapshot.children) == 2, "inactive child must be absent"
by_name = {child.child_name: child for child in snapshot.children}
ada = by_name["Ada"]
assert ada.due_today == 1
assert ada.completed_today == 0
assert snapshot.cycle_day == 3
assert len(snapshot.children) == 2
ada, bo = snapshot.children
assert (ada.child_id, ada.child_name) == (1, "Ada")
assert ada.due_today == 2
assert ada.completed_today == 1
assert ada.remaining_today == 1
assert ada.completion_pct == 0
instance = ada.instances[0]
assert instance.state == "open"
assert instance.overdue is False
assert instance.title == "Brush teeth"
assert instance.window == "morning"
assert ada.present is True, "no schedule means present every day"
assert ada.next_present is None, "present child has no returns date"
assert snapshot.cycle_day == 0, "no schedule anywhere"
assert ada.completion_pct == 50
assert ada.present is False
assert ada.next_present == "2026-09-24"
done_view, open_view = ada.instances
assert done_view.instance_id == 7
assert done_view.title == "Brush teeth"
assert done_view.icon == "🦷"
assert done_view.window == "morning"
assert done_view.due_time == "08:00"
assert done_view.due_date == _today_iso()
# The API's "completed" maps back to the core view's "done".
assert done_view.state == "done"
assert done_view.overdue is False
assert done_view.completed_at == f"{_today_iso()}T07:55:00+00:00"
assert done_view.was_on_time is True
assert open_view.state == "open"
assert open_view.overdue is True
assert open_view.completed_at is None
assert open_view.was_on_time is None
assert (bo.child_id, bo.child_name) == (2, "Bo")
assert bo.due_today == 1
assert bo.completed_today == 1
assert bo.remaining_today == 0
assert bo.completion_pct == 100
assert bo.present is True
assert bo.next_present is None
# Household rollups: the sums the household sensors compute.
assert sum(child.due_today for child in snapshot.children) == 3
assert sum(child.completed_today for child in snapshot.children) == 2
assert sum(child.remaining_today for child in snapshot.children) == 1
async def test_coordinator_next_present_for_away_child(
hass, make_entry
) -> None:
"""A child absent today carries the ISO date of their next present
day; an always-absent pattern carries None instead."""
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
database = entry.runtime_data.database
await create_child(database, "Ada")
ada = (await list_children(database))[0]
today = datetime.date.today()
pattern = ",".join(
str(day) for day in range(7) if day != today.weekday()
)
await PresenceSchedulesDao(database).upsert_by_child(
ada.id, 1, today.isoformat(), pattern
)
await create_child(database, "Bo")
bo = (await list_children(database))[1]
await PresenceSchedulesDao(database).upsert_by_child(
bo.id, 1, today.isoformat(), ""
)
async def test_coordinator_refresh_publishes_to_listeners(hass, make_entry) -> None:
"""The refresh pass notifies listeners and reports success — the
ONE refresh pass every entity reads from."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
called = []
coordinator.async_add_listener(lambda: called.append("entity"))
await coordinator.async_refresh()
assert called == ["entity"]
assert coordinator.last_update_success is True
assert coordinator.data is not None
assert coordinator.data.children[0].child_name == "Ada"
async def test_instance_payload_shape_is_unchanged(hass, make_entry) -> None:
"""The per-child instance attribute payload the due-today sensor
emits keeps its exact documented shape, and admin_instances now
equals instances (the API omits missed rows — D-009; the admin
surface for missed quests is the PWA)."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
snapshot = await coordinator._async_update_data()
by_name = {child.child_name: child for child in snapshot.children}
assert by_name["Ada"].present is False
assert by_name["Ada"].next_present == (
today + datetime.timedelta(days=1)
).isoformat()
assert by_name["Bo"].present is False
assert by_name["Bo"].next_present is None
ada = snapshot.children[0]
panel = instance_payload(ada.instances, include_missed=False)
assert panel == [
{
"id": 7,
"definition_id": 3,
"child_id": 1,
"title": "Brush teeth",
"icon": "🦷",
"window": "morning",
"due_time": "08:00",
"state": "completed",
"overdue": False,
"completed_at": f"{_today_iso()}T07:55:00+00:00",
"on_time": True,
},
{
"id": 8,
"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,
},
]
# The API payload omits missed rows, so include_missed cannot
# re-add them: admin_instances == instances, byte for byte.
assert instance_payload(ada.instances, include_missed=True) == panel
async def test_coordinator_reflects_completion_and_uncompletion(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
database = entry.runtime_data.database
await _seed_day(database)
ada = (
async def test_coordinator_entities_read_the_api_snapshot(hass, make_entry) -> None:
"""The registered sensors and binary sensors read the rebuilt
snapshot: entity values come from the fixture, with the SAME
entity_ids as ever."""
client = StubSnapshotClient(FIXTURE_SNAPSHOT)
entry, coordinator = await _wire_entry(hass, make_entry, client)
await coordinator.async_refresh()
ada_due = hass.entities[f"{DOMAIN}_child_1_quests_due_today"]
assert ada_due.native_value == 2
attributes = ada_due.extra_state_attributes
assert attributes["child_id"] == 1
assert attributes["child_name"] == "Ada"
assert attributes["present"] is False
assert [row["id"] for row in attributes["instances"]] == [7, 8]
assert (
hass.entities[f"{DOMAIN}_child_1_quests_completed_today"].native_value
== 1
)
assert (
hass.entities[f"{DOMAIN}_child_1_quests_remaining_today"].native_value
== 1
)
assert (
hass.entities[f"{DOMAIN}_child_1_completion_pct_today"].native_value
== 50
)
assert hass.entities[f"{DOMAIN}_child_1_next_quest"].native_value == (
"Pack bag"
)
# Ada owes one still: not all done; away today.
assert hass.entities[f"{DOMAIN}_child_1_all_done"].is_on is False
assert hass.entities[f"{DOMAIN}_child_1_present_today"].is_on is False
assert (
hass.entities[f"{DOMAIN}_child_2_completion_pct_today"].native_value
== 100
)
# Bo had a quest and owes none of it: all done, present today.
assert hass.entities[f"{DOMAIN}_child_2_all_done"].is_on is True
assert hass.entities[f"{DOMAIN}_child_2_present_today"].is_on is True
bo_attributes = hass.entities[
f"{DOMAIN}_child_2_present_today"
].extra_state_attributes
assert bo_attributes["next_present"] is None
assert (
hass.entities[f"{DOMAIN}_household_quests_due_today"].native_value == 3
)
assert (
hass.entities[f"{DOMAIN}_household_quests_completed_today"].native_value
== 2
)
cycle = hass.entities[f"{DOMAIN}_cycle_day"]
assert cycle.native_value == 3
assert cycle.extra_state_attributes == {
"cycle_day": 3,
"today": _today_iso(),
}
await async_unload_entry(hass, entry)
async def test_coordinator_failed_fetch_marks_unavailable(hass, make_entry) -> None:
"""A failed fetch (transport failure or non-2xx — both typed) is a
FAILED refresh pass: the update pass converts the typed error to
UpdateFailed (the exception HA's coordinator machinery records as a
failed pass, chaining the typed error as ``__cause__``), the refresh
records the failure instead of raising, and data keeps its last
good value."""
error = NestQuestApiError(
"NestQuest API GET /api/v1/panel/snapshot failed with HTTP 503",
status=503,
)
# The stub serves the good snapshot for setup's first refresh, then
# the scripted failure.
client = StubSnapshotClient(FIXTURE_SNAPSHOT, error)
_entry, coordinator = await _wire_entry(hass, make_entry, client)
# First refresh (at setup) served the good snapshot.
assert coordinator.data is not None
assert coordinator.last_update_success is True
# The next pass hits the scripted failure: the update pass converts
# the typed error to UpdateFailed, with the typed error chained.
with pytest.raises(UpdateFailed, match="HTTP 503") as excinfo:
await coordinator._async_update_data()
).children[0]
instance_id = ada.instances[0].instance_id
assert isinstance(excinfo.value.__cause__, NestQuestApiError)
await complete_instance(
database,
instance_id,
actor_source="panel",
actor_child_id=ada.child_id,
)
snapshot = await coordinator._async_update_data()
child = snapshot.children[0]
assert child.completed_today == 1
assert child.remaining_today == 0
assert child.completion_pct == 100
view = child.instances[0]
assert view.state == "done"
assert view.completed_at is not None
assert view.was_on_time is not None
from custom_components.nestquest.completion import uncomplete_instance
await uncomplete_instance(
database, instance_id, actor_source="user", actor_user_id="admin-1"
)
snapshot = await coordinator._async_update_data()
child = snapshot.children[0]
assert child.completed_today == 0
assert child.remaining_today == 1
assert child.instances[0].state == "open"
# The refresh pass records the failure the HA way — it does NOT
# raise: last_update_success goes False, data keeps the last GOOD
# snapshot, and the failed pass lands on last_exception as the
# converted UpdateFailed with the typed error chained.
await coordinator.async_refresh()
assert coordinator.last_update_success is False
assert coordinator.data.children[0].child_name == "Ada"
assert isinstance(coordinator.last_exception, UpdateFailed)
assert isinstance(coordinator.last_exception.__cause__, NestQuestApiError)
async def test_coordinator_zero_quests_reports_full_percentage(
async def test_failed_first_refresh_does_not_crash_setup(
hass, make_entry
) -> None:
"""A failed fetch at the FIRST refresh (at setup) does not crash
setup: the coordinator reports last_update_success=False, the setup
catches the ConfigEntryNotReady that HA's real first refresh raises
for a failed pass, and the platforms still forward so the entities
come up unavailable."""
# The stub fails every snapshot fetch (HTTP 503), including setup's
# first refresh.
client = StubSnapshotClient(
NestQuestApiError(
"NestQuest API GET /api/v1/panel/snapshot failed with HTTP 503",
status=503,
)
)
entry = wire_entry_to_registry(
make_entry(
data={"api_base_url": "http://api.test:8000", "panel_token": "tok"}
),
hass.registry,
)
set_coordinator_client(entry, client)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
assert coordinator.api_client is client
# The failed first pass: the entities' surface stays unavailable.
assert coordinator.last_update_success is False
assert coordinator.data is None
# Setup survived the failed first refresh — the platforms were
# forwarded (the regression the review caught: a first-refresh
# failure used to abort setup before any entity was created).
assert hass.config_entries.forwarded_platforms == [
"sensor",
"binary_sensor",
]
# The unload after the failed-first-refresh setup stays clean.
assert await async_unload_entry(hass, entry) is True
async def test_coordinator_unconfigured_api_does_not_crash_setup(
hass, make_entry, monkeypatch
) -> None:
"""An entry with no panel token set up fine: the coordinator gets
the typed-error stub, the first refresh fails cleanly (unavailable
entities), and setup does NOT crash."""
# Restore the production factory: the conftest default would serve
# the local snapshot, but this test exercises the token guard.
import custom_components.nestquest as nq
monkeypatch.setattr(
nq, "coordinator_client_from_entry", coordinator_client_from_entry
)
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
await create_child(entry.runtime_data.database, "Ada")
snapshot = await coordinator._async_update_data()
child = snapshot.children[0]
assert child.due_today == 0
assert child.completion_pct == 100, "zero quests is a perfect day"
assert child.instances == ()
from custom_components.nestquest.coordinator import _UnconfiguredApiClient
assert isinstance(coordinator.api_client, _UnconfiguredApiClient)
# A real NestQuestApiClient was never built (it would ValueError).
assert not isinstance(coordinator.api_client, NestQuestApiClient)
# The first refresh failed the HA way: the update pass converted
# the stub's typed error to UpdateFailed, the first refresh raised
# ConfigEntryNotReady, and setup tolerated it — platforms forwarded
# so the entities come up unavailable.
assert coordinator.last_update_success is False
assert coordinator.data is None
with pytest.raises(UpdateFailed, match="not configured"):
await coordinator._async_update_data()
assert hass.config_entries.forwarded_platforms == [
"sensor",
"binary_sensor",
]
assert await async_unload_entry(hass, entry) is True
async def test_coordinator_cycle_day_from_first_scheduled_child(
async def test_coordinator_configured_token_builds_real_client(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
"""An entry WITH a panel token builds the real client through the
production factory (a stub transport injected via session_factory,
the mock-only harness having no aiohttp module to hand in), and the
coordinator's refresh polls the snapshot route end to end."""
import json as json_module
class _OneShotSession:
"""A session-shaped stub answering every request with the fixture."""
def __init__(self, payload) -> None:
self._body = json_module.dumps(payload)
def request(self, method, url, **kwargs):
body = self._body
class _Response:
def __init__(self) -> None:
self.status = 200
async def text(self):
return body
class _Context:
async def __aenter__(self):
return _Response()
async def __aexit__(self, *exc):
return False
return _Context()
def _session_factory(_hass):
return _OneShotSession(FIXTURE_SNAPSHOT)
entry = wire_entry_to_registry(
make_entry(
data={"api_base_url": "http://api.test:8000", "panel_token": "tok"}
),
hass.registry,
)
real_client = coordinator_client_from_entry(
hass, entry, session_factory=_session_factory
)
assert isinstance(real_client, NestQuestApiClient)
set_coordinator_client(entry, real_client)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
database = entry.runtime_data.database
await create_child(database, "Ada")
ada = (await list_children(database))[0]
today = datetime.date.today()
# A 2-week cycle anchored 3 days ago: today is day 4.
anchor = today - datetime.timedelta(days=3)
await PresenceSchedulesDao(database).upsert_by_child(
ada.id, 2, anchor.isoformat(), "0,1,2,3,4,5,6|"
)
snapshot = await coordinator._async_update_data()
assert snapshot.cycle_day == 4
async def test_coordinator_presence_resolves_schedule_and_override(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
database = entry.runtime_data.database
await create_child(database, "Ada")
ada = (await list_children(database))[0]
today = datetime.date.today()
# Scheduled absent today (empty weekday set for this cycle week).
await PresenceSchedulesDao(database).upsert_by_child(
ada.id, 1, today.isoformat(), ""
)
snapshot = await coordinator._async_update_data()
assert snapshot.children[0].present is False
# An override beats the pattern.
from custom_components.nestquest.dao_presence import (
PresenceOverridesDao,
)
await PresenceOverridesDao(database).create(
ada.id, today.isoformat(), today.isoformat(), True
)
snapshot = await coordinator._async_update_data()
assert snapshot.children[0].present is True
async def test_coordinator_next_present_honours_future_override(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
database = entry.runtime_data.database
await create_child(database, "Ada")
ada = (await list_children(database))[0]
today = datetime.date.today()
tomorrow = today + datetime.timedelta(days=1)
# Scheduled present tomorrow (single-day cycle, weekday set set below
# covers every day), but overridden absent tomorrow.
await PresenceSchedulesDao(database).upsert_by_child(
ada.id, 1, today.isoformat(), "0,1,2,3,4,5,6"
)
from custom_components.nestquest.dao_presence import (
PresenceOverridesDao,
)
await PresenceOverridesDao(database).create(
ada.id, tomorrow.isoformat(), tomorrow.isoformat(), False
)
snapshot = await coordinator._async_update_data()
child = snapshot.children[0]
assert child.present is True
assert child.next_present != tomorrow.isoformat(), (
"a future absent override must remove tomorrow from the preview"
)
async def test_coordinator_derives_missed_for_past_due_open(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
snapshot = await coordinator._async_update_data()
# No children: empty but valid snapshot.
assert snapshot.children == ()
assert isinstance(coordinator.api_client, NestQuestApiClient)
# The real client polls the stub transport; the snapshot lands.
await coordinator.async_refresh()
assert coordinator.data.children[0].child_name == "Ada"
async def test_setup_creates_coordinator_with_configured_interval(
@@ -270,12 +519,9 @@ async def test_setup_creates_coordinator_with_configured_interval(
coordinator = entry.runtime_data.coordinator
assert isinstance(coordinator, NestQuestCoordinator)
assert coordinator.update_interval == datetime.timedelta(seconds=60)
assert coordinator.data is not None, "first refresh ran at setup"
async def test_setup_defaults_to_five_minute_interval(
hass, make_entry
) -> None:
async def test_setup_defaults_to_five_minute_interval(hass, make_entry) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
@@ -298,24 +544,23 @@ async def test_unload_shuts_coordinator_down(hass, make_entry) -> None:
assert torn_down == [coordinator]
async def test_coordinator_notifies_listeners_on_refresh(
hass, make_entry
) -> None:
entry = wire_entry_to_registry(make_entry(), hass.registry)
assert await async_setup_entry(hass, entry) is True
coordinator = entry.runtime_data.coordinator
called = []
coordinator.async_add_listener(lambda: called.append("entity"))
await coordinator.async_refresh()
assert called == ["entity"]
assert coordinator.last_update_success is True
async def test_failed_first_refresh_shuts_coordinator_down(
async def test_setup_failure_after_registration_shuts_coordinator_down(
hass, make_entry, monkeypatch
) -> None:
"""A setup failure after coordinator creation must not leak its
listeners against the database the except path closes."""
listeners or leave a stale runtime record behind.
The runtime record registers in ``hass.data`` BEFORE the platform
forward, so a forward failure reaches the unwind with the record
already registered. The unwind must shut the coordinator down,
clear ``entry.runtime_data``, AND remove the stale ``hass.data``
entry — otherwise a later unload picks up the stale, already-torn-
down record and re-invokes the already-consumed update-listener
remover, which raises. (A failed first refresh is NOT such a
failure: HA's real first refresh wraps it in ConfigEntryNotReady,
which setup deliberately tolerates so entities come up
unavailable.)
"""
from custom_components.nestquest import async_setup_entry
import custom_components.nestquest.coordinator as coordinator_module
@@ -330,17 +575,37 @@ async def test_failed_first_refresh_shuts_coordinator_down(
_record_shutdown,
)
async def _boom(self):
raise RuntimeError("refresh failed")
async def _boom(entry, platforms):
raise RuntimeError("platform forward failed")
monkeypatch.setattr(
coordinator_module.NestQuestCoordinator,
"_async_update_data",
hass.config_entries,
"async_forward_entry_setups",
_boom,
)
entry = wire_entry_to_registry(make_entry(), hass.registry)
with pytest.raises(RuntimeError, match="refresh failed"):
with pytest.raises(RuntimeError, match="platform forward failed"):
await async_setup_entry(hass, entry)
assert shutdown_calls == [entry.entry_id]
assert getattr(entry, "runtime_data", None) is None
# The stale hass.data registration was removed with the runtime
# record: the unload below finds nothing stale, must not raise,
# and must leave no residue behind.
assert await async_unload_entry(hass, entry) is True
assert DOMAIN not in hass.data
assert hass.data == {}
assert hass.registry.size == 0
assert hass.time_change.size == 0
assert entry.runtime_data is None
async def test_snapshot_from_api_payload_empty_household() -> None:
"""An empty household (no children) is a valid snapshot: empty
children tuple, cycle day passes through."""
snapshot = snapshot_from_api_payload(
{"today_iso": "2026-09-23", "cycle_day": 0, "children": []}
)
assert snapshot.children == ()
assert snapshot.cycle_day == 0
assert snapshot.today_iso == "2026-09-23"
+37 -10
View File
@@ -123,9 +123,11 @@ 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).
today = coordinator._local_now().date()
# 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.
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(
@@ -294,13 +296,38 @@ async def test_build_snapshot_explicit_values(hass, make_entry, monkeypatch) ->
== "missed"
)
# 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)
# 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())
await coordinator.async_refresh()
assert coordinator.data == built
+19 -7
View File
@@ -14,7 +14,7 @@ from __future__ import annotations
import datetime
import re
from conftest import wire_entry_to_registry
from conftest import refire_api_transitions, wire_entry_to_registry
from custom_components.nestquest import async_setup_entry
from custom_components.nestquest.children import (
@@ -154,6 +154,11 @@ 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"])
@@ -180,9 +185,12 @@ 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 while
# admin_instances includes them (D-009): pin the derive clock
# forward in the TEST ONLY so today's open instances read missed.
# --- 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.
original_refresh = coordinator._async_update_data
async def _future_dated_refresh():
@@ -203,9 +211,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 len(attributes["admin_instances"]) == 2
assert all(
entry["state"] == "missed" for entry in attributes["admin_instances"]
assert attributes["admin_instances"] == [], (
"the API payload omits missed rows, so the admin payload "
"carries none either (the PWA is the admin surface now)"
)
coordinator._async_update_data = original_refresh
await coordinator.async_refresh()
@@ -231,6 +239,10 @@ 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)),
+116 -26
View File
@@ -1,9 +1,21 @@
"""Tests for the NestQuest bus events (Feature 10, task 6)."""
"""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.
"""
from __future__ import annotations
import re
from conftest import wire_entry_to_registry
from conftest import refire_api_transitions, wire_entry_to_registry
from custom_components.nestquest import async_setup_entry
from custom_components.nestquest.children import create_child
@@ -86,6 +98,11 @@ 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]
@@ -111,6 +128,15 @@ 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
@@ -128,6 +154,9 @@ 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]
@@ -153,6 +182,12 @@ 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,
@@ -163,6 +198,9 @@ 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
@@ -223,6 +261,15 @@ 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
@@ -263,6 +310,9 @@ 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(
@@ -280,12 +330,19 @@ 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 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.
# 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.
assert len(hass.bus.fired(EVENT_QUEST_COMPLETED)) == len(instances)
@@ -324,6 +381,10 @@ 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
)
@@ -538,17 +599,32 @@ 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."""
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
entry, child, instances = await _setup_and_seed(hass, make_entry)
database = entry.runtime_data.database
for instance in instances:
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{
"instance_id": instance.id,
"actor": "panel",
"actor_child_id": child.id,
},
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,
)
ordered = [
etype for etype, _payload in hass.bus.events
@@ -732,21 +808,31 @@ 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 day-complete rule reads ``hass.config.time_zone`` and
derives ``today`` in HA-local time, not UTC.
"""The shim's 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), completes it
through the service path, and asserts
instance for THAT zone's today (not UTC's today), fires the shim
directly for a completion of that instance, 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
@@ -775,15 +861,19 @@ 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:
await hass.services.call(
DOMAIN,
SERVICE_COMPLETE_QUEST,
{
"instance_id": instance.id,
"actor": "panel",
"actor_child_id": child.id,
},
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,
)
day_complete = hass.bus.fired(EVENT_CHILD_DAY_COMPLETE)
assert len(day_complete) == 1, (
+110 -8
View File
@@ -15,22 +15,26 @@ 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_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_UPDATE_INTERVAL,
DEFAULT_DAY_ROLLOVER_TIME,
DEFAULT_HORIZON_DAYS,
@@ -49,8 +53,9 @@ VALID_INPUT = {
}
#: What a schema-validated submit produces from a bare VALID_INPUT on a
#: fresh entry: the Feature 10 refresh interval and the Feature 11
#: notification section fill from their defaults.
#: fresh entry: the Feature 10 refresh interval, the Feature 11
#: notification section, and the Feature 18 API connection fill from
#: their defaults.
FULL_INPUT = {
**VALID_INPUT,
CONF_UPDATE_INTERVAL: DEFAULT_UPDATE_INTERVAL,
@@ -62,6 +67,8 @@ 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,
}
@@ -259,6 +266,8 @@ 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: [],
}
@@ -272,6 +281,9 @@ 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])
@@ -326,8 +338,10 @@ 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_four_keys() -> None:
"""A valid submit creates an entry with exactly the four validated keys."""
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)."""
flow = _make_flow(_make_entry())
result = _run(flow.async_step_init(dict(VALID_INPUT)))
assert result["type"] == "create_entry"
@@ -799,3 +813,91 @@ 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"]
+21 -14
View File
@@ -221,19 +221,23 @@ 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_for_panel_includes_for_admin(
async def test_due_today_payload_omits_missed_admin_equals_panel(
hass, make_entry, monkeypatch
) -> None:
"""D-009: the panel payload omits missed instances, the admin one keeps them.
"""D-009: the panel payload omits missed instances.
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.
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.
"""
entry, coordinator, (ada, bo, cory) = await _setup_seeded_entry(
hass, make_entry
@@ -256,10 +260,13 @@ async def test_due_today_payload_omits_missed_for_panel_includes_for_admin(
assert attributes["instances"] == [], (
"missed instances are omitted from the panel payload"
)
[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 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.
assert _sensor(hass, bo.id, "quests_completed_today").native_value == 0
assert _sensor(hass, bo.id, "quests_remaining_today").native_value == 1
+619
View File
@@ -0,0 +1,619 @@
"""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