session_checkpoints.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/session_checkpoints.py
 
"""
Soft caps: the checkpoints that pause a long session instead of ending it.
 
Purpose
-------
CAPS-01 (plan 237). A per-turn tool-call count is a poor runaway detector — it
fired constantly during real work and made the long, semi-autonomous sessions
this product exists for impossible. The operator's ruling: the COUNT becomes a
checkpoint with a higher default, and the bound that actually matters becomes
MONEY. This module holds both checkpoints and nothing else:
 
- :func:`tool_call_checkpoint` — at the per-turn tool-call cap, on a real TTY,
  show a mini report and ask ``Continue this turn? (y/N/a)``. ``y`` buys one
  more cap-width; ``a`` stops asking for the rest of the session; ``N``, Enter
  or an interrupt fall through to the EXISTING per-call denial, byte for byte.
- :func:`evaluate_cost_checkpoints` — after each priced sample, fire the hard
  cap first and then any newly crossed dollar checkpoint, each of which writes
  a ``session-alerts`` lane event, notifies the operator best-effort, and (on a
  TTY) asks ``Continue this session? (y/N/a)``.
 
Headless is deliberately unchanged. With no TTY there is no prompt, the
tool-call cap stays the hard deny it has always been (fail-closed, F38/F52),
and the cost path alerts and logs but never blocks — the session runs to the
hard cap, which is the one bound that stops it either way.
 
The two "don't ask again" latches are PROCESS-scoped, exactly like the F67
external-content trust set and the plan-235 Tier-2 grant, and drop at the same
two session seams (:meth:`ChatSession.clear` and
``session._reset_trust_before_exposure``). They are deliberately NOT persisted:
a suppression that survived a reload would be a standing authority nobody
granted twice.
 
The two latches are also deliberately INDEPENDENT. ``a`` at the tool-call
checkpoint silences the tool-call prompt alone — never a cost prompt, never the
alerts, and never the hard cap. That separation is the ruling's exact shape.
 
Responsibilities
----------------
- :class:`MiniReport` — the five facts an operator needs to answer the prompt,
  and their rendering.
- :func:`tool_call_checkpoint` / :func:`evaluate_cost_checkpoints` — the two
  decisions, returning a verdict rather than acting on the loop themselves.
- :func:`reset_session_checkpoints` and the latch accessors.
 
Contracts
---------
- No prompt is ever shown when *is_interactive* is false. The caller passes the
  answer from the ONE TTY seam (``react_loop._stdin_is_interactive``) so this
  module has no hidden global to disagree with it.
- A verdict never mutates the loop. ``tool_call_checkpoint`` returns whether to
  extend the budget; ``evaluate_cost_checkpoints`` returns whether to stop. The
  react-loop seam owns every side effect on ``session.messages``.
- An interrupt at either prompt is the CONSERVATIVE answer (deny / stop), never
  a silent continue.
- A failure to write or relay an alert never changes a verdict.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.session_alerts (the credential-free lane writer)
- oc_chat.core.defaults (lazy; shipped top-tools count for the mini report)
- oc_chat.core.formatting (lazy; format_duration for the mini report lines)
- oc_ui.terminal.input (lazy; read_single_key for the checkpoint prompt)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L1 — a checkpoint stopped a turn or a session (override: operator-facing)
    L2 — a checkpoint fired; a latch was set
    L3 — per-decision detail
    L4 — module load
"""
from __future__ import annotations
 
from dataclasses import dataclass
from typing import Any
 
from oc_diagnostics import _dbg
 
from oc_chat.core import session_alerts
from oc_chat.core.session_alerts import (
    KIND_COST_HARD_CAP, KIND_COST_THRESHOLD, KIND_STALL_HALT,
    KIND_TOOL_CALL_CHECKPOINT,
)
 
_dbg("OC_CHAT.CORE", "module", "loaded", 4)
 
 
def _emit(emit: Any, **fields: Any) -> None:
    """Write one alert through *emit*, or through the live writer when ``None``.
 
    Resolved HERE rather than as a default argument on purpose. A default is
    bound once at function definition, so ``monkeypatch.setattr(module,
    "emit_session_alert", ...)`` would silently fail to take and a test would
    write into the real comms bus while believing it had faked it — which is
    exactly what happened while this plan was being written.
    """
    (emit or session_alerts.emit_session_alert)(**fields)
 
 
