feat: reduce authoritative player mqtt state

controller/mqtt_adapter.py

@@ -0,0 +1,353 @@
+"""Resilient MQTT transport with explicit Player command guarantees."""
+
+from __future__ import annotations
+
+import json
+import logging
+import queue
+import threading
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+import paho.mqtt.client as mqtt
+
+from .config import ControllerConfig
+from .contracts import (
+    PLAYER_COMMAND_QOS,
+    PLAYER_COMMAND_RETAIN,
+    PLAYER_STATE_QOS,
+    PLAYER_STATUS_QOS,
+    PLAYER_TELEMETRY_QOS,
+    ROUTER_COMMAND_QOS,
+    ROUTER_COMMAND_RETAIN,
+)
+
+
+LOG = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class MqttEvent:
+    kind: str
+    topic: str | None = None
+    payload: bytes = b""
+    qos: int = 0
+    retain: bool = False
+    message_id: int | None = None
+    connection_generation: int | None = None
+    reason: str | None = None
+
+
+@dataclass(frozen=True)
+class PublishReceipt:
+    accepted: bool
+    message_id: int | None = None
+    reason: str | None = None
+
+
+class MqttAdapter:
+    """Own Paho lifecycle while keeping all domain work off its callbacks."""
+
+    def __init__(
+        self,
+        config: ControllerConfig,
+        *,
+        client: Any | None = None,
+        events: "queue.Queue[MqttEvent] | None" = None,
+    ):
+        self.config = config
+        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._generation_lock = threading.Lock()
+        self._connection_generation = 0
+        self._connected = threading.Event()
+        self._stop_event = threading.Event()
+        self._lifecycle_lock = threading.Lock()
+        self._started = False
+        self._loop_started = False
+        self._bootstrap_thread: threading.Thread | None = None
+        self.last_error: str | None = None
+        self._configure_client()
+
+    def _configure_client(self) -> None:
+        if self.config.mqtt.username is not None:
+            self.client.username_pw_set(
+                self.config.mqtt.username,
+                self.config.mqtt.password,
+            )
+        self.client.on_connect = self._on_connect
+        self.client.on_connect_fail = self._on_connect_fail
+        self.client.on_disconnect = self._on_disconnect
+        self.client.on_message = self._on_message
+        self.client.suppress_exceptions = True
+        self.client.manual_ack_set(True)
+        self.client.max_queued_messages_set(self.config.mqtt.command_queue_size)
+        self.client.reconnect_delay_set(
+            min_delay=self.config.mqtt.reconnect_min_seconds,
+            max_delay=self.config.mqtt.reconnect_max_seconds,
+        )
+
+    @property
+    def connected(self) -> bool:
+        return self._connected.is_set()
+
+    @property
+    def connection_generation(self) -> int:
+        with self._generation_lock:
+            return self._connection_generation
+
+    @property
+    def started(self) -> bool:
+        with self._lifecycle_lock:
+            return self._started
+
+    def start(self) -> None:
+        """Start without requiring the broker or network to be reachable."""
+        with self._lifecycle_lock:
+            if self._started:
+                return
+            self._started = True
+            self._stop_event.clear()
+        try:
+            self._begin_async_connection()
+        except (OSError, mqtt.WebsocketConnectionError) 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(
+            self.config.mqtt.host,
+            self.config.mqtt.port,
+            keepalive=self.config.mqtt.keepalive_seconds,
+        )
+        with self._lifecycle_lock:
+            if not self._loop_started:
+                result = self.client.loop_start()
+                if result != mqtt.MQTT_ERR_SUCCESS:
+                    raise RuntimeError(f"failed to start MQTT network loop: {result}")
+                self._loop_started = True
+
+    def _start_bootstrap_retry(self) -> None:
+        with self._lifecycle_lock:
+            if self._bootstrap_thread is not None and self._bootstrap_thread.is_alive():
+                return
+            self._bootstrap_thread = threading.Thread(
+                target=self._bootstrap_retry,
+                name="mqtt-bootstrap-retry",
+                daemon=True,
+            )
+            self._bootstrap_thread.start()
+
+    def _bootstrap_retry(self) -> None:
+        delay = float(self.config.mqtt.reconnect_min_seconds)
+        maximum = float(self.config.mqtt.reconnect_max_seconds)
+        while not self._stop_event.wait(delay):
+            try:
+                self._begin_async_connection()
+                self.last_error = None
+                return
+            except (OSError, mqtt.WebsocketConnectionError) as exc:
+                self.last_error = f"MQTT setup retry failed: {exc}"
+                LOG.warning("%s", self.last_error)
+                delay = min(maximum, max(delay * 2, 1.0))
+            except RuntimeError as exc:
+                self.last_error = str(exc)
+                LOG.error("%s", self.last_error)
+                return
+
+    def stop(self) -> None:
+        with self._lifecycle_lock:
+            if not self._started:
+                return
+            self._started = False
+            loop_started = self._loop_started
+            self._loop_started = False
+        self._stop_event.set()
+        self._connected.clear()
+        if loop_started:
+            try:
+                self.client.disconnect()
+            except Exception:
+                LOG.exception("MQTT disconnect failed")
+            self.client.loop_stop()
+        thread = self._bootstrap_thread
+        if thread is not None and thread is not threading.current_thread():
+            thread.join(timeout=1.0)
+
+    @staticmethod
+    def _reason_success(reason_code: Any) -> bool:
+        try:
+            return int(reason_code) == 0
+        except (TypeError, ValueError):
+            return reason_code == 0
+
+    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)
+                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",
+                    connection_generation=generation,
+                )
+            )
+        except BaseException:
+            LOG.exception("unexpected MQTT on_connect failure")
+
+    def _on_connect_fail(self, client, userdata) -> None:
+        self.last_error = (
+            f"MQTT connection to {self.config.mqtt.host}:{self.config.mqtt.port} failed"
+        )
+        LOG.warning("%s; Paho will retry", self.last_error)
+
+    def _on_disconnect(
+        self,
+        client,
+        userdata,
+        disconnect_flags,
+        reason_code,
+        properties=None,
+    ) -> None:
+        self._connected.clear()
+        try:
+            self._enqueue(
+                MqttEvent(
+                    kind="disconnected",
+                    connection_generation=self.connection_generation,
+                    reason=str(reason_code),
+                )
+            )
+        except BaseException:
+            LOG.exception("unexpected MQTT on_disconnect failure")
+
+    def _on_message(self, client, userdata, message) -> None:
+        try:
+            self._enqueue(
+                MqttEvent(
+                    kind="message",
+                    topic=str(message.topic),
+                    payload=bytes(message.payload),
+                    qos=int(getattr(message, "qos", 0)),
+                    retain=bool(getattr(message, "retain", False)),
+                    message_id=getattr(message, "mid", None),
+                    connection_generation=self.connection_generation,
+                )
+            )
+        except BaseException:
+            LOG.exception("unexpected MQTT on_message failure")
+
+    def _enqueue(self, event: MqttEvent) -> bool:
+        while not self._stop_event.is_set():
+            try:
+                self.events.put(event, timeout=0.2)
+                return True
+            except queue.Full:
+                continue
+        return False
+
+    def acknowledge(self, event: MqttEvent) -> bool:
+        if event.message_id is None or event.qos <= 0:
+            return True
+        if event.connection_generation != self.connection_generation:
+            LOG.warning(
+                "skipping stale MQTT ack mid=%s from generation=%s",
+                event.message_id,
+                event.connection_generation,
+            )
+            return False
+        try:
+            result = self.client.ack(event.message_id, event.qos)
+        except BaseException:
+            LOG.exception("MQTT ack failed mid=%s", event.message_id)
+            return False
+        return result == mqtt.MQTT_ERR_SUCCESS
+
+    def _publish(
+        self,
+        topic: str,
+        payload: str,
+        *,
+        qos: int,
+        retain: bool,
+        require_connection: bool,
+    ) -> PublishReceipt:
+        if require_connection and not self.connected:
+            return PublishReceipt(False, reason="MQTT is disconnected")
+        try:
+            info = self.client.publish(topic, payload, qos=qos, retain=retain)
+        except BaseException as exc:
+            LOG.exception("MQTT publish failed topic=%s", topic)
+            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}")
+        return PublishReceipt(True, message_id=getattr(info, "mid", None))
+
+    @staticmethod
+    def _json_payload(payload: Mapping[str, Any]) -> str:
+        return json.dumps(payload, allow_nan=False, separators=(",", ":"))
+
+    def publish_player_command(self, payload: Mapping[str, Any]) -> PublishReceipt:
+        return self._publish(
+            self.config.topics.player_command,
+            self._json_payload(payload),
+            qos=PLAYER_COMMAND_QOS,
+            retain=PLAYER_COMMAND_RETAIN,
+            require_connection=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,
+        )
+
+    def publish_legacy(self, topic: str, payload: str, *, retain: bool = True) -> PublishReceipt:
+        return self._publish(
+            topic,
+            payload,
+            qos=0,
+            retain=retain,
+            require_connection=True,
+        )
+
+    def publish_json(
+        self,
+        topic: str,
+        payload: Mapping[str, Any],
+        *,
+        qos: int = 0,
+        retain: bool = False,
+    ) -> PublishReceipt:
+        return self._publish(
+            topic,
+            self._json_payload(payload),
+            qos=qos,
+            retain=retain,
+            require_connection=True,
+        )

