feat(web): a start that waits for the display answers 202 and is delivered in the background

The start route held a request open for up to 45 s while a cold-started
display loaded its plugins; the MQTT bridge (15 s timeout) and browsers
reported a failure for a request that was then delivered.

Now, when no display is listening, the route starts the service if asked
and answers 202 with status "starting" at once. A single worker in the web
process (web_interface/on_demand_dispatch.py) sends the request until the
display acknowledges it or the wait runs out (45 s cold start, 10 s for a
running service without a socket yet). A newer start supersedes the
pending one; a stop cancels it (and succeeds, with cancelled_request_id,
even with no display listening). The outcome is reported by
/display/on-demand/status (source "web": starting, or error with
start-timeout or the socket's reason, until the display publishes
something newer) and by /display/current-status as on_demand_pending.

Callers: the web UI's on-demand modal and "Preview on display" treat
"starting" as taken (an info toast); the MQTT bridge already treats any
non-error 2xx as success (now pinned by a test).

Tests: the dispatcher (ack, retry then ack, start-timeout, other failures,
superseded, an in-flight ack for a superseded start, stop while pending,
a per-start wait, outcome lifetime); the routes (202, status routes while
pending and after a timeout, a later display state replacing the failure,
stop while pending, a new start superseding); a JS suite for app.js.
Mutation check: 20 mutants on the worker, the routes and app.js, 20 killed.

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

test/conftest.py

@@ -328,6 +328,20 @@ def _hermetic_control_socket(monkeypatch):
     monkeypatch.setenv(SOCKET_PATH_ENV, 'off')
 
 
+@pytest.fixture(autouse=True)
+def _no_pending_on_demand_dispatch():
+    """Drop the web process's on-demand dispatcher after each test, so a
+    start one test left pending is not still being sent in the next."""
+    yield
+    module = sys.modules.get('web_interface.on_demand_dispatch')
+    if module is None:
+        return
+    dispatcher = module.current()
+    if dispatcher is not None:
+        dispatcher.cancel('test-teardown')
+    module.reset_for_tests()
+
+
 @pytest.fixture(autouse=True)
 def _hermetic_unit_refresh(monkeypatch, tmp_path_factory):
     """Keep updates' systemd unit refresh off the host.

test/js/run_all.js

@@ -32,7 +32,8 @@ const UNIT = ['unit/test_list_filter.js', 'unit/test_render_cards.js',
               'unit/test_page_registry.js', 'unit/test_core_modules.js',
               'unit/test_overview_reconciliation_poll.js',
               'unit/test_display_partial_ids.js',
-              'unit/test_general_web_login_token.js'];
+              'unit/test_general_web_login_token.js',
+              'unit/test_on_demand_starting.js'];
 const DOM = ['dom/test_installed_dom.js', 'dom/test_store_dom.js', 'dom/test_no_double_fetch.js',
              'dom/test_tools_sections.js', 'dom/test_cache_page.js',
              'dom/test_durations_page.js', 'dom/test_operation_history_page.js',

test/js/unit/test_on_demand_starting.js

@@ -0,0 +1,69 @@
+// POST /api/v3/display/on-demand/start answers 202 with status "starting"
+// when the display service has to be started first: the request is taken,
+// and the web process sends it once the display listens. "Preview on
+// display" (app.js) must read that as taken -- an info toast and the
+// floating preview opened -- not as a failure. Runs the shipped app.js in a
+// vm with a minimal fake DOM, as test_restart_banner.js does.
+
+const fs = require('fs');
+const path = require('path');
+const vm = require('vm');
+const V3 = path.resolve(__dirname, '../../../web_interface/static/v3');
+
+let pass = 0, fail = 0;
+const ok = (label, cond, extra) => cond
+  ? (pass++, console.log('  ok   ' + label))
+  : (fail++, console.log('  FAIL ' + label + (extra !== undefined ? '  ' + JSON.stringify(extra) : '')));
+
+function load(answer) {
+  const notes = [];
+  const opened = [];
+  const noop = () => {};
+  const document = {
+    body: { addEventListener: noop },
+    addEventListener: noop,
+    getElementById: () => null,
+    querySelector: () => null,
+    querySelectorAll: () => [],
+  };
+  const window = { addEventListener: noop, getApp: () => null };
+  const context = {
+    window, document, console,
+    sessionStorage: { setItem: noop, removeItem: noop, getItem: () => null },
+    showNotification: (m, t) => notes.push([m, t]),
+    setTimeout: () => 0,
+    fetch: () => Promise.resolve({ json: () => Promise.resolve(answer) }),
+  };
+  vm.createContext(context);
+  vm.runInContext(fs.readFileSync(path.join(V3, 'app.js'), 'utf8'), context);
+  window.toggleFloatingPreview = (open) => opened.push(open);
+  return { window, notes, opened };
+}
+
+async function preview(answer) {
+  const t = load(answer);
+  t.window.previewPluginNow('weather');
+  for (let i = 0; i < 5; i++) await Promise.resolve();
+  return t;
+}
+
+(async () => {
+  console.log('\npreviewPluginNow');
+  {
+    const t = await preview({ status: 'starting', message: 'The display service is starting',
+                              data: { request_id: 'r1', pending: true } });
+    ok('a 202 "starting" answer is an info toast, not an error',
+       t.notes.length === 1 && t.notes[0][1] === 'info', t.notes);
+    ok('and the preview opens', t.opened.length === 1 && t.opened[0] === true, t.opened);
+  }
+  {
+    const t = await preview({ status: 'success', data: { request_id: 'r1' } });
+    ok('a 200 success still opens it', t.opened.length === 1 && t.notes[0][1] === 'success', t.notes);
+  }
+  {
+    const t = await preview({ status: 'error', message: 'no display' });
+    ok('an error does not', t.opened.length === 0 && t.notes[0][1] === 'error', t.notes);
+  }
+  console.log(`\n${pass} passed, ${fail} failed`);
+  process.exit(fail ? 1 : 0);
+})();

test/test_api_v3_on_demand_restart.py

@@ -16,7 +16,8 @@ This file previously pinned that restart path (it guarded a broken
 ``import _pkg.time`` inside it). The path is gone; these tests pin its
 replacement: a running service is left alone, a stopped one is started (only
 when start_service is set), and the request goes over the control socket
-either way -- to a stopped display once it has started and its socket is up.
+either way -- to a stopped display once it has started and its socket is up,
+sent by the web process's dispatcher after the route has answered 202.
 Nothing is ever written to the cache: the file mailbox is gone (stage 5).
 
 The service helpers are patched where they run. display.py binds
@@ -26,6 +27,7 @@ _run_systemctl_command is the one place a systemctl command is issued.
 """
 
 import sys
