hub_lane.py

Go to the documentation of this file.

"""The hub lane: addressed messages at rest, their relay, and the reply path.
 
Plan 259 P5, from the factory-decomposition design note §4. The hub has
three carriers and this is the third: the connector, for whoever is remote.
A message addressed to ``operator@remote`` is written into
``comms/lanes/hub/`` by ``oc_chat`` -- which holds no credential and knows
nothing about this package -- and ``oc_connect relay hub`` is what puts it on
the operator's phone.
 
**This module is the schema authority for the lane.** The writer lives in
``oc_chat/oc_chat/core/hub_file.py`` and deliberately does NOT import this
package, exactly as the worker-report and session-alert writers do not. The
two sides agree by a written contract rather than by a shared import, and the
contract is :class:`HubLaneEvent`::
 
    {
      "schema_version":    "1.0",
      "event_id":          "<26-character ULID; IS the message_id>",
      "lane":              "hub",
      "direction":         "outbound",
      "risk_class":        "low",
      "created_at_utc":    "2026-09-13T12:00:00Z",
      "summary":           "<the first line of the body>",
      "relayed_at_utc":    null,
      "relay_chunk_count": null,
 
      "message_id":        "<same as event_id>",
      "at":                "2026-09-13T12:00:00+02:00",
      "kind":              "dispatch"|"report"|"alert"|"ask"|"ruling"|"instruction",
      "sender":            "orchestrator"|"operator@local"|"operator@remote"|"agent-NNN",
      "recipient":         <the same four forms>,
      "subject":           "259-hub" | null,
      "body":              "<the text>",
      "in_reply_to":       "<an ask's id>" | null,
      "requires":          "human"|"orchestrator" | null,
      "answer":            "YES"|"NO"|"ACK"|"HOLD" | null,
      "provenance":        {actor, actor_type, channel, correlation}
    }
 
File naming follows the §3 bus convention, ``<utc-stamp>_hub_<ulid>.json``.
 
Three differences from the two lanes it mirrors are deliberate.
 
* **The message id is the idempotency key, unchunked.** A worker report keys
  each chunk (``<event_id>-001``) because it carries a whole closing summary;
  a hub message is ONE message, and keying it by its own id is what lets the
  operator's reply correlate back to the MESSAGE rather than merely to a send.
  A body past the transport limit therefore fails loudly instead of being
  split -- the same call :mod:`~oc_connect.session_alerts` makes, and for the
  same reason.
* **Only messages addressed to** ``operator@remote`` **are relayed.** The lane
  carries every hub message, because it is also the durable hand-off between
  local parties; the relay is the half addressed away from here.
* **A human-required ask carries the four reply buttons.** Nothing else does:
  a report, an alert, an instruction and an orchestrator-required ask are not
  questions the operator is being asked to answer. Note that plan 236's
  ``CHOICES_RULE`` still governs WHAT may be asked with buttons -- a security
  ruling is never sent with them -- and that rule is the orchestrator's to
  apply when it composes the ask, not this relay's to enforce after the fact.
 
The reply path is the EXISTING inbox. :func:`resolve_hub_reply` walks the
correlation the poller already recorded -- audit id, outbound record,
idempotency key -- back to the hub message id, and refuses by name when the
walk does not land on this lane. Building the ruling from that is
``oc_chat``'s job (``hub_manage rule --from-inbox``), because a ruling is a
hub message and this package does not know what one is.
 
Dependencies
------------
- oc_connect.audit (AuditStore, content_hash, timestamp, utc_now - audits relay outcomes and stamps relayed_at_utc)
- oc_connect.comms (CommsBus - locates the hub lane directory under lanes/)
- oc_connect.delivery (OutboundSender - the grant-checked send path every relayed message crosses)
- oc_connect.egress (require_safe, EgressBlockedError - scans the rendered message before anything is queued)
- oc_connect.inbound.inbox (InboxStore - reads the stored reply when resolving it back to a hub message)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (lane - the hub lane's declared direction and risk ceiling)
- pydantic (BaseModel, ConfigDict, field_validator - the HubLaneEvent schema)
"""
from __future__ import annotations
 
