event_bus.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/event_bus.py
 
"""
In-process publish/subscribe event bus — M2.1 (event substrate).
 
Purpose
-------
A tiny, synchronous pub/sub bus that the M2.x stack wires together: the React
loop and sub-loop (M2.3/M2.4) publish ``TurnEvent`` envelopes; the NDJSON
writer (M2.2), ``/trace`` and ``/packet`` renderers (M2.6), and the replay CLI
(M2.7) subscribe. M2.1 ships only the surface — nothing here imports anything
writer- or loop-shaped.
 
Responsibilities
----------------
- Provide ``EventBus`` with ``subscribe`` / ``unsubscribe`` / ``publish`` and a
  ``Subscription`` handle.
- Dispatch synchronously and thread-safely, isolating subscriber exceptions.
- Provide the lazily-constructed ``default_bus()`` convenience singleton plus a
  test-only ``_reset_default_bus_for_tests()`` hook.
 
Semantics (see plan 095 §5 ADRs)
--------------------------------
- ADR-1: ``publish`` is synchronous — callbacks run on the caller's thread,
  before ``publish`` returns. Events are observability, not control flow.
- ADR-3: each callback runs under ``try / except Exception`` — a buggy
  subscriber is logged and skipped, never propagated. ``KeyboardInterrupt`` and
  ``SystemExit`` are intentionally NOT caught so they still escape.
- ADR-4: a lazily-constructed module-level default bus (``default_bus()``) is a
  convenience for the common one-bus-per-process case; every consumer can also
  accept an injected ``EventBus`` for isolation (tests, sub-agent scopes).
 
Dispatch order: for a published event, kind-specific subscribers fire first (in
subscription order), then ``"*"`` wildcard subscribers (in subscription order).
``publish`` snapshots the subscriber lists under a lock and releases the lock
before invoking callbacks, so a callback may safely subscribe/unsubscribe
without deadlocking — the change takes effect on the *next* publish.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — subscriber callback raised (lifecycle-worthy at default verbosity).
    L3 — unsubscribe of an unknown handle (no-op).
    L4 — module-load coverage marker.
 
Contracts
---------
- ``publish`` is synchronous and never propagates subscriber ``Exception``s;
  ``KeyboardInterrupt`` / ``SystemExit`` are intentionally re-raised.
- Subscriber lists are snapshotted under a lock before dispatch, so callbacks
  may subscribe/unsubscribe during dispatch without effect on the in-flight
  event.
- ``default_bus()`` is a convenience only; every consumer also accepts an
  injected ``EventBus``.
- **Publisher topology (NARROWED, plan 152 R11-02).** The thread-safety contract
  above is about internal consistency, NOT a promise that arbitrary threads
  publish concurrently. In production every ``default_bus().publish(`` call site
  lives on the MAIN thread inside ``run_turn`` (``react_loop.py`` and
  ``sub_loop.py`` — the only two authorised publisher modules; zero publishers
  exist under ``oc_chat/tools/``, so a tool worker outliving a hard-cancelled
  turn still cannot publish). The bus is therefore QUIESCENT between turns by
  design. ``replace_subscription`` relies on exactly this topology: it swaps a
  subscription atomically for callers that hold the between-turns main-thread
  lease, and does NOT drain a publisher that is already mid-dispatch against a
  pre-swap snapshot. A static guard test pins the publisher call sites; adding a
  publisher outside those two modules — or on another thread — breaks that test
  and re-opens the retirement-handle question tracked in plan ``152-b``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.events (TurnEvent / EventKind types, via the contracts bootstrap)
"""
from __future__ import annotations
 
import itertools
import threading
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Literal, Optional, Union
 
from oc_diagnostics import _dbg
 
# Importing from the sibling module reuses its contracts-import bootstrap, so we
# never duplicate the sys.path shim here.
from oc_chat.core.events import EventKind, TurnEvent
 
_dbg("OC_CHAT.MAIN", "event_bus", "module", 4)
 
# The wildcard subscription key: receives every published event.
WILDCARD: Literal["*"] = "*"
 
SubscriptionKey = Union[EventKind, Literal["*"]]
Callback = Callable[[TurnEvent], None]
 
 
@dataclass
class Subscription:
    """Opaque handle returned by :meth:`EventBus.subscribe`.
 
    ``seq`` is a process-monotonic counter that gives deterministic dispatch
    order within a key even when two subscriptions compare equal otherwise.
    """
 
    kind: str
    callback: Callback
    seq: int = field(default=0)
 
 
