engine: pass stream URL to mpv; support modern mpv JSON-IPC

Fixes the 'mpv exited before playback could start' failure: play()
never appended the media URL to the spawn args, so mpv exited
immediately with rc=0 (no socket, no stderr) for every episode.

Real-world verification against mpv v0.41 revealed the engine also
assumed an older JSON-IPC dialect:
  * acks are {"request_id", "data", "error"} — no command_id/type
  * property renames: "position" -> "playback-time",
    "length" -> "duration"; no state property at all
  * "volume" is a 0-100 percentage, not 0.0-1.0
  * properties only answer once media metadata is loaded
  * select() requires an explicit xlist sequence, not None

Changes:
  - play(): URL appended to args; per-build property discovery
    (_PROP_CANDIDATES / _discover_property), re-run while
    _wait_ready() probes for builds whose properties appear late
  - _cmd(): understands both old and new ack shapes; rejects commands
    that error out
  - status(): derives playing/paused/buffering/eof from position
    progress when no state property exists (_derived_state)
  - stop(): uses the ["stop"] command
  - set_volume()/status() normalize the 0-1 <-> 0-100 volume scale
  - mpv stderr is captured to <tmpdir>/.asot-ipc-<pid>.log and referenced
    in error messages so future stream failures are diagnosable

Verified: real mpv playback of a live episode (position advancing 1:1,
correct duration), state save + resume on reopen. 35/35 unit tests green.

asot_player/engine.py

@@ -71,6 +71,35 @@ class Status:
     volume: float     # 0.0 – 1.0
 
 
+# Property names vary across mpv releases ("position" vs "playback-time",
+# "length" vs "duration"), and newer builds (>= ~0.41) have no state property
+# at all.  Candidates are tried in order once per process start; the first one
+# that answers with a usable value wins.
+_PROP_CANDIDATES: dict[str, tuple[str, ...]] = {
+    "state": ("playback-state", "state"),
+    "pos":   ("position", "playback-time", "time-pos"),
+    "dur":   ("length", "duration"),
+}
+
+
+def _as_float(x) -> float:
+    if x is None or isinstance(x, bool):
+        return -1.0
+    try:
+        return float(x)
+    except (TypeError, ValueError):
+        return -1.0
+
+
+def _as_bool(x) -> bool:
+    if isinstance(x, bool):
+        return x
+    try:
+        return bool(float(x))
+    except (TypeError, ValueError):
+        return False
+
+
 class MpEngine:
     """An mpv process controlled through its JSON IPC unix socket."""
 
@@ -81,11 +110,17 @@ class MpEngine:
         self.sock_path = sock_path or os.path.join(
             tempfile.gettempdir(), f".asot-ipc-{os.getpid()}.sock"
         )
+        self.mpv_log = os.path.join(
+            tempfile.gettempdir(), f".asot-ipc-{os.getpid()}.log"
+        )
+        self._errlog: "io.TextIOWrapper | None" = None
         self.proc: subprocess.Popen | None = None
         self._sock: socket.socket | None = None
         self._cid = 0
         self.url = ""
         self.last_event: dict = {}
+        self._prop_names: dict[str, str] = {}
+        self._adv: tuple[float, float] | None = None
 
     # ----------------------------------------------------------------------
 
@@ -100,7 +135,6 @@ class MpEngine:
             os.unlink(self.sock_path)
         except OSError:
             pass
-
         args = [
             self.bin,
             "--no-video",
@@ -108,17 +142,28 @@ class MpEngine:
             "--really-quiet",
             f"--input-ipc-server={self.sock_path}",
             "--loop=0",
+            url,
         ]
         if self.no_config:
             args.append("--no-config")
         self.url = url
+        # Keep mpv's own stderr for diagnostics; referenced from the error
+        # messages when the stream cannot be started.
+        self._errlog = open(self.mpv_log, "w")
         self.proc = subprocess.Popen(
             args,
             stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
+            stderr=self._errlog,
             stdin=subprocess.DEVNULL,
         )
         self._connect()
+        self._errlog.flush()
+        # Fresh mpv process → re-discover which property names this build
+        # speaks (newer mpv dropped "playback-state" and friends).
+        self._prop_names = {}
+        self._adv = None
+        for logical, cands in _PROP_CANDIDATES.items():
+            self._prop_names[logical] = self._discover_property(logical, cands)
         try:
             self.set_mute(muted)
             self.set_volume(volume)