+import time
 from pathlib import Path
 from unittest.mock import patch
 
@@ -157,10 +159,18 @@ class TestStartWhileTheServiceIsStopped:
     def test_start_service_starts_it_once_and_never_stops_it(self, api_v3_client, service):
         service["state"]["active"] = False
         response = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
-        assert response.status_code == 200, response.get_json()
+        # Answered at once: the web process sends the request in the
+        # background once the started display listens.
+        assert response.status_code == 202, response.get_json()
+        assert response.get_json()["status"] == "starting"
         assert _systemctl_verbs(service["systemctl"]) == ["start"]
         service["stop_service"].assert_not_called()
-        # Sent once the started display's socket answered.
+        from web_interface import on_demand_dispatch
+        dispatcher = on_demand_dispatch.current()
+        deadline = time.monotonic() + 5
+        while dispatcher.pending() and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert dispatcher.status()["status"] == "delivered"
         assert [s[0] for s in service["sent"]] == ["start"]
         assert _mailbox_writes(service["cache"]) == []
 

test/test_api_v3_on_demand_socket.py

@@ -14,6 +14,7 @@ runs a real server on a temp socket (Linux/macOS only).
 
 import os
 import sys
+import time
 from pathlib import Path
 from unittest.mock import patch
 
@@ -139,89 +140,175 @@ def _attempts(*outcomes, calls=None, clock=None):
     return attempt
 
 
+def _until(predicate, timeout=5.0):
+    end = time.monotonic() + timeout
+    while time.monotonic() < end:
+        if predicate():
+            return True
+        time.sleep(0.005)
+    return False
+
+
 def _no_socket():
     return control_client.ControlError("no_socket", "x", sent=False)
 
 
 class TestNoDisplayListening:
