src/genro_asgi/tasks/hub.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """EventHub — in-memory per-session event fan-out (the live courier, ◆D22).16 17 The spool on storage is the SOURCE OF TRUTH for task progress (the worker writes18 ``progress.json`` at each tick); the hub is the LIVE COURIER that carries the same19 event to whoever is watching NOW. A subscriber is a bounded ``asyncio.Queue`` keyed20 by ``session_id`` (the launching MCP session). The executor pairs each21 ``spool.write_progress(...)`` with a ``hub.publish(session_id, event)`` (wired in22 Phase 6); the MCP push channel (Phase 6) subscribes on a GET and drains the queue23 into an SSE stream.24 25 Fire-and-forget, shaped like ``ChannelClient.send`` (channel/client.py) but pure26 in-memory — no transport, no frames, nothing to import. A ``publish`` to a session27 with no subscriber is a no-op: progress is not lost, it lives on the spool; the hub28 only serves live watchers. The queue is bounded and DROPS THE OLDEST event when29 full: progress is idempotent (each event is a snapshot superseding the previous), so30 a slow reader loses intermediate frames, never the meaning.31 32 No durable replay log lives here (that reads as orchestration, D22): resumability is33 snapshot-baseline — a late subscriber replays the current ``progress.json`` snapshot34 (Phase 5/6), then follows the live queue.35 """36 37 from __future__ import annotations38 39 import asyncio40 from typing import Any41 42 __all__ = ["EventHub", "QUEUE_MAXSIZE"]43 44 QUEUE_MAXSIZE = 256 # bounded per-subscriber buffer; drop-oldest when full45 46 47 class EventHub:48 """Per-``session_id`` fan-out of progress events over bounded queues.49 50 Note:51 Pure in-memory state on the instance (``self._subscribers``): a session52 maps to the set of its live queues (one server may serve several SSE53 streams for the same session). No locks — every method runs on the one54 server event loop.55 """56 57 __slots__ = ("_subscribers",)58 59 def __init__(self) -> None:60 """Start with no subscribers."""61 self._subscribers: dict[str, set[asyncio.Queue[Any]]] = {}62 63 def subscribe(self, session_id: str) -> asyncio.Queue[Any]:64 """Register a fresh bounded queue for ``session_id`` and return it.65 66 The caller (the SSE handler) drains the queue until it unsubscribes.67 """68 queue: asyncio.Queue[Any] = asyncio.Queue(maxsize=QUEUE_MAXSIZE)69 self._subscribers.setdefault(session_id, set()).add(queue)70 return queue71 72 def unsubscribe(self, session_id: str, queue: asyncio.Queue[Any]) -> None:73 """Drop ``queue`` from ``session_id`` (the SSE stream closed); tidy up empties."""74 queues = self._subscribers.get(session_id)75 if queues is None:76 return77 queues.discard(queue)78 if not queues:79 del self._subscribers[session_id]80 81 def publish(self, session_id: str, event: Any) -> None:82 """Deliver ``event`` to every live queue of ``session_id`` (no-op if none).83 84 Full queue → drop the oldest event and enqueue the new one: progress is a85 snapshot, so the freshest event is the one that matters.86 """87 for queue in self._subscribers.get(session_id, ()):88 if queue.full():89 queue.get_nowait()90 queue.put_nowait(event)