feat: overhaul video player controller

controller/downloads.py

@@ -90,6 +90,7 @@ class DownloadManager:
         self._lock = threading.Lock()
         self._futures: set[Future[Any]] = set()
         self._processes: set[subprocess.Popen[str]] = set()
+        self._deletions_in_flight: set[tuple[int, int]] = set()
         self._stopping = False
 
     def enqueue(
@@ -484,21 +485,43 @@ class DownloadManager:
         slot_directory = canonical_slot_directory(self.config.storage.media_base, token.slot)
         trash_root = self.config.storage.media_base / ".controller-trash"
         trash_path: Path | None = None
+        if self._slot_is_guarded(token.slot):
+            return False
         if os.path.lexists(slot_directory):
             if slot_directory.is_symlink():
                 slot_directory.unlink()
             else:
                 trash_root.mkdir(parents=True, exist_ok=True)
+                self._fsync_directory(self.config.storage.media_base)
                 trash_path = trash_root / (
                     f"slot{token.slot}.generation{token.generation}.{uuid.uuid4().hex}"
                 )
                 os.replace(slot_directory, trash_path)
                 self._fsync_directory(self.config.storage.media_base)
+                self._fsync_directory(trash_root)
+        if self._slot_is_guarded(token.slot):
+            if (
+                trash_path is not None
+                and not os.path.lexists(slot_directory)
+                and os.path.lexists(trash_path)
+            ):
+                os.replace(trash_path, slot_directory)
+                self._fsync_directory(self.config.storage.media_base)
+                self._fsync_directory(trash_root)
+            return False
         if trash_path is not None:
             shutil.rmtree(trash_path)
+            self._fsync_directory(trash_root)
         return self.store.finish_delete(token)
 
+    def _slot_is_guarded(self, slot: int) -> bool:
+        return any(
+            guard.slot == slot for guard in self.store.list_live_guards()
+        )
+
     def expire_due_slots(self) -> list[int]:
+        """Synchronously expire slots for maintenance callers and tests."""
+
         deleted: list[int] = []
         for token in self.store.begin_expired_deletions():
             try:
@@ -508,47 +531,164 @@ class DownloadManager:
                 LOG.exception("could not finish expired deletion for slot %s", token.slot)
         return deleted
 
+    def schedule_delete_slot(
+        self,
+        slot: int,
+        *,
+        reason: str,
+        expected_url: str | None = None,
+    ) -> DeleteToken:
+        """Atomically fence a manual deletion, then finish it off-thread."""
+
+        token = self.store.begin_delete(
+            slot,
+            reason=reason,
+            expected_url=expected_url,
+        )
+        self._schedule_deletion(token)
+        return token
+
+    def schedule_expired_deletions(self) -> list[DeleteToken]:
+        """Fence all currently due slots and queue filesystem work."""
+
+        tokens = self.store.begin_expired_deletions()
+        for token in tokens:
+            self._schedule_deletion(token)
+        return tokens
+
     def recover_pending_deletions(self) -> list[int]:
         recovered: list[int] = []
-        trash_root = self.config.storage.media_base / ".controller-trash"
         for token in self.store.pending_deletions():
-            if any(
-                guard.slot == token.slot
-                for guard in self.store.list_live_guards()
-            ):
-                LOG.warning(
-                    "deferring interrupted deletion for protected slot %s",
-                    token.slot,
-                )
-                continue
-            if trash_root.is_dir():
-                prefix = f"slot{token.slot}.generation{token.generation}."
-                for candidate in trash_root.iterdir():
-                    if candidate.name.startswith(prefix):
-                        if candidate.is_symlink() or candidate.is_file():
-                            candidate.unlink()
-                        elif candidate.is_dir():
-                            shutil.rmtree(candidate)
-            if self.finish_deletion(token):
+            if self._recover_deletion(token):
                 recovered.append(token.slot)
         return recovered
 
     def schedule_pending_deletion_recovery(self) -> bool:
-        """Run recoverable trash cleanup without delaying service startup."""
+        """Queue every safe pending deletion without blocking the coordinator."""
 
         with self._lock:
             if self._stopping:
                 return False
+        scheduled = False
+        for token in self.store.pending_deletions():
+            scheduled = self._schedule_deletion(token) or scheduled
+        return scheduled
+
+    def _schedule_deletion(self, token: DeleteToken) -> bool:
+        key = (token.slot, token.generation)
+        with self._lock:
+            if self._stopping or key in self._deletions_in_flight:
+                return False
+            self._deletions_in_flight.add(key)
         try:
-            future = self._executor.submit(self.recover_pending_deletions)
-        except BaseException:
-            LOG.exception("could not schedule interrupted deletion recovery")
+            future = self._executor.submit(self._deletion_worker, token)
+        except BaseException as exc:
+            with self._lock:
+                self._deletions_in_flight.discard(key)
+            self._emit(
+                DownloadEvent(
+                    slot=token.slot,
+                    state="delete_error",
+                    message=f"could not start deletion worker: {exc}",
+                    generation=token.generation,
+                )
+            )
             return False
         with self._lock:
             self._futures.add(future)
         future.add_done_callback(self._forget_future)
         return True
 
+    def _deletion_worker(self, token: DeleteToken) -> None:
+        key = (token.slot, token.generation)
+        try:
+            if self._recover_deletion(token):
+                self._emit(
+                    DownloadEvent(
+                        slot=token.slot,
+                        state="deleted",
+                        generation=token.generation,
+                    )
+                )
+            elif self.store.get_slot(token.slot).media_state == "deleting":
+                self._emit(
+                    DownloadEvent(
+                        slot=token.slot,
+                        state="delete_deferred",
+                        message="slot deletion was deferred by playback protection",
+                        generation=token.generation,
+                    )
+                )
+        except BaseException as exc:
+            LOG.exception(
+                "could not finish deletion for slot %s generation %s",
+                token.slot,
+                token.generation,
+            )
+            self._emit(
+                DownloadEvent(
+                    slot=token.slot,
+                    state="delete_error",
+                    message=f"deletion remains pending: {exc}",
+                    generation=token.generation,
+                )
+            )
+        finally:
+            with self._lock:
+                self._deletions_in_flight.discard(key)
+
+    def _recover_deletion(self, token: DeleteToken) -> bool:
+        trash_root = self.config.storage.media_base / ".controller-trash"
+        slot_directory = canonical_slot_directory(
+            self.config.storage.media_base,
+            token.slot,
+        )
+        if self._slot_is_guarded(token.slot):
+            LOG.warning(
+                "deferring interrupted deletion for protected slot %s",
+                token.slot,
+            )
+            self._restore_guarded_trash(token, trash_root, slot_directory)
+            return False
+        if trash_root.is_dir():
+            prefix = f"slot{token.slot}.generation{token.generation}."
+            for candidate in trash_root.iterdir():
+                if not candidate.name.startswith(prefix):
+                    continue
+                if self._slot_is_guarded(token.slot):
+                    self._restore_guarded_trash(
+                        token,
+                        trash_root,
+                        slot_directory,
+                    )
+                    return False
+                if candidate.is_symlink() or candidate.is_file():
+                    candidate.unlink()
+                elif candidate.is_dir():
+                    shutil.rmtree(candidate)
+                self._fsync_directory(trash_root)
+        return self.finish_deletion(token)
+
+    def _restore_guarded_trash(
+        self,
+        token: DeleteToken,
+        trash_root: Path,
+        slot_directory: Path,
+    ) -> None:
+        if not trash_root.is_dir() or os.path.lexists(slot_directory):
+            return
+        prefix = f"slot{token.slot}.generation{token.generation}."
+        candidates = sorted(
+            candidate
+            for candidate in trash_root.iterdir()
+            if candidate.name.startswith(prefix)
+        )
+        if not candidates:
+            return
+        os.replace(candidates[0], slot_directory)
+        self._fsync_directory(self.config.storage.media_base)
+        self._fsync_directory(trash_root)
+
     def shutdown(self, *, wait: bool = True) -> None:
         with self._lock:
             self._stopping = True

controller/playback.py

@@ -18,6 +18,7 @@ from .store import ControllerStore
 TERMINAL_REASONS = frozenset(
     {"stop", "finished", "mpv_error", "sigterm", "sigint", "shutdown"}
 )
+TERMINAL_ERROR_REASONS = frozenset({"startup_error", "worker_start_error"})
 CONTROL_ACTIONS = frozenset({"play", "stop", "pause", "resume", "toggle"})
 
 
@@ -485,13 +486,31 @@ class PlaybackReducer:
             return effects
 
         if state == "busy":
-            if slot is not None or path is not None:
+            previous = self.store.player_observation()
+            if (
+                slot is not None
+                or path is not None
+                or previous.active_slot is not None
+                or previous.active_path is not None
+            ):
                 self._active_evidence(
                     slot=slot,
                     path=path,
                     paused=None,
                     status=payload,
                 )
+            else:
+                self.store.set_player_observation(
+                    availability="online",
+                    phase="unknown",
+                    active_slot=None,
+                    active_path=None,
+                    paused=None,
+                    reconciliation_state="unknown",
+                    reconciliation_due_at=None,
+                    reconciliation_id=None,
+                    status_payload=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"))
@@ -499,40 +518,87 @@ class PlaybackReducer:
 
         if state == "error":
             previous = self.store.player_observation()
+            message = str(payload.get("message") or "Player error")
+            reason = payload.get("reason")
+            if reason in TERMINAL_ERROR_REASONS:
+                terminal_slot = slot if slot is not None else previous.active_slot
+                self._idle_evidence(status=payload)
+                effects.extend(
+                    (
+                        PlaybackEffect(
+                            "player_error",
+                            slot=terminal_slot,
+                            message=message,
+                        ),
+                        PlaybackEffect(
+                            "player_idle",
+                            slot=terminal_slot,
+                            message=str(reason),
+                        ),
+                    )
+                )
+                if self.pending is not None:
+                    effects.append(self._finish_operation("failed", message))
+                return effects
+
             active_slot = slot if slot is not None else previous.active_slot
             active_path = path if path is not None else previous.active_path
+            due = None
+            if self.config.storage.uncertain_review_after_seconds > 0:
+                due = (
+                    self._wall_clock()
+                    + self.config.storage.uncertain_review_after_seconds
+                )
             if active_slot is not None:
-                self.store.reconcile_active_guard(
+                guard = self.store.mark_slot_uncertain(
                     active_slot,
-                    kind="player_active",
-                    details={"error": payload.get("message")},
+                    review_after=due,
+                    details={"reason": "Player error", "message": message},
+                )
+                due = guard.review_after
+                self.store.set_player_observation(
+                    availability="online",
+                    phase="error",
+                    active_slot=active_slot,
+                    active_path=active_path,
+                    paused=(
+                        previous.paused
+                        if self._same_playback_identity(
+                            previous,
+                            active_slot,
+                            active_path,
+                        )
+                        else None
+                    ),
+                    reconciliation_state=(
+                        "operator_review"
+                        if guard.state == "review_required"
+                        else "awaiting_evidence"
+                    ),
+                    reconciliation_due_at=due,
+                    reconciliation_id=guard.guard_id,
+                    status_payload=payload,
                 )
             elif active_path is not None:
-                self.store.reconcile_external_active()
-            self.store.set_player_observation(
-                availability="online",
-                phase="error",
-                active_slot=active_slot,
-                active_path=active_path,
-                paused=(
-                    previous.paused
-                    if self._same_playback_identity(
-                        previous,
-                        active_slot,
-                        active_path,
-                    )
-                    else None
-                ),
-                reconciliation_state=(
-                    "confirmed"
-                    if active_slot is not None or active_path is not None
-                    else "unknown"
-                ),
-                reconciliation_due_at=None,
-                reconciliation_id=None,
-                status_payload=payload,
-            )
-            message = str(payload.get("message") or "Player error")
+                self.store.mark_external_uncertain(
+                    active_path,
+                    review_after=due,
+                    availability="online",
+                    details={"reason": "Player error", "message": message},
+                    status_payload=payload,
+                )
+            else:
+                self.store.set_player_observation(
+                    availability="online",
+                    phase="unknown",
+                    active_slot=None,
+                    active_path=None,
+                    paused=None,
+                    reconciliation_state="unknown",
+                    reconciliation_due_at=None,
+                    reconciliation_id=None,
+                    status_payload=payload,
+                )
             effects.append(PlaybackEffect("player_error", slot=active_slot, message=message))
             if self.pending is not None:
                 effects.append(self._finish_operation("failed", message))
@@ -645,6 +711,19 @@ class PlaybackReducer:
                         "reason": "stop timeout",
                     },
                 )
+        elif operation.action in {"pause", "resume", "toggle"}:
+            previous = self.store.player_observation()
+            if previous.active_slot is not None or previous.active_path is not None:
+                self.store.set_player_observation(
+                    availability=previous.availability,
+                    phase="unknown",
+                    active_slot=previous.active_slot,
+                    active_path=previous.active_path,
+                    paused=None,
+                    reconciliation_state=previous.reconciliation_state,
+                    reconciliation_due_at=previous.reconciliation_due_at,
+                    reconciliation_id=previous.reconciliation_id,
+                )
         return self._finish_operation("timed_out", "no authoritative Player confirmation")
 
     def _mark_play_uncertain(self, operation: PlayerOperation, reason: str) -> None:
@@ -753,8 +832,6 @@ class PlaybackReducer:
 
     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":
@@ -771,9 +848,9 @@ class PlaybackReducer:
         )
         if guard is not None:
             if guard.kind == "pending_play":
