hub.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/hub.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_orchestrator/tests/test_259_hub.py"
"""
The hub's contract: one address grammar, six kinds, one human rule.
 
Purpose
-------
Plan 259 P2, from the factory-decomposition design note §4. The hub is a
MESSAGE CONTRACT, not a transport. Two pipes and a reply path already
exist -- the worker-report lane, the session-alert lane, and the
connector's correlated inbound reply -- and what turns them into a hub is
three additions, all declared here:
 
**Addresses**, so a message is ROUTED rather than deposited. Four forms
and no others: ``orchestrator``, ``operator@local`` (the control centre,
which today is the CLI in the operator's own session), ``operator@remote``
(the connector), and ``agent-NNN`` for each worker instance.
 
**Kinds with replies.** ``report`` and ``alert`` flow one way. An ``ask``
expects an answer and a ``ruling`` IS that answer, naming the ask it
answers, so a question any human channel can answer is a first-class
thing. ``dispatch`` and ``instruction`` are records of what was done and
what was told, so the orchestrator's picture of the fleet is the log and
not a guess.
 
**Provenance on every message, and a rule about rulings.** A ruling
carries who answered and through which channel. This is not a nicety: it
derives from an operator ruling already in force, that a security
approval must be in the operator's own voice and never relayed by an
agent. In a hub where the orchestrator agent, the control centre and
Telegram can all answer, an unsigned reply makes an agent-relayed yes
indistinguishable from the human's. :func:`check_ruling_provenance` is
the ONE place that rule lives.
 
Responsibilities
----------------
- :func:`parse_address` -- the whole grammar, refusing by name. Agent
  ids are validated by
  :func:`oc_orchestrator.agent_instances.validate_agent_id`, IMPORTED
  rather than reimplemented, so there is one definition of ``agent-NNN``
  in the tree.
- :data:`SUPPORTED_KINDS` -- the six, checked in one place.
- :class:`Provenance` / :class:`HubMessage` -- frozen value types;
  :func:`new_message` is the one construction path.
- :func:`check_ruling_provenance` -- the human rule.
- :class:`HubLog` / :class:`Carrier` -- the two Protocols the backends
  and the carriers satisfy.
- :class:`Hub` -- ``send`` (record, then route) and ``rule`` (build the
  ruling from the ask and send it).
- :class:`HubError` and its named subclasses -- every refusal.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (a message sent, a carrier chosen)
    L3 - semantic details (a refusal, with what it refused)
    L4 - deep tracing (module load)
 
Contracts
---------
- **The log is the truth.** :meth:`Hub.send` RECORDS BEFORE IT DELIVERS,
  and a carrier failure after the record is a :class:`CarrierError` with
  the entry left standing -- the file lane redelivers, and a message
  that reached no carrier is still a message that was sent. That is
  design-note invariant 1 made mechanical: direct does not mean unlogged.
- **A ruling on a human-required ask is refused from any agent actor.**
  Not discouraged, not flagged: refused, by construction, in
  :func:`check_ruling_provenance`, which :meth:`Hub.send` calls before
  it records. A refused ruling is never recorded and never delivered.
- ``answer`` is one of :data:`RULING_ANSWERS` -- the connector's reply
  vocabulary -- so a Telegram tap, a CLI word and a control-centre
  button all produce the same four values.
- :class:`Provenance.correlation` is an IMMUTABLE mapping. A correlation
  a caller can edit after the check is not evidence.
- ``message_id`` is a ULID: :func:`new_ulid` and the Crockford alphabet
  are lifted VERBATIM from ``plan_status.py:450-464`` and ``:132``,
  waiver text included, because the lane readers compare these bytes and
  a second algorithm here would be a second answer. The two now sit in
  one package and could in principle be deduplicated; plan 272 P8 leaves
  that to the follow-up that also settles the comms-root rule, because a
  shared ULID module is a decision about the package's shape rather than
  a consequence of the move.
- This module imports nothing but the standard library, the diagnostics
  runtime and the one agent-id validator beside it -- no registry, no
  consumer, no toolchain. A witness measures that with the committed
  extractor, and ``tests/test_t1_leaf_guard.py`` measures it statically.
- Nothing here reads the process working directory, opens a file, or
  starts a subprocess. Routing is a decision; moving bytes is a carrier's
  job.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.agent_instances (the ONE definition of ``agent-NNN``)
"""
 
