Files
CASAN/AINative_OKR_CASAN5/.specify/scripts/bash/cost-spike-detect.sh
T
thanhnvandClaude Opus 4.8 7e998f67c2 feat(h5-h6-hardening): Plan-07 Track A — A4 telemetry integrity, A5 cost controls
A4 (V9): telemetry-integrity.sh binds provider-usage.jsonl + cost/metrics.jsonl
  to a signed manifest head. Tampering a token flips the head (MISMATCH); an
  attacker who rewrites the head cannot re-sign it (SIGNATURE_INVALID) without
  the off-repo key. sign-audit-head.sh now also signs telemetry (best-effort).
  Supports CASAN_AUDIT_PRIV/PUB overrides for self-contained verification.
A5 (V12/V13/V14): cost-spike-detect.sh adds an absolute per-call cap
  (CASAN_COST_ABSOLUTE_MAX_TOKENS, enforced from record #1 → catches slow-boil
  and cold-start) and a cumulative budget (CASAN_COST_CUMULATIVE_BUDGET_TOKENS →
  catches under-threshold spray), keeping the existing median×mult spike test.
  Backward compatible: <3 records with no caps still exits 3.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-03 22:25:00 +09:00

89 lines
3.6 KiB
Bash
Executable File

#!/usr/bin/env bash
set -uo pipefail
# CASAN H6 cost detector (WP-C + Track A cost controls A5 / V12,V13,V14).
# Answers the H6 key question: "if a step suddenly costs 3x the tokens, does
# anyone know?". Reads real per-step usage from provider-usage.jsonl and applies
# three complementary controls:
# 1. Relative spike — any step > MULTIPLIER x median (needs >=3 records).
# 2. Absolute cap — any single step > CASAN_COST_ABSOLUTE_MAX_TOKENS. Works
# from the very first record, so a slow-boil that drags the median up
# (V12) and a cold-start with no history (V14) are BOTH still caught.
# 3. Cumulative budget — sum of tokens > CASAN_COST_CUMULATIVE_BUDGET_TOKENS,
# catching a spray of many under-threshold calls (V13).
#
# Usage: cost-spike-detect.sh [provider-usage.jsonl] [multiplier]
# Env: CASAN_COST_ABSOLUTE_MAX_TOKENS per-call hard cap (0/unset = off)
# CASAN_COST_CUMULATIVE_BUDGET_TOKENS run/session budget (0/unset = off)
# Exit: 0 no violation, 2 violation (spike/absolute/cumulative), 64 usage,
# 3 not enough data for the RELATIVE test and no absolute/cumulative cap set.
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
LOG="${1:-$PROJECT_ROOT/.specify/logs/level5/provider-usage.jsonl}"
MULT="${2:-3.0}"
[[ -f "$LOG" ]] || { echo "COST_SPIKE_NO_DATA file=$LOG" >&2; exit 3; }
CASAN_COST_ABSOLUTE_MAX_TOKENS="${CASAN_COST_ABSOLUTE_MAX_TOKENS:-0}" \
CASAN_COST_CUMULATIVE_BUDGET_TOKENS="${CASAN_COST_CUMULATIVE_BUDGET_TOKENS:-0}" \
python - "$LOG" "$MULT" <<'PY'
import json, os, sys, statistics
path, mult = sys.argv[1], float(sys.argv[2])
abs_max = int(os.environ.get("CASAN_COST_ABSOLUTE_MAX_TOKENS", "0") or "0")
cum_budget = int(os.environ.get("CASAN_COST_CUMULATIVE_BUDGET_TOKENS", "0") or "0")
rows = []
for line in open(path, encoding="utf-8"):
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
except ValueError:
continue
if "total_tokens" in r:
rows.append((r.get("step", "?"), int(r["total_tokens"])))
tokens = [t for _, t in rows]
total = sum(tokens)
violations = []
# 2. Absolute per-call cap — enforced from the first record (cold-start safe).
if abs_max > 0:
over = [(s, t) for s, t in rows if t > abs_max]
for s, t in over:
violations.append(f"ABSOLUTE step={s} tokens={t} (> cap {abs_max})")
# 3. Cumulative budget over the whole run/session.
if cum_budget > 0 and total > cum_budget:
violations.append(f"CUMULATIVE total={total} (> budget {cum_budget})")
# 1. Relative median spike — needs enough history.
median = threshold = None
if len(rows) >= 3:
median = statistics.median(tokens)
threshold = median * mult
for s, t in rows:
if t > threshold:
violations.append(f"SPIKE step={s} tokens={t} (> {threshold:.0f} = median x{mult})")
med_str = f"{median}" if median is not None else "n/a(<3 records)"
print(f"records={len(rows)} total_tokens={total} median_tokens={med_str} "
f"abs_cap={abs_max or 'off'} cum_budget={cum_budget or 'off'}")
for v in violations:
print(v)
if violations:
sys.stderr.write(f"COST_VIOLATION_DETECTED count={len(violations)}\n")
raise SystemExit(2)
# No violations. If we could not run the relative test AND no absolute/cumulative
# cap was configured, we truly had nothing to check -> preserve the old rc=3.
if len(rows) < 3 and abs_max == 0 and cum_budget == 0:
sys.stderr.write(f"COST_SPIKE_NO_DATA records={len(rows)} (need >=3, no absolute/cumulative cap set)\n")
raise SystemExit(3)
print("COST_SPIKE_NONE")
PY