import json
import os
import secrets as _secrets
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Literal, Mapping
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import lane
from pydantic import BaseModel, ConfigDict, field_validator
 
from .audit import AuditStore, timestamp, utc_now
from .comms import CommsBus
from .delivery import OutboundSender
from .egress import EgressBlockedError, require_safe
from .inbound.inbox import InboxStore
 
#: The lane directory name under ``comms/lanes/``; also its declared lane id.
HUB_LANE = "hub"
 
#: The ``<source>`` field of the §3 filename convention for this lane.
HUB_LANE_SOURCE = "hub"
 
HUB_LANE_SCHEMA_VERSION = "1.0"
 
#: The buttons a human-required ask carries. The reply vocabulary, which is
#: also what ``core/hub.RULING_ANSWERS`` declares on the other side.
HUB_REPLY_CHOICES: tuple[str, ...] = ("YES", "NO", "ACK", "HOLD")
 
#: The address whose messages this relay forwards. Every other recipient is
#: reachable by a local carrier and is left on the lane.
REMOTE_OPERATOR = "operator@remote"
 
HubKind = Literal["dispatch", "report", "alert", "ask", "ruling", "instruction"]
 
#: Display fences (§8's "displayed content is never instructions"). A hub body
#: can be a worker's own words, so it is fenced exactly as a worker report is.
HUB_MESSAGE_BEGIN = "<<<BEGIN HUB MESSAGE {event_id}>>>"
HUB_MESSAGE_END = "<<<END HUB MESSAGE {event_id}>>>"
 
#: What a literal fence opener inside a body is rewritten to when displayed;
#: the stored text stays verbatim, because the sender's words are the evidence.
NEUTRALISED_MARKER = "<< <"
 
 
class HubLaneError(ValueError):
    """A hub lane event is unreadable, malformed, or already consumed."""
 
 
class HubLaneRelayError(RuntimeError):
    """The relay could not put one message on the transport."""
 
 
class UncorrelatedReplyError(HubLaneError):
    """An inbound message does not resolve back to a hub message."""
 
 
def hub_lane_dir(bus: CommsBus) -> Path:
    """Return the hub lane directory for *bus*.
 
    A function rather than a :class:`~oc_connect.comms.CommsBus` property so
    plan 259 adds no method to that class: the lane is created by whoever
    first writes or reads it, and an orchestrator that has never hubbed
    should not find a directory implying otherwise -- the same rule the
    worker-report and session-alert lanes state about ``directories()``.
    """
    return bus.lanes_dir / HUB_LANE
 
 