controller/playback.py

@@ -0,0 +1,495 @@
+"""Authoritative Player status/telemetry reduction and operation fencing."""
+
+from __future__ import annotations
+
+import math
+import time
+import uuid
+from dataclasses import dataclass, replace
+from typing import Any, Mapping
+
+from .config import ControllerConfig
+from .contracts import SLOT_IDS
+from .models import PlayerObservation
+from .store import ControllerStore
+
+
+TERMINAL_REASONS = frozenset(
+    {"stop", "finished", "mpv_error", "sigterm", "sigint", "shutdown"}
+)
+CONTROL_ACTIONS = frozenset({"play", "stop", "pause", "resume", "toggle"})
+
+
+@dataclass(frozen=True)
+class PlayerOperation:
+    operation_id: str
+    generation: int
+    action: str
+    slot: int | None
+    expected_path: str | None
+    deadline: float
+    state: str = "awaiting_player"
+    message: str | None = None
+
+
+@dataclass(frozen=True)
+class PlaybackEffect:
+    kind: str
+    slot: int | None = None
+    operation_id: str | None = None
+    message: str | None = None
+
+
+class OperationInProgressError(RuntimeError):
+    pass
+
+
+class PlaybackReducer:
+    """Apply Player evidence without projecting Controller intent as success."""
+
+    def __init__(
+        self,
+        config: ControllerConfig,
+        store: ControllerStore,
+        *,
+        monotonic=time.monotonic,
+        wall_clock=time.time,
+        operation_timeout_seconds: float = 10.0,
+    ):
+        self.config = config
+        self.store = store
+        self._monotonic = monotonic
+        self._wall_clock = wall_clock
+        self.operation_timeout_seconds = operation_timeout_seconds
+        self._operation_generation = 0
+        self.pending: PlayerOperation | None = None
+        self.last_operation: PlayerOperation | None = None
+
+    def begin_operation(
+        self,
+        action: str,
+        *,
+        slot: int | None = None,
+        expected_path: str | None = None,
+    ) -> PlayerOperation:
+        if action not in CONTROL_ACTIONS:
+            raise ValueError(f"unsupported Player action {action!r}")
+        if isinstance(slot, bool) or (slot is not None and slot not in SLOT_IDS):
+            raise ValueError("slot must be in 1..10")
+        if self.pending is not None:
+            raise OperationInProgressError(
+                f"Player operation {self.pending.operation_id} is still pending"
+            )
+        self._operation_generation += 1
+        operation = PlayerOperation(
+            operation_id=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,
+                "pending_play",
+                guard_id=f"operation-{operation.operation_id}",
+                details={"operation_id": operation.operation_id},
+            )
+        return operation
+
+    def publish_failed(self, reason: str) -> PlaybackEffect | None:
+        if self.pending is None:
+            return None
+        return self._finish_operation("failed", reason)
+
+    def _finish_operation(self, state: str, message: str | None = None) -> PlaybackEffect:
+        assert self.pending is not None
+        operation = replace(self.pending, state=state, message=message)
+        self.pending = None
+        self.last_operation = operation
+        if operation.action == "play" and operation.slot is not None and state != "confirmed":
+            self.store.clear_playback_guards(
+                slot=operation.slot,
+                kind="pending_play",
+                reason=f"play operation {state}: {message or ''}".strip(),
+            )
+        return PlaybackEffect(
+            kind=f"operation_{state}",
+            slot=operation.slot,
+            operation_id=operation.operation_id,
+            message=message,
+        )
+
+    @staticmethod
+    def _slot(value: Any) -> int | None:
+        if value is None or isinstance(value, bool):
+            return None
+        try:
+            converted = int(value)
+        except (TypeError, ValueError, OverflowError):
+            return None
+        if isinstance(value, float) and (not math.isfinite(value) or not value.is_integer()):
+            return None
+        return converted if converted in SLOT_IDS else None
+
+    @staticmethod
+    def _path(value: Any) -> str | None:
+        return value if isinstance(value, str) and value else None
+
+    def _matches_pending(self, slot: int | None, path: str | None) -> bool:
+        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
+
+    def _active_evidence(
+        self,
+        *,
+        slot: int | None,
+        path: str | None,
+        paused: bool | None,
+        status: Mapping[str, Any] | None = None,
+        telemetry: Mapping[str, Any] | None = None,
+    ) -> PlayerObservation:
+        previous = self.store.player_observation()
+        if slot is None:
+            slot = previous.active_slot
+        if path is None:
+            path = previous.active_path
+        if paused is None and previous.active_slot == slot:
+            paused = previous.paused
+        phase = "paused" if paused is True else "playing"
+        if slot is not None:
+            self.store.reconcile_active_guard(
+                slot,
+                kind="player_active",
+                details={"path": path},
+            )
+        return self.store.set_player_observation(
+            availability="online",
+            phase=phase,
+            active_slot=slot,
+            active_path=path,
+            paused=paused,
+            reconciliation_state="confirmed",
+            reconciliation_due_at=None,
+            status_payload=status,
+            telemetry_payload=telemetry,
+        )
+
+    def _idle_evidence(
+        self,
+        *,
+        status: Mapping[str, Any] | None = None,
+        telemetry: Mapping[str, Any] | None = None,
+    ) -> PlayerObservation:
+        # Persist idle before removing protection. A crash between the two
+        # steps can leave harmless over-protection, never unsafe under-protection.
+        observation = self.store.set_player_observation(
+            availability="online",
+            phase="idle",
+            active_slot=None,
+            active_path=None,
+            paused=False,
+            reconciliation_state="clear",
+            reconciliation_due_at=None,
+            status_payload=status,
+            telemetry_payload=telemetry,
+        )
+        self.store.clear_playback_guards(reason="fresh Player idle evidence")
+        return observation
+
+    def handle_status(self, payload: Any) -> list[PlaybackEffect]:
+        if not isinstance(payload, Mapping):
+            return [PlaybackEffect("warning", message="Player status must be an object")]
+        state = payload.get("state")
+        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"))
+        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(
+                    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,
+            )
+            effects.append(
+                PlaybackEffect(
+                    "player_offline",
+                    slot=previous.active_slot,
+                    message="Player availability is offline; playback is unknown",
+                )
+            )
+            return effects
+
+        if state == "idle" and payload.get("message") == "player online":
+            if path is not None:
+                self._active_evidence(slot=slot, path=path, paused=None, status=payload)
+                effects.append(
+                    PlaybackEffect(
+                        "player_reconnected_active",
+                        slot=slot,
+                        message="Reconnect idle announcement carried an active path",
+                    )
+                )
+            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"))
+            return effects
+
+        if state == "idle":
+            reason = payload.get("reason")
+            self._idle_evidence(status=payload)
+            effects.append(PlaybackEffect("player_idle", slot=slot, message=str(reason or "idle")))
+            if self.pending is not None:
+                if self.pending.action == "stop":
+                    effects.append(self._finish_operation("confirmed", str(reason or "idle")))
+                elif self.pending.action == "play":
+                    effects.append(self._finish_operation("failed", str(reason or "Player remained idle")))
+            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,
+                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"))
+            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,
+                status=payload,
+            )
+            effects.append(
+                PlaybackEffect(
+                    "control_issued",
+                    slot=slot,
+                    message=f"Player issued {state}; awaiting telemetry",
+                )
+            )
+            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,
+                    status=payload,
+                )
+            effects.append(PlaybackEffect("player_busy", slot=slot, message=payload.get("message")))
+            if self.pending is not None and self.pending.action == "play":
+                effects.append(self._finish_operation("failed", "Player is busy"))
+            return effects
+
+        if state == "error":
+            previous = self.store.player_observation()
+            active_slot = slot if slot is not None else previous.active_slot
+            active_path = path if path is not None else previous.active_path
+            if active_slot is not None:
+                self.store.reconcile_active_guard(
+                    active_slot,
+                    kind="player_active",
+                    details={"error": payload.get("message")},
+                )
+            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"),
+                reconciliation_due_at=None,
+                status_payload=payload,
+            )
+            message = str(payload.get("message") or "Player error")
+            effects.append(PlaybackEffect("player_error", slot=active_slot, message=message))
+            if self.pending is not None:
+                effects.append(self._finish_operation("failed", message))
+            return effects
+
+        effects.append(PlaybackEffect("warning", slot=slot, message=f"unknown Player state {state!r}"))
+        return effects
+
+    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"))
+        effects: list[PlaybackEffect] = []
+        if path is None:
+            self._idle_evidence(telemetry=payload)
+            effects.append(PlaybackEffect("player_idle", slot=slot, message="cleared telemetry"))
+            if self.pending is not None:
+                if self.pending.action == "stop":
+                    effects.append(self._finish_operation("confirmed", "cleared telemetry"))
+                elif self.pending.action == "play":
+                    effects.append(self._finish_operation("failed", "playback cleared"))
+            return effects
+
+        paused_value = payload.get("paused")
+        paused = paused_value if isinstance(paused_value, bool) else None
+        observation = self._active_evidence(
+            slot=slot,
+            path=path,
+            paused=paused,
+            telemetry=payload,
+        )
+        effects.append(PlaybackEffect("player_telemetry", slot=observation.active_slot))
+        operation = self.pending
+        if operation is None or not self._matches_pending(slot, path):
+            return effects
+        if operation.action == "play":
+            effects.append(self._finish_operation("confirmed", "matching telemetry"))
+        elif operation.action == "pause" and paused is True:
+            effects.append(self._finish_operation("confirmed", "telemetry paused=true"))
+        elif operation.action == "resume" and paused is False:
+            effects.append(self._finish_operation("confirmed", "telemetry paused=false"))
+        elif operation.action == "toggle" and paused is not None:
+            effects.append(self._finish_operation("confirmed", f"telemetry paused={paused}"))
+        return effects
+
+    def check_timeout(self) -> PlaybackEffect | None:
+        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,
+            )
+        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(
+                    previous.active_slot,
+                    kind="restart_uncertain",
+                    review_after=due,
+                    details={"operation_id": operation.operation_id, "reason": "stop timeout"},
+                )
+                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_due_at=due,
+                )
+        return self._finish_operation("timed_out", "no authoritative Player confirmation")
+
+    def projected_slot_state(self, slot: int) -> str:
+        record = self.store.get_slot(slot)
+        if record.media_state != "ready":
+            return record.media_state
+        observation = self.store.player_observation()
+        if observation.active_slot == slot:
+            if observation.reconciliation_state == "operator_review":
+                return "operator_review"
+            if observation.phase == "paused" and observation.paused is True:
+                return "paused"
+            if observation.phase == "playing":
+                return "playing"
+            if observation.phase in {"unknown", "error"}:
+                return "unknown"
+        guard = next(
+            (guard for guard in self.store.list_live_guards() if guard.slot == slot),
+            None,
+        )
+        if guard is not None:
+            if guard.kind == "pending_play":
+                return "ready"
+            return "operator_review" if guard.state == "review_required" else "unknown"
+        return "ready"
+
+    def mark_due_reconciliation(self) -> list[PlaybackEffect]:
+        due = self.store.mark_due_guards_for_review(self._wall_clock())
+        if not due:
+            return []
+        observation = self.store.player_observation()
+        if observation.active_slot is not None and any(
+            guard.slot == observation.active_slot for guard in due
+        ):
+            self.store.set_player_observation(
+                availability=observation.availability,
+                phase="unknown",
+                active_slot=observation.active_slot,
+                active_path=observation.active_path,
+                paused=observation.paused,
+                reconciliation_state="operator_review",
+                reconciliation_due_at=observation.reconciliation_due_at,
+            )
+        return [
+            PlaybackEffect(
+                "operator_review",
+                slot=guard.slot,
+                message=(
+                    "Playback remains unknown and protected; fresh Player evidence "
+                    "or an administrative clear is required"
+                ),
+            )
+            for guard in due
+        ]

