tegwick
c63caf5568
Some checks are pending
Governed runtime contract / contract (push) Waiting to run
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""Claude CLI controls and bounded terminal accounting, not a spend admission grant.
|
|
|
|
Daily/total reservation and provider cap semantics remain the operating owner's
|
|
responsibility. A post-run cost check cannot undo an already charged request.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import math
|
|
from typing import Any
|
|
|
|
|
|
def validate_limits(max_budget_usd: Any, max_turns: Any) -> None:
|
|
if max_budget_usd is not None and (
|
|
isinstance(max_budget_usd, bool)
|
|
or not isinstance(max_budget_usd, (int, float))
|
|
or not math.isfinite(max_budget_usd)
|
|
or max_budget_usd <= 0
|
|
):
|
|
raise ValueError("max_budget_usd must be a finite positive USD amount")
|
|
if max_turns is not None and (
|
|
isinstance(max_turns, bool) or not isinstance(max_turns, int) or max_turns <= 0
|
|
):
|
|
raise ValueError("max_turns must be a positive integer")
|
|
|
|
|
|
class NativeLimitError(RuntimeError):
|
|
def __init__(self, message: str, *, cost_usd: float | None = None) -> None:
|
|
super().__init__(message)
|
|
self.cost_usd = cost_usd
|
|
|
|
|
|
def terminal_accounting(
|
|
result: Any, *, max_budget_usd: float | None, max_turns: int | None
|
|
) -> tuple[dict, float]:
|
|
if not isinstance(result, dict) or result.get("type") != "result":
|
|
raise NativeLimitError(
|
|
"controlled Claude run did not return terminal accounting"
|
|
)
|
|
cost = result.get("total_cost_usd")
|
|
if (
|
|
isinstance(cost, bool)
|
|
or not isinstance(cost, (int, float))
|
|
or not math.isfinite(cost)
|
|
or cost < 0
|
|
):
|
|
raise NativeLimitError("controlled Claude run returned invalid cost accounting")
|
|
cost = float(cost)
|
|
if result.get("subtype") != "success" or result.get("is_error") is not False:
|
|
raise NativeLimitError(
|
|
"controlled Claude run did not complete successfully", cost_usd=cost
|
|
)
|
|
if max_budget_usd is not None and cost > max_budget_usd:
|
|
raise NativeLimitError(
|
|
"controlled Claude run exceeded its native USD limit", cost_usd=cost
|
|
)
|
|
turns = result.get("num_turns")
|
|
if (
|
|
isinstance(turns, bool)
|
|
or not isinstance(turns, int)
|
|
or turns < 0
|
|
or (max_turns is not None and turns > max_turns)
|
|
):
|
|
raise NativeLimitError(
|
|
"controlled Claude run returned invalid turn accounting", cost_usd=cost
|
|
)
|
|
usage = result.get("usage")
|
|
if not isinstance(usage, dict):
|
|
raise NativeLimitError(
|
|
"controlled Claude run returned no token accounting", cost_usd=cost
|
|
)
|
|
counts = {}
|
|
for name in (
|
|
"input_tokens",
|
|
"output_tokens",
|
|
"cache_creation_input_tokens",
|
|
"cache_read_input_tokens",
|
|
):
|
|
value = usage.get(name, 0 if name.startswith("cache_") else None)
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise NativeLimitError(
|
|
"controlled Claude run returned invalid token accounting", cost_usd=cost
|
|
)
|
|
counts[name] = value
|
|
return {
|
|
"prompt_tokens": counts["input_tokens"]
|
|
+ counts["cache_creation_input_tokens"]
|
|
+ counts["cache_read_input_tokens"],
|
|
"completion_tokens": counts["output_tokens"],
|
|
"total_tokens": sum(counts.values()),
|
|
}, cost
|