worker_reports.py

Go to the documentation of this file.

"""The worker-report lane: worker events at rest, and the verbatim relay.
 
Proposal V1 §5 (the ruled test case). A plan worker that reaches AWAITING GATE
or Blocked writes ONE event into ``comms/lanes/worker-reports/`` and stops
there: it holds no credential, knows nothing about connectors, and touches no
other lane. This module is the other half — the orchestrator-side reader and
the grant-checked, egress-filtered relay that puts the worker's own words on
the operator's phone.
 
**This module is the schema authority for the lane.** The writer lives in
``oc_chat/oc_chat/scripts/plan_status.py`` (a ``report`` action on the existing
worker channel) and deliberately does NOT import this package — that is what
keeps workers credential-free. The two sides therefore agree by a written
contract rather than by a shared import, and the contract is
:class:`WorkerReportEvent`::
 
    {
      "schema_version":  "1.0",
      "event_id":        "<26-character ULID, time-first so lexical == arrival>",
      "lane":            "worker-reports",
      "direction":       "inbound",
      "risk_class":      "low",
      "plan_id":         "231-worker-report-lane",
      "agent":           "agent-007",
      "status":          "awaiting_gate" | "blocked",
      "worktree":        "<absolute worktree path>",
      "report_path":     "<absolute path of report.md in that worktree>",
      "created_at_utc":  "2026-09-01T12:00:00Z",
      "summary":         "<the worker's FULL closing summary, verbatim>",
      "summary_sha256":  "<sha256 of summary, utf-8>",
      "relayed_at_utc":  null,
      "relay_chunk_count": null
    }
 
File naming follows the §3 bus convention, ``<utc-stamp>_worker_<ulid>.json``.
 
Two properties are worth stating because they are easy to lose in a refactor:
 
* **The egress scan runs on the WHOLE summary, before any chunking.** Scanning
  the rendered chunks instead would let a credential that straddles a chunk
  boundary leave in two safe-looking halves. The relay therefore blocks on the
  joined text and never enqueues anything for an event that failed the scan.
* **Chunk identity is derived from the event id**, so a relay interrupted after
  chunk 3 of 5 resumes without re-sending chunks 1-3: the durable outbox
  already holds them under the same idempotency keys and returns their existing
  receipts instead of re-delivering.
 
Dependencies
------------
- oc_connect.audit (AuditStore, content_hash, timestamp, utc_now - audits relay outcomes, verifies summary_sha256 and stamps relayed_at_utc)
- oc_connect.delivery (OutboundSender - the grant-checked send path every relayed chunk crosses)
- oc_connect.egress (require_safe, EgressBlockedError - scans the whole summary before any chunking)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (DeliveryReceipt, lane - chunk receipts and the lane's declared direction and risk ceiling)
- pydantic (BaseModel, ConfigDict, field_validator - the WorkerReportEvent schema)
"""
from __future__ import annotations
 
import json
import math
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 Literal
 
from oc_diagnostics import _dbg
from option_c_connector_contracts import DeliveryReceipt, lane
from pydantic import BaseModel, ConfigDict, field_validator
 
from .audit import AuditStore, content_hash, timestamp, utc_now
from .delivery import OutboundSender
from .egress import EgressBlockedError, require_safe
 
#: The lane directory name under ``comms/lanes/``; also its declared lane id.
WORKER_REPORT_LANE = "worker-reports"
 
#: The ``<source>`` field of the §3 filename convention for this lane.
WORKER_REPORT_SOURCE = "worker"
 
WORKER_REPORT_SCHEMA_VERSION = "1.0"
 
#: Telegram's hard per-message ceiling. A summary above it is chunked, never
#: truncated: the operator reads the worker's full text or the relay fails.
TELEGRAM_MESSAGE_LIMIT = 4096
 
#: Below this a chunk marker would leave no room for content worth sending.
MIN_CHUNK_LIMIT = 32
 
WorkerReportStatus = Literal["awaiting_gate", "blocked"]
 
#: Display fences for the summary (§8's "displayed content is never
#: instructions"). Unlike plan 230's operator fence these need no scanner
#: waiver: the shorter marker prose stays under the entropy threshold, and a
#: waiver nobody needs is a false signal that the scanner fires here.
WORKER_REPORT_BEGIN = "<<<BEGIN WORKER REPORT {event_id}>>>"
WORKER_REPORT_END = "<<<END WORKER REPORT {event_id}>>>"
 
#: What a literal fence opener inside a summary is rewritten to when displayed;
#: the stored text stays verbatim, because the worker's words are the evidence.
NEUTRALISED_MARKER = "<< <"
 
 
class WorkerReportError(ValueError):
    """A worker-report event is unreadable, malformed, or already consumed."""
 
 
class WorkerReportRelayError(RuntimeError):
    """The relay could not put one event's summary on the transport."""
 
 