controller/store.py

@@ -1060,6 +1060,79 @@ class ControllerStore:
             ).fetchall()
             return [self._row_to_guard(row) for row in rows]
 
+    def reconcile_active_guard(
+        self,
+        slot: int,
+        *,
+        kind: str = "player_active",
+        review_after: float | None = None,
+        details: Mapping[str, Any] | None = None,
+    ) -> SlotGuard:
+        """Replace all uncertain/pending guards with one authoritative guard."""
+        if slot not in SLOT_IDS:
+            raise ValueError("slot must be in 1..10")
+        with self._transaction() as connection:
+            now = self._now()
+            connection.execute(
+                """
+                UPDATE slot_guards
+                SET state = 'cleared', cleared_at = ?, cleared_by = 'player',
+                    clear_reason = 'reconciled by fresh active evidence'
+                WHERE state IN ('active','review_required')
+                """,
+                (now,),
+            )
+            guard_id = uuid.uuid4().hex
+            connection.execute(
+                """
+                INSERT INTO slot_guards
+                    (guard_id, slot_number, kind, state, created_at,
+                     review_after, details_json)
+                VALUES (?, ?, ?, 'active', ?, ?, ?)
+                """,
+                (
+                    guard_id,
+                    slot,
+                    kind,
+                    now,
+                    review_after,
+                    json.dumps(details or {}, sort_keys=True),
+                ),
+            )
+            row = connection.execute(
+                "SELECT * FROM slot_guards WHERE guard_id = ?", (guard_id,)
+            ).fetchone()
+            assert row is not None
+            return self._row_to_guard(row)
+
+    def clear_playback_guards(
+        self,
+        *,
+        reason: str,
+        slot: int | None = None,
+        kind: str | None = None,
+    ) -> int:
+        clauses = ["state IN ('active','review_required')"]
+        parameters: list[Any] = []
+        if slot is not None:
+            clauses.append("slot_number = ?")
+            parameters.append(slot)
+        if kind is not None:
+            clauses.append("kind = ?")
+            parameters.append(kind)
+        with self._transaction() as connection:
+            now = self._now()
+            result = connection.execute(
+                f"""
+                UPDATE slot_guards
+                SET state = 'cleared', cleared_at = ?, cleared_by = 'controller',
+                    clear_reason = ?
+                WHERE {' AND '.join(clauses)}
+                """,
+                (now, reason, *parameters),
+            )
+            return result.rowcount
+
     def mark_due_guards_for_review(self, now: float | None = None) -> list[SlotGuard]:
         now = self._now() if now is None else now
         with self._transaction() as connection:

