orchestrata_ultimate.core.patterns.observer

Reference › orchestrata_ultimate › core › patterns › observer

A generic, thread-safe Observer/Observable implementation in synchronous and asynchronous forms, imported by the base volk class, the pytest live observer in core.threading and the codemod observability events. The event type is generic, so the module carries no opinion about its payload. It is not tied to a specification; the module notes that it is generic layer-1 infrastructure and is deliberately untraced.

Observer and AsyncObserver are protocols with one method, on_event. The subscription list and its RLock are written once in _SubscriptionRegistry, whose subscribe is idempotent and returns an unsubscribe callable, and whose snapshot returns the observers in order; both observables compose it instead of inheriting. Observable is the abstract base with subscribe, unsubscribe, an observers snapshot and an abstract emit. SyncObservable.emit calls each observer in turn, and AsyncObservable.emit awaits them concurrently with asyncio.gather(..., return_exceptions=True). Both apply one error policy: an observer's exception is logged and swallowed, so one failing observer cannot stop the others and emit never raises because of an observer.

At a glance

Class diagram

Dependency diagram

API

Classes

Observer (protocol)

Bases: Protocol[T] · Decorators: runtime_checkable · Line 48

Synchronous observer interface (GoF Observer).

AsyncObserver (protocol)

Bases: Protocol[T] · Decorators: runtime_checkable · Line 65

Asynchronous observer interface (GoF Observer, async variant).

_SubscriptionRegistry

Bases: Generic[T] · Line 77

Thread-safe observer bookkeeping shared, by composition, by both Observable and AsyncObservable.

Observable (abstract)

Bases: ABC, Generic[T] · Line 117

Abstract base for a sync-dispatch Observable subject (GoF Observer).

SyncObservable

Bases: Observable[T] · Line 158

Concrete sync Observable.

AsyncObservable

Bases: Generic[T] · Line 184

Concrete async Observable (GoF Observer, async variant).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/patterns/observer.py

"""Generic, thread-safe Observer/Observable pattern (GoF) — sync and async.

Layer-1 reusable infrastructure, sibling to `core.patterns.error_handling`.
The event/payload type is generic (`TypeVar("T")`) so this module has no
opinion about what it carries; its first real consumer is expected to be a
live test-run event stream (an `Observable[TestEvent]` feeding an NDJSON
sink and/or a live UI), but nothing here is test-run specific.

Subscription bookkeeping (the subscribe/unsubscribe list + lock) is
implemented once, in `_SubscriptionRegistry`, and COMPOSED into both
`Observable` (sync) and `AsyncObservable` rather than duplicated across
them — composition over inheritance, per repo convention.

Thread-safety: this repo runs free-threaded CPython (3.14t), so the GIL
cannot be relied upon to serialize list mutation. `_SubscriptionRegistry`
guards its observer list with a `threading.RLock` around every read and
write.

Error-isolation policy (documented once, applied identically to both
variants): a single misbehaving observer must never prevent the others
from being notified. Both `SyncObservable.emit` and `AsyncObservable.emit`
use **log-and-continue** — an observer's exception is logged
(`logger.exception` / `logger.error`) and swallowed; `emit` itself never
raises because of an observer failure. `AsyncObservable.emit` gets this
"for free" from `asyncio.gather(..., return_exceptions=True)`, which is
also this codebase's existing convention (see `kernel.event_bus`).

Spec: no spec in the requirement index defines a generic GoF Observer
pattern; this module is generic layer-1 infra with no requirement to
trace to, so it is intentionally left untraced (the REQ-100-001 tag
this docstring previously referenced never existed in any spec's
requirements checklist and has been removed).
"""

import asyncio
import logging
import threading
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Generic, Protocol, TypeVar, runtime_checkable

logger = logging.getLogger(__name__)

T = TypeVar("T", contravariant=True)


@runtime_checkable
class Observer(Protocol[T]):
    """Synchronous observer interface (GoF Observer).

    Any object with a matching `on_event` method structurally satisfies
    this protocol — no explicit inheritance required.
    """

    def on_event(self, event: T) -> None:
        """Handle an emitted event.

        Args:
            event: The event payload emitted by the Observable.
        """
        ...  # pragma: no cover


@runtime_checkable
class AsyncObserver(Protocol[T]):
    """Asynchronous observer interface (GoF Observer, async variant)."""

    async def on_event(self, event: T) -> None:
        """Handle an emitted event asynchronously.

        Args:
            event: The event payload emitted by the AsyncObservable.
        """
        ...  # pragma: no cover