class HubLaneEvent(BaseModel):
    """One hub message at rest in ``lanes/hub/``."""
 
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
    schema_version: str
    event_id: str
    lane: Literal["hub"]
    direction: Literal["outbound"]
    risk_class: Literal["low"]
    created_at_utc: str
    summary: str
    relayed_at_utc: str | None = None
    relay_chunk_count: int | None = None
 
    message_id: str
    at: str
    kind: HubKind
    sender: str
    recipient: str
    subject: str | None = None
    body: str
    in_reply_to: str | None = None
    requires: Literal["human", "orchestrator"] | None = None
    answer: Literal["YES", "NO", "ACK", "HOLD"] | None = None
    provenance: dict[str, Any]
 
    @field_validator(
        "schema_version", "event_id", "created_at_utc", "message_id", "at",
        "sender", "recipient",
    )
    @classmethod
    def _required_text_is_nonblank(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("must not be blank")
        return value
 
    @field_validator("message_id")
    @classmethod
    def _message_id_is_the_event_id(cls, value: str, info: Any) -> str:
        """The one identity, asserted at the boundary.
 
        Keeping these two equal is what makes the relay's idempotency key the
        message id; a file where they differ would correlate a reply to a send
        that names a different message.
        """
        event_id = info.data.get("event_id")
        if event_id is not None and value != event_id:
            raise ValueError(
                f"message_id {value!r} must equal event_id {event_id!r}"
            )
        return value
 
    @property
    def is_relayed(self) -> bool:
        """True once this message reached the transport."""
        return self.relayed_at_utc is not None
 
    @property
    def is_human_ask(self) -> bool:
        """True when this is a question only the operator may answer."""
        return self.kind == "ask" and self.requires == "human"
 
    @property
    def is_for_remote_operator(self) -> bool:
        """True when this message is addressed away from the control centre."""
        return self.recipient == REMOTE_OPERATOR
 
 
def render_event(event: HubLaneEvent) -> str:
    """Return the text an operator reads for one hub message, fenced.
 
    The header names the ROUTE, because on a hub the question "who is asking
    me this, about what" is most of the message. The body is fenced with the
    event's own id and any literal ``<<<`` inside it is neutralised, so a body
    cannot forge its own closing fence. The rendered copy is the only thing
    that changes; the stored text does not.
    """
    state = "relayed" if event.is_relayed else "unrelayed"
    header = (
        f"[{state}] {event.event_id}  {event.created_at_utc}  "
        f"lane={event.lane} kind={event.kind} "
        f"{event.sender} -> {event.recipient}"
    )
    if event.subject:
        header += f" subject={event.subject}"
    if event.requires:
        header += f" requires={event.requires}"
    if event.in_reply_to:
        header += f" answers={event.in_reply_to}"
    body = event.body.replace("<<<", NEUTRALISED_MARKER)
    begin = HUB_MESSAGE_BEGIN.format(event_id=event.event_id)
    end = HUB_MESSAGE_END.format(event_id=event.event_id)
    return f"{header}\n{begin}\n{body}\n{end}"
 
 
def _atomic_write(path: Path, payload: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".tmp-{path.name}-{_secrets.token_hex(8)}")
    try:
        with temporary.open("w", encoding="utf-8", newline="\n") as handle:
            handle.write(payload)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, path)
    finally:
        if temporary.exists():
            try:
                temporary.unlink()
            except OSError:
                pass
 
 
