session_alerts.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/session_alerts.py
 
"""
The ``session-alerts`` lane writer — money events the operator reads on a phone.
 
Purpose
-------
CAPS-01 (plan 237 P5/P6). When a session crosses a cost checkpoint, hits the
dollar hard cap, or pauses at the per-turn tool-call checkpoint, the operator
should learn about it wherever they are — without the product ever holding a
Telegram credential. This module is the product-side half of that, and it is
deliberately the SAME shape plan 231 gave the worker-report lane:
 
- it writes ONE json file per event into ``<comms-root>/lanes/session-alerts/``
  and stops there;
- it imports no connector and no network package (asserted structurally by the
  W7 witness, against a negative control, exactly as the 231 tests do);
- ``python -m oc_connect relay session-alerts`` — a separate process, run by
  the operator, holding the grant — is what puts the text on the transport.
 
The one addition over 231 is :func:`notify_operator`: a best-effort
``python -m oc_connect send`` SUBPROCESS, run only when ``oc_connect`` is
importable in this environment, under the scrubbed environment every other
child process gets. It is a convenience, never a guarantee: any failure is
logged and swallowed, and the lane file on disk remains the durable record the
relay will pick up. The product still holds no token — the send succeeds only
where the operator's own sidecar grant already exists.
 
Responsibilities
----------------
- :func:`new_ulid` — the 26-character, time-first event id (the lane's ordering
  is a plain directory sort).
- :func:`find_comms_root` — the workspace-anchored bus root, by the same walk
  ``scripts/plan_status.py`` uses.
- :func:`session_alert_event` — build one validated event dict.
- :func:`write_session_alert` — write it, atomically; return ``None`` rather
  than raising when the bus is unreachable.
- :func:`notify_operator` — the best-effort direct send described above.
- :func:`emit_session_alert` — write + notify in one fail-soft call, which is
  what the react-loop seam uses.
 
Contracts
---------
- **Fail-SOFT everywhere.** An unreachable, unwritable or nonexistent comms root
  logs at L2 and returns ``None``. A money alert that cannot be delivered must
  never be the thing that kills the session it is reporting on.
- The ``summary`` carries NUMBERS ONLY — totals, the threshold, counts. No
  model content, no file paths, no prompt text ever reaches the transport from
  here (§8: displayed content is never instructions, and an alert is not a
  transcript).
- The event's ``lane``/``direction``/``risk_class`` mirror the
  ``session-alerts`` descriptor declared in ``option_c_connector_contracts``.
  The mirror is intentional and is what buys the credential-free write path;
  the drift risk is paid for by the W7 cross-implementation assertions.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.defaults (the shipped notification timeout)
- oc_chat.tools._sandbox (lazy; scrubbed_env for the notification subprocess)
- oc_connect (not imported; run as python -m oc_connect send for the direct notification)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L1 — an alert could not be written (override: the operator must see it)
    L2 — an alert was written; a direct notification was attempted
    L3 — per-field detail
    L4 — module load
"""
from __future__ import annotations
 
import json
import os
import secrets
import string
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
 
from oc_diagnostics import _dbg
 
from oc_chat.core.defaults import default
 
_dbg("OC_CHAT.CORE", "module", "loaded", 4)
 
 
#: 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.
SESSION_ALERT_LANE = "session-alerts"
SESSION_ALERT_SOURCE = "product"
SESSION_ALERT_SCHEMA_VERSION = "1.0"
SESSION_ALERT_DIRECTION = "inbound"
SESSION_ALERT_RISK_CLASS = "low"
 
#: The four things worth waking an operator for.
KIND_COST_THRESHOLD = "cost_threshold"
KIND_COST_HARD_CAP = "cost_hard_cap"
KIND_TOOL_CALL_CHECKPOINT = "tool_call_checkpoint"
KIND_STALL_HALT = "stall_halt"
SESSION_ALERT_KINDS = (
    KIND_COST_THRESHOLD, KIND_COST_HARD_CAP, KIND_TOOL_CALL_CHECKPOINT,
    KIND_STALL_HALT,
)
 
