comms.py

Go to the documentation of this file.

"""The comms bus: one workspace-anchored root with lane-separated storage.
 
Proposal V1 §3, verbatim layout::
 
    oc_chat/.option_c/comms/
      outbound/            # the plan-219 durable outbox, relocated here
      inbound/
        telegram/          # operator replies, one JSON file per message
      lanes/
        worker-reports/    # created by plan 231, not here
      audit/               # hash-only audit trail across all lanes
 
Lane separation is structural, not advisory: the directories exist, each
consumer is handed only the directory it may touch, and every stored record
names its lane, direction and risk class (§9's binding rule).
 
The plan-219 state moves in through :meth:`CommsBus.migrate_from`, which is a
*verified* move: hash the source, write the destination, re-read and re-hash the
destination, and only then unlink the source. A mismatch aborts with the source
still in place, so a failed migration loses nothing.
 
Dependencies
------------
- oc_connect.workspace (comms_root - the workspace-anchored root of the bus)
- oc_diagnostics (_dbg structured logger)
"""
from __future__ import annotations
 
import hashlib
import json
import os
import secrets as _secrets
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from .workspace import comms_root
 
#: Crockford base32, the ULID alphabet: no I, L, O or U.
# 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)"
_CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
 
OUTBOX_FILENAME = "outbox.json"
AUDIT_FILENAME = "audit.jsonl"
 
 
class CommsMigrationError(RuntimeError):
    """A one-shot state migration could not be completed and verified."""
 
 
