feat: playlist shuffle, lazy resolution, TTS ducking, kept repair

Music:
- #random URL fragment shuffles playlist tracks before enqueuing
- Lazy playlist resolution: first 10 tracks resolve immediately,
  remaining are fetched in a background task
- !kept repair re-downloads kept tracks with missing local files
- !kept shows [MISSING] marker for tracks without local files
- TTS ducking: music ducks when merlin speaks via voice peer,
  smooth restore after TTS finishes

Performance (from profiling):
- Connection pool: preload_content=True for SOCKS connection reuse
- Pool tuning: 30 pools / 8 connections (up from 20/4)
- _PooledResponse wrapper for stdlib-compatible read interface
- Iterative _extract_videos (replace 51K-deep recursion with stack)
- proxy=False for local SearXNG

Voice + multi-bot:
- Per-bot voice config lookup ([<username>.voice] in TOML)
- Mute detection: skip duck silence when all users muted
- Autoplay shuffle deck (no repeats until full cycle)
- Seek clamp to track duration (prevent seek-past-end stall)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tests/test_http.py

@@ -203,11 +203,15 @@ class TestUrlopen:
         pool = MagicMock()
         resp = MagicMock()
         resp.status = 200
+        resp.data = b"ok"
+        resp.reason = "OK"
+        resp.headers = {}
         pool.request.return_value = resp
         mock_pool_fn.return_value = pool
 
         result = urlopen("https://example.com/")
-        assert result is resp
+        assert result.status == 200
+        assert result.read() == b"ok"
 
     @patch.object(derp.http, "_get_pool")
     def test_context_falls_back_to_opener(self, mock_pool_fn):

tests/test_music.py

@@ -563,6 +563,48 @@ class TestPlaylistExpansion:
         assert "list=PLxyz" in called_url
         assert len(tracks) == 2
 
+    def test_random_fragment_shuffles(self):
+        """#random fragment shuffles resolved playlist tracks."""
+        bot = _FakeBot()
+        msg = _Msg(text="!play https://example.com/playlist#random")
+        tracks = [(f"https://example.com/{i}", f"Track {i}") for i in range(20)]
+        with patch.object(_mod, "_resolve_tracks", return_value=list(tracks)) as mock_rt:
+            with patch.object(_mod, "_ensure_loop"):
+                asyncio.run(_mod.cmd_play(bot, msg))
+        # Fragment stripped before passing to resolver
+        called_url = mock_rt.call_args[0][0]
+        assert "#random" not in called_url
+        ps = _mod._ps(bot)
+        assert len(ps["queue"]) == 20
+        # Extremely unlikely (1/20!) that shuffle preserves exact order
+        titles = [t.title for t in ps["queue"]]
+        assert titles != [f"Track {i}" for i in range(20)] or len(titles) == 1
+        # Announces shuffle
+        assert any("shuffled" in r for r in bot.replied)
+
+    def test_random_fragment_single_track_no_error(self):
+        """#random on a single-video URL works fine (nothing to shuffle)."""
+        bot = _FakeBot()
+        msg = _Msg(text="!play https://example.com/video#random")
+        tracks = [("https://example.com/video", "Solo Track")]
+        with patch.object(_mod, "_resolve_tracks", return_value=tracks):
+            with patch.object(_mod, "_ensure_loop"):
+                asyncio.run(_mod.cmd_play(bot, msg))
+        ps = _mod._ps(bot)
+        assert len(ps["queue"]) == 1
+        assert ps["queue"][0].title == "Solo Track"
+
+    def test_random_fragment_ignored_for_search(self):
+        """#random is not treated specially for search queries."""
+        bot = _FakeBot()
+        msg = _Msg(text="!play jazz #random")
+        tracks = [("https://example.com/1", "Result")]
+        with patch.object(_mod, "_resolve_tracks", return_value=tracks) as mock_rt:
+            with patch.object(_mod, "_ensure_loop"):
+                asyncio.run(_mod.cmd_play(bot, msg))
+        # Search query passed as-is (not a URL, fragment not stripped)
+        assert mock_rt.call_args[0][0] == "ytsearch10:jazz #random"
+
     def test_resolve_tracks_error_fallback(self):
         """On error, returns [(url, url)]."""
         with patch("subprocess.run", side_effect=Exception("fail")):
@@ -577,6 +619,136 @@ class TestPlaylistExpansion:
             tracks = _mod._resolve_tracks("https://example.com/empty")
         assert tracks == [("https://example.com/empty", "https://example.com/empty")]
 