from __future__ import annotations
 
import secrets
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Any, Callable, Protocol, runtime_checkable
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.agent_instances import validate_agent_id
 
_dbg("OC_CHAT.MAIN", "module", "hub loaded", 4)
 
 
# ── Addresses ────────────────────────────────────────────────────────────
 
 
#: The orchestrator agent.
ADDRESS_ORCHESTRATOR = "orchestrator"
 
#: The operator at the control centre. Today that is the CLI in the
#: operator's own session; when a web control centre exists it is that,
#: and nothing in this module changes.
ADDRESS_OPERATOR_LOCAL = "operator@local"
 
#: The operator, reachable only through the connector.
ADDRESS_OPERATOR_REMOTE = "operator@remote"
 
#: The three fixed addresses. A worker is ``agent-NNN`` and is validated
#: by the instance module rather than by a fourth literal.
FIXED_ADDRESSES: tuple[str, ...] = (
    ADDRESS_ORCHESTRATOR,
    ADDRESS_OPERATOR_LOCAL,
    ADDRESS_OPERATOR_REMOTE,
)
 
 
# ── Kinds ────────────────────────────────────────────────────────────────
 
 
KIND_DISPATCH = "dispatch"
KIND_REPORT = "report"
KIND_ALERT = "alert"
KIND_ASK = "ask"
KIND_RULING = "ruling"
KIND_INSTRUCTION = "instruction"
 
#: The six kinds, closed. Read at call time, so a seventh kind would be
#: one edit here plus whatever that kind's own rule needs.
SUPPORTED_KINDS: frozenset[str] = frozenset(
    {KIND_DISPATCH, KIND_REPORT, KIND_ALERT, KIND_ASK, KIND_RULING, KIND_INSTRUCTION}
)
 
#: What an ask may require of its answerer.
REQUIRES_HUMAN = "human"
REQUIRES_ORCHESTRATOR = "orchestrator"
SUPPORTED_REQUIRES: tuple[str, ...] = (REQUIRES_HUMAN, REQUIRES_ORCHESTRATOR)
 
#: The connector's reply vocabulary, restated so a ruling built in any
#: channel carries a value the Telegram keyboard could also have produced.
#: Mirrored rather than imported: this module is a dispatch-layer leaf and
#: must not pull the connector contracts into that closure.
RULING_ANSWERS: tuple[str, ...] = ("YES", "NO", "ACK", "HOLD")
 
#: The registry's three actor types, checked by name for the same reason:
#: the registry log is one backend, and a claim written in the file log
#: must use the registry's own vocabulary or the two cannot be compared.
ACTOR_TYPES: tuple[str, ...] = (
    "human_operator",
    "orchestrator_agent",
    "worker_agent",
)
 
#: How a message reached the hub. ``session`` is where an AGENT speaks;
#: ``cli`` and ``lane`` and ``telegram`` all leave a trace off-process.
CHANNELS: tuple[str, ...] = ("session", "cli", "lane", "telegram")
 
#: The channels a HUMAN answer may arrive on. A human at the control
#: centre types into a terminal (``cli``) or taps on their phone
#: (``telegram``); ``session`` is the agent's own channel and ``lane`` is
#: a file an agent wrote, so neither can carry a human's voice.
HUMAN_CHANNELS: tuple[str, ...] = ("cli", "telegram")
 
#: The actor type a human-required ruling must carry.
HUMAN_ACTOR_TYPE = "human_operator"
 
 
# ── ULID ─────────────────────────────────────────────────────────────────
 
 
#: Crockford base32, the ULID alphabet: no I, L, O or U. Lifted VERBATIM
#: from ``scripts/plan_status.py:126-127`` including its ruled waiver,
#: because the lane readers compare these bytes and a derivation here
#: would be a second answer to a settled question.
# OCT-LINT: disable=no_hardcoded_secrets reason="Public standard encoding alphabet; its 5.00 bits/char is the arithmetic maximum for any 32-distinct-character string and carries no credential. Operator-ruled 2026-09-01 (plan 230 W1 escalation), same constant as oc_connect.comms and scripts/plan_status.py"
CROCKFORD_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
 
 
def _encode(value: int, length: int) -> str:
    digits = []
    for _ in range(length):
        value, remainder = divmod(value, 32)
        digits.append(CROCKFORD_ALPHABET[remainder])
    return "".join(reversed(digits))
 
 
