engine/main: never crash when mpv dies mid-playback

Resuming deep into an episode (e.g. 618 at 2:00:19) put playback at the
very end, where mpv exits at EOF; its IPC socket closed between the
alive() check and sendall, so a raw BrokenPipeError — not an EngineError —
escaped status() and killed the player with a traceback.

- _cmd: wrap sendall in OSError -> EngineError carrying mpv's exit code and
  the mpv log path.
- _status no longer sys.exit(2)s; the main loop and key handler own it.
- Player.recover(): restart the same episode at the saved position, and
  advance to the next episode only when the position is within 30 s of the
  true end.  The loop budgets 5 consecutive drops with a 1 s pause so a
  persistently dead stream ends with a message instead of a restart storm.

Tests: 47 (socket-death -> EngineError, status(), recover restart/advance/
failure paths).  Verified live by killing mpv mid-playback: 6 "Recovered"
frames, playback continued from the saved position, no traceback.

asot_player/engine.py

@@ -342,8 +342,17 @@ 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())
+        try:
+            self._sock.sendall(
+                (json.dumps({"command": cmd, "command_id": cmd_id}) + "\n").encode())
+        except OSError as e:
+            # mpv died between the alive() check and this write: its IPC
+            # server closed the socket.  Must surface as EngineError — a raw
+            # BrokenPipeError here crashes the whole player.
+            rc = self.proc.poll() if self.proc is not None else "?"
+            raise EngineError(
+                f"mpv IPC socket closed (mpv exit code {rc}) — the stream "
+                f"dropped or mpv exited (see {self.mpv_log})") from e
         deadline = time.time() + timeout
         while True:
             remaining = max(0.01, deadline - time.time())

asot_player/main.py

@@ -186,14 +186,17 @@ class Player:
         self.msg_until = time.time() + 3.0
 
     def _status(self):
-        try:
-            return self.engine.status()
-        except EngineError as e:
-            print(f"[asot] engine error: {e}", file=sys.stderr)
-            sys.exit(2)
+        """Current engine status.  Raises EngineError when mpv is gone — the
+        main loop and key handler recover from that, they never crash."""
+        return self.engine.status()
 
     def handle_key(self, key):
-        st = self._status()
+        try:
+            st = self._status()
+        except EngineError:
+            # mpv died; the keypress that uncovered it is dropped
+            self.recover()
+            return
         if key in ("q", "\x03"):             # q or Ctrl-C (cbreak mode)
             self.finished = True
         elif key in (" ", "t"):              # play / pause
@@ -223,6 +226,28 @@ class Player:
         except EngineError:
             pass
 
+    def recover(self) -> bool:
+        """Restart playback after mpv died mid-episode (stream drop, network
+        loss, mpv reaching EOF).  Returns True when playback resumed (or the
+        episode truly ended and we advanced to the next one)."""
+        if self.ep is None:
+            return False
+        st = self.state
+        if st.length > 10.0 and st.position >= st.length - 30.0:
+            return self.advance(1)          # the episode really ended
+        try:
+            self.engine.play(
+                self.ep.url,
+                seek_to=max(0.0, float(st.position)),
+                muted=bool(st.muted),
+                volume=max(0.0, min(1.0, float(st.volume))))
+        except EngineError:
+            return False
+        st.paused = False
+        self.msg_override = f"Recovered \u2192 ASOT {self.ep.number:04d}"
+        self.msg_until = time.time() + 3.0
+        return True
+
     def render_once(self, st):
         now = time.time()
         if self.msg_override and now < self.msg_until:
@@ -325,6 +350,7 @@ def main():
                 seek_to=resume_pos if resume else 0.0)
 
         tick = 0
+        drops = 0                       # consecutive mpv deaths recovered from
         last_save = time.time()
         while not p.finished and tick < 50_000:
             t0 = time.time()
@@ -336,7 +362,23 @@ def main():
                     break
                 p.handle_key(k)
 
-            st = p._status()
+            try:
+                st = p._status()
+                drops = 0                      # healthy tick resets the budget
+            except EngineError as e:
+                # mpv died (stream drop / EOF race / network loss).  Do not
+                # crash: resume from the saved position, end only when the
+                # episode has truly finished.
+                drops += 1
+                print(f"[asot] playback dropped ({e.__class__.__name__}); "
+                      f"recovery attempt {drops}…", file=sys.stderr)
+                if drops > 5 or not p.recover():
+                    p.msg_override = "Playback failed — stream unavailable"
+                    p.msg_until = time.time() + 6.0
+                    break
+                tick += 1                      # keep the global tick cap binding
+                time.sleep(1.0)                # don't hammer a dead stream
+                continue
 
             # eof → next episode (auto-advance)
             if st.state == "eof":

