feat: coordinate authoritative playback and audio routing

config.yaml

@@ -8,6 +8,8 @@ mqtt:
   reconnect_min_seconds: 1
   reconnect_max_seconds: 30
   command_queue_size: 512
+  # QoS 1 is accepted only after the broker PUBACKs within this bound.
+  publish_ack_timeout_seconds: 2
 
 topics:
   player_command: youtube/player/cmd
@@ -19,8 +21,9 @@ topics:
   playback_base: youtube/media/playback
   router_mode_set: audio/player/mode/set
   router_mode_state: audio/player/mode/state
-  # Optional additive result topic. Existing consumers do not depend on it.
-  controller_result: null
+  # Optional additive result/warning topic. Set null to disable it.
+  # Existing consumers never depend on this topic.
+  controller_result: youtube/controller/v2/events
 
 storage:
   media_base: /mnt/Media/YTPlayer

controller/config.py

@@ -31,6 +31,7 @@ class MqttConfig:
     reconnect_min_seconds: int = 1
     reconnect_max_seconds: int = 30
     command_queue_size: int = 512
+    publish_ack_timeout_seconds: float = 2.0
 
 
 @dataclass(frozen=True)
@@ -195,6 +196,11 @@ def load_config(
         reconnect_min_seconds=_integer(mqtt_raw.get("reconnect_min_seconds", 1), "mqtt.reconnect_min_seconds"),
         reconnect_max_seconds=_integer(mqtt_raw.get("reconnect_max_seconds", 30), "mqtt.reconnect_max_seconds"),
         command_queue_size=_integer(mqtt_raw.get("command_queue_size", 512), "mqtt.command_queue_size"),
+        publish_ack_timeout_seconds=_number(
+            mqtt_raw.get("publish_ack_timeout_seconds", 2.0),
+            "mqtt.publish_ack_timeout_seconds",
+            minimum=0.1,
+        ),
     )
     if mqtt.reconnect_max_seconds < mqtt.reconnect_min_seconds:
         raise ConfigError("mqtt.reconnect_max_seconds must be >= reconnect_min_seconds")
@@ -211,7 +217,11 @@ def load_config(
         playback_base=_text(topics_raw.get("playback_base", "youtube/media/playback"), "topics.playback_base"),
         router_mode_set=_text(topics_raw.get("router_mode_set", "audio/player/mode/set"), "topics.router_mode_set"),
         router_mode_state=_text(topics_raw.get("router_mode_state", "audio/player/mode/state"), "topics.router_mode_state"),
-        controller_result=_text(topics_raw.get("controller_result"), "topics.controller_result", allow_none=True),
+        controller_result=_text(
+            topics_raw.get("controller_result", "youtube/controller/v2/events"),
+            "topics.controller_result",
+            allow_none=True,
+        ),
     )
 
     configured_slots = _integer(storage_raw.get("slot_count", 10), "storage.slot_count")

controller/contracts.py

@@ -25,7 +25,7 @@ class Topics:
     playback_base: str = "youtube/media/playback"
     router_mode_set: str = "audio/player/mode/set"
     router_mode_state: str = "audio/player/mode/state"
-    controller_result: str | None = None
+    controller_result: str | None = "youtube/controller/v2/events"
 
     def slot_base(self, slot: int) -> str:
         if slot not in SLOT_IDS:

controller/downloads.py

@@ -4,9 +4,11 @@ from __future__ import annotations
 
 import logging
 import os
+import signal
 import shutil
 import subprocess
 import threading
+import time
 import uuid
 from concurrent.futures import Future, ThreadPoolExecutor
 from dataclasses import dataclass
@@ -32,6 +34,10 @@ class DownloadEvent:
     message: str | None = None
     job_id: str | None = None
     generation: int | None = None
+    url: str | None = None
+    submitter: str | None = None
+    note: str | None = None
+    reused_slot: bool = False
 
 
 EventCallback = Callable[[DownloadEvent], None]
@@ -66,7 +72,7 @@ class DownloadManager:
         store: ControllerStore,
         *,
         on_event: EventCallback | None = None,
-        runner: Runner = subprocess.run,
+        runner: Runner | None = None,
         executor: ThreadPoolExecutor | None = None,
     ):
         self.config = config
@@ -80,6 +86,7 @@ class DownloadManager:
         self._owns_executor = executor is None
         self._lock = threading.Lock()
         self._futures: set[Future[Any]] = set()
+        self._processes: set[subprocess.Popen[str]] = set()
         self._stopping = False
 
     def enqueue(
@@ -106,6 +113,10 @@ class DownloadManager:
                 state="queued",
                 job_id=reservation.job_id,
                 generation=reservation.generation,
+                url=url,
+                submitter=submitter,
+                note=note,
+                reused_slot=reservation.reused_slot,
             )
         )
         try:
@@ -184,7 +195,7 @@ class DownloadManager:
             url,
         ]
         try:
-            result = self.runner(
+            result = self._run_command(
                 arguments,
                 check=True,
                 capture_output=True,
@@ -246,7 +257,7 @@ class DownloadManager:
                 ),
                 url,
             ]
