Add the permission test matrix across every registered service

Admin succeeds on all ten admin-only services; a non-admin and a
no-context call are refused on each; complete_quest stays open; a
kiosk-style id is refused on uncomplete_quest and regenerate; the
registry coverage assertion fails the suite on any gap.

custom_components/nestquest/permissions.py

@@ -0,0 +1,90 @@
+"""Permission gate for NestQuest services (Feature 09).
+
+A wrapper factory, applied to EVERY registered service at registration
+time (no handler is exempt), reads ``call.context.user_id``, consults
+the :data:`~.service_policy.SERVICE_POLICY` registry and
+:func:`~.admin_allowlist.is_admin`, and raises
+:class:`homeassistant.exceptions.Unauthorized` for an admin-only
+operation called by a non-admin — including a call with NO user
+context, which fails closed.  ``complete_quest`` (policy ``open``)
+proceeds regardless of the allowlist.
+
+Logging (later Feature 09 tasks): every denied admin-only call logs a
+warning and every successful one logs a debug line, both carrying the
+service, the user id (or ``none``), and a UTC timestamp.
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from homeassistant.exceptions import Unauthorized
+
+from .admin_allowlist import is_admin
+from .service_policy import ADMIN_ONLY
+
+LOGGER = logging.getLogger(__name__)
+
+ServiceHandler = Callable[[Any], Awaitable[None]]
+
+
+def _context_user_id(call: Any) -> str | None:
+    context = getattr(call, "context", None)
+    user_id = getattr(context, "user_id", None) if context is not None else None
+    if isinstance(user_id, str) and user_id:
+        return user_id
+    return None
+
+
+def _stamp() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def permission_gate(
+    service: str,
+    policy: str,
+    database_of: Callable[[Any], Any],
+) -> Callable[[ServiceHandler], ServiceHandler]:
+    """Wrap ``handler`` with the Feature 09 permission check.
+
+    ``database_of`` resolves the live database from the call-time
+    runtime (the same ``find_runtime`` the handler uses), so the gate
+    never opens its own connection.  The verdict is fail-closed:
+    absent allowlist row, unknown id, no user context — all not admin.
+    """
+
+    def _wrap(handler: ServiceHandler) -> ServiceHandler:
+        if policy != ADMIN_ONLY:
+            return handler
+
+        async def _gated(call: Any) -> None:
+            user_id = _context_user_id(call)
+            allowed = user_id is not None and await is_admin(
+                database_of(call), user_id
+            )
+            if not allowed:
+                LOGGER.warning(
+                    "NestQuest denied %s: caller user_id=%s at %s",
+                    service,
+                    user_id if user_id is not None else "none",
+                    _stamp(),
+                )
+                raise Unauthorized(
+                    f"User '{user_id if user_id is not None else 'none'}' "
+                    f"is not allowed to call nestquest.{service}"
+                )
+            LOGGER.debug(
+                "NestQuest allowed %s: caller user_id=%s at %s",
+                service,
+                user_id,
+                _stamp(),
+            )
+            await handler(call)
+
+        return _gated
+
+    return _wrap

custom_components/nestquest/service_policy.py

@@ -0,0 +1,46 @@
+"""Explicit admin-only versus open service registry (Feature 09).
+
+One mapping, one file: every registered ``nestquest`` service name is
+either ``open`` or ``admin_only``.  The permission decorator (later
+Feature 09 tasks) reads this table; it does not take per-handler
+policy arguments.
+"""
+from __future__ import annotations
+
+from .const import (
+    DOMAIN_SERVICES,
+    SERVICE_COMPLETE_QUEST,
+    SERVICE_CREATE_PRESENCE_OVERRIDE,
+    SERVICE_CREATE_QUEST_DEFINITION,
+    SERVICE_DELETE_PRESENCE_OVERRIDE,
+    SERVICE_EXPORT_HISTORY_CSV,
+    SERVICE_MANAGE_CHILD,
+    SERVICE_REGENERATE,
+    SERVICE_SET_PRESENCE_PATTERN,
+    SERVICE_SET_QUEST_DEFINITION_ACTIVE,
+    SERVICE_UNCOMPLETE_QUEST,
+    SERVICE_UPDATE_QUEST_DEFINITION,
+)
+
+OPEN = "open"
+ADMIN_ONLY = "admin_only"
+
+SERVICE_POLICY: dict[str, str] = {
+    SERVICE_COMPLETE_QUEST: OPEN,
+    SERVICE_UNCOMPLETE_QUEST: ADMIN_ONLY,
+    SERVICE_CREATE_QUEST_DEFINITION: ADMIN_ONLY,
+    SERVICE_UPDATE_QUEST_DEFINITION: ADMIN_ONLY,
+    SERVICE_SET_QUEST_DEFINITION_ACTIVE: ADMIN_ONLY,
+    SERVICE_SET_PRESENCE_PATTERN: ADMIN_ONLY,
+    SERVICE_CREATE_PRESENCE_OVERRIDE: ADMIN_ONLY,
+    SERVICE_DELETE_PRESENCE_OVERRIDE: ADMIN_ONLY,
+    SERVICE_EXPORT_HISTORY_CSV: ADMIN_ONLY,
+    SERVICE_MANAGE_CHILD: ADMIN_ONLY,
+    SERVICE_REGENERATE: ADMIN_ONLY,
+}
+
+if frozenset(SERVICE_POLICY) != frozenset(DOMAIN_SERVICES):
+    raise RuntimeError(
+        "SERVICE_POLICY must name exactly the registered nestquest services, "
+        f"got {sorted(SERVICE_POLICY)} vs {list(DOMAIN_SERVICES)}"
+    )

custom_components/nestquest/services.py

@@ -43,8 +43,10 @@ from .const import (
 )
 from .dao_presence import PresenceOverridesDao, PresenceSchedulesDao
 from .materialize import regenerate_for_child
+from .permissions import permission_gate
 from .presence import PresenceSchedule
 from .recurrence import ScheduleRule
+from .service_policy import SERVICE_POLICY
 
 FindRuntime = Callable[[HomeAssistant], Any]
 ServiceHandler = Callable[[Any], Awaitable[None]]
@@ -521,72 +523,78 @@ def async_register_services(
     find_runtime: FindRuntime,
     regenerate_handler: ServiceHandler,
 ) -> None:
-    """Register every canonical service once at domain scope."""
+    """Register every canonical service once at domain scope.
+
+    Every service — regenerate included — passes through the Feature
+    09 permission gate before its handler runs; the gate reads
+    :data:`~.service_policy.SERVICE_POLICY`, so no service can be
+    registered exempt from the policy.
+    """
     if hass.services.has_service(DOMAIN, SERVICE_REGENERATE):
         return
-    hass.services.async_register(
-        DOMAIN, SERVICE_REGENERATE, regenerate_handler
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_COMPLETE_QUEST,
-        _complete_quest(hass, find_runtime),
-        schema=SCHEMA_COMPLETE_QUEST,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_UNCOMPLETE_QUEST,
-        _uncomplete_quest(hass, find_runtime),
-        schema=SCHEMA_UNCOMPLETE_QUEST,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_CREATE_QUEST_DEFINITION,
-        _create_quest_definition(hass, find_runtime),
-        schema=SCHEMA_CREATE_QUEST_DEFINITION,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_UPDATE_QUEST_DEFINITION,
-        _update_quest_definition(hass, find_runtime),
-        schema=SCHEMA_UPDATE_QUEST_DEFINITION,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_SET_QUEST_DEFINITION_ACTIVE,
-        _set_quest_definition_active(hass, find_runtime),
-        schema=SCHEMA_SET_QUEST_DEFINITION_ACTIVE,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_SET_PRESENCE_PATTERN,
-        _set_presence_pattern(hass, find_runtime),
-        schema=SCHEMA_SET_PRESENCE_PATTERN,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_CREATE_PRESENCE_OVERRIDE,
-        _create_presence_override(hass, find_runtime),
-        schema=SCHEMA_CREATE_PRESENCE_OVERRIDE,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_DELETE_PRESENCE_OVERRIDE,
-        _delete_presence_override(hass, find_runtime),
-        schema=SCHEMA_DELETE_PRESENCE_OVERRIDE,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_EXPORT_HISTORY_CSV,
-        _export_history_csv,
-        schema=SCHEMA_EXPORT_HISTORY_CSV,
-    )
-    hass.services.async_register(
-        DOMAIN,
-        SERVICE_MANAGE_CHILD,
-        _manage_child(hass, find_runtime),
-        schema=SCHEMA_MANAGE_CHILD,
-    )
+    gate = permission_gate
+    specs: list[tuple[str, ServiceHandler, object]] = [
+        (SERVICE_REGENERATE, regenerate_handler, None),
+        (
+            SERVICE_COMPLETE_QUEST,
+            _complete_quest(hass, find_runtime),
+            SCHEMA_COMPLETE_QUEST,
+        ),
+        (
+            SERVICE_UNCOMPLETE_QUEST,
+            _uncomplete_quest(hass, find_runtime),
+            SCHEMA_UNCOMPLETE_QUEST,
+        ),
+        (
+            SERVICE_CREATE_QUEST_DEFINITION,
+            _create_quest_definition(hass, find_runtime),
+            SCHEMA_CREATE_QUEST_DEFINITION,
+        ),
+        (
+            SERVICE_UPDATE_QUEST_DEFINITION,
+            _update_quest_definition(hass, find_runtime),
+            SCHEMA_UPDATE_QUEST_DEFINITION,
+        ),
+        (
+            SERVICE_SET_QUEST_DEFINITION_ACTIVE,
+            _set_quest_definition_active(hass, find_runtime),
+            SCHEMA_SET_QUEST_DEFINITION_ACTIVE,
+        ),
+        (
+            SERVICE_SET_PRESENCE_PATTERN,
+            _set_presence_pattern(hass, find_runtime),
+            SCHEMA_SET_PRESENCE_PATTERN,
+        ),
+        (
+            SERVICE_CREATE_PRESENCE_OVERRIDE,
+            _create_presence_override(hass, find_runtime),
+            SCHEMA_CREATE_PRESENCE_OVERRIDE,
+        ),
+        (
+            SERVICE_DELETE_PRESENCE_OVERRIDE,
+            _delete_presence_override(hass, find_runtime),
+            SCHEMA_DELETE_PRESENCE_OVERRIDE,
+        ),
+        (
+            SERVICE_EXPORT_HISTORY_CSV,
+            _export_history_csv,
+            SCHEMA_EXPORT_HISTORY_CSV,
+        ),
+        (
+            SERVICE_MANAGE_CHILD,
+            _manage_child(hass, find_runtime),
+            SCHEMA_MANAGE_CHILD,
+        ),
+    ]
+
+    def _database_of(_call: Any) -> Any:
+        return _require_runtime(hass, find_runtime).database
+
+    for service, handler, schema in specs:
+        gated = gate(
+            service, SERVICE_POLICY[service], _database_of
+        )(handler)
+        hass.services.async_register(DOMAIN, service, gated, schema=schema)
 
 
 def async_deregister_services(hass: HomeAssistant) -> None:

tests/conftest.py

@@ -189,8 +189,13 @@ class ConfigEntryNotReady(HomeAssistantError):
     """
 
 
