subscriber

subscriber

Render clearance-hub events as desktop notifications.

EventSubscriber turns the event stream from one ClearanceClient into calls on an injected Notifier: a block arrives, the operator sees a popup with Allow/Deny actions, clicks route back to the hub as a Verdict. Live-block dedup and shield-down popup tracking live here because they're presentation concerns; identity resolution does not — the shield reader resolves the orchestrator-supplied dossier at emit time and ships it on every event, so the renderer just reads it.

MultiSocketSubscriber fans several per-container hubs into one notifier: it owns an EventSubscriber per hub socket and reconciles the set as containers come and go.

NOTIFY_BLOCKED = 'blocked' module-attribute

Category for connection_blocked Allow/Deny prompts (interactive).

NOTIFY_VERDICT = 'verdict' module-attribute

Category for verdict_applied resolution popups (ack of a verdict).

NOTIFY_SHIELD_UP = 'shield_up' module-attribute

Category for ShieldUp confirmation popups.

NOTIFY_SHIELD_DOWN = 'shield_down' module-attribute

Category for ShieldDown / ShieldDisengaged security-alert popups.

NOTIFY_CONTAINER_STARTED = 'container_started' module-attribute

Category for ContainerStarted lifecycle popups.

NOTIFY_CONTAINER_EXITED = 'container_exited' module-attribute

Category for ContainerExited lifecycle popups.

ALL_NOTIFY_CATEGORIES = frozenset({NOTIFY_BLOCKED, NOTIFY_VERDICT, NOTIFY_SHIELD_UP, NOTIFY_SHIELD_DOWN, NOTIFY_CONTAINER_STARTED, NOTIFY_CONTAINER_EXITED}) module-attribute

Every category the subscriber recognises — the default opt-in set.

EventSubscriber(notifier, client=None, *, socket_path=None, enabled_categories=None)

Bridge clearance-hub events into desktop notifications.

Owns the presentation-layer state a rendering client needs: live-block dedup keyed on (container, target), the tracked ShieldDown popup per container so ShieldUp can retire it, and verdict routing through notifier action callbacks.

Parameters:

Name	Type	Description	Default
notifier	Notifier	Desktop notification backend (any Notifier works).	required
client	ClearanceClient | None	Pre-configured ClearanceClient. When omitted, one is created on start pointing at socket_path (defaulting to default_clearance_socket_path).	None
socket_path	Path | None	Clearance-socket override when client isn't supplied (tests).	None
enabled_categories	Set[str] | None	Subset of ALL_NOTIFY_CATEGORIES whose popups should reach the notifier. None (the default) enables every category and matches the historical "render everything" behaviour — appropriate for the in-TUI subscriber that builds its own UI on top of the full event stream. The desktop notifier daemon narrows this to the interactive subset {blocked, verdict} so passive shield/container popups don't overrun the operator's tray. Lifecycle hook dispatch (_dispatch_lifecycle) and pending-state cleanup remain unconditional — the gate fires only on the notify() call itself, so an embedded consumer's lifecycle callbacks keep firing regardless of which categories are silenced.	None

Initialise the subscriber with a notifier and transport.

Source code in src/terok_clearance/client/subscriber.py

def __init__(
    self,
    notifier: Notifier,
    client: ClearanceClient | None = None,
    *,
    socket_path: Path | None = None,
    enabled_categories: AbstractSet[str] | None = None,
) -> None:
    """Initialise the subscriber with a notifier and transport."""
    self._notifier = notifier
    self._client = client or ClearanceClient(socket_path=socket_path)
    self._enabled_categories: frozenset[str] = (
        ALL_NOTIFY_CATEGORIES
        if enabled_categories is None
        else frozenset(enabled_categories) & ALL_NOTIFY_CATEGORIES
    )
    # request_id → pending block + its notification.
    self._pending: dict[str, _PendingBlock] = {}
    # container → notification_id of the active ShieldDown popup, so
    # ShieldUp can close the matching one before firing its brief
    # confirmation.  A stale "Shield DOWN" popup after shield is back
    # is a security hazard, not a benign leftover.
    self._shield_down_notifs: dict[str, int] = {}
    # Background action / lifecycle tasks we spawn.
    self._tasks: set[asyncio.Task[None]] = set()

start() async

Connect to the clearance hub and begin rendering its event stream.

Source code in src/terok_clearance/client/subscriber.py

async def start(self) -> None:
    """Connect to the clearance hub and begin rendering its event stream."""
    await self._client.start(self._on_event)
    _log.info("clearance subscriber online")

stop() async

Drain pending tasks and close the transport.

Closes the client first so no new handler tasks are scheduled, then awaits the currently-tracked tasks to settle (with their own CancelledError suppressed). A bare sleep(0) would yield only one loop turn — not enough for cancellation to propagate through chained awaits — and tasks.clear() on its own would drop references to tasks still writing to handles we then close.

Source code in src/terok_clearance/client/subscriber.py

async def stop(self) -> None:
    """Drain pending tasks and close the transport.

    Closes the client first so no new handler tasks are scheduled,
    then awaits the currently-tracked tasks to settle (with their
    own ``CancelledError`` suppressed).  A bare ``sleep(0)`` would
    yield only one loop turn — not enough for cancellation to
    propagate through chained awaits — and ``tasks.clear()`` on its
    own would drop references to tasks still writing to handles we
    then close.
    """
    tasks = list(self._tasks)
    for task in tasks:
        task.cancel()
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)
    self._tasks.clear()
    await self._client.stop()
    self._pending.clear()
    self._shield_down_notifs.clear()

