complete_quest proxies to the API (task 47c7c90e) (#169)

custom_components/nestquest/services.py

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

tests/conftest.py

@@ -757,6 +757,16 @@ class LazyLocalSnapshotClient:
     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
@@ -767,6 +777,29 @@ class LazyLocalSnapshotClient:
         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
@@ -776,14 +809,14 @@ class LazyLocalSnapshotClient:
         from custom_components.nestquest.core import snapshot as core_snapshot
 
         self.calls += 1
-        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):
+        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
@@ -813,6 +846,44 @@ class LazyLocalSnapshotClient:
             ],
         }
 
+    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():
@@ -858,6 +929,59 @@ def set_coordinator_client(entry, client):
     return client
 
 
+async def refire_api_transitions(hass, entry, client):
+    """Deliver the SSE frames the stand-in client has published so far
+    through the integration's SSE subscription — the production event
+    path.
+
+    Since Feature 18 the ``complete_quest`` service fires no local
+    bus event: the API service builds and publishes the transition
+    frames on its SSE stream, and the integration's subscription
+    (:mod:`custom_components.nestquest.sse`) re-fires each frame on
+    the bus.  This helper streams the frames the stand-in client
+    published (``client.published_frames``) through a real
+    :class:`~custom_components.nestquest.sse.NestQuestEventStream`
+    and waits (bounded) until the bus has re-fired every one of them.
+    """
+    import asyncio
+
+    from custom_components.nestquest.sse import NestQuestEventStream
+
+    frames, client.published_frames = client.published_frames, []
+
+    class _ScriptedStream:
+        """A healthy stream: the frames, then it stays open."""
+
+        def __init__(self, frames) -> None:
+            self._frames = frames
+
+        def stream_events(self):
+            async def _gen():
+                for frame in self._frames:
+                    yield frame
+                await asyncio.Event().wait()
+
+            return _gen()
+
+    before = {etype: len(hass.bus.fired(etype)) for etype, _ in frames}
+    expected = len(frames)
+    manager = NestQuestEventStream(hass, _ScriptedStream(frames))
+    manager.start()
+    try:
+        async with asyncio.timeout(2):
+            while True:
+                delivered = sum(
+                    len(hass.bus.fired(etype)) - before[etype]
+                    for etype, _ in frames
+                )
+                if delivered >= expected:
+                    break
+                await asyncio.sleep(0.01)
+    finally:
+        await manager.stop()
+    return frames
+
+
 class _HassNamespace(SimpleNamespace):
     """SimpleNamespace subclass so the hass stub can hold weak references."""
 

tests/test_complete_quest_proxy.py

@@ -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 == []

tests/test_entity_matrix.py

@@ -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"])
@@ -234,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)),

tests/test_events.py

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