#: The two process-scoped "don't ask again this session" latches. Module state,
#: not session state, for the same reason the F67 trust set is: they must not
#: survive a save/load, and every seam that drops session trust drops these.
_TOOL_CALLS_AUTO_CONTINUE: bool = False
_COST_PROMPTS_SUPPRESSED: bool = False
_STALL_PROMPTS_SUPPRESSED: bool = False
 
 
def reset_session_checkpoints() -> None:
    """Drop both "don't ask again" latches (the session-change seam)."""
    func = "reset_session_checkpoints"
    global _TOOL_CALLS_AUTO_CONTINUE, _COST_PROMPTS_SUPPRESSED
    global _STALL_PROMPTS_SUPPRESSED
    _TOOL_CALLS_AUTO_CONTINUE = False
    _COST_PROMPTS_SUPPRESSED = False
    _STALL_PROMPTS_SUPPRESSED = False
    _dbg("OC_CHAT.CORE", func, "checkpoint latches cleared", 3)
 
 
def tool_calls_auto_continue() -> bool:
    """True when the operator answered ``a`` at the tool-call checkpoint."""
    return _TOOL_CALLS_AUTO_CONTINUE
 
 
def cost_prompts_suppressed() -> bool:
    """True when the operator answered ``a`` at a cost checkpoint.
 
    Suppresses further cost PROMPTS only. Alerts still fire and the hard cap
    still stops the session — that is the point of a hard cap."""
    return _COST_PROMPTS_SUPPRESSED
 
 
def stall_prompts_suppressed() -> bool:
    """True when the operator answered ``a`` at a stall checkpoint."""
    return _STALL_PROMPTS_SUPPRESSED
 
 
def _set_tool_calls_auto_continue() -> None:
    global _TOOL_CALLS_AUTO_CONTINUE
    _TOOL_CALLS_AUTO_CONTINUE = True
 
 
def _set_cost_prompts_suppressed() -> None:
    global _COST_PROMPTS_SUPPRESSED
    _COST_PROMPTS_SUPPRESSED = True
 
 
def _set_stall_prompts_suppressed() -> None:
    global _STALL_PROMPTS_SUPPRESSED
    _STALL_PROMPTS_SUPPRESSED = True
 
 
@dataclass(frozen=True)
class MiniReport:
    """What the turn has spent so far, in the five terms the ruling names."""
 
    turn_tool_calls: int
    """Tool calls made so far this turn."""
    top_tools: tuple
    """The most-called tools this turn as ``(name, count)``, descending."""
    files_written: tuple
    """Paths successfully written this turn, in first-write order."""
    priced_usd: float
    """Session spend that could be priced, in USD."""
    unpriced_samples: int
    """Samples that could NOT be priced — counted, never treated as zero."""
    elapsed_ms: int
    """Milliseconds elapsed in this turn."""
 
    def lines(self) -> list:
        """Render the report as operator-facing lines (no ANSI, no markup)."""
        from oc_chat.core.formatting import format_duration
 
        tools = ", ".join(f"{name} x{count}" for name, count in self.top_tools) or "none"
        if self.files_written:
            shown = ", ".join(self.files_written[:5])
            more = len(self.files_written) - 5
            files = shown + (f" (+{more} more)" if more > 0 else "")
        else:
            files = "none"
        unpriced = (f"  (+{self.unpriced_samples} unpriced sample"
                    f"{'s' if self.unpriced_samples != 1 else ''})"
                    if self.unpriced_samples else "")
        return [
            f"  tool calls this turn : {self.turn_tool_calls}",
            f"  top tools            : {tools}",
            f"  files written        : {files}",
            f"  session cost         : ${self.priced_usd:.2f}{unpriced}",
            f"  elapsed this turn    : {format_duration(self.elapsed_ms)}",
        ]
 
 
def build_mini_report(
    *,
    turn_tool_calls: int,
    tool_name_counts: dict,
    files_written,
    priced_usd: float,
    unpriced_samples: int,
    elapsed_ms: int,
    top_n: int | None = None,
) -> MiniReport:
    """Assemble a :class:`MiniReport`; *top_n* defaults to the shipped value."""
    if top_n is None:
        from oc_chat.core.defaults import default
        top_n = int(default("checkpoint_top_tools"))
    ranked = sorted(tool_name_counts.items(), key=lambda kv: (-kv[1], kv[0]))
    return MiniReport(
        turn_tool_calls=turn_tool_calls,
        top_tools=tuple(ranked[:max(0, top_n)]),
        files_written=tuple(files_written),
        priced_usd=float(priced_usd),
        unpriced_samples=int(unpriced_samples),
        elapsed_ms=int(elapsed_ms),
    )
 
 
def _ask(presenter, header: str, report: MiniReport, prompt: str) -> str:
    """Show the mini report and read one key; ``""``/interrupt answer ``n``."""
    func = "_ask"
    presenter.show_info("\n".join([header, *report.lines()]))
    from oc_ui.terminal.input import read_single_key
    try:
        answer = read_single_key(prompt, valid="yna")
    except (EOFError, KeyboardInterrupt):
        # Conservative: an interrupt at a checkpoint is not consent.
        _dbg("OC_CHAT.CORE", func, "interrupt at checkpoint -> treated as 'n'", 2)
        return "n"
    return answer or "n"
 
 