class HubLaneStore:
    """Read-and-mark access to one ``lanes/hub/`` directory.
 
    Lane separation is physical: this store is handed the lane directory and
    has no path arithmetic that could reach ``inbound/`` or ``outbound/``.
    """
 
    def __init__(
        self,
        directory: str | Path,
        *,
        now: Callable[[], datetime] = utc_now,
    ) -> None:
        self.directory = Path(directory)
        self._now = now
 
    def paths(self) -> tuple[Path, ...]:
        """Return event files in arrival (filename) order, ignoring dotfiles."""
        if not self.directory.is_dir():
            return ()
        return tuple(
            sorted(
                path
                for path in self.directory.glob("*.json")
                if path.is_file() and not path.name.startswith(".")
            )
        )
 
    def _read(self, path: Path) -> HubLaneEvent:
        try:
            return HubLaneEvent.model_validate_json(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise HubLaneError(
                f"could not parse hub lane event at {path} ({type(exc).__name__})"
            ) from None
 
    def list_events(self, *, relayed: bool | None = None) -> tuple[HubLaneEvent, ...]:
        """Return stored events in arrival order, optionally by relay state."""
        events = tuple(self._read(path) for path in self.paths())
        if relayed is None:
            return events
        return tuple(event for event in events if event.is_relayed is relayed)
 
    def path_for(self, event_id: str) -> Path:
        """Return the file holding one event or fail with its public id."""
        suffix = f"_{event_id}.json"
        for path in self.paths():
            if path.name.endswith(suffix):
                return path
        raise HubLaneError(f"hub lane event {event_id!r} is not in the lane")
 
    def get(self, event_id: str) -> HubLaneEvent:
        """Return one stored event or fail with its public id."""
        return self._read(self.path_for(event_id))
 
    def mark_relayed(self, event_id: str, *, chunk_count: int = 1) -> HubLaneEvent:
        """Record that one message reached the transport."""
        if chunk_count < 1:
            raise ValueError("a relayed message must have sent at least one message")
        path = self.path_for(event_id)
        event = self._read(path)
        if event.is_relayed:
            raise HubLaneError(
                f"hub lane event {event_id!r} was already relayed at "
                f"{event.relayed_at_utc}"
            )
        updated = event.model_copy(
            update={
                "relayed_at_utc": timestamp(self._now()),
                "relay_chunk_count": chunk_count,
            }
        )
        _atomic_write(
            path, json.dumps(updated.model_dump(mode="json"), indent=2, sort_keys=True)
        )
        _dbg("OC_CONNECT.HUB", "mark_relayed",
             f"event={event_id} kind={event.kind}", 2)
        return updated
 
 
@dataclass(frozen=True)
class HubLaneOutcome:
    """What the relay did with one message, in terms a CLI can print."""
 
    message_id: str
    kind: str
    sender: str
    subject: str | None
    choices: tuple[str, ...]
    audit_id: str
 
 
class HubLaneRelay:
    """Forward hub messages addressed to the remote operator, verbatim.
 
    Every message still crosses the plan-219 pipeline -- grant check, egress
    scan, durable outbox, audit -- because the relay delegates to
    :class:`~oc_connect.delivery.OutboundSender` rather than reimplementing it.
    """
 
    def __init__(
        self,
        *,
        store: HubLaneStore,
        sender: OutboundSender,
        audit: AuditStore,
        grantee: str,
        destination: str,
        grant_id: str,
        risk_class: str | None = None,
        now: Callable[[], datetime] = utc_now,
    ) -> None:
        self.store = store
        self.sender = sender
        self.audit = audit
        self.grantee = grantee
        self.destination = destination
        self.grant_id = grant_id
        self.risk_class = risk_class
        self._now = now
 
    def _audit(
        self,
        event: HubLaneEvent,
        outcome: str,
        *,
        attempts: int,
        error_code: str | None = None,
    ) -> None:
        from .audit import content_hash
 
        descriptor = lane(HUB_LANE)
        self.audit.append(
            connector="telegram",
            grant_id=self.grant_id,
            destination=self.destination,
            outcome=outcome,  # type: ignore[arg-type]
            digest=content_hash(event.body),
            idempotency_key=event.event_id,
            attempts=attempts,
            error_code=error_code,
            lane=HUB_LANE,
            direction=descriptor.direction,
            risk_class=descriptor.risk_ceiling,
        )
 
    def pending(self) -> tuple[HubLaneEvent, ...]:
        """Unrelayed messages addressed to the remote operator, oldest first."""
        return tuple(
            event
            for event in self.store.list_events(relayed=False)
            if event.is_for_remote_operator
        )
 
    def relay_one(self, event: HubLaneEvent) -> HubLaneOutcome:
        """Relay one message, or leave it unrelayed and raise.
 
        The egress scan runs on the WHOLE rendered text before anything is
        queued. One message per hub message, keyed by its id: see the module
        docstring for why this lane does not chunk.
        """
        func = "relay_one"
        text = render_event(event)
        try:
            require_safe(text)
        except EgressBlockedError:
            self._audit(event, "blocked", attempts=0, error_code="egress_finding")
            _dbg("OC_CONNECT.HUB", func,
                 # OCT-LINT: disable=no_hardcoded_secrets reason="Diagnostic prose naming the S6 class SECURITY_ERROR on the egress-refusal path; scores 4.59 bits/char and carries no credential, the one interpolated value being a ULID. Operator-ruled 2026-09-13 (plan 259 escalation, line-scoped)"
                 f"SECURITY_ERROR: blocked hub message event={event.event_id}",
                 1, override=True)
            raise
        choices = HUB_REPLY_CHOICES if event.is_human_ask else None
        receipt = self.sender.send(
            grantee=self.grantee,
            destination=self.destination,
            text=text,
            grant_id=self.grant_id,
            risk_class=self.risk_class,
            # The message id UNCHANGED: this is what a reply correlates to.
            idempotency_key=event.event_id,
            choices=choices,
        )
        if receipt.outcome != "sent":
            self._audit(event, "failed", attempts=1, error_code=receipt.outcome)
            raise HubLaneRelayError(
                f"hub message {event.event_id!r} was not delivered (outcome "
                f"{receipt.outcome!r}); the message stays unrelayed and "
                "re-running resumes from the durable outbox"
            )
        self.store.mark_relayed(event.event_id)
        self._audit(event, "sent", attempts=1)
        _dbg("OC_CONNECT.HUB", func,
             f"relayed event={event.event_id} kind={event.kind} "
             f"choices={len(choices) if choices else 0}", 2)
        return HubLaneOutcome(
            message_id=event.message_id,
            kind=event.kind,
            sender=event.sender,
            subject=event.subject,
            choices=tuple(choices or ()),
            audit_id=receipt.audit_id,
        )
 
    def relay_pending(self) -> tuple[HubLaneOutcome, ...]:
        """Relay every pending message, oldest first, stopping on the first refusal."""
        return tuple(self.relay_one(event) for event in self.pending())
 
 
@dataclass(frozen=True)
class HubReply:
    """One inbound reply, resolved back to the hub message it answers."""
 
    #: The hub message this reply answers -- for an ask, the ask's id.
    hub_message_id: str
    #: The reply-vocabulary token the operator sent or tapped.
    answer: str
    #: The three fields a ``telegram``-channel provenance must carry.
    correlation: Mapping[str, str]
 
 
def resolve_hub_reply(
    inbound_message_id: str,
    *,
    inbox: InboxStore,
    audit: AuditStore,
) -> HubReply:
    """Resolve one stored inbound message back to the hub message it answers.
 
    The walk is the poller's own, read backwards: the stored message carries
    the audit id of the send it replied to; that record carries the
    idempotency key; and for this lane the key IS the hub message id. Every
    missing link, and every link that lands on a different lane, is refused by
    name rather than guessed -- turning an uncorrelated message into a ruling
    is precisely the failure the hub's provenance rule exists to prevent.
 
    Returns the three correlation fields the hub's ``telegram`` channel
    requires. Building the ruling from them is ``oc_chat``'s job: a ruling is
    a hub message, and this package does not know what one is.
 
    Raises:
        UncorrelatedReplyError: VALIDATION_ERROR (L3) -- the message is not
            stored, carries no correlation, names an audit record that does
            not exist, or answers a send on another lane.
    """
    func = "resolve_hub_reply"
    try:
        message = inbox.get(inbound_message_id)
    except (KeyError, ValueError) as exc:
        raise UncorrelatedReplyError(
            f"inbound message {inbound_message_id!r} is not in the inbox"
        ) from exc
 
    audit_id = message.correlation.audit_id
    if not audit_id:
        raise UncorrelatedReplyError(
            f"inbound message {inbound_message_id!r} carries no correlation; a "
            "freely typed message answers no question and is not made into one"
        )
 
    named = next(
        (record for record in audit.read_all() if record.audit_id == audit_id),
        None,
    )
    if named is None:
        raise UncorrelatedReplyError(
            f"inbound message {inbound_message_id!r} names audit record "
            f"{audit_id!r}, which the audit trail does not hold"
        )
    if named.lane != HUB_LANE:
        raise UncorrelatedReplyError(
            f"inbound message {inbound_message_id!r} answers a send on lane "
            f"{named.lane!r}, not {HUB_LANE!r}; a reply to a worker report is "
            "not a ruling on a hub ask"
        )
 
    answer = message.vocabulary_token or message.text.strip()
    reply = HubReply(
        hub_message_id=named.idempotency_key,
        answer=answer,
        correlation={
            "message_id": message.message_id,
            "audit_id": audit_id,
            "content_hash": message.content_hash,
        },
    )
    _dbg("OC_CONNECT.HUB", func,
         f"message={inbound_message_id} answers hub={reply.hub_message_id} "
         f"with {answer}", 2)
    return reply
 
 
__all__ = [
    "HUB_LANE",
    "HUB_LANE_SCHEMA_VERSION",
    "HUB_LANE_SOURCE",
    "HUB_MESSAGE_BEGIN",
    "HUB_MESSAGE_END",
    "HUB_REPLY_CHOICES",
    "NEUTRALISED_MARKER",
    "REMOTE_OPERATOR",
    "HubKind",
    "HubLaneError",
    "HubLaneEvent",
    "HubLaneOutcome",
    "HubLaneRelay",
    "HubLaneRelayError",
    "HubLaneStore",
    "HubReply",
    "UncorrelatedReplyError",
    "hub_lane_dir",
    "render_event",
    "resolve_hub_reply",
]