class EventBus:
    """A synchronous, thread-safe, in-process pub/sub bus for ``TurnEvent``s."""
 
    def __init__(self) -> None:
        self._subscribers: Dict[str, List[Subscription]] = {}
        self._lock = threading.RLock()
        self._seq = itertools.count()
 
    def subscribe(self, kind: SubscriptionKey, callback: Callback) -> Subscription:
        """Register ``callback`` for ``kind`` (an ``EventKind`` or ``"*"``).
 
        Returns a :class:`Subscription` handle; pass it to :meth:`unsubscribe`
        to remove the registration.
        """
        with self._lock:
            sub = Subscription(kind=str(kind), callback=callback, seq=next(self._seq))
            self._subscribers.setdefault(str(kind), []).append(sub)
            return sub
 
    def unsubscribe(self, subscription: Subscription) -> None:
        """Remove a previously returned handle. Unknown handles are a no-op (logged at L3)."""
        with self._lock:
            bucket = self._subscribers.get(subscription.kind)
            if bucket is not None:
                for i, existing in enumerate(bucket):
                    if existing is subscription:
                        del bucket[i]
                        if not bucket:
                            del self._subscribers[subscription.kind]
                        return
        func = "unsubscribe"
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"ignored unknown handle kind={subscription.kind!r} seq={subscription.seq}",
            3,
        )
 
    def replace_subscription(
        self,
        old: Subscription | None,
        kind: SubscriptionKey,
        callback: Callback,
    ) -> Subscription:
        """Atomically swap *old* for a new ``kind``/``callback`` registration.
 
        Remove-then-add under ONE acquisition of the bus lock, so a concurrent
        :meth:`publish` — which snapshots the subscriber lists under the same
        lock before dispatching outside it — observes EXACTLY the old
        registration or EXACTLY the new one, never both and never neither. That
        is the property the session-transition event-writer swap needs (plan 152
        ADR-6): a wildcard writer must never be duplicated or momentarily absent.
 
        *old* may be ``None`` (nothing to remove) or an unknown handle (removal
        is a no-op, matching :meth:`unsubscribe`). The new subscription is
        appended in the normal way, so dispatch order is unchanged.
 
        This is a lock-protected list mutation only — it performs no I/O and
        cannot fail, which is what lets it sit inside a no-fail commit block.
        See the module Contracts note on publisher topology: this method does not
        drain a publisher already mid-dispatch against a pre-swap snapshot,
        because production publishers are main-thread and between-turns
        quiescent.
        """
        func = "replace_subscription"
        with self._lock:
            if old is not None:
                bucket = self._subscribers.get(old.kind)
                if bucket is not None:
                    for i, existing in enumerate(bucket):
                        if existing is old:
                            del bucket[i]
                            if not bucket:
                                del self._subscribers[old.kind]
                            break
            sub = Subscription(kind=str(kind), callback=callback, seq=next(self._seq))
            self._subscribers.setdefault(str(kind), []).append(sub)
        _dbg("OC_CHAT.MAIN", func, f"swapped kind={str(kind)!r} seq={sub.seq}", 3)
        return sub
 
    def publish(self, event: TurnEvent) -> None:
        """Dispatch ``event`` to matching subscribers synchronously.
 
        Snapshots the kind-specific and wildcard subscriber lists under the lock,
        releases the lock, then invokes callbacks (kind-specific first, then
        wildcard, each in subscription order). Subscriber exceptions are caught
        and logged; ``KeyboardInterrupt`` / ``SystemExit`` propagate.
        """
        with self._lock:
            specific = list(self._subscribers.get(str(event.kind), ()))
            wildcard = list(self._subscribers.get(WILDCARD, ()))
 
        for sub in specific:
            self._dispatch_one(sub, event)
        for sub in wildcard:
            self._dispatch_one(sub, event)
 
    @staticmethod
    def _dispatch_one(sub: Subscription, event: TurnEvent) -> None:
        func = "_dispatch_one"
        try:
            sub.callback(event)
        except (KeyboardInterrupt, SystemExit):
            raise
        except Exception as exc:  # noqa: BLE001 - bus must isolate subscriber faults
            _dbg(
                "OC_CHAT.MAIN",
                func,
                f"subscriber_error kind={sub.kind!r} event_kind={event.kind!r} "
                f"seq={sub.seq} exc={type(exc).__name__}: {exc}",
                2,
            )
 
 
_DEFAULT_BUS: Optional[EventBus] = None
_DEFAULT_BUS_LOCK = threading.Lock()
 
 
def default_bus() -> EventBus:
    """Return the lazily-constructed process-wide default bus.
 
    Callers wanting isolation (tests, sub-agent scopes) should construct their
    own ``EventBus()`` and never touch this singleton.
    """
    global _DEFAULT_BUS
    if _DEFAULT_BUS is None:
        with _DEFAULT_BUS_LOCK:
            if _DEFAULT_BUS is None:
                _DEFAULT_BUS = EventBus()
    return _DEFAULT_BUS
 
 
def _reset_default_bus_for_tests() -> None:
    """Clear the default-bus singleton. Test-only — production code never calls this."""
    global _DEFAULT_BUS
    with _DEFAULT_BUS_LOCK:
        _DEFAULT_BUS = None
 
 
__all__ = [
    "EventBus",
    "Subscription",
    "default_bus",
    "WILDCARD",
]