-                return "ready"
+                return record.media_state
             return "operator_review" if guard.state == "review_required" else "unknown"
-        return "ready"
+        return record.media_state
 
     def mark_due_reconciliation(self) -> list[PlaybackEffect]:
         now = self._wall_clock()

controller/service.py

@@ -99,7 +99,7 @@ class ControllerService:
         self._processed_packets: dict[tuple[int, str, int], float] = {}
         self._player_evidence_ready = False
         self._operator_cleanup_override = False
-        self._recovery_scheduled = False
+        self._cleanup_review_id: str | None = None
 
     @property
     def started(self) -> bool:
@@ -149,9 +149,7 @@ class ControllerService:
                     )
                 # Cleanup is deliberately deferred until fresh Player evidence
                 # (or an explicit administrative reconciliation) makes it safe.
-                self._player_evidence_ready = False
-                self._operator_cleanup_override = False
-                self._recovery_scheduled = False
+                self._reset_cleanup_evidence()
                 self._stop_event.clear()
                 self._next_expiry = (
                     self._monotonic()
@@ -271,8 +269,13 @@ class ControllerService:
             guards = self.store.list_live_guards() if started else []
             player_operation_pending = bool(
                 started
-                and self.playback is not None
-                and self.playback.pending is not None
+                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] = []
@@ -321,6 +324,22 @@ class ControllerService:
                     "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"
@@ -346,6 +365,7 @@ class ControllerService:
                 }
             ),
             "reconciliation_reviews": reviews,
