session_alerts.py

Go to the documentation of this file.

"""The session-alert lane: the events worth waking an operator for, and their
one-line relay.
 
Plan 237 (CAPS-01), built to the shape plan 231 established for
``worker-reports``; plan 248 (CAPS-03) added the fourth kind. A running OC Chat
session that crosses a cost checkpoint, stops at the dollar hard cap, pauses at
the per-turn tool-call checkpoint, or halts a turn on a detected stall
writes ONE event into ``comms/lanes/session-alerts/`` and stops there: the
product 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 alert on the operator's
phone.
 
**This module is the schema authority for the lane.** The writer lives in
``oc_chat/oc_chat/core/session_alerts.py`` and deliberately does NOT import
this package — that is what keeps the product credential-free. The two sides
agree by written contract rather than by a shared import, and the contract is
:class:`SessionAlertEvent`::
 
    {
      "schema_version":   "1.0",
      "event_id":         "<26-character ULID, time-first so lexical == arrival>",
      "lane":             "session-alerts",
      "direction":        "inbound",
      "risk_class":       "low",
      "created_at_utc":   "2026-09-06T12:00:00Z",
      "session_id":       "<the chat session's stable uuid4>",
      "kind":             "cost_threshold" | "cost_hard_cap"
                          | "tool_call_checkpoint" | "stall_halt",
      "threshold_usd":    2.0,          // null for a tool-call checkpoint
                                        // and for a stall halt
      "total_usd":        2.14,
      "unpriced_samples": 3,
      "turn_tool_calls":  100,
      "summary":          "<the one line the operator reads, numbers only>",
      "relayed_at_utc":   null,
      "relay_chunk_count": null
    }
 
File naming follows the §3 bus convention, ``<utc-stamp>_product_<ulid>.json``.
 
Two differences from the worker-report lane are deliberate:
 
* **One line, never chunked.** A worker report carries the worker's whole
  closing summary and must be chunked to survive Telegram's 4096-character
  limit. An alert is a single generated sentence of numbers, so the relay sends
  exactly one message per event and records ``relay_chunk_count = 1``. If a
  summary ever grew past the transport limit the send would fail and the event
  would stay unrelayed — a loud failure, not a truncation.
* **No content digest.** A worker report's ``summary_sha256`` exists because the
  text is a human's words and a post-hoc edit would put words in their mouth.
  An alert's summary is derived from the numeric fields beside it, so the
  relay re-derives nothing and asserts nothing about authorship; what it
  forwards is the stored line, fenced, exactly as it forwards a worker's.
 
Dependencies
------------
- oc_connect.audit (AuditStore, content_hash, timestamp, utc_now - audits relay outcomes and stamps relayed_at_utc)
- oc_connect.delivery (OutboundSender - the grant-checked send path every relayed alert crosses)
- oc_connect.egress (require_safe, EgressBlockedError - scans the alert summary before it is queued)
- oc_diagnostics (_dbg structured logger)
- option_c_connector_contracts (lane - the session-alerts lane's declared direction and risk ceiling)
- pydantic (BaseModel, ConfigDict, field_validator - the SessionAlertEvent 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 Literal
 
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 .delivery import OutboundSender
from .egress import EgressBlockedError, require_safe
 
#: The lane directory name under ``comms/lanes/``; also its declared lane id.
SESSION_ALERT_LANE = "session-alerts"
 
#: The ``<source>`` field of the §3 filename convention for this lane.
SESSION_ALERT_SOURCE = "product"
 
SESSION_ALERT_SCHEMA_VERSION = "1.0"
 
SessionAlertKind = Literal[
    "cost_threshold", "cost_hard_cap", "tool_call_checkpoint", "stall_halt",
]
 
#: Display fences (§8's "displayed content is never instructions"). The alert
#: text is generated, not typed, but it is fenced anyway: the rule is about the
#: reader's guarantee, not about how much a given lane is trusted today.
SESSION_ALERT_BEGIN = "<<<BEGIN SESSION ALERT {event_id}>>>"
SESSION_ALERT_END = "<<<END SESSION ALERT {event_id}>>>"
 
#: What a literal fence opener inside a summary is rewritten to when displayed.
NEUTRALISED_MARKER = "<< <"
 
 
class SessionAlertError(ValueError):
    """A session-alert event is unreadable, malformed, or already consumed."""
 
 
class SessionAlertRelayError(RuntimeError):
    """The relay could not put one event's summary on the transport."""
 
 