-            self.runner(
+            self._run_command(
                 arguments,
                 check=True,
                 text=True,
@@ -322,6 +333,90 @@ class DownloadManager:
             return f"yt-dlp failed with exit code {exc.returncode}"
         return f"download failed: {exc}"
 
+    def _run_command(
+        self,
+        arguments: list[str],
+        *,
+        check: bool,
+        text: bool,
+        timeout: float,
+        capture_output: bool = False,
+    ) -> subprocess.CompletedProcess[str]:
+        if self.runner is not None:
+            return self.runner(
+                arguments,
+                check=check,
+                text=text,
+                timeout=timeout,
+                capture_output=capture_output,
+            )
+
+        with self._lock:
+            if self._stopping:
+                raise RuntimeError("Controller is shutting down")
+            process = subprocess.Popen(
+                arguments,
+                stdout=subprocess.PIPE if capture_output else None,
+                stderr=subprocess.PIPE if capture_output else None,
+                text=text,
+                start_new_session=True,
+            )
+            self._processes.add(process)
+        try:
+            try:
+                stdout, stderr = process.communicate(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                self._terminate_process(process, force=False)
+                try:
+                    stdout, stderr = process.communicate(timeout=2.0)
+                except subprocess.TimeoutExpired:
+                    self._terminate_process(process, force=True)
+                    stdout, stderr = process.communicate()
+                raise subprocess.TimeoutExpired(
+                    arguments,
+                    timeout,
+                    output=stdout,
+                    stderr=stderr,
+                )
+            completed = subprocess.CompletedProcess(
+                arguments,
+                process.returncode,
+                stdout=stdout,
+                stderr=stderr,
+            )
+            if check and process.returncode:
+                raise subprocess.CalledProcessError(
+                    process.returncode,
+                    arguments,
+                    output=stdout,
+                    stderr=stderr,
+                )
+            return completed
+        finally:
+            with self._lock:
+                self._processes.discard(process)
+
+    @staticmethod
+    def _terminate_process(
+        process: subprocess.Popen[str],
+        *,
+        force: bool,
+    ) -> None:
+        if process.poll() is not None:
+            return
+        try:
+            os.killpg(
+                process.pid,
+                signal.SIGKILL if force else signal.SIGTERM,
+            )
+        except ProcessLookupError:
+            return
+        except OSError:
+            try:
+                process.kill() if force else process.terminate()
+            except ProcessLookupError:
+                pass
+
     @staticmethod
     def _fsync_directory(path: Path) -> None:
         descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
@@ -387,6 +482,17 @@ class DownloadManager:
     def shutdown(self, *, wait: bool = True) -> None:
         with self._lock:
             self._stopping = True
+            processes = tuple(self._processes)
+        for process in processes:
+            self._terminate_process(process, force=False)
+        deadline = time.monotonic() + 2.0
+        for process in processes:
+            if process.poll() is not None:
+                continue
+            try:
+                process.wait(timeout=max(0.0, deadline - time.monotonic()))
+            except subprocess.TimeoutExpired:
+                self._terminate_process(process, force=True)
         if self._owns_executor:
             self._executor.shutdown(wait=wait, cancel_futures=True)
 

controller/models.py

@@ -84,10 +84,38 @@ class PlayerObservation:
     generation: int
     reconciliation_state: str
     reconciliation_due_at: float | None
+    reconciliation_id: str | None
     last_status_at: float | None
     last_telemetry_at: float | None
 
 
+@dataclass(frozen=True)
+class RouterObservation:
+    mode: str | None
+    freshness: str
+    observed_at: float | None
+    last_warning: str | None
+
+
+@dataclass(frozen=True)
+class AudioLeaseRecord:
+    lease_id: str
+    operation_id: str | None
+    playback_generation: int | None
+    state: str
+    previous_mode: str | None
+    target_mode: str
+    restoration_mode: str | None
+    restoration_source: str
+    controller_changed: bool
+    confirmation_deadline: float | None
+    created_at: float
+    acquired_at: float | None
+    released_at: float | None
+    warning: str | None
+    error: str | None
+
+
 @dataclass(frozen=True)
 class MigrationReport:
     migrated: bool

controller/mqtt_adapter.py

@@ -6,6 +6,7 @@ import json
 import logging
 import queue
 import threading
+import uuid
 from dataclasses import dataclass
 from typing import Any, Mapping
 
@@ -33,9 +34,11 @@ class MqttEvent:
     payload: bytes = b""
     qos: int = 0
     retain: bool = False
+    duplicate: bool = False
     message_id: int | None = None
     connection_generation: int | None = None
     reason: str | None = None
+    controller_echo: bool = False
 
 
 @dataclass(frozen=True)
@@ -43,6 +46,7 @@ class PublishReceipt:
     accepted: bool
     message_id: int | None = None
     reason: str | None = None
+    uncertain: bool = False
 
 
 class MqttAdapter:
@@ -59,11 +63,8 @@ class MqttAdapter:
         self.events: "queue.Queue[MqttEvent]" = events or queue.Queue(
             maxsize=config.mqtt.command_queue_size
         )
-        self.client = client or mqtt.Client(
-            mqtt.CallbackAPIVersion.VERSION2,
-            client_id=config.mqtt.client_id,
-            clean_session=True,
-        )
+        self._injected_client = client is not None
+        self.client = client or self._new_client()
         self._generation_lock = threading.Lock()
         self._connection_generation = 0
         self._connected = threading.Event()
@@ -72,9 +73,18 @@ class MqttAdapter:
         self._started = False
         self._loop_started = False
         self._bootstrap_thread: threading.Thread | None = None
+        self._router_echo_lock = threading.Lock()
+        self._pending_router_echoes: dict[str, bytes] = {}
         self.last_error: str | None = None
         self._configure_client()
 
+    def _new_client(self):
+        return mqtt.Client(
+            mqtt.CallbackAPIVersion.VERSION2,
+            client_id=self.config.mqtt.client_id,
+            clean_session=True,
+        )
+
     def _configure_client(self) -> None:
         if self.config.mqtt.username is not None:
             self.client.username_pw_set(
@@ -116,20 +126,32 @@ class MqttAdapter:
             self._stop_event.clear()
         try:
             self._begin_async_connection()
-        except (OSError, mqtt.WebsocketConnectionError) as exc:
+        except (OSError, mqtt.WebsocketConnectionError, RuntimeError) as exc:
             self.last_error = f"initial MQTT setup failed: {exc}"
             LOG.warning("%s; retrying in background", self.last_error)
             self._start_bootstrap_retry()
 
     def _begin_async_connection(self) -> None:
-        self.client.connect_async(
+        with self._lifecycle_lock:
+            if not self._started or self._stop_event.is_set():
+                return
+            client = self.client
+        result = client.connect_async(
             self.config.mqtt.host,
             self.config.mqtt.port,
             keepalive=self.config.mqtt.keepalive_seconds,
         )
+        if result != mqtt.MQTT_ERR_SUCCESS:
+            raise RuntimeError(f"failed to configure MQTT connection: {result}")
         with self._lifecycle_lock:
+            if (
+                not self._started
+                or self._stop_event.is_set()
+                or client is not self.client
+            ):
+                return
             if not self._loop_started:
-                result = self.client.loop_start()
+                result = client.loop_start()
                 if result != mqtt.MQTT_ERR_SUCCESS:
                     raise RuntimeError(f"failed to start MQTT network loop: {result}")
                 self._loop_started = True
@@ -159,17 +181,21 @@ class MqttAdapter:
                 delay = min(maximum, max(delay * 2, 1.0))
             except RuntimeError as exc:
                 self.last_error = str(exc)
-                LOG.error("%s", self.last_error)
-                return
+                LOG.warning("%s", self.last_error)
+                delay = min(maximum, max(delay * 2, 1.0))
 
     def stop(self) -> None:
+        # Signal callbacks/retry work before waiting for the lifecycle lock.
+        # A callback already inside subscribe/connect then observes shutdown
+        # and cannot publish a post-stop connected state.
+        self._stop_event.set()
         with self._lifecycle_lock:
             if not self._started:
+                self._connected.clear()
                 return
             self._started = False
             loop_started = self._loop_started
             self._loop_started = False
-        self._stop_event.set()
         self._connected.clear()
         if loop_started:
             try:
@@ -190,24 +216,43 @@ class MqttAdapter:
 
     def _on_connect(self, client, userdata, flags, reason_code, properties=None) -> None:
         try:
-            if not self._reason_success(reason_code):
-                self.last_error = f"MQTT connect failed: {reason_code}"
-                LOG.warning("%s", self.last_error)
+            with self._lifecycle_lock:
+                if (
+                    client is not self.client
+                    or not self._started
+                    or self._stop_event.is_set()
+                ):
+                    return
+                if not self._reason_success(reason_code):
+                    self.last_error = f"MQTT connect failed: {reason_code}"
+                    LOG.warning("%s", self.last_error)
+                    return
+                if self._stop_event.is_set():
+                    return
+                with self._generation_lock:
+                    self._connection_generation += 1
+                    generation = self._connection_generation
+                self._connected.set()
+                self.last_error = None
+                subscriptions = (
+                    (self.config.topics.player_status, PLAYER_STATUS_QOS),
+                    (self.config.topics.player_telemetry, PLAYER_TELEMETRY_QOS),
+                    (self.config.topics.player_state, PLAYER_STATE_QOS),
+                    (self.config.topics.controller_command, 1),
+                    (self.config.topics.router_mode_set, ROUTER_COMMAND_QOS),
+                    (self.config.topics.router_mode_state, 1),
+                )
+                for topic, qos in subscriptions:
+                    if self._stop_event.is_set():
+                        self._connected.clear()
+                        return
+                    client.subscribe(topic, qos=qos)
+                    if self._stop_event.is_set():
+                        self._connected.clear()
+                        return
+            if self._stop_event.is_set():
+                self._connected.clear()
                 return
-            with self._generation_lock:
-                self._connection_generation += 1
-                generation = self._connection_generation
-            self._connected.set()
-            self.last_error = None
-            subscriptions = (
-                (self.config.topics.player_status, PLAYER_STATUS_QOS),
-                (self.config.topics.player_telemetry, PLAYER_TELEMETRY_QOS),
-                (self.config.topics.player_state, PLAYER_STATE_QOS),
-                (self.config.topics.controller_command, 1),
-                (self.config.topics.router_mode_state, 1),
-            )
-            for topic, qos in subscriptions:
-                client.subscribe(topic, qos=qos)
             self._enqueue(
                 MqttEvent(
                     kind="connected",
@@ -231,6 +276,8 @@ class MqttAdapter:
         reason_code,
         properties=None,
     ) -> None:
+        if client is not self.client:
+            return
         self._connected.clear()
         try:
             self._enqueue(
@@ -245,15 +292,33 @@ class MqttAdapter:
 
     def _on_message(self, client, userdata, message) -> None:
         try:
+            # Capture client identity and generation atomically with session
+            # replacement. A callback that was already running when an
+            # uncertain command discards the transport must be able to leave a
+            # full queue so loop_stop() cannot deadlock waiting for it.
+            with self._lifecycle_lock:
+                if client is not self.client:
+                    return
+                with self._generation_lock:
+                    generation = self._connection_generation
+            topic = str(message.topic)
+            payload = bytes(message.payload)
+            retained = bool(getattr(message, "retain", False))
             self._enqueue(
                 MqttEvent(
                     kind="message",
-                    topic=str(message.topic),
-                    payload=bytes(message.payload),
+                    topic=topic,
+                    payload=payload,
                     qos=int(getattr(message, "qos", 0)),
-                    retain=bool(getattr(message, "retain", False)),
+                    retain=retained,
+                    duplicate=bool(getattr(message, "dup", False)),
                     message_id=getattr(message, "mid", None),
-                    connection_generation=self.connection_generation,
+                    connection_generation=generation,
+                    controller_echo=(
+                        topic == self.config.topics.router_mode_set
+                        and not retained
+                        and self._claim_router_echo(payload)
+                    ),
                 )
             )
         except BaseException:
@@ -261,6 +326,11 @@ class MqttAdapter:
 
     def _enqueue(self, event: MqttEvent) -> bool:
         while not self._stop_event.is_set():
+            if (
+                event.connection_generation is not None
+                and event.connection_generation != self.connection_generation
+            ):
+                return False
             try:
                 self.events.put(event, timeout=0.2)
                 return True
@@ -293,8 +363,9 @@ class MqttAdapter:
         qos: int,
         retain: bool,
         require_connection: bool,
+        invalidate_transport_on_uncertain: bool = False,
     ) -> PublishReceipt:
-        if require_connection and not self.connected:
+        if require_connection and (not self.started or not self.connected):
             return PublishReceipt(False, reason="MQTT is disconnected")
         try:
             info = self.client.publish(topic, payload, qos=qos, retain=retain)
@@ -303,8 +374,79 @@ class MqttAdapter:
             return PublishReceipt(False, reason=str(exc))
         if info.rc != mqtt.MQTT_ERR_SUCCESS:
             return PublishReceipt(False, message_id=getattr(info, "mid", None), reason=f"MQTT rc={info.rc}")
+        if qos > 0 and hasattr(info, "wait_for_publish"):
+            try:
+                info.wait_for_publish(
+                    timeout=self.config.mqtt.publish_ack_timeout_seconds
+                )
+                acknowledged = bool(info.is_published())
+            except BaseException as exc:
+                acknowledged = False
+                detail = str(exc) or exc.__class__.__name__
+            else:
+                detail = (
+                    f"no broker PUBACK within "
+                    f"{self.config.mqtt.publish_ack_timeout_seconds:g} seconds"
+                )
+            if not acknowledged:
+                reason = f"MQTT QoS 1 publication is uncertain: {detail}"
+                if invalidate_transport_on_uncertain:
+                    self._discard_unacknowledged_transport(reason)
+                return PublishReceipt(
+                    False,
+                    message_id=getattr(info, "mid", None),
+                    reason=reason,
+                    uncertain=True,
+                )
         return PublishReceipt(True, message_id=getattr(info, "mid", None))
 
+    def _discard_unacknowledged_transport(self, reason: str) -> None:
+        """Drop the Paho session so stale QoS 1 commands cannot replay."""
+
+        if self._injected_client:
+            # Test transports do not own a recreatable Paho session. Production
+            # always follows the replacement path below.
+            with self._generation_lock:
+                self._connection_generation += 1
+            self._connected.clear()
+            return
+        with self._lifecycle_lock:
+            old_client = self.client
+            old_loop_started = self._loop_started
+            self._loop_started = False
+            should_restart = self._started and not self._stop_event.is_set()
+            self._connected.clear()
+            with self._generation_lock:
+                previous_generation = self._connection_generation
+                self._connection_generation += 1
+            self.client = self._new_client()
+            self._configure_client()
+        try:
+            old_client.disconnect()
+        except BaseException:
+            LOG.exception("failed to disconnect uncertain MQTT transport")
+        if old_loop_started:
+            try:
+                old_client.loop_stop()
+            except BaseException:
+                LOG.exception("failed to stop uncertain MQTT transport")
+        self.last_error = reason
+        try:
+            self.events.put_nowait(
+                MqttEvent(
+                    kind="disconnected",
+                    connection_generation=previous_generation,
+                    reason=reason,
+                )
+            )
+        except queue.Full:
+            # This method can run on the sole coordinator thread; blocking on
+            # its own queue would deadlock. The uncertain PublishReceipt still
+            # drives the operation-specific conservative transition.
+            LOG.error("MQTT event queue was full while discarding uncertain transport")
+        if should_restart:
+            self._start_bootstrap_retry()
+
     @staticmethod
     def _json_payload(payload: Mapping[str, Any]) -> str:
         return json.dumps(payload, allow_nan=False, separators=(",", ":"))
@@ -316,16 +458,41 @@ class MqttAdapter:
             qos=PLAYER_COMMAND_QOS,
             retain=PLAYER_COMMAND_RETAIN,
             require_connection=True,
+            invalidate_transport_on_uncertain=True,
         )
 
     def publish_router_mode(self, mode: str) -> PublishReceipt:
-        return self._publish(
-            self.config.topics.router_mode_set,
-            mode,
-            qos=ROUTER_COMMAND_QOS,
-            retain=ROUTER_COMMAND_RETAIN,
-            require_connection=True,
-        )
+        # 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.
+        echo_token = uuid.uuid4().hex
+        with self._router_echo_lock:
+            self._pending_router_echoes[echo_token] = mode.encode("utf-8")
+        try:
+            return self._publish(
+                self.config.topics.router_mode_set,
+                mode,
+                qos=ROUTER_COMMAND_QOS,
+                retain=ROUTER_COMMAND_RETAIN,
+                require_connection=True,
+                invalidate_transport_on_uncertain=True,
+            )
+        finally:
+            with self._router_echo_lock:
+                self._pending_router_echoes.pop(echo_token, None)
+
+    def _claim_router_echo(self, payload: bytes) -> bool:
+        """Claim one exact mode-set echo from an in-flight local publish."""
+
+        with self._router_echo_lock:
+            for token, expected in tuple(self._pending_router_echoes.items()):
+                if expected == payload:
+                    del self._pending_router_echoes[token]
+                    return True
+        return False
 
     def publish_legacy(self, topic: str, payload: str, *, retain: bool = True) -> PublishReceipt:
         return self._publish(

controller/playback.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import math
+import os
 import time
 import uuid
 from dataclasses import dataclass, replace
@@ -28,6 +29,7 @@ class PlayerOperation:
     slot: int | None
     expected_path: str | None
     deadline: float
+    reconciliation_id: str | None = None
     state: str = "awaiting_player"
     message: str | None = None
 
@@ -71,6 +73,7 @@ class PlaybackReducer:
         *,
         slot: int | None = None,
         expected_path: str | None = None,
+        operation_id: str | None = None,
     ) -> PlayerOperation:
         if action not in CONTROL_ACTIONS:
             raise ValueError(f"unsupported Player action {action!r}")
@@ -82,14 +85,13 @@ class PlaybackReducer:
             )
         self._operation_generation += 1
         operation = PlayerOperation(
-            operation_id=uuid.uuid4().hex,
+            operation_id=operation_id or uuid.uuid4().hex,
             generation=self._operation_generation,
             action=action,
             slot=slot,
             expected_path=expected_path,
             deadline=self._monotonic() + self.operation_timeout_seconds,
         )
-        self.pending = operation
         if action == "play" and slot is not None:
             self.store.protect_slot(
                 slot,
@@ -97,11 +99,40 @@ class PlaybackReducer:
                 guard_id=f"operation-{operation.operation_id}",
                 details={"operation_id": operation.operation_id},
             )
+        elif action == "play" and expected_path is not None:
+            due = None
+            if self.config.storage.uncertain_review_after_seconds > 0:
+                due = (
+                    self._wall_clock()
+                    + self.config.storage.uncertain_review_after_seconds
+                )
+            observation = self.store.mark_external_uncertain(
+                expected_path,
+                review_after=due,
+                details={
+                    "operation_id": operation.operation_id,
+                    "reason": "play command pending Player evidence",
+                },
+            )
+            operation = replace(
+                operation,
+                reconciliation_id=observation.reconciliation_id,
+            )
+        self.pending = operation
         return operation
 
     def publish_failed(self, reason: str) -> PlaybackEffect | None:
         if self.pending is None:
             return None
+        if (
+            self.pending.action == "play"
+            and self.pending.slot is None
+            and self.pending.reconciliation_id is not None
+        ):
+            self.store.cancel_external_intent(
+                self.pending.reconciliation_id,
+                reason=reason,
+            )
         return self._finish_operation("failed", reason)
 
     def _finish_operation(self, state: str, message: str | None = None) -> PlaybackEffect:
@@ -142,11 +173,64 @@ class PlaybackReducer:
         operation = self.pending
         if operation is None:
             return False
-        slot_matches = operation.slot is None or operation.slot == slot
-        path_matches = operation.expected_path is None or operation.expected_path == path
-        if operation.slot is not None and operation.expected_path is not None:
-            return slot_matches or path_matches
-        return slot_matches and path_matches
+        positive_match = False
+        if operation.slot is not None and slot is not None:
+            if operation.slot != slot:
+                return False
+            positive_match = True
+        if operation.expected_path is not None and path is not None:
+            if os.path.abspath(operation.expected_path) != os.path.abspath(path):
+                return False
+            positive_match = True
+        return positive_match
+
+    def _canonical_slot_for_path(self, path: str | None) -> int | None:
+        if path is None:
+            return None
+        candidate = os.path.abspath(path)
+        for slot in SLOT_IDS:
+            slot_directory = os.path.abspath(
+                self.config.storage.media_base / f"slot{slot}"
+            )
+            try:
+                if os.path.commonpath((candidate, slot_directory)) == slot_directory:
+                    return slot
+            except ValueError:
+                continue
+        return None
+
+    def _normalized_identity(
+        self,
+        payload: Mapping[str, Any],
+    ) -> tuple[int | None, str | None, str | None]:
+        slot = self._slot(payload.get("slot"))
+        path = self._path(payload.get("path"))
+        derived = self._canonical_slot_for_path(path)
+        if slot is not None and derived is not None and slot != derived:
+            return (
+                slot,
+                path,
+                f"Player identity conflicts: slot {slot} does not contain path {path!r}",
+            )
+        return (derived if slot is None else slot), path, None
+
+    @staticmethod
+    def _same_playback_identity(
+        previous: PlayerObservation,
+        slot: int | None,
+        path: str | None,
+    ) -> bool:
+        if previous.active_slot != slot:
+            return False
+        if slot is None:
+            return (
+                previous.active_path is not None
+                and path is not None
+                and os.path.abspath(previous.active_path) == os.path.abspath(path)
+            )
+        if previous.active_path is None or path is None:
+            return True
+        return os.path.abspath(previous.active_path) == os.path.abspath(path)
 
     def _active_evidence(
         self,
@@ -158,11 +242,11 @@ class PlaybackReducer:
         telemetry: Mapping[str, Any] | None = None,
     ) -> PlayerObservation:
         previous = self.store.player_observation()
-        if slot is None:
+        if slot is None and (path is None or path == previous.active_path):
             slot = previous.active_slot
-        if path is None:
+        if path is None and (slot is None or slot == previous.active_slot):
             path = previous.active_path
-        if paused is None and previous.active_slot == slot:
+        if paused is None and self._same_playback_identity(previous, slot, path):
             paused = previous.paused
         phase = "paused" if paused is True else "playing"
         if slot is not None:
@@ -171,6 +255,8 @@ class PlaybackReducer:
                 kind="player_active",
                 details={"path": path},
             )
+        elif path is not None:
+            self.store.reconcile_external_active()
         return self.store.set_player_observation(
             availability="online",
             phase=phase,
@@ -179,6 +265,7 @@ class PlaybackReducer:
             paused=paused,
             reconciliation_state="confirmed",
             reconciliation_due_at=None,
+            reconciliation_id=None,
             status_payload=status,
             telemetry_payload=telemetry,
         )
@@ -199,6 +286,7 @@ class PlaybackReducer:
             paused=False,
             reconciliation_state="clear",
             reconciliation_due_at=None,
+            reconciliation_id=None,
             status_payload=status,
             telemetry_payload=telemetry,
         )
@@ -212,36 +300,63 @@ class PlaybackReducer:
         if not isinstance(state, str) or not state:
             return [PlaybackEffect("warning", message="Player status is missing state")]
         state = state.strip()
-        slot = self._slot(payload.get("slot"))
-        path = self._path(payload.get("path"))
+        slot, path, identity_warning = self._normalized_identity(payload)
+        if identity_warning is not None:
+            return [PlaybackEffect("warning", slot=slot, message=identity_warning)]
         effects: list[PlaybackEffect] = []
 
         if state == "offline":
             previous = self.store.player_observation()
             due = None
-            if (
-                previous.active_slot is not None
-                and self.config.storage.uncertain_review_after_seconds > 0
-            ):
-                due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
-                self.store.reconcile_active_guard(
+            active_known = (
+                previous.active_slot is not None or previous.active_path is not None
+            )
+            if previous.active_slot is not None:
+                if self.config.storage.uncertain_review_after_seconds > 0:
+                    due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+                guard = self.store.mark_slot_uncertain(
                     previous.active_slot,
-                    kind="restart_uncertain",
                     review_after=due,
                     details={"reason": "Player offline"},
                 )
-            self.store.set_player_observation(
-                availability="offline",
-                phase="unknown" if previous.active_slot is not None else previous.phase,
-                active_slot=previous.active_slot,
-                active_path=previous.active_path,
-                paused=previous.paused,
-                reconciliation_state=(
-                    "awaiting_evidence" if previous.active_slot is not None else "unknown"
-                ),
-                reconciliation_due_at=due,
-                status_payload=payload,
-            )
+                due = guard.review_after
+                self.store.set_player_observation(
+                    availability="offline",
+                    phase="unknown",
+                    active_slot=previous.active_slot,
+                    active_path=previous.active_path,
+                    paused=previous.paused,
+                    reconciliation_state=(
+                        "operator_review"
+                        if guard.state == "review_required"
+                        else "awaiting_evidence"
+                    ),
+                    reconciliation_due_at=due,
+                    reconciliation_id=guard.guard_id,
+                    status_payload=payload,
+                )
+            elif previous.active_path is not None:
+                if self.config.storage.uncertain_review_after_seconds > 0:
+                    due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+                self.store.mark_external_uncertain(
+                    previous.active_path,
+                    review_after=due,
+                    availability="offline",
+                    details={"reason": "Player offline"},
+                    status_payload=payload,
+                )
+            else:
+                self.store.set_player_observation(
+                    availability="offline",
+                    phase=previous.phase,
+                    active_slot=None,
+                    active_path=None,
+                    paused=previous.paused,
+                    reconciliation_state="unknown",
+                    reconciliation_due_at=None,
+                    reconciliation_id=None,
+                    status_payload=payload,
+                )
             effects.append(
                 PlaybackEffect(
                     "player_offline",
@@ -261,15 +376,56 @@ class PlaybackReducer:
                         message="Reconnect idle announcement carried an active path",
                     )
                 )
+                if self.pending is not None and self.pending.action == "play":
+                    if self._matches_pending(slot, path):
+                        effects.append(
+                            self._finish_operation(
+                                "confirmed", "matching reconnect evidence"
+                            )
+                        )
+                    else:
+                        effects.append(
+                            self._finish_operation(
+                                "failed",
+                                "Player reconnected with different active playback",
+                            )
+                        )
             else:
                 self._idle_evidence(status=payload)
                 effects.append(PlaybackEffect("player_idle"))
-                if self.pending is not None and self.pending.action == "stop":
-                    effects.append(self._finish_operation("confirmed"))
+                if self.pending is not None:
+                    if self.pending.action == "stop":
+                        effects.append(self._finish_operation("confirmed"))
+                    elif self.pending.action == "play":
+                        effects.append(
+                            self._finish_operation(
+                                "failed", "Player reconnected without active playback"
+                            )
+                        )
             return effects
 
         if state == "idle":
             reason = payload.get("reason")
+            if reason not in TERMINAL_REASONS:
+                previous = self.store.player_observation()
+                self.store.set_player_observation(
+                    availability="online",
+                    phase=previous.phase,
+                    active_slot=previous.active_slot,
+                    active_path=previous.active_path,
+                    paused=previous.paused,
+                    reconciliation_state=previous.reconciliation_state,
+                    reconciliation_due_at=previous.reconciliation_due_at,
+                    reconciliation_id=previous.reconciliation_id,
+                    status_payload=payload,
+                )
+                return [
+                    PlaybackEffect(
+                        "warning",
+                        slot=previous.active_slot,
+                        message="ambiguous Player idle status was not treated as terminal",
+                    )
+                ]
             self._idle_evidence(status=payload)
             effects.append(PlaybackEffect("player_idle", slot=slot, message=str(reason or "idle")))
             if self.pending is not None:
@@ -280,29 +436,33 @@ class PlaybackReducer:
             return effects
 
         if state == "playing":
-            previous = self.store.player_observation()
-            actual_paused = previous.paused if previous.active_slot == slot else None
             self._active_evidence(
                 slot=slot,
                 path=path,
-                paused=actual_paused,
+                paused=None,
                 status=payload,
             )
             effects.append(PlaybackEffect("player_active", slot=slot))
             if (
                 self.pending is not None
                 and self.pending.action == "play"
-                and self._matches_pending(slot, path)
             ):
-                effects.append(self._finish_operation("confirmed"))
+                if self._matches_pending(slot, path):
+                    effects.append(self._finish_operation("confirmed"))
+                else:
+                    effects.append(
+                        self._finish_operation(
+                            "failed",
+                            "Player became active with a different slot or path",
+                        )
+                    )
             return effects
 
         if state in {"paused", "toggled"}:
-            previous = self.store.player_observation()
             self._active_evidence(
                 slot=slot,
                 path=path,
-                paused=previous.paused if previous.active_slot == slot else None,
+                paused=None,
                 status=payload,
             )
             effects.append(
@@ -312,15 +472,24 @@ class PlaybackReducer:
                     message=f"Player issued {state}; awaiting telemetry",
                 )
             )
+            if self.pending is not None and self.pending.action == "play":
+                if self._matches_pending(slot, path):
+                    effects.append(self._finish_operation("confirmed", state))
+                else:
+                    effects.append(
+                        self._finish_operation(
+                            "failed",
+                            "Player control status identified different playback",
+                        )
+                    )
             return effects
 
         if state == "busy":
             if slot is not None or path is not None:
-                previous = self.store.player_observation()
                 self._active_evidence(
                     slot=slot,
                     path=path,
-                    paused=previous.paused if previous.active_slot == slot else None,
+                    paused=None,
                     status=payload,
                 )
             effects.append(PlaybackEffect("player_busy", slot=slot, message=payload.get("message")))
@@ -338,14 +507,29 @@ class PlaybackReducer:
                     kind="player_active",
                     details={"error": payload.get("message")},
                 )
+            elif active_path is not None:
+                self.store.reconcile_external_active()
             self.store.set_player_observation(
                 availability="online",
                 phase="error",
                 active_slot=active_slot,
                 active_path=active_path,
-                paused=previous.paused,
-                reconciliation_state=("confirmed" if active_slot is not None else "unknown"),
+                paused=(
+                    previous.paused
+                    if self._same_playback_identity(
+                        previous,
+                        active_slot,
+                        active_path,
+                    )
+                    else None
+                ),
+                reconciliation_state=(
+                    "confirmed"
+                    if active_slot is not None or active_path is not None
+                    else "unknown"
+                ),
                 reconciliation_due_at=None,
+                reconciliation_id=None,
                 status_payload=payload,
             )
             message = str(payload.get("message") or "Player error")
@@ -360,8 +544,9 @@ class PlaybackReducer:
     def handle_telemetry(self, payload: Any) -> list[PlaybackEffect]:
         if not isinstance(payload, Mapping) or payload.get("state") != "telemetry":
             return [PlaybackEffect("warning", message="invalid Player telemetry envelope")]
-        slot = self._slot(payload.get("slot"))
-        path = self._path(payload.get("path"))
+        slot, path, identity_warning = self._normalized_identity(payload)
+        if identity_warning is not None:
+            return [PlaybackEffect("warning", slot=slot, message=identity_warning)]
         effects: list[PlaybackEffect] = []
         if path is None:
             self._idle_evidence(telemetry=payload)
@@ -383,7 +568,15 @@ class PlaybackReducer:
         )
         effects.append(PlaybackEffect("player_telemetry", slot=observation.active_slot))
         operation = self.pending