+class Unauthorized(HomeAssistantError):
+    """Stand-in mirroring homeassistant.exceptions.Unauthorized."""
+
+
 _exceptions_mock.HomeAssistantError = HomeAssistantError
 _exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
+_exceptions_mock.Unauthorized = Unauthorized
 
 import voluptuous as vol
 

tests/test_lifecycle.py

@@ -9,6 +9,7 @@ from conftest import make_config_entry, make_hass, wire_entry_to_registry
 
 from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry
 from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
     CONF_DAY_ROLLOVER_TIME,
     CONF_HORIZON_DAYS,
     DEFAULT_HORIZON_DAYS,
@@ -16,6 +17,13 @@ from custom_components.nestquest.const import (
     DOMAIN as DOMAIN_CONST,
 )
 
+ADMIN_ID = "admin-1"
+ADMIN_CTX = {"user_id": ADMIN_ID}
+
+
+def _admin_data():
+    return {CONF_ADMIN_USER_IDS: [ADMIN_ID]}
+
 
 def _wire(entry, registry):
     return wire_entry_to_registry(entry, registry)
@@ -546,7 +554,7 @@ async def test_regenerate_service_registered_and_materializes(
 
     from custom_components.nestquest.dao_instances import QuestInstancesDao
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(make_entry(data=_admin_data()), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
 
@@ -555,7 +563,7 @@ async def test_regenerate_service_registered_and_materializes(
     database = entry.runtime_data.database
     child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
 
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
 
     today = _local_today(hass)
     end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
@@ -574,14 +582,15 @@ async def test_regenerate_service_uses_configured_horizon_days(
     from custom_components.nestquest.dao_instances import QuestInstancesDao
 
     entry = _wire(
-        make_entry(options={CONF_HORIZON_DAYS: 5}), hass.registry
+        make_entry(options={CONF_HORIZON_DAYS: 5}, data=_admin_data()),
+        hass.registry,
     )
     assert await async_setup_entry(hass, entry) is True
 
     database = entry.runtime_data.database
     child_id = await _seed_daily_child_and_definition(database, _local_today(hass))
 
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
 
     today = _local_today(hass)
     end = (today + datetime.timedelta(days=5)).isoformat()
@@ -622,8 +631,12 @@ async def test_regenerate_service_is_domain_scoped_across_entries(
 
     from custom_components.nestquest.dao_instances import QuestInstancesDao
 
-    entry_a = _wire(make_entry(entry_id="entry_a"), hass.registry)
-    entry_b = _wire(make_entry(entry_id="entry_b"), hass.registry)
+    entry_a = _wire(
+        make_entry(entry_id="entry_a", data=_admin_data()), hass.registry
+    )
+    entry_b = _wire(
+        make_entry(entry_id="entry_b", data=_admin_data()), hass.registry
+    )
 
     assert await async_setup_entry(hass, entry_a) is True
     assert await async_setup_entry(hass, entry_b) is True
@@ -634,7 +647,7 @@ async def test_regenerate_service_is_domain_scoped_across_entries(
     child_id = await _seed_daily_child_and_definition(
         database_b, _local_today(hass)
     )
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
     today = _local_today(hass)
     end = (today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)).isoformat()
     records = await QuestInstancesDao(database_b).list_by_date_range(
@@ -647,7 +660,7 @@ async def test_regenerate_service_is_domain_scoped_across_entries(
     assert hass.services.has_service(DOMAIN, SERVICE_REGENERATE)
 
     # It still works for the remaining entry.
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
 
     # Unloading the final entry removes the domain-global service.
     assert await async_unload_entry(hass, entry_b) is True

tests/test_permission_matrix.py

@@ -0,0 +1,238 @@
+"""Feature 09 permission test matrix across every registered service."""
+from __future__ import annotations
+
+import datetime
+
+import pytest
+from homeassistant.exceptions import Unauthorized
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
+    DEFAULT_HORIZON_DAYS,
+    DOMAIN,
+    DOMAIN_SERVICES,
+    SERVICE_COMPLETE_QUEST,
+    SERVICE_CREATE_PRESENCE_OVERRIDE,
+    SERVICE_CREATE_QUEST_DEFINITION,
+    SERVICE_DELETE_PRESENCE_OVERRIDE,
+    SERVICE_EXPORT_HISTORY_CSV,
+    SERVICE_MANAGE_CHILD,
+    SERVICE_REGENERATE,
+    SERVICE_SET_PRESENCE_PATTERN,
+    SERVICE_SET_QUEST_DEFINITION_ACTIVE,
+    SERVICE_UNCOMPLETE_QUEST,
+    SERVICE_UPDATE_QUEST_DEFINITION,
+)
+from custom_components.nestquest.service_policy import (
+    ADMIN_ONLY,
+    OPEN,
+    SERVICE_POLICY,
+)
+
+ADMIN_ID = "admin-1"
+ADMIN_CTX = {"user_id": ADMIN_ID}
+NON_ADMIN_ID = "non-admin"
+NON_ADMIN_CTX = {"user_id": NON_ADMIN_ID}
+KIOSK_ID = "kiosk-panel"
+KIOSK_CTX = {"user_id": KIOSK_ID}
+
+
+def test_every_registered_service_is_in_the_operation_registry() -> None:
+    """The registry is total: a service outside it could never be gated."""
+    missing = [
+        service for service in DOMAIN_SERVICES if service not in SERVICE_POLICY
+    ]
+    extra = [
+        service for service in SERVICE_POLICY if service not in DOMAIN_SERVICES
+    ]
+    assert not missing, f"registered but ungated: {missing}"
+    assert not extra, f"gated but never registered: {extra}"
+    assert set(SERVICE_POLICY.values()) == {OPEN, ADMIN_ONLY}
+
+
+async def _setup(hass, make_entry, *, admin=True):
+    data = {CONF_ADMIN_USER_IDS: [ADMIN_ID]} if admin else {}
+    entry = wire_entry_to_registry(make_entry(data=data), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    return entry
+
+
+async def _seed(hass, entry) -> dict:
+    """Seed one child, one definition, and materialized instances."""
+    from custom_components.nestquest.children import list_children
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    database = entry.runtime_data.database
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_MANAGE_CHILD,
+        {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
+    )
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_CREATE_QUEST_DEFINITION,
+        {
+            "title": "Brush teeth",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [(await list_children(database))[0].id],
+            "windows": ["morning"],
+        },
+        context=ADMIN_CTX,
+    )
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
+    child = (await list_children(database))[0]
+    today = datetime.date.today().isoformat()
+    end = (
+        datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    instances = await QuestInstancesDao(database).list_by_date_range(
+        child.id, today, end
+    )
+    assert instances
+    from custom_components.nestquest.dao_presence import (
+        PresenceOverridesDao,
+    )
+
+    return {
+        "child_id": child.id,
+        "instance_id": instances[0].id,
+        "definition_id": instances[0].definition_id,
+        "override": await PresenceOverridesDao(database).create(
+            child.id, "2027-01-01", "2027-01-02", False
+        ),
+        "today": today,
+        "database": database,
+    }
+
+
+def _admin_success_payload(service: str, ctx: dict) -> dict | None:
+    """A schema-valid, end-to-end-successful payload per admin service."""
+    if service == SERVICE_UNCOMPLETE_QUEST:
+        return {"instance_id": ctx["instance_id"], "actor": "user"}
+    if service == SERVICE_CREATE_QUEST_DEFINITION:
+        return {
+            "title": "Second quest",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [ctx["child_id"]],
+            "windows": ["afternoon"],
+        }
+    if service == SERVICE_UPDATE_QUEST_DEFINITION:
+        return {"definition_id": ctx["definition_id"], "title": "Brush teeth 2"}
+    if service == SERVICE_SET_QUEST_DEFINITION_ACTIVE:
+        return {"definition_id": ctx["definition_id"], "is_active": False}
+    if service == SERVICE_SET_PRESENCE_PATTERN:
+        return {
+            "child_id": ctx["child_id"],
+            "cycle_length_weeks": 1,
+            "anchor_date": ctx["today"],
+            "pattern": {0: [0, 1, 2, 3, 4, 5, 6]},
+        }
+    if service == SERVICE_CREATE_PRESENCE_OVERRIDE:
+        return {
+            "child_id": ctx["child_id"],
+            "start_date": "2027-03-01",
+            "end_date": "2027-03-02",
+            "is_present": False,
+        }
+    if service == SERVICE_DELETE_PRESENCE_OVERRIDE:
+        return {"override_id": ctx["override"].id}
+    if service == SERVICE_EXPORT_HISTORY_CSV:
+        return None  # raises HomeAssistantError (Feature 13) — not Unauthorized
+    if service == SERVICE_MANAGE_CHILD:
+        return {"action": "create", "display_name": "Bo"}
+    if service == SERVICE_REGENERATE:
+        return {}
+    raise AssertionError(f"unexpected service {service}")
+
+
+_ADMIN_ONLY = [
+    service for service, policy in SERVICE_POLICY.items() if policy == ADMIN_ONLY
+]
+
+
+@pytest.mark.parametrize("service", _ADMIN_ONLY)
+async def test_admin_succeeds_on_every_admin_only_service(
+    hass, make_entry, service
+) -> None:
+    from homeassistant.exceptions import HomeAssistantError
+
+    entry = await _setup(hass, make_entry)
+    ctx = await _seed(hass, entry)
+    if service == SERVICE_EXPORT_HISTORY_CSV:
+        # The gate passes; the Feature 13 stub raises its "not
+        # implemented" HomeAssistantError — NOT Unauthorized.
+        with pytest.raises(HomeAssistantError, match="Feature 13"):
+            await hass.services.call(
+                DOMAIN, service, None, context=ADMIN_CTX
+            )
+        return
+    await hass.services.call(
+        DOMAIN, service, _admin_success_payload(service, ctx), context=ADMIN_CTX
+    )
+
+
+@pytest.mark.parametrize("service", _ADMIN_ONLY)
+async def test_non_admin_refused_on_every_admin_only_service(
+    hass, make_entry, service
+) -> None:
+    entry = await _setup(hass, make_entry)
+    ctx = await _seed(hass, entry)
+    if service != SERVICE_EXPORT_HISTORY_CSV:
+        await hass.services.call(
+            DOMAIN,
+            service,
+            _admin_success_payload(service, ctx),
+            context=ADMIN_CTX,
+        )
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            service,
+            _admin_success_payload(service, ctx),
+            context=NON_ADMIN_CTX,
+        )
+
+
+async def test_non_admin_allowed_on_complete_quest(hass, make_entry) -> None:
+    from custom_components.nestquest.completion import instance_state
+
+    entry = await _setup(hass, make_entry)
+    ctx = await _seed(hass, entry)
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {
+            "instance_id": ctx["instance_id"],
+            "actor": "panel",
+            "actor_child_id": ctx["child_id"],
+        },
+        context=NON_ADMIN_CTX,
+    )
+    assert (
+        await instance_state(ctx["database"], ctx["instance_id"]) == "done"
+    )
+
+
+async def test_kiosk_refused_on_uncomplete_and_regenerate(
+    hass, make_entry
+) -> None:
+    entry = await _setup(hass, make_entry)
+    ctx = await _seed(hass, entry)
+    from custom_components.nestquest.admin_allowlist import list_admin_ids
+
+    assert KIOSK_ID not in await list_admin_ids(ctx["database"])
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_UNCOMPLETE_QUEST,
+            {"instance_id": ctx["instance_id"], "actor": "user"},
+            context=KIOSK_CTX,
+        )
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN, SERVICE_REGENERATE, context=KIOSK_CTX
+        )

tests/test_permissions.py

@@ -0,0 +1,122 @@
+"""Tests for the Feature 09 permission gate's audit logging."""
+from __future__ import annotations
+
+import logging
+import re
+
+import pytest
+from homeassistant.exceptions import Unauthorized
+
+from conftest import wire_entry_to_registry
+
+from custom_components.nestquest import async_setup_entry
+from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
+    DOMAIN,
+    SERVICE_MANAGE_CHILD,
+)
+
+ADMIN_ID = "admin-1"
+ADMIN_CTX = {"user_id": ADMIN_ID}
+_TIMESTAMP = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}")
+
+
+async def _set_up(hass, make_entry):
+    entry = wire_entry_to_registry(
+        make_entry(data={CONF_ADMIN_USER_IDS: [ADMIN_ID]}), hass.registry
+    )
+    assert await async_setup_entry(hass, entry) is True
+    return entry
+
+
+async def test_denied_call_logs_warning_with_fields(
+    hass, make_entry, caplog
+) -> None:
+    entry = await _set_up(hass, make_entry)
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+            context={"user_id": "kiosk-panel"},
+        )
+    denials = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.WARNING
+        and "denied" in record.getMessage()
+    ]
+    assert len(denials) == 1
+    message = denials[0].getMessage()
+    assert SERVICE_MANAGE_CHILD in message
+    assert "kiosk-panel" in message
+    assert _TIMESTAMP.search(message), message
+
+
+async def test_denied_no_context_logs_user_none(
+    hass, make_entry, caplog
+) -> None:
+    entry = await _set_up(hass, make_entry)
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+        )
+    denials = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.WARNING
+        and "denied" in record.getMessage()
+    ]
+    assert len(denials) == 1
+    message = denials[0].getMessage()
+    assert "none" in message
+    assert _TIMESTAMP.search(message), message
+
+
+async def test_successful_admin_call_logs_debug_with_fields(
+    hass, make_entry, caplog
+) -> None:
+    caplog.set_level(logging.DEBUG, logger="custom_components.nestquest")
+    entry = await _set_up(hass, make_entry)
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_MANAGE_CHILD,
+        {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
+    )
+    allowed = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.DEBUG
+        and "allowed" in record.getMessage()
+    ]
+    assert len(allowed) == 1
+    message = allowed[0].getMessage()
+    assert SERVICE_MANAGE_CHILD in message
+    assert ADMIN_ID in message
+    assert _TIMESTAMP.search(message), message
+
+
+async def test_denied_log_carries_no_personal_data_beyond_user_id(
+    hass, make_entry, caplog
+) -> None:
+    """The log line names the caller's user id only: the payload's
+    display name must not leak into the audit trail."""
+    entry = await _set_up(hass, make_entry)
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+            context={"user_id": "kiosk-panel"},
+        )
+    denials = [
+        record.getMessage()
+        for record in caplog.records
+        if record.levelno == logging.WARNING and "denied" in record.getMessage()
+    ]
+    assert denials, "expected a denial log line"
+    assert "Ada" not in denials[0]
+    assert "display_name" not in denials[0]

