hub_log.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/hub_log.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_259_hub_log.py"
"""
The registry backend for the hub log: one message, one operation-log row.
 
Purpose
-------
Plan 259 P4. The hub has one log contract and two backends, and this is
the one oc_chat plugs in. It lives on the REGISTRY side on purpose,
exactly as ``registry/queue_backend.py`` does: the operation log already
carries actor, actor type, credential id and an append-only guarantee
enforced by a database trigger, so the hub inherits those instead of
growing a second copy of them.
 
A hub message is ONE row of ``operation_log`` with
``operation_type='hub.message'``, ``command`` = the hub kind and
``related_plan_id`` = the subject when the subject is a plan id. There is
no projection: a message changes no projected state, it IS the state, and
``SQLiteBackend._apply_domain_projection`` is an ``if``/``elif`` chain
with no ``else``, so an unmatched type projects nothing. **No projector
was added, because none is required** -- a witness asserts the
observable half of that.
 
**The row's signature and the message's claim are different things.**
``actor_type`` and ``hmac_key_id`` are written by the storage layer from
the SLOT, and a caller cannot choose them; the message's ``provenance``
is a payload field a caller writes freely. The hub's human rule is
enforced on the claim, in ``core/hub.check_ruling_provenance``, and this
module does not second-guess it -- but the row beside it records who the
process was, which is the cross-check the design note asked for.
 
The measured preflight fact
---------------------------
A ``human_operator`` row written from a CLI process that supplies no
``auth_proof`` **is accepted, unsigned**. ``Registry._preflight_auth``
(``registry/api.py:115``) runs the role guard and returns ``None`` the
moment ``auth_proof`` is ``None``; ``hub.message`` matches no entry in
``ROLE_ALLOWSETS`` and no prefix in ``PREFIX_ALLOWSETS``, so the guard
permits all three slot kinds. The row still lands carrying an
``hmac_key_id``, because the storage layer falls back to the slot's
ACTIVE credential when the caller supplies none -- so that key id names a
credential which did not sign this row.
 
So the registry mode's signature is a **claim**, like the file mode's,
and the hub's own human rule is what does the work in both. That is a
finding, not a design; it is filed rather than fixed here, because
demanding a credential would be a security-layer change and those are
the operator's.
 
Responsibilities
----------------
- :class:`RegistryHubLog` -- the four :class:`~oc_orchestrator.hub.HubLog`
  operations over the operation log.
- :data:`HUB_OPERATION_TYPE` -- the one operation type, pinned as a
  constant so a witness can assert it.
- :class:`UnregisteredSenderError` -- the adapter's named refusal.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (a message recorded)
    L3 - semantic details (a refusal, a filter)
    L4 - deep tracing (module load)
 
Contracts
---------
- This module MUTATES only through ``registry.domain.hub_message``: no
  SQL write, no direct table update.
- The READ is one ``SELECT`` in this file rather than a fifth function in
  ``registry/readers.py``. That is the honest place: ``readers.py``
  returns ``registry/dataclasses.py`` types, and a hub reader must return
  :class:`~oc_orchestrator.hub.HubEntry` -- a DISPATCH-layer type. Putting
  that import into the general reader library would push a dispatch type
  into every consumer of it, including the registry CLI. The adapter is
  already the one place the two vocabularies meet, so the read lives
  here, exactly as ``queue_backend`` keeps its own ``plans`` projection
  query rather than widening ``readers``.
- The import direction is registry -> dispatch, which is the permitted
  one and the one ``registry/mirror_observer.py`` and
  ``registry/queue_backend.py`` already take. Nothing in the dispatch
  layer imports this module; the CLI reaches it through one lazy import
  behind ``--backend registry``.
- ``seq`` is the row's ``state_version``: monotonic within the store, as
  the contract requires. It is NOT comparable with the file log's
  line count, and the equivalence witness compares the FOLD rather than
  the numbering for that reason.
 
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 registry-side backend of the seam, not a caller that always
  spelled the old path)
- oc_chat.registry.api (the Registry facade)
- oc_chat.registry.domain (the typed hub operation)
"""
 
from __future__ import annotations
 
import json
import re
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.hub import (
    KIND_ASK,
    KIND_RULING,
    HubEntry,
    HubError,
    HubMessage,
)
from oc_chat.registry.api import Registry
from oc_chat.registry.domain import hub_message
 
_dbg("OC_CHAT.MAIN", "module", "hub_log loaded", 4)
 
 
#: The one operation type this adapter writes and reads.
HUB_OPERATION_TYPE = "hub.message"
 
#: What ``related_plan_id`` will accept: a plan id is ``NNN-stem``. A
#: subject outside this shape stays in the payload and leaves the column
#: NULL -- the column is plan-shaped, and the design note §5 names that
#: as the seam a second work-item kind will find.
# OCT-LINT: disable=no_hardcoded_secrets reason="Plan-id shape regex (NNN-stem) for the related_plan_id column; no credential content"
_PLAN_ID_RE = re.compile(r"^[0-9]{3,}-[A-Za-z0-9][A-Za-z0-9_\-.]*$")
 
#: The registry's actor type for each slot kind, mirroring
#: ``backends/sqlite._ACTOR_TYPES_BY_SLOT_KIND``. The storage layer
#: asserts the same mapping before commit; deriving it here is what lets
#: the adapter supply the RIGHT value rather than make the caller guess
#: and be refused with ``ActorTypeMismatchError``.
_ACTOR_TYPE_BY_SLOT_KIND = {
    "operator": "human_operator",
    "orchestrator": "orchestrator_agent",
    "worker": "worker_agent",
}
 
 
class UnregisteredSenderError(HubError):
    """The actor slot this log writes as is not registered in this registry."""
 
 
