resource_caps.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/resource_caps.py
 
# OCT-LINT: disable=no_hardcoded_secrets reason="Module docstring prose trips Shannon entropy (5.05 bits/char, pre-existing at BASE); no secret (operator ruling 2026-09-05)"
"""
Normative resource caps — hard ceilings model args cannot exceed (M8.0 WU-9).
 
Purpose
-------
Appendix B of the locked V1.0 spec (LD-3) fixes hard ceilings that bound the
transcript-flood / provider-egress-cost / runaway-call DoS surface (F38/F52/F19;
every turn re-ships the whole transcript, F72). A *model-supplied* argument can
never exceed them: ``timeout``, ``max_bytes``/``max_chars``, ``max_results`` are
clamped at the tool boundary; the per-turn tool-call count and the parallel-worker
pool are bounded in the react-loop. The values are unified at **5 MB** for the
read / fetch-body / output ceilings (LD-3). The per-turn tool-call ceiling is an
**absolute ceiling a settings write may LOWER but never RAISE** (amended from the
V1.0 Appendix-B 10 on operator approval). Plan 238 (CONFIG-01) moved every value
below into the shipped, read-only ``oc_chat.defaults.toml``, hard-denied for
writes by name in ``tools/_sandbox.py``: the names here are import-time aliases
bound from :mod:`oc_chat.core.defaults`, and no value is written twice. The workspace
``max_tool_calls_per_turn`` setting is a model-reachable, human-confirm-tier write,
so F33/§5.4 bar it from raising a security ceiling; raising the ceiling above the
default is done only via the operator-out-of-band env knob
``OC_CHAT_MAX_TOOL_CALLS_CEILING`` (``_env_guard``-enrolled, stripped from a
workspace ``.env``). The F52 "non-overridable by model args" DoS bound thus holds
against the settings-file path too — see :func:`resolve_max_tool_calls_per_turn`.
 
Responsibilities
----------------
- The cap constants (:data:`MAX_TOOL_CALLS_PER_TURN`, :data:`MAX_PARALLEL_WORKERS`,
  :data:`MAX_TIMEOUT_S`, :data:`MAX_BYTES`, :data:`MAX_SEARCH_RESULTS`).
- :func:`clamp_timeout` / :func:`clamp_bytes` / :func:`clamp_results` /
  :func:`clamp_workers` — clamp a (possibly hostile) model value to its ceiling.
- :func:`partition_call_budget` — split this iteration's tool calls into
  (allowed, denied) given the per-turn budget already spent.
- :func:`resolve_session_cost_hard_cap_usd` / :func:`resolve_cost_thresholds_usd`
  — the plan-237 (CAPS-01) money bounds, resolved under the same F33/§5.4 rule.
 
Contracts
---------
- A clamp NEVER returns a value above its ceiling; a non-int / unparseable value
  fails to the ceiling (bounded), never to "unbounded".
- Clamps only bound the UPPER side; a small/negative value passes through (a tiny
  read/timeout is not a DoS).
- :func:`partition_call_budget` is the deterministic per-turn cap helper; the
  react-loop denies (does not execute) calls beyond the budget but still answers
  them so the tool-call/result pairing stays well-formed.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.defaults (shipped cap values, ceilings, env-var names and setting keys)
- oc_chat.core.settings (lazy; runtime_settings for workspace cap and checkpoint requests)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L2 — a model arg was clamped below its requested value
    L3 — per-call clamp detail
    L4 — deep tracing
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
import os
 
from oc_chat.core.defaults import ceiling, env_var, setting_key, shipped
 
#: Appendix B (LD-3) — normative hard ceilings. The VALUES live in the shipped,
#: read-only ``oc_chat.defaults.toml`` (CONFIG-01, plan 238), which is hard-denied
#: for writes by name; these names remain as import-time aliases so every existing
#: call site keeps working.
#: Per-turn tool-call ceiling DEFAULT (amended from the V1.0 Appendix-B value of
#: 10 → 50, operator-approved 2026-06-13). This is also the default ABSOLUTE
#: ceiling: a workspace ``max_tool_calls_per_turn`` setting may LOWER the effective
#: cap below it but can NEVER RAISE it (the settings file is model-reachable;
#: F33/§5.4). Raising the ceiling above the default is done only via the OOB env
#: knob :data:`_CAP_CEILING_ENV_VAR` — see :func:`resolve_max_tool_calls_per_turn`.
MAX_TOOL_CALLS_PER_TURN = shipped("tool_calls_per_turn")
MAX_PARALLEL_WORKERS = shipped("parallel_workers")
MAX_TIMEOUT_S = shipped("timeout_seconds")
MAX_BYTES = shipped("byte_budget")          # 5 MB read / fetch-body / output ceiling
MAX_SEARCH_RESULTS = shipped("search_results")
MAX_DIR_DEPTH = shipped("dir_depth")        # directory-walk recursion depth (list_dir)
 
#: §4.4(a) OOB env knob: raises the absolute per-turn-cap ceiling above the code
#: default. Operator-out-of-band only — enrolled in ``_env_guard._AUTHORITY_EXACT``
#: (stripped from a workspace ``.env``, preserved from the real process env),
#: never settable from the workspace settings file. Absent / non-positive / non-int
#: ⇒ the code default :data:`MAX_TOOL_CALLS_PER_TURN`. The name is declared by the
#: ``tool_calls_per_turn`` table's ``env`` field, not restated here.
_CAP_CEILING_ENV_VAR = env_var("tool_calls_per_turn")
 
#: CAPS-01 (plan 237): the OOB env knob that raises the session dollar hard cap.
#: Same class of authority as :data:`_CAP_CEILING_ENV_VAR` and enrolled in
#: ``_env_guard._AUTHORITY_EXACT`` beside it. The name is declared by the
#: ``session_cost_hard_cap_usd`` table's ``env`` field, not restated here — the
#: refusal message reads it from here so the text can never name a stale knob.
_COST_CEILING_ENV_VAR = env_var("session_cost_hard_cap_usd")
 
 
def _clamp_upper(value: object, ceiling: int, *, label: str) -> int:
    """Return ``int(value)`` bounded above by *ceiling*; an unparseable value fails
    to *ceiling* (bounded, never unbounded). Values at or below the ceiling pass."""
    func = "_clamp_upper"
    try:
        v = int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError, OverflowError):  # OverflowError: int(float('inf'))
        _dbg("OC_CHAT.CORE", func, f"{label}: non-int {value!r} -> ceiling {ceiling}", 2)
        return ceiling
    if v > ceiling:
        _dbg("OC_CHAT.CORE", func, f"{label}: clamped {v} -> {ceiling}", 2)
        return ceiling
    return v
 
 
def clamp_timeout(timeout: object) -> int:
    """Clamp a model ``timeout`` (seconds) to :data:`MAX_TIMEOUT_S`."""
    return _clamp_upper(timeout, MAX_TIMEOUT_S, label="timeout")
 
 
def clamp_bytes(n: object) -> int:
    """Clamp a model byte/char budget to :data:`MAX_BYTES` (5 MB)."""
    return _clamp_upper(n, MAX_BYTES, label="bytes")
 
 
def clamp_results(n: object) -> int:
    """Clamp a model result count to :data:`MAX_SEARCH_RESULTS`."""
    return _clamp_upper(n, MAX_SEARCH_RESULTS, label="results")
 
 
def clamp_depth(n: object) -> int:
    """Clamp a model directory-walk depth to :data:`MAX_DIR_DEPTH`."""
    return _clamp_upper(n, MAX_DIR_DEPTH, label="depth")
 
 
def clamp_workers(n: int) -> int:
    """Bound a parallel-worker pool to :data:`MAX_PARALLEL_WORKERS` (≥ 1)."""
    try:
        v = int(n)
    except (TypeError, ValueError, OverflowError):
        return 1
    return max(1, min(MAX_PARALLEL_WORKERS, v))
 
 
def resolve_max_tool_calls_per_turn() -> int:
    """The effective per-turn tool-call ceiling — an ABSOLUTE ceiling a settings
    write may LOWER but never RAISE (F33/§5.4).
 
    The workspace ``max_tool_calls_per_turn`` setting (a model-reachable,
    human-confirm-tier write) can only *lower* the effective cap; it can never
    raise it above the operator's out-of-band ceiling (the shipped value
    :data:`MAX_TOOL_CALLS_PER_TURN`, or the OOB env knob
    :data:`_CAP_CEILING_ENV_VAR` when set, which ``_env_guard`` keeps off the
    ``.env`` path). So the F52 "non-overridable by model args" DoS bound holds
    against the settings-file path too. Degrades to the bounded default on a
    non-int / unavailable settings read.
 
    Plan 238 (CONFIG-01) moved the rule itself into
    :func:`oc_chat.core.defaults.ceiling`, which applies it to every ceiling in
    the system; this function keeps its name, its callers and its semantics.
    """
    return ceiling("tool_calls_per_turn")
 
 
def _settings_request(setting: str) -> "float | None":
    """The workspace settings' numeric REQUEST for *setting*, or ``None``.
 
    ``None`` means "no request was made" — the settings subsystem is
    unavailable, the key is absent, the value is unparseable, or it is
    non-positive (``0`` is the documented "no request" spelling). Every such
    answer degrades the caller to its bounded default, never to "unbounded".
    """
    func = "_settings_request"
    if not setting:
        return None
    try:
        from oc_chat.core.settings import runtime_settings
        raw = getattr(runtime_settings(), setting, None)
    except Exception as exc:  # noqa: BLE001 — settings unavailable ⇒ bounded default
        _dbg("OC_CHAT.CORE", func, f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
        return None
    try:
        value = float(raw)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None
    return value if value > 0 else None
 
 
def _env_raised_ceiling(key: str) -> int:
    """The shipped ceiling for *key*, raised ONLY by the env variable it names.
 
    This is :func:`oc_chat.core.defaults.ceiling`'s first half, reached through
    the public accessors. It exists because the two ceilings compose the F33
    rule from OPPOSITE starting points, and one function cannot be both:
 
    - ``tool_calls_per_turn``'s settings key carries a shipped-valued default
      (``DEFAULT_SETTINGS["max_tool_calls_per_turn"] == shipped(...)``), so
      "unset" is indistinguishable from "requested exactly the shipped value".
      Its resolver is therefore ``min(request, env_or_shipped)`` and the env
      raises the ceiling the settings request is clamped by — unchanged here.
    - ``session_cost_hard_cap_usd``'s settings key defaults to ``0`` = "no
      request", so an operator who exports the env variable and writes nothing
      else must get the raised cap. That is P4's "raisable only out of band".
 
    Both are the same rule — settings may LOWER, never RAISE; the operator's
    out-of-band environment raises — applied to a key whose settings default is
    a value, and to one whose settings default is a sentinel.
    """
    func = "_env_raised_ceiling"
    absolute = int(shipped(key))
    name = env_var(key)
    if not name:
        return absolute
    raw = os.environ.get(name)
    if raw is None:
        return absolute
    try:
        candidate = int(raw)
    except (TypeError, ValueError):
        return absolute
    if candidate <= 0:
        return absolute
    _dbg("OC_CHAT.CORE", func, f"{key}: OOB ceiling {name}={candidate}", 2)
    return candidate
 
 
def resolve_session_cost_hard_cap_usd() -> int:
    """The effective session dollar hard cap — settings may LOWER, env raises.
 
    CAPS-01 / P4. The shipped ``session_cost_hard_cap_usd`` (50) is the
    adopters' protection; the workspace settings key the table names may only
    lower it (``0`` = no request); the ceiling itself moves only when the
    operator exports :data:`_COST_CEILING_ENV_VAR` out of band, which
    ``_env_guard`` keeps off the workspace ``.env`` path. Whole dollars: the cap
    is an int so it reuses the same env parsing every other ceiling uses, and a
    fractional cap would suggest a precision the honesty gate cannot deliver.
    """
    absolute = _env_raised_ceiling("session_cost_hard_cap_usd")
    request = _settings_request(setting_key("session_cost_hard_cap_usd"))
    if request is None:
        return absolute
    return min(absolute, int(request))
 
 
def resolve_cost_thresholds_usd() -> tuple[float, ...]:
    """The effective session-cost checkpoint ladder, ascending.
 
    CAPS-01 / P3, under the same F33 rule as every ceiling: the shipped ladder
    is a FLOOR of protection. A workspace settings list may add checkpoints
    BELOW the shipped ceiling — which only makes the session ask sooner — and
    can neither remove a shipped checkpoint nor add one above the top of the
    shipped ladder (that would move the last warning later, i.e. raise a bound
    from a model-reachable file). An unavailable or malformed settings value
    leaves the shipped ladder exactly as it is.
    """
    func = "resolve_cost_thresholds_usd"
    ladder = [float(v) for v in shipped("session_cost_thresholds_usd")]
    top = max(ladder)
    setting = setting_key("session_cost_thresholds_usd")
    extra: list[float] = []
    if setting:
        try:
            from oc_chat.core.settings import runtime_settings
            raw = getattr(runtime_settings(), setting, None)
        except Exception as exc:  # noqa: BLE001 — settings unavailable ⇒ shipped ladder
            _dbg("OC_CHAT.CORE", func, f"VALIDATION_ERROR: {type(exc).__name__}: {exc}", 3)
            raw = None
        if isinstance(raw, (list, tuple)):
            for item in raw:
                try:
                    value = float(item)
                except (TypeError, ValueError):
                    continue
                if 0 < value < top:
                    extra.append(value)
                else:
                    _dbg("OC_CHAT.CORE", func,
                         f"settings checkpoint {value!r} ignored: not below the "
                         f"shipped ceiling {top}", 3)
    return tuple(sorted(set(ladder) | set(extra)))
 
 
def partition_call_budget(prior: int, n: int, cap: "int | None" = None) -> tuple[int, int]:
    """Return ``(allowed, denied)`` for *n* new tool calls this iteration given
    *prior* calls already spent this turn, capped at *cap* (default: the resolved
    per-turn ceiling, :func:`resolve_max_tool_calls_per_turn`)."""
    if cap is None:
        cap = resolve_max_tool_calls_per_turn()
    allowed = max(0, min(n, cap - max(0, prior)))
    return allowed, n - allowed
 
 
__all__ = [
    "MAX_TOOL_CALLS_PER_TURN",
    "MAX_PARALLEL_WORKERS",
    "MAX_TIMEOUT_S",
    "MAX_BYTES",
    "MAX_SEARCH_RESULTS",
    "MAX_DIR_DEPTH",
    "clamp_timeout",
    "clamp_bytes",
    "clamp_results",
    "clamp_depth",
    "clamp_workers",
    "partition_call_budget",
    "resolve_cost_thresholds_usd",
    "resolve_max_tool_calls_per_turn",
    "resolve_session_cost_hard_cap_usd",
]