+    def test_resolve_tracks_start_param(self):
+        """start= passes --playlist-start to yt-dlp."""
+        result = MagicMock()
+        result.stdout = "https://example.com/6\nTrack 6\n"
+        with patch("subprocess.run", return_value=result) as mock_run:
+            tracks = _mod._resolve_tracks("https://example.com/pl",
+                                          max_tracks=5, start=6)
+        cmd = mock_run.call_args[0][0]
+        assert "--playlist-start=6" in cmd
+        assert "--playlist-end=10" in cmd
+        assert tracks == [("https://example.com/6", "Track 6")]
+
+    def test_resolve_tracks_start_empty_returns_empty(self):
+        """Paginated call with no results returns [] (not fallback)."""
+        result = MagicMock()
+        result.stdout = ""
+        with patch("subprocess.run", return_value=result):
+            tracks = _mod._resolve_tracks("https://example.com/pl",
+                                          start=100)
+        assert tracks == []
+
+    def test_resolve_tracks_start_error_returns_empty(self):
+        """Paginated call on error returns [] (not fallback)."""
+        with patch("subprocess.run", side_effect=Exception("fail")):
+            tracks = _mod._resolve_tracks("https://example.com/pl",
+                                          start=10)
+        assert tracks == []
+
+    def test_playlist_url_triggers_batched_resolve(self):
+        """Playlist URL resolves initial batch, spawns feeder for rest."""
+        bot = _FakeBot()
+        batch = _mod._PLAYLIST_BATCH
+        initial = [(f"https://example.com/{i}", f"T{i}")
+                    for i in range(batch)]
+        spawned = []
+        orig_spawn = bot._spawn
+
+        def spy_spawn(coro, *, name=None):
+            spawned.append(name)
+            return orig_spawn(coro, name=name)
+
+        bot._spawn = spy_spawn
+        msg = _Msg(text="!play https://example.com/watch?v=a&list=PLxyz")
+        with patch.object(_mod, "_resolve_tracks", return_value=initial):
+            with patch.object(_mod, "_ensure_loop"):
+                asyncio.run(_mod.cmd_play(bot, msg))
+        ps = _mod._ps(bot)
+        assert len(ps["queue"]) == batch
+        assert "music-playlist-feeder" in spawned
+        assert any("resolving more" in r.lower() for r in bot.replied)
+
+    def test_non_playlist_url_no_feeder(self):
+        """Single video URL does not spawn background feeder."""
+        bot = _FakeBot()
+        spawned = []
+        orig_spawn = bot._spawn
+
+        def spy_spawn(coro, *, name=None):
+            spawned.append(name)
+            return orig_spawn(coro, name=name)
+
+        bot._spawn = spy_spawn
+        tracks = [("https://example.com/v", "Video")]
+        msg = _Msg(text="!play https://example.com/v")
+        with patch.object(_mod, "_resolve_tracks", return_value=tracks):
+            with patch.object(_mod, "_ensure_loop"):
+                asyncio.run(_mod.cmd_play(bot, msg))
+        assert "music-playlist-feeder" not in spawned
+
+    def test_playlist_feeder_appends_to_queue(self):
+        """Background feeder resolves remaining tracks into queue."""
+        bot = _FakeBot()
+        ps = _mod._ps(bot)
+        remaining = [("https://example.com/6", "Track 6"),
+                     ("https://example.com/7", "Track 7")]
+
+        async def _check():
+            with patch.object(_mod, "_resolve_tracks",
+                              return_value=remaining):
+                await _mod._playlist_feeder(
+                    bot, "https://example.com/pl", 6, 10,
+                    False, "Alice", "https://example.com/pl",
+                )
+            assert len(ps["queue"]) == 2
+            assert ps["queue"][0].title == "Track 6"
+            assert ps["queue"][1].requester == "Alice"
+
+        asyncio.run(_check())
+
+    def test_playlist_feeder_shuffles(self):
+        """Background feeder shuffles when shuffle=True."""
+        bot = _FakeBot()
+        ps = _mod._ps(bot)
+        remaining = [(f"https://example.com/{i}", f"T{i}")
+                     for i in range(20)]
+
+        async def _check():
+            with patch.object(_mod, "_resolve_tracks",
+                              return_value=list(remaining)):
+                await _mod._playlist_feeder(
+                    bot, "https://example.com/pl", 6, 20,
+                    True, "Alice", "",
+                )
+            titles = [t.title for t in ps["queue"]]
+            assert len(titles) == 20
+            # Extremely unlikely shuffle preserves order
+            assert titles != [f"T{i}" for i in range(20)]
+
+        asyncio.run(_check())
+
+    def test_playlist_feeder_respects_queue_cap(self):
+        """Background feeder stops at _MAX_QUEUE."""
+        bot = _FakeBot()
+        ps = _mod._ps(bot)
+        # Pre-fill queue to near capacity
+        ps["queue"] = [_mod._Track(url="x", title="t", requester="a")
+                       for _ in range(_mod._MAX_QUEUE - 2)]
+        remaining = [(f"https://example.com/{i}", f"T{i}")
+                     for i in range(10)]
+
+        async def _check():
+            with patch.object(_mod, "_resolve_tracks",
+                              return_value=remaining):
+                await _mod._playlist_feeder(
+                    bot, "url", 6, 10, False, "a", "",
+                )
+            assert len(ps["queue"]) == _mod._MAX_QUEUE
+
+        asyncio.run(_check())
+
 
 # ---------------------------------------------------------------------------
 # TestResumeState