-        if operation is None or not self._matches_pending(slot, path):
+        if operation is None:
+            return effects
+        if not self._matches_pending(slot, path):
+            effects.append(
+                self._finish_operation(
+                    "failed",
+                    "Player telemetry identified a different slot or path",
+                )
+            )
             return effects
         if operation.action == "play":
             effects.append(self._finish_operation("confirmed", "matching telemetry"))
@@ -399,48 +592,165 @@ class PlaybackReducer:
         operation = self.pending
         if operation is None or self._monotonic() < operation.deadline:
             return None
-        if operation.action == "play" and operation.slot is not None:
-            due = None
-            if self.config.storage.uncertain_review_after_seconds > 0:
-                due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
-            self.store.reconcile_active_guard(
-                operation.slot,
-                kind="restart_uncertain",
-                review_after=due,
-                details={"operation_id": operation.operation_id, "reason": "play timeout"},
-            )
-            self.store.set_player_observation(
-                availability="unknown",
-                phase="unknown",
-                active_slot=operation.slot,
-                active_path=operation.expected_path,
-                paused=None,
-                reconciliation_state="awaiting_evidence",
-                reconciliation_due_at=due,
+        if operation.action == "play":
+            previous = self.store.player_observation()
+            has_confirmed_active = (
+                previous.reconciliation_state == "confirmed"
+                and (previous.active_slot is not None or previous.active_path is not None)
             )
+            if has_confirmed_active and self._matches_pending(
+                previous.active_slot,
+                previous.active_path,
+            ):
+                return self._finish_operation(
+                    "confirmed", "matching authoritative Player evidence"
+                )
+            if not has_confirmed_active:
+                self._mark_play_uncertain(operation, "play timeout")
         elif operation.action == "stop":
             previous = self.store.player_observation()
             if previous.active_slot is not None:
                 due = None
                 if self.config.storage.uncertain_review_after_seconds > 0:
                     due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
-                self.store.reconcile_active_guard(
+                guard = self.store.mark_slot_uncertain(
                     previous.active_slot,
-                    kind="restart_uncertain",
                     review_after=due,
                     details={"operation_id": operation.operation_id, "reason": "stop timeout"},
                 )
+                due = guard.review_after
                 self.store.set_player_observation(
                     availability="unknown",
                     phase="unknown",
                     active_slot=previous.active_slot,
                     active_path=previous.active_path,
                     paused=previous.paused,
-                    reconciliation_state="awaiting_evidence",
+                    reconciliation_state=(
+                        "operator_review"
+                        if guard.state == "review_required"
+                        else "awaiting_evidence"
+                    ),
                     reconciliation_due_at=due,
+                    reconciliation_id=guard.guard_id,
+                )
+            elif previous.active_path is not None:
+                due = None
+                if self.config.storage.uncertain_review_after_seconds > 0:
+                    due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+                self.store.mark_external_uncertain(
+                    previous.active_path,
+                    review_after=due,
+                    details={
+                        "operation_id": operation.operation_id,
+                        "reason": "stop timeout",
+                    },
                 )
         return self._finish_operation("timed_out", "no authoritative Player confirmation")
 
+    def _mark_play_uncertain(self, operation: PlayerOperation, reason: str) -> None:
+        due = None
+        if self.config.storage.uncertain_review_after_seconds > 0:
+            due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+        slot = operation.slot or self._canonical_slot_for_path(operation.expected_path)
+        if slot is not None:
+            guard = self.store.mark_slot_uncertain(
+                slot,
+                review_after=due,
+                details={"operation_id": operation.operation_id, "reason": reason},
+            )
+            due = guard.review_after
+            self.store.set_player_observation(
+                availability="unknown",
+                phase="unknown",
+                active_slot=slot,
+                active_path=operation.expected_path,
+                paused=None,
+                reconciliation_state=(
+                    "operator_review"
+                    if guard.state == "review_required"
+                    else "awaiting_evidence"
+                ),
+                reconciliation_due_at=due,
+                reconciliation_id=guard.guard_id,
+            )
+        elif operation.expected_path is not None:
+            self.store.mark_external_uncertain(
+                operation.expected_path,
+                review_after=due,
+                details={"operation_id": operation.operation_id, "reason": reason},
+            )
+
+    def publication_uncertain(self, reason: str) -> PlaybackEffect | None:
+        if self.pending is None:
+            return None
+        operation = self.pending
+        if operation.action == "play":
+            self._mark_play_uncertain(operation, reason)
+        return self._finish_operation("uncertain", reason)
+
+    def mark_transport_uncertain(self, reason: str) -> PlaybackEffect:
+        previous = self.store.player_observation()
+        pending_play = (
+            self.pending
+            if self.pending is not None and self.pending.action == "play"
+            else None
+        )
+        active_slot = previous.active_slot
+        active_path = previous.active_path
+        if active_slot is None and active_path is None and pending_play is not None:
+            active_slot = pending_play.slot or self._canonical_slot_for_path(
+                pending_play.expected_path
+            )
+            active_path = pending_play.expected_path
+        due = None
+        if active_slot is not None:
+            if self.config.storage.uncertain_review_after_seconds > 0:
+                due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+            guard = self.store.mark_slot_uncertain(
+                active_slot,
+                review_after=due,
+                details={"reason": reason},
+            )
+            due = guard.review_after
+            self.store.set_player_observation(
+                availability="unknown",
+                phase="unknown",
+                active_slot=active_slot,
+                active_path=active_path,
+                paused=previous.paused,
+                reconciliation_state=(
+                    "operator_review"
+                    if guard.state == "review_required"
+                    else "awaiting_evidence"
+                ),
+                reconciliation_due_at=due,
+                reconciliation_id=guard.guard_id,
+            )
+        elif active_path is not None:
+            if self.config.storage.uncertain_review_after_seconds > 0:
+                due = self._wall_clock() + self.config.storage.uncertain_review_after_seconds
+            self.store.mark_external_uncertain(
+                active_path,
+                review_after=due,
+                details={"reason": reason},
+            )
+        else:
+            self.store.set_player_observation(
+                availability="unknown",
+                phase=previous.phase,
+                active_slot=None,
+                active_path=None,
+                paused=previous.paused,
+                reconciliation_state="unknown",
+                reconciliation_due_at=None,
+                reconciliation_id=None,
+            )
+        return PlaybackEffect(
+            "transport_uncertain",
+            slot=active_slot,
+            message=reason,
+        )
+
     def projected_slot_state(self, slot: int) -> str:
         record = self.store.get_slot(slot)
         if record.media_state != "ready":
@@ -466,9 +776,9 @@ class PlaybackReducer:
         return "ready"
 
     def mark_due_reconciliation(self) -> list[PlaybackEffect]:
-        due = self.store.mark_due_guards_for_review(self._wall_clock())
-        if not due:
-            return []
+        now = self._wall_clock()
+        due = self.store.mark_due_guards_for_review(now)
+        external_due = self.store.mark_external_reconciliation_due(now)
         observation = self.store.player_observation()
         if observation.active_slot is not None and any(
             guard.slot == observation.active_slot for guard in due
@@ -481,15 +791,32 @@ class PlaybackReducer:
                 paused=observation.paused,
                 reconciliation_state="operator_review",
                 reconciliation_due_at=observation.reconciliation_due_at,
+                reconciliation_id=observation.reconciliation_id,
             )
-        return [
+        effects = [
             PlaybackEffect(
                 "operator_review",
                 slot=guard.slot,
+                operation_id=guard.guard_id,
                 message=(
                     "Playback remains unknown and protected; fresh Player evidence "
-                    "or an administrative clear is required"
+                    "or an administrative clear is required "
+                    f"(review_id={guard.guard_id})"
                 ),
             )
             for guard in due
         ]
+        if external_due is not None:
+            review_id, path = external_due
+            effects.append(
+                PlaybackEffect(
+                    "operator_review",
+                    operation_id=review_id,
+                    message=(
+                        f"Playback at {path!r} remains unknown and protected; "
+                        "fresh Player evidence or an administrative clear is required "
+                        f"(review_id={review_id})"
+                    ),
+                )
+            )
+        return effects

controller/router.py

@@ -0,0 +1,422 @@
+"""Timeout-bounded Audio Router acquisition and restoration leases."""
+
+from __future__ import annotations
+
+import time
+import uuid
+from dataclasses import dataclass
+from typing import Callable
+
+from .config import ControllerConfig
+from .models import AudioLeaseRecord
+from .mqtt_adapter import PublishReceipt
+from .store import ControllerStore
+
+
+ModePublisher = Callable[[str], PublishReceipt]
+
+
+@dataclass(frozen=True)
+class RouterDecision:
+    action: str
+    lease_id: str | None = None
+    warning: str | None = None
+    continued: bool = False
+
+
+class AudioRouterCoordinator:
+    """Own one conservative Router lease for one playback session."""
+
+    def __init__(
+        self,
+        config: ControllerConfig,
+        store: ControllerStore,
+        publish_mode: ModePublisher,
+        *,
+        clock=time.monotonic,
+    ):
+        self.config = config
+        self.store = store
+        self.publish_mode = publish_mode
+        self._clock = clock
+
+    def observe_mode(self, mode: str) -> RouterDecision:
+        mode = mode.strip()
+        if not mode:
+            warning = "Audio Router published an empty mode"
+            self._record_warning(warning)
+            return RouterDecision("warning", warning=warning)
+        self.store.set_router_observation(mode=mode, freshness="current")
+        lease = self.store.live_audio_lease()
+        if lease is None:
+            return RouterDecision("observed")
+
+        if lease.state == "acquiring":
+            if mode == lease.target_mode:
+                self.store.update_audio_lease(
+                    lease.lease_id,
+                    state="held",
+                    confirmation_deadline=None,
+                    acquired=True,
+                )
+                return RouterDecision("ready", lease_id=lease.lease_id)
+            # State is evidence of the Router's current output, not evidence
+            # that an operator issued a command. It may simply be a periodic
+            # repeat of the pre-acquisition state, so keep waiting for the
+            # finite confirmation deadline.
+            return RouterDecision("observed", lease_id=lease.lease_id)
+
+        if lease.state == "held":
+            if mode != lease.target_mode:
+                warning = (
+                    f"Audio Router changed from Controller-owned {lease.target_mode!r} "
+                    f"to {mode!r}; automatic restoration was relinquished"
+                )
+                self.store.update_audio_lease(
+                    lease.lease_id,
+                    state="ownership_lost",
+                    confirmation_deadline=None,
+                    warning=warning,
+                    released=True,
+                )
+                self._record_warning(warning)
+                return RouterDecision("ownership_lost", lease.lease_id, warning)
+            return RouterDecision("observed", lease_id=lease.lease_id)
+
+        if lease.state == "unconfirmed":
+            # A late state event must never resurrect a lease that already
+            # timed out or deliberately continued without confirmation.
+            return RouterDecision("observed", lease_id=lease.lease_id)
+
+        if lease.state == "restore_pending":
+            if mode == lease.restoration_mode:
+                self.store.update_audio_lease(
+                    lease.lease_id,
+                    state="released",
+                    confirmation_deadline=None,
+                    released=True,
+                )
+                return RouterDecision("restored", lease_id=lease.lease_id)
+            # The Router may repeat its current state before applying the
+            # restoration command. Wait for the bounded deadline; explicit
+            # mode-set intent is handled separately below.
+            return RouterDecision("observed", lease_id=lease.lease_id)
+        return RouterDecision("observed", lease_id=lease.lease_id)
+
+    def observe_mode_command(
+        self,
+        mode: str,
+        *,
+        retained: bool = False,
+        controller_echo: bool = False,
+    ) -> RouterDecision:
+        """Observe a Router mode-set publication without treating state as intent.
+
+        The MQTT adapter marks an exact echo received while a Controller QoS 1
+        publication is in flight. Every other live mode-set publication is
+        conservatively treated as operator intent. Retained commands are old
+        broker state and can never establish current operator intent.
+        """
+
+        mode = mode.strip()
+        if not mode:
+            warning = "Audio Router published an empty mode command"
+            self._record_warning(warning)
+            return RouterDecision("warning", warning=warning)
+
+        lease = self.store.live_audio_lease()
+        lease_id = None if lease is None else lease.lease_id
+        if retained:
+            warning = (
+                f"Retained Audio Router mode command {mode!r} was ignored as stale intent"
+            )
+            self._record_warning(warning)
+            return RouterDecision("retained_ignored", lease_id, warning)
+        if controller_echo:
+            return RouterDecision("self_echo", lease_id=lease_id)
+        if lease is None:
+            return RouterDecision("operator_intent")
+
+        if lease.state == "acquiring":
+            return self._acquisition_failed(
+                lease.lease_id,
+                (
+                    f"Audio Router received an operator mode command for {mode!r} "
+                    f"while {lease.target_mode!r} confirmation was pending; "
+                    "automatic restoration ownership was relinquished"
+                ),
+            )
+
+        if lease.state in {"held", "unconfirmed", "restore_pending"}:
+            warning = (
+                f"Audio Router received an operator mode command for {mode!r}; "
+                "automatic restoration was relinquished"
+            )
+            self.store.update_audio_lease(
+                lease.lease_id,
+                state="ownership_lost",
+                confirmation_deadline=None,
+                warning=warning,
+                released=True,
+            )
+            self._record_warning(warning)
+            return RouterDecision("ownership_lost", lease.lease_id, warning)
+        return RouterDecision("operator_intent", lease_id=lease.lease_id)
+
+    def acquire(self, operation_id: str) -> RouterDecision:
+        observation = self.store.router_observation()
+        known = observation.freshness == "current" and observation.mode is not None
+        target = self.config.router.required_mode
+        previous = observation.mode if known else None
+        lease_id = uuid.uuid4().hex
+
+        if known and previous == target:
+            self.store.create_audio_lease(
+                lease_id=lease_id,
+                operation_id=operation_id,
+                state="held",
+                previous_mode=previous,
+                target_mode=target,
+                restoration_mode=None,
+                restoration_source="none",
+                controller_changed=False,
+                confirmation_deadline=None,
+            )
+            self.store.update_audio_lease(
+                lease_id,
+                state="held",
+                acquired=True,
+            )
+            return RouterDecision("ready", lease_id=lease_id)
+
+        if known:
+            restoration_mode = previous
+            restoration_source = "observed_previous"
+        elif self.config.router.fallback_restore_mode is not None:
+            restoration_mode = self.config.router.fallback_restore_mode
+            restoration_source = "configured_fallback"
+        else:
+            restoration_mode = None
+            restoration_source = "none"
+
+        policy = self.config.router.when_state_unknown
+        if not known and policy == "reject":
+            warning = "Audio Router state is unavailable; playback was rejected by policy"
+            self.store.create_audio_lease(
+                lease_id=lease_id,
+                operation_id=operation_id,
+                state="abandoned",
+                previous_mode=None,
+                target_mode=target,
+                restoration_mode=restoration_mode,
+                restoration_source=restoration_source,
+                controller_changed=False,
+                confirmation_deadline=None,
+                warning=warning,
+            )
+            self._record_warning(warning)
+            return RouterDecision("rejected", lease_id, warning)
+
+        deadline = self._clock() + self.config.router.confirmation_timeout_seconds
+        initial_state = "unconfirmed" if not known and policy == "continue" else "acquiring"
+        self.store.create_audio_lease(
+            lease_id=lease_id,
+            operation_id=operation_id,
+            state=initial_state,
+            previous_mode=previous,
+            target_mode=target,
+            restoration_mode=restoration_mode,
+            restoration_source=restoration_source,
+            controller_changed=True,
+            confirmation_deadline=(deadline if initial_state == "acquiring" else None),
+        )
+        receipt = self.publish_mode(target)
+        if not receipt.accepted:
+            return self._acquisition_failed(
+                lease_id,
+                f"could not request Audio Router mode {target!r}: {receipt.reason}",
+            )
+
+        if initial_state == "unconfirmed":
+            warning = (
+                f"Audio Router state is unavailable; requested {target!r} and "
+                "continued immediately without restoration ownership"
+            )
+            self.store.update_audio_lease(
+                lease_id,
+                state="unconfirmed",
+                warning=warning,
+            )
+            self._record_warning(warning)
+            return RouterDecision("ready", lease_id, warning, continued=True)
+        return RouterDecision("waiting", lease_id=lease_id)
+
+    def check_deadline(self) -> RouterDecision | None:
+        lease = self.store.live_audio_lease()
+        if (
+            lease is None
+            or lease.confirmation_deadline is None
+            or self._clock() < lease.confirmation_deadline
+        ):
+            return None
+        if lease.state == "acquiring":
+            return self._acquisition_failed(
+                lease.lease_id,
+                (
+                    f"Audio Router did not confirm {lease.target_mode!r} within "
+                    f"{self.config.router.confirmation_timeout_seconds:g} seconds"
+                ),
+            )
+        if lease.state == "restore_pending":
+            warning = (
+                f"Audio Router did not confirm restoration to {lease.restoration_mode!r}; "
+                "no further automatic change will be attempted"
+            )
+            self.store.update_audio_lease(
+                lease.lease_id,
+                state="abandoned",
+                confirmation_deadline=None,
+                warning=warning,
+                released=True,
+            )
+            self._record_warning(warning)
+            return RouterDecision("restore_failed", lease.lease_id, warning)
+        return None
+
+    def _acquisition_failed(self, lease_id: str, warning: str) -> RouterDecision:
+        continue_playback = self.config.router.continue_on_confirmation_failure
+        self.store.update_audio_lease(
+            lease_id,
+            state="unconfirmed" if continue_playback else "abandoned",
+            confirmation_deadline=None,
+            warning=warning,
+            released=not continue_playback,
+        )
+        self._record_warning(warning)
+        return RouterDecision(
+            "ready" if continue_playback else "rejected",
+            lease_id,
+            warning,
+            continued=continue_playback,
+        )
+
+    def release(self, lease_id: str | None) -> RouterDecision:
+        if lease_id is None:
+            return RouterDecision("released")
+        lease = self.store.get_audio_lease(lease_id)
+        if lease is None:
+            return RouterDecision("released")
+        if lease.state == "held" and not lease.controller_changed:
+            self.store.update_audio_lease(
+                lease_id,
+                state="released",
+                confirmation_deadline=None,
+                released=True,
+            )
+            return RouterDecision("released", lease_id)
+        if lease.state != "held":
+            if lease.state in {"unconfirmed", "acquiring"}:
+                self.store.update_audio_lease(
+                    lease_id,
+                    state="abandoned",
+                    confirmation_deadline=None,
+                    warning="Audio Router was never confirmed; restoration was skipped",
+                    released=True,
+                )
+            return RouterDecision("released", lease_id)
+
+        observation = self.store.router_observation()
+        if observation.freshness != "current" or observation.mode != lease.target_mode:
+            warning = "Audio Router ownership is no longer current; restoration was skipped"
+            self.store.update_audio_lease(
+                lease_id,
+                state="ownership_lost",
+                confirmation_deadline=None,
+                warning=warning,
+                released=True,
+            )
+            self._record_warning(warning)
+            return RouterDecision("ownership_lost", lease_id, warning)
+        if lease.restoration_mode is None:
+            self.store.update_audio_lease(
+                lease_id,
+                state="released",
+                confirmation_deadline=None,
+                released=True,
+            )
+            return RouterDecision("released", lease_id)
+
+        receipt = self.publish_mode(lease.restoration_mode)
+        if not receipt.accepted:
+            warning = f"could not request Audio Router restoration: {receipt.reason}"
+            self.store.update_audio_lease(
+                lease_id,
+                state="abandoned",
+                confirmation_deadline=None,
+                warning=warning,
+                released=True,
+            )
+            self._record_warning(warning)
+            return RouterDecision("restore_failed", lease_id, warning)
+        self.store.update_audio_lease(
+            lease_id,
+            state="restore_pending",
+            confirmation_deadline=(
+                self._clock() + self.config.router.confirmation_timeout_seconds
+            ),
+        )
+        return RouterDecision("restore_waiting", lease_id)
+
+    def connection_lost(self) -> RouterDecision | None:
+        observation = self.store.router_observation()
+        self.store.set_router_observation(
+            mode=observation.mode,
+            freshness="stale",
+            warning="MQTT continuity was lost",
+        )
+        lease = self.store.live_audio_lease()
+        if lease is None:
+            return None
+        if lease.state == "acquiring":
+            return self._acquisition_failed(
+                lease.lease_id,
+                "MQTT disconnected while waiting for Audio Router confirmation",
+            )
+        warning = "MQTT continuity was lost; automatic Audio Router restoration was relinquished"
+        self.store.update_audio_lease(
+            lease.lease_id,
+            state="ownership_lost",
+            confirmation_deadline=None,
+            warning=warning,
+            released=True,
+        )
+        self._record_warning(warning)
+        return RouterDecision("ownership_lost", lease.lease_id, warning)
+
+    def abandon(self, lease_id: str | None, reason: str) -> None:
+        if lease_id is None:
+            return
+        lease = self.store.get_audio_lease(lease_id)
+        if lease is None or lease.state in {"released", "abandoned", "ownership_lost"}:
+            return
+        self.store.update_audio_lease(
+            lease_id,
+            state="abandoned",
+            confirmation_deadline=None,
+            warning=reason,
+            released=True,
+        )
+        self._record_warning(reason)
+
+    def _record_warning(self, warning: str) -> None:
+        observation = self.store.router_observation()
+        self.store.set_router_observation(
+            mode=observation.mode,
+            freshness=observation.freshness,
+            warning=warning,
+        )
+        self.store.record_audit_event(
+            "audio_router_warning",
+            actor="controller",
+            details={"message": warning},
+        )

tests/test_config_contracts.py

@@ -29,6 +29,7 @@ class ContractCharacterizationTests(unittest.TestCase):
         self.assertEqual(topics.playback_base, "youtube/media/playback")
         self.assertEqual(topics.router_mode_set, "audio/player/mode/set")
         self.assertEqual(topics.router_mode_state, "audio/player/mode/state")
+        self.assertEqual(topics.controller_result, "youtube/controller/v2/events")
         self.assertEqual(PLAYER_COMMAND_QOS, 1)
         self.assertIs(PLAYER_COMMAND_RETAIN, False)
         self.assertEqual(SLOT_IDS, tuple(range(1, 11)))
@@ -79,6 +80,7 @@ class ConfigurationTests(unittest.TestCase):
         self.assertIs(config.router.continue_on_confirmation_failure, True)
         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)
 
     def test_documented_environment_overrides_apply(self):
         env = {

tests/test_downloads.py

@@ -1,6 +1,8 @@
 import os
 import subprocess
 import tempfile
+import threading
+import time
 import unittest
 from concurrent.futures import Future
 from pathlib import Path
@@ -194,6 +196,42 @@ class DownloadManagerTests(unittest.TestCase):
         self.assertFalse(slot.exists())
         self.assertEqual(self.store.get_slot(1).media_state, "empty")
 
+    def test_shutdown_terminates_tracked_process_without_waiting_for_job_timeout(self):
+        manager = DownloadManager(self.config, self.store)
+        errors = []
+
+        def run_long_process():
+            try:
+                manager._run_command(
+                    ["/bin/sh", "-c", "sleep 60"],
+                    check=True,
+                    text=True,
+                    timeout=60,
+                    capture_output=True,
+                )
+            except BaseException as exc:
+                errors.append(exc)
+
+        worker = threading.Thread(target=run_long_process)
+        worker.start()
+        deadline = time.monotonic() + 2
+        while time.monotonic() < deadline:
+            with manager._lock:
+                if manager._processes:
+                    break
+            time.sleep(0.01)
+        else:
+            self.fail("test process did not start")
+
+        started = time.monotonic()
+        manager.shutdown(wait=False)
+        worker.join(timeout=3)
+        self.assertFalse(worker.is_alive())
+        self.assertLess(time.monotonic() - started, 3)
+        self.assertTrue(errors)
+        with manager._lock:
+            self.assertEqual(manager._processes, set())
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_mqtt_adapter.py

@@ -1,5 +1,6 @@
 import json
 import queue
+import threading
 import unittest
 from types import SimpleNamespace
 
@@ -9,6 +10,20 @@ from controller.config import ControllerConfig, MqttConfig
 from controller.mqtt_adapter import MqttAdapter
 
 
+class FakeMessageInfo:
+    def __init__(self, mid, published):
+        self.rc = mqtt.MQTT_ERR_SUCCESS
+        self.mid = mid
+        self.published = published
+        self.waited = []
+
+    def wait_for_publish(self, timeout=None):
+        self.waited.append(timeout)
+
+    def is_published(self):
+        return self.published
+
+
 class FakeClient:
     def __init__(self):
         self.calls = []
@@ -17,6 +32,9 @@ class FakeClient:
         self.acks = []
         self.suppress_exceptions = False
         self.connect_error = None
+        self.loop_result = mqtt.MQTT_ERR_SUCCESS
+        self.publish_acknowledged = None
+        self.publish_hook = None
 
     def username_pw_set(self, username, password):
         self.calls.append(("username_pw_set", username, password))
@@ -42,7 +60,7 @@ class FakeClient:
 
     def loop_start(self):
         self.calls.append(("loop_start",))
-        return mqtt.MQTT_ERR_SUCCESS
+        return self.loop_result
 
     def loop_stop(self):
         self.calls.append(("loop_stop",))
@@ -56,6 +74,12 @@ class FakeClient:
 
     def publish(self, topic, payload, qos=0, retain=False):
         self.published.append((topic, payload, qos, retain))
+        if self.publish_hook is not None:
+            self.publish_hook(topic, payload, qos, retain)
+        if self.publish_acknowledged is not None:
+            return FakeMessageInfo(
+                len(self.published), self.publish_acknowledged
+            )
         return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=len(self.published))
 
     def ack(self, message_id, qos):
@@ -106,6 +130,66 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertIn("network unreachable", self.adapter.last_error)
         self.assertNotIn("loop_start", [call[0] for call in self.client.calls])
 
+    def test_loop_start_failure_does_not_escape_startup(self):
+        self.client.loop_result = mqtt.MQTT_ERR_UNKNOWN
+        self.adapter.start()
+        self.assertTrue(self.adapter.started)
+        self.assertIn("failed to start MQTT network loop", self.adapter.last_error)
+
+    def test_stop_during_blocked_connect_never_starts_network_loop_afterward(self):
+        entered = threading.Event()
+        release = threading.Event()
+        original_connect = self.client.connect_async
+
+        def blocked_connect(*args, **kwargs):
+            entered.set()
+            release.wait(timeout=2)
+            return original_connect(*args, **kwargs)
+
+        self.client.connect_async = blocked_connect
+        starter = threading.Thread(target=self.adapter.start)
+        starter.start()
+        self.assertTrue(entered.wait(timeout=1))
+        self.adapter.stop()
+        release.set()
+        starter.join(timeout=2)
+        self.assertFalse(starter.is_alive())
+        self.assertNotIn("loop_start", [call[0] for call in self.client.calls])
+        self.assertFalse(self.adapter.started)
+
+    def test_stop_wins_against_connect_callback_blocked_during_subscribe(self):
+        self.adapter.start()
+        entered = threading.Event()
+        release = threading.Event()
+        original_subscribe = self.client.subscribe
+
+        def blocked_subscribe(topic, qos=0):
+            entered.set()
+            release.wait(timeout=2)
+            return original_subscribe(topic, qos=qos)
+
+        self.client.subscribe = blocked_subscribe
+        callback = threading.Thread(
+            target=self.adapter._on_connect,
+            args=(self.client, None, None, 0),
+        )
+        callback.start()
+        self.assertTrue(entered.wait(timeout=1))
+        stopper = threading.Thread(target=self.adapter.stop)
+        stopper.start()
+        release.set()
+        callback.join(timeout=2)
+        stopper.join(timeout=2)
+        self.assertFalse(callback.is_alive())
+        self.assertFalse(stopper.is_alive())
+        self.assertFalse(self.adapter.started)
+        self.assertFalse(self.adapter.connected)
+        self.assertLess(len(self.client.subscriptions), 6)
+        self.assertTrue(self.events.empty())
+        self.assertFalse(
+            self.adapter.publish_player_command({"action": "stop"}).accepted
+        )
+
     def test_connect_subscribes_to_all_authoritative_inputs(self):
         self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
@@ -117,6 +201,7 @@ class MqttAdapterTests(unittest.TestCase):
                 ("youtube/player/telemetry", 0),
                 ("youtube/player/state", 0),
                 ("youtube/controller/cmd", 1),
+                ("audio/player/mode/set", 1),
                 ("audio/player/mode/state", 1),
             ],
         )