def tool_call_checkpoint(
    *,
    presenter,
    report: MiniReport,
    effective_cap: int,
    is_interactive: bool,
    session_id: str = "",
    emit: Any = None,
) -> bool:
    """Decide whether to buy one more cap-width of tool calls this turn.
 
    Returns ``True`` to extend the budget, ``False`` to fall through to the
    existing per-call denial. Headless always returns ``False`` — the cap is a
    hard deny with no human present, exactly as before this plan.
    """
    func = "tool_call_checkpoint"
    if not is_interactive:
        _dbg("OC_CHAT.CORE", func,
             f"no TTY at the {effective_cap}-call checkpoint -> hard deny (unchanged)", 2)
        return False
    if tool_calls_auto_continue():
        _dbg("OC_CHAT.CORE", func,
             f"tool-call checkpoint at {effective_cap} auto-continued ('a' this session)", 2)
        return True
    # The operator is about to be BLOCKED at a prompt: tell them wherever they
    # are, before the terminal starts waiting. Auto-continued crossings emit
    # nothing — they answered "stop asking", and an alert is asking.
    _emit(
        emit, session_id=session_id, kind=KIND_TOOL_CALL_CHECKPOINT,
        total_usd=report.priced_usd, unpriced_samples=report.unpriced_samples,
        turn_tool_calls=report.turn_tool_calls,
    )
    answer = _ask(
        presenter,
        f"[checkpoint] {report.turn_tool_calls} tool calls this turn reaches the "
        f"per-turn budget of {effective_cap}.",
        report,
        "Continue this turn? (y/N/a) ",
    )
    if answer == "a":
        _set_tool_calls_auto_continue()
        _dbg("OC_CHAT.CORE", func,
             "tool-call checkpoint set to auto-continue for this session", 2)
        return True
    if answer == "y":
        _dbg("OC_CHAT.CORE", func,
             f"tool-call budget extended past {effective_cap} by the operator", 2)
        return True
    _dbg("OC_CHAT.CORE", func,
         f"operator declined at the {effective_cap}-call checkpoint -> deny as before",
         1, override=True)
    return False
 
 
def stall_checkpoint(
    *,
    presenter,
    report: MiniReport,
    sentence: str,
    is_interactive: bool,
    session_id: str = "",
    emit: Any = None,
) -> bool:
    """Alert on a stall and decide whether the turn may continue.
 
    The offending proposal remains denied regardless of the answer. ``y`` and
    ``a`` let the next ReAct iteration run; ``a`` suppresses only later stall
    prompts in this process-scoped session. Headless and ``n`` stop the turn.
    """
    func = "stall_checkpoint"
    _emit(
        emit,
        session_id=session_id,
        kind=KIND_STALL_HALT,
        total_usd=report.priced_usd,
        unpriced_samples=report.unpriced_samples,
        turn_tool_calls=report.turn_tool_calls,
        stall_reason=sentence,
    )
    if stall_prompts_suppressed():
        _dbg("OC_CHAT.CORE", func,
             "stall checkpoint auto-continued ('a' this session)", 2)
        return True
    if not is_interactive:
        _dbg("OC_CHAT.CORE", func, "stall halted a headless turn", 1, override=True)
        return False
    answer = _ask(
        presenter,
        sentence,
        report,
        "Continue this turn? (y/N/a) ",
    )
    if answer == "a":
        _set_stall_prompts_suppressed()
        _dbg("OC_CHAT.CORE", func,
             "stall prompts suppressed for this session", 2)
        return True
    if answer == "y":
        _dbg("OC_CHAT.CORE", func, "stall checkpoint continued by operator", 2)
        return True
    _dbg("OC_CHAT.CORE", func, "operator ended the turn at a stall checkpoint",
         1, override=True)
    return False
 
 
@dataclass(frozen=True)
class CostVerdict:
    """Whether the session may continue, and the operator-facing reason if not."""
 
    stop: bool
    reason: str = ""
 
 
