fix(web): a delivered on-demand start reads as starting until the display acts on it

On ledpi (three cold starts) the display acknowledged the start as its
socket opened, then took ~5 s to act on it while Vegas built its first
strip; the status routes meanwhile showed the display's own idle state, so
a UI polling every 700 ms flashed idle.

The display's on-demand state now names the request it answers
(request_id). A delivered start keeps reading as status "starting" with
delivered: true, in /display/on-demand/status and as on_demand_pending in
/display/current-status, until the display publishes state for that
request id (a display without the field: any state newer than the
delivery), for at most DELIVERED_SHOWN_SECONDS (30 s). The display's
startup state, which can be published after the acknowledgement, names no
request and does not end it.

Tests: stays starting against the startup idle state (no id, an older id);
the matching active state and the matching error take over; an older
display's newer state takes over; the 30 s cap; the display's state names
its request. Mutation check: 11 mutants, 11 killed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

test/test_api_v3_on_demand_socket.py

@@ -15,6 +15,7 @@ runs a real server on a temp socket (Linux/macOS only).
 import os
 import sys
 import time
+import types
 from pathlib import Path
 from unittest.mock import patch
 
@@ -294,6 +295,86 @@ class TestNoDisplayListening:
         status = api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
         assert status["state"]["status"] == "idle" and status["source"] != "web"
 
+    # -- delivered, but the display has not acted on it yet ----------------
+    # The display acknowledges a start when its socket opens and acts on it
+    # seconds later (Vegas builds its first strip first: ~5 s on ledpi);
+    # meanwhile it publishes its own idle state, which must not flash.
+
+    def _delivered(self, api_v3_client, service):
+        with patch(f"{CLIENT}.on_demand_start",
+                   side_effect=_attempts(_no_socket(), "ack")):
+            data = api_v3_client.post(START_URL, json={"plugin_id": "weather"})                 .get_json()["data"]
+            outcome = self._outcome()
+        assert outcome["status"] == "delivered"
+        return data["request_id"], outcome["last_updated"]
+
+    @staticmethod
+    def _status(api_v3_client):
+        return api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
+
+    def test_a_delivered_start_stays_starting_until_the_display_answers(
+            self, api_v3_client, service):
+        rid, delivered_at = self._delivered(api_v3_client, service)
+        # The display's startup state: idle, published after the ack, for
+        # no request (or an older one).
+        for older in (None, "an-older-request"):
+            service["cache"].get.return_value = {
+                "active": False, "status": "idle", "request_id": older,
+                "last_updated": delivered_at + 1}
+            data = self._status(api_v3_client)
+            assert data["source"] == "web", older
+            assert data["state"]["status"] == "starting"
+            assert data["state"]["delivered"] is True
+            assert data["state"]["request_id"] == rid
+            current = api_v3_client.get("/api/v3/display/current-status").get_json()["data"]
+            assert current["on_demand_pending"]["delivered"] is True
+
+    def test_the_display_state_for_the_request_takes_over(self, api_v3_client, service):
+        rid, delivered_at = self._delivered(api_v3_client, service)
+        service["cache"].get.return_value = {
+            "active": True, "status": "active", "plugin_id": "weather",
+            "request_id": rid, "last_updated": delivered_at + 5}
+        data = self._status(api_v3_client)
+        assert data["source"] == "cache" and data["state"]["status"] == "active"
+        current = api_v3_client.get("/api/v3/display/current-status").get_json()["data"]
+        assert "on_demand_pending" not in current
+
+    def test_its_error_takes_over_too(self, api_v3_client, service):
+        rid, delivered_at = self._delivered(api_v3_client, service)
+        service["cache"].get.return_value = {
+            "active": False, "status": "error", "error": "load-failed",
+            "request_id": rid, "last_updated": delivered_at + 5}
+        assert self._status(api_v3_client)["state"]["error"] == "load-failed"
+
+    def test_an_older_display_answers_with_any_newer_state(self, api_v3_client, service):
+        # A display before request_id was published: timestamps decide.
+        _, delivered_at = self._delivered(api_v3_client, service)
+        service["cache"].get.return_value = {"active": False, "status": "idle",
+                                             "last_updated": delivered_at - 1}
+        assert self._status(api_v3_client)["source"] == "web"
+        service["cache"].get.return_value = {"active": True, "status": "active",
+                                             "last_updated": delivered_at + 1}
+        assert self._status(api_v3_client)["source"] == "cache"
+
+    def test_a_delivered_start_is_shown_for_a_limited_time(
+            self, api_v3_client, service, monkeypatch):
+        from web_interface import on_demand_dispatch
+        _, delivered_at = self._delivered(api_v3_client, service)
+        service["cache"].get.return_value = {"active": False, "status": "idle",
+                                             "request_id": None,
+                                             "last_updated": delivered_at + 1}
+        cap = on_demand_dispatch.DELIVERED_SHOWN_SECONDS
+        assert cap == 30.0
+        clock = types.SimpleNamespace(time=lambda: delivered_at + cap - 1,
+                                      monotonic=time.monotonic, sleep=time.sleep)
+        monkeypatch.setattr("web_interface.blueprints.api_v3.time", clock)
+        assert self._status(api_v3_client)["source"] == "web"
+        clock.time = lambda: delivered_at + cap + 1
+        data = self._status(api_v3_client)
+        assert data["source"] == "cache" and data["state"]["status"] == "idle"
+        current = api_v3_client.get("/api/v3/display/current-status").get_json()["data"]
+        assert "on_demand_pending" not in current
+
     def test_a_new_start_supersedes_the_pending_one(self, api_v3_client, service):
         service["state"]["active"] = False
         with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()):