fix(worker): stop leaving other workers' probe logs on the host while Activity is off

app/worker_api.py

@@ -944,13 +944,28 @@ def _seconds_to_next_window(now: float | None = None) -> float:
     return activity.WINDOW_SECONDS - (now % activity.WINDOW_SECONDS) + 1
 
 
+def _sweep_probes() -> None:
+    """Open a client, sweep, close it, all in one thread: a cancelled await cannot drop the client."""
+    client = activity._client()
+    try:
+        activity.sweep(client)
+    finally:
+        with contextlib.suppress(Exception):
+            client.close()
+
+
 async def _activity_cycle() -> None:
-    """One round, if the owner switched Activity on and Docker answers."""
+    """One round, if the owner switched Activity on and Docker answers.
+
+    With Activity off there is no round, but probes other workers left behind
+    are still swept: their logs hold raw addresses and nothing else removes them.
+    """
     global _pending_activity, _activity_seq
-    if not _activity_enabled:
-        return
     if not await asyncio.to_thread(orchestrator.docker_available):
         return
+    if not _activity_enabled:
+        await asyncio.to_thread(_sweep_probes)
+        return
     switches = _activity_switches
     result = await asyncio.to_thread(activity.run_round)
     if not _activity_enabled or _activity_switches != switches:
@@ -973,12 +988,7 @@ async def _activity_loop() -> None:
     """
     with contextlib.suppress(Exception):
         if await asyncio.to_thread(orchestrator.docker_available):
-            client = await asyncio.to_thread(activity._client)
-            try:
-                await asyncio.to_thread(activity.sweep, client)
-            finally:
-                with contextlib.suppress(Exception):
-                    client.close()
+            await asyncio.to_thread(_sweep_probes)
     while True:
         await asyncio.sleep(_seconds_to_next_window())
         try:

tests/test_activity_worker.py

@@ -484,9 +484,11 @@ def wapi():
 with (
     patch("app.worker_api.orchestrator.docker_available", return_value=True) as docker_ok,
     patch("app.worker_api.activity.run_round", return_value=None) as run_round,
+    patch("app.worker_api.activity._client", return_value=MagicMock()),
+    patch("app.worker_api.activity.sweep", return_value=[]),
 ):
     asyncio.run(worker_api._activity_cycle())
-    assert not docker_ok.called and not run_round.called, "a round ran before any reply"
+    assert not run_round.called, "a round ran before any reply"
     reply = MagicMock()
     reply.json.return_value = {"activity": {"enabled": True}}
     worker_api._apply_activity_reply(reply, None)
@@ -523,10 +525,39 @@ def test_disabled_never_runs_a_round(self, wapi):
         with (
             patch("app.worker_api.orchestrator.docker_available", return_value=True),
             patch("app.worker_api.activity.run_round") as run_round,
+            patch("app.worker_api.activity._client", return_value=MagicMock()),
+            patch("app.worker_api.activity.sweep", return_value=[]),
         ):
             asyncio.run(wapi._activity_cycle())
         assert not run_round.called
 
+    def test_disabled_still_sweeps_probes_other_workers_left(self, wapi):
+        """With Activity off the sweep at worker start is not the last one: a probe another
+        worker leaves later (it died and came back under a new id) holds raw addresses in
+        its log, and only a sweep removes it."""
+        wapi._activity_enabled = False
+        client = MagicMock()
+        with (
+            patch("app.worker_api.orchestrator.docker_available", return_value=True),
+            patch("app.worker_api.activity.run_round") as run_round,
+            patch("app.worker_api.activity._client", return_value=client),
+            patch("app.worker_api.activity.sweep", return_value=[]) as sweep,
+        ):
+            asyncio.run(wapi._activity_cycle())
+        sweep.assert_called_once_with(client)
+        client.close.assert_called_once()
+        assert not run_round.called
+
+    def test_disabled_without_docker_does_not_sweep(self, wapi):
+        wapi._activity_enabled = False
+        with (
+            patch("app.worker_api.orchestrator.docker_available", return_value=False),
+            patch("app.worker_api.activity._client") as make_client,
+            patch("app.worker_api.activity.sweep") as sweep,
+        ):
+            asyncio.run(wapi._activity_cycle())
+        assert not make_client.called and not sweep.called
+
     def test_enabled_runs_a_round_negative_control(self, wapi):
         """The control for the test above: the same call with the switch on runs."""
         wapi._activity_enabled = True