#: Returned when nothing crossed — the overwhelmingly common case.
_CONTINUE = CostVerdict(stop=False)
 
 
def hard_cap_refusal(
    *, priced_usd: float, unpriced_samples: int, cap_usd: int, env_var_name: str,
) -> str:
    """The refusal text: the total, the cap, and the exact knob that raises it.
 
    Named rather than vague on purpose (P4). An operator who hits this must not
    have to read the source to find out how to continue, and must not be left
    believing the total is the whole story when samples went unpriced.
    """
    unpriced = ""
    if unpriced_samples:
        plural = unpriced_samples != 1
        unpriced = (
            f" {unpriced_samples} sample{'s' if plural else ''} could not be "
            f"priced and {'are' if plural else 'is'} NOT included in that "
            f"total, so real spend may be higher."
        )
    raise_it = (
        f" Raising it is out-of-band only: set {env_var_name}=<whole dollars> in "
        f"the launch environment (a workspace settings file can only LOWER it)."
        if env_var_name else
        " This cap has no out-of-band raise path."
    )
    return (
        f"[cost] Session spend ${priced_usd:.2f} has reached the ${cap_usd} hard "
        f"cap; no further provider call will be made in this session.{unpriced}"
        f"{raise_it}"
    )
 
 
def evaluate_cost_checkpoints(
    *,
    session_id: str,
    cost: Any,
    presenter,
    report: MiniReport,
    is_interactive: bool,
    hard_cap_usd: int,
    thresholds: tuple,
    env_var_name: str,
    emit: Any = None,
) -> CostVerdict:
    """Fire the hard cap and any newly crossed checkpoint; return the verdict.
 
    *cost* is the session's :class:`~oc_chat.core.session.SessionCost` fold —
    duck-typed rather than imported, because ``session.py`` imports THIS module
    at its reset seams.
 
    Order is deliberate: the hard cap is evaluated first and returns
    immediately, so a sample that vaults several checkpoints at once ends the
    session on the cap rather than after a queue of prompts nobody will answer.
    """
    func = "evaluate_cost_checkpoints"
    priced = float(getattr(cost, "priced_usd", 0.0) or 0.0)
    unpriced = int(getattr(cost, "unpriced_samples", 0) or 0)
 
    if hard_cap_usd > 0 and priced >= float(hard_cap_usd):
        reason = hard_cap_refusal(
            priced_usd=priced, unpriced_samples=unpriced,
            cap_usd=hard_cap_usd, env_var_name=env_var_name,
        )
        _emit(
            emit, session_id=session_id, kind=KIND_COST_HARD_CAP, total_usd=priced,
            unpriced_samples=unpriced, threshold_usd=float(hard_cap_usd),
            turn_tool_calls=report.turn_tool_calls,
        )
        _dbg("OC_CHAT.CORE", func,
             f"SECURITY_ERROR: session hard cap reached (${priced:.2f} >= "
             f"${hard_cap_usd}); refusing further provider calls", 1, override=True)
        return CostVerdict(stop=True, reason=reason)
 
    crossed = [
        t for t in sorted(thresholds)
        if priced >= float(t) and not cost.has_fired(t)
    ]
    if not crossed:
        return _CONTINUE
 
    # One alert per checkpoint passed — each is a distinct fact the record
    # should carry. ONE notification, for the highest, because three phone
    # messages for one sample is noise, and three 15-second subprocesses on the
    # turn's critical path is worse.
    highest = crossed[-1]
    for threshold in crossed:
        cost.mark_threshold_fired(threshold)
        _emit(
            emit, session_id=session_id, kind=KIND_COST_THRESHOLD, total_usd=priced,
            unpriced_samples=unpriced, threshold_usd=float(threshold),
            turn_tool_calls=report.turn_tool_calls,
            notify=(threshold == highest),
        )
    _dbg("OC_CHAT.CORE", func,
         f"cost checkpoint(s) crossed: {crossed} at ${priced:.2f}", 2)
 
    if not is_interactive or cost_prompts_suppressed():
        # Headless, or "don't ask again": alerted and logged, never blocked.
        return _CONTINUE
 
    answer = _ask(
        presenter,
        f"[checkpoint] Session spend has passed ${highest:.2f}.",
        report,
        "Continue this session? (y/N/a) ",
    )
    if answer == "a":
        _set_cost_prompts_suppressed()
        _dbg("OC_CHAT.CORE", func,
             "cost prompts suppressed for this session (alerts and hard cap stand)", 2)
        return _CONTINUE
    if answer == "y":
        return _CONTINUE
    _dbg("OC_CHAT.CORE", func,
         f"operator ended the turn at the ${highest:.2f} checkpoint", 1, override=True)
    return CostVerdict(
        stop=True,
        reason=(
            f"[cost] You ended this turn at the ${highest:.2f} checkpoint; session "
            f"spend is ${priced:.2f}. Nothing further was sent to the provider."
        ),
    )
 
 
__all__ = [
    "CostVerdict",
    "MiniReport",
    "build_mini_report",
    "cost_prompts_suppressed",
    "evaluate_cost_checkpoints",
    "hard_cap_refusal",
    "reset_session_checkpoints",
    "stall_checkpoint",
    "stall_prompts_suppressed",
    "tool_call_checkpoint",
    "tool_calls_auto_continue",
]