@@ -127,6 +212,7 @@ class MqttAdapterTests(unittest.TestCase):
     def test_player_commands_are_exact_qos_one_and_nonretained(self):
         disconnected = self.adapter.publish_player_command({"action": "stop"})
         self.assertFalse(disconnected.accepted)
+        self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
         receipt = self.adapter.publish_player_command(
@@ -140,6 +226,7 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertIs(retain, False)
 
     def test_router_commands_are_qos_one_nonretained_plain_strings(self):
+        self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
         receipt = self.adapter.publish_router_mode("video")
@@ -149,7 +236,75 @@ 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):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+
+        def echo(topic, payload, qos, retain):
+            self.adapter._on_message(
+                self.client,
+                None,
+                SimpleNamespace(
+                    topic=topic,
+                    payload=payload.encode("utf-8"),
+                    qos=qos,
+                    retain=retain,
+                    dup=False,
+                    mid=31,
+                ),
+            )
+
+        self.client.publish_hook = echo
+        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,
+            None,
+            SimpleNamespace(
+                topic="audio/player/mode/set",
+                payload=b"video",
+                qos=1,
+                retain=False,
+                dup=False,
+                mid=32,
+            ),
+        )
+        self.assertFalse(self.events.get_nowait().controller_echo)
+
+    def test_retained_router_set_never_consumes_controller_echo_marker(self):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+
+        def retained_then_live_echo(topic, payload, qos, retain):
+            for message_retain, mid in ((True, 41), (False, 42)):
+                self.adapter._on_message(
+                    self.client,
+                    None,
+                    SimpleNamespace(
+                        topic=topic,
+                        payload=payload.encode("utf-8"),
+                        qos=qos,
+                        retain=message_retain,
+                        dup=False,
+                        mid=mid,
+                    ),
+                )
+
+        self.client.publish_hook = retained_then_live_echo
+        self.assertTrue(self.adapter.publish_router_mode("video").accepted)
+        retained_event = self.events.get_nowait()
+        live_event = self.events.get_nowait()
+        self.assertTrue(retained_event.retain)
+        self.assertFalse(retained_event.controller_echo)
+        self.assertFalse(live_event.retain)
+        self.assertTrue(live_event.controller_echo)
+
     def test_qos_message_is_acked_only_after_consumer_calls_acknowledge(self):
