fix: close Audio Router self-echo PUBACK/subscription-delivery race

publish_router_mode registered its self-echo correlation marker before
publishing but removed it in a finally block the instant the QoS 1 publish()
call observed its PUBACK. A broker may legally deliver a Controller's own
subscription copy of that same publish after the PUBACK, since MQTT does not
order the two; when that happened, the Controller mistook its own late echo
for live operator intent and relinquished automatic restoration ownership.

Keep the marker claimable for a short, bounded mqtt.router_echo_grace_seconds
window (default 2s) after the PUBACK instead of clearing it immediately.
Genuine external/operator commands are unaffected: once the grace window
elapses, a matching payload is still correctly treated as operator intent
and ownership is relinquished as before.

Verified with deterministic fake-clock unit tests plus a real Mosquitto
broker integration test (tests/test_router_echo_mosquitto_integration.py,
skipped automatically when Docker is unavailable).

README.md

@@ -345,8 +345,12 @@ when all of these remain true:
 - No live, non-self mode command and no conflicting state change indicated an
   operator override.
 
-Controller subscribes to both Router mode set and state topics. Its exact
-in-flight echo is recognized; every other live mode-set publication is treated
+Controller subscribes to both Router mode set and state topics. Its own
+mode-set publication is correlated by exact payload and stays recognized as
+a self-echo both while the publish is in flight and, because a broker may
+legally deliver a subscription echo after the PUBACK that acknowledged the
+same publish, for a short `mqtt.router_echo_grace_seconds` window afterward
+(default 2 seconds). Every other live mode-set publication is treated
 conservatively as operator intent, even if it repeats `video`. A retained set
 message is ignored as stale intent. Controller does not publish Router state;
 it accepts a received live or retained state message as the Router's current
@@ -927,8 +931,11 @@ PYTHONDONTWRITEBYTECODE=1 .venv/bin/python -m unittest discover -s tests -v
 
 - The web form is unauthenticated and listens on `0.0.0.0` by default. Restrict
   the published port with host firewall/network ACLs or place it behind an
-  authenticated reverse proxy. URL validation checks only the `http`/`https`
-  shape; yt-dlp performs the outbound request.
+  authenticated reverse proxy. Submitted URLs must use `http`/`https`, carry no
+  embedded userinfo, and resolve (by exact, normalized hostname match) to an
+  official YouTube host: `youtube.com`, `www.youtube.com`, `m.youtube.com`,
+  `music.youtube.com`, or `youtu.be`. URL, submitter, note, and total request
+  size are also server-side bounded. yt-dlp performs the outbound request.
 - MQTT username/password support is available, but application-level TLS is not
   configured here. Keep this connection on a trusted/protected network or add a
   separately managed secure tunnel/proxy, and grant the Controller only the

app.py

@@ -224,12 +224,36 @@ HTML_FORM = """
 </html>
 """
 
+ALLOWED_YOUTUBE_HOSTS = frozenset(
+    {
+        "youtube.com",
+        "www.youtube.com",
+        "m.youtube.com",
+        "music.youtube.com",
+        "youtu.be",
+    }
+)
+
+MAX_URL_LENGTH = 2048
+MAX_SUBMITTER_LENGTH = 200
+MAX_NOTE_LENGTH = 1000
+MAX_REQUEST_CONTENT_LENGTH = 32 * 1024
+
+
 def _valid_media_url(value: str) -> bool:
+    if not value or len(value) > MAX_URL_LENGTH:
+        return False
     try:
         parsed = urlsplit(value)
+        hostname = parsed.hostname
+        has_userinfo = parsed.username is not None or parsed.password is not None
     except ValueError:
         return False
-    return parsed.scheme.lower() in {"http", "https"} and bool(parsed.netloc)
+    if parsed.scheme.lower() not in {"http", "https"}:
+        return False
+    if has_userinfo or not hostname:
+        return False
+    return hostname.lower() in ALLOWED_YOUTUBE_HOSTS
 
 
 def _submission_message(result: dict[str, object]) -> str:
