hub_file.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/hub_file.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_orchestrator/tests/test_259_hub_file.py"
"""
The hub's default backend: an append-only log, a lane carrier, and ingest.
 
Purpose
-------
Plan 259 P3, from the factory-decomposition design note §4. This is the
DEFAULT hub backend and the mode the package's own tests run in, which is
the condition that stops it rotting into dead code kept warm for
strangers. It is three things:
 
``hub.ops.jsonl``
    The append-only log of recorded messages, one JSON object per line in
    the :class:`~oc_orchestrator.hub.HubEntry` shape. Opened for APPEND
    only, flushed and fsynced per write, never rewritten. **This file is
    the truth**, and every carrier moves a copy of something already in
    it.
 
``hub.cursors.json``
    One last-read sequence number per address, beside the log. It is what
    makes ``inbox`` idempotent and what lets a worker that lost its
    session resume against an ask it has not yet answered -- design-note
    invariant 2, an ask outliving its asker.
 
``<comms>/lanes/hub/``
    The durable carrier: one JSON file per message, in the SAME envelope
    the worker-report and session-alert lanes already use, so
    ``oc_connect relay hub`` is the relay those two lanes already have
    rather than a second one.
 
And one reader, :func:`ingest_lanes`, which folds the two EXISTING lanes
into the hub by READING them. A worker report at a gate becomes a
``report``; a blocked worker becomes an OPEN ASK, which is how a worker
waiting on a decision becomes visible as waiting; a session alert becomes
an ``alert`` to the operator. Neither existing writer is asked to change
and no lane file is ever modified.
 
Responsibilities
----------------
- :class:`HubPaths` -- the injected filesystem inputs, in the
  :class:`~oc_orchestrator.work_queue_file.QueuePaths` shape.
- :func:`read_entries` / :func:`append_entry` -- the log.
- :class:`FileHubLog` -- the :class:`~oc_orchestrator.hub.HubLog` contract
  plus the per-recipient cursor.
- :func:`hub_lane_event` / :func:`write_hub_lane_event` /
  :class:`FileLaneCarrier` -- the durable carrier.
- :func:`ingest_lanes` -- the two existing lanes, folded in.
- :func:`find_comms_root` -- the workspace-anchored bus root.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (a record, a lane write, an ingest pass)
    L3 - semantic details (a refusal, a mapping)
    L4 - deep tracing (module load, per-line parse)
 
Contracts
---------
- NO LOCKING, stated exactly as ``work_queue_file`` states it. Two
  writers can lose an update: each reads the log, decides, and appends,
  so a concurrent pair can both append against the same prefix. That is
  accepted while the operator session is the sole writer. A lock file or
  a daemon is forbidden by operator ruling, not merely out of scope.
- The log is APPEND-ONLY. No code path here opens the ops file for
  writing, truncation or replacement.
- A corrupt log line is :class:`CorruptHubLogError` naming the file and
  the 1-BASED line number. The reader never silently skips a line, and a
  corrupt CURSOR file refuses for the same reason: resetting to zero in
  silence would replay every ask a worker had already answered.
- **Ingest never writes to a lane.** It opens the two lane directories
  read-only, and a witness hashes every file before and after.
- :func:`read_entries` and :func:`append_entry` are
  ``work_queue_file.read_ops`` and ``append_op`` LIFTED over
  :class:`~oc_orchestrator.hub.HubEntry`. Those two are typed on ``QueueOp``
  (``work_queue_file.py:214`` and ``:300``); widening the queue's types
  to carry hub messages would make one shape serve two contracts, so the
  bodies are lifted and this line says so.
- :func:`find_comms_root` is ``plan_status.find_comms_root``
  LIFTED for the same reason in the other direction: importing it would
  pull the status tool's pydantic, YAML and observer closure into the
  hub's, and the extractor is what decides such questions here.
- The lane constants MIRROR ``option_c_connector_contracts.LANES``
  exactly as the other two lane writers mirror theirs; a witness
  compares the two rather than trusting the copy.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_orchestrator.hub (the contract)
- oc_orchestrator.work_queue_file (the ONE ``DEFAULT_QUEUE_DIR``)
"""
 