+            "cleanup_reconciliation": cleanup_reconciliation,
             "degraded_reasons": degraded,
         }
 
@@ -416,16 +436,14 @@ class ControllerService:
                 handled_successfully = True
                 return
             if event.kind == "connected":
-                self._player_evidence_ready = False
-                self._operator_cleanup_override = False
+                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._player_evidence_ready = False
-                self._operator_cleanup_override = False
+                self._reset_cleanup_evidence()
                 assert self.playback is not None
                 assert self.router is not None
                 playback_effect = self.playback.mark_transport_uncertain(
@@ -441,30 +459,41 @@ class ControllerService:
                 handled_successfully = True
                 return
             if event.topic == self.config.topics.player_status:
-                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
-                    self._handle_playback_effects(self.playback.handle_status(payload))
-                    self._refresh_player_evidence(before_generation)
+                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:
-                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
-                    self._handle_playback_effects(self.playback.handle_telemetry(payload))
-                    self._refresh_player_evidence(before_generation)
+                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:
-                try:
-                    flat_state = event.payload.decode("utf-8").strip()
-                except UnicodeDecodeError:
-                    flat_state = ""
-                # Retained flat idle is sufficient only as a quiescence gate;
-                # it never mutates detailed playback state or clears guards.
-                self._player_evidence_ready = flat_state == "idle"
-                self._operator_cleanup_override = False
-                if self._player_evidence_ready:
-                    self._schedule_deletion_recovery()
+                # 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(
@@ -511,8 +540,9 @@ class ControllerService:
                     else:
                         decision = self.router.observe_mode(mode)
                     self._handle_router_decision(decision)
-            # Flat state is only a cold-start cleanup gate. Detailed playback
-            # truth continues to come exclusively from status and telemetry.
+            # 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:
@@ -873,17 +903,20 @@ class ControllerService:
         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.delete_slot(
+        self.downloads.schedule_delete_slot(
             slot,
             reason="manual",
             expected_url=expected_url,
         )
         self._announce_slot(slot, force=True)
-        return {"status": "deleted", "slot": slot}
+        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
@@ -899,7 +932,36 @@ class ControllerService:
             raise ValueError(
                 "playback reconciliation cannot be cleared while a Player operation is pending"
             )
-        if slot is None:
+        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(),
@@ -933,7 +995,9 @@ class ControllerService:
             and observation.active_slot is None
             and observation.active_path is None
         ):
-            self._operator_cleanup_override = True
+            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(
@@ -954,6 +1018,8 @@ class ControllerService:
             "status": "cleared",
             "review_id": review_id.strip(),
         }
+        if scope == "cleanup":
+            result["scope"] = "cleanup"
         if slot is not None:
             result["slot"] = slot
         return result
@@ -996,7 +1062,11 @@ class ControllerService:
         no_active_player = (
             observation.active_slot is None and observation.active_path is None
         )
-        if terminal or (failed_play and no_active_player):
+        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:
@@ -1028,6 +1098,27 @@ class ControllerService:
         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)