@@ -925,6 +1097,56 @@ class TestDuckMonitor:
                 pass
         asyncio.run(_check())
 
+    def test_tts_active_ducks(self):
+        """TTS activity from voice peer triggers ducking."""
+        bot = _FakeBot()
+        ps = _mod._ps(bot)
+        ps["duck_enabled"] = True
+        ps["duck_floor"] = 5
+        ps["duck_restore"] = 1  # fast restore for test
+        bot.registry._voice_ts = 0.0
+        bot.registry._tts_active = True
+
+        async def _check():
+            task = asyncio.create_task(_mod._duck_monitor(bot))
+            await asyncio.sleep(1.5)
+            assert ps["duck_vol"] == 5.0
+            # TTS ends -- restore should begin and complete quickly
+            bot.registry._tts_active = False
+            await asyncio.sleep(2.5)
+            assert ps["duck_vol"] is None
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        asyncio.run(_check())
+
+    def test_tts_active_overrides_all_muted(self):
+        """TTS ducks even when all users are muted."""
+        bot = _FakeBot()
+        ps = _mod._ps(bot)
+        ps["duck_enabled"] = True
+        ps["duck_floor"] = 5
+        bot.registry._voice_ts = time.monotonic()
+        bot.registry._tts_active = True
+        # Simulate all users muted
+        bot._mumble = MagicMock()
+        bot._mumble.users = {1: {"name": "human", "self_mute": True,
+                                  "mute": False, "self_deaf": False}}
+        bot.registry._bots = {}
+
+        async def _check():
+            task = asyncio.create_task(_mod._duck_monitor(bot))
+            await asyncio.sleep(1.5)
+            assert ps["duck_vol"] == 5.0
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        asyncio.run(_check())
+
 
 # ---------------------------------------------------------------------------
 # TestAutoResume
@@ -1109,33 +1331,21 @@ class TestAutoResume:
 
 
 class TestAutoplayKept:
-    def test_shuffles_kept_tracks(self, tmp_path):
-        """Autoplay loads kept tracks, shuffles, and starts playback."""
+    def test_starts_loop_with_kept_tracks(self, tmp_path):
+        """Autoplay starts play loop when kept tracks exist."""
         bot = _FakeBot()
         bot.registry._voice_ts = 0.0
         music_dir = tmp_path / "music"
         music_dir.mkdir()
-        # Create two kept files
         (music_dir / "a.opus").write_bytes(b"audio")
-        (music_dir / "b.opus").write_bytes(b"audio")
         bot.state.set("music", "keep:1", json.dumps({
             "url": "https://example.com/a", "title": "Track A",
             "filename": "a.opus", "id": 1,
         }))
-        bot.state.set("music", "keep:2", json.dumps({
-            "url": "https://example.com/b", "title": "Track B",
-            "filename": "b.opus", "id": 2,
-        }))
         with patch.object(_mod, "_MUSIC_DIR", music_dir), \
              patch.object(_mod, "_ensure_loop") as mock_loop:
             asyncio.run(_mod._autoplay_kept(bot))
             mock_loop.assert_called_once_with(bot)
-        ps = _mod._ps(bot)
-        assert len(ps["queue"]) == 2
-        titles = {t.title for t in ps["queue"]}
-        assert titles == {"Track A", "Track B"}
-        # All tracks marked keep=True
-        assert all(t.keep for t in ps["queue"])
 
     def test_skips_when_already_playing(self):
         bot = _FakeBot()