@@ -141,7 +186,8 @@ class MpEngine:
                 rc = self.proc.returncode
                 raise EngineError(
                     f"mpv exited before playback could start "
-                    f"(return code {rc}) — the stream may be unavailable."
+                    f"(return code {rc}) — the stream may be unavailable "
+                    f"(see {self.mpv_log})."
                 )
             try:
                 self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
@@ -154,17 +200,34 @@ class MpEngine:
         raise EngineError("timed out waiting for mpv to create its IPC socket")
 
     def _wait_ready(self):
-        """Block until playback begins (or fails)."""
+        """Block until playback begins (or fails).  While waiting, re-probe
+        property names for this mpv build: newer builds only report some of
+        them once the media metadata has loaded, so the one-shot probe at
+        connect time can miss."""
         deadline = time.time() + 18.0
         while time.time() < deadline:
-            state = self._prop("playback-state")
-            if isinstance(state, str) and state in ("playing", "buffering", "paused", "eof"):
+            # Re-discover any property we still haven't found; candidates are
+            # retried until one answers or the deadline passes.
+            for logical, cands in _PROP_CANDIDATES.items():
+                if self._prop_names.get(logical) is None:
+                    self._prop_names[logical] = self._discover_property(
+                        logical, cands)
+            state_name = self._prop_names.get("state")
+            state = self._get(state_name) if state_name else None
+            if isinstance(state, str) and state in ("playing", "buffering",
+                                                   "paused", "eof"):
                 return
+            dur_name = self._prop_names.get("dur")
+            if dur_name:
+                d = _as_float(self._get(dur_name))
+                if d > 60.0:
+                    return
             if self.proc is not None and self.proc.poll() is not None:
                 rc = self.proc.returncode
                 raise EngineError(
                     f"mpv failed to load '{self.url}' (return code {rc}) — "
-                    "check your network or the stream URL."
+                    f"check your network or the stream URL "
+                    f"(see {self.mpv_log})."
                 )
             time.sleep(0.35)
         raise EngineError("timed out waiting for mpv to become ready")
@@ -172,37 +235,29 @@ class MpEngine:
     # ----------------------------------------------------------------------
 
     def status(self) -> Status:
-        state = self._prop("playback-state")
-        paused = self._prop("pause")
-        muted = self._prop("mute")
-        volume = self._prop("volume")
-        pos = self._prop("playback-time")
-        length = self._prop("length")
-
-        def _num(x) -> float:
-            if x is None or isinstance(x, bool):
-                return -1.0
-            try:
-                return float(x)
-            except (TypeError, ValueError):
-                return -1.0
-
-        def _bool(x) -> bool:
-            if isinstance(x, bool):
-                return x
-            try:
-                return bool(float(x))
-            except (TypeError, ValueError):
-                return False
-
-        state_s = state if isinstance(state, str) else "stop"
+        """Snapshot of playback state."""
+        state_name = self._prop_names.get("state")
+        state = self._get(state_name) if state_name else None
+        paused = _as_bool(self._get("pause"))
+        muted = _as_bool(self._get("mute"))
+        volume = _as_float(self._get("volume"))
+        # mpv reports "volume" as a 0-100 percentage; normalize to 0-1.
+        if volume > 0.0:
+            volume /= 100.0
+        pos_name = self._prop_names.get("pos")
+        dur_name = self._prop_names.get("dur")
+        pos = _as_float(self._get(pos_name)) if pos_name else -1.0
+        length = _as_float(self._get(dur_name)) if dur_name else -1.0
+
+        state_s = state if (isinstance(state, str) and state) \
+                    else self._derived_state(paused, pos, length)
         return Status(
-            position=_num(pos),
-            length=_num(length),
+            position=pos,
+            length=length,
             state=state_s,
-            paused=_bool(paused),
-            muted=_bool(muted),
-            volume=_num(volume),
+            paused=paused,
+            muted=muted,
+            volume=volume,
         )
 
     def seek(self, seconds: float):
@@ -216,12 +271,63 @@ class MpEngine:
         self._cmd(["set_property", "mute", muted])
 
     def set_volume(self, volume: float):