+        self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
         message = SimpleNamespace(
@@ -167,6 +322,7 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertEqual(self.client.acks, [(42, 1)])
 
     def test_stale_mid_from_previous_connection_is_not_acked(self):
+        self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
         message = SimpleNamespace(
@@ -184,6 +340,7 @@ class MqttAdapterTests(unittest.TestCase):
         self.assertEqual(self.client.acks, [])
 
     def test_retained_flag_is_preserved_for_coordinator_rejection(self):
+        self.adapter.start()
         self.adapter._on_connect(self.client, None, None, 0)
         self.events.get_nowait()
         message = SimpleNamespace(
@@ -196,6 +353,82 @@ class MqttAdapterTests(unittest.TestCase):
         self.adapter._on_message(self.client, None, message)
         self.assertIs(self.events.get_nowait().retain, True)
 
+    def test_duplicate_flag_is_preserved_for_coordinator_deduplication(self):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        message = SimpleNamespace(
+            topic="youtube/controller/cmd",
+            payload=b'{"action":"toggle"}',
+            qos=1,
+            retain=False,
+            dup=True,
+            mid=9,
+        )
+        self.adapter._on_message(self.client, None, message)
+        self.assertIs(self.events.get_nowait().duplicate, True)
+
+    def test_missing_puback_is_uncertain_and_invalidates_connection(self):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        self.client.publish_acknowledged = False
+        receipt = self.adapter.publish_player_command({"action": "stop"})
+        self.assertFalse(receipt.accepted)
+        self.assertTrue(receipt.uncertain)
+        self.assertIn("no broker PUBACK", receipt.reason)
+        self.assertFalse(self.adapter.connected)
+
+    def test_optional_qos_event_puback_failure_does_not_invalidate_transport(self):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        self.client.publish_acknowledged = False
+        receipt = self.adapter.publish_json(
+            "youtube/controller/v2/events",
+            {"version": 2, "type": "warning"},
+            qos=1,
+            retain=False,
+        )
+        self.assertFalse(receipt.accepted)
+        self.assertTrue(receipt.uncertain)
+        self.assertTrue(self.adapter.connected)
+
+    def test_discarded_transport_releases_callback_blocked_on_full_queue(self):
+        events = queue.Queue(maxsize=1)
+        client = FakeClient()
+        adapter = MqttAdapter(self.config, client=client, events=events)
+        try:
+            adapter.start()
+            adapter._on_connect(client, None, None, 0)
+            self.assertTrue(events.full())
+            callback = threading.Thread(
+                target=adapter._on_message,
+                args=(
+                    client,
+                    None,
+                    SimpleNamespace(
+                        topic="youtube/controller/cmd",
+                        payload=b'{"action":"stop"}',
+                        qos=1,
+                        retain=False,
+                        dup=False,
+                        mid=99,
+                    ),
+                ),
+            )
+            callback.start()
+            callback.join(timeout=0.05)
+            self.assertTrue(callback.is_alive())
+
+            adapter._discard_unacknowledged_transport("test transport discard")
+            callback.join(timeout=1)
+            self.assertFalse(callback.is_alive())
+            self.assertEqual(events.qsize(), 1)
+            self.assertEqual(events.get_nowait().kind, "connected")
+        finally:
+            adapter.stop()
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_playback.py

@@ -230,6 +230,154 @@ class PlaybackReducerTests(unittest.TestCase):
             effects = self.reducer.handle_status(payload)
             self.assertEqual(effects[0].kind, "warning")
 
+    def test_contradictory_slot_and_path_cannot_confirm_play(self):
+        expected = str(self.config.storage.media_base / "slot1" / "playlist.m3u")
+        self.reducer.begin_operation("play", slot=1, expected_path=expected)
+        effects = self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": str(
+                    self.config.storage.media_base / "slot2" / "playlist.m3u"
+                ),
+            }
+        )
+        self.assertEqual(effects[0].kind, "warning")
+        self.assertIsNotNone(self.reducer.pending)
+        self.assertEqual(self.store.player_observation().phase, "unknown")
+
+    def test_mismatched_active_evidence_fails_intent_and_is_not_overwritten(self):
+        self.reducer.begin_operation(
+            "play",
+            slot=1,
+            expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
+        )
+        effects = self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 2,
+                "path": "/mnt/Media/YTPlayer/slot2/playlist.m3u",
+            }
+        )
+        self.assertIn("operation_failed", [effect.kind for effect in effects])
+        self.assertIsNone(self.reducer.pending)
+        self.monotonic.value += 100
+        self.assertIsNone(self.reducer.check_timeout())
+        observation = self.store.player_observation()
+        self.assertEqual(observation.active_slot, 2)
+        self.assertEqual(observation.phase, "playing")
+        self.assertEqual(self.store.list_live_guards()[0].slot, 2)
+
+    def test_bare_idle_does_not_clear_live_playback(self):
+        path = "/mnt/Media/YTPlayer/slot1/playlist.m3u"
+        self.reducer.handle_status({"state": "playing", "slot": 1, "path": path})
+        effects = self.reducer.handle_status({"state": "idle", "slot": 1})
+        self.assertEqual(effects[0].kind, "warning")
+        self.assertEqual(self.store.player_observation().phase, "playing")
+        self.assertEqual(self.store.list_live_guards()[0].slot, 1)
+
+    def test_path_only_evidence_derives_and_protects_managed_slot(self):
+        path = str(self.config.storage.media_base / "slot2" / "playlist.m3u")
+        self.reducer.handle_telemetry(
+            {"state": "telemetry", "slot": None, "path": path, "paused": False}
+        )
+        observation = self.store.player_observation()
+        self.assertEqual(observation.active_slot, 2)
+        self.assertEqual(self.store.list_live_guards()[0].slot, 2)
+
+    def test_duplicate_offline_evidence_does_not_postpone_review(self):
+        path = "/mnt/Media/YTPlayer/slot1/playlist.m3u"
+        self.reducer.handle_status({"state": "playing", "slot": 1, "path": path})
+        self.reducer.handle_status({"state": "offline"})
+        first_deadline = self.store.list_live_guards()[0].review_after
+        self.wall.value += 20
+        self.reducer.handle_status({"state": "offline"})
+        self.assertEqual(
+            self.store.list_live_guards()[0].review_after,
+            first_deadline,
+        )
+
+    def test_repeated_active_telemetry_reuses_guard_without_history_churn(self):
+        path = str(self.config.storage.media_base / "slot1" / "playlist.m3u")
+        payload = {
+            "state": "telemetry",
+            "slot": 1,
+            "path": path,
+            "paused": False,
+        }
+        self.reducer.handle_telemetry(payload)
+        first_guard = self.store.list_live_guards()[0]
+        for _ in range(20):
+            self.reducer.handle_telemetry(payload)
+        self.assertEqual(self.store.list_live_guards()[0].guard_id, first_guard.guard_id)
+        count = self.store.connection.execute(
+            "SELECT COUNT(*) FROM slot_guards"
+        ).fetchone()[0]
+        self.assertEqual(count, 1)
+
+    def test_external_play_timeout_transitions_to_review_without_assuming_stop(self):
+        path = "/srv/operator-playlists/special-event.m3u"
+        self.reducer.begin_operation("play", expected_path=path)
+        self.monotonic.value += 6
+        timeout = self.reducer.check_timeout()
+        self.assertEqual(timeout.kind, "operation_timed_out")
+        uncertain = self.store.player_observation()
+        self.assertEqual(uncertain.active_path, path)
+        self.assertEqual(uncertain.reconciliation_state, "awaiting_evidence")
+        self.assertIsNotNone(uncertain.reconciliation_id)
+
+        self.wall.value += 31
+        effects = self.reducer.mark_due_reconciliation()
+        self.assertEqual(len(effects), 1)
+        self.assertEqual(effects[0].kind, "operator_review")
+        self.assertEqual(effects[0].operation_id, uncertain.reconciliation_id)
+        review = self.store.player_observation()
+        self.assertEqual(review.reconciliation_state, "operator_review")
+        self.assertEqual(review.active_path, path)
+
+    def test_external_pause_does_not_bleed_to_a_different_path(self):
+        first_path = "/srv/operator-playlists/first.m3u"
+        second_path = "/srv/operator-playlists/second.m3u"
+        self.reducer.handle_telemetry(
+            {
+                "state": "telemetry",
+                "path": first_path,
+                "paused": True,
+            }
+        )
+        self.assertIs(self.store.player_observation().paused, True)
+
+        self.reducer.handle_status(
+            {"state": "playing", "path": second_path}
+        )
+        observation = self.store.player_observation()
+        self.assertEqual(observation.active_path, second_path)
+        self.assertIsNone(observation.paused)
+        self.assertEqual(observation.phase, "playing")
+
+    def test_stop_timeout_preserves_existing_operator_review_and_token(self):
+        path = str(self.config.storage.media_base / "slot1" / "playlist.m3u")
+        self.reducer.handle_status(
+            {"state": "playing", "slot": 1, "path": path}
+        )
+        self.reducer.handle_status({"state": "offline"})
+        self.wall.value += 31
+        self.reducer.mark_due_reconciliation()
+        before = self.store.player_observation()
+        guard_before = self.store.list_live_guards()[0]
+        self.assertEqual(before.reconciliation_state, "operator_review")
+
+        self.reducer.begin_operation("stop", slot=1, expected_path=path)
+        self.monotonic.value += 6
+        effect = self.reducer.check_timeout()
+        self.assertEqual(effect.kind, "operation_timed_out")
+        after = self.store.player_observation()
+        guard_after = self.store.list_live_guards()[0]
+        self.assertEqual(after.reconciliation_state, "operator_review")
+        self.assertEqual(after.reconciliation_id, before.reconciliation_id)
+        self.assertEqual(guard_after.guard_id, guard_before.guard_id)
+        self.assertEqual(guard_after.state, "review_required")
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_router.py