from __future__ import annotations
 
import json
import os
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
 
from oc_diagnostics import _dbg
 
from oc_orchestrator.hub import (
    ADDRESS_ORCHESTRATOR,
    ADDRESS_OPERATOR_LOCAL,
    ADDRESS_OPERATOR_REMOTE,
    KIND_ALERT,
    KIND_ASK,
    KIND_REPORT,
    KIND_RULING,
    REQUIRES_ORCHESTRATOR,
    DeliveryReceipt,
    HubEntry,
    HubError,
    HubMessage,
    Provenance,
    check_kind,
)
from oc_orchestrator.work_queue_file import DEFAULT_QUEUE_DIR
 
_dbg("OC_CHAT.MAIN", "module", "hub_file loaded", 4)
 
 
# ── Constants ────────────────────────────────────────────────────────────
 
 
# Where the two files live, relative to the caller's root; injected
# through :class:`HubPaths`, never discovered. IMPORTED from the queue
# rather than restated: the hub log and the queue log are the same
# directory because both are the orchestration state of one workspace,
# and before plan 272 that fact was two literals in two subprojects that
# nothing held together. Now it is one, re-exported through
# :data:`__all__` so ``hub_file.DEFAULT_QUEUE_DIR`` still resolves and a
# monkeypatch of it still reaches :class:`HubPaths`'s default factory.
 
#: The append-only log of recorded messages.
HUB_OPS_FILENAME = "hub.ops.jsonl"
 
#: Last-read sequence number per address.
HUB_CURSORS_FILENAME = "hub.cursors.json"
 
#: Lane constants. These mirror ``option_c_connector_contracts.LANES`` on
#: purpose -- see the module Contracts for why the mirror, rather than an
#: import, is the point.
HUB_LANE = "hub"
HUB_LANE_SOURCE = "hub"
HUB_LANE_SCHEMA_VERSION = "1.0"
HUB_LANE_DIRECTION = "outbound"
HUB_LANE_RISK_CLASS = "low"
 
#: The two lanes ingest reads. Names only -- ingest never imports either
#: writer, exactly as the writers never import a connector.
WORKER_REPORT_LANE = "worker-reports"
SESSION_ALERT_LANE = "session-alerts"
 
#: Workspace walk constants, mirroring ``plan_status.py``.
WORKSPACE_MARKER = ".option_c_workspace.json"
PROJECT_MARKER = ".option_c"
COMMS_ANCHOR = "oc_chat"
COMMS_SUBPATH = (".option_c", "comms")
 
 
class CorruptHubLogError(HubError):
    """A stored line, a cursor file, or a workspace walk did not resolve."""
 
 
# ── Paths ────────────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class HubPaths:
    """All caller-controlled filesystem inputs to one hub log.
 
    The :class:`~oc_orchestrator.work_queue_file.QueuePaths` shape: a frozen
    dataclass whose defaults are today's values, constructed at the
    boundary and passed whole. ``queue_dir`` is late-bound through a
    default factory so a caller that monkeypatches
    :data:`DEFAULT_QUEUE_DIR` is still honoured.
    """
 
    queue_dir: Path = field(default_factory=lambda: DEFAULT_QUEUE_DIR)
    #: Overrides the derived ``<queue_dir>/hub.ops.jsonl``.
    ops_path_override: Path | None = None
    #: Overrides the derived ``<queue_dir>/hub.cursors.json``.
    cursors_path_override: Path | None = None
 
    @property
    def ops_path(self) -> Path:
        """The append-only log: the override, else derived."""
        if self.ops_path_override is not None:
            return self.ops_path_override
        return self.queue_dir / HUB_OPS_FILENAME
 
    @property
    def cursors_path(self) -> Path:
        """The per-recipient cursors: the override, else derived."""
        if self.cursors_path_override is not None:
            return self.cursors_path_override
        return self.queue_dir / HUB_CURSORS_FILENAME
 
 
