"""Single-coordinator Controller v2 service.""" from __future__ import annotations import json import logging import os import queue import threading import time import uuid from dataclasses import dataclass, replace from pathlib import Path from typing import Any, Mapping from .config import ControllerConfig from .contracts import SLOT_IDS from .downloads import ( DownloadEvent, DownloadManager, canonical_playlist_path, ) from .ha import SlotAnnouncement, discovery_messages, slot_messages from .models import SlotProtectedError, StaleOperationError, StoreError from .mqtt_adapter import MqttAdapter, MqttEvent from .playback import OperationInProgressError, PlaybackEffect, PlaybackReducer from .router import AudioRouterCoordinator, RouterDecision from .store import ControllerStore LOG = logging.getLogger(__name__) @dataclass class _ServiceCall: action: str payload: Any response: "queue.Queue[Any]" @dataclass(frozen=True) class _InternalEvent: kind: str payload: Any @dataclass(frozen=True) class _PendingPlay: intent_id: str command: Mapping[str, Any] slot: int | None expected_path: str | None lease_id: str | None = None class ControllerUnavailableError(RuntimeError): pass class ControllerService: """Coordinates MQTT, Player evidence, Router leases, and media inventory.""" def __init__( self, config: ControllerConfig, *, store: ControllerStore | None = None, mqtt_adapter: MqttAdapter | None = None, event_queue: "queue.Queue[Any] | None" = None, download_manager: DownloadManager | None = None, playback: PlaybackReducer | None = None, router: AudioRouterCoordinator | None = None, monotonic=time.monotonic, wall_clock=time.time, ): self.config = config self.events: "queue.Queue[Any]" = event_queue or queue.Queue( maxsize=config.mqtt.command_queue_size ) self.store = store or ControllerStore(config.storage, now=wall_clock) self.mqtt = mqtt_adapter or MqttAdapter( config, events=self.events, monotonic=monotonic, ) self.downloads = download_manager self.playback = playback self.router = router self._monotonic = monotonic self._wall_clock = wall_clock self._lifecycle_lock = threading.Lock() self._stop_event = threading.Event() self._coordinator: threading.Thread | None = None self._started = False self._pending_play: _PendingPlay | None = None self._session_lease_id: str | None = None self._last_announced: dict[int, tuple[Any, ...]] = {} self._next_expiry = 0.0 self._next_reconciliation = 0.0 self._processed_packets: dict[tuple[int, str, int], float] = {} self._player_evidence_ready = False self._operator_cleanup_override = False self._cleanup_review_id: str | None = None @property def started(self) -> bool: with self._lifecycle_lock: return self._started def start(self) -> None: with self._lifecycle_lock: if self._started: return if self._coordinator is not None and self._coordinator.is_alive(): raise ControllerUnavailableError( "previous Controller coordinator has not stopped" ) try: self.store.open() self.config.storage.media_base.mkdir(parents=True, exist_ok=True) self.store.abandon_audio_leases_on_restart() router_observation = self.store.router_observation() self.store.set_router_observation( mode=router_observation.mode, freshness=( "stale" if router_observation.mode is not None else "unknown" ), warning=None, ) self.store.mark_restart_uncertain() if self.playback is None: self.playback = PlaybackReducer( self.config, self.store, monotonic=self._monotonic, wall_clock=self._wall_clock, ) if self.router is None: self.router = AudioRouterCoordinator( self.config, self.store, self.mqtt.publish_router_mode, clock=self._monotonic, ) if self.downloads is None: self.downloads = DownloadManager( self.config, self.store, on_event=self._queue_download_event, ) try: self.downloads.reconcile_export_artifacts() except BaseException: LOG.exception( "utility export reconciliation failed; Controller startup continues" ) # Cleanup is deliberately deferred until fresh Player evidence # (or an explicit administrative reconciliation) makes it safe. self._reset_cleanup_evidence() self._stop_event.clear() self._next_expiry = ( self._monotonic() + self.config.storage.expiry_interval_seconds ) self._next_reconciliation = self._monotonic() + 1.0 self._coordinator = threading.Thread( target=self._run, name="controller-coordinator", daemon=True, ) self._coordinator.start() self.mqtt.start() self._started = True except BaseException: self._stop_event.set() try: self.mqtt.stop() except BaseException: LOG.exception("MQTT cleanup failed after Controller startup error") coordinator = self._coordinator if coordinator is not None and coordinator.is_alive(): coordinator.join(timeout=5.0) if coordinator is None or not coordinator.is_alive(): self.store.close() self._coordinator = None raise def stop(self) -> None: with self._lifecycle_lock: if not self._started: coordinator = self._coordinator if coordinator is not None and not coordinator.is_alive(): self.store.close() self._coordinator = None return self._started = False self._stop_event.set() self.mqtt.stop() if self.downloads is not None: self.downloads.shutdown(wait=True) try: self.events.put_nowait(_InternalEvent("shutdown", None)) except queue.Full: pass coordinator = self._coordinator if coordinator is threading.current_thread(): LOG.error("Controller stop was requested on its coordinator thread") return if coordinator is not None: coordinator.join(timeout=5.0) if coordinator is not None and coordinator.is_alive(): LOG.error( "Controller coordinator did not stop; store remains open for safety" ) return self.store.close() self._coordinator = None def _queue_download_event(self, event: DownloadEvent) -> None: if threading.current_thread() is self._coordinator: self._handle_download_event(event) return self._put_event(_InternalEvent("download", event)) def _put_event(self, event: Any) -> 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 _call(self, action: str, payload: Any, timeout: float = 10.0) -> Any: if not self.started: raise ControllerUnavailableError("Controller service is not started") if threading.current_thread() is self._coordinator: return self._handle_call(action, payload) response: "queue.Queue[Any]" = queue.Queue(maxsize=1) if not self._put_event(_ServiceCall(action, payload, response)): raise ControllerUnavailableError("Controller service is stopping") try: result = response.get(timeout=timeout) except queue.Empty as exc: raise ControllerUnavailableError("Controller coordinator did not respond") from exc if isinstance(result, BaseException): raise result return result def enqueue_download( self, url: str, submitter: str | None, note: str | None, ) -> dict[str, Any]: return self._call( "enqueue_download", {"url": url, "submitter": submitter, "note": note}, ) def enqueue_export( self, url: str, mode: str, submitter: str | None, note: str | None, ) -> dict[str, Any]: return self._call( "enqueue_export", { "url": url, "mode": mode, "submitter": submitter, "note": note, }, ) def list_export_jobs(self, *, limit: int = 10) -> list[dict[str, Any]]: return self._call("list_export_jobs", {"limit": limit}) def submit_control(self, payload: Mapping[str, Any]) -> dict[str, Any]: return self._call("control", dict(payload)) def health(self) -> dict[str, Any]: with self._lifecycle_lock: started = self._started coordinator_alive = ( self._coordinator is not None and self._coordinator.is_alive() ) router_observation = ( self.store.router_observation() if started else None ) player_observation = ( self.store.player_observation() if started else None ) guards = self.store.list_live_guards() if started else [] player_operation_pending = bool( started and ( self._pending_play is not None or ( self.playback is not None and self.playback.pending is not None ) ) ) mqtt_connected = bool(started and self.mqtt.connected) degraded: list[str] = [] if started and not mqtt_connected: degraded.append("mqtt_disconnected") if started and not self._cleanup_allowed(): degraded.append("awaiting_player_evidence") if ( started and router_observation is not None and router_observation.freshness != "current" ): degraded.append("router_state_unavailable") reviews = [ { "review_id": guard.guard_id, "slot": guard.slot, "kind": guard.kind, "state": ( "operator_review" if guard.state == "review_required" else "awaiting_evidence" ), "review_after": guard.review_after, "clearable": not player_operation_pending, } for guard in guards if guard.kind in {"restart_uncertain", "legacy_uncertain"} ] if ( player_observation is not None and player_observation.active_slot is None and player_observation.active_path is not None and player_observation.reconciliation_id is not None and player_observation.reconciliation_state in {"awaiting_evidence", "operator_review"} ): reviews.append( { "review_id": player_observation.reconciliation_id, "slot": None, "path": player_observation.active_path, "kind": "external_playback", "state": player_observation.reconciliation_state, "review_after": player_observation.reconciliation_due_at, "clearable": not player_operation_pending, } ) cleanup_reconciliation = None if ( started and not self._cleanup_allowed() and self._cleanup_review_id is not None and not guards and player_observation is not None and player_observation.active_slot is None and player_observation.active_path is None ): cleanup_reconciliation = { "review_id": self._cleanup_review_id, "kind": "player_evidence", "state": "awaiting_evidence", "clearable": not player_operation_pending, } return { "status": ( "stopped" if not started else "degraded" if degraded else "ok" ), "started": started, "coordinator_alive": coordinator_alive, "mqtt_connected": mqtt_connected, "router_state": ( None if router_observation is None else router_observation.mode ), "player": ( None if player_observation is None else { "availability": player_observation.availability, "phase": player_observation.phase, "active_slot": player_observation.active_slot, "active_path": player_observation.active_path, "paused": player_observation.paused, "reconciliation_state": player_observation.reconciliation_state, } ), "reconciliation_reviews": reviews, "cleanup_reconciliation": cleanup_reconciliation, "degraded_reasons": degraded, } def _run(self) -> None: while not self._stop_event.is_set(): try: event = self.events.get(timeout=0.1) except queue.Empty: event = None if event is not None: try: self._handle_event(event) except BaseException: LOG.exception("Controller coordinator contained an event failure") try: self._run_timers() except BaseException: LOG.exception("Controller coordinator contained a timer failure") def _handle_event(self, event: Any) -> None: if isinstance(event, _ServiceCall): try: result = self._handle_call(event.action, event.payload) except BaseException as exc: event.response.put(exc) else: event.response.put(result) return if isinstance(event, _InternalEvent): if event.kind == "download": self._handle_download_event(event.payload) return if isinstance(event, MqttEvent): self._handle_mqtt_event(event) def _handle_call(self, action: str, payload: Any) -> Any: if action == "enqueue_export": assert self.downloads is not None reservation = self.downloads.enqueue_export( payload["url"], payload["mode"], payload.get("submitter"), payload.get("note"), ) return { "kind": "export", "status": reservation.status, "job_id": reservation.job_id, "mode": payload["mode"], } if action == "list_export_jobs": jobs = self.store.list_export_jobs(limit=payload["limit"]) return [ { "job_id": job.job_id, "mode": job.mode, "url": job.url, "status": job.status, "requested_at": job.requested_at, "finished_at": job.finished_at, "output_paths": list(job.output_paths), "error": job.error, } for job in jobs ] if action == "enqueue_download": assert self.downloads is not None url = payload["url"] if not self._cleanup_allowed() and self._ingest_requires_evidence(url): raise ControllerUnavailableError( "refreshing or replacing existing playlist media is deferred " "until fresh Player evidence or administrative playback " "reconciliation" ) reservation = self.downloads.enqueue( url, payload.get("submitter"), payload.get("note") ) result: dict[str, Any] = { "status": "queued" if reservation.status == "reserved" else reservation.status, "slot": reservation.slot, "reused_slot": reservation.reused_slot, } if reservation.oldest_expires_at is not None: result["oldest_expires_at"] = reservation.oldest_expires_at result["oldest_expires_in_seconds"] = max( 0, int(reservation.oldest_expires_at - self._wall_clock()), ) return result if action == "control": return self._handle_control_command(payload) raise ValueError(f"unknown service call {action!r}") def _handle_mqtt_event(self, event: MqttEvent) -> None: packet_key = self._mqtt_packet_key(event) handled_successfully = False try: if packet_key is not None and self._is_duplicate_packet(event, packet_key): handled_successfully = True return if event.kind == "connected": self._reset_cleanup_evidence() self._announce_discovery() self._sync_slot_states(force=True) self._announce_reconciliation_reviews() handled_successfully = True return if event.kind == "disconnected": self._reset_cleanup_evidence() assert self.playback is not None assert self.router is not None playback_effect = self.playback.mark_transport_uncertain( f"Controller MQTT connection was lost: {event.reason}" ) self._handle_playback_effects([playback_effect]) decision = self.router.connection_lost() if decision is not None: self._handle_router_decision(decision) handled_successfully = True return if event.kind != "message" or event.topic is None: handled_successfully = True return if event.topic == self.config.topics.player_status: if event.retain: self._publish_event( "retained_player_evidence_rejected", "warning", "Retained Player status was ignored as historical evidence", ) else: payload = self._decode_json(event.payload, "Player status") if payload is not None: assert self.playback is not None before_generation = self.store.player_observation().generation effects = self.playback.handle_status(payload) self._handle_playback_effects(effects) self._refresh_player_evidence(before_generation, effects) elif event.topic == self.config.topics.player_telemetry: if event.retain: self._publish_event( "retained_player_evidence_rejected", "warning", "Retained Player telemetry was ignored as historical evidence", ) else: payload = self._decode_json(event.payload, "Player telemetry") if payload is not None: assert self.playback is not None before_generation = self.store.player_observation().generation effects = self.playback.handle_telemetry(payload) self._handle_playback_effects(effects) self._refresh_player_evidence(before_generation, effects) elif event.topic == self.config.topics.player_state: # The Player owns this retained Companion topic. Its value is # historical across outages and reconnect deliberately emits # flat ``idle`` even while MPV is active, so it is never fresh # cleanup authority. pass elif event.topic == self.config.topics.controller_command: if event.retain: self._publish_event( "retained_command_rejected", "warning", "Retained Controller commands are rejected to prevent replay", ) else: payload = self._decode_json(event.payload, "Controller command") if payload is not None: try: self._handle_control_command(payload) except ( ValueError, StoreError, ControllerUnavailableError, OperationInProgressError, ) as exc: self._publish_event( "controller_command_rejected", "warning", str(exc), ) elif event.topic in { self.config.topics.router_mode_set, self.config.topics.router_mode_state, }: try: mode = event.payload.decode("utf-8").strip() except UnicodeDecodeError: self._publish_event( "router_state_invalid", "warning", "Audio Router mode payload was not valid UTF-8", ) else: assert self.router is not None if event.topic == self.config.topics.router_mode_set: decision = self.router.observe_mode_command( mode, retained=event.retain, controller_echo=event.controller_echo, ) else: decision = self.router.observe_mode(mode) self._handle_router_decision(decision) # The flat state subscription is compatibility observation only. # Playback and cleanup truth come exclusively from detailed status # and telemetry or an explicit administrative reconciliation. handled_successfully = True finally: if handled_successfully: if packet_key is not None: self._remember_packet(packet_key) self.mqtt.acknowledge(event) @staticmethod def _mqtt_packet_key(event: MqttEvent) -> tuple[int, str, int] | None: if ( event.kind != "message" or event.qos <= 0 or event.message_id is None or event.connection_generation is None or event.topic is None ): return None return (event.connection_generation, event.topic, event.message_id) def _is_duplicate_packet( self, event: MqttEvent, key: tuple[int, str, int], ) -> bool: return event.duplicate and key in self._processed_packets def _remember_packet(self, key: tuple[int, str, int]) -> None: now = self._monotonic() self._processed_packets[key] = now cutoff = now - 300.0 if len(self._processed_packets) > 1024: stale = sorted( self._processed_packets.items(), key=lambda item: item[1] )[: len(self._processed_packets) - 1024] for candidate, _ in stale: self._processed_packets.pop(candidate, None) for candidate, processed_at in tuple(self._processed_packets.items()): if processed_at < cutoff: self._processed_packets.pop(candidate, None) def _decode_json(self, payload: bytes, description: str) -> Any | None: try: value = json.loads(payload.decode("utf-8")) except (UnicodeDecodeError, ValueError, RecursionError) as exc: self._publish_event( "malformed_mqtt_payload", "warning", f"{description} was rejected: {exc}", ) return None return value @staticmethod def _parse_slot(value: Any) -> int: if value is None or isinstance(value, bool): raise ValueError("slot must be an integer from 1 through 10") if isinstance(value, float) and not value.is_integer(): raise ValueError("slot must be an integer from 1 through 10") try: slot = int(value) except (TypeError, ValueError, OverflowError) as exc: raise ValueError("slot must be an integer from 1 through 10") from exc if slot not in SLOT_IDS: raise ValueError("slot must be an integer from 1 through 10") return slot def _handle_control_command(self, payload: Any) -> dict[str, Any]: if not isinstance(payload, Mapping): raise ValueError("Controller command must be a JSON object") action = payload.get("action") if not isinstance(action, str) or not action: raise ValueError("Controller command is missing a string action") if action == "play": return self._request_play(payload) if action in {"stop", "pause", "resume", "toggle"}: return self._request_player_control(action) if action == "delete_slot": return self._request_delete(payload) if action == "reconcile_playback": return self._request_reconciliation(payload) raise ValueError(f"unknown Controller action {action!r}") def _request_play(self, payload: Mapping[str, Any]) -> dict[str, Any]: assert self.playback is not None assert self.router is not None if self._pending_play is not None or self.playback.pending is not None: raise ValueError("another Player transition is still pending") observation = self.store.player_observation() if ( observation.phase != "idle" and ( observation.active_slot is not None or observation.active_path is not None ) ): raise ValueError( f"Player state is {observation.phase} for " f"{observation.active_path or f'slot {observation.active_slot}'}; stop first" ) if self.store.list_live_guards(): raise SlotProtectedError( "playback protection is unresolved; reconcile it before starting another slot" ) slot = ( self._parse_slot(payload.get("slot")) if payload.get("slot") is not None else None ) expected_path: str | None = None command: dict[str, Any] = {"action": "play"} path_value = payload.get("path") playlist_value = payload.get("playlist") explicit_key: str | None = None explicit_value: Any = None if path_value is not None: explicit_key, explicit_value = "path", path_value elif playlist_value is not None: explicit_key, explicit_value = "playlist", playlist_value if explicit_key is not None: if not isinstance(explicit_value, str) or not explicit_value: raise ValueError(f"play {explicit_key} must be a non-empty string") expected_path = os.path.abspath(explicit_value) if not Path(expected_path).is_file(): raise ValueError(f"playback path is missing: {expected_path}") command[explicit_key] = expected_path if slot is not None: record = self.store.get_slot(slot) if record.media_state != "ready": raise ValueError(f"slot {slot} is {record.media_state}, not ready") if expected_path is not None: path_slot = self._managed_slot_for_path(expected_path) if path_slot is not None and path_slot != slot: raise ValueError( f"slot {slot} conflicts with managed playback path for slot {path_slot}" ) if expected_path is None: playlist = canonical_playlist_path( self.config.storage.media_base, slot ) if not playlist.is_file(): raise ValueError(f"slot {slot} playlist is missing: {playlist}") expected_path = os.path.abspath(playlist) command["slot"] = slot label = payload.get("label") or record.title if isinstance(label, str) and label: command["label"] = label elif expected_path is not None: derived_slot = self._managed_slot_for_path(expected_path) if derived_slot is not None: record = self.store.get_slot(derived_slot) if record.media_state != "ready": raise ValueError( f"slot {derived_slot} is {record.media_state}, not ready" ) slot = derived_slot command["slot"] = slot label = payload.get("label") or payload.get("title") if isinstance(label, str) and label: command["label"] = label else: raise ValueError("play command requires slot, path, or playlist") intent_id = uuid.uuid4().hex if slot is not None: self.store.protect_slot( slot, "pending_play", guard_id=f"router-{intent_id}", details={"intent_id": intent_id}, ) self._pending_play = _PendingPlay( intent_id=intent_id, command=command, slot=slot, expected_path=expected_path, ) try: decision = self.router.acquire(intent_id) except BaseException: if slot is not None: self.store.clear_playback_guards( slot=slot, kind="pending_play", reason="Audio Router acquisition failed", ) self._pending_play = None raise self._pending_play = replace( self._pending_play, lease_id=decision.lease_id, ) self._handle_router_decision(decision) if decision.action == "rejected": return { "status": "rejected", "operation_id": intent_id, "message": decision.warning, } if decision.action == "ready": pending = self.playback.pending if pending is not None and pending.operation_id == intent_id: return { "status": "pending", "operation_id": intent_id, "router": decision.action, "warning": decision.warning, } completed = self.playback.last_operation if completed is not None and completed.operation_id == intent_id: return { "status": ( "unknown" if completed.state == "uncertain" else "unavailable" ), "operation_id": intent_id, "router": decision.action, "warning": completed.message or decision.warning, } return { "status": "pending", "operation_id": intent_id, "router": decision.action, "warning": decision.warning, } def _managed_slot_for_path(self, path: str) -> int | None: candidate = os.path.abspath(path) for slot in SLOT_IDS: directory = os.path.abspath( self.config.storage.media_base / f"slot{slot}" ) try: if os.path.commonpath((candidate, directory)) == directory: return slot except ValueError: continue return None def _publish_pending_play(self) -> None: intent = self._pending_play if intent is None: return assert self.playback is not None assert self.router is not None try: operation = self.playback.begin_operation( "play", slot=intent.slot, expected_path=intent.expected_path, operation_id=intent.intent_id, ) except OperationInProgressError as exc: self.router.abandon(intent.lease_id, str(exc)) if intent.slot is not None: self.store.clear_playback_guards( slot=intent.slot, kind="pending_play", reason="Player operation could not begin", ) self._pending_play = None return receipt = self.mqtt.publish_player_command(intent.command) self._pending_play = None if not receipt.accepted: reason = receipt.reason or "Player command publish failed" if receipt.uncertain: self._session_lease_id = intent.lease_id effect = self.playback.publication_uncertain(reason) else: effect = self.playback.publish_failed(reason) if effect is not None: self._handle_playback_effects([effect]) if not receipt.uncertain: self._handle_router_decision(self.router.release(intent.lease_id)) return self._session_lease_id = intent.lease_id self._publish_event( "player_command_pending", "info", "Play command was published; awaiting authoritative Player state", slot=intent.slot, operation_id=operation.operation_id, ) def _request_player_control(self, action: str) -> dict[str, Any]: assert self.playback is not None if self._pending_play is not None or self.playback.pending is not None: raise ValueError("another Player transition is still pending") observation = self.store.player_observation() if ( action != "stop" and observation.active_slot is None and observation.active_path is None ): raise ValueError(f"nothing is known to be playing for {action}") player_action = action if action == "toggle": if observation.paused is None: raise ValueError( "toggle requires authoritative pause telemetry; use pause or resume" ) player_action = "resume" if observation.paused else "pause" operation = self.playback.begin_operation( player_action, slot=observation.active_slot, expected_path=observation.active_path, ) receipt = self.mqtt.publish_player_command({"action": player_action}) if not receipt.accepted: reason = receipt.reason or "Player command publish failed" effects: list[PlaybackEffect] = [] if receipt.uncertain: effects.append(self.playback.mark_transport_uncertain(reason)) uncertain = self.playback.publication_uncertain(reason) if uncertain is not None: effects.append(uncertain) else: failed = self.playback.publish_failed(reason) if failed is not None: effects.append(failed) self._handle_playback_effects(effects) if receipt.uncertain: return { "status": "unknown", "operation_id": operation.operation_id, "requested_action": action, "player_action": player_action, "warning": reason, } raise ControllerUnavailableError(reason) self._publish_event( "player_command_pending", "info", f"{player_action} command was published; awaiting authoritative Player state", slot=observation.active_slot, operation_id=operation.operation_id, ) result = {"status": "pending", "operation_id": operation.operation_id} if action != player_action: result.update( {"requested_action": action, "player_action": player_action} ) return result def _request_delete(self, payload: Mapping[str, Any]) -> dict[str, Any]: assert self.downloads is not None if ( not self._cleanup_allowed() and payload.get("confirm_player_unknown") is not True ): raise SlotProtectedError( "delete requires fresh Player evidence or " "confirm_player_unknown=true" ) slot = self._parse_slot(payload.get("slot")) expected_url = payload.get("url") if expected_url is not None and not isinstance(expected_url, str): raise ValueError("delete URL guard must be a string") self.downloads.schedule_delete_slot( slot, reason="manual", expected_url=expected_url, ) self._announce_slot(slot, force=True) return {"status": "deleting", "slot": slot} def _request_reconciliation(self, payload: Mapping[str, Any]) -> dict[str, Any]: if payload.get("decision") != "clear" or payload.get("confirm") is not True: raise ValueError("reconciliation clear requires decision=clear and confirm=true") scope = payload.get("scope") if scope not in {None, "cleanup"}: raise ValueError("reconciliation scope must be cleanup when provided") slot = ( self._parse_slot(payload.get("slot")) if payload.get("slot") is not None else None ) review_id = payload.get("review_id") actor = payload.get("actor") reason = payload.get("reason") if not all(isinstance(value, str) and value.strip() for value in (review_id, actor, reason)): raise ValueError("reconciliation requires review_id, actor, and reason") assert self.playback is not None if self._pending_play is not None or self.playback.pending is not None: raise ValueError( "playback reconciliation cannot be cleared while a Player operation is pending" ) if scope == "cleanup": if slot is not None: raise ValueError("cleanup reconciliation does not accept a slot") observation = self.store.player_observation() if ( self._cleanup_allowed() or self._cleanup_review_id is None or review_id.strip() != self._cleanup_review_id ): raise StaleOperationError("cleanup reconciliation review_id is stale") if ( self.store.list_live_guards() or observation.active_slot is not None or observation.active_path is not None ): raise SlotProtectedError( "cleanup reconciliation cannot bypass known playback protection" ) self.store.record_audit_event( "player_evidence_admin_override", actor=actor.strip(), details={ "review_id": review_id.strip(), "reason": reason.strip(), }, ) self._operator_cleanup_override = True self._cleanup_review_id = None cleared = True elif slot is None: cleared = self.store.clear_external_reconciliation( review_id.strip(), actor=actor.strip(), reason=reason.strip(), ) else: guard = next( ( candidate for candidate in self.store.list_live_guards() if candidate.slot == slot ), None, ) if guard is None or guard.kind not in { "restart_uncertain", "legacy_uncertain", }: raise ValueError(f"slot {slot} has no uncertain playback review") cleared = self.store.clear_guard( slot, review_id.strip(), actor=actor.strip(), reason=reason.strip(), ) if not cleared: raise StaleOperationError("reconciliation review_id is stale") observation = self.store.player_observation() if ( not self.store.list_live_guards() and observation.active_slot is None and observation.active_path is None ): if scope != "cleanup": self._operator_cleanup_override = True self._cleanup_review_id = None self._schedule_deletion_recovery() assert self.router is not None self.router.abandon( self._session_lease_id, "Administrative playback reconciliation cleared uncertain ownership", ) self._session_lease_id = None if slot is not None: self._announce_slot(slot, force=True) self._publish_event( "playback_reconciled", "warning", f"Operator cleared uncertain playback protection: {reason.strip()}", slot=slot, operation_id=review_id.strip(), ) result: dict[str, Any] = { "status": "cleared", "review_id": review_id.strip(), } if scope == "cleanup": result["scope"] = "cleanup" if slot is not None: result["slot"] = slot return result def _handle_router_decision(self, decision: RouterDecision) -> None: if decision.warning: self._publish_event( "audio_router_warning", "warning" if decision.action != "rejected" else "error", decision.warning, slot=(None if self._pending_play is None else self._pending_play.slot), operation_id=( None if self._pending_play is None else self._pending_play.intent_id ), continued=decision.continued, ) if decision.action == "ready" and self._pending_play is not None: if decision.lease_id == self._pending_play.lease_id: self._publish_pending_play() elif decision.action == "rejected" and self._pending_play is not None: intent = self._pending_play if intent.slot is not None: self.store.clear_playback_guards( slot=intent.slot, kind="pending_play", reason="Audio Router policy rejected playback", ) self._pending_play = None def _handle_playback_effects(self, effects: list[PlaybackEffect]) -> None: assert self.router is not None terminal = any(effect.kind == "player_idle" for effect in effects) failed_play = ( any(effect.kind == "operation_failed" for effect in effects) and self.playback is not None and self.playback.last_operation is not None and self.playback.last_operation.action == "play" ) observation = self.store.player_observation() no_active_player = ( observation.active_slot is None and observation.active_path is None ) unidentified_busy = ( no_active_player and any(effect.kind == "player_busy" for effect in effects) ) if terminal or (failed_play and no_active_player and not unidentified_busy): self._handle_router_decision(self.router.release(self._session_lease_id)) self._session_lease_id = None for effect in effects: if effect.kind in { "warning", "player_error", "player_offline", "transport_uncertain", "operation_failed", "operation_uncertain", "operation_timed_out", "operator_review", }: self._publish_event( effect.kind, "warning" if effect.kind != "player_error" else "error", effect.message or effect.kind, slot=effect.slot, operation_id=effect.operation_id, ) elif effect.kind == "operation_confirmed": self._publish_event( "player_operation_confirmed", "info", "Player confirmed the requested operation", slot=effect.slot, operation_id=effect.operation_id, ) self._sync_slot_states() def _handle_download_event(self, event: DownloadEvent) -> None: if event.state == "deleted": self._announce_slot(event.slot, force=True) return if event.state == "delete_deferred": self._announce_slot(event.slot, force=True) return if event.state == "delete_error": message = event.message or "slot deletion remains pending" self._publish_event( "slot_deletion_pending", "warning", message, slot=event.slot, ) self._announce_slot(event.slot, force=True) retry_delay = max( 1.0, min(30.0, self.config.storage.expiry_interval_seconds), ) self._next_expiry = self._monotonic() + retry_delay return payload: dict[str, Any] = {"slot": event.slot, "state": event.state} for key in ("title", "count", "message", "url", "submitter", "note"): value = getattr(event, key) if value is not None: payload[key] = value if event.state == "queued": payload["reused_slot"] = event.reused_slot self.mqtt.publish_json( self.config.topics.ingest_progress, payload, qos=0, retain=False, ) self._announce_slot(event.slot) def _announce_discovery(self) -> None: if not self.config.homeassistant.enabled: return for topic, payload, _, retain in discovery_messages( self.config.topics, self.config.homeassistant, ): self.mqtt.publish_legacy(topic, payload, retain=retain) def _announce_reconciliation_reviews(self) -> None: """Re-announce durable review tokens after each broker connection.""" for guard in self.store.list_live_guards(): if guard.kind not in {"restart_uncertain", "legacy_uncertain"}: continue self._publish_event( "playback_reconciliation_required", "warning", ( f"Slot {guard.slot} playback is uncertain and protected " f"(review_id={guard.guard_id}); fresh Player evidence or an " "explicit administrative clear is required" ), slot=guard.slot, operation_id=guard.guard_id, ) observation = self.store.player_observation() if ( observation.active_slot is None and observation.active_path is not None and observation.reconciliation_id is not None and observation.reconciliation_state in {"awaiting_evidence", "operator_review"} ): self._publish_event( "playback_reconciliation_required", "warning", ( f"Playback at {observation.active_path!r} is uncertain and " f"protected (review_id={observation.reconciliation_id}); fresh " "Player evidence or an explicit administrative clear is required" ), operation_id=observation.reconciliation_id, ) def _sync_slot_states(self, *, force: bool = False) -> None: for slot in SLOT_IDS: self._announce_slot(slot, force=force) def _announce_slot(self, slot: int, *, force: bool = False) -> None: assert self.playback is not None record = self.store.get_slot(slot) item_records = self.store.get_slot_items(slot) item_titles = tuple(title for title, _ in item_records) effective_state = self.playback.projected_slot_state(slot) signature = (record.title, effective_state, record.item_count, item_titles) if not force and self._last_announced.get(slot) == signature: return authoritative_items = ( record.media_state == "empty" or len(item_titles) == record.item_count ) clear_through = ( max(record.published_item_count, len(item_titles)) if authoritative_items else 0 ) all_accepted = True for topic, payload, _, retain in slot_messages( SlotAnnouncement( slot=slot, title=record.title, state=effective_state, count=record.item_count, items=item_titles, ), self.config.topics, clear_items_through=clear_through, ): receipt = self.mqtt.publish_legacy(topic, payload, retain=retain) all_accepted = all_accepted and receipt.accepted if all_accepted: self._last_announced[slot] = signature if authoritative_items: self.store.set_published_item_count(slot, len(item_titles)) def _publish_event( self, event_type: str, severity: str, message: str, *, slot: int | None = None, operation_id: str | None = None, continued: bool | None = None, ) -> None: event_id = uuid.uuid4().hex payload: dict[str, Any] = { "version": 2, "event_id": event_id, "type": event_type, "severity": severity, "message": message, "ts": self._wall_clock(), } if slot is not None: payload["slot"] = slot if operation_id is not None: payload["operation_id"] = operation_id if continued is not None: payload["continued"] = continued self.store.record_audit_event( event_type, slot=slot, actor="controller", details=payload, ) if severity in {"warning", "error"}: LOG.warning("%s: %s", event_type, message) else: LOG.info("%s: %s", event_type, message) topic = self.config.topics.controller_result if topic is not None: self.mqtt.publish_json(topic, payload, qos=1, retain=False) def _cleanup_allowed(self) -> bool: return self._player_evidence_ready or self._operator_cleanup_override def _ingest_requires_evidence(self, url: Any) -> bool: """True when ingesting ``url`` could reuse/refresh occupied media. A URL that does not already occupy a slot can only land in a slot that ``reserve_download`` independently proves is empty and unguarded, so it cannot mutate, delete, or replace media the Player may currently be using. Only the reuse/refresh path needs the same conservative evidence gate as deletion and expiry. """ if not isinstance(url, str): return False return self.store.find_occupied_slot_for_url(url) is not None def _schedule_deletion_recovery(self) -> None: if not self._cleanup_allowed(): return assert self.downloads is not None self.downloads.schedule_pending_deletion_recovery() def _refresh_player_evidence( self, before_generation: int, effects: list[PlaybackEffect], ) -> None: observation = self.store.player_observation() if observation.generation == before_generation: return kinds = {effect.kind for effect in effects} identity_known = ( observation.active_slot is not None or observation.active_path is not None ) safe_evidence = "player_idle" in kinds or ( identity_known and bool( kinds & { "player_active", "player_busy", "player_reconnected_active", "player_telemetry", "control_issued", } ) ) if safe_evidence and observation.availability == "online": self._player_evidence_ready = True self._operator_cleanup_override = False self._cleanup_review_id = None self._schedule_deletion_recovery() return self._mark_cleanup_unknown() def _reset_cleanup_evidence(self) -> None: self._player_evidence_ready = False self._operator_cleanup_override = False self._cleanup_review_id = uuid.uuid4().hex def _mark_cleanup_unknown(self) -> None: self._player_evidence_ready = False self._operator_cleanup_override = False if self._cleanup_review_id is None: self._cleanup_review_id = uuid.uuid4().hex def _run_timers(self) -> None: assert self.playback is not None assert self.router is not None assert self.downloads is not None playback_timeout = self.playback.check_timeout() if playback_timeout is not None: self._handle_playback_effects([playback_timeout]) router_timeout = self.router.check_deadline() if router_timeout is not None: self._handle_router_decision(router_timeout) now = self._monotonic() if now >= self._next_reconciliation: self._handle_playback_effects(self.playback.mark_due_reconciliation()) self._next_reconciliation = now + 1.0 if now >= self._next_expiry and self._cleanup_allowed(): self._schedule_deletion_recovery() for token in self.downloads.schedule_expired_deletions(): self._announce_slot(token.slot, force=True) self._next_expiry = now + self.config.storage.expiry_interval_seconds