class RegistryHubLog:
    """The hub's log over the registry's operation log."""
 
    def __init__(self, registry: Registry, *, actor_slot: str) -> None:
        func = "RegistryHubLog.__init__"
        self.registry = registry
        self.actor_slot = actor_slot
        _dbg("OC_CHAT.MAIN", func, f"as {actor_slot}", 2)
 
    # -- writes -----------------------------------------------------------
 
    def record(self, message: HubMessage) -> HubEntry:
        """Write *message* as one operation-log row and return its entry.
 
        Raises:
            UnregisteredSenderError: VALIDATION_ERROR (L3) -- the actor
                slot is not in ``agents``. The registry would refuse it
                anyway; refusing here names the slot instead of raising
                an ``UnknownActorError`` from three frames down.
        """
        func = "RegistryHubLog.record"
        actor_type = self._actor_type()
        row = hub_message(
            self.registry,
            message=message.as_dict(),
            kind=message.kind,
            actor_slot=self.actor_slot,
            actor_type=actor_type,  # type: ignore[arg-type]
            related_plan_id=_plan_subject(message.subject),
        )
        _dbg("OC_CHAT.MAIN", func,
             f"{message.kind} as {self.actor_slot}/{actor_type} "
             f"(state_version {row.state_version})", 2)
        return HubEntry(seq=int(row.state_version), message=message)
 
    # -- reads ------------------------------------------------------------
 
    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.
 
        ``kind`` narrows in SQL, because it is the ``command`` column;
        the other two narrow in Python, because they live inside the
        payload and a JSON predicate in SQL would be a parser in
        disguise -- the same call the queue adapter made about its own
        free-text column.
        """
        func = "RegistryHubLog.entries"
        sql = (
            # OCT-LINT: disable=no_hardcoded_secrets reason="Static SQL projection over the operation log; no credential content"
            "SELECT state_version, args_json FROM operation_log "
            "WHERE operation_type = ?"
        )
        parameters: list[object] = [HUB_OPERATION_TYPE]
        if kind is not None:
            sql += " AND command = ?"
            parameters.append(kind)
        if since_seq is not None:
            sql += " AND state_version > ?"
            parameters.append(int(since_seq))
        sql += " ORDER BY state_version"
 
        with self._connect() as conn:
            rows = conn.execute(sql, tuple(parameters)).fetchall()
 
        entries = [
            HubEntry(seq=int(row["state_version"]), message=_message_from_row(row["args_json"]))
            for row in rows
        ]
        chosen = tuple(
            entry
            for entry in entries
            if (recipient is None or entry.message.recipient == recipient)
            and (subject is None or entry.message.subject == subject)
        )
        _dbg("OC_CHAT.MAIN", func, f"{len(chosen)} of {len(entries)} row(s)", 3)
        return chosen
 
    def open_asks(self, *, recipient: str | None = None) -> tuple[HubEntry, ...]:
        """Asks with no ruling in the log -- the questions still waiting."""
        ruled = {
            entry.message.in_reply_to for entry in self.entries(kind=KIND_RULING)
        }
        return tuple(
            entry
            for entry in self.entries(kind=KIND_ASK, recipient=recipient)
            if entry.message.message_id not in ruled
        )
 
    def ruling_for(self, ask_id: str) -> HubEntry | None:
        """The ruling that answered *ask_id*, or ``None``."""
        for entry in self.entries(kind=KIND_RULING):
            if entry.message.in_reply_to == ask_id:
                return entry
        return None
 
    # -- internals --------------------------------------------------------
 
    def _actor_type(self) -> str:
        """The actor type the storage layer will demand for this slot."""
        with self._connect() as conn:
            row = conn.execute(
                "SELECT slot_kind FROM agents WHERE slot = ?",
                (self.actor_slot,),
            ).fetchone()
        if row is None:
            raise UnregisteredSenderError(
                f"actor slot {self.actor_slot!r} is not registered in this "
                "registry; the hub cannot write a row it cannot attribute"
            )
        slot_kind = str(row["slot_kind"])
        try:
            return _ACTOR_TYPE_BY_SLOT_KIND[slot_kind]
        except KeyError:
            raise UnregisteredSenderError(
                f"actor slot {self.actor_slot!r} has slot_kind {slot_kind!r}, "
                f"which maps to no actor type; the registry knows "
                f"{', '.join(sorted(_ACTOR_TYPE_BY_SLOT_KIND))}"
            ) from None
 
    def _connect(self) -> object:
        """A read connection through the storage backend."""
        connect = getattr(self.registry.backend, "connect", None)
        if connect is None:
            raise HubError("the registry backend does not expose read connections")
        return connect()
 
 
def _plan_subject(subject: str | None) -> str | None:
    """*subject* if it is plan-shaped, else ``None``. See the constant above."""
    if subject and _PLAN_ID_RE.fullmatch(subject):
        return subject
    return None
 
 
def _message_from_row(args_json: object) -> HubMessage:
    """Rebuild the message one ``hub.message`` payload carried."""
    payload = json.loads(str(args_json or "{}"))
    stored = payload.get("message")
    if not isinstance(stored, dict):
        raise HubError(
            f"a {HUB_OPERATION_TYPE} row carries no message object; the payload "
            f"has {sorted(payload)}"
        )
    return HubMessage.from_mapping(stored)
 
 
__all__ = [
    "HUB_OPERATION_TYPE",
    "RegistryHubLog",
    "UnregisteredSenderError",
]