# ── The log ──────────────────────────────────────────────────────────────
 
 
def read_entries(ops_path: Path) -> tuple[HubEntry, ...]:
    """Read every recorded entry in *ops_path*, oldest first.
 
    A missing file is an empty log, never a ``FileNotFoundError`` -- the
    same tolerance ``work_queue_file.read_ops`` has, whose body this is,
    lifted over :class:`~oc_orchestrator.hub.HubEntry`.
 
    Raises:
        CorruptHubLogError: SYSTEM_ERROR (L1 for the caller) -- a
            non-blank line is not a JSON object carrying a sequence
            number and a message with a known kind. The message names
            the file and the 1-based line number; the reader never
            silently skips it.
    """
    func = "read_entries"
    path = Path(ops_path)
    try:
        text = path.read_text(encoding="utf-8")
    except FileNotFoundError:
        _dbg("OC_CHAT.MAIN", func, f"{path}: absent; empty log", 2)
        return ()
    entries: list[HubEntry] = []
    for number, line in enumerate(text.splitlines(), start=1):
        if not line.strip():
            continue
        entries.append(_decode_entry(line, path, number))
    _dbg("OC_CHAT.MAIN", func, f"{path}: {len(entries)} entr(ies)", 2)
    return tuple(entries)
 
 
def _decode_entry(line: str, path: Path, number: int) -> HubEntry:
    """One log line to a :class:`HubEntry`, or a refusal that names it."""
    func = "_decode_entry"
    try:
        raw = json.loads(line)
    except json.JSONDecodeError as exc:
        _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: line {number}: {exc}", 1, override=True)
        raise CorruptHubLogError(
            f"{path.name}: line {number} is not JSON: {exc}"
        ) from exc
    if not isinstance(raw, dict):
        raise CorruptHubLogError(
            f"{path.name}: line {number} is not a JSON object: {type(raw).__name__}"
        )
    missing = {"seq", "message"} - set(raw)
    if missing:
        raise CorruptHubLogError(
            f"{path.name}: line {number} is missing {', '.join(sorted(missing))}"
        )
    payload = raw["message"]
    if not isinstance(payload, dict):
        raise CorruptHubLogError(
            f"{path.name}: line {number} has a non-object message"
        )
    absent = {"message_id", "at", "kind", "sender", "recipient"} - set(payload)
    if absent:
        raise CorruptHubLogError(
            f"{path.name}: line {number} message is missing "
            f"{', '.join(sorted(absent))}"
        )
    try:
        check_kind(str(payload["kind"]))
    except HubError as exc:
        raise CorruptHubLogError(f"{path.name}: line {number}: {exc}") from exc
    try:
        message = HubMessage.from_mapping(payload)
    except (HubError, KeyError, TypeError, ValueError) as exc:
        raise CorruptHubLogError(
            f"{path.name}: line {number} does not decode: {exc}"
        ) from exc
    return HubEntry(seq=int(raw["seq"]), message=message)
 
 
def _encode_entry(entry: HubEntry) -> str:
    """One :class:`HubEntry` to its log line, without the newline."""
    return json.dumps(
        {"seq": entry.seq, "message": entry.message.as_dict()},
        ensure_ascii=False,
        sort_keys=True,
    )
 
 
