From e4cba8ed38c806f5596170408541865dee081a31 Mon Sep 17 00:00:00 2001
From: Andres Lopez <190146319+AndresL230@users.noreply.github.com>
Date: Wed, 13 May 2026 03:35:33 -0400
Subject: [PATCH 1/3] fix(interceptor): guard install/uninstall with RLock

Two threads racing install() could both observe _installed=False and
both run the patch path, with the second wrapping the first. A later
uninstall() would unwrap only one layer, leaving a recost wrapper in
place even though _installed=False.

Add a module-level RLock around install/uninstall/is_installed so the
patched-method state cannot drift away from the _installed flag.

Adds a regression test that reliably reproduced the wrapper-leak on
the unfixed code.

Refs #4
---
 recost/_interceptor.py    |  40 +++++++++------
 tests/test_interceptor.py | 103 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 128 insertions(+), 15 deletions(-)

diff --git a/recost/_interceptor.py b/recost/_interceptor.py
index 0f2b6ea..4bcaa03 100644
--- a/recost/_interceptor.py
+++ b/recost/_interceptor.py
@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import contextvars
+import threading
 import time
 from datetime import datetime, timezone
 from typing import Callable, Optional
@@ -30,6 +31,12 @@
 _installed: bool = False
 _callback: Optional[EventCallback] = None
 
+# Guards install / uninstall / is_installed so the patched-method state
+# cannot drift away from the _installed flag under concurrent callers.
+# RLock so the same thread can re-enter (e.g. from a callback that
+# triggers reinstall). See issue #4.
+_install_lock: threading.RLock = threading.RLock()
+
 # Original function references — restored on uninstall
 _original_urllib3_urlopen = None
 _original_httpx_send = None
@@ -372,29 +379,32 @@ def _unpatch_aiohttp() -> None:
 def install(callback: EventCallback) -> None:
     """Install patches on urllib3, httpx, and aiohttp. No-op if already installed."""
     global _installed, _callback
-    if _installed:
-        return
+    with _install_lock:
+        if _installed:
+            return
 
-    _callback = callback
-    _patch_urllib3()
-    _patch_httpx()
-    _patch_aiohttp()
-    _installed = True
+        _callback = callback
+        _patch_urllib3()
+        _patch_httpx()
+        _patch_aiohttp()
+        _installed = True
 
 
 def uninstall() -> None:
     """Restore all patched functions to their originals. No-op if not installed."""
     global _installed, _callback
-    if not _installed:
-        return
+    with _install_lock:
+        if not _installed:
+            return
 
-    _unpatch_urllib3()
-    _unpatch_httpx()
-    _unpatch_aiohttp()
-    _callback = None
-    _installed = False
+        _unpatch_urllib3()
+        _unpatch_httpx()
+        _unpatch_aiohttp()
+        _callback = None
+        _installed = False
 
 
 def is_installed() -> bool:
     """Returns True if patches are currently active."""
-    return _installed
+    with _install_lock:
+        return _installed
diff --git a/tests/test_interceptor.py b/tests/test_interceptor.py
index 7d0e1a6..e7e3158 100644
--- a/tests/test_interceptor.py
+++ b/tests/test_interceptor.py
@@ -188,3 +188,106 @@ def test_uninstall_restores_originals(self, test_server):
         import requests
         requests.get(f"{test_server}/after-uninstall")
         assert len(events) == 0
