broker: a swallowed cancel spun until the host died (v0.3.5)

A full `pytest tests/` run reached 46 GB RSS plus 20 GB of swap on a 54 GB
machine. Swap filled, the OOM killer took coderai-nvidia — which was serving —
and the box needed a hard reset, which lost the last forty minutes of
kern.log with it. atop's history is what identified the process: a pytest
worker, carrying the `coderai-front` process title because importing the front
sets it, growing 7.8 GB → 34.8 GB → 46.6 GB across three ten-minute samples.

The defect is `except asyncio.CancelledError: pass` wrapped around an await
whose job is to collect a CHILD task. asyncio delivers a cancellation as a
CancelledError at the next await point, so that handler catches two unrelated
things: the child reporting it stopped (fine to swallow) and the caller asking
us to stop (never fine). run_forever's reconnect handler sat on exactly such an
await, so `task.cancel()` was absorbed and the loop carried on. With the
reconnect delay mocked to zero by the test driving it, that became a busy loop
allocating ~78 MB/s — AsyncMock records every call it receives — and 400,000
iterations in the five seconds I measured.

being_cancelled() (Task.cancelling(), 3.11+) tells the two apart, and all four
sites now re-raise ours: _stop_heartbeat_task, _cancel_inflight_tasks, the two
keepalive collects in handle_message, and BrokerService.stop. The inflight ones
matter in production independently of the leak — absorbing a cancel there means
a shutdown waits out a whole video generation instead of stopping it, and
run_forever simply reconnected forever instead of exiting. run_forever's own
cancellation path clears the cancel for the duration of its cleanup and
re-raises at the end, so honouring the cancel does not cost the websocket close.

The test that drove it now brakes itself: the zero-delay sleep stand-in counts,
and raises a BaseException (anything else is caught by the reconnect handler)
well past the two reconnects it needs. With the product bug reintroduced the
file fails in 1.5 s at 441 MB instead of eating the machine.

tests/conftest.py adds what was missing in general: an RSS ceiling (6 GB,
CODERAI_TEST_RSS_LIMIT_MB) and a per-test stall ceiling (180 s,
CODERAI_TEST_STALL_SECONDS), both dumping every thread's stack to a file before
exiting — pytest captures stderr per test and drops the buffer when the process
does not unwind, so a watchdog that only printed would have died silently. No
new dependency. "A test can consume the host" is a property of the run, not of
one test.

tests/test_broker_protocol.py could not finish before this — it hung at the
14th test. The whole suite now completes for the first time: 78 s, 1.57 GB peak,
1635 passed. The 25 failures are the 17 that already failed plus the 8 stale
register-ack expectations in that file, which were unreachable behind the hang
and are a separate question (the fixtures send `accepted: true`; the client has
required `status: "ok"` for some time).
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019sRvvg5WSdLgcFUM96pmE4

codai/__init__.py

@@ -16,7 +16,7 @@
 
 # Canonical product version for CoderAI — single source of truth. Both the API
 # metadata and the admin web UI read from here.
-__version__ = "0.3.4"
+__version__ = "0.3.5"
 __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
 
 # Configure the CUDA caching allocator BEFORE torch is imported anywhere.

codai/broker/client.py

@@ -26,6 +26,32 @@ Dispatcher = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]
 logger = logging.getLogger(__name__)
 
 
+def being_cancelled() -> bool:
+    """True when the CURRENT coroutine has itself been cancelled.
+
+    asyncio delivers a cancellation as a CancelledError at the next await point —
+    including an await whose whole purpose is to collect a CHILD task we just
+    cancelled. ``except asyncio.CancelledError: pass`` around that await therefore
+    catches two completely different things: the child reporting that it stopped
+    (fine to swallow) and the caller asking US to stop (never fine). Swallowing the
+    second one means a shutdown is silently ignored and the loop carries on.
+
+    ``Task.cancelling()`` (3.11+) counts cancel() calls made against the current
+    task, which is what tells them apart. Guarded with getattr so an older runtime
+    degrades to the previous behaviour rather than raising.
+    """
+    current = asyncio.current_task()
+    counter = getattr(current, "cancelling", None)
+    try:
+        return bool(counter and counter() > 0)
+    except Exception:
+        return False
+
+
+# The name the first fix used; kept so nothing that imported it breaks.
+_being_cancelled = being_cancelled
+
+
 class BrokerClient:
     def __init__(self, runtime, dispatcher: Dispatcher | None = None):
         self.runtime = runtime