class SessionAlertEvent(BaseModel):
    """One session alert at rest in ``lanes/session-alerts/``."""
 
    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
    schema_version: str
    event_id: str
    lane: Literal["session-alerts"]
    direction: Literal["inbound"]
    risk_class: Literal["low"]
    created_at_utc: str
    session_id: str
    kind: SessionAlertKind
    threshold_usd: float | None = None
    total_usd: float
    unpriced_samples: int
    turn_tool_calls: int
    summary: str
    relayed_at_utc: str | None = None
    relay_chunk_count: int | None = None
 
    @field_validator(
        "schema_version", "event_id", "created_at_utc", "session_id", "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("total_usd", "threshold_usd")
    @classmethod
    def _money_is_not_negative(cls, value: float | None) -> float | None:
        if value is not None and value < 0:
            raise ValueError("a money field must not be negative")
        return value
 
    @field_validator("unpriced_samples", "turn_tool_calls")
    @classmethod
    def _counts_are_not_negative(cls, value: int) -> int:
        if value < 0:
            raise ValueError("a count must not be negative")
        return value
 
    @property
    def is_relayed(self) -> bool:
        """True once this alert reached the transport."""
        return self.relayed_at_utc is not None
 
 
def render_event(event: SessionAlertEvent) -> str:
    """Return the one line an operator reads for this alert, fenced."""
    state = "relayed" if event.is_relayed else "unrelayed"
    threshold = "-" if event.threshold_usd is None else f"${event.threshold_usd:.2f}"
    header = (
        f"[{state}] {event.event_id}  {event.created_at_utc}  "
        f"lane={event.lane} kind={event.kind} threshold={threshold} "
        f"session={event.session_id}"
    )
    body = event.summary.replace("<<<", NEUTRALISED_MARKER)
    begin = SESSION_ALERT_BEGIN.format(event_id=event.event_id)
    end = SESSION_ALERT_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 SessionAlertStore:
    """Read-and-mark access to one ``lanes/session-alerts/`` 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) -> SessionAlertEvent:
        try:
            return SessionAlertEvent.model_validate_json(
                path.read_text(encoding="utf-8")
            )
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise SessionAlertError(
                f"could not parse session alert event at {path} "
                f"({type(exc).__name__})"
            ) from None
 
    def list_events(self, *, relayed: bool | None = None) -> tuple[SessionAlertEvent, ...]:
        """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 SessionAlertError(f"session alert event {event_id!r} is not in the lane")
 
    def get(self, event_id: str) -> SessionAlertEvent:
        """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) -> SessionAlertEvent:
        """Record that one event's summary reached the transport."""
        if chunk_count < 1:
            raise ValueError("a relayed event must have sent at least one message")
        path = self.path_for(event_id)
        event = self._read(path)
        if event.is_relayed:
            raise SessionAlertError(
                f"session alert 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.SESSION_ALERTS",
            "mark_relayed",
            f"event={event_id} kind={event.kind}",
            2,
        )
        return updated
 
 
@dataclass(frozen=True)
class SessionAlertOutcome:
    """What the relay did with one alert, in terms a CLI can print."""
 
    event_id: str
    kind: str
    total_usd: float
    audit_id: str
 
 
class SessionAlertRelay:
    """Forward unrelayed session alerts through the standard send path.
 
    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: SessionAlertStore,
        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: SessionAlertEvent,
        outcome: str,
        *,
        attempts: int,
        error_code: str | None = None,
    ) -> None:
        descriptor = lane(SESSION_ALERT_LANE)
        from .audit import content_hash
 
        self.audit.append(
            connector="telegram",
            grant_id=self.grant_id,
            destination=self.destination,
            outcome=outcome,  # type: ignore[arg-type]
            digest=content_hash(event.summary),
            idempotency_key=event.event_id,
            attempts=attempts,
            error_code=error_code,
            lane=SESSION_ALERT_LANE,
            direction=descriptor.direction,
            risk_class=descriptor.risk_ceiling,
        )
 
    def relay_one(self, event: SessionAlertEvent) -> SessionAlertOutcome:
        """Relay one alert, or leave it unrelayed and raise."""
        func = "relay_one"
        try:
            require_safe(event.summary)
        except EgressBlockedError:
            self._audit(event, "blocked", attempts=0, error_code="egress_finding")
            _dbg(
                "OC_CONNECT.SESSION_ALERTS",
                func,
                f"SECURITY_ERROR: blocked session alert event={event.event_id}",
                1,
                override=True,
            )
            raise
        receipt = self.sender.send(
            grantee=self.grantee,
            destination=self.destination,
            text=event.summary,
            grant_id=self.grant_id,
            risk_class=self.risk_class,
            idempotency_key=event.event_id,
        )
        if receipt.outcome != "sent":
            self._audit(event, "failed", attempts=1, error_code=receipt.outcome)
            raise SessionAlertRelayError(
                f"session alert {event.event_id!r} was not delivered "
                f"(outcome {receipt.outcome!r}); the event 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.SESSION_ALERTS",
            func,
            f"relayed event={event.event_id} kind={event.kind}",
            2,
        )
        return SessionAlertOutcome(
            event_id=event.event_id,
            kind=event.kind,
            total_usd=event.total_usd,
            audit_id=receipt.audit_id,
        )
 
    def relay_pending(self) -> tuple[SessionAlertOutcome, ...]:
        """Relay every unrelayed alert, oldest first, stopping on the first refusal."""
        return tuple(
            self.relay_one(event)
            for event in self.store.list_events(relayed=False)
        )
 
 
__all__ = [
    "NEUTRALISED_MARKER",
    "SESSION_ALERT_BEGIN",
    "SESSION_ALERT_END",
    "SESSION_ALERT_LANE",
    "SESSION_ALERT_SCHEMA_VERSION",
    "SESSION_ALERT_SOURCE",
    "SessionAlertError",
    "SessionAlertEvent",
    "SessionAlertKind",
    "SessionAlertOutcome",
    "SessionAlertRelay",
    "SessionAlertRelayError",
    "SessionAlertStore",
    "render_event",
]