-    """No socket to talk to: the display is stopped, or still starting."""
+    """No socket to talk to: the display is stopped, or still starting.
+
+    The route answers at once (202, ``status: "starting"``) and the web
+    process's dispatcher (web_interface/on_demand_dispatch.py) sends the
+    request until the display acknowledges it; the status routes report the
+    outcome. The dispatcher runs on real time with short waits here.
+    """
+
+    @pytest.fixture(autouse=True)
+    def quick(self, monkeypatch, service):
+        from web_interface import on_demand_dispatch
+        monkeypatch.setattr(on_demand_dispatch, "RETRY_INTERVAL", 0.01)
+        monkeypatch.setattr(on_demand_dispatch, "START_WAIT_SECONDS", 2.0)
+        monkeypatch.setattr(f"{DISPLAY}.ON_DEMAND_SOCKET_WAIT_RUNNING_SECONDS", 1.0)
+        service["cache"].get.return_value = None   # the display has published nothing
+
+    @staticmethod
+    def _outcome(status=None):
+        from web_interface import on_demand_dispatch
+        d = on_demand_dispatch.current()
+        assert d is not None
+        assert _until(lambda: not d.pending()), "the dispatcher never finished"
+        return d.status()
 
     def test_a_display_still_starting_gets_the_request_once_it_listens(
-            self, api_v3_client, service, clock):
+            self, api_v3_client, service):
         calls = []
         with patch(f"{CLIENT}.on_demand_start",
-                   side_effect=_attempts(_no_socket(), _no_socket(), "ack",
-                                         calls=calls, clock=clock)):
+                   side_effect=_attempts(_no_socket(), _no_socket(), "ack", calls=calls)):
             resp = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
-        assert resp.status_code == 200, resp.get_json()
-        data = resp.get_json()["data"]
-        assert data["transport"] == "socket" and "socket_error" not in data
-        assert data["service"]["started"] is False
+            assert resp.status_code == 202, resp.get_json()
+            body = resp.get_json()
+            assert body["status"] == "starting"
+            data = body["data"]
+            assert data["pending"] is True and data["socket_error"] == "no_socket"
+            assert data["service"]["started"] is False
+            outcome = self._outcome()
+        assert outcome["status"] == "delivered"
+        assert outcome["request_id"] == data["request_id"]
         assert len(calls) == 3
         assert not [c for c in service["calls"] if c[0] == "systemctl"]
         assert _mailbox_writes(service["cache"]) == []
 
-    def test_a_stopped_display_is_started_and_then_sent_the_request(
-            self, api_v3_client, service, clock):
+    def test_a_stopped_display_is_started_and_answered_at_once(self, api_v3_client, service):
         service["state"]["active"] = False
-        calls = []
         with patch(f"{CLIENT}.on_demand_start",
-                   side_effect=_attempts(*[_no_socket()] * 6, "ack",
-                                         calls=calls, clock=clock)) as start:
+                   side_effect=_attempts(*[_no_socket()] * 6, "ack")) as start:
+            started = time.monotonic()
             resp = api_v3_client.post(START_URL, json={"plugin_id": "weather",
                                                        "duration": 30})
-        assert resp.status_code == 200, resp.get_json()
-        data = resp.get_json()["data"]
-        assert data["transport"] == "socket"
-        assert data["service"]["started"] is True
+            answered = time.monotonic() - started
+            assert resp.status_code == 202, resp.get_json()
+            data = resp.get_json()["data"]
+            assert data["service"]["started"] is True
+            from web_interface import on_demand_dispatch
+            assert data["wait_seconds"] == on_demand_dispatch.START_WAIT_SECONDS
+            outcome = self._outcome()
+        assert answered < 1.0, "the route waited for the display"
         assert service["calls"] == [("systemctl", "start")]
-        assert len(calls) == 7
+        assert outcome["status"] == "delivered"
         # The same request, the same id, every time.
         assert {call.args[0] for call in start.call_args_list} == {data["request_id"]}
         assert _mailbox_writes(service["cache"]) == []
 
-    def test_a_stopped_display_without_start_service_is_an_error(
-            self, api_v3_client, service, clock):
+    def test_while_pending_the_status_routes_say_starting(self, api_v3_client, service):
         service["state"]["active"] = False