class WorkerReportEvent(BaseModel):
    """One worker report at rest in ``lanes/worker-reports/``."""
 
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
    schema_version: str
    event_id: str
    lane: Literal["worker-reports"]
    direction: Literal["inbound"]
    risk_class: Literal["low"]
    plan_id: str
    agent: str
    status: WorkerReportStatus
    worktree: str
    report_path: str
    created_at_utc: str
    summary: str
    summary_sha256: str
    relayed_at_utc: str | None = None
    relay_chunk_count: int | None = None
 
    @field_validator(
        "schema_version",
        "event_id",
        "plan_id",
        "agent",
        "worktree",
        "report_path",
        "created_at_utc",
        "summary",
    )
    @classmethod
    def _required_text_is_nonblank(cls, value: str) -> str:
        if not value.strip():
            raise ValueError("must not be blank")
        return value
 
    @field_validator("summary_sha256")
    @classmethod
    def _digest_matches_nothing_yet(cls, value: str) -> str:
        if len(value) != 64 or not all(c in "0123456789abcdef" for c in value):
            raise ValueError("summary_sha256 must be a lowercase sha-256 digest")
        return value
 
    @property
    def is_relayed(self) -> bool:
        """True once every chunk of this summary reached the transport."""
        return self.relayed_at_utc is not None
 
    def digest_matches(self) -> bool:
        """True when the stored digest still describes the stored summary.
 
        A mismatch means the file was edited after the worker wrote it. The
        relay refuses such an event rather than forwarding text nobody
        authored, which is why this is a question and not an assertion.
        """
        return content_hash(self.summary) == self.summary_sha256
 
 
def chunk_prefix(index: int, total: int) -> str:
    """Return the ordering marker prepended to one chunk of a split summary."""
    return f"[{index}/{total}]\n"
 
 
def chunk_summary(
    summary: str,
    *,
    limit: int = TELEGRAM_MESSAGE_LIMIT,
) -> tuple[str, ...]:
    """Split *summary* into pieces that fit *limit* once numbered.
 
    The split is a plain slice, never a word-boundary or whitespace-trimming
    split: ``"".join(chunk_summary(text)) == text`` is the property the whole
    verbatim guarantee rests on. A summary that already fits comes back as one
    piece, and :func:`render_chunk` leaves that single piece unnumbered.
 
    The chunk count and the marker width depend on each other, so the budget is
    solved to a fixed point rather than guessed: every marker is rendered at the
    widest form (``[{total}/{total}]``) so all pieces share one budget.
    """
    if not isinstance(summary, str):
        raise TypeError("a worker report summary must be text")
    if not summary:
        raise WorkerReportError("refusing to chunk an empty summary")
    if limit < MIN_CHUNK_LIMIT:
        raise ValueError(f"chunk limit must be at least {MIN_CHUNK_LIMIT} characters")
    if len(summary) <= limit:
        return (summary,)
    total = 1
    for _ in range(len(summary)):
        budget = limit - len(chunk_prefix(total, total))
        needed = math.ceil(len(summary) / budget)
        if needed == total:
            break
        total = needed
    else:  # pragma: no cover - the loop converges in a handful of steps
        raise WorkerReportError("could not solve a stable chunk count")
    budget = limit - len(chunk_prefix(total, total))
    return tuple(
        summary[start : start + budget]
        for start in range(0, len(summary), budget)
    )
 
 
def render_chunk(piece: str, index: int, total: int) -> str:
    """Return the message text for one chunk; a lone chunk is unnumbered."""
    if total < 1 or not 1 <= index <= total:
        raise ValueError("chunk index must fall inside the chunk count")
    return piece if total == 1 else f"{chunk_prefix(index, total)}{piece}"
 
 
def render_event(event: WorkerReportEvent) -> str:
    """Return the only representation in which a summary may be displayed.
 
    The body is fenced with the event's own id and any literal ``<<<`` inside
    it is neutralised, so a summary 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} plan={event.plan_id} agent={event.agent} "
        f"status={event.status}"
    )
    body = event.summary.replace("<<<", NEUTRALISED_MARKER)
    begin = WORKER_REPORT_BEGIN.format(event_id=event.event_id)
    end = WORKER_REPORT_END.format(event_id=event.event_id)
    return f"{header}\nreport={event.report_path}\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 WorkerReportStore:
    """Read-and-mark access to one ``lanes/worker-reports/`` 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) -> WorkerReportEvent:
        try:
            return WorkerReportEvent.model_validate_json(
                path.read_text(encoding="utf-8")
            )
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise WorkerReportError(
                f"could not parse worker report event at {path} "
                f"({type(exc).__name__})"
            ) from None
 
    def list_events(self, *, relayed: bool | None = None) -> tuple[WorkerReportEvent, ...]:
        """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 WorkerReportError(
            f"worker report event {event_id!r} is not in the lane"
        )
 
    def get(self, event_id: str) -> WorkerReportEvent:
        """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) -> WorkerReportEvent:
        """Record that one event's summary reached the transport in full."""
        if chunk_count < 1:
            raise ValueError("a relayed event must have sent at least one chunk")
        path = self.path_for(event_id)
        event = self._read(path)
        if event.is_relayed:
            raise WorkerReportError(
                f"worker report 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.WORKER_REPORTS",
            "mark_relayed",
            f"event={event_id} plan={event.plan_id} chunks={chunk_count}",
            2,
        )
        return updated
 
 