def new_ulid(now: datetime, *, entropy: bytes | None = None) -> str:
    """Return a 26-character ULID: 48 bits of time, 80 bits of randomness.
 
    Lifted VERBATIM from ``scripts/plan_status.py:445-460``, which says
    the same of ``oc_connect.comms.new_ulid``. Time-first encoding makes
    lexical order arrival order, which is what lets a reader walk the hub
    lane with a plain directory sort, exactly as the other two lanes are
    walked.
    """
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    milliseconds = int(now.astimezone(timezone.utc).timestamp() * 1000)
    raw = entropy if entropy is not None else secrets.token_bytes(10)
    if len(raw) != 10:
        raise ValueError("ULID entropy must be exactly ten bytes")
    return _encode(milliseconds, 10) + _encode(int.from_bytes(raw, "big"), 16)
 
 
def _now_iso() -> str:
    """A local-with-offset ISO 8601 stamp, seconds resolution.
 
    The project-wide convention, matching ``work_queue_file._now_iso``
    and ``plan_inventory._now_iso``. An offset is not decoration: a
    stamp without one is not a point in time.
    """
    return datetime.now().astimezone().replace(microsecond=0).isoformat()
 
 
# ── Refusals ─────────────────────────────────────────────────────────────
 
 
class HubError(Exception):
    """Base class for every refusal the hub makes."""
 
 
class AddressError(HubError):
    """A string is not one of the four address forms."""
 
 
class KindError(HubError):
    """A message names a kind the hub does not carry, or breaks that kind's shape."""
 
 
class ProvenanceError(HubError):
    """A message's provenance does not support what the message claims."""
 
 
class CarrierError(HubError):
    """No carrier would take the message, or the one that did failed.
 
    Raised AFTER the entry is recorded, always. The log is the truth and
    a message nobody carried is still a message that was sent.
    """
 
 
class UnknownAskError(HubError):
    """A ruling names something that is not an open ask."""
 
 
# ── The grammar ──────────────────────────────────────────────────────────
 
 
def parse_address(value: str) -> str:
    """Return *value* if it is an address, else refuse it by name.
 
    The whole grammar, in one function. ``agent-NNN`` is delegated to
    :func:`oc_orchestrator.agent_instances.validate_agent_id` -- imported,
    never reimplemented -- so the tree has exactly one definition of what
    an instance identity looks like and a change there is a change here.
 
    Raises:
        AddressError: VALIDATION_ERROR (L3) -- the message names the
            value it refused and every form it accepts.
    """
    func = "parse_address"
    if isinstance(value, str) and value in FIXED_ADDRESSES:
        return value
    if isinstance(value, str) and value.startswith("agent-"):
        try:
            return validate_agent_id(value)
        except ValueError:
            pass
    _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: address {value!r}", 3)
    raise AddressError(
        f"{value!r} is not a hub address; expected one of "
        f"{', '.join(FIXED_ADDRESSES)}, or an agent-NNN instance identity"
    )
 
 
def check_kind(kind: str) -> None:
    """Refuse *kind* unless it is in :data:`SUPPORTED_KINDS`, by name."""
    func = "check_kind"
    if kind not in SUPPORTED_KINDS:
        accepted = ", ".join(sorted(SUPPORTED_KINDS))
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: kind {kind!r}", 3)
        raise KindError(f"unknown hub kind {kind!r}; the hub carries: {accepted}")
 
 
