fix(tests): stop transport flush thread between tests so it doesn't race respx

PR #60 landed the cancellable-sleep fix in Transport._flush_loop and
expected CI wall-clock to drop to 3-5 minutes. The first green run
on PR #60 (PR #60 run #1) actually took 9m 47s — the test step
dominated by a retry storm:

  Request failed (attempt 5/11), retrying in 8.46s: ConnectError
  Request failed (attempt 6/11), retrying in 9.16s: ConnectError
  ...
  Circuit breaker OPEN. Batch of 10 events will be re-queued.

Root cause: `tests/conftest.py:reset_runtime` teardown nulled the
runtime reference WITHOUT calling `runtime.shutdown()`. The
transport flush thread therefore kept running across tests, the
buffer drained through httpx with no respx context active, and the
xdist workers spent the next 9 minutes retry-sending the buffer
against the real (unreachable in CI) backend. `_retry_with_backoff
(max_retries=10, max_delay=10s)` is 65s of pure sleep per failed
batch, and with 4 xdist workers and many buffered batches this
multiplied into 9m 47s — i.e. a CI-noise fix that hid a deeper
lifecycle bug.

Pre-fix CI was already paying this cost (5s shutdown-sleep × 200+
tests ≈ 17 min of teardown per Python leg); the retry storm was
always there but masked by the dominant 5s cost. PR #60's 5s fix
exposed it.