class _SubscriptionRegistry(Generic[T]):  # noqa: UP046 (brief: typing.Generic[T])
    """Thread-safe observer bookkeeping shared, by composition, by both
    `Observable` and `AsyncObservable`.

    Not part of the public surface of this module — subclasses/consumers
    should go through `Observable`/`AsyncObservable`, not this class
    directly.
    """

    def __init__(self) -> None:
        self._observers: list[T] = []
        self._lock = threading.RLock()

    def subscribe(self, observer: T) -> Callable[[], None]:
        """Subscribe `observer`; idempotent if already subscribed.

        Returns:
            A zero-arg callable that unsubscribes `observer` when invoked.
            Calling it more than once is safe (unsubscribe is a no-op for
            non-members).
        """
        with self._lock:
            if observer not in self._observers:
                self._observers.append(observer)
        return lambda: self.unsubscribe(observer)

    def unsubscribe(self, observer: T) -> None:
        """Unsubscribe `observer`; a no-op if it isn't currently subscribed."""
        with self._lock:
            try:
                self._observers.remove(observer)
            except ValueError:
                pass

    def snapshot(self) -> tuple[T, ...]:
        """Read-only snapshot of currently subscribed observers, in order."""
        with self._lock:
            return tuple(self._observers)


class Observable(ABC, Generic[T]):  # noqa: UP046 (brief: typing.Generic[T])
    """Abstract base for a sync-dispatch Observable subject (GoF Observer).

    Subscription machinery is composed from a shared `_SubscriptionRegistry`
    rather than reimplemented per subclass; concrete subclasses need only
    implement `emit`.
    """

    def __init__(self) -> None:
        self._registry: _SubscriptionRegistry[Observer[T]] = _SubscriptionRegistry()

    def subscribe(self, observer: Observer[T]) -> Callable[[], None]:
        """Subscribe `observer` to future emitted events.

        Args:
            observer: The sync observer to notify on `emit`.

        Returns:
            An idempotent zero-arg unsubscribe handle.
        """
        return self._registry.subscribe(observer)

    def unsubscribe(self, observer: Observer[T]) -> None:
        """Unsubscribe `observer` (a no-op if not currently subscribed)."""
        self._registry.unsubscribe(observer)

    @property
    def observers(self) -> tuple[Observer[T], ...]:
        """Read-only snapshot of currently subscribed observers, in order."""
        return self._registry.snapshot()

    @abstractmethod
    def emit(self, event: T) -> None:
        """Emit `event` to all subscribed observers.

        Args:
            event: The event payload to deliver.
        """
        raise NotImplementedError


class SyncObservable(Observable[T]):
    """Concrete sync Observable.

    Error-isolation policy (log-and-continue): `emit` notifies every
    subscribed observer, in subscription order, even if an earlier one
    raises. A raising observer's exception is logged via
    `logger.exception` and swallowed — it is never re-raised, so one
    misbehaving observer can never block delivery to the rest.
    """

    def emit(self, event: T) -> None:
        """Notify all subscribed observers of `event`.

        See the class docstring for the error-isolation policy.
        """
        for observer in self.observers:
            try:
                observer.on_event(event)
            except Exception:
                logger.exception(
                    "Observer %r raised while handling event; continuing "
                    "with remaining observers",
                    observer,
                )


class AsyncObservable(Generic[T]):  # noqa: UP046 (brief: typing.Generic[T])
    """Concrete async Observable (GoF Observer, async variant).

    Mirrors `Observable`'s subscribe/unsubscribe contract but dispatches to
    `AsyncObserver`s concurrently. Composes the same `_SubscriptionRegistry`
    used by `Observable`/`SyncObservable` rather than duplicating the
    lock/list bookkeeping.

    Error-isolation policy (log-and-continue): `emit` awaits every
    subscribed observer concurrently via
    `asyncio.gather(..., return_exceptions=True)`, so one failing
    observer's exception can never prevent the others from running to
    completion. Exceptions are logged via `logger.error` and swallowed;
    `emit` itself never raises because of an observer failure.
    """

    def __init__(self) -> None:
        self._registry: _SubscriptionRegistry[AsyncObserver[T]] = (
            _SubscriptionRegistry()
        )

    def subscribe(self, observer: AsyncObserver[T]) -> Callable[[], None]:
        """Subscribe `observer` to future emitted events.

        Args:
            observer: The async observer to notify on `emit`.

        Returns:
            An idempotent zero-arg unsubscribe handle.
        """
        return self._registry.subscribe(observer)

    def unsubscribe(self, observer: AsyncObserver[T]) -> None:
        """Unsubscribe `observer` (a no-op if not currently subscribed)."""
        self._registry.unsubscribe(observer)

    @property
    def observers(self) -> tuple[AsyncObserver[T], ...]:
        """Read-only snapshot of currently subscribed observers, in order."""
        return self._registry.snapshot()

    async def emit(self, event: T) -> None:
        """Notify all subscribed observers of `event`, concurrently.

        See the class docstring for the error-isolation policy.
        """
        observers = self.observers
        if not observers:
            return
        results = await asyncio.gather(
            *(observer.on_event(event) for observer in observers),
            return_exceptions=True,
        )
        for observer, result in zip(observers, results, strict=True):
            if isinstance(result, BaseException):
                logger.error(
                    "Observer %r raised while handling event: %s",
                    observer,
                    result,
                    exc_info=result,
                )