-        with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()) as start:
-            resp = api_v3_client.post(START_URL, json={"plugin_id": "weather",
-                                                       "start_service": False})
-        assert resp.status_code == 400
-        body = resp.get_json()
-        assert "not running" in body["message"]
-        assert body["data"]["socket_error"] == "no_socket"
-        assert start.call_count == 1 and clock.sleeps == 0
-        assert service["calls"] == []
+        with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()):
+            rid = api_v3_client.post(START_URL, json={"plugin_id": "weather"}) \
+                .get_json()["data"]["request_id"]
+            status = api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
+            assert status["source"] == "web"
+            assert status["state"]["status"] == "starting"
+            assert status["state"]["request_id"] == rid
+            assert status["state"]["plugin_id"] == "weather"
+            current = api_v3_client.get("/api/v3/display/current-status").get_json()["data"]
+            assert current["on_demand_pending"]["status"] == "starting"
 
-    def test_a_display_that_never_comes_up_is_given_up_on(self, api_v3_client, service, clock):
-        from web_interface.blueprints.api_v3 import display
+    def test_a_display_that_never_comes_up_is_a_start_timeout(self, api_v3_client, service):
         service["state"]["active"] = False
-        calls = []
-        with patch(f"{CLIENT}.on_demand_start",
-                   side_effect=_attempts(_no_socket(), calls=calls, clock=clock)):
+        with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()):
             resp = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
-        assert resp.status_code == 503
-        body = resp.get_json()
-        assert body["data"]["socket_error"] == "no_socket"
-        assert body["data"]["service"]["started"] is True
-        assert "did not answer within 45 seconds" in body["message"]
-        waited = calls[-1] - calls[1]
-        assert display.ON_DEMAND_SOCKET_WAIT_SECONDS - 1 <= waited \
-            <= display.ON_DEMAND_SOCKET_WAIT_SECONDS
-        assert _mailbox_writes(service["cache"]) == []
+            assert resp.status_code == 202
+            outcome = self._outcome()
+        assert outcome["status"] == "error" and outcome["error"] == "start-timeout"
+        status = api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
+        assert status["state"]["status"] == "error"
+        assert status["state"]["error"] == "start-timeout"
+        current = api_v3_client.get("/api/v3/display/current-status").get_json()["data"]
+        assert current["on_demand_pending"]["error"] == "start-timeout"
+
+    def test_a_later_display_state_replaces_the_failure(self, api_v3_client, service):
+        service["state"]["active"] = False
+        with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()):
+            api_v3_client.post(START_URL, json={"plugin_id": "weather"})
+            failed_at = self._outcome()["last_updated"]
+        later = {"active": True, "status": "active", "plugin_id": "clock",
+                 "last_updated": failed_at + 5}
+        service["cache"].get.return_value = later
+        status = api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
+        assert status["state"]["plugin_id"] == "clock" and status["source"] == "cache"
 
     def test_a_running_service_without_a_socket_is_waited_for_less(
-            self, api_v3_client, service, clock):
-        from web_interface.blueprints.api_v3 import display
+            self, api_v3_client, service):
+        with patch(f"{CLIENT}.on_demand_start", side_effect=_no_socket()):
+            resp = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
+            assert resp.status_code == 202
+            assert resp.get_json()["data"]["wait_seconds"] == 1.0
+            assert self._outcome()["error"] == "start-timeout"
+        assert not [c for c in service["calls"] if c[0] == "systemctl"]
+
+    def test_a_different_failure_while_waiting_ends_it(self, api_v3_client, service):
+        calls = []
+        busy = control_client.ControlError("busy", "x", sent=True)
+        with patch(f"{CLIENT}.on_demand_start",
+                   side_effect=_attempts(_no_socket(), _no_socket(), busy, calls=calls)):
+            assert api_v3_client.post(START_URL, json={"plugin_id": "weather"}).status_code == 202
+            outcome = self._outcome()
+        assert outcome["status"] == "error" and outcome["error"] == "busy"
+        assert len(calls) == 3
+
+    def test_a_stop_while_pending_cancels_it(self, api_v3_client, service):
+        service["state"]["active"] = False
         calls = []
         with patch(f"{CLIENT}.on_demand_start",
-                   side_effect=_attempts(_no_socket(), calls=calls, clock=clock)):
-            resp = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
-        assert resp.status_code == 503
-        assert f"within {int(display.ON_DEMAND_SOCKET_WAIT_RUNNING_SECONDS)} seconds" \
-            in resp.get_json()["message"]
-        assert calls[-1] - calls[0] <= display.ON_DEMAND_SOCKET_WAIT_RUNNING_SECONDS
-        assert not [c for c in service["calls"] if c[0] == "systemctl"]
+                   side_effect=_attempts(_no_socket(), calls=calls)), \
+             patch(f"{CLIENT}.on_demand_stop", side_effect=_no_socket()):
+            rid = api_v3_client.post(START_URL, json={"plugin_id": "weather"}) \
+                .get_json()["data"]["request_id"]
+            resp = api_v3_client.post(STOP_URL, json={})
+            assert resp.status_code == 200, resp.get_json()
+            data = resp.get_json()["data"]
+            assert data["cancelled_request_id"] == rid
+            outcome = self._outcome()
+            n = len(calls)
+            time.sleep(0.05)
+            assert len(calls) == n, "the cancelled start was still being sent"
+        assert outcome["status"] == "idle" and outcome["last_event"] == "requested-stop"
+        status = api_v3_client.get("/api/v3/display/on-demand/status").get_json()["data"]
+        assert status["state"]["status"] == "idle" and status["source"] != "web"
+
+    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()):
+            old = api_v3_client.post(START_URL, json={"plugin_id": "weather"}) \
+                .get_json()["data"]["request_id"]
+        sent = []
+        service["state"]["active"] = True
+        with patch(f"{CLIENT}.on_demand_start",
+                   side_effect=lambda rid, *a: sent.append(rid) or {"accepted": True}):
+            resp = api_v3_client.post(START_URL, json={"plugin_id": "clock"})
+            assert resp.status_code == 200
+            outcome = self._outcome()
+        assert sent == [resp.get_json()["data"]["request_id"]]
+        assert old not in sent
+        assert outcome["status"] == "idle" and outcome["last_event"] == "superseded"
 
     def test_a_service_that_will_not_start_is_an_error(self, api_v3_client, service, clock):
         service["state"]["active"] = False
