wf(wf-52ref): review R1 — 2 issues fixed (self-join guard, worker_crashed coverage)

- stop() skips self-join when called from worker thread (deadlock guard)
- remove pragma; test refresh.worker_crashed guard + re-entrant stop

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: usmanabbas7

src/convert_sdk/config_loader/refresh.py

@@ -125,11 +125,22 @@ def start(self) -> None:
         thread.start()
 
     def stop(self, *, timeout: float = 5.0) -> None:
-        """Signal the worker to stop and join the thread (idempotent)."""
+        """Signal the worker to stop and join the thread (idempotent).
+
+        Safe to call re-entrantly from the worker thread itself (e.g. if a host's
+        terminal-failure callback calls ``Core.close()``): a thread cannot join
+        itself, so the self-join is skipped — the daemon loop observes the
+        ``_stopping`` flag and exits on its own. The thread is a daemon, so even
+        an un-joined worker never blocks interpreter exit.
+        """
         self._stopping.set()
         self._wake.set()
         thread = self._thread
-        if thread is not None and thread.is_alive():
+        if (
+            thread is not None
+            and thread.is_alive()
+            and thread is not threading.current_thread()
+        ):
             thread.join(timeout=timeout)
         self._thread = None
 
@@ -177,7 +188,7 @@ def _run(self) -> None:
                 break
             try:
                 self._do_refresh()
-            except Exception:  # pragma: no cover - defended; see Task 3 guard
+            except Exception:
                 # Outer resilience guard. _do_refresh handles its own transport
                 # failures (Task 3); reaching here means a logging/clock/callback
                 # subsystem failure escaped. Emit and keep the worker alive.

tests/test_config_refresh.py

@@ -353,6 +353,61 @@ def test_failure_does_not_crash_core_host(self) -> None:
             core.close()
 
 
+class TestWorkerResilience:
+    def test_swap_callback_exception_does_not_kill_worker(self) -> None:
+        """An error escaping _do_refresh (e.g. swap callback) is caught and the
+        worker survives (refresh.worker_crashed guard)."""
+        from convert_sdk.config_loader.refresh import ConfigRefresher
+
+        def _boom(_snapshot: Any) -> None:
+            raise RuntimeError("swap blew up")
+
+        transport = _FakeTransport([_config_payload(), _config_payload()])
+        refresher = ConfigRefresher(
+            config=_remote_config(RefreshConfig(interval_seconds=300)),
+            transport=transport,
+            on_snapshot=_boom,
+        )
+        refresher.start()
+        try:
+            refresher.trigger_now()
+            assert refresher.wait_for_next_refresh(timeout=5.0)
+            # Worker absorbed the callback failure and is still running.
+            assert refresher.is_alive()
+            # And it can still complete a subsequent cycle.
+            refresher.trigger_now()
+            assert refresher.wait_for_next_refresh(timeout=5.0)
+        finally:
+            refresher.stop()
+
+    def test_stop_is_safe_from_worker_thread(self) -> None:
+        """Calling stop() re-entrantly from the worker thread must not deadlock."""
+        from convert_sdk.config_loader.refresh import ConfigRefresher
+        from convert_sdk.errors import ConfigLoadError
+
+        policy = RefreshConfig(interval_seconds=300, backoff_max_seconds=300)
+        holder: Dict[str, Any] = {}
+
+        def _terminal(_exc: BaseException) -> None:
+            # Re-entrant stop from inside the worker thread.
+            holder["refresher"].stop()
+
+        transport = _FakeTransport([ConfigLoadError("boom")])
+        refresher = ConfigRefresher(
+            config=_remote_config(policy),
+            transport=transport,
+            on_snapshot=lambda _s: None,
+            on_terminal_failure=_terminal,
+        )
+        holder["refresher"] = refresher
+        refresher.start()
+        refresher.trigger_now()
+        # If self-join were attempted this would hang; the seam must resolve.
+        assert refresher.wait_for_next_refresh(timeout=5.0)
+        refresher.stop()
+        assert not refresher.is_alive()
+
+
 class TestConfigUpdatedEvent:
     def test_config_updated_emitted_on_successful_swap(self) -> None:
         from convert_sdk import LifecycleEvent