# ── Value types ──────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class Provenance:
    """Who is speaking, through which channel, and what act proves it.
 
    ``correlation`` is what makes the claim checkable, and what it holds
    depends on the channel: for ``telegram`` the inbound message id, the
    audit id and the content hash, which together name one tap the
    connector recorded; for ``lane`` the source lane and event id; for
    ``cli`` the process user and working directory; empty for
    ``session``, because an agent speaking in its own session has no
    off-process act to point at -- which is exactly why the human rule
    refuses that channel.
 
    It is stored as an immutable mapping: a correlation a caller can edit
    after :func:`check_ruling_provenance` has passed is not evidence.
    """
 
    #: An address -- who the claim is made on behalf of.
    actor: str
    #: One of :data:`ACTOR_TYPES`, the registry's own three.
    actor_type: str
    #: One of :data:`CHANNELS`.
    channel: str
    #: Channel-specific identifiers; immutable once constructed.
    correlation: Mapping[str, str] = field(default_factory=dict)
 
    def __post_init__(self) -> None:
        """Validate the three vocabularies and freeze the correlation."""
        func = "Provenance.__post_init__"
        parse_address(self.actor)
        if self.actor_type not in ACTOR_TYPES:
            _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: actor_type {self.actor_type!r}", 3)
            raise ProvenanceError(
                f"actor_type {self.actor_type!r} is not one of the registry's "
                f"three: {', '.join(ACTOR_TYPES)}"
            )
        if self.channel not in CHANNELS:
            _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: channel {self.channel!r}", 3)
            raise ProvenanceError(
                f"channel {self.channel!r} is not declared; the hub knows: "
                f"{', '.join(CHANNELS)}"
            )
        frozen = MappingProxyType(
            {str(key): str(value) for key, value in dict(self.correlation).items()}
        )
        object.__setattr__(self, "correlation", frozen)
 
    def as_dict(self) -> dict[str, object]:
        """This provenance as plain JSON-safe data, for a log line or a row."""
        return {
            "actor": self.actor,
            "actor_type": self.actor_type,
            "channel": self.channel,
            "correlation": dict(self.correlation),
        }
 
    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> "Provenance":
        """Rebuild a provenance from what :meth:`as_dict` wrote."""
        correlation = raw.get("correlation") or {}
        if not isinstance(correlation, Mapping):
            raise ProvenanceError("provenance correlation must be a mapping")
        return cls(
            actor=str(raw.get("actor", "")),
            actor_type=str(raw.get("actor_type", "")),
            channel=str(raw.get("channel", "")),
            correlation={str(k): str(v) for k, v in correlation.items()},
        )
 
 
@dataclass(frozen=True)
class HubMessage:
    """One addressed message. Construct through :func:`new_message`."""
 
    #: A 26-character ULID; lexical order is arrival order.
    message_id: str
    #: ISO 8601 WITH an explicit offset.
    at: str
    #: One of :data:`SUPPORTED_KINDS`.
    kind: str
    #: An address.
    sender: str
    #: An address.
    recipient: str
    #: The work item this is about -- a plan id such as ``258-work-queue``
    #: -- or ``None`` when the message is about nothing in particular.
    subject: str | None
    #: The text. The hub never interprets it.
    body: str
    #: The ask a ruling answers; ``None`` for every other kind.
    in_reply_to: str | None = None
    #: For an ask, one of :data:`SUPPORTED_REQUIRES`; ``None`` otherwise.
    requires: str | None = None
    #: For a ruling, one of :data:`RULING_ANSWERS`; ``None`` otherwise.
    answer: str | None = None
    #: Who is speaking and what proves it.
    provenance: Provenance = field(
        default_factory=lambda: Provenance(
            actor=ADDRESS_ORCHESTRATOR, actor_type="orchestrator_agent",
            channel="session",
        )
    )
 
    @property
    def summary(self) -> str:
        """The first non-blank line of the body -- what a lane envelope carries."""
        for line in self.body.splitlines():
            if line.strip():
                return line.strip()
        return ""
 
    def as_dict(self) -> dict[str, object]:
        """This message as plain JSON-safe data."""
        return {
            "message_id": self.message_id,
            "at": self.at,
            "kind": self.kind,
            "sender": self.sender,
            "recipient": self.recipient,
            "subject": self.subject,
            "body": self.body,
            "in_reply_to": self.in_reply_to,
            "requires": self.requires,
            "answer": self.answer,
            "provenance": self.provenance.as_dict(),
        }
 
    @classmethod
    def from_mapping(cls, raw: Mapping[str, object]) -> "HubMessage":
        """Rebuild a message from what :meth:`as_dict` wrote.
 
        Deliberately NOT re-validated through :func:`new_message`: this
        reads back what the hub itself recorded, and a stored message
        that no longer passes today's rules is a fact about the log, not
        an input to reject. The readers that care assert their own
        invariants; this one reconstructs.
        """
        return cls(
            message_id=str(raw["message_id"]),
            at=str(raw["at"]),
            kind=str(raw["kind"]),
            sender=str(raw["sender"]),
            recipient=str(raw["recipient"]),
            subject=_optional_str(raw.get("subject")),
            body=str(raw.get("body", "")),
            in_reply_to=_optional_str(raw.get("in_reply_to")),
            requires=_optional_str(raw.get("requires")),
            answer=_optional_str(raw.get("answer")),
            provenance=Provenance.from_mapping(
                raw.get("provenance") or {}  # type: ignore[arg-type]
            ),
        )
 
 
