Add 5 test additions JohnBaumb suggested

Pinning gaps he flagged after his second pass:

- register_plugin when set_engine() raises: registration must succeed
  + plugin stays in the registry (download() raises later, surfacing
  the error to the user via download_with_fallback). Pin so a future
  refactor can't accidentally propagate the set_engine exception and
  crash boot.
- engine.get_all_downloads exclude actually doesn't invoke the plugin:
  ID-only check would pass even if soulseek's get_all was called and
  returned []; sentinel proves the plugin isn't touched at all.
- Cancel mid-flight observable from inside _download_sync: existing
  tests pin Cancelled-preserve AFTER impl returns, this pins the
  contract plugins rely on (engine.get_record reflecting Cancelled
  state during the impl thread's polling loop).
- configured_clients() with broken is_configured(): the try/except
  guard exists but had no test — broken plugin is silently skipped,
  healthy ones still surface.
- Per-source delay independence: YouTube's 3s rate-limit delay must
  not block a Tidal download starting in parallel. Companion to the
  per-source-locks test.

tests/downloads/test_background_download_worker.py

@@ -406,6 +406,108 @@ def test_semaphore_concurrency_can_be_increased():
 # ---------------------------------------------------------------------------
 
 
+def test_impl_can_observe_cancel_mid_flight_via_state_check():
+    """Per JohnBaumb: existing tests cover Cancelled-preserve AFTER impl
+    returns — but plugins also poll engine state mid-download (via
+    ``_is_cancelled`` helpers) to abort partial transfers. Pin that
+    contract: a cancel landing while impl is mid-flight must be
+    visible to a subsequent ``engine.get_record()`` from the impl
+    thread.
+    """
+    engine = DownloadEngine()
+    impl_started = threading.Event()
+    impl_can_finish = threading.Event()
+    observed_state_during_impl = []
+
+    def impl(download_id, target_id, display_name):
+        impl_started.set()
+        # Wait for the test thread to write Cancelled, then check
+        # what we can observe from inside the impl callback.
+        impl_can_finish.wait(timeout=2.0)
+        record = engine.get_record('youtube', download_id)
+        observed_state_during_impl.append(record.get('state') if record else None)
+        return None  # impl noticed cancel + bailed out
+
+    download_id = engine.worker.dispatch(
+        source_name='youtube',
+        target_id='vid',
+        display_name='X',
+        original_filename='vid||X',
+        impl_callable=impl,
+    )
+
+    # Wait for impl to start, then inject a cancel.
+    impl_started.wait(timeout=1.0)
+    engine.update_record('youtube', download_id, {'state': 'Cancelled'})
+    impl_can_finish.set()
+
+    # Wait for impl to finish (its append populates observed_state).
+    # Engine state is already Cancelled at this point, so polling on
+    # state would race: it'd break before impl ran the get_record line.
+    deadline = time.time() + 2.0
+    while not observed_state_during_impl and time.time() < deadline:
+        time.sleep(0.01)
+
+    # impl observed the Cancelled state mid-flight via get_record,
+    # AND the worker preserved Cancelled after impl returned None.
+    assert observed_state_during_impl == ['Cancelled']
+    assert engine.get_record('youtube', download_id)['state'] == 'Cancelled'
+
+
+def test_per_source_delays_dont_block_other_sources():
+    """Per JohnBaumb: per-source semaphores + delays must not let one
+    slow source stall another. YouTube's 3s rate-limit delay should
+    not delay a Tidal download starting in parallel.
+
+    Configure YouTube with a 0.5s delay, dispatch one YouTube download
+    (which holds the source's serial slot + arms the next-call delay),
+    then immediately dispatch a Tidal download. Tidal must complete
+    well before YouTube's delay window would have elapsed.
+    """
+    engine = DownloadEngine()
+    engine.worker.set_delay('youtube', 0.5)
+
+    yt_completed = threading.Event()
+    td_completed = threading.Event()
+
+    def yt_impl(download_id, target_id, display_name):
+        time.sleep(0.05)
+        yt_completed.set()
+        return '/tmp/yt.mp3'
+
+    def td_impl(download_id, target_id, display_name):
+        td_completed.set()
+        return '/tmp/td.flac'
+
+    # First YouTube call. After it finishes, the worker arms the
+    # 0.5s delay BEFORE the next youtube dispatch can run.
+    engine.worker.dispatch(
+        source_name='youtube', target_id='a', display_name='A',
+        original_filename='a||A', impl_callable=yt_impl,
+    )
+    yt_completed.wait(timeout=1.0)
+
+    # Second YouTube would now block 0.5s on the rate-limit delay.
+    # Dispatch one to occupy that wait, then dispatch Tidal — Tidal
+    # must NOT wait on YouTube's delay.
+    engine.worker.dispatch(
+        source_name='youtube', target_id='b', display_name='B',
+        original_filename='b||B', impl_callable=lambda *a: '/tmp/y.mp3',
+    )
+
+    td_start = time.time()
+    engine.worker.dispatch(
+        source_name='tidal', target_id='t1', display_name='T',
+        original_filename='t1||T', impl_callable=td_impl,
+    )
+    td_completed.wait(timeout=0.4)
+    td_elapsed = time.time() - td_start
+
+    # Tidal must have finished in well under 0.5s (the YouTube delay).
+    assert td_completed.is_set(), "Tidal blocked on YouTube's per-source delay"
+    assert td_elapsed < 0.4, f"Tidal took {td_elapsed:.2f}s — should be near-instant"
+
+
 def test_delay_enforces_minimum_gap_between_downloads():
     """Pinning: YouTube uses 3s delay today (legacy
     `_download_delay`). Worker-driven delay must enforce the same

tests/downloads/test_download_engine.py

@@ -34,6 +34,29 @@ def test_get_plugin_returns_none_for_unknown_source():
     assert engine.get_plugin('made_up') is None
 
 
+def test_register_plugin_swallows_set_engine_failure(caplog):
+    """Per JohnBaumb: if a plugin's ``set_engine`` callback raises,
+    registration shouldn't take down the engine — the failure gets
+    logged + the plugin stays registered. The plugin's own download()
+    method is responsible for surfacing the missing-engine state to
+    the user (see ``test_download_raises_when_engine_not_wired`` per
+    source). Pinning this so a future refactor can't accidentally
+    propagate the set_engine exception and crash boot.
+    """
+    class _BrokenSetEngine:
+        def set_engine(self, engine):
+            raise RuntimeError("plugin's set_engine blew up")
+
+    engine = DownloadEngine()
+    plugin = _BrokenSetEngine()
+    # Should not raise.
+    engine.register_plugin('flaky', plugin)
+
+    # Plugin still registered + lookupable.
+    assert engine.get_plugin('flaky') is plugin
+    assert 'flaky' in engine.registered_sources()
+
+
 def test_register_plugin_overwrites_on_duplicate(caplog):
     """Re-registering under the same name overwrites and warns. Not a
     common path but useful so test fixtures that build a fresh engine
@@ -320,6 +343,36 @@ def test_engine_get_all_downloads_aggregates_across_plugins():
     assert {r.id for r in result} == {'yt-1', 'td-1', 'td-2'}
 
 
+def test_engine_get_all_downloads_excludes_dont_invoke_plugin():
+    """Per JohnBaumb: the monitor calls engine.get_all_downloads(
+    exclude=('soulseek',)) AFTER already pulling slskd transfers via
+    the transfers/downloads endpoint. The whole point of the exclude
+    is to NOT touch soulseek's plugin a second time — so the plugin's
+    get_all_downloads must literally not be called when its name is
+    in the exclude list. Pin that semantic explicitly (a test that
+    just checks IDs would pass even if soulseek was called and
+    returned [])."""
+    engine = DownloadEngine()
+    sl_plugin = _FakePlugin('soulseek', downloads=[_FakeStatus('sl-1', 'soulseek')])
+    yt_plugin = _FakePlugin('youtube', downloads=[_FakeStatus('yt-1', 'youtube')])
+    engine.register_plugin('soulseek', sl_plugin)
+    engine.register_plugin('youtube', yt_plugin)
+
+    # Sentinel — flips True if get_all_downloads is called on soulseek.
+    soulseek_called = []
+    original_get_all = sl_plugin.get_all_downloads
+    async def _tracking_get_all():
+        soulseek_called.append(True)
+        return await original_get_all()
+    sl_plugin.get_all_downloads = _tracking_get_all
+
+    _run_async(engine.get_all_downloads(exclude=('soulseek',)))
+    assert soulseek_called == [], (
+        "soulseek's get_all_downloads was invoked despite being in exclude — "
+        "monitor would still double-fetch slskd"
+    )
+
+
 def test_engine_get_all_downloads_skips_excluded_sources():
     """Per JohnBaumb: monitor pulls slskd transfers via the
     transfers/downloads endpoint earlier in its loop, so engine

tests/downloads/test_download_orchestrator.py

@@ -127,6 +127,27 @@ def test_configured_clients_excludes_unconfigured_sources():
     assert result['soulseek'] is configured
 
 
+def test_configured_clients_skips_clients_whose_is_configured_raises():
+    """Per JohnBaumb: configured_clients() has a try/except so a single
+    broken is_configured() call doesn't crash the whole iteration —
+    pin it so a future refactor can't quietly drop the guard. The
+    broken plugin is skipped; the rest still come back."""
+
+    class _BrokenIsConfigured(_FakeClient):
+        def is_configured(self):
+            raise RuntimeError("is_configured blew up")
+
+    broken = _BrokenIsConfigured()
+    healthy = _FakeClient(configured=True)
+    orch = _build_orchestrator(soulseek=healthy, youtube=broken)
+
+    result = orch.configured_clients()
+    # Healthy plugin still surfaces; broken one is silently skipped.
+    assert 'soulseek' in result
+    assert result['soulseek'] is healthy
+    assert 'youtube' not in result
+
+
 def test_reload_instances_dispatches_to_named_source():
     """Generic dispatch — caller passes source name instead of
     reaching for orch.hifi.reload_instances() directly."""