@@ -1429,6 +1639,20 @@ class TestKeptCommand:
         assert not list(music_dir.iterdir())
         assert bot.state.get("music", "keep:1") is None
 
+    def test_kept_shows_missing_marker(self, tmp_path):
+        """Tracks with missing files show [MISSING] in listing."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        bot.state.set("music", "keep:1", json.dumps({
+            "title": "Gone Track", "artist": "", "duration": 0,
+            "filename": "gone.opus", "id": 1,
+        }))
+        with patch.object(_mod, "_MUSIC_DIR", music_dir):
+            msg = _Msg(text="!kept")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("MISSING" in r for r in bot.replied)
+
     def test_kept_non_mumble(self):
         bot = _FakeBot(mumble=False)
         msg = _Msg(text="!kept")
@@ -1908,3 +2132,102 @@ class TestFetchMetadata:
         assert meta["title"] == ""
         assert meta["artist"] == ""
         assert meta["duration"] == 0
+
+
+# ---------------------------------------------------------------------------
+# TestKeptRepair
+# ---------------------------------------------------------------------------
+
+
+class TestKeptRepair:
+    def test_repair_nothing_missing(self, tmp_path):
+        """Repair reports all present when files exist."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        (music_dir / "song.opus").write_bytes(b"audio")
+        bot.state.set("music", "keep:1", json.dumps({
+            "url": "https://example.com/v", "title": "Song",
+            "filename": "song.opus", "id": 1,
+        }))
+        with patch.object(_mod, "_MUSIC_DIR", music_dir):
+            msg = _Msg(text="!kept repair")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("nothing to repair" in r.lower() for r in bot.replied)
+
+    def test_repair_downloads_missing(self, tmp_path):
+        """Repair re-downloads missing files."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        bot.state.set("music", "keep:1", json.dumps({
+            "url": "https://example.com/v", "title": "Song",
+            "filename": "song.opus", "id": 1,
+        }))
+
+        dl_path = tmp_path / "cache" / "dl.opus"
+        dl_path.parent.mkdir()
+        dl_path.write_bytes(b"audio")
+
+        with patch.object(_mod, "_MUSIC_DIR", music_dir), \
+             patch.object(_mod, "_download_track", return_value=dl_path):
+            msg = _Msg(text="!kept repair")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("1 restored" in r for r in bot.replied)
+        assert (music_dir / "song.opus").is_file()
+
+    def test_repair_counts_failures(self, tmp_path):
+        """Repair reports failed downloads."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        bot.state.set("music", "keep:1", json.dumps({
+            "url": "https://example.com/v", "title": "Song",
+            "filename": "song.opus", "id": 1,
+        }))
+
+        with patch.object(_mod, "_MUSIC_DIR", music_dir), \
+             patch.object(_mod, "_download_track", return_value=None):
+            msg = _Msg(text="!kept repair")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("1 failed" in r for r in bot.replied)
+
+    def test_repair_no_url_skips(self, tmp_path):
+        """Repair skips entries with no URL."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        bot.state.set("music", "keep:1", json.dumps({
+            "url": "", "title": "No URL",
+            "filename": "nourl.opus", "id": 1,
+        }))
+
+        with patch.object(_mod, "_MUSIC_DIR", music_dir):
+            msg = _Msg(text="!kept repair")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("1 failed" in r for r in bot.replied)
+
+    def test_repair_extension_mismatch(self, tmp_path):
+        """Repair updates metadata when download extension differs."""
+        bot = _FakeBot()
+        music_dir = tmp_path / "music"
+        music_dir.mkdir()
+        bot.state.set("music", "keep:1", json.dumps({
+            "url": "https://example.com/v", "title": "Song",
+            "filename": "song.opus", "id": 1,
+        }))
+
+        dl_path = tmp_path / "cache" / "dl.webm"
+        dl_path.parent.mkdir()
+        dl_path.write_bytes(b"audio")
+
+        with patch.object(_mod, "_MUSIC_DIR", music_dir), \
+             patch.object(_mod, "_download_track", return_value=dl_path):
+            msg = _Msg(text="!kept repair")
+            asyncio.run(_mod.cmd_kept(bot, msg))
+        assert any("1 restored" in r for r in bot.replied)
+        # Filename updated to new extension
+        raw = bot.state.get("music", "keep:1")
+        stored = json.loads(raw)
+        assert stored["filename"] == "song.webm"
+        assert (music_dir / "song.webm").is_file()