Fix: add `flush: bool = True` to both `Transport.stop()` and
`NullRunRuntime.shutdown()`. When False, the transport thread is
cancelled WITHOUT a final `_do_flush()` / `_persist_to_wal()`.
`tests/conftest.py:reset_runtime` teardown now calls
`inst.shutdown(flush=False)` before nilling the reference. This
makes the conftest teardown a true no-op for the buffer — the test
that wrote the events is responsible for asserting on what it
cared about. The production default (`flush=True`) is preserved,
so the `nullrun.shutdown()` audit contract ("drain in-flight
events") is unchanged.

Pins:

  * `tests/test_transport.py::test_stop_flush_false_skips_final_flush
    ` — buffers an event, calls `stop(flush=False)` with no
    respx active, asserts the call returns in <1s AND the buffer
    is left untouched. Pre-fix this would have hung for 65s+ on
    the first retry.

  * `tests/test_init_contract.py::TestShutdownFlushKwarg::
    test_runtime_shutdown_flush_false_skips_final_flush` — same
    contract at the `NullRunRuntime` level: `shutdown(flush=False
    )` propagates the `flush=False` flag to
    `Transport.stop()`.

Public API additions:

  * `Transport.stop(timeout=10.0, flush: bool = True)` — `flush
    =False` is the new flag.
  * `NullRunRuntime.shutdown(flush: bool = True)` — propagates.
  * `nullrun.shutdown(timeout=2.0, flush: bool = True)` — passes
    `flush` through to the runtime.

No on-wire or production behaviour change. CI step is expected to
drop from ~9m 47s (PR #60 run #1) to ~30-60s on the next run.

Author: maltsev-dev

src/nullrun/__init__.py

@@ -41,7 +41,7 @@ def my_agent(query):
 from nullrun.runtime import track_event, track_llm, track_tool
 
 
-def shutdown(timeout: float = 2.0) -> None:
+def shutdown(timeout: float = 2.0, flush: bool = True) -> None:
     """Gracefully shut down the NullRun runtime.
 
     Sends a clean WebSocket close frame, drains in-flight events, and
@@ -62,6 +62,13 @@ def shutdown(timeout: float = 2.0) -> None:
             ``NullRunRuntime.shutdown `` already caps WS join at
             0.5s and the WS close at 2.0s — this parameter is
             reserved for future expansion and is currently unused.
+        flush: when True (default) the transport drains any
+            buffered events to the backend on the way out. Pass
+            False to cancel the flush thread without a final
+            network call — used by the test conftest to teardown
+            between tests without racing the respx context exit
+            (see ``NullRunRuntime.shutdown(flush=False)`` for the
+            full rationale).
 
     Example::
 
@@ -75,7 +82,7 @@ def shutdown(timeout: float = 2.0) -> None:
     runtime = NullRunRuntime._instance  # type: ignore[attr-defined]
     if runtime is None:
         return
-    runtime.shutdown()
+    runtime.shutdown(flush=flush)
 
 
 def status():

src/nullrun/runtime.py

@@ -1822,8 +1822,20 @@ def _auth_headers(self) -> dict[str, str]:
         headers[HEADER_PROTOCOL] = _protocol_header_value()
         return headers
 
-    def shutdown(self) -> None:
-        """Shutdown runtime gracefully."""
+    def shutdown(self, flush: bool = True) -> None:
+        """Shutdown runtime gracefully.
+
+        Args:
+            flush: when True (default) the transport drains any
+                buffered events to the backend on the way out — the
+                production "send everything you have before we go"
+                contract. When False, the transport thread is
+                cancelled without a final ``_do_flush()``. Used by
+                the test conftest to teardown between tests without
+                racing the respx context exit
+                (see ``Transport.stop(flush=False)`` for the full
+                rationale; observed 9m 47s CI noise on PR #60).
+        """
         # Stop the HTTP poller (legacy path) if it was started.
         self._poll_running = False
         if self._poll_thread and self._poll_thread.is_alive():
@@ -1847,7 +1859,7 @@ def shutdown(self) -> None:
             self._ws_thread.join(timeout=0.5)
 
         if self._transport:
-            self._transport.stop()
+            self._transport.stop(flush=flush)
         NullRunRuntime._instance = None
         logger.info("NullRun Runtime shutdown")
 

src/nullrun/transport.py

@@ -807,8 +807,25 @@ def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         except Exception as e:  # noqa: BLE001 — best-effort on context exit
             logger.debug(f"Transport.__exit__: stop() raised: {e}")
 
-    def stop(self, timeout: float = 10.0) -> None:
-        """Stop background flush thread and flush remaining events."""
+    def stop(self, timeout: float = 10.0, flush: bool = True) -> None:
+        """Stop background flush thread and flush remaining events.
+
+        Args:
+            timeout: max seconds to wait for the flush thread to exit.
+            flush: when True (default) the final ``_do_flush()`` and
+                ``_persist_to_wal()`` run after the thread joins — the
+                production "drain on the way out" contract. When
+                False, the thread is cancelled but the buffer is left
+                alone. The test conftest uses ``flush=False`` to
+                teardown between tests without a final httpx call —
+                in tests the respx context has already exited by the
+                time the conftest's teardown runs, so a final
+                ``_do_flush()`` would race respx and trigger a
+                ``ConnectError`` retry storm
+                (observed: 9m 47s of "Request failed (attempt N/11),
+                retrying in 10s" on PR #60, dominating the
+                otherwise-fast xdist wall clock).
+        """
         self._running = False
         self._stopped = True  # Mark as stopped to prevent double flush
         # Wake the flush thread out of its cancellable sleep so join()
@@ -820,8 +837,9 @@ def stop(self, timeout: float = 10.0) -> None:
         self._stop_event.set()
         if self._flush_thread:
             self._flush_thread.join(timeout=timeout)
-        self._do_flush()  # Final flush
-        self._persist_to_wal()  # WAL any remaining events
+        if flush:
+            self._do_flush()  # Final flush
+            self._persist_to_wal()  # WAL any remaining events
         self._client.close()
         # Detach the weakref finalizer — stop is the canonical
         # "I am done" path. After this point the finalizer will

tests/conftest.py

@@ -42,8 +42,24 @@ def reset_runtime():
 
     yield
 
-    # Just clear references, don't call shutdown which may try HTTP calls
-    # after respx mock context has already exited
+    # Stop any running transport flush thread BEFORE we drop the
+    # reference. Without this the thread keeps running across tests,
+    # the buffer drains through httpx with no respx context active,
+    # and the worker logs a ``ConnectError`` retry storm for the rest
+    # of the xdist session — observed 9m 47s of "Request failed
+    # (attempt N/11), retrying in 10s" on PR #60, which dwarfed the
+    # actual test time. ``flush=False`` skips the final ``_do_flush``
+    # / ``_persist_to_wal`` so the teardown is a true no-op even when
+    # the buffer still has events; the test that wrote them is
+    # responsible for asserting on what it cared about. Best-effort:
+    # the runtime may be in any state at teardown, and we don't want
+    # a flaky shutdown to mask the real test failure that just ran.
+    inst = NullRunRuntime._instance
+    if inst is not None:
+        try:
+            inst.shutdown(flush=False)
+        except Exception:
+            pass
     NullRunRuntime._instance = None
     _dec._runtime = None
     _act._action_handler = None

tests/test_init_contract.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import threading
+import time
 
 import pytest
 
@@ -316,3 +317,57 @@ def test_init_logs_debug_when_probe_raises(
                 rt.shutdown()
         finally:
             _caps_mod.probe_capabilities = original_probe
+
+
+class TestShutdownFlushKwarg:
+    """Regression pin for the PR #60 follow-up: ``shutdown(flush=...)``
+    must propagate to ``Transport.stop(flush=...)`` so the test
+    conftest can teardown between tests without racing the respx
+    context exit. Pre-this-pin, the conftest's teardown just nulled
+    the runtime reference; the transport flush thread kept running
+    with a non-empty buffer, the next ``_do_flush`` raced respx and
+    hit the real network, and CI logged 9m 47s of
+    "Request failed (attempt N/11), retrying in 10s" — dominating
+    the otherwise-fast xdist wall clock.
+    """
+
+    def test_runtime_shutdown_flush_false_skips_final_flush(self, mock_api):
+        """``runtime.shutdown(flush=False)`` cancels the transport
+        thread WITHOUT triggering a final ``_do_flush()``.
+
+        We use ``_test_mode=True`` so init skips auth, then buffer
+        an event directly into the transport (bypassing
+        ``track()``'s auth path), then call ``shutdown(flush=False
+        )`` AFTER the respx context has exited. The whole call must
+        return in well under 1s; a regression to
+        ``shutdown(flush=...)`` not propagating would push the
+        assertion past the 5s connect timeout × retry budget.
+        """
+        # _test_mode skips auth but still starts the transport thread.
+        rt = NullRunRuntime(
+            api_key="test-key-12345678",
+            _test_mode=True,
+            polling=False,
+        )
+        # Buffer an event so a final _do_flush() would have
+        # something to attempt to send. mock_api is a function-
+        # scoped fixture; we drop the reference so the respx
+        # context exits before we call shutdown.
+        rt._transport._buffer.append({"event_id": "x", "event": "test"})
+
+        started = time.monotonic()
+        rt.shutdown(flush=False)
+        elapsed = time.monotonic() - started
+
+        assert elapsed < 1.0, (
+            f"shutdown(flush=False) took {elapsed:.2f}s; expected "
+            f"<1s. The flush=False kwarg did not propagate to "
+            f"Transport.stop() — the conftest teardown regression "
+            f"is back."
+        )
+        # And the buffer is left alone — the test that wrote it
+        # is responsible for asserting on what it cared about.
+        assert len(rt._transport._buffer) == 1, (
+            f"shutdown(flush=False) should leave the buffer alone; "
+            f"expected 1 event, got {len(rt._transport._buffer)}."
+        )

tests/test_transport.py

@@ -111,6 +111,59 @@ def test_stop_interrupts_flush_sleep(self):
             f"cancellable-wait fix regressed."
         )
 
+    def test_stop_flush_false_skips_final_flush(self):
+        """``stop(flush=False)`` cancels the thread WITHOUT a final
+        ``_do_flush()`` so the conftest can teardown between tests
+        without racing the respx context exit.
+
+        Regression pin for the second CI-noise fix (PR #60 follow-up):
+        the conftest previously nulled the runtime reference without
+        calling ``shutdown()`` so the transport flush thread kept
+        running with a non-empty buffer; on the next ``_do_flush``
+        (after respx exited) httpx hit the real network, got
+        ``ConnectError``, retried 11 times with up-to-10s backoff,
+        and dominated the xdist wall clock (9m 47s of
+        "Request failed (attempt N/11), retrying in 10s").
+
+        The contract being pinned here: with ``flush=False``,
+        ``_do_flush`` is NOT called from ``stop()`` even when the
+        buffer is non-empty. The teardown is a true no-op apart
+        from the thread join.
+        """
+        from nullrun.transport import FlushConfig
+
+        t = Transport(
+            api_url="https://api.test.nullrun.io",
+            api_key="test-key-12345678",
+            config=FlushConfig(flush_interval=30.0),
+        )
+        t.start()
+        # Buffer an event so a final _do_flush() would have something
+        # to attempt to send (and therefore would race respx).
+        t._buffer.append({"event_id": "x", "event": "test"})
+        # No respx mock active here — if stop() tries to flush, httpx
+        # will block for the 5s connect timeout per attempt and
+        # multiply by the retry budget. The whole point of
+        # ``flush=False`` is to skip that path entirely.
+        started = time.monotonic()
+        t.stop(flush=False)
+        elapsed = time.monotonic() - started
+        # Generous bound: thread join is the only blocking step. A
+        # regression to "stop() always flushes" would push this
+        # past 60s on the first failure.
+        assert elapsed < 1.0, (
+            f"stop(flush=False) took {elapsed:.2f}s; expected < 1s. "
+            f"The final _do_flush() ran despite flush=False — the "
+            f"conftest teardown is back to racing respx and the "
+            f"CI retry-storm regression is open again."
+        )
+        # And the buffer is left alone — the conftest contract is
+        # "we don't care, the test that wrote it is responsible".
+        assert len(t._buffer) == 1, (
+            f"stop(flush=False) should leave the buffer untouched; "
+            f"expected 1 event, got {len(t._buffer)}."
+        )
+
     def test_ssl_verification_enabled(self, transport):
         # httpx 0.28+ doesn't expose verify as a direct attribute
         # SSL verification is enabled by default (verify=True)