@@ -1171,21 +1262,55 @@ class ControllerService:
         return self._player_evidence_ready or self._operator_cleanup_override
 
     def _schedule_deletion_recovery(self) -> None:
-        if self._recovery_scheduled or not self._cleanup_allowed():
+        if not self._cleanup_allowed():
             return
         assert self.downloads is not None
-        self._recovery_scheduled = (
-            self.downloads.schedule_pending_deletion_recovery()
-        )
+        self.downloads.schedule_pending_deletion_recovery()
 
-    def _refresh_player_evidence(self, before_generation: int) -> None:
+    def _refresh_player_evidence(
+        self,
+        before_generation: int,
+        effects: list[PlaybackEffect],
+    ) -> None:
         observation = self.store.player_observation()
         if observation.generation == before_generation:
             return
-        self._operator_cleanup_override = False
-        self._player_evidence_ready = observation.availability == "online"
-        if self._player_evidence_ready and observation.phase == "idle":
+        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
@@ -1203,6 +1328,7 @@ class ControllerService:
             self._handle_playback_effects(self.playback.mark_due_reconciliation())
             self._next_reconciliation = now + 1.0
         if now >= self._next_expiry and self._cleanup_allowed():
-            for slot in self.downloads.expire_due_slots():
-                self._announce_slot(slot, force=True)
+            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

