hub_bus.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/hub_bus.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/unit/test_259_hub_carriers.py"
"""
The in-process hub carrier: a direct call, and a published event beside it.
 
Purpose
-------
Plan 259 P5, from the factory-decomposition design note §4. The hub has
three carriers, and this is the first one tried: when both parties are
in one process -- the orchestrator and an agent it is driving -- the
message goes straight to the recipient's callback rather than out to a
file and back. That is the direct path the operator asked for.
 
Two things keep it honest.
 
**It is chosen by PRESENCE, not by preference.** A party is present here
only while something is attached for its address, so a session that has
gone away simply falls through to the file lane and the message waits on
disk. Nothing is dropped because a listener vanished.
 
**Direct does not mean unlogged.** The hub records before it routes, so
the log's entry is identical whether this carrier or the file lane moved
the message -- design-note invariant 1, and a witness measures it by
comparing two logs rather than reading it off this file. On top of that,
when an event bus is injected the carrier publishes ONE
``TurnEvent(kind="hub_message")`` per delivery, so a session's own event
writer records the fleet traffic that crossed it for free.
 
That published event carries ROUTING ONLY -- no body, no provenance, no
answer. The hub's durable log is the record of what was said; this event
is the record that something was said. A telemetry event anything on the
bus may read must not become a second place a ruling's authority can be
reconstructed from.
 
Responsibilities
----------------
- :class:`BusCarrier` -- ``attach`` / ``detach`` / ``present`` /
  ``deliver``, the :class:`~oc_chat.core.hub.Carrier` contract plus the
  two attachment methods :meth:`~oc_chat.core.hub.Hub.attach` delegates
  to.
- :class:`NoAttachmentError` -- the carrier's named refusal.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (an attach, a delivery)
    L3 - semantic details (a refusal, a publish)
    L4 - deep tracing (module load)
 
Contracts
---------
- **The callback runs BEFORE the event is published.** Delivery is the
  job; telemetry is the by-product, and a bus fault must never eat a
  message that a recipient has already had.
- ``present`` is false for ``operator@remote`` even when something is
  attached for it. An in-process callback cannot be a phone, and saying
  "present" would swallow a message the operator is waiting for. The
  attach is still ALLOWED, so a control centre may watch that traffic;
  it simply never wins the routing decision.
- One address holds ONE callback. Re-attaching replaces; two would
  deliver the same message twice.
- The event is built by ``core/events.make_hub_message``, the one
  factory in ``EVENT_FACTORIES``, so this module cannot drift from the
  kind's declared payload shape.
- The bus is DUCK-TYPED: anything with ``publish``. That keeps the
  hub's closure free of the runtime's event stack while still letting a
  live session pass its real :class:`~oc_chat.core.event_bus.EventBus`.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.hub (the contract; moved there by plan 272, and named
  directly rather than through the shim because this module is the
  consumer-side carrier that STAYED, not a caller that always spelled
  the old path)
- oc_chat.core.events (the one hub_message factory)
"""
 
from __future__ import annotations
 
from typing import Any, Callable
 
from oc_diagnostics import _dbg
 
from oc_chat.core.events import make_hub_message, new_event_id
from oc_orchestrator.hub import (
    ADDRESS_OPERATOR_REMOTE,
    DeliveryReceipt,
    HubError,
    HubMessage,
    parse_address,
)
 
_dbg("OC_CHAT.MAIN", "module", "hub_bus loaded", 4)
 
 
#: The session a published event belongs to when the carrier was built
#: without one. A live session injects its own; this is what a headless
#: orchestrator process reports as.
DEFAULT_BUS_SESSION_ID = "hub"
 
#: The parent every published hub event hangs from. ``TurnEvent`` makes
#: ``parent_event_id`` mandatory for every kind but ``session_started``,
#: and a hub message has no turn to parent onto, so the carrier declares
#: one root rather than inventing a fake turn.
DEFAULT_BUS_PARENT_EVENT_ID = "hub-root"
 
#: ``(message) -> None``; raising is how a reader refuses.
HubCallback = Callable[[HubMessage], None]
 
 
class NoAttachmentError(HubError):
    """No in-process reader is attached for that address."""
 
 