+
+
+# ---------------------------------------------------------------------------
+# Thread safety — install/uninstall race
+# ---------------------------------------------------------------------------
+
+import sys
+import threading
+
+
+class TestInstallUninstallRace:
+    """Regression tests for issue #4 — install/uninstall must be safe under
+    concurrent calls so the patch state cannot drift away from the
+    ``_installed`` flag."""
+
+    def test_concurrent_install_uninstall_leaves_consistent_state(self):
+        """N threads each loop install/uninstall many times. After all
+        threads finish, the interceptor must be uninstalled and no exception
+        should have been raised."""
+        from recost._interceptor import install, uninstall, is_installed
+
+        iterations_per_thread = 200
+        num_threads = 4
+        exceptions: list[BaseException] = []
+
+        def _noop(_event) -> None:
+            pass
+
+        def worker() -> None:
+            try:
+                for _ in range(iterations_per_thread):
+                    install(_noop)
+                    uninstall()
+            except BaseException as exc:  # noqa: BLE001
+                exceptions.append(exc)
+
+        # Force aggressive GIL yields so the race fires reliably on modern
+        # CPython. Restored in finally.
+        original_switch = sys.getswitchinterval()
+        sys.setswitchinterval(0.000001)
+        try:
+            threads = [threading.Thread(target=worker) for _ in range(num_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=30.0)
+        finally:
+            sys.setswitchinterval(original_switch)
+
+        assert not exceptions, f"unexpected exceptions: {exceptions!r}"
+        assert not is_installed(), "interceptor should be uninstalled after the race"
+
+    def test_concurrent_install_uninstall_does_not_leak_patches(self):
+        """Pre-fix: two threads can both enter ``install()`` before either has
+        set ``_installed = True``, both patch the target, and the second
+        wraps the first. A subsequent ``uninstall()`` unwraps only the
+        outer layer — the inner wrapper persists even after ``_installed``
+        is False. This test verifies the original urllib3 callable is
+        restored after a concurrent install/uninstall race."""
+        try:
+            import urllib3
+        except ImportError:
+            import pytest
+
+            pytest.skip("urllib3 not installed")
+
+        from recost._interceptor import install, uninstall, is_installed
+
+        original_urlopen = urllib3.HTTPConnectionPool.urlopen
+
+        def _noop(_event) -> None:
+            pass
+
+        iterations_per_thread = 200
+        num_threads = 4
+        exceptions: list[BaseException] = []
+
+        def worker() -> None:
+            try:
+                for _ in range(iterations_per_thread):
+                    install(_noop)
+                    uninstall()
+            except BaseException as exc:  # noqa: BLE001
+                exceptions.append(exc)
+
+        original_switch = sys.getswitchinterval()
+        sys.setswitchinterval(0.000001)
+        try:
+            threads = [threading.Thread(target=worker) for _ in range(num_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=30.0)
+        finally:
+            sys.setswitchinterval(original_switch)
+
+        assert not exceptions, f"unexpected exceptions: {exceptions!r}"
+        assert not is_installed(), "interceptor should be uninstalled"
+        # The critical invariant: urllib3's method is back to its real
+        # original, not a recost wrapper.
+        assert urllib3.HTTPConnectionPool.urlopen is original_urlopen, (
+            "concurrent install/uninstall left a recost wrapper in place"
+        )

From c622198b88d4b1d8826bd1976c16da6d53f0932b Mon Sep 17 00:00:00 2001
From: Andres Lopez <190146319+AndresL230@users.noreply.github.com>
Date: Wed, 13 May 2026 03:58:38 -0400
Subject: [PATCH 2/3] fix(init): guard init/dispose with RLock
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two threads racing init() could both observe _handle=None and both
proceed past the dispose-previous guard, with the first thread's
handle orphaned (its timer thread + transport leak) once the second
overwrites _handle.

Add a module-level RLock around init() and RecostHandle.dispose() so
_handle mutation and the install/uninstall calls they make stay
atomic against each other.

The lock nests safely with the interceptor's _install_lock — _init_lock
is always acquired first, then _install_lock via the transitive
install()/uninstall() call.

Adds a regression test that reliably reproduced the orphan-handle leak
on the unfixed code (17 of 80 handles orphaned in a typical run).

Tests use RecostConfig(enabled=False) so init/dispose pairs do not
spin up transport/timer threads — the lock invariant under test does
not depend on those resources being live, and avoiding them keeps the
test under one second instead of four minutes.

Refs #4
---
 recost/_init.py    | 292 +++++++++++++++++++++++----------------------
 tests/test_init.py | 113 ++++++++++++++++++
 2 files changed, 263 insertions(+), 142 deletions(-)

diff --git a/recost/_init.py b/recost/_init.py
index 4567250..67a4ece 100644
--- a/recost/_init.py
+++ b/recost/_init.py
@@ -51,32 +51,39 @@ def dispose(self) -> None:
         the final flush settles or its timeout elapses, so an in-flight
         cloud POST is not cut off mid-request.
         """
-        if self._disposed:
-            return
-        self._disposed = True
+        with _init_lock:
+            if self._disposed:
+                return
+            self._disposed = True
 
-        self._timer_stop.set()
-        if self._timer_thread is not None:
-            self._timer_thread.join(timeout=5.0)
+            self._timer_stop.set()
+            if self._timer_thread is not None:
+                self._timer_thread.join(timeout=5.0)
 
-        if self._final_flush is not None:
-            flush_thread = threading.Thread(target=self._final_flush, daemon=True)
-            flush_thread.start()
-            flush_thread.join(timeout=self._shutdown_flush_timeout_ms / 1000.0)
+            if self._final_flush is not None:
+                flush_thread = threading.Thread(target=self._final_flush, daemon=True)
+                flush_thread.start()
+                flush_thread.join(timeout=self._shutdown_flush_timeout_ms / 1000.0)
 
-        uninstall()
+            uninstall()
 
-        if self._transport is not None:
-            self._transport.dispose()
+            if self._transport is not None:
+                self._transport.dispose()
 
-        global _handle
-        if _handle is self:
-            _handle = None
+            global _handle
+            if _handle is self:
+                _handle = None
 
 
 # Module-level handle so a second init() call disposes the first.
 _handle: Optional[RecostHandle] = None
 
+# Guards init() and dispose() so the _handle global cannot become
+# inconsistent under concurrent callers. RLock so a wrapping caller
+# (e.g. dispose() running on the timer thread which itself owns the
+# init lock) does not deadlock. See issue #4.
+_init_lock: threading.RLock = threading.RLock()
+
 
 def init(config: Optional[RecostConfig] = None) -> RecostHandle:
     """
@@ -87,112 +94,127 @@ def init(config: Optional[RecostConfig] = None) -> RecostHandle:
     - Returns a handle with a dispose() method for explicit cleanup.
     """
     global _handle
-    if _handle is not None:
-        _handle.dispose()
-
-    config = config or RecostConfig()
-
-    # Resolve flush interval: prefer the new ms-based field, but if a caller
-    # still passes the legacy seconds-based flush_interval, honor it with a
-    # deprecation warning so existing code keeps working until they migrate.
-    if config.flush_interval is not None:
-        warnings.warn(
-            "flush_interval is deprecated, use flush_interval_ms instead",
-            DeprecationWarning,
-            stacklevel=2,
-        )
-        flush_interval_seconds = float(config.flush_interval)
-    else:
-        flush_interval_seconds = config.flush_interval_ms / 1000.0
-
-    if not config.enabled:
-        stop_event = threading.Event()
-        stop_event.set()
-        noop = RecostHandle(timer_stop=stop_event, timer_thread=None, transport=None)
-        _handle = noop
-        return noop
-
-    registry = ProviderRegistry(config.custom_providers or None)
-    aggregator = Aggregator(
-        project_id=config.project_id or "",
-        environment=config.environment,
-        sdk_version="0.1.0",
-        max_buckets=config.max_buckets,
-    )
-    transport = Transport(config)
-    debug = config.debug
-    max_batch_size = config.max_batch_size
-
-    # Build the set of URL substrings to exclude from tracking.
-    exclude_patterns = list(config.exclude_patterns)
-    if config.api_key:
-        exclude_patterns.append(config.base_url.rstrip("/"))
-    else:
-        exclude_patterns.append(f"127.0.0.1:{config.local_port}")
-        exclude_patterns.append(f"localhost:{config.local_port}")
-
-    def flush_and_send() -> None:
-        summary = aggregator.flush()
-        if summary is None:
-            return
-        if debug:
-            print(
-                f"[recost] flush: {len(summary.metrics)} metric group(s), "
-                f"window {summary.window_start} → {summary.window_end}",
-                file=sys.stderr,
+    with _init_lock:
+        if _handle is not None:
+            _handle.dispose()
+
+        config = config or RecostConfig()
+
+        # Resolve flush interval: prefer the new ms-based field, but if a caller
+        # still passes the legacy seconds-based flush_interval, honor it with a
+        # deprecation warning so existing code keeps working until they migrate.
+        if config.flush_interval is not None:
+            warnings.warn(
+                "flush_interval is deprecated, use flush_interval_ms instead",
+                DeprecationWarning,
+                stacklevel=2,
             )
-        transport.send(summary)
-
-    def on_event(event: RawEvent) -> None:
-        # Drop excluded URLs
-        for pattern in exclude_patterns:
-            if pattern in event.url or pattern in event.host:
+            flush_interval_seconds = float(config.flush_interval)
+        else:
+            flush_interval_seconds = config.flush_interval_ms / 1000.0
+
+        if not config.enabled:
+            stop_event = threading.Event()
+            stop_event.set()
+            noop = RecostHandle(timer_stop=stop_event, timer_thread=None, transport=None)
+            _handle = noop
+            return noop
+
+        registry = ProviderRegistry(config.custom_providers or None)
+        aggregator = Aggregator(
+            project_id=config.project_id or "",
+            environment=config.environment,
+            sdk_version="0.1.0",
+            max_buckets=config.max_buckets,
+        )
+        transport = Transport(config)
+        debug = config.debug
+        max_batch_size = config.max_batch_size
+
+        # Build the set of URL substrings to exclude from tracking.
+        exclude_patterns = list(config.exclude_patterns)
+        if config.api_key:
+            exclude_patterns.append(config.base_url.rstrip("/"))
+        else:
+            exclude_patterns.append(f"127.0.0.1:{config.local_port}")
+            exclude_patterns.append(f"localhost:{config.local_port}")
+
+        def flush_and_send() -> None:
+            summary = aggregator.flush()
+            if summary is None:
                 return
+            if debug:
+                print(
+                    f"[recost] flush: {len(summary.metrics)} metric group(s), "
+                    f"window {summary.window_start} → {summary.window_end}",
+                    file=sys.stderr,
+                )
+            transport.send(summary)
+
+        def on_event(event: RawEvent) -> None:
+            # Drop excluded URLs
+            for pattern in exclude_patterns:
+                if pattern in event.url or pattern in event.host:
+                    return
+
+            # Enrich with provider/endpoint from the registry
+            match = registry.match(event.url)
+            if match is not None:
+                event.provider = match.provider
+                event.endpoint_category = match.endpoint_category
+
+            if debug:
+                print(
+                    f"[recost] captured {event.method} {event.url} "
+                    f"{event.status_code} ({event.latency_ms}ms)",
+                    file=sys.stderr,
+                )
+
+            # If this event would push us past the bucket cap, flush the current
+            # window first so it's preserved, then ingest into a fresh window.
+            if aggregator.would_overflow(event):
+                try:
+                    flush_and_send()
+                except Exception as err:
+                    if config.on_error:
+                        config.on_error(err)
+                    elif debug:
+                        print(f"[recost] flush error: {err}", file=sys.stderr)
+
+            cost = match.cost_per_request_cents if match is not None else 0.0
+            aggregator.ingest(event, cost)
+
+            # Trigger an early flush if the batch size threshold is reached
+            if aggregator.size >= max_batch_size:
+                try:
+                    flush_and_send()
+                except Exception as err:
+                    if config.on_error:
+                        config.on_error(err)
+                    elif debug:
+                        print(f"[recost] flush error: {err}", file=sys.stderr)
+
+        install(on_event)
+
+        # Flush timer using a background thread with Event-based stopping
+        stop_event = threading.Event()
 
-        # Enrich with provider/endpoint from the registry
-        match = registry.match(event.url)
-        if match is not None:
-            event.provider = match.provider
-            event.endpoint_category = match.endpoint_category
-
-        if debug:
-            print(
-                f"[recost] captured {event.method} {event.url} "
-                f"{event.status_code} ({event.latency_ms}ms)",
-                file=sys.stderr,
-            )
-
-        # If this event would push us past the bucket cap, flush the current
-        # window first so it's preserved, then ingest into a fresh window.
-        if aggregator.would_overflow(event):
-            try:
-                flush_and_send()
-            except Exception as err:
-                if config.on_error:
-                    config.on_error(err)
-                elif debug:
-                    print(f"[recost] flush error: {err}", file=sys.stderr)
-
-        cost = match.cost_per_request_cents if match is not None else 0.0
-        aggregator.ingest(event, cost)
-
-        # Trigger an early flush if the batch size threshold is reached
-        if aggregator.size >= max_batch_size:
-            try:
-                flush_and_send()
-            except Exception as err:
-                if config.on_error:
-                    config.on_error(err)
-                elif debug:
-                    print(f"[recost] flush error: {err}", file=sys.stderr)
-
-    install(on_event)
-
-    # Flush timer using a background thread with Event-based stopping
-    stop_event = threading.Event()
-
-    def _timer_loop() -> None:
-        while not stop_event.wait(timeout=flush_interval_seconds):
+        def _timer_loop() -> None:
+            while not stop_event.wait(timeout=flush_interval_seconds):
+                try:
+                    flush_and_send()
+                except Exception as err:
+                    if config.on_error:
+                        config.on_error(err)
+                    elif debug:
+                        print(f"[recost] flush error: {err}", file=sys.stderr)
+
+        timer_thread = threading.Thread(target=_timer_loop, daemon=True)
+        timer_thread.start()
+
+        def _final_flush() -> None:
+            # Errors during the final flush are logged / forwarded the same way
+            # as a normal tick — we never want dispose() to surface them.
             try:
                 flush_and_send()
             except Exception as err:
@@ -201,26 +223,12 @@ def _timer_loop() -> None:
                 elif debug:
                     print(f"[recost] flush error: {err}", file=sys.stderr)
 
-    timer_thread = threading.Thread(target=_timer_loop, daemon=True)
-    timer_thread.start()
-
-    def _final_flush() -> None:
-        # Errors during the final flush are logged / forwarded the same way
-        # as a normal tick — we never want dispose() to surface them.
-        try:
-            flush_and_send()
-        except Exception as err:
-            if config.on_error:
-                config.on_error(err)
-            elif debug:
-                print(f"[recost] flush error: {err}", file=sys.stderr)
-
-    handle = RecostHandle(
-        timer_stop=stop_event,
-        timer_thread=timer_thread,
-        transport=transport,
-        final_flush=_final_flush,
-        shutdown_flush_timeout_ms=config.shutdown_flush_timeout_ms,
-    )
-    _handle = handle
-    return handle
+        handle = RecostHandle(
+            timer_stop=stop_event,
+            timer_thread=timer_thread,
+            transport=transport,
+            final_flush=_final_flush,
+            shutdown_flush_timeout_ms=config.shutdown_flush_timeout_ms,
+        )
+        _handle = handle
+        return handle
diff --git a/tests/test_init.py b/tests/test_init.py
index 1c58dc7..78476da 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -72,3 +72,116 @@ def test_local_mode_excludes_localhost(self):
         ))
         assert is_installed()
         handle.dispose()
+
+
+# ---------------------------------------------------------------------------
+# Thread safety — init/dispose race
+# ---------------------------------------------------------------------------
+
+import sys
+
+
+class TestInitDisposeRace:
+    """Regression tests for issue #4 — concurrent init() / dispose() must
+    not orphan handles or leave the SDK in an inconsistent state."""
+
+    def _drain_module_state(self) -> None:
+        """Ensure no prior test left a handle in place before we start."""
+        from recost import _init as init_module
+
+        if init_module._handle is not None:
+            init_module._handle.dispose()
+            init_module._handle = None
+
+    def test_concurrent_init_does_not_orphan_handles(self):
+        """N threads call init() concurrently. After all threads return,
+        exactly one handle should be ``_handle`` (the last installer wins)
+        and every other returned handle should be disposed by the
+        next init() in line. Pre-fix the race lets two threads pass the
+        ``_handle is None`` guard simultaneously and both install — the
+        first thread's handle is then orphaned (not disposed)."""
+        self._drain_module_state()
+
+        from recost import _init as init_module
+        from recost._init import init
+
+        # Disabled config avoids actually starting transport / timer threads
+        # so the test stays fast and isolated. The lock invariant we are
+        # testing does not depend on those resources being live.
+        config_factory = lambda: RecostConfig(enabled=False)
+        num_threads = 4
+        iterations_per_thread = 20
+        produced: list = []
+        produced_lock = threading.Lock()
+        exceptions: list[BaseException] = []
+
+        def worker() -> None:
+            try:
+                for _ in range(iterations_per_thread):
+                    handle = init(config_factory())
+                    with produced_lock:
+                        produced.append(handle)
+            except BaseException as exc:  # noqa: BLE001
+                exceptions.append(exc)
+
+        original_switch = sys.getswitchinterval()
+        sys.setswitchinterval(0.000001)
+        try:
+            threads = [threading.Thread(target=worker) for _ in range(num_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=30.0)
+        finally:
+            sys.setswitchinterval(original_switch)
+
+        assert not exceptions, f"unexpected exceptions: {exceptions!r}"
+        # Exactly one handle should be undisposed — the final winner.
+        undisposed = [h for h in produced if not h._disposed]
+        assert len(undisposed) == 1, (
+            f"expected exactly 1 undisposed handle, got {len(undisposed)} "
+            f"out of {len(produced)} total — earlier handles were orphaned"
+        )
+        assert init_module._handle is undisposed[0], (
+            "module-level _handle does not point at the active handle"
+        )
+
+        # Cleanup
+        undisposed[0].dispose()
+
+    def test_concurrent_init_dispose_leaves_module_clean(self):
+        """N threads each loop init()/dispose() many times. After all
+        threads finish, the module must report a clean state: ``_handle``
+        is None and the interceptor is uninstalled."""
+        self._drain_module_state()
+
+        from recost import _init as init_module
+        from recost._init import init
+        from recost._interceptor import is_installed
+
+        num_threads = 4
+        iterations_per_thread = 20
+        exceptions: list[BaseException] = []
+
+        def worker() -> None:
+            try:
+                for _ in range(iterations_per_thread):
+                    handle = init(RecostConfig(enabled=False))
+                    handle.dispose()
+            except BaseException as exc:  # noqa: BLE001
+                exceptions.append(exc)
+
+        original_switch = sys.getswitchinterval()
+        sys.setswitchinterval(0.000001)
+        try:
+            threads = [threading.Thread(target=worker) for _ in range(num_threads)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=30.0)
+        finally:
+            sys.setswitchinterval(original_switch)
+
+        assert not exceptions, f"unexpected exceptions: {exceptions!r}"
+        assert init_module._handle is None, "module-level _handle should be None"
+        assert not is_installed(), "interceptor should be uninstalled"

From 77125f01eaf2630a776048bdeb172366c8560bfe Mon Sep 17 00:00:00 2001
From: Andres Lopez <190146319+AndresL230@users.noreply.github.com>
Date: Wed, 13 May 2026 04:03:05 -0400
Subject: [PATCH 3/3] style(tests): hoist imports + replace lambda in race
 tests

Move `sys` / `threading` imports introduced by the new race-test
classes to the top of their files (ruff E402) and rewrite the
`config_factory` lambda as a `def` (ruff E731). No behavior change.

Refs #4
---
 tests/test_init.py        | 7 ++++---
 tests/test_interceptor.py | 4 +---
 2 files changed, 5 insertions(+), 6 deletions(-)

diff --git a/tests/test_init.py b/tests/test_init.py
index 78476da..21c9404 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -2,6 +2,7 @@
 Tests for recost/_init.py
 """
 
+import sys
 import threading
 import time
 
@@ -78,8 +79,6 @@ def test_local_mode_excludes_localhost(self):
 # Thread safety — init/dispose race
 # ---------------------------------------------------------------------------
 
-import sys
-
 
 class TestInitDisposeRace:
     """Regression tests for issue #4 — concurrent init() / dispose() must
@@ -108,7 +107,9 @@ def test_concurrent_init_does_not_orphan_handles(self):
         # Disabled config avoids actually starting transport / timer threads
         # so the test stays fast and isolated. The lock invariant we are
         # testing does not depend on those resources being live.
-        config_factory = lambda: RecostConfig(enabled=False)
+        def config_factory() -> RecostConfig:
+            return RecostConfig(enabled=False)
+
         num_threads = 4
         iterations_per_thread = 20
         produced: list = []
diff --git a/tests/test_interceptor.py b/tests/test_interceptor.py
index e7e3158..a8a6534 100644
--- a/tests/test_interceptor.py
+++ b/tests/test_interceptor.py
@@ -4,6 +4,7 @@
 Tests urllib3 (via requests), httpx sync, httpx async interception.
 """
 
+import sys
 import threading
 from http.server import BaseHTTPRequestHandler, HTTPServer
 
@@ -194,9 +195,6 @@ def test_uninstall_restores_originals(self, test_server):
 # Thread safety — install/uninstall race
 # ---------------------------------------------------------------------------
 
-import sys
-import threading
-
 
 class TestInstallUninstallRace:
     """Regression tests for issue #4 — install/uninstall must be safe under