@@ -277,6 +301,7 @@ def create_app(
     web_app.config.update(
         TESTING=testing,
         SECRET_KEY=secret_key or "controller-test-only-secret",
+        MAX_CONTENT_LENGTH=MAX_REQUEST_CONTENT_LENGTH,
     )
     web_app.extensions["controller_service"] = service
     web_app.extensions["controller_config"] = config
@@ -285,6 +310,11 @@ def create_app(
     def index():
         return render_template_string(HTML_FORM)
 
+    @web_app.errorhandler(413)
+    def _submission_too_large(_error):
+        flash("The submission was too large.")
+        return redirect(url_for("index"))
+
     @web_app.post("/submit")
     def submit():
         url = request.form.get("url", "").strip()
@@ -292,7 +322,16 @@ def create_app(
         note = request.form.get("note", "").strip() or None
 
         if not _valid_media_url(url):
-            flash("Invalid URL. Use a complete http:// or https:// address.")
+            flash(
+                "Invalid URL. Only YouTube links are accepted "
+                "(e.g. https://www.youtube.com/... or https://youtu.be/...)."
+            )
+            return redirect(url_for("index"))
+        if submitter is not None and len(submitter) > MAX_SUBMITTER_LENGTH:
+            flash(f"Your name must be {MAX_SUBMITTER_LENGTH} characters or fewer.")
+            return redirect(url_for("index"))
+        if note is not None and len(note) > MAX_NOTE_LENGTH:
+            flash(f"The note must be {MAX_NOTE_LENGTH} characters or fewer.")
             return redirect(url_for("index"))
 
         try:

config.yaml

@@ -10,6 +10,11 @@ mqtt:
   command_queue_size: 512
   # QoS 1 is accepted only after the broker PUBACKs within this bound.
   publish_ack_timeout_seconds: 2
+  # Audio Router self-echo correlation stays claimable for this long after
+  # a Controller-originated mode-set publish is PUBACK'd, so a subscription
+  # echo that is delivered after the PUBACK is still recognized as ours
+  # instead of being mistaken for operator intent.
+  router_echo_grace_seconds: 2
 
 topics:
   player_command: youtube/player/cmd

controller/config.py

@@ -33,6 +33,7 @@ class MqttConfig:
     reconnect_max_seconds: int = 30
     command_queue_size: int = 512
     publish_ack_timeout_seconds: float = 2.0
+    router_echo_grace_seconds: float = 2.0
 
 
 @dataclass(frozen=True)
@@ -219,6 +220,11 @@ def load_config(
             "mqtt.publish_ack_timeout_seconds",
             minimum=0.1,
         ),
+        router_echo_grace_seconds=_number(
+            mqtt_raw.get("router_echo_grace_seconds", 2.0),
+            "mqtt.router_echo_grace_seconds",
+            minimum=0.0,
+        ),
     )
     if mqtt.reconnect_max_seconds < mqtt.reconnect_min_seconds:
         raise ConfigError("mqtt.reconnect_max_seconds must be >= reconnect_min_seconds")

controller/mqtt_adapter.py

@@ -6,6 +6,7 @@ import json
 import logging
 import queue
 import threading
+import time
 import uuid
 from dataclasses import dataclass
 from typing import Any, Mapping
@@ -58,6 +59,7 @@ class MqttAdapter:
         *,
         client: Any | None = None,
         events: "queue.Queue[MqttEvent] | None" = None,
+        monotonic=time.monotonic,
     ):
         self.config = config
         self.events: "queue.Queue[MqttEvent]" = events or queue.Queue(
@@ -73,8 +75,16 @@ class MqttAdapter:
         self._started = False
         self._loop_started = False
         self._bootstrap_thread: threading.Thread | None = None
+        self._monotonic = monotonic
         self._router_echo_lock = threading.Lock()
-        self._pending_router_echoes: dict[str, bytes] = {}
+        # token -> (exact payload, grace deadline). The deadline is ``None``
+        # while the correlated publish() call has not yet observed its
+        # PUBACK; such an entry never expires on its own, bounded instead by
+        # the publish ack timeout. Once PUBACK is observed, the entry keeps
+        # a short grace window so a subscription echo delivered just after
+        # the PUBACK is still recognized as ours rather than mistaken for
+        # operator intent.
+        self._pending_router_echoes: dict[str, tuple[bytes, float | None]] = {}
         self.last_error: str | None = None
         self._configure_client()
 
@@ -466,15 +476,23 @@ class MqttAdapter:
     def publish_router_mode(self, mode: str) -> PublishReceipt:
         # A broker normally sends our publication back through the matching
         # subscription. Register the exact payload before publishing because
-        # Paho's network thread may receive that echo before publish() returns.
-        # The marker is removed as soon as the QoS 1 publish call completes;
-        # this intentionally prefers a false ownership loss over mistaking a
-        # later same-value operator command for a Controller echo.
+        # Paho's network thread may receive that echo before publish()
+        # returns. A PUBACK only proves broker receipt of our publish, not
+        # that the broker has already forwarded (or even scheduled) our own
+        # subscription copy of it; that copy can legally arrive after the
+        # PUBACK. So the marker is not removed the instant publish()
+        # completes: it stays claimable for a short bounded grace window
+        # afterward, and is removed immediately once actually claimed or
+        # once the grace window elapses. This closes the specific race where
+        # the Controller's own late echo would otherwise be mistaken for
+        # live operator intent and cause automatic restoration ownership to
+        # be relinquished.
         echo_token = uuid.uuid4().hex
+        payload = mode.encode("utf-8")
         with self._router_echo_lock:
-            self._pending_router_echoes[echo_token] = mode.encode("utf-8")
-        try:
-            return self._publish(
+            self._prune_router_echoes_locked()
+            self._pending_router_echoes[echo_token] = (payload, None)
+        receipt = self._publish(
             self.config.topics.router_mode_set,
             mode,
             qos=ROUTER_COMMAND_QOS,
@@ -482,20 +500,39 @@ class MqttAdapter:
             require_connection=True,
             invalidate_transport_on_uncertain=True,
         )
-        finally:
+        grace_deadline = (
+            self._monotonic() + self.config.mqtt.router_echo_grace_seconds
+        )
         with self._router_echo_lock:
-                self._pending_router_echoes.pop(echo_token, None)
+            if echo_token in self._pending_router_echoes:
+                self._pending_router_echoes[echo_token] = (payload, grace_deadline)
+        return receipt
 
     def _claim_router_echo(self, payload: bytes) -> bool:
-        """Claim one exact mode-set echo from an in-flight local publish."""
+        """Claim one exact mode-set echo from an in-flight or recent local publish."""
 
         with self._router_echo_lock:
-            for token, expected in tuple(self._pending_router_echoes.items()):
+            self._prune_router_echoes_locked()
+            for token, (expected, _deadline) in tuple(
+                self._pending_router_echoes.items()
+            ):
                 if expected == payload:
                     del self._pending_router_echoes[token]
                     return True
         return False
 
+    def _prune_router_echoes_locked(self) -> None:
+        """Drop expired grace-window entries. Caller holds ``_router_echo_lock``."""
+
+        now = self._monotonic()
+        expired = [
+            token
+            for token, (_, deadline) in self._pending_router_echoes.items()
+            if deadline is not None and deadline <= now
+        ]
+        for token in expired:
+            del self._pending_router_echoes[token]
+
     def publish_legacy(self, topic: str, payload: str, *, retain: bool = True) -> PublishReceipt:
         return self._publish(
             topic,

controller/service.py

@@ -81,6 +81,7 @@ class ControllerService:
         self.mqtt = mqtt_adapter or MqttAdapter(
             config,
             events=self.events,
+            monotonic=monotonic,
         )
         self.downloads = download_manager
         self.playback = playback
@@ -404,13 +405,15 @@ class ControllerService:
     def _handle_call(self, action: str, payload: Any) -> Any:
         if action == "enqueue_download":
             assert self.downloads is not None
-            if not self._cleanup_allowed():
+            url = payload["url"]
+            if not self._cleanup_allowed() and self._ingest_requires_evidence(url):
                 raise ControllerUnavailableError(
-                    "downloads are deferred until fresh Player evidence or "
-                    "administrative playback reconciliation"
+                    "refreshing or replacing existing playlist media is deferred "
+                    "until fresh Player evidence or administrative playback "
+                    "reconciliation"
                 )
             reservation = self.downloads.enqueue(
-                payload["url"], payload.get("submitter"), payload.get("note")
+                url, payload.get("submitter"), payload.get("note")
             )
             result: dict[str, Any] = {
                 "status": "queued" if reservation.status == "reserved" else reservation.status,
@@ -1261,6 +1264,20 @@ class ControllerService:
     def _cleanup_allowed(self) -> bool:
         return self._player_evidence_ready or self._operator_cleanup_override
 
+    def _ingest_requires_evidence(self, url: Any) -> bool:
+        """True when ingesting ``url`` could reuse/refresh occupied media.
+
+        A URL that does not already occupy a slot can only land in a slot
+        that ``reserve_download`` independently proves is empty and
+        unguarded, so it cannot mutate, delete, or replace media the Player
+        may currently be using. Only the reuse/refresh path needs the same
+        conservative evidence gate as deletion and expiry.
+        """
+
+        if not isinstance(url, str):
+            return False
+        return self.store.find_occupied_slot_for_url(url) is not None
+
     def _schedule_deletion_recovery(self) -> None:
         if not self._cleanup_allowed():
             return

controller/store.py

@@ -962,6 +962,31 @@ class ControllerStore:
             (slot,),
         ).fetchone() is not None
 
+    def find_occupied_slot_for_url(self, url: str) -> int | None:
+        """Return the slot already holding ``url``, if any.
+
+        A non-``None`` result means ingest for this exact URL would reuse or
+        refresh a slot that already holds media, rather than land in a slot
+        that is genuinely empty. Callers use this to decide whether an
+        ingest request needs fresh Player evidence before proceeding.
+        """
+
+        if not isinstance(url, str):
+            return None
+        url = url.strip()
+        if not url:
+            return None
+        with self._lock:
+            row = self.connection.execute(
+                """
+                SELECT slot_number FROM slots
+                WHERE url = ? AND media_state <> 'empty'
+                ORDER BY slot_number LIMIT 1
+                """,
+                (url,),
+            ).fetchone()
+            return None if row is None else int(row["slot_number"])
+
     def mark_download_running(self, job_id: str, slot: int, generation: int) -> bool:
         with self._transaction() as connection:
             now = self._now()

tests/test_app.py

@@ -53,7 +53,13 @@ class FlaskFactoryTests(unittest.TestCase):
     def _app(self, service):
         return self._factory()(self.config, service, testing=True)
 
-    def _post(self, service, *, url="https://example.com/playlist", **fields):
+    def _post(
+        self,
+        service,
+        *,
+        url="https://www.youtube.com/playlist?list=PL1234567890",
+        **fields,
+    ):
         app = self._app(service)
         return app.test_client().post(
             "/submit",
@@ -71,15 +77,29 @@ class FlaskFactoryTests(unittest.TestCase):
         self.assertIn(b'name="submitter"', response.data)
         self.assertIn(b'name="note"', response.data)
 
-    def test_submit_accepts_only_http_or_https_urls_with_a_network_location(self):
+    def test_submit_rejects_non_youtube_and_malformed_urls(self):
         invalid_urls = (
             "",
             "http",
             "https:///missing-host",
-            "httpfoo://example.com/video",
-            "ftp://example.com/video",
+            "httpfoo://youtube.com/video",
+            "ftp://youtube.com/video",
             "javascript:alert(1)",
             "/relative/path",
+            # Arbitrary external hosts, including lookalikes and substring
+            # tricks, must be rejected by exact normalized hostname match.
+            "https://example.com/video",
+            "https://notyoutube.com/video",
+            "https://youtube.com.evil.com/video",
+            "https://evil.com/?u=youtube.com",
+            "https://xyoutube.com/watch?v=abc",
+            "https://youtube.com.attacker.net/watch?v=abc",
+            # Embedded userinfo must be rejected even when the host matches.
+            "https://user:[email protected]/watch?v=abc",
+            "https://[email protected]/watch?v=abc",
+            "https://[email protected]/watch?v=abc",
+            # Oversized URL.
+            "https://www.youtube.com/watch?v=" + ("a" * 3000),
         )
         for url in invalid_urls:
             with self.subTest(url=url):
@@ -89,7 +109,18 @@ class FlaskFactoryTests(unittest.TestCase):
                 self.assertIn(b"Invalid URL", response.data)
                 self.assertEqual(service.enqueue_calls, [])
 
-        for url in ("http://example.com/video", "https://example.com/video"):
+    def test_submit_accepts_only_explicit_official_youtube_hosts(self):
+        valid_urls = (
+            "https://youtube.com/watch?v=abc",
+            "https://www.youtube.com/watch?v=abc",
+            "http://www.youtube.com/watch?v=abc",
+            "https://m.youtube.com/watch?v=abc",
+            "https://music.youtube.com/watch?v=abc",
+            "https://youtu.be/abc",
+            "https://YouTube.com/watch?v=abc",
+            "https://WWW.YOUTUBE.COM/watch?v=abc",
+        )
+        for url in valid_urls:
             with self.subTest(url=url):
                 service = FakeControllerService()
                 response = self._post(service, url=url)
@@ -103,7 +134,7 @@ class FlaskFactoryTests(unittest.TestCase):
 
         response = self._post(
             service,
-            url="https://example.com/playlist",
+            url="https://www.youtube.com/playlist?list=PL1234567890",
             submitter="  Alice  ",
             note="  Before service  ",
         )
@@ -113,8 +144,49 @@ class FlaskFactoryTests(unittest.TestCase):
         self.assertIn(b"start downloading shortly", response.data)
         self.assertEqual(
             service.enqueue_calls,
-            [("https://example.com/playlist", "Alice", "Before service")],
+            [
+                (
+                    "https://www.youtube.com/playlist?list=PL1234567890",
+                    "Alice",
+                    "Before service",
                 )
+            ],
+        )
+
+    def test_submit_rejects_oversized_submitter_and_note(self):
+        module = importlib.import_module("app")
+        service = FakeControllerService()
+        response = self._post(
+            service,
+            submitter="x" * (module.MAX_SUBMITTER_LENGTH + 1),
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"must be", response.data)
+        self.assertEqual(service.enqueue_calls, [])
+
+        service = FakeControllerService()
+        response = self._post(
+            service,
+            note="x" * (module.MAX_NOTE_LENGTH + 1),
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"must be", response.data)
+        self.assertEqual(service.enqueue_calls, [])
+
+    def test_submit_rejects_oversized_request_body(self):
+        service = FakeControllerService()
+        app = self._app(service)
+        response = app.test_client().post(
+            "/submit",
+            data={
+                "url": "https://www.youtube.com/watch?v=abc",
+                "note": "x" * 64_000,
+            },
+            follow_redirects=True,
+        )
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"too large", response.data.lower())
+        self.assertEqual(service.enqueue_calls, [])
 
     def test_reused_submission_preserves_existing_slot_message(self):
         service = FakeControllerService(

tests/test_config_contracts.py

@@ -81,6 +81,7 @@ class ConfigurationTests(unittest.TestCase):
         self.assertIsNone(config.router.fallback_restore_mode)
         self.assertEqual(config.storage.uncertain_review_after_seconds, 900.0)
         self.assertEqual(config.mqtt.publish_ack_timeout_seconds, 2.0)
+        self.assertEqual(config.mqtt.router_echo_grace_seconds, 2.0)
 
     def test_documented_environment_overrides_apply(self):
         env = {
@@ -152,6 +153,15 @@ class ConfigurationTests(unittest.TestCase):
                 with self.assertRaisesRegex(ConfigError, "must be finite"):
                     load_config(path, environ={})
 
+    def test_negative_router_echo_grace_is_rejected(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "config.yaml"
+            path.write_text(
+                "mqtt:\n  router_echo_grace_seconds: -1\n", encoding="utf-8"
+            )
+            with self.assertRaisesRegex(ConfigError, "router_echo_grace_seconds"):
+                load_config(path, environ={})
+
     def test_runtime_paths_must_be_absolute(self):
         cases = (
             ("storage:\n  media_base: relative/media\n", "storage.media_base"),

tests/test_mqtt_adapter.py

@@ -10,6 +10,14 @@ from controller.config import ControllerConfig, MqttConfig
 from controller.mqtt_adapter import MqttAdapter
 
 
+class MutableClock:
+    def __init__(self, value=0.0):
+        self.value = value
+
+    def __call__(self):
+        return self.value
+
+
 class FakeMessageInfo:
     def __init__(self, mid, published):
         self.rc = mqtt.MQTT_ERR_SUCCESS
@@ -245,7 +253,7 @@ class MqttAdapterTests(unittest.TestCase):
             ("audio/player/mode/set", "video", 1, False),
         )
 
-    def test_router_self_echo_is_marked_only_while_publish_is_in_flight(self):
+    def test_router_self_echo_is_marked_while_publish_is_in_flight(self):
         self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
@@ -268,9 +276,28 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertTrue(self.adapter.publish_router_mode("video").accepted)
         self.assertTrue(self.events.get_nowait().controller_echo)
 
-        self.client.publish_hook = None
-        self.adapter._on_message(
-            self.client,
+    def test_router_self_echo_delivered_after_puback_is_still_claimed_within_grace(
+        self,
+    ):
+        # Regression: a broker may legally deliver our own subscription echo
+        # after the PUBACK that already acknowledged the same publish. If the
+        # echo marker were cleared the instant publish() observes that
+        # PUBACK, this late echo would be mistaken for live operator intent
+        # and cause automatic Router restoration ownership to be relinquished.
+        clock = MutableClock(1000.0)
+        client = FakeClient()
+        events = queue.Queue()
+        adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
+        try:
+            adapter.start()
+            adapter._on_connect(client, None, None, 0)
+            events.get_nowait()
+
+            self.assertTrue(adapter.publish_router_mode("video").accepted)
+            clock.value += self.config.mqtt.router_echo_grace_seconds - 0.5
+
+            adapter._on_message(
+                client,
                 None,
                 SimpleNamespace(
                     topic="audio/player/mode/set",
@@ -281,7 +308,41 @@ class MqttAdapterTests(unittest.TestCase):
                     mid=32,
                 ),
             )
-        self.assertFalse(self.events.get_nowait().controller_echo)
+            self.assertTrue(events.get_nowait().controller_echo)
+        finally:
+            adapter.stop()
+
+    def test_router_self_echo_marker_expires_after_grace_window(self):
+        # A genuine external/operator command that happens to repeat the same
+        # value must still be recognized as operator intent once the bounded
+        # grace window has elapsed; the marker must not linger forever.
+        clock = MutableClock(1000.0)
+        client = FakeClient()
+        events = queue.Queue()
+        adapter = MqttAdapter(self.config, client=client, events=events, monotonic=clock)
+        try:
+            adapter.start()
+            adapter._on_connect(client, None, None, 0)
+            events.get_nowait()
+
+            self.assertTrue(adapter.publish_router_mode("video").accepted)
+            clock.value += self.config.mqtt.router_echo_grace_seconds + 0.5
+
+            adapter._on_message(
+                client,
+                None,
+                SimpleNamespace(
+                    topic="audio/player/mode/set",
+                    payload=b"video",
+                    qos=1,
+                    retain=False,
+                    dup=False,
+                    mid=32,
+                ),
+            )
+            self.assertFalse(events.get_nowait().controller_echo)
+        finally:
+            adapter.stop()
 
     def test_retained_router_set_never_consumes_controller_echo_marker(self):
         self.adapter.start()

tests/test_router_echo_mosquitto_integration.py

@@ -0,0 +1,206 @@
+"""Real-broker smoke test for Audio Router self-echo correlation.
+
+The fake MQTT client used by ``tests/test_mqtt_adapter.py`` delivers a
+publish's own echo synchronously, inside the ``publish()`` call itself, so it
+cannot exercise the actual ordering risk: a real broker may legally deliver a
+Controller-originated Router mode-set message back through the Controller's
+own subscription either before or after the PUBACK that already acknowledged
+the same publish. This test runs the real ``MqttAdapter`` against a real
+Mosquitto broker in Docker to prove that ordering no longer matters.
+
+Skipped automatically when Docker (or a Mosquitto broker container) is not
+available, so it never blocks the primary regression suite.
+"""
+
+from __future__ import annotations
+
+import queue
+import shutil
+import socket
+import subprocess
+import tempfile
+import time
+import unittest
+import uuid
+from pathlib import Path
+
+import paho.mqtt.client as mqtt
+
+from controller.config import ControllerConfig, MqttConfig
+from controller.mqtt_adapter import MqttAdapter
+
+
+def _docker_available() -> bool:
+    return shutil.which("docker") is not None
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@unittest.skipUnless(
+    _docker_available(), "docker is not available for a real Mosquitto broker"
+)
+class RouterSelfEchoMosquittoIntegrationTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.port = _free_port()
+        cls.container_name = f"controller-echo-test-{uuid.uuid4().hex[:10]}"
+        cls.tempdir = tempfile.TemporaryDirectory()
+        conf_path = Path(cls.tempdir.name) / "mosquitto.conf"
+        conf_path.write_text("listener 1883\nallow_anonymous true\n", encoding="utf-8")
+        try:
+            subprocess.run(
+                [
+                    "docker",
+                    "run",
+                    "-d",
+                    "--rm",
+                    "--name",
+                    cls.container_name,
+                    "-p",
+                    f"127.0.0.1:{cls.port}:1883",
+                    "-v",
+                    f"{conf_path}:/mosquitto/config/mosquitto.conf:ro",
+                    "eclipse-mosquitto",
+                ],
+                check=True,
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+        except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
+            cls.tempdir.cleanup()
+            raise unittest.SkipTest(
+                f"could not start a real Mosquitto broker container: {exc}"
+            )
+        try:
+            cls._wait_for_broker(cls.port)
+        except AssertionError as exc:
+            subprocess.run(
+                ["docker", "rm", "-f", cls.container_name],
+                capture_output=True,
+                timeout=15,
+            )
+            cls.tempdir.cleanup()
+            raise unittest.SkipTest(str(exc))
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        subprocess.run(
+            ["docker", "rm", "-f", cls.container_name],
+            capture_output=True,
+            timeout=15,
+        )
+        cls.tempdir.cleanup()
+
+    @staticmethod
+    def _wait_for_broker(port: int, timeout: float = 15.0) -> None:
+        deadline = time.monotonic() + timeout
+        last_exc: OSError | None = None
+        while time.monotonic() < deadline:
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.5):
+                    return
+            except OSError as exc:
+                last_exc = exc
+                time.sleep(0.2)
+        raise AssertionError(f"Mosquitto broker never became reachable: {last_exc}")
+
+    def _config(self) -> ControllerConfig:
+        return ControllerConfig(
+            mqtt=MqttConfig(
+                host="127.0.0.1",
+                port=self.port,
+                client_id=f"controller-echo-{uuid.uuid4().hex[:10]}",
+                reconnect_min_seconds=1,
+                reconnect_max_seconds=1,
+                command_queue_size=32,
+                publish_ack_timeout_seconds=5.0,
+                router_echo_grace_seconds=2.0,
+            )
+        )
+
+    def test_controller_recognizes_its_own_router_echo_over_a_real_broker(self) -> None:
+        config = self._config()
+        adapter = MqttAdapter(config)
+        adapter.start()
+        try:
+            self._wait_until(
+                lambda: adapter.connected,
+                "Controller MQTT adapter never connected to the real broker",
+            )
+            receipt = adapter.publish_router_mode("video")
+            self.assertTrue(receipt.accepted, receipt.reason)
+
+            event = self._next_message_event(adapter, timeout=5.0)
+            self.assertEqual(event.topic, config.topics.router_mode_set)
+            self.assertEqual(event.payload, b"video")
+            self.assertTrue(
+                event.controller_echo,
+                "Controller did not recognize its own Router mode-set echo "
+                "delivered by a real broker; the PUBACK-vs-subscription-echo "
+                "race was not closed",
+            )
+        finally:
+            adapter.stop()
+
+    def test_genuine_external_router_command_is_not_treated_as_self_echo(self) -> None:
+        config = self._config()
+        adapter = MqttAdapter(config)
+        adapter.start()
+        operator = mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2,
+            client_id=f"operator-{uuid.uuid4().hex[:10]}",
+        )
+        try:
+            self._wait_until(
+                lambda: adapter.connected,
+                "Controller MQTT adapter never connected to the real broker",
+            )
+            operator.connect(config.mqtt.host, config.mqtt.port, keepalive=30)
+            operator.loop_start()
+            info = operator.publish(
+                config.topics.router_mode_set, "service", qos=1, retain=False
+            )
+            info.wait_for_publish(timeout=5.0)
+
+            event = self._next_message_event(adapter, timeout=5.0)
+            self.assertEqual(event.payload, b"service")
+            self.assertFalse(
+                event.controller_echo,
+                "A genuine external Router command was mistaken for a "
+                "Controller echo; automatic restoration ownership would be "
+                "incorrectly preserved instead of relinquished",
+            )
+        finally:
+            operator.loop_stop()
+            operator.disconnect()
+            adapter.stop()
+
+    @staticmethod
+    def _wait_until(predicate, message: str, timeout: float = 10.0) -> None:
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.05)
+        raise AssertionError(message)
+
+    @staticmethod
+    def _next_message_event(adapter: MqttAdapter, *, timeout: float):
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                event = adapter.events.get(timeout=0.2)
+            except queue.Empty:
+                continue
+            if event.kind == "message":
+                return event
+        raise AssertionError("no MQTT message event was observed within the timeout")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -88,6 +88,11 @@ class FakeDownloadManager:
         self.expiry_calls = 0
         self.stopped = False
         self.delete_requests = []
+        self.enqueue_calls = []
+
+    def enqueue(self, url, submitter, note):
+        self.enqueue_calls.append((url, submitter, note))
+        return self.store.reserve_download(url, submitter, note)
 
     def recover_pending_deletions(self):
         self.recovered = True
@@ -492,22 +497,24 @@ class ControllerServiceTests(unittest.TestCase):
         self.assertIn("mqtt_disconnected", health["degraded_reasons"])
         self.assertIn("awaiting_player_evidence", health["degraded_reasons"])
 
-    def test_cleanup_and_downloads_wait_for_post_start_player_evidence(self):
+    def test_cleanup_waits_for_post_start_player_evidence_but_new_ingest_does_not(self):
         self.start_service()
         self.service._next_expiry = self.monotonic()
         time.sleep(0.2)
 
         self.assertEqual(self.downloads.expiry_calls, 0)
         self.assertFalse(self.downloads.recovered)
-        with self.assertRaisesRegex(
-            ControllerUnavailableError,
-            "deferred until fresh Player evidence",
-        ):
-            self.service.enqueue_download(
-                "https://example.test/deferred",
+
+        # A genuinely new playlist cannot mutate, delete, or replace any
+        # occupied slot, so it is not blocked by the missing evidence that
+        # still defers cleanup/expiry below.
+        result = self.service.enqueue_download(
+            "https://example.test/brand-new",
             None,
             None,
         )
+        self.assertEqual(result["status"], "queued")
+        self.assertFalse(result["reused_slot"])
 
         self.send_message(
             self.config.topics.player_state,
@@ -540,6 +547,106 @@ class ControllerServiceTests(unittest.TestCase):
             self.service.health()["degraded_reasons"],
         )
 
+    def test_ingest_matching_an_occupied_slot_url_remains_gated_without_evidence(self):
+        self.start_service()
+        self.ready_slot(1)
+
+        with self.assertRaisesRegex(
+            ControllerUnavailableError,
+            "fresh Player evidence",
+        ):
+            self.service.enqueue_download("https://example.test/1", None, None)
+
+    def test_controller_only_restart_with_idle_connected_player_allows_new_ingest(self):
+        """Regression for the too-broad evidence gate deadlock.
+
+        A Controller-only restart with the Player still connected and idle
+        never produces fresh detailed evidence, because the Player has no
+        reason to reconnect or emit unsolicited status. New ingest into an
+        empty slot must still work; refresh/reuse of already-occupied media,
+        expiry, cleanup recovery, and manual deletion must all stay gated
+        until fresh evidence or an explicit administrative override arrives.
+        """
+
+        self.start_service()
+        self.ready_slot(1)
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "idle", "message": "player online"},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: "awaiting_player_evidence"
+            not in self.service.health()["degraded_reasons"],
+            "evidence did not clear before the simulated restart",
+        )
+        self.service.stop()
+
+        # Simulate a Controller-only restart against the same durable store:
+        # the Player process itself never restarted, stays connected and
+        # idle, and therefore never sends another fresh status.
+        self.events = queue.Queue()
+        restarted_store = ControllerStore(self.config.storage, now=self.wall)
+        restarted_mqtt = FakeMqttAdapter()
+        restarted_downloads = FakeDownloadManager(restarted_store)
+        restarted_service = ControllerService(
+            self.config,
+            store=restarted_store,
+            mqtt_adapter=restarted_mqtt,
+            event_queue=self.events,
+            download_manager=restarted_downloads,
+            monotonic=self.monotonic,
+            wall_clock=self.wall,
+        )
+        restarted_service.start()
+        self.addCleanup(
+            lambda: restarted_service.stop() if restarted_service.started else None
+        )
+        self.assertIn(
+            "awaiting_player_evidence",
+            restarted_service.health()["degraded_reasons"],
+        )
+
+        # A genuinely new playlist still ingests safely with no evidence.
+        result = restarted_service.enqueue_download(
+            "https://example.test/after-restart", None, None
+        )
+        self.assertEqual(result["status"], "queued")
+        self.assertFalse(result["reused_slot"])
+        self.assertNotEqual(result["slot"], 1)
+
+        # Refreshing the URL already occupying slot 1 remains gated.
+        with self.assertRaisesRegex(
+            ControllerUnavailableError,
+            "fresh Player evidence",
+        ):
+            restarted_service.enqueue_download("https://example.test/1", None, None)
+
+        # Expiry and cleanup recovery remain deferred.
+        restarted_service._next_expiry = self.monotonic()
+        time.sleep(0.2)
+        self.assertEqual(restarted_downloads.expiry_calls, 0)
+        self.assertFalse(restarted_downloads.recovered)
+
+        # Manual deletion of occupied slot 1 still requires the explicit
+        # confirm_player_unknown override; it is not silently unlocked.
+        with self.assertRaisesRegex(
+            SlotProtectedError,
+            "confirm_player_unknown=true",
+        ):
+            restarted_service.submit_control({"action": "delete_slot", "slot": 1})
+
+        # Fresh Player evidence still unlocks cleanup normally afterward.
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "idle", "message": "player online"},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: restarted_downloads.recovered,
+            "fresh evidence did not unlock deferred recovery after restart",
+        )
+
     def test_administrator_can_explicitly_unlock_cleanup_without_player_evidence(self):
         self.start_service()
         review = self.service.health()["cleanup_reconciliation"]

tests/test_store.py

@@ -98,6 +98,20 @@ class FreshStoreTests(StoreTestCase):
             self.assertTrue(store.mark_download_running("job-one", 1, 1))
             self.assertEqual(store.get_slot(1).media_state, "downloading")
 
+    def test_find_occupied_slot_for_url_only_matches_non_empty_slots(self):
+        with self.open_store() as store:
+            url = "https://www.youtube.com/playlist?list=one"
+            self.assertIsNone(store.find_occupied_slot_for_url(url))
+            self.assertIsNone(store.find_occupied_slot_for_url("  "))
+            self.assertIsNone(store.find_occupied_slot_for_url(None))
+
+            store.reserve_download(url, None, None, job_id="job-one")
+            self.assertEqual(store.find_occupied_slot_for_url(url), 1)
+            self.assertEqual(store.find_occupied_slot_for_url(f"  {url}  "), 1)
+            self.assertIsNone(
+                store.find_occupied_slot_for_url("https://example.test/other")
+            )
+
     def test_concurrent_requests_get_only_one_remaining_slot(self):
         with self.open_store() as store:
             for index in range(1, 10):