def _optional_str(value: object) -> str | None:
    return None if value is None else str(value)
 
 
@dataclass(frozen=True)
class HubEntry:
    """One recorded message: its position in the log, and the message."""
 
    #: 1-based, monotonic within one backend's log.
    seq: int
    message: HubMessage
 
 
@dataclass(frozen=True)
class DeliveryReceipt:
    """What a carrier reports when it moved one message."""
 
    carrier: str
    message_id: str
    #: Whatever the carrier wants the caller to see -- a path, a send id.
    detail: str | None = None
 
 
# ── Construction ─────────────────────────────────────────────────────────
 
 
def new_message(
    *,
    kind: str,
    sender: str,
    recipient: str,
    body: str,
    provenance: Provenance,
    subject: str | None = None,
    in_reply_to: str | None = None,
    requires: str | None = None,
    answer: str | None = None,
    now: datetime | None = None,
    entropy: bytes | None = None,
) -> HubMessage:
    """Build one validated :class:`HubMessage`, or refuse it by name.
 
    The ONE construction path, so every shape rule is checked in one
    place: the kind is in the closed set, the two addresses parse, and
    the three kind-conditional fields agree with the kind. A ruling must
    name the ask it answers and carry an answer from the reply
    vocabulary; only an ask may declare what it requires.
 
    Raises:
        AddressError: VALIDATION_ERROR (L3) -- a party is not an address.
        KindError: VALIDATION_ERROR (L3) -- the kind is unknown, or a
            kind-conditional field is present where it may not be, or
            absent where it must be.
        ProvenanceError: VALIDATION_ERROR (L3) -- from
            :class:`Provenance`'s own validation.
    """
    func = "new_message"
    check_kind(kind)
    parse_address(sender)
    parse_address(recipient)
 
    if kind == KIND_RULING:
        if not in_reply_to:
            raise KindError(
                "a ruling must name the ask it answers in in_reply_to; a "
                "ruling that answers nothing is a claim, not an answer"
            )
        if answer not in RULING_ANSWERS:
            raise KindError(
                f"a ruling's answer must be one of {', '.join(RULING_ANSWERS)}; "
                f"got {answer!r}. That vocabulary is the connector's, so any "
                "human channel can produce it."
            )
    else:
        if in_reply_to is not None:
            raise KindError(f"only a ruling may set in_reply_to; {kind!r} may not")
        if answer is not None:
            raise KindError(f"only a ruling may carry an answer; {kind!r} may not")
 
    if kind == KIND_ASK:
        if requires not in SUPPORTED_REQUIRES:
            raise KindError(
                f"an ask must require one of {', '.join(SUPPORTED_REQUIRES)}; "
                f"got {requires!r}"
            )
    elif requires is not None:
        raise KindError(f"only an ask may set requires; {kind!r} may not")
 
    message = HubMessage(
        message_id=new_ulid(now or datetime.now(timezone.utc), entropy=entropy),
        at=_now_iso(),
        kind=kind,
        sender=sender,
        recipient=recipient,
        subject=subject,
        body=body,
        in_reply_to=in_reply_to,
        requires=requires,
        answer=answer,
        provenance=provenance,
    )
    _dbg("OC_CHAT.MAIN", func, f"{kind} {sender} -> {recipient} ({message.message_id})", 3)
    return message
 
 