@@ -233,17 +320,6 @@ class TestNoDisplayListening:
         assert "Failed to start display service" in resp.get_json()["message"]
         assert start.call_count == 1
 
-    def test_a_different_failure_while_waiting_ends_the_wait(self, api_v3_client, service,
-                                                             clock):
-        calls = []
-        busy = control_client.ControlError("busy", "x", sent=True)
-        with patch(f"{CLIENT}.on_demand_start",
-                   side_effect=_attempts(_no_socket(), busy, calls=calls, clock=clock)):
-            resp = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
-        assert resp.status_code == 503
-        assert resp.get_json()["data"]["socket_error"] == "busy"
-        assert len(calls) == 2
-
     def test_stop_with_no_display_running_is_an_error(self, api_v3_client, service, clock):
         service["state"]["active"] = False
         with patch(f"{CLIENT}.on_demand_stop", side_effect=_no_socket()) as stop:

test/test_mqtt_bridge.py

@@ -327,3 +327,27 @@ class TestCleartextIsCalledOut:
     def test_tls_on_does_not_warn(self, bridge_module):
         assert bridge_module.warn_if_cleartext(
             {"mqtt_tls": True, "mqtt_password": "hunter2"}) is False
+
+
+class TestTheApiClientReadsStartingAsTaken:
+    """A cold start answers 202 with ``status: "starting"``: the request is
+    taken and the web process delivers it once the display listens. The
+    bridge must report that as success, not as a failure."""
+
+    def _client(self, bridge_module, status_code, body):
+        response = MagicMock(status_code=status_code)
+        response.json.return_value = body
+        session = MagicMock()
+        session.request.return_value = response
+        return bridge_module.LEDMatrixClient("http://pi:5000", session=session)
+
+    def test_202_starting_is_returned_not_raised(self, bridge_module):
+        api = self._client(bridge_module, 202, {
+            "status": "starting", "message": "starting",
+            "data": {"request_id": "r1", "pending": True}})
+        assert api.start_on_demand(mode="clock") == {"request_id": "r1", "pending": True}
+
+    def test_an_error_is_still_raised(self, bridge_module):
+        api = self._client(bridge_module, 503, {"status": "error", "message": "no display"})
+        with pytest.raises(RuntimeError, match="no display"):
+            api.start_on_demand(mode="clock")

test/test_on_demand_dispatch.py