controller/store.py

@@ -704,7 +704,16 @@ class ControllerStore:
         title = str(legacy.get("title") or "")
         count = _coerce_count(legacy.get("item_count", 0), slot, warnings)
         created_at = _utc_timestamp(legacy.get("created_at"), f"slot {slot} created_at", warnings)
-        expires_at = _utc_timestamp(legacy.get("expires_at"), f"slot {slot} expires_at", warnings)
+        expires_at = _utc_timestamp(
+            legacy.get("expires_at"),
+            f"slot {slot} expires_at",
+            warnings,
+        )
+        if media_state == "error" and expires_at is None:
+            expires_at = now + self.config.retention_seconds
+            warnings.append(
+                f"slot {slot}: missing error expiry defaulted to retention window"
+            )
         generation = 1
         connection.execute(
             """
@@ -816,10 +825,17 @@ class ControllerStore:
                 connection.execute(
                     """
                     UPDATE slots
-                    SET media_state = 'error', last_error = ?, updated_at = ?
+                    SET media_state = 'error', last_error = ?,
+                        expires_at = COALESCE(expires_at, ?), updated_at = ?
                     WHERE slot_number = ? AND generation = ?
                     """,
-                    (message, now, row["slot_number"], row["generation"]),
+                    (
+                        message,
+                        now + self.config.retention_seconds,
+                        now,
+                        row["slot_number"],
+                        row["generation"],
+                    ),
                 )
 
     def list_slots(self) -> list[SlotRecord]:
@@ -1060,10 +1076,17 @@ class ControllerStore:
             connection.execute(
                 """
                 UPDATE slots
-                SET media_state = 'error', last_error = ?, updated_at = ?
+                SET media_state = 'error', last_error = ?,
+                    expires_at = COALESCE(expires_at, ?), updated_at = ?
                 WHERE slot_number = ? AND generation = ?
                 """,
-                (error, now, slot, generation),
+                (
+                    error,
+                    now + self.config.retention_seconds,
+                    now,
+                    slot,
+                    generation,
+                ),
             )
             return True