tests/test_mqtt_adapter.py

@@ -0,0 +1,201 @@
+import json
+import queue
+import unittest
+from types import SimpleNamespace
+
+import paho.mqtt.client as mqtt
+
+from controller.config import ControllerConfig, MqttConfig
+from controller.mqtt_adapter import MqttAdapter
+
+
+class FakeClient:
+    def __init__(self):
+        self.calls = []
+        self.published = []
+        self.subscriptions = []
+        self.acks = []
+        self.suppress_exceptions = False
+        self.connect_error = None
+
+    def username_pw_set(self, username, password):
+        self.calls.append(("username_pw_set", username, password))
+
+    def manual_ack_set(self, enabled):
+        self.calls.append(("manual_ack_set", enabled))
+
+    def max_queued_messages_set(self, count):
+        self.calls.append(("max_queued_messages_set", count))
+
+    def reconnect_delay_set(self, **kwargs):
+        self.calls.append(("reconnect_delay_set", kwargs))
+
+    def connect_async(self, *args, **kwargs):
+        self.calls.append(("connect_async", args, kwargs))
+        if self.connect_error is not None:
+            raise self.connect_error
+        return mqtt.MQTT_ERR_SUCCESS
+
+    def connect(self, *args, **kwargs):
+        self.calls.append(("connect", args, kwargs))
+        raise AssertionError("synchronous connect must never be used")
+
+    def loop_start(self):
+        self.calls.append(("loop_start",))
+        return mqtt.MQTT_ERR_SUCCESS
+
+    def loop_stop(self):
+        self.calls.append(("loop_stop",))
+
+    def disconnect(self):
+        self.calls.append(("disconnect",))
+
+    def subscribe(self, topic, qos=0):
+        self.subscriptions.append((topic, qos))
+        return mqtt.MQTT_ERR_SUCCESS, len(self.subscriptions)
+
+    def publish(self, topic, payload, qos=0, retain=False):
+        self.published.append((topic, payload, qos, retain))
+        return SimpleNamespace(rc=mqtt.MQTT_ERR_SUCCESS, mid=len(self.published))
+
+    def ack(self, message_id, qos):
+        self.acks.append((message_id, qos))
+        return mqtt.MQTT_ERR_SUCCESS
+
+
+class MqttAdapterTests(unittest.TestCase):
+    def setUp(self):
+        self.client = FakeClient()
+        self.events = queue.Queue()
+        self.config = ControllerConfig(
+            mqtt=MqttConfig(
+                host="broker",
+                port=1884,
+                username="user",
+                password="pass",
+                client_id="controller-test",
+                reconnect_min_seconds=1,
+                reconnect_max_seconds=2,
+                command_queue_size=8,
+            )
+        )
+        self.adapter = MqttAdapter(
+            self.config,
+            client=self.client,
+            events=self.events,
+        )
+
+    def tearDown(self):
+        self.adapter.stop()
+
+    def test_start_is_asynchronous_and_configures_manual_ack(self):
+        self.adapter.start()
+        names = [call[0] for call in self.client.calls]
+        self.assertIn("manual_ack_set", names)
+        self.assertIn("connect_async", names)
+        self.assertIn("loop_start", names)
+        self.assertNotIn("connect", names)
+        connect = next(call for call in self.client.calls if call[0] == "connect_async")
+        self.assertEqual(connect[1], ("broker", 1884))
+        self.assertEqual(connect[2], {"keepalive": 30})
+
+    def test_initial_setup_oserror_does_not_escape_startup(self):
+        self.client.connect_error = OSError("network unreachable")
+        self.adapter.start()
+        self.assertTrue(self.adapter.started)
+        self.assertIn("network unreachable", self.adapter.last_error)
+        self.assertNotIn("loop_start", [call[0] for call in self.client.calls])
+
+    def test_connect_subscribes_to_all_authoritative_inputs(self):
+        self.adapter.start()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.assertTrue(self.adapter.connected)
+        self.assertEqual(
+            self.client.subscriptions,
+            [
+                ("youtube/player/status", 1),
+                ("youtube/player/telemetry", 0),
+                ("youtube/player/state", 0),
+                ("youtube/controller/cmd", 1),
+                ("audio/player/mode/state", 1),
+            ],
+        )
+        event = self.events.get_nowait()
+        self.assertEqual(event.kind, "connected")
+        self.assertEqual(event.connection_generation, 1)
+
+    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._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        receipt = self.adapter.publish_player_command(
+            {"action": "play", "slot": 3, "label": "Opening"}
+        )
+        self.assertTrue(receipt.accepted)
+        topic, payload, qos, retain = self.client.published[-1]
+        self.assertEqual(topic, "youtube/player/cmd")
+        self.assertEqual(json.loads(payload), {"action": "play", "slot": 3, "label": "Opening"})
+        self.assertEqual(qos, 1)
+        self.assertIs(retain, False)
+
+    def test_router_commands_are_qos_one_nonretained_plain_strings(self):
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        receipt = self.adapter.publish_router_mode("video")
+        self.assertTrue(receipt.accepted)
+        self.assertEqual(
+            self.client.published[-1],
+            ("audio/player/mode/set", "video", 1, False),
+        )
+
+    def test_qos_message_is_acked_only_after_consumer_calls_acknowledge(self):
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        message = SimpleNamespace(
+            topic="youtube/controller/cmd",
+            payload=b'{"action":"stop"}',
+            qos=1,
+            retain=False,
+            mid=42,
+        )
+        self.adapter._on_message(self.client, None, message)
+        event = self.events.get_nowait()
+        self.assertEqual(self.client.acks, [])
+        self.assertEqual(event.message_id, 42)
+        self.assertTrue(self.adapter.acknowledge(event))
+        self.assertEqual(self.client.acks, [(42, 1)])
+
+    def test_stale_mid_from_previous_connection_is_not_acked(self):
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        message = SimpleNamespace(
+            topic="youtube/controller/cmd",
+            payload=b'{"action":"stop"}',
+            qos=1,
+            retain=False,
+            mid=7,
+        )
+        self.adapter._on_message(self.client, None, message)
+        old_event = self.events.get_nowait()
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        self.assertFalse(self.adapter.acknowledge(old_event))
+        self.assertEqual(self.client.acks, [])
+
+    def test_retained_flag_is_preserved_for_coordinator_rejection(self):
+        self.adapter._on_connect(self.client, None, None, 0)
+        self.events.get_nowait()
+        message = SimpleNamespace(
+            topic="youtube/controller/cmd",
+            payload=b'{"action":"delete_slot","slot":1}',
+            qos=1,
+            retain=True,
+            mid=8,
+        )
+        self.adapter._on_message(self.client, None, message)
+        self.assertIs(self.events.get_nowait().retain, True)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_playback.py