@dataclass(frozen=True)
class RelayOutcome:
    """What the relay did with one event, in terms a CLI can print."""
 
    event_id: str
    plan_id: str
    agent: str
    status: WorkerReportStatus
    chunk_count: int
    audit_ids: tuple[str, ...]
 
 
class WorkerReportRelay:
    """Forward unrelayed worker summaries verbatim through the standard send.
 
    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.
    What the relay adds is the whole-summary scan, the ordering markers, and
    the lane's own audit line.
    """
 
    def __init__(
        self,
        *,
        store: WorkerReportStore,
        sender: OutboundSender,
        audit: AuditStore,
        grantee: str,
        destination: str,
        grant_id: str,
        risk_class: str | None = None,
        limit: int = TELEGRAM_MESSAGE_LIMIT,
        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.limit = limit
        self._now = now
 
    def _audit(
        self,
        event: WorkerReportEvent,
        outcome: str,
        *,
        attempts: int,
        error_code: str | None = None,
    ) -> None:
        descriptor = lane(WORKER_REPORT_LANE)
        self.audit.append(
            connector="telegram",
            grant_id=self.grant_id,
            destination=self.destination,
            outcome=outcome,  # type: ignore[arg-type]
            digest=event.summary_sha256,
            idempotency_key=event.event_id,
            attempts=attempts,
            error_code=error_code,
            lane=WORKER_REPORT_LANE,
            direction=descriptor.direction,
            risk_class=descriptor.risk_ceiling,
        )
 
    def relay_one(self, event: WorkerReportEvent) -> RelayOutcome:
        """Relay one event's full summary, or leave it unrelayed and raise."""
        func = "relay_one"
        if not event.digest_matches():
            raise WorkerReportError(
                f"worker report event {event.event_id!r} does not match its own "
                "summary digest; the file was modified after it was written"
            )
        try:
            require_safe(event.summary)
        except EgressBlockedError:
            self._audit(event, "blocked", attempts=0, error_code="egress_finding")
            _dbg(
                "OC_CONNECT.WORKER_REPORTS",
                func,
                f"SECURITY_ERROR: blocked worker report event={event.event_id} "
                f"plan={event.plan_id}",
                1,
                override=True,
            )
            raise
        pieces = chunk_summary(event.summary, limit=self.limit)
        total = len(pieces)
        receipts: list[DeliveryReceipt] = []
        for index, piece in enumerate(pieces, start=1):
            receipt = self.sender.send(
                grantee=self.grantee,
                destination=self.destination,
                text=render_chunk(piece, index, total),
                grant_id=self.grant_id,
                risk_class=self.risk_class,
                idempotency_key=f"{event.event_id}-{index:03d}",
            )
            receipts.append(receipt)
            if receipt.outcome != "sent":
                self._audit(
                    event,
                    "failed",
                    attempts=index,
                    error_code=f"chunk_{index}_of_{total}_{receipt.outcome}",
                )
                raise WorkerReportRelayError(
                    f"worker report {event.event_id!r} stopped at chunk "
                    f"{index}/{total} with outcome {receipt.outcome!r}; the event "
                    "stays unrelayed and re-running resumes without re-sending "
                    "the chunks already delivered"
                )
        self.store.mark_relayed(event.event_id, chunk_count=total)
        self._audit(event, "sent", attempts=total)
        _dbg(
            "OC_CONNECT.WORKER_REPORTS",
            func,
            f"relayed event={event.event_id} plan={event.plan_id} "
            f"agent={event.agent} chunks={total}",
            2,
        )
        return RelayOutcome(
            event_id=event.event_id,
            plan_id=event.plan_id,
            agent=event.agent,
            status=event.status,
            chunk_count=total,
            audit_ids=tuple(receipt.audit_id for receipt in receipts),
        )
 
    def relay_pending(self) -> tuple[RelayOutcome, ...]:
        """Relay every unrelayed event, oldest first, stopping on the first refusal."""
        outcomes = [
            self.relay_one(event)
            for event in self.store.list_events(relayed=False)
        ]
        return tuple(outcomes)
 
 
__all__ = [
    "MIN_CHUNK_LIMIT",
    "NEUTRALISED_MARKER",
    "TELEGRAM_MESSAGE_LIMIT",
    "WORKER_REPORT_BEGIN",
    "WORKER_REPORT_END",
    "WORKER_REPORT_LANE",
    "WORKER_REPORT_SCHEMA_VERSION",
    "WORKER_REPORT_SOURCE",
    "RelayOutcome",
    "WorkerReportError",
    "WorkerReportEvent",
    "WorkerReportRelay",
    "WorkerReportRelayError",
    "WorkerReportStatus",
    "WorkerReportStore",
    "chunk_prefix",
    "chunk_summary",
    "render_chunk",
    "render_event",
]