poke_reconnect()

Cut short any in-flight reconnect back-off — forwards to the client.

Source code in src/terok_clearance/client/subscriber.py

def poke_reconnect(self) -> None:
    """Cut short any in-flight reconnect back-off — forwards to the client."""
    self._client.poke_reconnect()

MultiSocketSubscriber(notifier, *, socket_glob=None, enabled_categories=None, rescan_interval_s=2.0)

Subscribe to every per-container clearance hub socket under a glob.

The per-container-supervisor model gives every container its own hub socket at $XDG_RUNTIME_DIR/terok/clearance/<container_id>.sock, so an operator UI that wants the union of every supervisor's event stream must multiplex across the set. This class watches socket_glob (default $XDG_RUNTIME_DIR/terok/clearance/*.sock), opens an EventSubscriber against each matching path on start, re-scans every rescan_interval_s seconds to pick up sockets that appeared after start, and tears down child subscribers whose sockets vanish. All events fan into the same notifier.

A simple poll loop drives the rescan — no inotify dependency. The overhead is one glob.glob call per interval, which is cheap compared to the cost of carrying a fanotify watch fd for what is effectively a directory that mutates only at container start / stop.

Parameters:

Name	Type	Description	Default
notifier	Notifier	Desktop notification backend (any Notifier works). All child subscribers fan their events into this single notifier so the operator sees one merged stream.	required
socket_glob	str | None	Filesystem glob of clearance hub sockets to track. None (the default) derives $XDG_RUNTIME_DIR/terok/clearance/*.sock from the current runtime dir.	None
enabled_categories	Set[str] | None	Subset of ALL_NOTIFY_CATEGORIES propagated to every child EventSubscriber. None enables every category.	None
rescan_interval_s	float	Seconds between glob rescans. Must be strictly positive. The default (2.0) balances startup latency for a freshly-spawned supervisor against background polling cost when the host is idle.	2.0

Capture notifier + scan parameters; sockets are discovered in start.

Raises:

Type	Description
ValueError	If rescan_interval_s is not strictly positive. 0 would spin the rescan loop with no delay; a negative value would make asyncio.sleep raise.

Source code in src/terok_clearance/client/subscriber.py

def __init__(
    self,
    notifier: Notifier,
    *,
    socket_glob: str | None = None,
    enabled_categories: AbstractSet[str] | None = None,
    rescan_interval_s: float = 2.0,
) -> None:
    """Capture *notifier* + scan parameters; sockets are discovered in `start`.

    Raises:
        ValueError: If *rescan_interval_s* is not strictly positive.
            ``0`` would spin the rescan loop with no delay; a negative
            value would make ``asyncio.sleep`` raise.
    """
    if rescan_interval_s <= 0:
        raise ValueError(f"rescan_interval_s must be > 0, got {rescan_interval_s!r}")
    self._notifier = notifier
    self._socket_glob = socket_glob or _default_socket_glob()
    self._enabled_categories: frozenset[str] | None = (
        None if enabled_categories is None else frozenset(enabled_categories)
    )
    self._rescan_interval_s = rescan_interval_s
    self._subscribers: dict[str, EventSubscriber] = {}
    self._rescan_task: asyncio.Task[None] | None = None
    self._stopping = False

start() async

Connect to every currently-matching socket and start the rescan loop.

Sockets that fail to connect at start are logged and skipped — a flaky supervisor doesn't stop the rest from coming online. The rescan loop runs as a background task so newly-spawned supervisors join the merged stream without a manual restart.

Idempotent: a second start() while the rescan loop is still live is a no-op (the alternative would orphan the existing task and leak a loop that keeps reconciling).

Source code in src/terok_clearance/client/subscriber.py

async def start(self) -> None:
    """Connect to every currently-matching socket and start the rescan loop.

    Sockets that fail to connect at start are logged and skipped —
    a flaky supervisor doesn't stop the rest from coming online.
    The rescan loop runs as a background task so newly-spawned
    supervisors join the merged stream without a manual restart.

    Idempotent: a second ``start()`` while the rescan loop is still
    live is a no-op (the alternative would orphan the existing task
    and leak a loop that keeps reconciling).
    """
    if self._rescan_task is not None and not self._rescan_task.done():
        return
    self._stopping = False
    await self._reconcile()
    self._rescan_task = asyncio.create_task(self._rescan_loop())
    _log.info(
        "multi-socket subscriber online (glob=%s, %d socket(s))",
        self._socket_glob,
        len(self._subscribers),
    )

stop() async

Cancel the rescan loop and stop every child subscriber.

Stops are awaited in parallel because each child holds its own transports — serialising would extend shutdown latency by the sum of every disconnect timeout instead of the worst-case one.

Source code in src/terok_clearance/client/subscriber.py

async def stop(self) -> None:
    """Cancel the rescan loop and stop every child subscriber.

    Stops are awaited in parallel because each child holds its own
    transports — serialising would extend shutdown latency by the
    sum of every disconnect timeout instead of the worst-case one.
    """
    self._stopping = True
    if self._rescan_task is not None:
        self._rescan_task.cancel()
        with contextlib.suppress(asyncio.CancelledError, Exception):
            await self._rescan_task
        self._rescan_task = None
    children = list(self._subscribers.values())
    self._subscribers.clear()
    if children:
        await asyncio.gather(*(child.stop() for child in children), return_exceptions=True)