class BusCarrier:
    """Deliver to an attached callback, and publish one event beside it."""
 
    #: What a receipt and a trace call this carrier.
    name = "bus"
 
    def __init__(
        self,
        *,
        bus: Any | None = None,
        session_id: str = DEFAULT_BUS_SESSION_ID,
        parent_event_id: str = DEFAULT_BUS_PARENT_EVENT_ID,
    ) -> None:
        func = "BusCarrier.__init__"
        self._callbacks: dict[str, HubCallback] = {}
        self.bus = bus
        self.session_id = session_id
        self.parent_event_id = parent_event_id
        _dbg("OC_CHAT.MAIN", func,
             f"bus={'injected' if bus is not None else 'none'}", 2)
 
    # -- attachment -------------------------------------------------------
 
    def attach(self, address: str, callback: HubCallback) -> None:
        """Register *callback* as the in-process reader for *address*.
 
        Replaces any existing one: two callbacks for one address would
        deliver the same message twice, which no reader expects.
 
        Raises:
            AddressError: VALIDATION_ERROR (L3) -- from the grammar.
        """
        func = "BusCarrier.attach"
        parse_address(address)
        self._callbacks[address] = callback
        _dbg("OC_CHAT.MAIN", func, f"attached {address}", 2)
 
    def detach(self, address: str) -> None:
        """Remove *address*'s reader.
 
        Raises:
            NoAttachmentError: VALIDATION_ERROR (L3) -- nothing was
                attached. Silence here would hide a mistyped address for
                the life of a session.
        """
        func = "BusCarrier.detach"
        parse_address(address)
        if address not in self._callbacks:
            raise NoAttachmentError(
                f"nothing is attached for {address}; there is no in-process "
                "reader to detach"
            )
        del self._callbacks[address]
        _dbg("OC_CHAT.MAIN", func, f"detached {address}", 2)
 
    # -- the Carrier contract ---------------------------------------------
 
    def present(self, address: str) -> bool:
        """True when something is attached AND the address is reachable here.
 
        ``operator@remote`` is never reachable in-process, whatever is
        attached for it: the address MEANS the operator is away from the
        control centre, and the connector is the only carrier that can
        reach them.
        """
        if address == ADDRESS_OPERATOR_REMOTE:
            return False
        return address in self._callbacks
 
    def deliver(self, message: HubMessage) -> DeliveryReceipt:
        """Call the recipient's reader, then publish one event.
 
        Raises:
            NoAttachmentError: VALIDATION_ERROR (L3) -- nothing is
                attached. The hub only reaches this after ``present``
                said yes, so hitting it means the reader went away
                between the two calls; refusing lets the log's entry
                stand and the file lane redeliver.
        """
        func = "BusCarrier.deliver"
        callback = self._callbacks.get(message.recipient)
        if callback is None:
            raise NoAttachmentError(
                f"nothing is attached for {message.recipient}; the bus carrier "
                f"cannot deliver {message.message_id}"
            )
        callback(message)
        _dbg("OC_CHAT.MAIN", func,
             f"delivered {message.message_id} to {message.recipient}", 2)
        self._publish(message)
        return DeliveryReceipt(
            carrier=self.name, message_id=message.message_id,
            detail=f"attached callback for {message.recipient}",
        )
 
    # -- telemetry --------------------------------------------------------
 
    def _publish(self, message: HubMessage) -> None:
        """Publish one ``hub_message`` event, if a bus was injected.
 
        Never raises into the delivery path: the recipient already has
        the message, and a telemetry failure must not be reported as a
        delivery failure. The bus itself already isolates subscriber
        faults; this guards the construction and the publish call.
        """
        func = "BusCarrier._publish"
        if self.bus is None:
            return
        try:
            self.bus.publish(make_hub_message(
                event_id=new_event_id(),
                parent_event_id=self.parent_event_id,
                session_id=self.session_id,
                ts=message.at,
                message_id=message.message_id,
                hub_kind=message.kind,
                sender=message.sender,
                recipient=message.recipient,
                subject=message.subject,
                in_reply_to=message.in_reply_to,
                summary=message.summary,
            ))
        except Exception as exc:  # noqa: BLE001 - EXTERNAL_ERROR (L2); telemetry is not delivery
            _dbg("OC_CHAT.MAIN", func,
                 # OCT-LINT: disable=no_hardcoded_secrets reason="Diagnostic prose naming the S6 class EXTERNAL_ERROR and the hub kind; scores 4.59 bits/char and carries no credential. Operator-ruled 2026-09-13 (plan 259 escalation, line-scoped), same class as oc_connect/inbound/poller.py:429"
                 f"EXTERNAL_ERROR: could not publish hub_message for "
                 f"{message.message_id} ({type(exc).__name__}: {exc}); the "
                 f"message was delivered regardless", 2)
            return
        _dbg("OC_CHAT.MAIN", func, f"published hub_message {message.message_id}", 3)
 
 
__all__ = [
    "DEFAULT_BUS_PARENT_EVENT_ID",
    "DEFAULT_BUS_SESSION_ID",
    "BusCarrier",
    "HubCallback",
    "NoAttachmentError",
]