@@ -0,0 +1,235 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from controller.config import ControllerConfig, StorageConfig
+from controller.playback import OperationInProgressError, PlaybackReducer
+from controller.store import ControllerStore
+
+
+class Clock:
+    def __init__(self, value):
+        self.value = value
+
+    def __call__(self):
+        return self.value
+
+
+class PlaybackReducerTests(unittest.TestCase):
+    def setUp(self):
+        self.tempdir = tempfile.TemporaryDirectory()
+        base = Path(self.tempdir.name)
+        self.wall = Clock(1_700_000_000.0)
+        self.monotonic = Clock(100.0)
+        self.config = ControllerConfig(
+            storage=StorageConfig(
+                media_base=base / "media",
+                data_base=base / "data",
+                retention_seconds=100,
+                uncertain_review_after_seconds=30,
+            )
+        )
+        self.store = ControllerStore(self.config.storage, now=self.wall).open()
+        self.reducer = PlaybackReducer(
+            self.config,
+            self.store,
+            monotonic=self.monotonic,
+            wall_clock=self.wall,
+            operation_timeout_seconds=5,
+        )
+        self._ready(1, "one")
+        self._ready(2, "two")
+
+    def tearDown(self):
+        self.store.close()
+        self.tempdir.cleanup()
+
+    def _ready(self, slot, suffix):
+        reservation = self.store.reserve_download(
+            f"https://example.test/{suffix}", None, None, job_id=f"job-{suffix}"
+        )
+        self.assertEqual(reservation.slot, slot)
+        self.store.mark_download_running(
+            reservation.job_id, reservation.slot, reservation.generation
+        )
+        self.store.complete_download(
+            reservation.job_id,
+            reservation.slot,
+            reservation.generation,
+            title=suffix.title(),
+            items=((suffix.title(), f"{suffix}.mp4"),),
+        )
+
+    def test_play_intent_is_not_optimistically_reported_playing(self):
+        operation = self.reducer.begin_operation(
+            "play",
+            slot=1,
+            expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
+        )
+        self.assertEqual(operation.state, "awaiting_player")
+        self.assertEqual(self.store.player_observation().phase, "unknown")
+        self.assertEqual(self.reducer.projected_slot_state(1), "ready")
+        with self.assertRaises(OperationInProgressError):
+            self.reducer.begin_operation("stop")
+
+    def test_matching_player_playing_confirms_play_and_protects_slot(self):
+        self.reducer.begin_operation(
+            "play",
+            slot=1,
+            expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
+        )
+        effects = self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+            }
+        )
+        self.assertIn("operation_confirmed", [effect.kind for effect in effects])
+        self.assertIsNone(self.reducer.pending)
+        self.assertEqual(self.store.player_observation().phase, "playing")
+        self.assertEqual(self.reducer.projected_slot_state(1), "playing")
+        guards = self.store.list_live_guards()
+        self.assertEqual([(guard.slot, guard.kind) for guard in guards], [(1, "player_active")])
+
+    def test_reconnect_online_idle_with_path_is_not_terminal(self):
+        self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+            }
+        )
+        effects = self.reducer.handle_status(
+            {
+                "state": "idle",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+                "message": "player online",
+            }
+        )
+        self.assertEqual(effects[0].kind, "player_reconnected_active")
+        self.assertEqual(self.store.player_observation().phase, "playing")
+        self.assertEqual(self.store.list_live_guards()[0].slot, 1)
+
+    def test_reconnect_online_idle_without_path_is_fresh_idle_evidence(self):
+        self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+            }
+        )
+        self.reducer.handle_status({"state": "idle", "message": "player online"})
+        self.assertEqual(self.store.player_observation().phase, "idle")
+        self.assertEqual(self.store.list_live_guards(), [])
+
+    def test_offline_without_slot_preserves_active_protection_as_unknown(self):
+        self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+            }
+        )
+        effects = self.reducer.handle_status({"state": "offline"})
+        observation = self.store.player_observation()
+        self.assertEqual(effects[0].kind, "player_offline")
+        self.assertEqual(observation.active_slot, 1)
+        self.assertEqual(observation.phase, "unknown")
+        self.assertEqual(self.reducer.projected_slot_state(1), "unknown")
+        self.assertEqual(self.store.list_live_guards()[0].kind, "restart_uncertain")
+
+    def test_busy_identifies_existing_playback_not_requested_slot(self):
+        self.reducer.handle_status(
+            {
+                "state": "playing",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+            }
+        )
+        self.reducer.begin_operation(
+            "play",
+            slot=2,
+            expected_path="/mnt/Media/YTPlayer/slot2/playlist.m3u",
+        )
+        effects = self.reducer.handle_status(
+            {
+                "state": "busy",
+                "slot": 1,
+                "path": "/mnt/Media/YTPlayer/slot1/playlist.m3u",
+                "message": "mpv already playing; stop first",
+            }
+        )
+        self.assertIn("operation_failed", [effect.kind for effect in effects])
+        self.assertEqual(self.store.player_observation().active_slot, 1)
+        self.assertEqual(self.reducer.projected_slot_state(1), "playing")
+        self.assertEqual(self.reducer.projected_slot_state(2), "ready")
+
+    def test_pause_status_does_not_override_actual_telemetry(self):
+        path = "/mnt/Media/YTPlayer/slot1/playlist.m3u"
+        self.reducer.handle_telemetry(
+            {"state": "telemetry", "slot": 1, "path": path, "paused": False}
+        )
+        self.reducer.begin_operation("pause", slot=1, expected_path=path)
+        effects = self.reducer.handle_status(
+            {"state": "paused", "slot": 1, "path": path}
+        )
+        self.assertIn("control_issued", [effect.kind for effect in effects])
+        self.assertIsNotNone(self.reducer.pending)
+        self.assertEqual(self.reducer.projected_slot_state(1), "playing")
+        effects = self.reducer.handle_telemetry(
+            {"state": "telemetry", "slot": 1, "path": path, "paused": True}
+        )
+        self.assertIn("operation_confirmed", [effect.kind for effect in effects])
+        self.assertEqual(self.reducer.projected_slot_state(1), "paused")
+
+    def test_terminal_idle_and_cleared_telemetry_release_protection(self):
+        path = "/mnt/Media/YTPlayer/slot1/playlist.m3u"
+        self.reducer.handle_telemetry(
+            {"state": "telemetry", "slot": 1, "path": path, "paused": False}
+        )
+        self.reducer.begin_operation("stop", slot=1, expected_path=path)
+        effects = self.reducer.handle_telemetry(
+            {"state": "telemetry", "slot": 1, "path": None, "paused": False}
+        )
+        self.assertIn("operation_confirmed", [effect.kind for effect in effects])
+        self.assertEqual(self.store.player_observation().phase, "idle")
+        self.assertEqual(self.store.list_live_guards(), [])
+        self.reducer.handle_status({"state": "idle", "slot": 1, "reason": "stop"})
+        self.assertEqual(self.reducer.projected_slot_state(1), "ready")
+
+    def test_error_does_not_release_owned_playback(self):
+        path = "/mnt/Media/YTPlayer/slot1/playlist.m3u"
+        self.reducer.handle_status({"state": "playing", "slot": 1, "path": path})
+        self.reducer.handle_status(
+            {"state": "error", "slot": 1, "path": path, "message": "cannot stop"}
+        )
+        self.assertEqual(self.store.player_observation().phase, "error")
+        self.assertEqual(self.store.list_live_guards()[0].slot, 1)
+        self.assertEqual(self.reducer.projected_slot_state(1), "unknown")
+
+    def test_play_timeout_becomes_unknown_then_operator_review_without_unprotecting(self):
+        self.reducer.begin_operation(
+            "play",
+            slot=1,
+            expected_path="/mnt/Media/YTPlayer/slot1/playlist.m3u",
+        )
+        self.monotonic.value += 6
+        effect = self.reducer.check_timeout()
+        self.assertEqual(effect.kind, "operation_timed_out")
+        self.assertEqual(self.reducer.projected_slot_state(1), "unknown")
+        self.wall.value += 31
+        review = self.reducer.mark_due_reconciliation()
+        self.assertEqual(review[0].kind, "operator_review")
+        self.assertEqual(self.reducer.projected_slot_state(1), "operator_review")
+        self.assertEqual(self.store.list_live_guards()[0].state, "review_required")
+
+    def test_malformed_and_unknown_statuses_are_isolated(self):
+        for payload in (None, [], {}, {"state": "mystery"}):
+            effects = self.reducer.handle_status(payload)
+            self.assertEqual(effects[0].kind, "warning")
+
+
+if __name__ == "__main__":
+    unittest.main()