#!/usr/bin/env python3 """Offline verifier: loop-ops pricing must match claude-api-ops's model table. loop-estimate.py reads assets/model-pricing.json. That table is a *copy* of the authoritative "Current Models" table in skills/claude-api-ops/SKILL.md — and a copy drifts silently (the exact §7 failure mode). This asserts every model in loop-ops' pricing exists in the claude-api-ops table with matching input/output prices. Both files are in-repo, so this is a pure OFFLINE consistency check and safe to gate PR CI (no network). Live model-id drift is owned by claude-api-ops/scripts/check-model-table.py. Usage: check-pricing-sync.py [--offline] [--pricing FILE] [--table FILE] [--json] Input: argv flags only (no stdin). Output: stdout = drift findings (plain rows, or --json envelope). Data only. Stderr: the verdict panel, notices, errors. Exit: 0 in sync, 2 usage, 3 a file missing, 4 unparseable, 10 drift found --offline is the default and only mode (accepted for parity with the other §7 verifiers invoked by tests/check-resources.sh). Examples: check-pricing-sync.py --offline check-pricing-sync.py --json | jq '.data[]' """ from __future__ import annotations import argparse import json import os import re import sys from pathlib import Path EX_OK = 0 EX_USAGE = 2 EX_NOTFOUND = 3 EX_UNPARSEABLE = 4 EX_DRIFT = 10 HERE = Path(__file__).resolve().parent DEFAULT_PRICING = HERE.parent / "assets" / "model-pricing.json" DEFAULT_TABLE = HERE.parent.parent / "claude-api-ops" / "SKILL.md" PRICE_RE = re.compile(r"\$?\s*([0-9]+(?:\.[0-9]+)?)") class Term: """Minimal ANSI helper (term.sh is bash-only; per TERMINAL-DESIGN.md §9 the Python port is inline). Honors FORCE_COLOR / NO_COLOR / TERM_ASCII and the bound stream's TTY + encoding so piped data stays plain ASCII.""" _C = {"green": "\033[32m", "red": "\033[31m", "cyan": "\033[36m", "dim": "\033[2m", "off": "\033[0m"} def __init__(self, stream=sys.stderr): enc = (getattr(stream, "encoding", "") or "").lower() self.ascii = os.environ.get("TERM_ASCII") == "1" or "utf" not in enc if os.environ.get("FORCE_COLOR"): self.color = True elif (os.environ.get("NO_COLOR") is not None or os.environ.get("TERM") == "dumb" or not getattr(stream, "isatty", lambda: False)()): self.color = False else: self.color = True def c(self, name, text): return f"{self._C.get(name,'')}{text}{self._C['off']}" if self.color else text def mark(self, ok): g = ("+" if self.ascii else "✓") if ok else ("x" if self.ascii else "✗") return self.c("green" if ok else "red", g) def parse_price(cell: str) -> float | None: m = PRICE_RE.search(cell) return float(m.group(1)) if m else None def load_pricing(path: Path) -> dict: """{model_id: (input_per_mtok, output_per_mtok)} from loop-ops' JSON.""" if not path.is_file(): print(f"error: pricing file not found: {path}", file=sys.stderr) raise SystemExit(EX_NOTFOUND) try: data = json.loads(path.read_text(encoding="utf-8")) out = {} for mid, pr in data.get("models", {}).items(): out[mid] = (float(pr["input_per_mtok"]), float(pr["output_per_mtok"])) if not out: print(f"error: no models in {path}", file=sys.stderr) raise SystemExit(EX_UNPARSEABLE) return out except (json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: print(f"error: could not parse pricing file: {exc}", file=sys.stderr) raise SystemExit(EX_UNPARSEABLE) def load_table(path: Path) -> dict: """{model_id: (input_price, output_price)} from the claude-api-ops markdown 'Current Models' table. Columns: Model | ID | Context | Max Output | Input | Output.""" if not path.is_file(): print(f"error: claude-api-ops table not found: {path}", file=sys.stderr) raise SystemExit(EX_NOTFOUND) table: dict = {} in_table = False for line in path.read_text(encoding="utf-8").splitlines(): s = line.strip() low = s.lower() if s.startswith("|") and "id" in low and "context" in low and "output" in low: in_table = True continue if in_table: if not s.startswith("|"): if table: # table ended break continue if set(s) <= set("|-: "): # separator row continue cells = [c.strip() for c in s.strip("|").split("|")] if len(cells) < 6: continue mid = cells[1].strip("`").strip() if not mid.startswith("claude-"): continue ip, op = parse_price(cells[4]), parse_price(cells[5]) if ip is not None and op is not None: table[mid] = (ip, op) if not table: print(f"error: no model rows parsed from {path}", file=sys.stderr) raise SystemExit(EX_UNPARSEABLE) return table def main(argv: list[str]) -> int: p = argparse.ArgumentParser( prog="check-pricing-sync.py", description="Verify loop-ops pricing matches claude-api-ops's model table (offline).", ) p.add_argument("--offline", action="store_true", help="offline consistency check (default/only mode)") p.add_argument("--pricing", default=str(DEFAULT_PRICING), help="loop-ops model-pricing.json") p.add_argument("--table", default=str(DEFAULT_TABLE), help="claude-api-ops SKILL.md with the model table") p.add_argument("--json", action="store_true", help="emit a JSON envelope") try: args = p.parse_args(argv) except SystemExit as exc: return EX_USAGE if exc.code not in (0, None) else (exc.code or EX_OK) pricing = load_pricing(Path(args.pricing)) table = load_table(Path(args.table)) findings = [] for mid, (ip, op) in sorted(pricing.items()): if mid not in table: findings.append({"model": mid, "issue": "absent from claude-api-ops table", "loop_ops": [ip, op], "authoritative": None}) continue tip, top = table[mid] if abs(ip - tip) > 1e-9 or abs(op - top) > 1e-9: findings.append({"model": mid, "issue": "price mismatch", "loop_ops": [ip, op], "authoritative": [tip, top]}) if args.json: print(json.dumps({ "data": findings, "meta": {"count": len(findings), "models_checked": len(pricing), "in_sync": not findings, "schema": "claude-mods.loop-ops.pricing-sync/v1"}, }, indent=2)) else: for f in findings: auth = f"authoritative {f['authoritative']}" if f["authoritative"] else "not in table" print(f"DRIFT {f['model']}: {f['issue']} (loop-ops {f['loop_ops']} vs {auth})") t = Term(sys.stderr) ok = not findings print(f"{t.mark(ok)} pricing-sync: {len(pricing)} model(s) checked, " f"{len(findings)} drift " f"{t.c('dim', '(authoritative: claude-api-ops/SKILL.md)')}", file=sys.stderr) return EX_DRIFT if findings else EX_OK if __name__ == "__main__": sys.exit(main(sys.argv[1:]))