fix: audit follow-ups for the next-free-GPU dispatcher

Lifecycle:
- Queue clears and deletes leave a run whose master prompt is currently
  executing alone, as ComfyUI leaves the running prompt alone; previously an
  executing orchestrator-only head run could be cancelled out from under its
  waiting Collector.
- The sweep now also drops worker-assigned runs whose queued delegate prompt
  left the queue without the Collector finishing, releasing their result
  queues instead of leaking them and polling forever.
- Completed runs release their result queues, and orchestration no longer
  pre-creates result queues when load balance picks the master alone.

Transport:
- A websocket timeout during connect, before anything was sent, is a plain
  failure the dispatcher may retry elsewhere; only timeouts after the prompt
  was sent remain AmbiguousDispatchError.

Efficiency and cleanup:
- One WorkerSnapshot per worker per cycle answers every run's "is my job
  there" question (queue first, history only when needed).
- Sweep and cancellation snapshot queued/running prompt ids once per pass.
- probe_worker_queues reuses dispatch._probe_worker_queue; the dead
  aggregate idle/busy probe helper is gone.

Rebased on "never replay load-balanced runs on missing history": an
unresolved submission stays on hold until delivery or positive evidence,
so there is no rejection path to fix and no bounded hold.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

tests/test_dispatch_selection.py

@@ -196,6 +196,19 @@ class ExplicitTargetTransportTests(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(calls[0]['partial_execution_targets'], ['9'])
         self.assertNotIn('partial_execution_targets', calls[1])
 
+    async def test_websocket_connect_timeout_is_a_definite_failure(self):
+        """A timeout before anything was sent must not read as ambiguous acceptance."""
+        never = asyncio.Event()
+
+        async def hang(*_args, **_kwargs):
+            await never.wait()
+
+        session = types.SimpleNamespace(ws_connect=hang)
+        with patch.object(dispatch, "get_client_session", AsyncMock(return_value=session)):
+            with self.assertRaisesRegex(RuntimeError, "nothing was sent") as caught:
+                await dispatch._dispatch_via_websocket("http://worker", {"prompt": {}}, "client", timeout=0.05)
+        self.assertNotIsInstance(caught.exception, dispatch.AmbiguousDispatchError)
+
     async def test_websocket_sends_targets_all_the_way_to_wire(self):
         import json
         sent = []