tests/test_service_policy.py

@@ -0,0 +1,32 @@
+"""Tests for the Feature 09 admin-only versus open service registry."""
+from __future__ import annotations
+
+from custom_components.nestquest.const import (
+    DOMAIN_SERVICES,
+    SERVICE_COMPLETE_QUEST,
+)
+from custom_components.nestquest.service_policy import (
+    ADMIN_ONLY,
+    OPEN,
+    SERVICE_POLICY,
+)
+
+
+def test_policy_covers_exactly_the_registered_services() -> None:
+    assert frozenset(SERVICE_POLICY) == frozenset(DOMAIN_SERVICES)
+
+
+def test_only_complete_quest_is_open() -> None:
+    open_services = {
+        name for name, access in SERVICE_POLICY.items() if access == OPEN
+    }
+    assert open_services == {SERVICE_COMPLETE_QUEST}
+    assert all(
+        access == ADMIN_ONLY
+        for name, access in SERVICE_POLICY.items()
+        if name != SERVICE_COMPLETE_QUEST
+    )
+
+
+def test_policy_values_are_only_open_or_admin_only() -> None:
+    assert set(SERVICE_POLICY.values()) <= {OPEN, ADMIN_ONLY}

tests/test_services.py

@@ -6,12 +6,13 @@ from pathlib import Path
 
 import pytest
 import voluptuous as vol