# ── The human rule ───────────────────────────────────────────────────────
 
 
def check_ruling_provenance(ask: HubMessage, ruling: HubMessage) -> None:
    """Refuse *ruling* unless its provenance supports answering *ask*.
 
    THE one place the human rule lives. An ask that requires a human is
    answerable only by a ``human_operator`` actor, on a channel a human
    can actually speak through (:data:`HUMAN_CHANNELS`), with a non-empty
    correlation naming the act that carried the answer. An
    ``orchestrator_agent`` or ``worker_agent`` actor can never satisfy
    it, which is the design note's "an agent-relayed yes is
    distinguishable from the human's" made mechanical rather than
    procedural.
 
    An ask that requires only the orchestrator is answerable by the
    orchestrator in its own session, because routing an ask to the human
    and waiting is autonomous while answering a human's ask is not.
 
    Raises:
        ProvenanceError: SECURITY_ERROR (L1) -- the message names the
            actor type, the channel and what was required.
    """
    func = "check_ruling_provenance"
    if ruling.in_reply_to != ask.message_id:
        raise ProvenanceError(
            f"ruling {ruling.message_id} has in_reply_to "
            f"{ruling.in_reply_to!r} and cannot answer ask {ask.message_id}"
        )
    if ask.requires != REQUIRES_HUMAN:
        return
 
    provenance = ruling.provenance
    if provenance.actor_type != HUMAN_ACTOR_TYPE:
        _dbg("OC_CHAT.MAIN", func,
             f"SECURITY_ERROR: {provenance.actor_type} ruled on a human ask", 1,
             override=True)
        raise ProvenanceError(
            f"ask {ask.message_id} requires a human, and a ruling from actor_type "
            f"{provenance.actor_type!r} cannot supply one: only "
            f"{HUMAN_ACTOR_TYPE!r} may. A security approval is in the operator's "
            "own voice and is never relayed by an agent."
        )
    if provenance.channel not in HUMAN_CHANNELS:
        _dbg("OC_CHAT.MAIN", func,
             f"SECURITY_ERROR: human ruling on channel {provenance.channel}", 1,
             override=True)
        raise ProvenanceError(
            f"ask {ask.message_id} requires a human, and channel "
            f"{provenance.channel!r} cannot carry one: a human answers through "
            f"{' or '.join(HUMAN_CHANNELS)}, each of which leaves a trace "
            "outside the answering process."
        )
    if not provenance.correlation:
        _dbg("OC_CHAT.MAIN", func, "SECURITY_ERROR: human ruling with no correlation",
             1, override=True)
        raise ProvenanceError(
            f"ask {ask.message_id} requires a human, and the ruling's correlation "
            "is empty: an answer that names no channel act is not attributable "
            "to anyone, which is the property the rule exists to keep."
        )
    _dbg("OC_CHAT.MAIN", func,
         f"human ruling on {ask.message_id} accepted via {provenance.channel}", 2)
 
 
# ── The two Protocols ────────────────────────────────────────────────────
 
 
@runtime_checkable
class HubLog(Protocol):
    """The log, defined once and implemented by both backends."""
 
    def record(self, message: HubMessage) -> HubEntry:
        """Append *message* and return the entry it became."""
 
    def entries(
        self,
        *,
        kind: str | None = None,
        recipient: str | None = None,
        subject: str | None = None,
        since_seq: int | None = None,
    ) -> tuple[HubEntry, ...]:
        """Recorded entries, oldest first, narrowed by whatever is given."""
 
    def open_asks(self, *, recipient: str | None = None) -> tuple[HubEntry, ...]:
        """Asks with no ruling in the log -- the questions still waiting."""
 
    def ruling_for(self, ask_id: str) -> HubEntry | None:
        """The ruling that answered *ask_id*, or ``None``."""
 
 
@runtime_checkable
class Carrier(Protocol):
    """Something that can move a message to a recipient."""
 
    #: What the carrier is called in a receipt and in a trace.
    name: str
 
    def present(self, address: str) -> bool:
        """Can this carrier reach *address* right now?"""
 
    def deliver(self, message: HubMessage) -> DeliveryReceipt:
        """Move *message*, or raise."""
 
 