-        self._cmd(["set_property", "volume", max(0.0, min(1.0, float(volume)))])
+        """Set volume; mpv's ``volume`` property is a 0-100 percentage."""
+        pct = max(0.0, min(1.0, float(volume))) * 100.0
+        self._cmd(["set_property", "volume", round(pct, 2)])
 
     # ----------------------------------------------------------------------
+    # low-level helpers
+    # ----------------------------------------------------------------------
+
+    def _derived_state(self, paused: bool, pos: float, length: float) -> str:
+        """Infer playback state for mpv builds without a state property
+        (e.g. newer mpv >= 0.41).  Position normally advances ~1:1 with real
+        time while playing; a position that stays flat for a long stretch means
+        buffering or end of stream."""
+        now = time.time()
+        if paused:
+            return "paused"
+        if pos < 0.0:
+            return "buffering" if length > 0.0 else "playing"
+        if length > 10.0 and pos >= length - 2.0:
+            return "eof"
+        if self._adv is None:
+            self._adv = (pos, now)
+            return "playing"
+        last_pos, last_t = self._adv
+        advanced = pos - last_pos
+        if advanced > 0.5:               # real progress → update reference
+            self._adv = (pos, now)
+            return "playing"
+        stall = now - last_t             # seconds without real progress
+        if stall > 8.0:
+            return "eof"                 # no progress for a while → end of stream
+        if stall > 2.0:
+            return "buffering"
+        return "playing"
+
+    def _get(self, name: str):
+        """Get a property by name; None when this mpv build lacks it."""
+        try:
+            return self._cmd(["get_property", name])
+        except EngineError as e:
+            if "not found" in str(e) or "unavailable" in str(e):
+                return None
+            raise
 
-    def _prop(self, name: str):
-        return self._cmd(["get_property", name])
+    def _discover_property(self, logical: str,
+                           candidates: tuple[str, ...]) -> str | None:
+        """Try ``candidates`` in order; return the first property this mpv
+        build reports a usable value for (None if none work)."""
+        for cand in candidates:
+            d = self._get(cand)
+            if logical == "state":
+                if isinstance(d, str) and d:
+                    return cand
+            else:
+                if _as_float(d) != -1.0:
+                    return cand
+        return None
 
     def _cmd(self, cmd, timeout: float = 4.0):
         """Send one JSON command; return its ``data`` (raises on failure)."""
@@ -229,11 +335,12 @@ class MpEngine:
             raise EngineError("mpv process is dead")
         self._cid += 1
         cmd_id = self._cid
-        self._sock.sendall((json.dumps({"command": cmd, "command_id": cmd_id}) + "\n").encode())
+        self._sock.sendall(
+            (json.dumps({"command": cmd, "command_id": cmd_id}) + "\n").encode())
         deadline = time.time() + timeout
         while True:
             remaining = max(0.01, deadline - time.time())
-            rlist, _, _ = select.select([self._sock], [], remaining)
+            rlist, _, _ = select.select([self._sock], [], [], remaining)
             if rlist:
                 try:
                     data = self._sock.recv(65536)
@@ -245,10 +352,18 @@ class MpEngine:
                         except json.JSONDecodeError:
                             continue
                         if obj.get("event") is not None:
+                            # Unsolicited event — remember it; keep waiting.
                             self.last_event = obj
                         else:
-                            if obj.get("command_id") == cmd_id:
-                                return obj.get("data")
+                            # Ack.  Older mpv acks carry command/command_id/
+                            # type; newer builds (>= ~0.41) use
+                            # request_id/data/error instead.  Both shapes
+                            # carry the result under "data" and no "event".
+                            error = obj.get("error")
+                            if error and error != "success":
+                                raise EngineError(
+                                    f"mpv rejected {cmd!r}: {error}")
+                            return obj.get("data")
                 except (ConnectionResetError, OSError):
                     raise EngineError("mpv IPC socket closed unexpectedly")
             if time.time() >= deadline:
@@ -259,7 +374,7 @@ class MpEngine:
 
     def stop(self):
         try:
-            self._cmd(["set_property", "stop", True], timeout=1.5)
+            self._cmd(["stop"], timeout=1.5)
         except Exception:
             pass
         try:
@@ -279,6 +394,12 @@ class MpEngine:
             os.unlink(self.sock_path)
         except OSError:
             pass
+        try:
+            if self._errlog is not None:
+                self._errlog.close()
+        except OSError:
+            pass
+        self._errlog = None
         self.proc = None
         self._sock = None