From 8b3bdae8a0d59dbaf63823460ae3d16fa499a1b3 Mon Sep 17 00:00:00 2001 From: Media Date: Thu, 6 Aug 2026 17:41:07 -0500 Subject: [PATCH] feat: reduce authoritative player mqtt state --- controller/mqtt_adapter.py | 353 ++++++++++++++++++++++++++ controller/playback.py | 495 +++++++++++++++++++++++++++++++++++++ controller/store.py | 73 ++++++ tests/test_mqtt_adapter.py | 201 +++++++++++++++ tests/test_playback.py | 235 ++++++++++++++++++ 5 files changed, 1357 insertions(+) create mode 100644 controller/mqtt_adapter.py create mode 100644 controller/playback.py create mode 100644 tests/test_mqtt_adapter.py create mode 100644 tests/test_playback.py diff --git a/controller/mqtt_adapter.py b/controller/mqtt_adapter.py new file mode 100644 index 0000000..5882c53 --- /dev/null +++ b/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, + ) diff --git a/controller/playback.py b/controller/playback.py new file mode 100644 index 0000000..4b67010 --- /dev/null +++ b/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 + ] diff --git a/controller/store.py b/controller/store.py index 7647c1d..00c13e0 100644 --- a/controller/store.py +++ b/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: diff --git a/tests/test_mqtt_adapter.py b/tests/test_mqtt_adapter.py new file mode 100644 index 0000000..45fbcc8 --- /dev/null +++ b/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() diff --git a/tests/test_playback.py b/tests/test_playback.py new file mode 100644 index 0000000..ea051ad --- /dev/null +++ b/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()