# ── The hub ──────────────────────────────────────────────────────────────
 
 
class Hub:
    """Validate, record, then route. In that order, always."""
 
    def __init__(self, *, log: HubLog, carriers: tuple[Carrier, ...] = ()) -> None:
        func = "Hub.__init__"
        self.log = log
        self.carriers = tuple(carriers)
        _dbg("OC_CHAT.MAIN", func,
             f"over {len(self.carriers)} carrier(s): "
             f"{', '.join(c.name for c in self.carriers) or 'none'}", 2)
 
    def send(self, message: HubMessage) -> DeliveryReceipt:
        """Record *message*, then hand it to the first present carrier.
 
        The order is the contract. A ruling is checked against its ask
        BEFORE the record, because a ruling the human rule refuses must
        leave no trace of having been accepted; everything else is
        recorded first, because the log is the truth and a carrier that
        fails afterwards has lost a delivery, not a message.
 
        Raises:
            ProvenanceError: SECURITY_ERROR (L1) -- a ruling the human
                rule refuses. Nothing is recorded and nothing is sent.
            UnknownAskError: VALIDATION_ERROR (L3) -- a ruling naming
                something the log does not hold as an ask.
            CarrierError: EXTERNAL_ERROR (L2) -- no carrier is present
                for the recipient, or the chosen one raised. The entry
                stands either way.
        """
        func = "Hub.send"
        if message.kind == KIND_RULING:
            check_ruling_provenance(self._ask_for(message.in_reply_to or ""), message)
 
        entry = self.log.record(message)
        _dbg("OC_CHAT.MAIN", func,
             f"recorded seq {entry.seq}: {message.kind} -> {message.recipient}", 2)
 
        for carrier in self.carriers:
            if not carrier.present(message.recipient):
                continue
            try:
                receipt = carrier.deliver(message)
            except Exception as exc:  # noqa: BLE001 - every carrier failure is one class
                _dbg("OC_CHAT.MAIN", func,
                     f"EXTERNAL_ERROR: {carrier.name} failed: {exc}", 2)
                raise CarrierError(
                    f"carrier {carrier.name!r} could not deliver "
                    f"{message.message_id} to {message.recipient}: {exc}. The log "
                    f"entry at seq {entry.seq} stands; the message was sent."
                ) from exc
            _dbg("OC_CHAT.MAIN", func,
                 f"{carrier.name} delivered {message.message_id}", 2)
            return receipt
 
        names = ", ".join(c.name for c in self.carriers) or "none"
        _dbg("OC_CHAT.MAIN", func,
             f"EXTERNAL_ERROR: no carrier present for {message.recipient}", 2)
        raise CarrierError(
            f"no carrier is present for {message.recipient}; tried: {names}. "
            f"The log entry at seq {entry.seq} stands; the message was sent."
        )
 
    def attach(self, address: str, callback: "Callable[[HubMessage], None]") -> None:
        """Register *callback* as the in-process reader for *address*.
 
        Delegated to the first carrier that accepts attachments, which
        today is ``oc_chat.core.hub_bus.BusCarrier`` -- a CONSUMER's
        module, injected through ``carriers=``, which this package neither
        imports nor names at any depth beyond this sentence. The
        delegation is duck-typed ON PURPOSE: this module declares what a
        hub IS, and a hub that imported a specific carrier to offer this
        method would have the routing decision baked into the contract
        instead of injected beside it. Plan 272 P4 is the proof that the
        duck typing was load-bearing rather than decorative: the bus
        carrier stayed behind when the hub moved, and nothing here
        changed.
 
        Raises:
            AddressError: VALIDATION_ERROR (L3) -- *address* is not one.
            CarrierError: VALIDATION_ERROR (L3) -- no injected carrier
                accepts attachments, so there is nowhere to put it.
        """
        parse_address(address)
        self._attachable().attach(address, callback)
 
    def detach(self, address: str) -> None:
        """Remove *address*'s in-process reader. See :meth:`attach`."""
        parse_address(address)
        self._attachable().detach(address)
 
    def _attachable(self) -> "Any":
        """The first carrier exposing ``attach``, or a refusal naming the set."""
        for carrier in self.carriers:
            if hasattr(carrier, "attach") and hasattr(carrier, "detach"):
                return carrier
        names = ", ".join(c.name for c in self.carriers) or "none"
        raise CarrierError(
            f"no injected carrier accepts an attach; tried: {names}. An "
            "in-process reader needs a carrier that can hold one."
        )
 
    def rule(
        self,
        ask_id: str,
        *,
        answer: str,
        body: str,
        provenance: Provenance,
        now: datetime | None = None,
        entropy: bytes | None = None,
    ) -> HubEntry:
        """Answer the open ask *ask_id* and send the ruling back to its asker.
 
        The ruling is BUILT from the ask rather than supplied beside it,
        so its recipient, subject and ``in_reply_to`` cannot disagree
        with the question it answers.
 
        Raises:
            UnknownAskError: VALIDATION_ERROR (L3) -- no such message, it
                is not an ask, or it already has a ruling.
            ProvenanceError: SECURITY_ERROR (L1) -- the human rule.
        """
        func = "Hub.rule"
        ask = self._ask_for(ask_id)
        existing = self.log.ruling_for(ask_id)
        if existing is not None:
            raise UnknownAskError(
                f"ask {ask_id} was already ruled by {existing.message.message_id} "
                f"({existing.message.answer}); a second ruling would make the log "
                "ambiguous about which answer was acted on"
            )
        ruling = new_message(
            kind=KIND_RULING,
            sender=provenance.actor,
            recipient=ask.sender,
            subject=ask.subject,
            body=body,
            in_reply_to=ask.message_id,
            answer=answer,
            provenance=provenance,
            now=now,
            entropy=entropy,
        )
        check_ruling_provenance(ask, ruling)
        entry = self.log.record(ruling)
        _dbg("OC_CHAT.MAIN", func, f"{answer} on {ask_id} (seq {entry.seq})", 2)
        self._deliver_recorded(ruling)
        return entry
 
    def _deliver_recorded(self, message: HubMessage) -> DeliveryReceipt | None:
        """Route an already-recorded message; a failure is still a refusal."""
        func = "Hub._deliver_recorded"
        for carrier in self.carriers:
            if not carrier.present(message.recipient):
                continue
            try:
                return carrier.deliver(message)
            except Exception as exc:  # noqa: BLE001 - one class for every carrier
                _dbg("OC_CHAT.MAIN", func,
                     f"EXTERNAL_ERROR: {carrier.name} failed: {exc}", 2)
                raise CarrierError(
                    f"carrier {carrier.name!r} could not deliver "
                    f"{message.message_id} to {message.recipient}: {exc}. The log "
                    "entry stands; the message was sent."
                ) from exc
        raise CarrierError(
            f"no carrier is present for {message.recipient}. The log entry "
            "stands; the message was sent."
        )
 
    def _ask_for(self, ask_id: str) -> HubMessage:
        """The ask *ask_id* names, or a refusal saying what it found instead."""
        for entry in self.log.entries():
            if entry.message.message_id != ask_id:
                continue
            if entry.message.kind != KIND_ASK:
                raise UnknownAskError(
                    f"{ask_id} is a {entry.message.kind!r}, not an ask; only an "
                    "ask can be ruled on"
                )
            return entry.message
        raise UnknownAskError(f"the hub log holds no message {ask_id!r}")
 
 
__all__ = [
    "ACTOR_TYPES",
    "ADDRESS_OPERATOR_LOCAL",
    "ADDRESS_OPERATOR_REMOTE",
    "ADDRESS_ORCHESTRATOR",
    "CHANNELS",
    "CROCKFORD_ALPHABET",
    "FIXED_ADDRESSES",
    "HUMAN_ACTOR_TYPE",
    "HUMAN_CHANNELS",
    "KIND_ALERT",
    "KIND_ASK",
    "KIND_DISPATCH",
    "KIND_INSTRUCTION",
    "KIND_REPORT",
    "KIND_RULING",
    "REQUIRES_HUMAN",
    "REQUIRES_ORCHESTRATOR",
    "RULING_ANSWERS",
    "SUPPORTED_KINDS",
    "SUPPORTED_REQUIRES",
    "AddressError",
    "Carrier",
    "CarrierError",
    "DeliveryReceipt",
    "Hub",
    "HubEntry",
    "HubError",
    "HubLog",
    "HubMessage",
    "KindError",
    "Provenance",
    "ProvenanceError",
    "UnknownAskError",
    "check_kind",
    "check_ruling_provenance",
    "new_message",
    "new_ulid",
    "parse_address",
    "validate_agent_id",
]