def append_entry(ops_path: Path, entry: HubEntry) -> None:
    """Append *entry* to the log, flushed and fsynced.
 
    The ONLY write path for the log, and it opens the file in append
    mode. ``work_queue_file.append_op``'s body, lifted: nothing in this
    module truncates or replaces the ops file.
    """
    func = "append_entry"
    path = Path(ops_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8", newline="\n") as handle:
        handle.write(_encode_entry(entry) + "\n")
        handle.flush()
        os.fsync(handle.fileno())
    _dbg("OC_CHAT.MAIN", func,
         f"{path.name} += {entry.message.kind} (seq {entry.seq})", 2)
 
 
class FileHubLog:
    """The default log: an append-only file plus per-recipient cursors."""
 
    def __init__(self, paths: HubPaths | None = None) -> None:
        func = "FileHubLog.__init__"
        self.paths = paths if paths is not None else HubPaths()
        _dbg("OC_CHAT.MAIN", func, f"over {self.paths.queue_dir}", 2)
 
    # -- the HubLog contract ----------------------------------------------
 
    def record(self, message: HubMessage) -> HubEntry:
        """Append *message* and return the entry it became."""
        entry = HubEntry(seq=len(read_entries(self.paths.ops_path)) + 1, message=message)
        append_entry(self.paths.ops_path, entry)
        return entry
 
    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."""
        return tuple(
            entry
            for entry in read_entries(self.paths.ops_path)
            if (kind is None or entry.message.kind == kind)
            and (recipient is None or entry.message.recipient == recipient)
            and (subject is None or entry.message.subject == subject)
            and (since_seq is None or entry.seq > since_seq)
        )
 
    def open_asks(self, *, recipient: str | None = None) -> tuple[HubEntry, ...]:
        """Asks with no ruling in the log -- the questions still waiting."""
        entries = read_entries(self.paths.ops_path)
        ruled = {
            entry.message.in_reply_to
            for entry in entries
            if entry.message.kind == KIND_RULING
        }
        return tuple(
            entry
            for entry in entries
            if entry.message.kind == KIND_ASK
            and entry.message.message_id not in ruled
            and (recipient is None or entry.message.recipient == recipient)
        )
 
    def ruling_for(self, ask_id: str) -> HubEntry | None:
        """The ruling that answered *ask_id*, or ``None``."""
        for entry in read_entries(self.paths.ops_path):
            if entry.message.kind == KIND_RULING and entry.message.in_reply_to == ask_id:
                return entry
        return None
 
    # -- the cursor -------------------------------------------------------
 
    def cursor_for(self, address: str) -> int:
        """The last sequence number *address* has read; ``0`` when never.
 
        Raises:
            CorruptHubLogError: SYSTEM_ERROR -- the cursor file does not
                parse. Resetting to zero in silence would replay every
                ask the reader had already answered, which is worse than
                stopping.
        """
        return int(self._cursors().get(address, 0))
 
    def unread_for(self, address: str) -> tuple[HubEntry, ...]:
        """Entries addressed to *address* that its cursor has not passed."""
        cursor = self.cursor_for(address)
        return self.entries(recipient=address, since_seq=cursor)
 
    def advance_cursor(self, address: str, seq: int) -> int:
        """Move *address*'s cursor forward to *seq*. Never backwards.
 
        Monotonic on purpose: two readers for one address (a resumed
        session beside a stale one) must not be able to un-read what the
        other consumed.
        """
        func = "FileHubLog.advance_cursor"
        cursors = self._cursors()
        current = int(cursors.get(address, 0))
        if seq <= current:
            _dbg("OC_CHAT.MAIN", func, f"{address}: {seq} <= {current}; unchanged", 3)
            return current
        cursors[address] = int(seq)
        self._write_cursors(cursors)
        _dbg("OC_CHAT.MAIN", func, f"{address}: {current} -> {seq}", 2)
        return int(seq)
 
    def _cursors(self) -> dict[str, int]:
        func = "FileHubLog._cursors"
        path = self.paths.cursors_path
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except FileNotFoundError:
            return {}
        except json.JSONDecodeError as exc:
            _dbg("OC_CHAT.MAIN", func, f"SYSTEM_ERROR: {path.name}: {exc}", 1, override=True)
            raise CorruptHubLogError(
                f"{path.name} does not parse ({exc}); refusing to reset every "
                "cursor to zero, which would replay answered asks"
            ) from exc
        if not isinstance(raw, dict):
            raise CorruptHubLogError(
                f"{path.name} is not a JSON object: {type(raw).__name__}"
            )
        return {str(key): int(value) for key, value in raw.items()}
 
    def _write_cursors(self, cursors: dict[str, int]) -> None:
        """Rewrite the cursor file atomically. The LOG is never rewritten."""
        path = self.paths.cursors_path
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = json.dumps(cursors, indent=2, sort_keys=True) + "\n"
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", newline="\n", dir=path.parent,
            prefix=".hub_cursors_", suffix=".tmp", delete=False,
        ) as tmp:
            tmp.write(payload)
            tmp_path = Path(tmp.name)
        tmp_path.replace(path)
 
 
# ── The comms-bus walk ───────────────────────────────────────────────────
 
 
def find_comms_root(start: Path) -> Path:
    """Return the workspace-anchored comms-bus root above *start*.
 
    ``plan_status.find_comms_root``'s body, LIFTED rather than
    imported -- see the module Contracts. The rules are unchanged: the
    workspace marker wins over a nearer project marker because connector
    state is workspace-wide, and the bus sits beside the dev engine's
    orchestration state when an ``oc_chat`` subproject is present.
    Guessing a root would post a message somewhere nobody is listening,
    so a failed walk raises.
 
    Raises:
        CorruptHubLogError: VALIDATION_ERROR (L3) -- no marker at or
            above *start*; the message names both markers it looked for.
    """
    func = "find_comms_root"
    origin = Path(start).expanduser().resolve()
    project_fallback: Path | None = None
    anchor: Path | None = None
    for parent in (origin, *origin.parents):
        if (parent / WORKSPACE_MARKER).is_file():
            anchor = parent
            break
        if project_fallback is None and (parent / PROJECT_MARKER).is_dir():
            project_fallback = parent
    if anchor is None:
        anchor = project_fallback
    if anchor is None:
        _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: no marker above {origin}", 3)
        raise CorruptHubLogError(
            f"no Option C workspace marker ({WORKSPACE_MARKER}) or project "
            f"marker ({PROJECT_MARKER}/) was found at or above {origin}"
        )
    if (anchor / COMMS_ANCHOR).is_dir():
        return anchor.joinpath(COMMS_ANCHOR, *COMMS_SUBPATH)
    return anchor.joinpath(*COMMS_SUBPATH)
 
 
# ── The lane carrier ─────────────────────────────────────────────────────
 
 
def _utc_timestamp(now: datetime) -> str:
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    return now.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
 
 
def utc_stamp(now: datetime) -> str:
    """Return a filename-safe UTC stamp; colons are illegal on Windows."""
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    return now.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
 
 
def hub_lane_event(message: HubMessage, *, now: datetime) -> dict[str, object]:
    """Build one ``lanes/hub/`` event: the shared envelope plus the message.
 
    The envelope is the one the worker-report and session-alert lanes
    already carry -- ``schema_version``, ``event_id``, ``lane``,
    ``direction``, ``risk_class``, ``created_at_utc``, ``summary``,
    ``relayed_at_utc``, ``relay_chunk_count`` -- so the relay that reads
    those two reads this one with the same code. ``event_id`` IS the
    message id, which is what lets a reply correlate back to the message
    rather than merely to a send.
    """
    event: dict[str, object] = {
        "schema_version": HUB_LANE_SCHEMA_VERSION,
        "event_id": message.message_id,
        "lane": HUB_LANE,
        "direction": HUB_LANE_DIRECTION,
        "risk_class": HUB_LANE_RISK_CLASS,
        "created_at_utc": _utc_timestamp(now),
        "summary": message.summary,
        "relayed_at_utc": None,
        "relay_chunk_count": None,
    }
    event.update(message.as_dict())
    return event
 
 
def write_hub_lane_event(comms_root: Path, event: dict[str, object], *, now: datetime) -> Path:
    """Write one event into ``lanes/hub/`` and return its path.
 
    The atomic temp-then-replace ``core/session_alerts.write_session_alert``
    uses. The lane directory is created when absent: a party must be able
    to hand off from a workspace whose bus has never been initialised.
    Nothing outside the lane is created, read, or written.
    """
    func = "write_hub_lane_event"
    lane_dir = Path(comms_root) / "lanes" / HUB_LANE
    lane_dir.mkdir(parents=True, exist_ok=True)
    name = f"{utc_stamp(now)}_{HUB_LANE_SOURCE}_{event['event_id']}.json"
    path = lane_dir / name
    if path.exists():
        raise CorruptHubLogError(
            f"refusing to overwrite an existing hub lane event at {path}"
        )
    payload = json.dumps(event, indent=2, sort_keys=True) + "\n"
    with tempfile.NamedTemporaryFile(
        mode="w", encoding="utf-8", newline="\n", dir=lane_dir,
        prefix=".hub_", suffix=".tmp", delete=False,
    ) as tmp:
        tmp.write(payload)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
    _dbg("OC_CHAT.MAIN", func, f"wrote hub lane event {path.name}", 2)
    return path
 
 
class FileLaneCarrier:
    """The durable carrier: one file per message under ``lanes/hub/``."""
 
    #: What a receipt and a trace call this carrier.
    name = "file-lane"
 
    def __init__(self, *, comms_root: Path) -> None:
        func = "FileLaneCarrier.__init__"
        self.comms_root = Path(comms_root)
        _dbg("OC_CHAT.MAIN", func, f"over {self.comms_root}", 2)
 
    def present(self, address: str) -> bool:
        """True for every party but the REMOTE operator.
 
        Durable hand-off is the floor every local party has: the
        orchestrator, each ``agent-NNN`` and the operator at the control
        centre all read this directory (or the log beside it) with no
        credential and no process running. ``operator@remote`` is the one
        address a file cannot reach, by definition -- that is the
        connector's carrier, and saying "present" here would silently
        swallow a message the operator is waiting for on their phone.
        """
        return address != ADDRESS_OPERATOR_REMOTE
 
    def deliver(self, message: HubMessage) -> DeliveryReceipt:
        """Write *message* into the lane and report where it landed."""
        now = datetime.now(timezone.utc).replace(microsecond=0)
        path = write_hub_lane_event(self.comms_root, hub_lane_event(message, now=now), now=now)
        return DeliveryReceipt(
            carrier=self.name, message_id=message.message_id, detail=str(path)
        )
 
 
# ── Ingest: the two existing lanes, folded in ────────────────────────────
 
 
def _lane_events(lane_dir: Path) -> list[dict[str, Any]]:
    """Every JSON event in *lane_dir*, in arrival (filename) order.
 
    Read-only by construction: this opens files and nothing else. A file
    that does not parse is SKIPPED rather than refused, because an
    unreadable event on somebody else's lane is not the hub's to fail
    over -- the lane's own reader is the authority on its contents.
    """
    func = "_lane_events"
    if not lane_dir.is_dir():
        return []
    events: list[dict[str, Any]] = []
    for path in sorted(lane_dir.glob("*.json")):
        if not path.is_file() or path.name.startswith("."):
            continue
        try:
            raw = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            # EXTERNAL_ERROR (L2): another writer's file, another writer's
            # problem. Ingest reports it and carries on.
            _dbg("OC_CHAT.MAIN", func,
                 f"EXTERNAL_ERROR: {path.name} did not parse ({type(exc).__name__}); skipped", 2)
            continue
        if isinstance(raw, dict) and raw.get("event_id"):
            events.append(raw)
    return events
 
 
def _message_from_worker_report(event: dict[str, Any]) -> HubMessage | None:
    """A worker-report event as a hub message, or ``None`` if unmappable."""
    func = "_message_from_worker_report"
    agent = str(event.get("agent") or "")
    status = str(event.get("status") or "")
    summary = str(event.get("summary") or "")
    subject = str(event.get("plan_id") or "") or None
    provenance = Provenance(
        actor=agent, actor_type="worker_agent", channel="lane",
        correlation={"lane": WORKER_REPORT_LANE, "event_id": str(event["event_id"])},
    )
    if status == "awaiting_gate":
        kind, requires = KIND_REPORT, None
    elif status == "blocked":
        # A blocked worker is WAITING ON A DECISION, not finished, so it
        # becomes an open ask rather than a report. That is what lets
        # ``fleet`` say "waiting" before anyone calls the worker stalled.
        kind, requires = KIND_ASK, REQUIRES_ORCHESTRATOR
    else:
        _dbg("OC_CHAT.MAIN", func, f"unmapped worker-report status {status!r}", 3)
        return None
    return HubMessage(
        message_id=str(event["event_id"]),
        at=str(event.get("created_at_utc") or ""),
        kind=kind,
        sender=agent,
        recipient=ADDRESS_ORCHESTRATOR,
        subject=subject,
        body=summary,
        requires=requires,
        provenance=provenance,
    )
 
 
def _message_from_session_alert(event: dict[str, Any]) -> HubMessage:
    """A session-alert event as an alert to the operator at the control centre."""
    return HubMessage(
        message_id=str(event["event_id"]),
        at=str(event.get("created_at_utc") or ""),
        kind=KIND_ALERT,
        sender=ADDRESS_ORCHESTRATOR,
        recipient=ADDRESS_OPERATOR_LOCAL,
        subject=None,
        body=str(event.get("summary") or ""),
        provenance=Provenance(
            actor=ADDRESS_ORCHESTRATOR, actor_type="orchestrator_agent", channel="lane",
            correlation={"lane": SESSION_ALERT_LANE, "event_id": str(event["event_id"])},
        ),
    )
 
 
def ingest_lanes(comms_root: Path, log: Any) -> tuple[HubEntry, ...]:
    """Record every lane event *log* does not already hold. Returns the new ones.
 
    The two existing pipes join the hub by being READ: no lane file is
    opened for writing, neither existing writer is asked to change, and
    running this twice records nothing the second time. Idempotency is by
    ``event_id``, which is also the recorded message's id -- one event,
    one message, for the life of the log.
 
    An event's OWN ``created_at_utc`` becomes the message's ``at``, so
    ingesting history does not rewrite it to the moment it was noticed.
    """
    func = "ingest_lanes"
    root = Path(comms_root)
    seen = {
        str(entry.message.provenance.correlation.get("event_id"))
        for entry in log.entries()
        if entry.message.provenance.channel == "lane"
    }
    recorded: list[HubEntry] = []
 
    for event in _lane_events(root / "lanes" / WORKER_REPORT_LANE):
        if str(event["event_id"]) in seen:
            continue
        message = _message_from_worker_report(event)
        if message is None:
            continue
        recorded.append(log.record(message))
 
    for event in _lane_events(root / "lanes" / SESSION_ALERT_LANE):
        if str(event["event_id"]) in seen:
            continue
        recorded.append(log.record(_message_from_session_alert(event)))
 
    if recorded:
        _dbg("OC_CHAT.MAIN", func, f"ingested {len(recorded)} lane event(s)", 2)
    return tuple(recorded)
 
 
__all__ = [
    "DEFAULT_QUEUE_DIR",
    "HUB_CURSORS_FILENAME",
    "HUB_LANE",
    "HUB_LANE_DIRECTION",
    "HUB_LANE_RISK_CLASS",
    "HUB_LANE_SCHEMA_VERSION",
    "HUB_LANE_SOURCE",
    "HUB_OPS_FILENAME",
    "SESSION_ALERT_LANE",
    "WORKER_REPORT_LANE",
    "CorruptHubLogError",
    "FileHubLog",
    "FileLaneCarrier",
    "HubPaths",
    "append_entry",
    "find_comms_root",
    "hub_lane_event",
    "ingest_lanes",
    "read_entries",
    "utc_stamp",
    "write_hub_lane_event",
]