def _encode(value: int, length: int) -> str:
    digits = []
    for _ in range(length):
        value, remainder = divmod(value, 32)
        digits.append(_CROCKFORD[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.
 
    Time-first encoding means lexical order is arrival order, which is what
    makes the flood bound's oldest-first rule a plain directory sort.
    *entropy* exists so tests can pin the random half; production leaves it
    ``None`` and draws from :mod:`secrets`.
    """
    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 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 message_filename(now: datetime, source: str, ulid: str) -> str:
    """Return the ``<utc-stamp>_<source>_<ulid>.json`` name from §3."""
    if not source.strip() or "_" in source:
        raise ValueError("message source must be non-blank and contain no underscore")
    return f"{utc_stamp(now)}_{source}_{ulid}.json"
 
 
def content_digest(payload: bytes) -> str:
    """Return the lowercase SHA-256 digest of raw file bytes."""
    return hashlib.sha256(payload).hexdigest()
 
 
@dataclass(frozen=True)
class MigratedFile:
    """One verified relocation: identical bytes at a new path."""
 
    name: str
    source: Path
    destination: Path
    sha256: str
    byte_count: int
    record_count: int
 
 
@dataclass(frozen=True)
class MigrationReport:
    """What the one-shot move did, in a form a ready report can quote."""
 
    moved: tuple[MigratedFile, ...]
    skipped: tuple[str, ...]
 
    def summary(self) -> str:
        """Return an ASCII one-line-per-file summary for operator reports."""
        if not self.moved:
            return "comms migration: nothing to move"
        lines = [
            f"{entry.name}: {entry.record_count} records, "
            f"{entry.byte_count} bytes, sha256={entry.sha256}"
            for entry in self.moved
        ]
        return "\n".join(lines)
 
 
def _atomic_write(destination: Path, payload: bytes) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_name(
        f".tmp-{destination.name}-{_secrets.token_hex(8)}"
    )
    try:
        with temporary.open("wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, destination)
    finally:
        if temporary.exists():
            try:
                temporary.unlink()
            except OSError:
                pass
 
 
def _count_records(name: str, payload: bytes) -> int:
    try:
        text = payload.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise CommsMigrationError(
            f"refusing to migrate {name}: it is not UTF-8 ({type(exc).__name__})"
        ) from None
    if name.endswith(".jsonl"):
        return sum(1 for line in text.splitlines() if line.strip())
    try:
        document = json.loads(text)
    except json.JSONDecodeError as exc:
        raise CommsMigrationError(
            f"refusing to migrate {name}: it is not valid JSON ({type(exc).__name__})"
        ) from None
    items = document.get("items") if isinstance(document, dict) else None
    return len(items) if isinstance(items, list) else 0
 
 
class CommsBus:
    """Resolve, create, and populate the comms bus for one workspace."""
 
    def __init__(self, root: str | Path | None = None) -> None:
        self.root = Path(comms_root(root))
 
    @property
    def outbound_dir(self) -> Path:
        """Return the outbound lane directory."""
        return self.root / "outbound"
 
    @property
    def inbound_dir(self) -> Path:
        """Return the Telegram inbound lane directory."""
        return self.root / "inbound" / "telegram"
 
    @property
    def lanes_dir(self) -> Path:
        """Return the local lane root; plan 231 creates ``worker-reports/``."""
        return self.root / "lanes"
 
    @property
    def worker_reports_dir(self) -> Path:
        """Return the worker-report lane directory (proposal V1 §5).
 
        Deliberately absent from :meth:`directories`: the lane is created by
        whoever first writes or reads it, because the worker-side helper must
        work in a workspace where the bus has never been initialised, and an
        orchestrator that has never received a worker report should not find a
        directory implying it has.
        """
        return self.lanes_dir / "worker-reports"
 
    @property
    def session_alerts_dir(self) -> Path:
        """Return the session-alert lane directory (plan 237 / CAPS-01).
 
        Absent from :meth:`directories` for the same reason as the worker-report
        lane above: the product-side writer must work in a workspace whose bus
        has never been initialised, and an operator who has never had a cost
        alert should not find a directory implying otherwise.
        """
        return self.lanes_dir / "session-alerts"
 
    @property
    def audit_dir(self) -> Path:
        """Return the cross-lane, hash-only audit directory."""
        return self.root / "audit"
 
    @property
    def outbox_path(self) -> Path:
        """Return the durable outbox file inside the outbound lane."""
        return self.outbound_dir / OUTBOX_FILENAME
 
    @property
    def audit_path(self) -> Path:
        """Return the append-only audit file."""
        return self.audit_dir / AUDIT_FILENAME
 
    @property
    def cursor_path(self) -> Path:
        """Return the inbound transport cursor; a dotfile, never a message."""
        return self.inbound_dir / ".cursor.json"
 
    def directories(self) -> tuple[Path, ...]:
        """Return every directory the §3 layout requires, in creation order."""
        return (self.outbound_dir, self.inbound_dir, self.lanes_dir, self.audit_dir)
 
    def initialise(self) -> tuple[Path, ...]:
        """Create the full layout; already-present directories are left alone."""
        func = "initialise"
        for directory in self.directories():
            directory.mkdir(parents=True, exist_ok=True)
        _dbg("OC_CONNECT.COMMS", func, f"comms layout ready at {self.root}", 2)
        return self.directories()
 
    def _plan(self, legacy_dir: Path) -> tuple[tuple[str, Path, Path], ...]:
        return (
            (OUTBOX_FILENAME, legacy_dir / OUTBOX_FILENAME, self.outbox_path),
            (AUDIT_FILENAME, legacy_dir / AUDIT_FILENAME, self.audit_path),
        )
 
    def migrate_from(
        self,
        legacy_dir: str | Path,
        *,
        _write: Callable[[Path, bytes], None] = _atomic_write,
    ) -> MigrationReport:
        """Move plan-219 outbox and audit state in, verifying every byte.
 
        The move is one-shot and idempotent: a source that is already gone, or
        a destination that already holds byte-identical content, is reported as
        skipped rather than repeated. A destination holding *different* content
        aborts the whole migration — silently choosing a winner between two
        divergent audit trails is never the right call.
 
        *_write* is injected only so the negative control can prove the
        verification step can fail; production callers never pass it.
        """
        func = "migrate_from"
        legacy = Path(legacy_dir)
        moved: list[MigratedFile] = []
        skipped: list[str] = []
        self.initialise()
        for name, source, destination in self._plan(legacy):
            if not source.is_file():
                skipped.append(name)
                continue
            payload = source.read_bytes()
            digest = content_digest(payload)
            if destination.exists():
                if content_digest(destination.read_bytes()) == digest:
                    source.unlink()
                    skipped.append(name)
                    continue
                raise CommsMigrationError(
                    f"refusing to migrate {name}: {destination} already holds "
                    "different content; resolve the two files by hand"
                )
            records = _count_records(name, payload)
            _write(destination, payload)
            written = destination.read_bytes()
            if content_digest(written) != digest:
                raise CommsMigrationError(
                    f"migration verification failed for {name}: {destination} "
                    f"does not match the source digest; {source} was left in place"
                )
            source.unlink()
            moved.append(
                MigratedFile(
                    name=name,
                    source=source,
                    destination=destination,
                    sha256=digest,
                    byte_count=len(payload),
                    record_count=records,
                )
            )
            _dbg(
                "OC_CONNECT.COMMS",
                func,
                f"migrated {name} records={records} bytes={len(payload)} "
                f"sha256={digest}",
                2,
            )
        return MigrationReport(moved=tuple(moved), skipped=tuple(skipped))
 
    def message_paths(self) -> tuple[Path, ...]:
        """Return stored inbound message files in arrival (filename) order."""
        return _message_paths(self.inbound_dir)
 
 
def _message_paths(directory: Path) -> tuple[Path, ...]:
    if not directory.is_dir():
        return ()
    candidates: Iterable[Path] = directory.glob("*.json")
    return tuple(
        sorted(
            path
            for path in candidates
            if path.is_file() and not path.name.startswith(".")
        )
    )
 
 
__all__ = [
    "AUDIT_FILENAME",
    "OUTBOX_FILENAME",
    "CommsBus",
    "CommsMigrationError",
    "MigratedFile",
    "MigrationReport",
    "content_digest",
    "message_filename",
    "new_ulid",
    "utc_stamp",
]