@@ -0,0 +1,249 @@
+"""The web process's on-demand dispatcher (web_interface/on_demand_dispatch.py).
+
+A start that finds no display listening -- the service was just started, or
+is still loading its plugins -- is answered at once (202), and the
+dispatcher's one worker thread sends it again until the display
+acknowledges it or the wait runs out. These tests drive the worker with a
+fake ``send`` and short waits:
+
+* acknowledged: delivered once, and reported as such;
+* nothing listening for the whole wait: ``start-timeout``;
+* any other failure: reported at once, not retried;
+* a newer start supersedes the pending one; a stop cancels it;
+* the outcome is reported for a while, then forgotten.
+"""
+
+import threading
+import time
+
+import pytest
+
+from src.ipc import client as control_client
+from web_interface import on_demand_dispatch
+from web_interface.on_demand_dispatch import OnDemandDispatcher
+
+
+def _not_listening():
+    return control_client.ControlError("no_socket", "x", sent=False)
+
+
+def _payload(rid, plugin_id="weather"):
+    return {"request_id": rid, "action": "start", "plugin_id": plugin_id,
+            "mode": plugin_id, "duration": 30, "pinned": False}
+
+
+class FakeSend:
+    """Answers with ``outcomes`` in turn (an exception is raised, anything
+    else acks); the last one repeats. Records the request ids it was sent."""
+
+    def __init__(self, *outcomes):
+        self.outcomes = list(outcomes) or ["ack"]
+        self.sent = []
+        self.lock = threading.Lock()
+
+    def __call__(self, payload):
+        with self.lock:
+            self.sent.append(payload["request_id"])
+            outcome = self.outcomes.pop(0) if len(self.outcomes) > 1 else self.outcomes[0]
+        if isinstance(outcome, BaseException):
+            raise outcome
+        if callable(outcome):
+            return outcome(payload)
+        return {"accepted": True}
+
+
+def _until(predicate, timeout=5.0):
+    end = time.monotonic() + timeout
+    while time.monotonic() < end:
+        if predicate():
+            return True
+        time.sleep(0.005)
+    return False
+
+
+def _settled(d):
+    return _until(lambda: not d.pending() and d._thread is None)
+
+
+@pytest.fixture
+def make():
+    made = []
+
+    def build(send, wait_seconds=2.0, retry_interval=0.01):
+        d = OnDemandDispatcher(send, wait_seconds=wait_seconds, retry_interval=retry_interval)
+        made.append(d)
+        return d
+    yield build
+    for d in made:
+        d.cancel("teardown")
+
+
+class TestDelivery:
+    def test_an_acknowledged_start_is_delivered_once(self, make):
+        send = FakeSend("ack")
+        d = make(send)
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert send.sent == ["r1"]
+        status = d.status()
+        assert status["status"] == "delivered" and status["request_id"] == "r1"
+        assert status["source"] == "web"
+
+    def test_it_is_sent_again_until_the_display_listens(self, make):
+        send = FakeSend(_not_listening(), _not_listening(), "ack")
+        d = make(send)
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert send.sent == ["r1", "r1", "r1"]
+        assert d.status()["status"] == "delivered"
+
+    def test_while_it_waits_it_reports_starting(self, make):
+        gate = threading.Event()
+
+        def blocked(payload):
+            gate.wait(5)
+            raise _not_listening()
+
+        send = FakeSend(blocked, "ack")
+        d = make(send)
+        d.submit(_payload("r1"))
+        status = d.status()
+        assert status["status"] == "starting" and status["active"] is False
+        assert (status["plugin_id"], status["mode"], status["duration"]) == ("weather", "weather", 30)
+        gate.set()
+        assert _settled(d)
+        assert d.status()["status"] == "delivered"
+
+
+class TestGivingUp:
+    def test_nothing_listening_for_the_whole_wait_is_a_start_timeout(self, make):
+        send = FakeSend(_not_listening())
+        d = make(send, wait_seconds=0.2)
+        started = time.monotonic()
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        waited = time.monotonic() - started
+        status = d.status()
+        assert status["status"] == "error" and status["error"] == "start-timeout"
+        assert 0.15 <= waited < 2.0
+        assert len(send.sent) > 3   # it kept trying in between
+
+    def test_a_start_can_carry_its_own_wait(self, make):
+        # The route passes a shorter wait for a service that was already
+        # running; the dispatcher's default must not override it.
+        d = make(FakeSend(_not_listening()), wait_seconds=30.0)
+        d.submit(_payload("r1"), wait_seconds=0.1)
+        assert _until(lambda: not d.pending(), timeout=3.0), "it waited the default"
+        assert d.status()["error"] == "start-timeout"
+
+    @pytest.mark.parametrize("reason,sent", [("busy", True), ("unknown_command", True),
+                                             ("timeout", True), ("forbidden", False)])
+    def test_any_other_failure_is_reported_at_once(self, make, reason, sent):
+        send = FakeSend(control_client.ControlError(reason, "x", sent=sent))
+        d = make(send)
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert send.sent == ["r1"]
+        status = d.status()
+        assert status["status"] == "error" and status["error"] == reason
+
+    def test_a_client_bug_is_internal(self, make):
+        d = make(FakeSend(RuntimeError("boom")))
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert d.status()["error"] == "internal"
+
+
+class TestOneAtATime:
+    def test_a_newer_start_supersedes_the_pending_one(self, make):
+        send = FakeSend(_not_listening())
+        d = make(send)
+        d.submit(_payload("old"))
+        assert _until(lambda: "old" in send.sent)
+        d.submit(_payload("new", plugin_id="clock"))
+        assert d.status()["request_id"] == "new"
+        n = len(send.sent)
+        send.outcomes = ["ack"]
+        assert _settled(d)
+        assert send.sent[n:] and set(send.sent[n + 1:]) <= {"new"}
+        assert send.sent[-1] == "new"
+        status = d.status()
+        assert status["status"] == "delivered" and status["plugin_id"] == "clock"
+
+    def test_an_answer_for_a_superseded_start_is_not_reported(self, make):
+        # The old start's send is in flight when the new one arrives; its
+        # ack must not mark the new one delivered.
+        in_flight, release = threading.Event(), threading.Event()
+
+        def slow(payload):
+            in_flight.set()
+            release.wait(5)
+            return {"accepted": True}
+
+        send = FakeSend(slow, _not_listening())
+        d = make(send, wait_seconds=0.3)
+        d.submit(_payload("old"))
+        assert in_flight.wait(5)
+        d.submit(_payload("new"))
+        release.set()
+        assert _settled(d)
+        status = d.status()
+        assert status["request_id"] == "new"
+        assert status["status"] == "error" and status["error"] == "start-timeout"
+
+    def test_a_stop_cancels_the_pending_start(self, make):
+        send = FakeSend(_not_listening())
+        d = make(send)
+        d.submit(_payload("r1"))
+        assert _until(lambda: send.sent)
+        assert d.cancel("requested-stop") == "r1"
+        assert _settled(d)
+        n = len(send.sent)
+        time.sleep(0.05)
+        assert len(send.sent) == n, "it kept sending a cancelled start"
+        status = d.status()
+        assert status["status"] == "idle" and status["last_event"] == "requested-stop"
+
+    def test_a_cancel_with_nothing_pending_does_nothing(self, make):
+        d = make(FakeSend("ack"))
+        assert d.cancel() is None
+        assert d.status() is None
+
+    def test_a_cancel_after_delivery_leaves_the_outcome(self, make):
+        d = make(FakeSend("ack"))
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert d.cancel() is None
+        assert d.status()["status"] == "delivered"
+
+
+class TestOutcomeLifetime:
+    def test_an_outcome_is_forgotten_after_a_while(self, make, monkeypatch):
+        d = make(FakeSend("ack"))
+        d.submit(_payload("r1"))
+        assert _settled(d)
+        assert d.status() is not None
+        monkeypatch.setattr(on_demand_dispatch, "OUTCOME_SECONDS", 0.0)
+        time.sleep(0.01)
+        assert d.status() is None
+
+    def test_a_pending_start_never_expires(self, make, monkeypatch):
+        monkeypatch.setattr(on_demand_dispatch, "OUTCOME_SECONDS", 0.0)
+        gate = threading.Event()
+        d = make(FakeSend(lambda p: gate.wait(5) and {"accepted": True}))
+        d.submit(_payload("r1"))
+        time.sleep(0.01)
+        assert d.status()["status"] == "starting"
+        gate.set()
+        assert _settled(d)
+
+
+def test_the_process_has_one_dispatcher():
+    on_demand_dispatch.reset_for_tests()
+    try:
+        assert on_demand_dispatch.current() is None
+        first = on_demand_dispatch.get_dispatcher(FakeSend())
+        assert on_demand_dispatch.get_dispatcher(FakeSend()) is first
+        assert on_demand_dispatch.current() is first
+    finally:
+        on_demand_dispatch.reset_for_tests()