@@ -218,9 +244,21 @@ class BrokerClient:
                     self._inflight_tasks.add(task)
                     task.add_done_callback(self._inflight_tasks.discard)
             except asyncio.CancelledError:
+                # Cleanup here awaits child tasks, and those awaits are themselves
+                # cancellation points: with the cancel still pending they would
+                # re-raise immediately (see _being_cancelled) and leave the
+                # websocket open. Clear it for the duration of the cleanup and
+                # re-raise at the end, so the caller still sees a cancelled task.
+                current = asyncio.current_task()
+                uncancel = getattr(current, "uncancel", None)
+                if uncancel is not None:
+                    uncancel()
+                try:
                     await self._stop_heartbeat_task()
                     await self._cancel_inflight_tasks()
                     await self._close_websocket()
+                except Exception:
+                    logger.debug("broker shutdown cleanup failed", exc_info=True)
                 raise
             except Exception as error:
                 logger.warning(
@@ -258,7 +296,11 @@ class BrokerClient:
         try:
             await task
         except asyncio.CancelledError:
-            pass
+            # The heartbeat reporting that it stopped is expected. OUR own
+            # cancellation arriving here is not, and must not be absorbed: it is
+            # how run_forever is shut down.
+            if being_cancelled():
+                raise
 
     async def _cancel_inflight_tasks(self):
         if not self._inflight_tasks:
@@ -271,7 +313,8 @@ class BrokerClient:
             try:
                 await task
             except asyncio.CancelledError:
-                pass
+                if being_cancelled():
+                    raise
             except Exception:
                 pass
 
@@ -297,7 +340,10 @@ class BrokerClient:
                 except Exception:
                     break
         except asyncio.CancelledError:
-            pass
+            # Re-raise: the caller cancels this deliberately when the request
+            # finishes, and a task that swallows its cancellation reports itself as
+            # having completed normally, which hides a stuck keepalive.
+            raise
 
     async def _heartbeat_loop(self):
         started_at = time.monotonic()
@@ -476,7 +522,12 @@ class BrokerClient:
                         try:
                             await keepalive_task
                         except asyncio.CancelledError:
-                            pass
+                            # Not ours to swallow: this runs inside an inflight
+                            # request task, and absorbing the cancel here means a
+                            # shutdown waits out a whole generation instead of
+                            # stopping it.
+                            if being_cancelled():
+                                raise
                 return None
 
             try:
@@ -493,7 +544,8 @@ class BrokerClient:
                     try:
                         await keepalive_task
                     except asyncio.CancelledError:
-                        pass
+                        if being_cancelled():
+                            raise
             reply_bytes = json.dumps(response)
             logger.info(
                 "CoderAI broker replied request_id=%s status=%s event=%s reply_bytes=%d",

codai/broker/service.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import asyncio
 
-from codai.broker.client import BrokerClient
+from codai.broker.client import BrokerClient, being_cancelled
 from codai.broker.dispatcher import execute_broker_request
 
 
@@ -36,6 +36,11 @@ class BrokerService:
         try:
             await task
         except asyncio.CancelledError:
-            pass
+            # The client reporting that it stopped is the expected outcome. A
+            # cancellation of THIS coroutine — stop() awaited from a task that is
+            # itself being torn down — is not, and swallowing it tells the caller
+            # the shutdown completed normally.
+            if being_cancelled():
+                raise
         finally:
             self._started = False

tests/conftest.py

+"""Guards so a runaway test cannot take the host down with it.
+
+On 2026-10-10 a full `pytest tests/` run grew to 46 GB RSS plus 20 GB of swap on a
+54 GB machine: a broker test mocked the reconnect delay to zero and the loop it was
+driving swallowed the cancellation meant to stop it, so it spun at full speed
+allocating ~78 MB/s (AsyncMock records every call it receives). Swap filled, the OOM
+killer took the coderai server — which was serving at the time — and the box needed a
+hard reset. The kernel log for the final forty minutes was lost with it.
+
+The leak itself is fixed (codai/broker/client.py) and that test now brakes itself. But
+"a test can consume the machine" is a property of the test RUN, not of one test, and
+the next one will be written by someone who has never heard of this incident. So:
+
+  RSS ceiling   — a watchdog thread samples this process's own RSS and, above the
+                  ceiling, dumps every thread's stack and exits 97. A test that wants
+                  several GB is a bug; dying with a traceback that names it beats an
+                  OOM kill that names whatever happened to be fattest.
+  stall ceiling — a hung test is how a slow leak gets the time to become a big one,
+                  and it blocks every test behind it. faulthandler records where it
+                  is stuck and ends the run rather than waiting forever.
+
+Both write their report to a FILE as well as to stderr, because the process exits
+hard: pytest captures stderr per test and discards the buffer when the process does
+not unwind. The path is printed on the way out and defaults to
+``$TMPDIR/coderai-test-abort-<pid>.log``.
+
+Both are on by default and tunable from the environment:
+
+  CODERAI_TEST_RSS_LIMIT_MB    ceiling in MB (default 6144; 0 disables)
+  CODERAI_TEST_STALL_SECONDS   per-test stall limit (default 180; 0 disables)
+  CODERAI_TEST_ABORT_LOG       where to write the report
+
+A test that legitimately needs more should raise its own ceiling explicitly rather
+than lift the whole suite's.
+"""
+
+import faulthandler
+import os
+import sys
+import tempfile
+import threading
+
+_PAGE = os.sysconf("SC_PAGE_SIZE") if hasattr(os, "sysconf") else 4096
+
+RSS_LIMIT_MB = int(os.environ.get("CODERAI_TEST_RSS_LIMIT_MB") or 6144)
+STALL_SECONDS = int(os.environ.get("CODERAI_TEST_STALL_SECONDS") or 180)
+ABORT_LOG = os.environ.get("CODERAI_TEST_ABORT_LOG") or os.path.join(
+    tempfile.gettempdir(), f"coderai-test-abort-{os.getpid()}.log")
+
+# Sampled often enough that the ceiling is overshot by little: the incident ran at
+# 78 MB/s, and even a pathological allocator only gets a fraction of a second.
+_SAMPLE_SECONDS = 0.25
+
+_current_test = ["(collection)"]
+_log_handle = None
+
+
+def _rss_mb():
+    """This process's resident size in MB, or None where /proc is unavailable."""
+    try:
+        with open("/proc/self/statm") as handle:
+            return int(handle.read().split()[1]) * _PAGE / 1048576
+    except Exception:
+        return None
+
+
+def _open_log():
+    """The report file, kept open: a watchdog must not need to allocate much to report.
+
+    Opened once up front for the same reason — by the time the ceiling is hit, opening
+    a file may itself be what fails.
+    """
+    global _log_handle
+    if _log_handle is None:
+        try:
+            _log_handle = open(ABORT_LOG, "w", buffering=1)
+        except Exception:
+            _log_handle = sys.__stderr__
+    return _log_handle
+
+
+def _shout(message):
+    """Say it everywhere it might survive the process exiting hard."""
+    for stream in (_open_log(), sys.__stderr__):
+        try:
+            stream.write(message)
+            stream.flush()
+        except Exception:
+            pass
+    try:
+        os.write(2, message.encode())
+    except Exception:
+        pass
+
+
+def _watch_rss(limit_mb):
+    while True:
+        rss = _rss_mb()
+        if rss is None:
+            return          # not Linux: nothing to watch with
+        if rss > limit_mb:
+            _shout(
+                f"\n\n*** test run aborted: {rss:.0f} MB resident, over the "
+                f"{limit_mb} MB ceiling, during {_current_test[0]}\n"
+                f"*** a test allocating this much is a leak, not a workload — the "
+                f"stacks in {ABORT_LOG} say where.\n"
+                f"*** raise CODERAI_TEST_RSS_LIMIT_MB if this one genuinely needs "
+                f"the memory. Exit code 97.\n\n")
+            try:
+                faulthandler.dump_traceback(file=_open_log(), all_threads=True)
+                _open_log().flush()
+            except Exception:
+                pass
+            os._exit(97)
+        threading.Event().wait(_SAMPLE_SECONDS)
+
+
+def pytest_configure(config):
+    if RSS_LIMIT_MB > 0 and _rss_mb() is not None:
+        threading.Thread(target=_watch_rss, args=(RSS_LIMIT_MB,),
+                         name="rss-watchdog", daemon=True).start()
+
+
+def pytest_runtest_logstart(nodeid, location):
+    _current_test[0] = nodeid
+    if STALL_SECONDS > 0:
+        # exit=True: the stack is written and the run ends. A hung test cannot be
+        # waited out — it blocks every test after it, and in CI it blocks forever.
+        faulthandler.dump_traceback_later(STALL_SECONDS, exit=True,
+                                          file=_open_log())
+
+
+def pytest_runtest_logfinish(nodeid, location):
+    if STALL_SECONDS > 0:
+        faulthandler.cancel_dump_traceback_later()
+
+
+def pytest_sessionfinish(session, exitstatus):
+    """Nothing went wrong: drop the empty report rather than leave litter behind."""
+    global _log_handle
+    if STALL_SECONDS > 0:
+        faulthandler.cancel_dump_traceback_later()
+    handle, _log_handle = _log_handle, None
+    if handle is None or handle is sys.__stderr__:
+        return
+    try:
+        empty = handle.tell() == 0
+        handle.close()
+        if empty:
+            os.unlink(ABORT_LOG)
+    except Exception:
+        pass

tests/test_broker_protocol.py

@@ -24,6 +24,42 @@ class Recorder:
         self.events.append(event)
 
 
+class RunawayLoop(BaseException):
+    """Raised to break a test's reconnect loop out of run_forever.
+
+    BaseException on purpose: an ordinary exception raised from the sleep stand-in
+    would be caught by run_forever's `except Exception` reconnect handler and bring
+    us straight back round to the same stand-in.
+    """
+
+
+def bounded_zero_sleep(calls, limit=50, what="run_forever"):
+    """A zero-delay stand-in for asyncio.sleep that refuses to spin forever.
+
+    Mocking the reconnect delay to zero is what makes these tests fast, and it is
+    also what makes them dangerous: with the delay gone, the reconnect loop is a
+    busy loop, and it only stops because the test cancels the task. When a
+    cancellation was swallowed (see _being_cancelled in codai/broker/client.py) the
+    spin never ended — it allocated ~78 MB/s, because AsyncMock records every call
+    it receives, reached 46 GB RSS, and the OOM killer took the server down with it.
+
+    So the stand-in counts. These tests need a couple of reconnects; well past that
+    means the loop is no longer stopping, and failing in milliseconds with a clear
+    message beats taking the host down.
+    """
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay):
+        calls.append(delay)
+        if len(calls) > limit:
+            raise RunawayLoop(
+                f"{what} slept {len(calls)} times without stopping — a "
+                "cancellation is being swallowed")
+        await real_sleep(0)
+
+    return fake_sleep
+
+
 class FakeWebSocket:
     def __init__(self, messages):
         self._messages = list(messages)
@@ -248,10 +284,8 @@ async def test_broker_client_run_forever_closes_registered_socket_before_reconne
         if len(attempts) >= 2:
             reconnected.set()
 
-    real_sleep = asyncio.sleep
-
-    async def fake_sleep(delay):
-        await real_sleep(0)
+    sleep_calls = []
+    fake_sleep = bounded_zero_sleep(sleep_calls, what="the reconnect loop")
 
     client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
 
@@ -498,11 +532,7 @@ async def test_broker_client_run_forever_reconnects_after_disconnect():
         if len(attempts) >= 2:
             reconnected.set()
 
-    real_sleep = asyncio.sleep
-
-    async def fake_sleep(delay):
-        sleep_calls.append(delay)
-        await real_sleep(0)
+    fake_sleep = bounded_zero_sleep(sleep_calls, what="the reconnect loop")
 
     client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
 
@@ -615,11 +645,7 @@ async def test_broker_client_run_forever_handles_heartbeat_before_reconnect():
         if len(attempts) >= 2:
             reconnected.set()
 
-    real_sleep = asyncio.sleep
-
-    async def fake_sleep(delay):
-        sleep_calls.append(delay)
-        await real_sleep(0)
+    fake_sleep = bounded_zero_sleep(sleep_calls, what="the reconnect loop")
 
     client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
 

tests/test_broker_shutdown.py

+"""Cancelling the broker client must actually stop it.
+
+The incident this file exists for: a full test run reached 46 GB RSS + 20 GB swap on
+a 54 GB machine and the OOM killer took the live coderai server with it. The shape was
+``except asyncio.CancelledError: pass`` wrapped around an await whose job was to
+collect a CHILD task. asyncio delivers a cancellation as a CancelledError at the next
+await point, so that handler caught two different things: the child saying it stopped,
+and the caller saying *we* must stop. Swallowing the second one left run_forever
+looping — and with the reconnect delay mocked to zero by the test that drove it, the
+loop spun at full speed allocating ~78 MB/s, because AsyncMock records every call.
+
+So these tests are about cancellation being observed, not about websockets. They each
+bound the loop, because the failure mode under test is precisely "the loop does not
+stop" and a test for that must not be the thing that hangs.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from codai.broker.client import BrokerClient, being_cancelled
+from codai.broker.config import BrokerRuntimeConfig
+
+pytestmark = pytest.mark.anyio("asyncio")
+
+# Generous next to the two reconnects these tests need, and tiny next to the
+# ~400,000 iterations the unfixed loop managed in the 5 seconds I measured.
+SPIN_LIMIT = 200
+
+
+class _Runaway(BaseException):
+    """BaseException: anything else is caught by run_forever's reconnect handler."""
+
+
+def _bounded_sleep(calls):
+    real_sleep = asyncio.sleep
+
+    async def fake_sleep(delay):
+        calls.append(delay)
+        if len(calls) > SPIN_LIMIT:
+            raise _Runaway(f"the reconnect loop slept {len(calls)} times")
+        await real_sleep(0)
+
+    return fake_sleep
+
+
+def _client(**kwargs):
+    runtime = BrokerRuntimeConfig(enabled=True, reconnect_initial_delay_seconds=1,
+                                  reconnect_max_delay_seconds=4, **kwargs)
+    return BrokerClient(runtime)
+
+
+class _DisconnectingWebSocket:
+    def __init__(self):
+        self.closed = False
+
+    async def recv(self):
+        raise RuntimeError("disconnect")
+
+    async def close(self):
+        self.closed = True
+
+
+# ------------------------------------------------------------------ the mechanism
+async def test_being_cancelled_is_false_normally():
+    assert being_cancelled() is False
+
+
+async def test_being_cancelled_is_true_once_we_have_been_cancelled():
+    """This is the whole distinction the fix rests on."""
+    seen = []
+
+    async def victim():
+        try:
+            await asyncio.sleep(10)
+        except asyncio.CancelledError:
+            seen.append(being_cancelled())
+            raise
+
+    task = asyncio.create_task(victim())
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert seen == [True]
+
+
+# ------------------------------------------------------------------- run_forever
+async def test_cancelling_run_forever_stops_it():
+    """The incident, in miniature: zero reconnect delay plus a cancel that was ignored."""
+    client = _client()
+    attempts = []
+    sleep_calls = []
+    reconnected = asyncio.Event()
+
+    async def fake_connect_and_register():
+        attempts.append("connect")
+        client.websocket = _DisconnectingWebSocket()
+        if len(attempts) >= 2:
+            reconnected.set()
+
+    client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
+
+    with patch("codai.broker.client.asyncio.sleep",
+               new=AsyncMock(side_effect=_bounded_sleep(sleep_calls))):
+        task = asyncio.create_task(client.run_forever())
+        await asyncio.wait_for(reconnected.wait(), timeout=1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, timeout=2)
+
+    assert len(attempts) < SPIN_LIMIT, (
+        f"{len(attempts)} reconnects: the cancel was swallowed and the loop spun on")
+    assert task.cancelled()
+
+
+async def test_the_cancel_is_honoured_even_while_the_heartbeat_is_being_collected():
+    """The exact await the old handler sat on: _stop_heartbeat_task's `await task`.
+
+    A heartbeat that takes a moment to finish is the window in which the cancellation
+    arrives, and the window the old code lost it in.
+    """
+    client = _client()
+    in_cleanup = asyncio.Event()
+
+    class _SlowStop(_DisconnectingWebSocket):
+        async def close(self):
+            in_cleanup.set()
+            await asyncio.sleep(0)
+            self.closed = True
+
+    socket = _SlowStop()
+
+    async def fake_connect_and_register():
+        client.websocket = socket
+
+    client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
+    task = asyncio.create_task(client.run_forever())
+    await asyncio.wait_for(in_cleanup.wait(), timeout=2)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(task, timeout=2)
+
+
+async def test_shutdown_still_closes_the_websocket():
+    """Honouring the cancel must not cost the cleanup.
+
+    run_forever's cancellation handler awaits three things, and every one of those
+    awaits is itself a cancellation point — so with the cancel left pending they would
+    re-raise at the first of them and strand an open socket. It clears the cancel for
+    the duration and re-raises at the end.
+    """
+    client = _client(heartbeat_interval_seconds=60)
+    connected = asyncio.Event()
+    socket = _DisconnectingWebSocket()
+
+    async def fake_connect_and_register():
+        client.websocket = socket
+        connected.set()
+        await asyncio.sleep(3600)       # hold the loop open, inside the try
+
+    client.connect_and_register = AsyncMock(side_effect=fake_connect_and_register)
+    task = asyncio.create_task(client.run_forever())
+    await asyncio.wait_for(connected.wait(), timeout=2)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await asyncio.wait_for(task, timeout=2)
+
+    assert socket.closed, "the websocket was left open on shutdown"
+    assert client.websocket is None
+    assert client._heartbeat_task is None
+
+
+# ------------------------------------------------------------------- the helpers
+async def test_stopping_the_heartbeat_swallows_the_childs_cancellation():
+    """The half that was always right: a child reporting that it stopped is expected."""
+    client = _client()
+
+    async def heartbeat():
+        await asyncio.sleep(3600)
+
+    client._heartbeat_task = asyncio.create_task(heartbeat())
+    await asyncio.sleep(0)
+    await client._stop_heartbeat_task()         # must not raise
+    assert client._heartbeat_task is None
+
+
+async def test_stopping_the_heartbeat_re_raises_our_own_cancellation():
+    client = _client()
+    outcome = []
+
+    async def heartbeat():
+        await asyncio.sleep(3600)
+
+    async def caller():
+        client._heartbeat_task = asyncio.create_task(heartbeat())
+        await asyncio.sleep(0)
+        try:
+            await client._stop_heartbeat_task()
+        except asyncio.CancelledError:
+            outcome.append("raised")
+            raise
+        outcome.append("swallowed")
+
+    task = asyncio.create_task(caller())
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert outcome == ["raised"], outcome
+
+
+async def test_cancelling_inflight_work_re_raises_our_own_cancellation():
+    """Same defect, same shape, one method along — and the one that holds request work."""
+    client = _client()
+    outcome = []
+
+    async def inflight():
+        await asyncio.sleep(3600)
+
+    async def caller():
+        client._inflight_tasks.add(asyncio.create_task(inflight()))
+        await asyncio.sleep(0)
+        try:
+            await client._cancel_inflight_tasks()
+        except asyncio.CancelledError:
+            outcome.append("raised")
+            raise
+        outcome.append("swallowed")
+
+    task = asyncio.create_task(caller())
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert outcome == ["raised"], outcome
+
+
+async def test_cancelling_inflight_work_still_ignores_a_failed_request():
+    """An inflight request that raised is not a shutdown problem."""
+    client = _client()
+
+    async def boom():
+        raise RuntimeError("request failed")
+
+    client._inflight_tasks.add(asyncio.create_task(boom()))
+    await asyncio.sleep(0)
+    await client._cancel_inflight_tasks()      # must not raise
+    assert not client._inflight_tasks
+
+
+async def test_a_cancelled_keepalive_reports_itself_as_cancelled():
+    """It swallowed its own cancellation and so looked like a clean completion, which
+    hides a keepalive still running against a finished request."""
+    client = _client()
+    task = asyncio.create_task(client._send_keepalives("req-1", interval=3600))
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert task.cancelled()
+
+
+async def test_stopping_the_service_re_raises_our_own_cancellation():
+    """BrokerService.stop() collects the client task the same way, and is itself
+    awaited from application shutdown — which may already be cancelled."""
+    from codai.broker.service import BrokerService
+
+    client = _client()
+
+    async def never_returns():
+        await asyncio.sleep(3600)
+
+    client.connect_and_register = AsyncMock(side_effect=never_returns)
+    service = BrokerService(client)
+    service.start()
+    await asyncio.sleep(0)
+    outcome = []
+
+    async def caller():
+        try:
+            await service.stop()
+        except asyncio.CancelledError:
+            outcome.append("raised")
+            raise
+        outcome.append("swallowed")
+
+    task = asyncio.create_task(caller())
+    await asyncio.sleep(0)
+    await asyncio.sleep(0)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert outcome == ["raised"], outcome
+
+
+async def test_stopping_the_service_normally_does_not_raise():
+    from codai.broker.service import BrokerService
+
+    client = _client()
+
+    async def never_returns():
+        await asyncio.sleep(3600)
+
+    client.connect_and_register = AsyncMock(side_effect=never_returns)
+    service = BrokerService(client)
+    service.start()
+    await asyncio.sleep(0)
+    await service.stop()                   # must not raise
+    assert service.task is None
+
+
+def test_no_cancellation_handler_in_the_broker_swallows_a_cancel_blind():
+    """Four sites had the same shape; a fifth will be written one day.
+
+    Every `except asyncio.CancelledError:` that wraps an await of a CHILD task has to
+    ask whose cancellation it caught. This reads the source because the point is to
+    catch a NEW handler, not the ones already covered above.
+    """
+    import re
+
+    for name in ("client.py", "service.py"):
+        src = (ROOT / "codai" / "broker" / name).read_text()
+        for match in re.finditer(r"except asyncio\.CancelledError:\n((?:[ \t]*(?:#[^\n]*)?\n)*)"
+                                 r"([ \t]*)([^\n]+)", src):
+            body = match.group(3).strip()
+            assert body != "pass", (
+                f"codai/broker/{name}: a CancelledError handler swallows the cancel "
+                f"blind. If it is collecting a child task it must re-raise when "
+                f"being_cancelled(); if it is its own task's cancellation it must "
+                f"re-raise outright.")
+
+
+# --------------------------------------------------------------- the host guards
+def test_the_test_run_has_a_memory_ceiling_and_a_stall_ceiling():
+    """Without these, the next swallowed cancel takes the machine down again."""
+    import importlib.util
+    spec = importlib.util.spec_from_file_location(
+        "coderai_test_guards", ROOT / "tests" / "conftest.py")
+    guards = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(guards)
+    assert guards.RSS_LIMIT_MB > 0
+    assert guards.STALL_SECONDS > 0
+    # Headroom for the real suite (it peaks well under 2 GB), far below a 54 GB host.
+    assert 2048 <= guards.RSS_LIMIT_MB <= 16384
+    assert hasattr(guards, "_watch_rss")
+
+
+def test_the_guards_can_be_turned_off_or_retuned_from_the_environment():
+    src = (ROOT / "tests/conftest.py").read_text()
+    assert "CODERAI_TEST_RSS_LIMIT_MB" in src
+    assert "CODERAI_TEST_STALL_SECONDS" in src
+    assert "CODERAI_TEST_ABORT_LOG" in src
+
+
+def test_the_report_survives_the_process_exiting_hard():
+    """pytest captures stderr per test and drops the buffer when the process does not
+    unwind, so a watchdog that only printed would die silently."""
+    src = (ROOT / "tests/conftest.py").read_text()
+    assert "file=_open_log()" in src
+    assert "os._exit(97)" in src