tests/test_asot.py

@@ -392,6 +392,92 @@ class MpDerivedStateTests(unittest.TestCase):
             engine_mod.time.time = real
 
 
+# --------------------------------------------------------------------------
+# mpv socket death: a raw BrokenPipeError from sendall used to escape as a
+# plain OSError and crash the player mid-playback on a remote machine.  It
+# must surface as EngineError and the player must recover, not die.
+# --------------------------------------------------------------------------
+
+def _dead_mpv_engine(poll_rc=None):
+    """MpEngine whose mpv has died: alive() passes, socket writes fail.
+    poll_rc=None models the race that matters: poll() still reports the
+    process alive while its IPC server has already closed the socket."""
+    eng = MpEngine.__new__(MpEngine)        # skip __init__: no mpv spawn
+    eng.mpv_bin = "mpv"
+    eng.mpv_log = "/tmp/.asot-ipc-99.log"
+    eng.proc = Mock()
+    eng.proc.poll.return_value = poll_rc
+    eng._cid = 0
+    eng._prop_names = {}
+    eng._adv = None
+    eng._sock = Mock()
+    eng._sock.sendall.side_effect = BrokenPipeError(32, "Broken pipe")
+    return eng
+
+
+class MpSocketDeathTests(unittest.TestCase):
+    def test_cmd_surfaces_engine_error(self):
+        eng = _dead_mpv_engine()
+        with self.assertRaises(engine_mod.EngineError) as cm:
+            eng._cmd(["get_property", "pause"])
+        msg = str(cm.exception)
+        self.assertIn("socket closed", msg)
+        self.assertIn("asot-ipc-99.log", msg)      # and where to look
+
+    def test_dead_process_is_reported_without_socket_write(self):
+        eng = _dead_mpv_engine(poll_rc=1)
+        with self.assertRaises(engine_mod.EngineError) as cm:
+            eng._cmd(["get_property", "pause"])
+        self.assertIn("mpv process is dead", str(cm.exception))
+
+    def test_status_surfaces_engine_error(self):
+        # alive() passes (poll → None), the write itself fails — the exact
+        # shape of the reported crash traceback
+        eng = _dead_mpv_engine()
+        with self.assertRaises(engine_mod.EngineError):
+            eng.status()
+
+
+class PlayerRecoverTests(unittest.TestCase):
+    def setUp(self):
+        self._fetch = main_mod.fetch_meta
+        main_mod.fetch_meta = lambda url: Meta()   # no network in tests
+        self.eps = fake_eps()
+
+    def tearDown(self):
+        main_mod.fetch_meta = self._fetch
+
+    def _player(self, ep, position, length):
+        engine = Mock()
+        engine.alive.return_value = False          # _load: restore saved vol
+        p = Player(self.eps, engine, Mock(), State())
+        p.ep = ep
+        p.state.ep, p.state.url = ep.number, ep.url
+        p.state.position, p.state.length = position, length
+        return p
+
+    def test_recover_restarts_same_episode_at_saved_position(self):
+        p = self._player(self.eps[1], 3_000.0, 5_400.0)   # mid-episode
+        self.assertTrue(p.recover())
+        self.assertEqual(p.ep.number, 1300)               # same episode
+        self.assertEqual(p.engine.play.call_args.kwargs["seek_to"], 3_000.0)
+
+    def test_recover_advances_only_at_true_end(self):
+        p = self._player(self.eps[1], 5_380.0, 5_400.0)   # within 30 s of end
+        self.assertTrue(p.recover())
+        self.assertEqual(p.ep.number, 1301)               # advanced
+        self.assertEqual(p.engine.play.call_args.kwargs["seek_to"], 0.0)
+
+    def test_recover_reports_failure_when_restart_fails(self):
+        p = self._player(self.eps[1], 1_000.0, 5_400.0)
+        p.engine.play.side_effect = engine_mod.EngineError("mpv gone")
+        self.assertFalse(p.recover())
+
+    def test_recover_without_episode_is_noop(self):
+        p = Player(self.eps, Mock(), Mock(), State())
+        self.assertFalse(p.recover())
+
+
 # --------------------------------------------------------------------------
 # Keyboard restore: Linux termios has no TCSU — using it silently skipped the
 # restore (AttributeError swallowed), leaving the shell in cbreak mode.