#: Workspace walk constants, mirroring ``scripts/plan_status.py``.
WORKSPACE_MARKER = ".option_c_workspace.json"
PROJECT_MARKER = ".option_c"
COMMS_ANCHOR = "oc_chat"
COMMS_SUBPATH = (".option_c", "comms")
 
#: Seconds a best-effort direct notification may take before it is abandoned.
#: The VALUE lives in the shipped, read-only defaults file (CONFIG-01, plan
#: 238); this name is an import-time alias, as in ``core/resource_caps``.
NOTIFY_TIMEOUT_S = default("session_alert_notify_timeout_s")
 
#: Crockford base32, the ULID alphabet: no I, L, O or U. DERIVED from the
#: stdlib vocabularies rather than written as a 32-character literal — the
#: literal scores the arithmetic maximum 5.00 bits/char and needs a scanner
#: waiver everywhere it appears (``plan_status``, ``oc_connect.comms``). The
#: derivation is provably the same alphabet, and the W7 witness compares the
#: two byte for byte so this can never silently drift from the lane's readers.
_CROCKFORD = "".join(
    ch for ch in string.digits + string.ascii_uppercase if ch not in "ILOU"
)
 
 
class SessionAlertError(ValueError):
    """A session-alert event was asked for in a shape the lane does not carry."""
 
 
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.
 
    The same algorithm ``plan_status.new_ulid`` and ``oc_connect.comms.new_ulid``
    implement. Time-first encoding makes lexical order arrival order, which is
    what lets a reader walk the lane with a plain directory sort.
    """
    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 _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 find_comms_root(start: Path) -> Path | None:
    """Return the workspace-anchored comms-bus root above *start*, or ``None``.
 
    The same walk ``plan_status.find_comms_root`` performs — the workspace
    marker wins over a nearer project marker, because connector state is
    workspace-wide. It differs in ONE way, deliberately: a failed walk returns
    ``None`` instead of raising. A worker reporting at a gate should fail loudly
    if nobody is listening; a live chat session must not die because it is
    running outside a workspace.
    """
    func = "find_comms_root"
    try:
        origin = Path(start).expanduser().resolve()
    except OSError as exc:  # EXTERNAL_ERROR (L2) — an unresolvable cwd
        _dbg("OC_CHAT.CORE", func, f"EXTERNAL_ERROR: {type(exc).__name__}: {exc}", 2)
        return None
    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.CORE", func,
             f"no workspace ({WORKSPACE_MARKER}) or project ({PROJECT_MARKER}) "
             f"marker at or above {origin}; session alerts stay local", 2)
        return None
    if (anchor / COMMS_ANCHOR).is_dir():
        return anchor.joinpath(COMMS_ANCHOR, *COMMS_SUBPATH)
    return anchor.joinpath(*COMMS_SUBPATH)
 
 
def format_summary(
    *,
    kind: str,
    total_usd: float,
    unpriced_samples: int,
    threshold_usd: float | None,
    turn_tool_calls: int,
    stall_reason: str | None = None,
) -> str:
    """Return the ONE line the operator reads, verbatim, on their phone.
 
    Numbers only — see the module Contracts. The unpriced count travels with
    every total because a total without it is a claim the pricing table cannot
    support: it is a lower bound, and the reader is entitled to know by how
    many samples it may be short.
    """
    if kind == KIND_STALL_HALT:
        return stall_reason or "OC Chat stopped: a tool-call stall halted the turn."
    unpriced = f" (+{unpriced_samples} unpriced samples)" if unpriced_samples else ""
    if kind == KIND_COST_HARD_CAP:
        cap = f"{threshold_usd:.2f}" if threshold_usd is not None else "?"
        return (
            f"OC Chat stopped: session spend ${total_usd:.2f}{unpriced} reached "
            f"the ${cap} hard cap. No further provider call will be made in this "
            f"session."
        )
    if kind == KIND_COST_THRESHOLD:
        crossed = f"{threshold_usd:.2f}" if threshold_usd is not None else "?"
        return (
            f"OC Chat cost checkpoint ${crossed}: session spend is "
            f"${total_usd:.2f}{unpriced}."
        )
    return (
        f"OC Chat tool-call checkpoint: {turn_tool_calls} tool calls this turn; "
        f"session spend ${total_usd:.2f}{unpriced}."
    )
 
 
def session_alert_event(
    *,
    session_id: str,
    kind: str,
    total_usd: float,
    unpriced_samples: int,
    now: datetime,
    threshold_usd: float | None = None,
    turn_tool_calls: int = 0,
    stall_reason: str | None = None,
    entropy: bytes | None = None,
) -> dict:
    """Build one ``session-alerts`` event; refuse an undeclared *kind* by name."""
    if kind not in SESSION_ALERT_KINDS:
        raise SessionAlertError(
            f"session alert kind must be one of "
            f"{', '.join(SESSION_ALERT_KINDS)}; got {kind!r}"
        )
    return {
        "schema_version": SESSION_ALERT_SCHEMA_VERSION,
        "event_id": new_ulid(now, entropy=entropy),
        "lane": SESSION_ALERT_LANE,
        "direction": SESSION_ALERT_DIRECTION,
        "risk_class": SESSION_ALERT_RISK_CLASS,
        "created_at_utc": _utc_timestamp(now),
        "session_id": str(session_id),
        "kind": kind,
        "threshold_usd": (None if threshold_usd is None else float(threshold_usd)),
        "total_usd": float(total_usd),
        "unpriced_samples": int(unpriced_samples),
        "turn_tool_calls": int(turn_tool_calls),
        "summary": format_summary(
            kind=kind, total_usd=total_usd, unpriced_samples=unpriced_samples,
            threshold_usd=threshold_usd, turn_tool_calls=turn_tool_calls,
            stall_reason=stall_reason,
        ),
        "relayed_at_utc": None,
        "relay_chunk_count": None,
    }
 
 
def write_session_alert(comms_root: Path, event: dict, *, now: datetime) -> Path:
    """Write one event into ``lanes/session-alerts/`` and return its path.
 
    The lane directory is created when absent: a session must be able to alert
    from a workspace whose bus has never been initialised. Nothing outside the
    lane is created, read, or written.
    """
    func = "write_session_alert"
    lane_dir = Path(comms_root) / "lanes" / SESSION_ALERT_LANE
    lane_dir.mkdir(parents=True, exist_ok=True)
    name = f"{utc_stamp(now)}_{SESSION_ALERT_SOURCE}_{event['event_id']}.json"
    path = lane_dir / name
    if path.exists():
        raise SessionAlertError(
            f"refusing to overwrite an existing session alert 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=".session_alert_", suffix=".tmp", delete=False,
    ) as tmp:
        tmp.write(payload)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
    _dbg("OC_CHAT.CORE", func,
         f"wrote {event['kind']} alert {event['event_id']} to {path.name}", 2)
    return path
 
 
def _oc_connect_available() -> bool:
    """True when ``oc_connect`` is importable in THIS environment.
 
    ``find_spec`` deliberately, not an import: the product must never load the
    connector package into its own process — that is what keeps the credential
    out of this address space. This only answers "is there a sidecar CLI to
    hand a summary to?".
    """
    func = "_oc_connect_available"
    try:
        from importlib.util import find_spec
        return find_spec("oc_connect") is not None
    except (ImportError, ValueError) as exc:
        _dbg("OC_CHAT.CORE", func, f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
        return False
 
 
def notify_operator(summary: str, *, destination: str = "operator") -> bool:
    """Best-effort direct notification of one alert *summary*. Never raises.
 
    Runs ``python -m oc_connect send <destination> "<summary>"`` as a
    subprocess under :func:`~oc_chat.tools._sandbox.scrubbed_env`, with a short
    timeout. Returns whether the child reported success. Every failure — no
    ``oc_connect``, no grant, no network, a timeout, a non-zero exit — is logged
    at L2 and swallowed: the lane file is the durable record, and the relay is
    the delivery guarantee. *summary* is the event's own fixed one-line text, so
    nothing but numbers can reach the transport through this path.
    """
    func = "notify_operator"
    if not _oc_connect_available():
        _dbg("OC_CHAT.CORE", func, "oc_connect not importable; skipping direct send", 3)
        return False
    try:
        import sys
        from oc_chat.tools._sandbox import scrubbed_env
        completed = subprocess.run(  # noqa: S603 — fixed argv, no shell, no model input
            [sys.executable, "-m", "oc_connect", "send", destination, summary],
            capture_output=True, text=True, timeout=NOTIFY_TIMEOUT_S,
            env=scrubbed_env(), shell=False, check=False,
        )
    except Exception as exc:  # noqa: BLE001 — EXTERNAL_ERROR (L2); best-effort by contract
        _dbg("OC_CHAT.CORE", func,
             f"EXTERNAL_ERROR: direct notification failed "
             f"({type(exc).__name__}: {exc}); the lane file stands", 2)
        return False
    if completed.returncode != 0:
        _dbg("OC_CHAT.CORE", func,
             f"EXTERNAL_ERROR: oc_connect send exited {completed.returncode}; "
             f"the lane file stands", 2)
        return False
    _dbg("OC_CHAT.CORE", func, "direct notification sent", 2)
    return True
 
 
def emit_session_alert(
    *,
    session_id: str,
    kind: str,
    total_usd: float,
    unpriced_samples: int,
    threshold_usd: float | None = None,
    turn_tool_calls: int = 0,
    stall_reason: str | None = None,
    comms_root: Path | None = None,
    now: datetime | None = None,
    notify: bool = True,
) -> Path | None:
    """Write one alert and (best-effort) notify the operator. Never raises.
 
    The one call the react-loop seam makes. Returns the written path, or
    ``None`` when the bus was unreachable — in which case the caller carries on
    regardless, because a session must not die of a failed alert.
    """
    func = "emit_session_alert"
    stamp = now or datetime.now(timezone.utc)
    try:
        root = comms_root if comms_root is not None else find_comms_root(Path(os.getcwd()))
        if root is None:
            return None
        event = session_alert_event(
            session_id=session_id, kind=kind, total_usd=total_usd,
            unpriced_samples=unpriced_samples, threshold_usd=threshold_usd,
            turn_tool_calls=turn_tool_calls, stall_reason=stall_reason, now=stamp,
        )
        path = write_session_alert(root, event, now=stamp)
    except Exception as exc:  # noqa: BLE001 — EXTERNAL_ERROR (L2); fail-soft by contract
        _dbg("OC_CHAT.CORE", func,
             f"EXTERNAL_ERROR: could not write a {kind} alert "
             f"({type(exc).__name__}: {exc}); the session continues", 2)
        return None
    if notify:
        notify_operator(event["summary"])
    return path
 
 
__all__ = [
    "KIND_COST_HARD_CAP",
    "KIND_COST_THRESHOLD",
    "KIND_STALL_HALT",
    "KIND_TOOL_CALL_CHECKPOINT",
    "NOTIFY_TIMEOUT_S",
    "SESSION_ALERT_DIRECTION",
    "SESSION_ALERT_KINDS",
    "SESSION_ALERT_LANE",
    "SESSION_ALERT_RISK_CLASS",
    "SESSION_ALERT_SCHEMA_VERSION",
    "SESSION_ALERT_SOURCE",
    "SessionAlertError",
    "emit_session_alert",
    "find_comms_root",
    "format_summary",
    "new_ulid",
    "notify_operator",
    "session_alert_event",
    "utc_stamp",
    "write_session_alert",
]