@@ -0,0 +1,316 @@
+import tempfile
+import unittest
+from dataclasses import replace
+from pathlib import Path
+
+from controller.config import ControllerConfig, RouterConfig, StorageConfig
+from controller.mqtt_adapter import PublishReceipt
+from controller.router import AudioRouterCoordinator
+from controller.store import ControllerStore
+
+
+class Clock:
+    def __init__(self, value=1000.0):
+        self.value = value
+
+    def __call__(self):
+        return self.value
+
+
+class ModePublisher:
+    def __init__(self):
+        self.modes = []
+        self.accepted = True
+        self.reason = None
+
+    def __call__(self, mode):
+        self.modes.append(mode)
+        return PublishReceipt(self.accepted, message_id=len(self.modes), reason=self.reason)
+
+
+class AudioRouterCoordinatorTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        base = Path(self.tempdir.name)
+        self.clock = Clock()
+        self.base_config = ControllerConfig(
+            storage=StorageConfig(
+                media_base=base / "media",
+                data_base=base / "data",
+            ),
+            router=RouterConfig(
+                required_mode="video",
+                confirmation_timeout_seconds=3,
+                when_state_unknown="attempt",
+                continue_on_confirmation_failure=True,
+                fallback_restore_mode=None,
+            ),
+        )
+        self.store = ControllerStore(self.base_config.storage, now=self.clock).open()
+        self.publisher = ModePublisher()
+
+    def tearDown(self):
+        self.store.close()
+        self.tempdir.cleanup()
+
+    def coordinator(self, router_config=None):
+        config = self.base_config
+        if router_config is not None:
+            config = replace(config, router=router_config)
+        return AudioRouterCoordinator(
+            config,
+            self.store,
+            self.publisher,
+            clock=self.clock,
+        )
+
+    def test_known_mode_is_confirmed_before_play_and_restored_afterward(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-one")
+        self.assertEqual(decision.action, "waiting")
+        self.assertEqual(self.publisher.modes, ["video"])
+        self.assertEqual(router.observe_mode("video").action, "ready")
+        lease = self.store.get_audio_lease(decision.lease_id)
+        self.assertEqual(lease.state, "held")
+        self.assertEqual(lease.previous_mode, "service")
+        restore = router.release(decision.lease_id)
+        self.assertEqual(restore.action, "restore_waiting")
+        self.assertEqual(self.publisher.modes, ["video", "service"])
+        self.assertEqual(router.observe_mode("service").action, "restored")
+        self.assertEqual(self.store.get_audio_lease(decision.lease_id).state, "released")
+
+    def test_default_unknown_state_timeout_continues_with_warning_not_hang(self):
+        router = self.coordinator()
+        decision = router.acquire("play-timeout")
+        self.assertEqual(decision.action, "waiting")
+        self.assertEqual(self.publisher.modes, ["video"])
+        self.clock.value += 2.9
+        self.assertIsNone(router.check_deadline())
+        self.clock.value += 0.2
+        decision = router.check_deadline()
+        self.assertEqual(decision.action, "ready")
+        self.assertTrue(decision.continued)
+        self.assertIn("did not confirm", decision.warning)
+        lease = self.store.get_audio_lease(decision.lease_id)
+        self.assertEqual(lease.state, "unconfirmed")
+        self.assertEqual(router.release(lease.lease_id).action, "released")
+        self.assertEqual(self.publisher.modes, ["video"])
+
+    def test_timeout_can_reject_when_continue_policy_is_disabled(self):
+        router = self.coordinator(
+            replace(
+                self.base_config.router,
+                continue_on_confirmation_failure=False,
+            )
+        )
+        decision = router.acquire("play-reject")
+        self.clock.value += 4
+        decision = router.check_deadline()
+        self.assertEqual(decision.action, "rejected")
+        self.assertFalse(decision.continued)
+        self.assertEqual(self.store.get_audio_lease(decision.lease_id).state, "abandoned")
+
+    def test_unknown_state_continue_policy_does_not_wait_or_gain_late_ownership(self):
+        router = self.coordinator(
+            replace(self.base_config.router, when_state_unknown="continue")
+        )
+        decision = router.acquire("play-immediate")
+        self.assertEqual(decision.action, "ready")
+        self.assertTrue(decision.continued)
+        self.assertEqual(self.store.get_audio_lease(decision.lease_id).state, "unconfirmed")
+        self.assertEqual(router.observe_mode("video").action, "observed")
+        self.assertEqual(self.store.get_audio_lease(decision.lease_id).state, "unconfirmed")
+
+    def test_unknown_state_reject_policy_does_not_publish(self):
+        router = self.coordinator(
+            replace(self.base_config.router, when_state_unknown="reject")
+        )
+        decision = router.acquire("play-reject")
+        self.assertEqual(decision.action, "rejected")
+        self.assertEqual(self.publisher.modes, [])
+
+    def test_already_video_requires_no_change_and_no_restore(self):
+        router = self.coordinator()
+        router.observe_mode("video")
+        decision = router.acquire("play-video")
+        self.assertEqual(decision.action, "ready")
+        self.assertEqual(self.publisher.modes, [])
+        self.assertEqual(router.release(decision.lease_id).action, "released")
+        self.assertEqual(self.publisher.modes, [])
+
+    def test_confirmed_unknown_acquisition_can_use_configured_fallback(self):
+        router = self.coordinator(
+            replace(self.base_config.router, fallback_restore_mode="service")
+        )
+        decision = router.acquire("play-fallback")
+        self.assertEqual(router.observe_mode("video").action, "ready")
+        lease = self.store.get_audio_lease(decision.lease_id)
+        self.assertEqual(lease.restoration_source, "configured_fallback")
+        self.assertEqual(router.release(decision.lease_id).action, "restore_waiting")
+        self.assertEqual(self.publisher.modes, ["video", "service"])
+
+    def test_operator_change_is_sticky_even_if_mode_later_returns_to_video(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-operator")
+        router.observe_mode("video")
+        lost = router.observe_mode("announcements")
+        self.assertEqual(lost.action, "ownership_lost")
+        router.observe_mode("video")
+        self.assertEqual(router.release(decision.lease_id).action, "released")
+        self.assertEqual(self.publisher.modes, ["video"])
+        self.assertEqual(
+            self.store.get_audio_lease(decision.lease_id).state,
+            "ownership_lost",
+        )
+
+    def test_connection_loss_relinquishes_confirmed_lease(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-disconnect")
+        router.observe_mode("video")
+        lost = router.connection_lost()
+        self.assertEqual(lost.action, "ownership_lost")
+        self.assertEqual(router.release(decision.lease_id).action, "released")
+        self.assertEqual(self.publisher.modes, ["video"])
+
+    def test_publish_failure_uses_confirmation_failure_policy(self):
+        self.publisher.accepted = False
+        self.publisher.reason = "MQTT disconnected"
+        router = self.coordinator()
+        decision = router.acquire("play-publish-fail")
+        self.assertEqual(decision.action, "ready")
+        self.assertTrue(decision.continued)
+        self.assertIn("MQTT disconnected", decision.warning)
+
+    def test_restoration_timeout_is_bounded_and_not_retried(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-restore-timeout")
+        router.observe_mode("video")
+        router.release(decision.lease_id)
+        self.clock.value += 4
+        timeout = router.check_deadline()
+        self.assertEqual(timeout.action, "restore_failed")
+        self.assertEqual(self.publisher.modes, ["video", "service"])
+        self.assertIsNone(router.check_deadline())
+
+    def test_repeated_previous_state_during_acquisition_waits_for_deadline(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        waiting = router.acquire("play-repeated-state")
+        self.assertEqual(router.observe_mode("service").action, "observed")
+        self.assertEqual(
+            self.store.get_audio_lease(waiting.lease_id).state,
+            "acquiring",
+        )
+        self.clock.value += 4
+        decision = router.check_deadline()
+        self.assertEqual(decision.action, "ready")
+        self.assertTrue(decision.continued)
+        self.assertEqual(
+            self.store.get_audio_lease(waiting.lease_id).state,
+            "unconfirmed",
+        )
+        router.release(waiting.lease_id)
+        self.assertEqual(self.publisher.modes, ["video"])
+
+    def test_repeated_target_state_is_observation_not_operator_takeover(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-repeated-state")
+        self.assertEqual(router.observe_mode("video").action, "ready")
+        self.assertEqual(router.observe_mode("video").action, "observed")
+        self.assertEqual(
+            self.store.get_audio_lease(decision.lease_id).state,
+            "held",
+        )
+        self.assertEqual(router.release(decision.lease_id).action, "restore_waiting")
+        self.assertEqual(self.publisher.modes, ["video", "service"])
+
+    def test_controller_mode_set_echo_does_not_relinquish_ownership(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-self-echo")
+        self.assertEqual(
+            router.observe_mode_command("video", controller_echo=True).action,
+            "self_echo",
+        )
+        self.assertEqual(router.observe_mode("video").action, "ready")
+        self.assertEqual(router.release(decision.lease_id).action, "restore_waiting")
+        self.assertEqual(
+            router.observe_mode_command("service", controller_echo=True).action,
+            "self_echo",
+        )
+        self.assertEqual(router.observe_mode("service").action, "restored")
+
+    def test_manual_mode_set_during_acquisition_uses_continue_policy(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        waiting = router.acquire("play-manual-acquire")
+        decision = router.observe_mode_command("announcements")
+        self.assertEqual(decision.action, "ready")
+        self.assertTrue(decision.continued)
+        self.assertEqual(
+            self.store.get_audio_lease(waiting.lease_id).state,
+            "unconfirmed",
+        )
+        self.assertIsNone(self.store.get_audio_lease(waiting.lease_id).confirmation_deadline)
+
+    def test_manual_mode_set_during_acquisition_can_reject(self):
+        router = self.coordinator(
+            replace(
+                self.base_config.router,
+                continue_on_confirmation_failure=False,
+            )
+        )
+        router.observe_mode("service")
+        waiting = router.acquire("play-manual-reject")
+        decision = router.observe_mode_command("announcements")
+        self.assertEqual(decision.action, "rejected")
+        self.assertFalse(decision.continued)
+        self.assertEqual(
+            self.store.get_audio_lease(waiting.lease_id).state,
+            "abandoned",
+        )
+
+    def test_manual_mode_set_while_held_relinquishes_restoration(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-manual-held")
+        router.observe_mode("video")
+        lost = router.observe_mode_command("video")
+        self.assertEqual(lost.action, "ownership_lost")
+        self.assertEqual(router.release(decision.lease_id).action, "released")
+        self.assertEqual(self.publisher.modes, ["video"])
+
+    def test_manual_mode_set_while_restoring_relinquishes_restoration(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-manual-restore")
+        router.observe_mode("video")
+        self.assertEqual(router.release(decision.lease_id).action, "restore_waiting")
+        lost = router.observe_mode_command("announcements")
+        self.assertEqual(lost.action, "ownership_lost")
+        self.assertEqual(
+            self.store.get_audio_lease(decision.lease_id).state,
+            "ownership_lost",
+        )
+        self.assertEqual(self.publisher.modes, ["video", "service"])
+
+    def test_retained_mode_set_does_not_claim_live_operator_intent(self):
+        router = self.coordinator()
+        router.observe_mode("service")
+        decision = router.acquire("play-retained-set")
+        router.observe_mode("video")
+        ignored = router.observe_mode_command("announcements", retained=True)
+        self.assertEqual(ignored.action, "retained_ignored")
+        self.assertEqual(
+            self.store.get_audio_lease(decision.lease_id).state,
+            "held",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_service.py

@@ -0,0 +1,742 @@
+import json
+import queue
+import tempfile
+import threading
+import time
+import unittest
+from dataclasses import replace
+from pathlib import Path
+
+from controller.config import (
+    ControllerConfig,
+    HomeAssistantConfig,
+    RouterConfig,
+    StorageConfig,
+)
+from controller.downloads import canonical_playlist_path
+from controller.mqtt_adapter import MqttEvent, PublishReceipt
+from controller.service import ControllerService
+from controller.store import ControllerStore
+
+
+class MutableClock:
+    def __init__(self, value):
+        self.value = value
+
+    def __call__(self):
+        return self.value
+
+
+class FakeMqttAdapter:
+    """In-memory transport boundary; it never starts a network client."""
+
+    def __init__(self, *, connected=True):
+        self.connected = connected
+        self.started = False
+        self.player_commands = []
+        self.router_modes = []
+        self.json_messages = []
+        self.legacy_messages = []
+        self.acknowledged = []
+
+    def start(self):
+        self.started = True
+
+    def stop(self):
+        self.started = False
+
+    def _receipt(self):
+        if self.connected:
+            return PublishReceipt(True, message_id=1)
+        return PublishReceipt(False, reason="MQTT is disconnected")
+
+    def publish_player_command(self, payload):
+        receipt = self._receipt()
+        if receipt.accepted:
+            self.player_commands.append(dict(payload))
+        return receipt
+
+    def publish_router_mode(self, mode):
+        receipt = self._receipt()
+        if receipt.accepted:
+            self.router_modes.append(mode)
+        return receipt
+
+    def publish_json(self, topic, payload, *, qos=0, retain=False):
+        receipt = self._receipt()
+        if receipt.accepted:
+            self.json_messages.append((topic, dict(payload), qos, retain))
+        return receipt
+
+    def publish_legacy(self, topic, payload, *, retain=True):
+        receipt = self._receipt()
+        if receipt.accepted:
+            self.legacy_messages.append((topic, payload, retain))
+        return receipt
+
+    def acknowledge(self, event):
+        self.acknowledged.append(event)
+        return True
+
+
+class FakeDownloadManager:
+    def __init__(self):
+        self.recovered = False
+        self.stopped = False
+
+    def recover_pending_deletions(self):
+        self.recovered = True
+
+    def expire_due_slots(self):
+        return []
+
+    def shutdown(self, *, wait):
+        self.stopped = wait
+
+
+class ControllerServiceTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        base = Path(self.tempdir.name)
+        self.wall = MutableClock(1_700_000_000.0)
+        self.monotonic = MutableClock(100.0)
+        self.config = ControllerConfig(
+            storage=StorageConfig(
+                media_base=base / "media",
+                data_base=base / "data",
+                retention_seconds=3600,
+                expiry_interval_seconds=3600,
+                uncertain_review_after_seconds=30,
+            ),
+            router=RouterConfig(
+                required_mode="video",
+                confirmation_timeout_seconds=3,
+                when_state_unknown="continue",
+                continue_on_confirmation_failure=True,
+                fallback_restore_mode=None,
+            ),
+            homeassistant=HomeAssistantConfig(enabled=False),
+        )
+        self.events = queue.Queue()
+        self.store = None
+        self.mqtt = None
+        self.downloads = None
+        self.service = None
+
+    def tearDown(self):
+        if self.service is not None and self.service.started:
+            self.service.stop()
+        self.tempdir.cleanup()
+
+    def start_service(self, *, config=None, mqtt=None):
+        config = config or self.config
+        self.events = queue.Queue()
+        self.store = ControllerStore(config.storage, now=self.wall)
+        self.mqtt = mqtt or FakeMqttAdapter()
+        self.downloads = FakeDownloadManager()
+        self.service = ControllerService(
+            config,
+            store=self.store,
+            mqtt_adapter=self.mqtt,
+            event_queue=self.events,
+            download_manager=self.downloads,
+            monotonic=self.monotonic,
+            wall_clock=self.wall,
+        )
+        self.service.start()
+        self.assertTrue(self.downloads.recovered)
+        return self.service
+
+    def ready_slot(self, slot=1):
+        reservation = self.store.reserve_download(
+            f"https://example.test/{slot}",
+            "tester",
+            "service test",
+            job_id=f"ready-{slot}",
+        )
+        self.assertEqual(reservation.slot, slot)
+        self.assertTrue(
+            self.store.mark_download_running(
+                reservation.job_id,
+                reservation.slot,
+                reservation.generation,
+            )
+        )
+        self.assertTrue(
+            self.store.complete_download(
+                reservation.job_id,
+                reservation.slot,
+                reservation.generation,
+                title=f"Slot {slot}",
+                items=(("One", "one.mp4"),),
+            )
+        )
+        playlist = canonical_playlist_path(self.service.config.storage.media_base, slot)
+        playlist.parent.mkdir(parents=True, exist_ok=True)
+        playlist.write_text("one.mp4\n", encoding="utf-8")
+        return playlist
+
+    def send_message(
+        self,
+        topic,
+        payload,
+        *,
+        qos=0,
+        retain=False,
+        message_id=None,
+        duplicate=False,
+        controller_echo=False,
+    ):
+        if not isinstance(payload, bytes):
+            payload = (
+                payload.encode("utf-8")
+                if isinstance(payload, str)
+                else json.dumps(payload).encode("utf-8")
+            )
+        self.events.put(
+            MqttEvent(
+                kind="message",
+                topic=topic,
+                payload=payload,
+                qos=qos,
+                retain=retain,
+                duplicate=duplicate,
+                message_id=message_id,
+                connection_generation=1,
+                controller_echo=controller_echo,
+            )
+        )
+
+    def wait_for(self, predicate, message):
+        deadline = time.monotonic() + 2.0
+        while time.monotonic() < deadline:
+            if predicate():
+                return
+            time.sleep(0.01)
+        self.fail(message)
+
+    def result_events(self):
+        return [
+            payload
+            for topic, payload, qos, retain in self.mqtt.json_messages
+            if topic == self.service.config.topics.controller_result
+        ]
+
+    def test_play_waits_for_router_confirmation_and_player_evidence(self):
+        config = replace(
+            self.config,
+            router=replace(self.config.router, when_state_unknown="attempt"),
+        )
+        self.start_service(config=config)
+        playlist = self.ready_slot(1)
+
+        self.send_message(config.topics.router_mode_state, "service", qos=1)
+        self.wait_for(
+            lambda: self.store.router_observation().mode == "service",
+            "Router state was not consumed",
+        )
+
+        result = self.service.submit_control({"action": "play", "slot": 1})
+        self.assertEqual(result["status"], "pending")
+        self.assertEqual(result["router"], "waiting")
+        self.assertEqual(self.mqtt.router_modes, ["video"])
+        self.assertEqual(self.mqtt.player_commands, [])
+        self.assertEqual(self.service.playback.projected_slot_state(1), "ready")
+
+        self.send_message(
+            config.topics.router_mode_set,
+            "video",
+            qos=1,
+            controller_echo=True,
+        )
+        self.wait_for(
+            lambda: len(self.mqtt.acknowledged) == 2,
+            "Controller Router command echo was not consumed",
+        )
+        self.assertEqual(self.store.live_audio_lease().state, "acquiring")
+        self.assertEqual(self.mqtt.player_commands, [])
+
+        self.send_message(config.topics.router_mode_state, "video", qos=1)
+        self.wait_for(
+            lambda: len(self.mqtt.player_commands) == 1,
+            "Player command was not released after Router confirmation",
+        )
+        self.assertEqual(
+            self.mqtt.player_commands,
+            [{"action": "play", "slot": 1, "label": "Slot 1"}],
+        )
+        self.assertEqual(self.service.playback.projected_slot_state(1), "ready")
+
+        self.send_message(
+            config.topics.player_status,
+            {"state": "playing", "slot": 1, "path": str(playlist)},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: self.store.player_observation().phase == "playing",
+            "authoritative Player status was not reduced",
+        )
+        self.assertEqual(self.service.playback.projected_slot_state(1), "playing")
+        self.assertIn(
+            "player_operation_confirmed",
+            {event["type"] for event in self.result_events()},
+        )
+
+    def test_manual_router_set_during_acquisition_resolves_by_failure_policy(self):
+        config = replace(
+            self.config,
+            router=replace(self.config.router, when_state_unknown="attempt"),
+        )
+        self.start_service(config=config)
+        self.ready_slot(1)
+        self.send_message(config.topics.router_mode_state, "service", qos=1)
+        self.wait_for(
+            lambda: self.store.router_observation().mode == "service",
+            "Router state was not consumed",
+        )
+
+        result = self.service.submit_control({"action": "play", "slot": 1})
+        self.assertEqual(result["router"], "waiting")
+        self.send_message(config.topics.router_mode_set, "announcements", qos=1)
+        self.wait_for(
+            lambda: len(self.mqtt.player_commands) == 1,
+            "manual Router intent did not resolve acquisition",
+        )
+        lease = self.store.get_audio_lease(self.service._session_lease_id)
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease.state, "unconfirmed")
+        warnings = [
+            event
+            for event in self.result_events()
+            if event["type"] == "audio_router_warning"
+        ]
+        self.assertTrue(warnings)
+        self.assertIs(warnings[-1]["continued"], True)
+
+    def test_pause_state_remains_telemetry_authoritative(self):
+        self.start_service()
+        playlist = self.ready_slot(1)
+        self.service.submit_control({"action": "play", "slot": 1})
+        self.wait_for(
+            lambda: self.mqtt.player_commands == [
+                {"action": "play", "slot": 1, "label": "Slot 1"}
+            ],
+            "play command was not published",
+        )
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "playing", "slot": 1, "path": str(playlist)},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: self.service.playback.pending is None,
+            "play operation was not confirmed",
+        )
+        playing_generation = self.store.player_observation().generation
+
+        self.service.submit_control({"action": "pause"})
+        self.send_message(
+            self.config.topics.player_status,
+            {"state": "paused", "slot": 1, "path": str(playlist)},
+            qos=1,
+        )
+        self.wait_for(
+            lambda: self.store.player_observation().generation > playing_generation,
+            "pause status was not consumed",
+        )
+        self.assertIsNotNone(self.service.playback.pending)
+        self.assertEqual(self.service.playback.projected_slot_state(1), "playing")
+
+        self.send_message(
+            self.config.topics.player_telemetry,
+            {
+                "state": "telemetry",
+                "slot": 1,
+                "path": str(playlist),
+                "paused": True,
+            },
+        )
+        self.wait_for(
+            lambda: self.service.playback.pending is None,
+            "pause was not confirmed by telemetry",
+        )
+        self.assertEqual(self.store.player_observation().paused, True)
+        self.assertEqual(self.service.playback.projected_slot_state(1), "paused")
+        self.assertEqual(self.mqtt.player_commands[-1], {"action": "pause"})
+
+    def test_retained_controller_command_is_rejected_then_acked_and_reported(self):
+        self.start_service()
+        self.send_message(
+            self.config.topics.controller_command,
+            {"action": "play", "path": "/must/not/run"},
+            qos=1,
+            retain=True,
+            message_id=42,
+        )
+        self.wait_for(
+            lambda: len(self.mqtt.acknowledged) == 1,
+            "retained command was not acknowledged after rejection",
+        )
+        self.assertEqual(self.mqtt.player_commands, [])
+        event = self.result_events()[-1]
+        self.assertEqual(event["version"], 2)
+        self.assertEqual(event["type"], "retained_command_rejected")
+        topic, _, qos, retain = self.mqtt.json_messages[-1]
+        self.assertEqual(topic, "youtube/controller/v2/events")
+        self.assertEqual((qos, retain), (1, False))
+        self.assertEqual(self.mqtt.acknowledged[0].message_id, 42)
+
+    def test_optional_result_topic_can_be_disabled_without_affecting_ack(self):
+        config = replace(
+            self.config,
+            topics=replace(self.config.topics, controller_result=None),
+        )
+        self.start_service(config=config)
+        self.send_message(
+            config.topics.controller_command,
+            {"action": "delete_slot", "slot": 1},
+            qos=1,
+            retain=True,
+            message_id=7,
+        )
+        self.wait_for(
+            lambda: len(self.mqtt.acknowledged) == 1,
+            "command was not acknowledged with result topic disabled",
+        )
+        self.assertEqual(self.mqtt.json_messages, [])
+
+    def test_broker_unavailable_does_not_prevent_service_startup(self):
+        mqtt = FakeMqttAdapter(connected=False)
+        self.start_service(mqtt=mqtt)
+        health = self.service.health()
+        self.assertTrue(self.service.started)
+        self.assertTrue(mqtt.started)
+        self.assertTrue(health["coordinator_alive"])
+        self.assertEqual(health["status"], "degraded")
+        self.assertIn("mqtt_disconnected", health["degraded_reasons"])
+
+    def test_administrator_can_clear_exact_uncertain_playback_review(self):
+        self.start_service()
+        playlist = self.ready_slot(1)
+        self.store.set_player_observation(
+            availability="unknown",
+            phase="unknown",
+            active_slot=1,
+            active_path=str(playlist),
+            paused=None,
+            reconciliation_state="operator_review",
+            reconciliation_due_at=None,
+            reconciliation_id="review-token",
+        )
+        guard = self.store.protect_slot(
+            1,
+            "restart_uncertain",
+            guard_id="review-token",
+        )
+        review = self.service.health()["reconciliation_reviews"]
+        self.assertEqual(review[0]["review_id"], "review-token")
+
+        result = self.service.submit_control(
+            {
+                "action": "reconcile_playback",
+                "decision": "clear",
+                "confirm": True,
+                "slot": 1,
+                "review_id": review[0]["review_id"],
+                "actor": "operator",
+                "reason": "Player host was replaced",
+            }
+        )
+        self.assertEqual(
+            result,
+            {"status": "cleared", "slot": 1, "review_id": "review-token"},
+        )
+        self.assertEqual(self.store.list_live_guards(), [])
+        observation = self.store.player_observation()
+        self.assertEqual(observation.reconciliation_state, "admin_cleared")
+        self.assertIsNone(observation.active_slot)
+        self.assertEqual(self.service.playback.projected_slot_state(1), "ready")
+        self.assertEqual(self.result_events()[-1]["type"], "playback_reconciled")
+
+    def test_busy_player_keeps_video_route_until_actual_playback_finishes(self):
+        config = replace(
+            self.config,
+            router=replace(self.config.router, when_state_unknown="attempt"),
+        )
+        self.start_service(config=config)
+        self.ready_slot(1)
+        second_playlist = self.ready_slot(2)
+        self.send_message(config.topics.router_mode_state, "service", qos=1)
+        self.wait_for(
+            lambda: self.store.router_observation().mode == "service",
+            "Router service mode was not observed",
+        )
+        self.service.submit_control({"action": "play", "slot": 1})
+        self.send_message(config.topics.router_mode_state, "video", qos=1)
+        self.wait_for(
+            lambda: len(self.mqtt.player_commands) == 1,
+            "play command was not released",
+        )
+        self.send_message(
+            config.topics.player_status,
+            {
+                "state": "busy",
+                "slot": 2,
+                "path": str(second_playlist),
+                "message": "mpv already playing; stop first",
+            },
+            qos=1,
+        )
+        self.wait_for(
+            lambda: self.service.playback.pending is None,
+            "busy response was not reduced",
+        )
+        self.assertEqual(self.store.player_observation().active_slot, 2)
+        self.assertEqual(self.mqtt.router_modes, ["video"])
+        lease = self.store.live_audio_lease()
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease.state, "held")
+
+    def test_disconnected_synchronous_play_failure_is_not_reported_pending(self):
+        self.start_service(mqtt=FakeMqttAdapter(connected=False))
+        self.ready_slot(1)
+        result = self.service.submit_control({"action": "play", "slot": 1})
+        self.assertEqual(result["status"], "unavailable")
+        self.assertIsNone(self.service.playback.pending)
+        self.assertEqual(self.store.list_live_guards(), [])
+
+    def test_router_timeout_uses_monotonic_clock_and_continues_with_warning(self):
+        config = replace(
+            self.config,
+            router=replace(self.config.router, when_state_unknown="attempt"),
+        )
+        self.start_service(config=config)
+        self.ready_slot(1)
+        result = self.service.submit_control({"action": "play", "slot": 1})
+        self.assertEqual(result["router"], "waiting")
+        self.wall.value -= 10_000
+        self.monotonic.value += 4
+        self.wait_for(
+            lambda: len(self.mqtt.player_commands) == 1,
+            "Router timeout did not release play within its finite deadline",
+        )
+        warnings = [
+            event
+            for event in self.result_events()
+            if event["type"] == "audio_router_warning"
+        ]
+        self.assertTrue(warnings)
+        self.assertIs(warnings[-1]["continued"], True)
+
+    def test_duplicate_toggle_is_suppressed_and_translated_idempotently(self):
+        self.start_service()
+        playlist = self.ready_slot(1)
+        self.send_message(
+            self.config.topics.player_telemetry,
+            {
+                "state": "telemetry",
+                "slot": 1,
+                "path": str(playlist),
+                "paused": False,
+            },
+        )
+        self.wait_for(
+            lambda: self.store.player_observation().active_slot == 1,
+            "active telemetry was not consumed",
+        )
+        command = {"action": "toggle"}
+        self.send_message(
+            self.config.topics.controller_command,
+            command,
+            qos=1,
+            message_id=71,
+        )
+        self.send_message(
+            self.config.topics.controller_command,
+            command,
+            qos=1,
+            message_id=71,
+            duplicate=True,
+        )
+        self.wait_for(
+            lambda: sum(
+                event.message_id == 71 for event in self.mqtt.acknowledged
+            )
+            == 2,
+            "duplicate command packets were not both acknowledged",
+        )
+        self.assertEqual(self.mqtt.player_commands, [{"action": "pause"}])
+
+    def test_path_only_managed_play_adds_stable_slot_identity(self):
+        self.start_service()
+        playlist = self.ready_slot(1)
+        result = self.service.submit_control(
+            {"action": "play", "playlist": str(playlist)}
+        )
+        self.assertEqual(result["status"], "pending")
+        self.assertEqual(
+            self.mqtt.player_commands,
+            [{"action": "play", "playlist": str(playlist), "slot": 1}],
+        )
+        self.assertEqual(self.service.playback.projected_slot_state(1), "ready")
+
+    def test_explicit_slot_cannot_be_paired_with_another_managed_slot_path(self):
+        self.start_service()
+        self.ready_slot(1)
+        second_playlist = self.ready_slot(2)
+        with self.assertRaisesRegex(ValueError, "conflicts"):
+            self.service.submit_control(
+                {
+                    "action": "play",
+                    "slot": 1,
+                    "playlist": str(second_playlist),
+                }
+            )
+        self.assertEqual(self.mqtt.player_commands, [])
+        self.assertEqual(self.store.list_live_guards(), [])
+
+    def test_administrator_can_clear_external_playback_review_without_slot(self):
+        config = replace(
+            self.config,
+            topics=replace(self.config.topics, controller_result=None),
+        )
+        self.start_service(config=config)
+        observation = self.store.mark_external_uncertain(
+            "/srv/operator-playlists/special-event.m3u",
+            review_after=self.wall.value,
+            details={"reason": "restart"},
+        )
+        self.assertIsNotNone(observation.reconciliation_id)
+        reviews = self.service.health()["reconciliation_reviews"]
+        self.assertEqual(len(reviews), 1)
+        review_id = reviews[0]["review_id"]
+
+        result = self.service.submit_control(
+            {
+                "action": "reconcile_playback",
+                "decision": "clear",
+                "confirm": True,
+                "review_id": review_id,
+                "actor": "operator",
+                "reason": "Player appliance was replaced",
+            }
+        )
+        self.assertEqual(
+            result,
+            {"status": "cleared", "review_id": review_id},
+        )
+        cleared = self.store.player_observation()
+        self.assertIsNone(cleared.active_path)
+        self.assertEqual(cleared.reconciliation_state, "admin_cleared")
+
+    def test_external_play_intent_is_durable_immediately_after_puback(self):
+        self.start_service()
+        playlist = Path(self.tempdir.name) / "operator-playlist.m3u"
+        playlist.write_text("/srv/media/opening.mp4\n", encoding="utf-8")
+        result = self.service.submit_control(
+            {"action": "play", "playlist": str(playlist)}
+        )
+        self.assertEqual(result["status"], "pending")
+        self.assertEqual(
+            self.mqtt.player_commands,
+            [{"action": "play", "playlist": str(playlist)}],
+        )
+        reviews = self.service.health()["reconciliation_reviews"]
+        self.assertEqual(len(reviews), 1)
+        review_id = reviews[0]["review_id"]
+        self.assertIs(reviews[0]["clearable"], False)
+
+        self.service.stop()
+        self.service = None
+        with ControllerStore(self.config.storage, now=self.wall).open() as reopened:
+            reopened.mark_restart_uncertain()
+            observation = reopened.player_observation()
+            self.assertEqual(observation.active_path, str(playlist))
+            self.assertEqual(observation.reconciliation_id, review_id)
+
+    def test_live_external_play_fence_cannot_be_admin_cleared(self):
+        self.start_service()
+        playlist = Path(self.tempdir.name) / "operator-playlist.m3u"
+        playlist.write_text("/srv/media/opening.mp4\n", encoding="utf-8")
+        self.service.submit_control(
+            {"action": "play", "playlist": str(playlist)}
+        )
+        review = self.service.health()["reconciliation_reviews"][0]
+        self.assertIs(review["clearable"], False)
+
+        with self.assertRaisesRegex(ValueError, "Player operation is pending"):
+            self.service.submit_control(
+                {
+                    "action": "reconcile_playback",
+                    "decision": "clear",
+                    "confirm": True,
+                    "review_id": review["review_id"],
+                    "actor": "operator",
+                    "reason": "clear requested too early",
+                }
+            )
+        self.assertIsNotNone(self.service.playback.pending)
+        observation = self.store.player_observation()
+        self.assertEqual(observation.reconciliation_id, review["review_id"])
+        self.assertEqual(observation.active_path, str(playlist))
+
+    def test_definite_external_publish_failure_clears_prepublication_guard(self):
+        self.start_service(mqtt=FakeMqttAdapter(connected=False))
+        playlist = Path(self.tempdir.name) / "operator-playlist.m3u"
+        playlist.write_text("/srv/media/opening.mp4\n", encoding="utf-8")
+        result = self.service.submit_control(
+            {"action": "play", "playlist": str(playlist)}
+        )
+        self.assertEqual(result["status"], "unavailable")
+        self.assertEqual(self.service.health()["reconciliation_reviews"], [])
+        observation = self.store.player_observation()
+        self.assertIsNone(observation.active_path)
+        self.assertIsNone(observation.reconciliation_id)
+
+    def test_unexpected_reducer_failure_is_not_acked_or_deduplicated(self):
+        self.start_service()
+        original = self.service.playback.handle_status
+        failure_seen = threading.Event()
+        attempts = 0
+
+        def fail_once(payload):
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                failure_seen.set()
+                raise RuntimeError("transient reducer failure")
+            return original(payload)
+
+        self.service.playback.handle_status = fail_once
+        payload = {
+            "state": "playing",
+            "slot": 1,
+            "path": str(self.config.storage.media_base / "slot1" / "playlist.m3u"),
+        }
+        self.send_message(
+            self.config.topics.player_status,
+            payload,
+            qos=1,
+            message_id=88,
+        )
+        self.assertTrue(failure_seen.wait(timeout=1))
+        time.sleep(0.05)
+        self.assertEqual(self.mqtt.acknowledged, [])
+
+        self.send_message(
+            self.config.topics.player_status,
+            payload,
+            qos=1,
+            message_id=88,
+            duplicate=True,
+        )
+        self.wait_for(
+            lambda: len(self.mqtt.acknowledged) == 1,
+            "retried status was not processed and acknowledged",
+        )
+        self.assertEqual(attempts, 2)
+        self.assertEqual(self.store.player_observation().phase, "playing")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_store.py

@@ -1,6 +1,7 @@
 import concurrent.futures
 import hashlib
 import os
+import sqlite3
 import tempfile
 import unittest
 from pathlib import Path
@@ -49,6 +50,37 @@ class FreshStoreTests(StoreTestCase):
             self.assertTrue(all(slot.media_state == "empty" for slot in slots))
             self.assertEqual(store.player_observation().phase, "unknown")
 
+    def test_schema_one_database_upgrades_external_reconciliation_atomically(self):
+        with self.open_store() as store:
+            database_path = store.path
+        connection = sqlite3.connect(database_path)
+        try:
+            connection.execute(
+                "ALTER TABLE player_observation DROP COLUMN reconciliation_id"
+            )
+            connection.execute("DELETE FROM schema_migrations WHERE version = 2")
+            connection.execute("PRAGMA user_version = 1")
+            connection.commit()
+        finally:
+            connection.close()
+
+        with self.open_store() as upgraded:
+            self.assertEqual(
+                upgraded.connection.execute("PRAGMA user_version").fetchone()[0],
+                2,
+            )
+            columns = {
+                row["name"]
+                for row in upgraded.connection.execute(
+                    "PRAGMA table_info(player_observation)"
+                ).fetchall()
+            }
+            self.assertIn("reconciliation_id", columns)
+            migration = upgraded.connection.execute(
+                "SELECT name FROM schema_migrations WHERE version = 2"
+            ).fetchone()
+            self.assertEqual(migration["name"], "external_playback_reconciliation")
+
     def test_slot_reservation_is_committed_before_worker_transition(self):
         with self.open_store() as store:
             reservation = store.reserve_download(
@@ -127,6 +159,25 @@ class FreshStoreTests(StoreTestCase):
             self.assertEqual(store.get_slot(2).slot, 2)
             self.assertEqual(store.get_slot(2).url, "https://example.test/two")
 
+    def test_concurrent_first_open_installs_only_one_valid_database(self):
+        stores = [ControllerStore(self.config, now=self.clock) for _ in range(2)]
+
+        def open_store(store):
+            store.open()
+            return [slot.slot for slot in store.list_slots()]
+
+        try:
+            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+                results = list(pool.map(open_store, stores))
+            self.assertEqual(results, [list(range(1, 11)), list(range(1, 11))])
+            markers = stores[0].connection.execute(
+                "SELECT name FROM data_migrations ORDER BY name"
+            ).fetchall()
+            self.assertEqual([row["name"] for row in markers], ["fresh_database_v1"])
+        finally:
+            for store in stores:
+                store.close()
+
 
 class GuardAndExpiryTests(StoreTestCase):
     def _ready_slot(self, store, *, slot_expected=1):
@@ -196,6 +247,7 @@ class GuardAndExpiryTests(StoreTestCase):
                 paused=False,
                 reconciliation_state="confirmed",
                 reconciliation_due_at=None,
+                reconciliation_id=None,
             )
             guard = store.mark_restart_uncertain()
             self.assertEqual(guard.slot, 1)
@@ -204,6 +256,150 @@ class GuardAndExpiryTests(StoreTestCase):
             self.assertEqual(observation.reconciliation_state, "awaiting_evidence")
             self.assertEqual(observation.active_slot, 1)
 
+    def test_external_playback_restart_gets_stable_review_token_and_exact_clear(self):
+        external_path = "/srv/operator-playlists/live.m3u"
+        with self.open_store() as store:
+            store.set_player_observation(
+                availability="online",
+                phase="playing",
+                active_slot=None,
+                active_path=external_path,
+                paused=False,
+                reconciliation_state="confirmed",
+                reconciliation_due_at=None,
+                reconciliation_id=None,
+            )
+            self.assertIsNone(store.mark_restart_uncertain())
+            first = store.player_observation()
+            self.assertEqual(first.reconciliation_state, "awaiting_evidence")
+            self.assertIsNotNone(first.reconciliation_id)
+            self.assertEqual(first.reconciliation_due_at, self.clock.value + 30)
+
+            self.clock.value += 10
+            self.assertIsNone(store.mark_restart_uncertain())
+            repeated = store.player_observation()
+            self.assertEqual(repeated.reconciliation_id, first.reconciliation_id)
+            self.assertEqual(repeated.reconciliation_due_at, first.reconciliation_due_at)
+
+            self.clock.value = first.reconciliation_due_at + 1
+            due = store.mark_external_reconciliation_due()
+            self.assertEqual(due, (first.reconciliation_id, external_path))
+            self.assertIsNone(store.mark_external_reconciliation_due())
+            review = store.player_observation()
+            self.assertEqual(review.reconciliation_state, "operator_review")
+            self.assertEqual(review.active_path, external_path)
+
+            self.assertFalse(
+                store.clear_external_reconciliation(
+                    "stale-review",
+                    actor="operator",
+                    reason="Player was replaced",
+                )
+            )
+            self.assertTrue(
+                store.clear_external_reconciliation(
+                    first.reconciliation_id,
+                    actor="operator",
+                    reason="Player was replaced",
+                )
+            )
+            cleared = store.player_observation()
+            self.assertEqual(cleared.reconciliation_state, "admin_cleared")
+            self.assertIsNone(cleared.active_path)
+            self.assertIsNone(cleared.reconciliation_id)
+
+    def test_restart_converts_existing_active_guard_and_preserves_deadline(self):
+        with self.open_store() as store:
+            self._ready_slot(store)
+            store.protect_slot(1, "player_active", guard_id="active-before-restart")
+            store.set_player_observation(
+                availability="online",
+                phase="playing",
+                active_slot=1,
+                active_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
+                paused=False,
+                reconciliation_state="confirmed",
+                reconciliation_due_at=None,
+                reconciliation_id=None,
+            )
+            guard = store.mark_restart_uncertain()
+            self.assertEqual(guard.guard_id, "active-before-restart")
+            self.assertEqual(guard.kind, "restart_uncertain")
+            first_deadline = guard.review_after
+            self.clock.value += 10
+            repeated = store.mark_restart_uncertain()
+            self.assertEqual(repeated.guard_id, guard.guard_id)
+            self.assertEqual(repeated.review_after, first_deadline)
+
+    def test_restart_converts_pre_evidence_pending_guard_for_admin_review(self):
+        with self.open_store() as store:
+            self._ready_slot(store)
+            store.protect_slot(1, "pending_play", guard_id="published-before-crash")
+            guard = store.mark_restart_uncertain()
+            self.assertEqual(guard.guard_id, "published-before-crash")
+            self.assertEqual(guard.kind, "restart_uncertain")
+            self.assertEqual(store.player_observation().active_slot, 1)
+            self.assertTrue(
+                store.clear_guard(
+                    1,
+                    guard.guard_id,
+                    actor="operator",
+                    reason="Player replaced",
+                )
+            )
+
+    def test_guard_review_transition_is_reported_only_once(self):
+        with self.open_store() as store:
+            self._ready_slot(store)
+            store.protect_slot(
+                1,
+                "restart_uncertain",
+                guard_id="one-warning",
+                review_after=self.clock.value,
+            )
+            self.assertEqual(len(store.mark_due_guards_for_review()), 1)
+            self.assertEqual(store.mark_due_guards_for_review(), [])
+
+    def test_clearing_one_legacy_guard_does_not_erase_other_observation(self):
+        with self.open_store() as store:
+            self._ready_slot(store)
+            second = store.reserve_download(
+                "https://example.test/second", None, None, job_id="second"
+            )
+            store.mark_download_running("second", 2, second.generation)
+            store.complete_download(
+                "second",
+                2,
+                second.generation,
+                title="Second",
+                items=(("Second", "second.mp4"),),
+            )
+            first_guard = store.protect_slot(
+                1, "legacy_uncertain", guard_id="legacy-one"
+            )
+            store.protect_slot(2, "player_active", guard_id="active-two")
+            store.set_player_observation(
+                availability="online",
+                phase="playing",
+                active_slot=2,
+                active_path="/mnt/Media/YTPlayer/slot2/playlist.m3u",
+                paused=False,
+                reconciliation_state="confirmed",
+                reconciliation_due_at=None,
+                reconciliation_id=None,
+            )
+            self.assertTrue(
+                store.clear_guard(
+                    1,
+                    first_guard.guard_id,
+                    actor="operator",
+                    reason="legacy evidence resolved",
+                )
+            )
+            observation = store.player_observation()
+            self.assertEqual(observation.active_slot, 2)
+            self.assertEqual(observation.phase, "playing")
+
 
 class LegacyMigrationTests(StoreTestCase):
     def write_legacy(self, data):