-from homeassistant.exceptions import HomeAssistantError
+from homeassistant.exceptions import HomeAssistantError, Unauthorized
 
 from conftest import wire_entry_to_registry
 
 from custom_components.nestquest import async_setup_entry, async_unload_entry
 from custom_components.nestquest.const import (
+    CONF_ADMIN_USER_IDS,
     DOMAIN,
     DOMAIN_SERVICES,
     SERVICE_COMPLETE_QUEST,
@@ -25,6 +26,7 @@ from custom_components.nestquest.const import (
     SERVICE_SET_QUEST_DEFINITION_ACTIVE,
     SERVICE_UNCOMPLETE_QUEST,
 )
+from custom_components.nestquest.service_policy import ADMIN_ONLY, SERVICE_POLICY
 
 SERVICES_PATH = (
     Path(__file__).parent.parent
@@ -38,6 +40,16 @@ def _wire(entry, registry):
     return wire_entry_to_registry(entry, registry)
 
 
+ADMIN_ID = "admin-1"
+ADMIN_CTX = {"user_id": ADMIN_ID}
+
+
+def _make_admin_entry(make_entry):
+    from custom_components.nestquest.const import CONF_ADMIN_USER_IDS
+
+    return make_entry(data={CONF_ADMIN_USER_IDS: [ADMIN_ID]})
+
+
 def _service_section(text: str, name: str) -> str:
     match = re.search(rf"^{name}:\s*$", text, re.MULTILINE)
     assert match is not None, f"services.yaml must declare {name!r}"
@@ -129,12 +141,13 @@ async def test_unload_one_of_two_entries_keeps_services(
 async def test_manage_child_create_persists_child(hass, make_entry) -> None:
     from custom_components.nestquest.children import list_children
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     await hass.services.call(
         DOMAIN,
         SERVICE_MANAGE_CHILD,
         {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
     )
     children = await list_children(entry.runtime_data.database)
     assert [child.display_name for child in children] == ["Ada"]
@@ -149,12 +162,13 @@ async def test_create_and_set_quest_definition_active(
     )
     from custom_components.nestquest.dao_rules import QuestDefinitionsDao
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     await hass.services.call(
         DOMAIN,
         SERVICE_MANAGE_CHILD,
         {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
     )
     child = (await list_children(entry.runtime_data.database))[0]
     await hass.services.call(
@@ -166,6 +180,7 @@ async def test_create_and_set_quest_definition_active(
             "assignee_child_ids": [child.id],
             "windows": ["morning"],
         },
+        context=ADMIN_CTX,
     )
     active = await list_active_definitions(entry.runtime_data.database)
     assert len(active) == 1
@@ -174,6 +189,7 @@ async def test_create_and_set_quest_definition_active(
         DOMAIN,
         SERVICE_SET_QUEST_DEFINITION_ACTIVE,
         {"definition_id": definition_id, "is_active": False},
+        context=ADMIN_CTX,
     )
     assert await list_active_definitions(entry.runtime_data.database) == []
     stored = await QuestDefinitionsDao(entry.runtime_data.database).get(
@@ -193,12 +209,13 @@ async def test_complete_and_uncomplete_quest_call_business_layer(
     from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
     from custom_components.nestquest.dao_instances import QuestInstancesDao
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     await hass.services.call(
         DOMAIN,
         SERVICE_MANAGE_CHILD,
         {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
     )
     child = (await list_children(entry.runtime_data.database))[0]
     await hass.services.call(
@@ -210,8 +227,9 @@ async def test_complete_and_uncomplete_quest_call_business_layer(
             "assignee_child_ids": [child.id],
             "windows": ["morning"],
         },
+        context=ADMIN_CTX,
     )
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
     today = datetime.date.today().isoformat()
     end = (
         datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
@@ -240,8 +258,9 @@ async def test_complete_and_uncomplete_quest_call_business_layer(
         {
             "instance_id": instance_id,
             "actor": "user",
-            "actor_user_id": "parent-1",
+            "actor_user_id": ADMIN_ID,
         },
+        context=ADMIN_CTX,
     )
     assert (
         await instance_state(entry.runtime_data.database, instance_id)
@@ -258,12 +277,13 @@ async def test_set_presence_pattern_regenerates_child_instances(
     from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
     from custom_components.nestquest.dao_instances import QuestInstancesDao
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     await hass.services.call(
         DOMAIN,
         SERVICE_MANAGE_CHILD,
         {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
     )
     child = (await list_children(entry.runtime_data.database))[0]
     await hass.services.call(
@@ -275,8 +295,9 @@ async def test_set_presence_pattern_regenerates_child_instances(
             "assignee_child_ids": [child.id],
             "windows": ["morning"],
         },
+        context=ADMIN_CTX,
     )
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
     today = datetime.date.today().isoformat()
     end = (
         datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
@@ -292,6 +313,7 @@ async def test_set_presence_pattern_regenerates_child_instances(
             "anchor_date": today,
             "pattern": {0: []},
         },
+        context=ADMIN_CTX,
     )
     assert await dao.list_by_date_range(child.id, today, end) == []
 
@@ -306,12 +328,13 @@ async def test_presence_override_create_and_delete_regenerate(
     from custom_components.nestquest.dao_instances import QuestInstancesDao
     from custom_components.nestquest.dao_presence import PresenceOverridesDao
 
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     await hass.services.call(
         DOMAIN,
         SERVICE_MANAGE_CHILD,
         {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
     )
     child = (await list_children(entry.runtime_data.database))[0]
     await hass.services.call(
@@ -323,8 +346,9 @@ async def test_presence_override_create_and_delete_regenerate(
             "assignee_child_ids": [child.id],
             "windows": ["morning"],
         },
+        context=ADMIN_CTX,
     )
-    await hass.services.call(DOMAIN, SERVICE_REGENERATE)
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
     today = datetime.date.today()
     end = today + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
     dao = QuestInstancesDao(entry.runtime_data.database)
@@ -341,6 +365,7 @@ async def test_presence_override_create_and_delete_regenerate(
             "end_date": end.isoformat(),
             "is_present": False,
         },
+        context=ADMIN_CTX,
     )
     assert (
         await dao.list_by_date_range(
@@ -359,6 +384,7 @@ async def test_presence_override_create_and_delete_regenerate(
         DOMAIN,
         SERVICE_DELETE_PRESENCE_OVERRIDE,
         {"override_id": override.id},
+        context=ADMIN_CTX,
     )
     after = await dao.list_by_date_range(
         child.id, today.isoformat(), end.isoformat()
@@ -369,10 +395,12 @@ async def test_presence_override_create_and_delete_regenerate(
 async def test_export_history_csv_raises_home_assistant_error(
     hass, make_entry
 ) -> None:
-    entry = _wire(make_entry(), hass.registry)
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
     assert await async_setup_entry(hass, entry) is True
     with pytest.raises(HomeAssistantError, match="Feature 13"):
-        await hass.services.call(DOMAIN, SERVICE_EXPORT_HISTORY_CSV)
+        await hass.services.call(
+            DOMAIN, SERVICE_EXPORT_HISTORY_CSV, context=ADMIN_CTX
+        )
 
 
 async def test_invalid_schema_data_surfaces_as_error(hass, make_entry) -> None:
@@ -392,3 +420,174 @@ async def test_invalid_schema_data_surfaces_as_error(hass, make_entry) -> None:
             SERVICE_MANAGE_CHILD,
             {"action": "create"},
         )
+
+
+async def test_admin_only_service_denies_non_admin(hass, make_entry) -> None:
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+            context={"user_id": "kiosk-panel"},
+        )
+
+
+async def test_admin_only_service_denies_missing_user_context(
+    hass, make_entry
+) -> None:
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+        )
+
+
+async def test_admin_only_service_denies_admin_on_empty_allowlist(
+    hass, make_entry
+) -> None:
+    """Even the allowlisted id is denied when the allowlist is empty."""
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+            context=ADMIN_CTX,
+        )
+
+
+async def test_unauthorized_is_raised_before_handler_runs(
+    hass, make_entry
+) -> None:
+    from custom_components.nestquest.children import list_children
+
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN,
+            SERVICE_MANAGE_CHILD,
+            {"action": "create", "display_name": "Ada"},
+            context={"user_id": "kiosk-panel"},
+        )
+    assert await list_children(entry.runtime_data.database) == []
+
+
+_MINIMAL_PAYLOADS = {
+    "uncomplete_quest": {"instance_id": 1, "actor": "user"},
+    "create_quest_definition": {
+        "title": "T",
+        "rule": {"rule_type": "daily"},
+        "assignee_child_ids": [1],
+        "windows": ["morning"],
+    },
+    "update_quest_definition": {"definition_id": 1},
+    "set_quest_definition_active": {"definition_id": 1, "is_active": True},
+    "set_presence_pattern": {
+        "child_id": 1,
+        "cycle_length_weeks": 1,
+        "anchor_date": "2026-09-19",
+        "pattern": {0: [0]},
+    },
+    "create_presence_override": {
+        "child_id": 1,
+        "start_date": "2026-09-19",
+        "end_date": "2026-09-19",
+        "is_present": False,
+    },
+    "delete_presence_override": {"override_id": 1},
+    "export_history_csv": {},
+    "manage_child": {"action": "create", "display_name": "Ada"},
+    "regenerate": {},
+}
+
+_ADMIN_ONLY_SERVICES = [
+    name for name, policy in SERVICE_POLICY.items() if policy == ADMIN_ONLY
+]
+
+
+@pytest.mark.parametrize("service", _ADMIN_ONLY_SERVICES)
+async def test_admin_only_service_fails_closed_without_user_context(
+    hass, make_entry, service
+) -> None:
+    """A call with no user_id is denied on every admin-only service."""
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    with pytest.raises(Unauthorized):
+        await hass.services.call(
+            DOMAIN, service, dict(_MINIMAL_PAYLOADS[service])
+        )
+
+
+async def test_complete_quest_proceeds_without_user_context(
+    hass, make_entry
+) -> None:
+    """The open service needs no user context: the kiosk panel taps a
+    child profile and the call goes through with no HA user at all."""
+    import datetime
+
+    from custom_components.nestquest.children import list_children
+    from custom_components.nestquest.completion import instance_state
+    from custom_components.nestquest.const import DEFAULT_HORIZON_DAYS
+    from custom_components.nestquest.dao_instances import QuestInstancesDao
+
+    entry = _wire(_make_admin_entry(make_entry), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_MANAGE_CHILD,
+        {"action": "create", "display_name": "Ada"},
+        context=ADMIN_CTX,
+    )
+    child = (await list_children(entry.runtime_data.database))[0]
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_CREATE_QUEST_DEFINITION,
+        {
+            "title": "Brush teeth",
+            "rule": {"rule_type": "daily"},
+            "assignee_child_ids": [child.id],
+            "windows": ["morning"],
+        },
+        context=ADMIN_CTX,
+    )
+    await hass.services.call(DOMAIN, SERVICE_REGENERATE, context=ADMIN_CTX)
+    today = datetime.date.today().isoformat()
+    end = (
+        datetime.date.today() + datetime.timedelta(days=DEFAULT_HORIZON_DAYS)
+    ).isoformat()
+    instance_id = (
+        await QuestInstancesDao(
+            entry.runtime_data.database
+        ).list_by_date_range(child.id, today, end)
+    )[0].id
+    await hass.services.call(
+        DOMAIN,
+        SERVICE_COMPLETE_QUEST,
+        {"instance_id": instance_id, "actor": "panel", "actor_child_id": child.id},
+    )
+    assert (
+        await instance_state(entry.runtime_data.database, instance_id)
+        == "done"
+    )
+
+
+async def test_scheduled_internals_never_call_the_service_layer(
+    hass, make_entry
+) -> None:
+    """The daily rollover and startup backfill run the materialization
+    as direct function calls; they never widen the service gate."""
+    import custom_components.nestquest as nestquest
+
+    assert not hasattr(nestquest, "_run_horizon_materialization_service")
+    source = Path(nestquest.__file__).read_text(encoding="utf-8")
+    assert "hass.services.call" not in source, (
+        "scheduled internals must call the materialization directly, "
+        "never through hass.services.call"
+    )