From 1aa2c68818a1fb9171076f68d5e49315c0fc38fa Mon Sep 17 00:00:00 2001 From: Evandro Fonseca Junior Date: Sat, 10 Oct 2026 04:05:58 -0300 Subject: [PATCH] refactor(catalog): split validators and staleness checks below CC 10, with path tests Adds qa/test_branch_paths.py pinning the main path and one inverted path per decision for validate_aa_snapshot, validate_availability_snapshot, validate_native_tariff_overrides, model_router._staleness and _native_cost_and_headroom, then splits each into small helpers. Messages, check order and results are unchanged. --- qa/test_branch_paths.py | 455 +++++++++++++++++++++++++++++++++++++++ scripts/model_catalog.py | 296 +++++++++++++++---------- scripts/model_router.py | 233 ++++++++++++-------- 3 files changed, 785 insertions(+), 199 deletions(-) create mode 100644 qa/test_branch_paths.py diff --git a/qa/test_branch_paths.py b/qa/test_branch_paths.py new file mode 100644 index 0000000..522e2ea --- /dev/null +++ b/qa/test_branch_paths.py @@ -0,0 +1,455 @@ +#!/usr/bin/env python3 +"""Characterization tests: one main path plus one path per decision for the validators and +the staleness/cost helpers. They pin current behavior so the functions can be split safely. + +python3 qa/test_branch_paths.py +""" +import copy +import os +import sys +import unittest +from datetime import datetime, timedelta, timezone +from decimal import Decimal + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(ROOT, "scripts")) +sys.path.insert(0, os.path.join(ROOT, "qa")) +import model_catalog as catalog +import model_router +from test_model_router import NOW, NOW_TEXT, candidate + +PAST = "2026-01-01T00:00:00Z" +FUTURE = "2099-01-01T00:00:00Z" +OLD = "2026-08-01T00:00:00Z" + + +def mutated(base, *edits): + """Deep copy of base with edits [(path, value)]; value DELETE removes the key.""" + result = copy.deepcopy(base) + for path, value in edits: + node = result + for key in path[:-1]: + node = node[key] + if value is DELETE: + node.pop(path[-1], None) + else: + node[path[-1]] = value + return result + + +DELETE = object() + + +class ValidationCases(unittest.TestCase): + def assert_rejected(self, function, base, cases): + for name, edits, message in cases: + with self.subTest(name): + with self.assertRaisesRegex(ValueError, message): + function(mutated(base, *edits)) + + +class NativeTariffTests(ValidationCases): + ZEN = {"opencode/m": { + "billing_mode": "zen_balance", "account_confirmed": True, "source_url": "https://opencode.ai/docs/zen", + "captured_at": PAST, "price_per_token_usd": {"input": "0.1", "output": "0.2"}, + "available_balance_usd": "2"}} + GO = {"opencode/m": { + "billing_mode": "go_quota", "account_confirmed": True, "source_url": "https://opencode.ai/docs/go", + "captured_at": PAST, + "quota_window_remaining": {"5h": "1", "weekly": "2", "monthly": "3"}, + "quota_window_debit": {"5h": "0.1", "weekly": "0.2", "monthly": "0.3"}}} + + def test_main_paths_normalize_records(self): + zen = catalog.validate_native_tariff_overrides(self.ZEN)["opencode/m"] + self.assertEqual(zen["price_per_token_usd"], {"input": "0.1", "output": "0.2"}) + self.assertEqual(zen["available_balance_usd"], "2") + self.assertNotIn("quota_window_remaining", zen) + go = catalog.validate_native_tariff_overrides(self.GO)["opencode/m"] + self.assertEqual(go["quota_window_remaining"], {"5h": "1", "weekly": "2", "monthly": "3"}) + self.assertEqual(go["quota_window_debit"]["monthly"], "0.3") + self.assertIs(go["use_balance_fallback"], False) + self.assertNotIn("available_balance_usd", go) + self.assertEqual(catalog.validate_native_tariff_overrides({}), {}) + + def test_omp_payg_follows_the_zen_branch_and_fallback_bool_is_kept(self): + omp = mutated(self.ZEN, (("opencode/m", "billing_mode"), "omp_payg")) + self.assertEqual(catalog.validate_native_tariff_overrides(omp)["opencode/m"]["billing_mode"], "omp_payg") + go = mutated(self.GO, (("opencode/m", "use_balance_fallback"), True)) + self.assertIs(catalog.validate_native_tariff_overrides(go)["opencode/m"]["use_balance_fallback"], True) + + def test_shared_decisions_reject(self): + z = "opencode/m" + self.assert_rejected(catalog.validate_native_tariff_overrides, self.ZEN, [ + ("unknown field", [((z, "extra"), 1)], "estrutura/campo inválido"), + ("value not a dict", [((z,), "x")], "estrutura/campo inválido"), + ("not confirmed", [((z, "account_confirmed"), False)], "confirmação explícita"), + ("bad mode", [((z, "billing_mode"), "other")], "billing_mode native inválido"), + ("http url", [((z, "source_url"), "http://opencode.ai/docs")], "source_url HTTPS"), + ("captured_at missing", [((z, "captured_at"), DELETE)], "captured_at de tarifa/cota native inválido"), + ("captured_at garbage", [((z, "captured_at"), "x")], "captured_at de tarifa/cota native inválido"), + ("captured_at naive", [((z, "captured_at"), "2026-01-01T00:00:00")], "aware e não futuro"), + ("captured_at future", [((z, "captured_at"), FUTURE)], "aware e não futuro"), + ]) + with self.assertRaisesRegex(ValueError, "deve ser objeto JSON"): + catalog.validate_native_tariff_overrides([]) + with self.assertRaisesRegex(ValueError, "estrutura/campo inválido"): + catalog.validate_native_tariff_overrides({1: self.ZEN["opencode/m"]}) + + def test_balance_branch_rejects(self): + z = "opencode/m" + self.assert_rejected(catalog.validate_native_tariff_overrides, self.ZEN, [ + ("rates missing", [((z, "price_per_token_usd"), DELETE)], "precisa de price_per_token_usd"), + ("rates empty", [((z, "price_per_token_usd"), {})], "precisa de price_per_token_usd"), + ("unknown component", [((z, "price_per_token_usd"), {"input": "1", "foo": "1"})], "componentes de tarifa"), + ("negative component", [((z, "price_per_token_usd"), {"input": "-1"})], "componentes de tarifa"), + ("empty component", [((z, "price_per_token_usd"), {"input": ""})], "componentes de tarifa"), + ("balance missing", [((z, "available_balance_usd"), DELETE)], "saldo disponível"), + ("balance negative", [((z, "available_balance_usd"), "-1")], "saldo disponível"), + ]) + + def test_go_branch_rejects(self): + z = "opencode/m" + self.assert_rejected(catalog.validate_native_tariff_overrides, self.GO, [ + ("remaining missing", [((z, "quota_window_remaining"), DELETE)], "5h/weekly/monthly"), + ("debit not dict", [((z, "quota_window_debit"), "1")], "5h/weekly/monthly"), + ("window missing in remaining", [((z, "quota_window_remaining"), {"5h": "1", "weekly": "2"})], "5h/weekly/monthly"), + ("window missing in debit", [((z, "quota_window_debit"), {"5h": "1", "monthly": "2"})], "5h/weekly/monthly"), + ("negative remaining", [((z, "quota_window_remaining", "5h"), "-1")], "não negativos"), + ("empty debit", [((z, "quota_window_debit", "weekly"), "")], "não negativos"), + ("fallback not bool", [((z, "use_balance_fallback"), "yes")], "use_balance_fallback precisa ser bool"), + ]) + + +class AvailabilityTests(ValidationCases): + BASE = { + "tool": "opencode", "ingestion": "manual_audited", "user_confirmed": True, "captured_at": PAST, + "records": [{"provider": "opencode", "model_id": "m", "variant": None, "local_listed": True, + "authenticated": True, "entitled": None, "verified_at": PAST, "source": "manual"}]} + + def rec(self, **edits): + return [(("records", 0, key), value) for key, value in edits.items()] + + def test_main_path_returns_audited_record_keyed_by_identity(self): + result = catalog.validate_availability_snapshot(self.BASE) + record = result[("opencode", "m", None)] + self.assertEqual((record["local_listed"], record["authenticated"], record["entitled"]), (True, True, None)) + self.assertEqual(record["source"], "manual_audited") + self.assertIsNone(record["balance_confirmed"]) + self.assertEqual(len(record["source_hash"]), 64) + + def test_balance_confirmed_with_valid_time_is_kept(self): + payload = mutated(self.BASE, *self.rec(balance_confirmed=True, balance_verified_at=PAST)) + record = catalog.validate_availability_snapshot(payload)[("opencode", "m", None)] + self.assertEqual((record["balance_confirmed"], record["balance_verified_at"]), (True, PAST)) + + def test_snapshot_level_decisions_reject(self): + self.assert_rejected(catalog.validate_availability_snapshot, self.BASE, [ + ("tool", [(("tool",), "claude")], "manual auditado"), + ("ingestion", [(("ingestion",), "auto")], "manual auditado"), + ("not confirmed", [(("user_confirmed",), False)], "user_confirmed=true"), + ("captured_at not str", [(("captured_at",), 5)], "captured_at de disponibilidade obrigatório"), + ("captured_at garbage", [(("captured_at",), "x")], "captured_at de disponibilidade inválido"), + ("captured_at naive", [(("captured_at",), "2026-01-01T00:00:00")], "UTC/offset-aware"), + ("captured_at future", [(("captured_at",), FUTURE)], "UTC/offset-aware"), + ("records not list", [(("records",), {})], "deve ser lista"), + ]) + with self.assertRaisesRegex(ValueError, "manual auditado"): + catalog.validate_availability_snapshot([]) + + def test_record_level_decisions_reject(self): + self.assert_rejected(catalog.validate_availability_snapshot, self.BASE, [ + ("record not dict", [(("records",), ["x"])], "campos não permitidos"), + ("extra field", self.rec(extra=1), "campos não permitidos"), + ("provider not str", self.rec(provider=1), "exatos são obrigatórios"), + ("model empty", self.rec(model_id=""), "exatos são obrigatórios"), + ("model with slash", self.rec(model_id="a/b"), "exatos são obrigatórios"), + ("model with hash", self.rec(model_id="a#b"), "exatos são obrigatórios"), + ("variant not str", self.rec(variant=1), "exatos são obrigatórios"), + ("field absent", self.rec(entitled=DELETE), "todos os campos"), + ("flag not bool", self.rec(authenticated="yes"), "authenticated deve ser bool ou null"), + ("balance flag not bool", self.rec(balance_confirmed="yes"), "balance_confirmed deve ser bool ou null"), + ("balance time without confirmation", self.rec(balance_verified_at=PAST), "requer confirmação explícita"), + ("balance time future", self.rec(balance_confirmed=True, balance_verified_at=FUTURE), "requer confirmação explícita"), + ("confirmed without time", self.rec(balance_confirmed=True), "exige balance_verified_at"), + ("verified_at invalid", self.rec(verified_at="x"), "source e verified_at"), + ("verified_at future", self.rec(verified_at=FUTURE), "source e verified_at"), + ("source empty", self.rec(source=""), "source e verified_at"), + ]) + + def test_duplicate_identity_rejects_but_variant_distinguishes(self): + first = self.BASE["records"][0] + with self.assertRaisesRegex(ValueError, "duplicado"): + catalog.validate_availability_snapshot(mutated(self.BASE, (("records",), [first, dict(first)]))) + other = dict(first, variant="high") + result = catalog.validate_availability_snapshot(mutated(self.BASE, (("records",), [first, other]))) + self.assertEqual(set(result), {("opencode", "m", None), ("opencode", "m", "high")}) + + +class AaSnapshotTests(ValidationCases): + URL = "https://artificialanalysis.ai/agents/coding-agents" + BASE = { + "metric_id": catalog.AA_METRIC, "version": "1.5", "ingestion": "manual_audited", "terms_accepted": True, + "source_url": URL, "attribution": "Artificial Analysis", "captured_at": PAST, + "records": [{"provider": "p", "model_id": "m", "score": "50", "match_grade": "exact", "agent": "a", + "harness": "h", "settings": {"effort": "max"}, "evidence_url": URL}]} + + def rec(self, **edits): + return [(("records", 0, key), value) for key, value in edits.items()] + + def test_main_path_adds_provenance_and_normalizes_score(self): + row = catalog.validate_aa_snapshot(self.BASE)[0] + self.assertEqual((row["score"], row["metric_id"], row["version"]), ("50", catalog.AA_METRIC, "1.5")) + self.assertEqual((row["observed_at"], row["source_url"], row["attribution"]), (PAST, self.URL, "Artificial Analysis")) + self.assertEqual(len(row["snapshot_hash"]), 64) + self.assertEqual(catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), []))), []) + + def test_snapshot_level_decisions_reject(self): + self.assert_rejected(catalog.validate_aa_snapshot, self.BASE, [ + ("extra root field", [(("extra",), 1)], "campo não permitido"), + ("metric", [(("metric_id",), "other")], "somente aa_coding_agent_index v1.5"), + ("version", [(("version",), "1.4")], "somente aa_coding_agent_index v1.5"), + ("ingestion", [(("ingestion",), "auto")], "importação manual consentida"), + ("terms", [(("terms_accepted",), False)], "importação manual consentida"), + ("source url", [(("source_url",), "https://example.com/x")], "source_url oficial"), + ("attribution missing", [(("attribution",), DELETE)], "atribuição AA obrigatória"), + ("attribution empty", [(("attribution",), "")], "atribuição AA obrigatória"), + ("captured_at invalid", [(("captured_at",), "x")], "captured_at obrigatório"), + ("captured_at future", [(("captured_at",), FUTURE)], "captured_at obrigatório"), + ("records not list", [(("records",), {})], "records deve ser lista"), + ]) + with self.assertRaisesRegex(ValueError, "deve ser objeto"): + catalog.validate_aa_snapshot([]) + + def test_record_level_decisions_reject(self): + self.assert_rejected(catalog.validate_aa_snapshot, self.BASE, [ + ("record not dict", [(("records",), ["x"])], "registro AA inválido"), + ("record extra field", self.rec(extra=1), "registro AA inválido"), + ("record metric", self.rec(metric_id="other"), "não pode ser misturado"), + ("record version", self.rec(version="1.4"), "não pode ser misturado"), + ("provider not str", self.rec(provider=1), "mapeamento provider/model_id"), + ("model not str", self.rec(model_id=1), "mapeamento provider/model_id"), + ("model empty", self.rec(model_id=""), "mapeamento provider/model_id"), + ("model slash", self.rec(model_id="a/b"), "mapeamento provider/model_id"), + ("model hash", self.rec(model_id="a#b"), "mapeamento provider/model_id"), + ("variant not str", self.rec(variant=1), "variante AA inválida"), + ("variant empty", self.rec(variant=""), "variante AA inválida"), + ("score missing", self.rec(score=DELETE), "intervalo 0..100"), + ("score negative", self.rec(score="-1"), "intervalo 0..100"), + ("score above 100", self.rec(score="101"), "intervalo 0..100"), + ("grade", self.rec(match_grade="fuzzy"), "exact ou proxy"), + ("agent missing", self.rec(agent=DELETE), "proveniência AA obrigatório: agent"), + ("harness empty", self.rec(harness=""), "proveniência AA obrigatório: harness"), + ("settings none", self.rec(settings=None), "proveniência AA obrigatório: settings"), + ("evidence missing", self.rec(evidence_url=DELETE), "proveniência AA obrigatório: evidence_url"), + ("evidence url", self.rec(evidence_url="http://artificialanalysis.ai/x"), "evidence_url AA inválida"), + ]) + + def test_boundaries_duplicates_and_variant(self): + for score in ("0", "100"): + row = catalog.validate_aa_snapshot(mutated(self.BASE, *self.rec(score=score)))[0] + self.assertEqual(row["score"], score) + proxy = catalog.validate_aa_snapshot(mutated(self.BASE, *self.rec(match_grade="proxy", variant="high")))[0] + self.assertEqual((proxy["match_grade"], proxy["variant"]), ("proxy", "high")) + first = self.BASE["records"][0] + with self.assertRaisesRegex(ValueError, "duplicado"): + catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), [first, dict(first)]))) + both = catalog.validate_aa_snapshot(mutated(self.BASE, (("records",), [first, dict(first, variant="high")]))) + self.assertEqual(len(both), 2) + + +TOKENS = {"input": 1000, "output": 100} +HOUR = timedelta(hours=1) + + +def stale(row, **kwargs): + return model_router._staleness(row, kwargs.get("now", NOW), Decimal(24), Decimal(30), kwargs.get("tokens", TOKENS)) + + +def ago(delta): + return (NOW - delta).strftime("%Y-%m-%dT%H:%M:%SZ") + + +class StalenessTests(unittest.TestCase): + def go_row(self): + return candidate("opencode/g", billing="go_quota") + + def check(self, row, expected, **kwargs): + self.assertEqual(stale(row, **kwargs), expected) + + def test_fresh_paid_rows_have_no_reasons(self): + self.check(candidate("opencode/m"), []) + self.check(self.go_row(), []) + + def test_unpaid_mode_is_never_stale(self): + row = mutated(candidate("opencode/m", billing="free", availability={}), (("score",), {})) + self.check(row, []) + + def test_reference_price(self): + no_price = candidate("opencode/m", openrouter={"captured_at": OLD, "price_per_token_usd": {}}) + self.check(no_price, []) + for name, value in (("old", OLD), ("future", FUTURE), ("missing", None)): + with self.subTest(name): + row = candidate("opencode/m", openrouter={"captured_at": value, "price_per_token_usd": {"input": "1"}}) + self.check(row, ["preco_referencial_obsoleto_ou_ausente"]) + + def test_benchmark_score(self): + for name, value in (("old", OLD), ("future", FUTURE), ("missing", None)): + with self.subTest(name): + row = mutated(candidate("opencode/m"), (("score", "observed_at"), value)) + self.check(row, ["benchmark_obsoleto_ou_ausente"]) + self.check(mutated(candidate("opencode/m"), (("score",), None)), ["benchmark_obsoleto_ou_ausente"]) + + def test_native_tariff_for_balance_modes(self): + base = candidate("opencode/m") + cases = [ + ("not confirmed", (("native_tariff", "account_confirmed"), False)), + ("http", (("native_tariff", "source_url"), "http://x")), + ("missing url", (("native_tariff", "source_url"), DELETE)), + ("old", (("native_tariff", "captured_at"), OLD)), + ("future", (("native_tariff", "captured_at"), FUTURE)), + ("missing time", (("native_tariff", "captured_at"), DELETE)), + ] + for name, edit in cases: + with self.subTest(name): + self.check(mutated(base, edit), ["tarifa_nativa_obsoleta_ou_nao_confirmada"]) + omp = candidate("opencode/m", billing="omp_payg") + self.check(mutated(omp, (("native_tariff", "account_confirmed"), False)), ["tarifa_nativa_obsoleta_ou_nao_confirmada"]) + + def test_go_quota_freshness_window_is_sixty_seconds(self): + base = self.go_row() + self.check(mutated(base, (("quota_verified_at",), ago(timedelta(seconds=59)))), []) + for name, edit in (("old", ago(timedelta(seconds=61))), ("missing", None), ("future", FUTURE)): + with self.subTest(name): + self.check(mutated(base, (("quota_verified_at",), edit)), ["quota_go_obsoleta_ou_nao_confirmada"]) + self.check(mutated(base, (("quota_source_url",), "http://x")), ["quota_go_obsoleta_ou_nao_confirmada"]) + self.check(mutated(base, (("quota_source_url",), DELETE)), ["quota_go_obsoleta_ou_nao_confirmada"]) + + def test_local_availability(self): + base = candidate("opencode/m") + for name, edit in (("old", ago(25 * HOUR)), ("future", FUTURE), ("missing", None)): + with self.subTest(name): + row = mutated(base, (("availability", "local_verified_at"), edit), (("availability", "verified_at"), edit)) + self.check(row, ["disponibilidade_local_precisa_revalidacao"]) + fallback = mutated(base, (("availability", "local_verified_at"), DELETE)) + self.check(fallback, []) + self.check(mutated(base, (("availability", "local_verified_at"), ago(23 * HOUR))), []) + + def test_entitlement_is_checked_only_when_authenticated_and_entitled(self): + base = candidate("opencode/m") + for name, edit in (("old", ago(25 * HOUR)), ("future", FUTURE), ("missing", DELETE)): + with self.subTest(name): + self.check(mutated(base, (("availability", "account_verified_at"), edit)), ["entitlement_precisa_revalidacao"]) + stale_account = (("availability", "account_verified_at"), DELETE) + self.check(mutated(base, stale_account, (("availability", "entitled"), False)), []) + self.check(mutated(base, stale_account, (("availability", "authenticated"), None)), []) + + def test_balance_confirmation_for_balance_modes_only(self): + base = candidate("opencode/m") + cases = [ + ("unconfirmed", (("availability", "balance_confirmed"), False)), + ("old", (("availability", "balance_verified_at"), ago(timedelta(seconds=61)))), + ("future", (("availability", "balance_verified_at"), FUTURE)), + ("missing time", (("availability", "balance_verified_at"), DELETE)), + ] + for name, edit in cases: + with self.subTest(name): + self.check(mutated(base, edit), ["saldo_nao_confirmado_ou_obsoleto"]) + go = self.go_row() + self.check(mutated(go, (("availability", "balance_confirmed"), False)), []) + + def test_reasons_accumulate_in_order(self): + row = mutated(candidate("opencode/m", openrouter={"captured_at": OLD, "price_per_token_usd": {"input": "1"}}, + availability={}), (("score",), {})) + self.check(row, ["preco_referencial_obsoleto_ou_ausente", "benchmark_obsoleto_ou_ausente", + "disponibilidade_local_precisa_revalidacao", "saldo_nao_confirmado_ou_obsoleto"]) + + def test_legacy_or_price_key_and_configurable_limits(self): + row = candidate("opencode/m", openrouter=None, or_price_per_token={"input": "1"}) + self.check(row, ["preco_referencial_obsoleto_ou_ausente"]) + row = mutated(candidate("opencode/m"), (("openrouter", "captured_at"), ago(2 * HOUR))) + self.assertEqual(model_router._staleness(row, NOW, Decimal(1), Decimal(30), TOKENS), + ["preco_referencial_obsoleto_ou_ausente"]) + self.assertEqual(model_router._staleness(row, NOW, Decimal(3), Decimal(30), TOKENS), []) + + +def cost(row, tokens=TOKENS): + return model_router._native_cost_and_headroom(row, tokens) + + +class NativeCostTests(unittest.TestCase): + def zen(self, *edits, billing="zen_balance"): + return mutated(candidate("opencode/m", billing=billing), *edits) + + def go(self, *edits, **extra): + remaining = {"5h": "10", "weekly": "20", "monthly": "30"} + debit = {"5h": "1", "weekly": "2", "monthly": "3"} + row = candidate("opencode/g", billing="go_quota", quota_window_remaining=remaining, quota_window_debit=debit, **extra) + return mutated(row, *edits) + + def test_balance_main_path_and_exact_balance(self): + amount, headroom, reason = cost(self.zen()) + self.assertEqual((amount, reason), (Decimal("0.0001") + Decimal("0.00002"), None)) + self.assertEqual(headroom, Decimal(2) - amount) + exact = self.zen((("native_tariff", "available_balance_usd"), str(amount))) + self.assertEqual(cost(exact), (amount, Decimal(0), None)) + self.assertEqual(cost(self.zen(billing="omp_payg"))[2], None) + + def test_balance_decisions(self): + cases = [ + ("no token profile", self.zen(), {}, (None, None, "perfil_tokens_native_nao_configurado")), + ("tariff missing", self.zen((("native_tariff",), None)), TOKENS, (None, None, "tarifa_native_nao_confirmada")), + ("rates not dict", self.zen((("native_tariff", "price_per_token_usd"), None)), TOKENS, + (None, None, "tarifa_native_nao_confirmada")), + ("not confirmed", self.zen((("native_tariff", "account_confirmed"), False)), TOKENS, + (None, None, "tarifa_native_nao_confirmada")), + ("rates incomplete", self.zen((("native_tariff", "price_per_token_usd"), {"input": "1"})), TOKENS, + (None, None, "tarifa_native_incompleta_para_perfil")), + ("balance not confirmed", self.zen((("availability", "balance_confirmed"), None)), TOKENS, + (None, None, "saldo_nao_confirmado")), + ("balance unknown", self.zen((("native_tariff", "available_balance_usd"), None)), TOKENS, + (None, None, "saldo_desconhecido")), + ("balance negative", self.zen((("native_tariff", "available_balance_usd"), "-1")), TOKENS, + (None, None, "saldo_desconhecido")), + ("balance short", self.zen((("native_tariff", "available_balance_usd"), "0.00001")), TOKENS, + (None, Decimal("0.00001"), "saldo_insuficiente")), + ] + for name, row, tokens, expected in cases: + with self.subTest(name): + self.assertEqual(cost(row, tokens), expected) + + def test_go_main_path_uses_monthly_debit_and_smallest_headroom(self): + self.assertEqual(cost(self.go()), (Decimal(3), Decimal(9), None)) + + def test_go_scalar_debit_applies_to_every_window(self): + row = self.go((("quota_window_debit",), "5")) + self.assertEqual(cost(row), (Decimal(5), Decimal(5), None)) + + def test_go_decisions(self): + cases = [ + ("no token profile", self.go(), {}, (None, None, "perfil_tokens_go_nao_configurado")), + ("balance fallback", self.go((("use_balance_fallback",), True)), TOKENS, + (None, None, "fallback_go_para_saldo_zen_nao_autorizado")), + ("remaining not dict", self.go((("quota_window_remaining",), None)), TOKENS, + (None, None, "quota_go_nao_confirmada")), + ("debit missing", self.go((("quota_window_debit",), None)), TOKENS, (None, None, "quota_go_nao_confirmada")), + ("window unknown", self.go((("quota_window_remaining", "weekly"), DELETE)), TOKENS, + (None, None, "janela_go_weekly_desconhecida")), + ("debit negative", self.go((("quota_window_debit", "5h"), "-1")), TOKENS, + (None, None, "janela_go_5h_desconhecida")), + ("insufficient", self.go((("quota_window_debit", "monthly"), "31")), TOKENS, + (None, None, "quota_go_monthly_insuficiente")), + ] + for name, row, tokens, expected in cases: + with self.subTest(name): + self.assertEqual(cost(row, tokens), expected) + + def test_claude_subscription_and_unknown_modes(self): + sub = candidate("claude/s", billing="claude_subscription", subscription_consumption_usd="1.5") + self.assertEqual(cost(sub), (Decimal("1.5"), None, None)) + unmeasured = candidate("claude/s", billing="claude_subscription") + self.assertEqual(cost(unmeasured), (None, None, "consumo_assinatura_nao_mensuravel")) + self.assertEqual(cost(candidate("x/y", billing="mystery")), (None, None, "modo_de_cobranca_desconhecido")) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/model_catalog.py b/scripts/model_catalog.py index a3ceee5..bf41a29 100755 --- a/scripts/model_catalog.py +++ b/scripts/model_catalog.py @@ -338,72 +338,98 @@ def reference_cost(price_per_token_usd: dict[str, Any], tokens: dict[str, Any]) return total -def validate_aa_snapshot(payload: Any) -> list[dict[str, Any]]: - """Aceita exclusivamente export manual auditável do Coding Agent Index v1.5.""" - if not isinstance(payload, dict): - raise ValueError("snapshot AA deve ser objeto") - allowed_root = {"metric_id", "version", "ingestion", "terms_accepted", "source_url", "attribution", "captured_at", "records"} - if set(payload) - allowed_root: - raise ValueError("snapshot AA contém campo não permitido") - if payload.get("metric_id") != AA_METRIC or str(payload.get("version")) != AA_VERSION: - raise ValueError("somente aa_coding_agent_index v1.5 é aceito; índices não são intercambiáveis") +_AA_ROOT_FIELDS = {"metric_id", "version", "ingestion", "terms_accepted", "source_url", "attribution", "captured_at", "records"} +_AA_RECORD_FIELDS = {"metric_id", "version", "provider", "model_id", "variant", "score", "agent", "harness", + "settings", "match_grade", "evidence_url", "components", "fallback", "effort", "reasoning", "configuration"} +_AA_PROVENANCE = ("agent", "harness", "settings", "evidence_url") + + +def _check_aa_provenance(payload: dict[str, Any]) -> None: + """Consentimento, fonte oficial, atribuição e data de captura do export AA.""" if payload.get("ingestion") != "manual_audited" or payload.get("terms_accepted") is not True: raise ValueError("snapshot precisa ser importação manual consentida (terms_accepted=true)") - source_url = payload.get("source_url") - if not _secure_url(source_url, "artificialanalysis.ai"): + if not _secure_url(payload.get("source_url"), "artificialanalysis.ai"): raise ValueError("source_url oficial da Artificial Analysis é obrigatório") if not isinstance(payload.get("attribution"), str) or not payload["attribution"]: raise ValueError("atribuição AA obrigatória") - captured_at = payload.get("captured_at") - captured_time = _parse_utc(captured_at) + captured_time = _parse_utc(payload.get("captured_at")) if captured_time is None or captured_time > dt.datetime.now(dt.timezone.utc): raise ValueError("captured_at obrigatório, UTC/offset-aware e não futuro") + + +def _aa_records(payload: Any) -> list[Any]: + """Valida o cabeçalho do export AA (índice e proveniência) e devolve os registros.""" + if not isinstance(payload, dict): + raise ValueError("snapshot AA deve ser objeto") + if set(payload) - _AA_ROOT_FIELDS: + raise ValueError("snapshot AA contém campo não permitido") + if payload.get("metric_id") != AA_METRIC or str(payload.get("version")) != AA_VERSION: + raise ValueError("somente aa_coding_agent_index v1.5 é aceito; índices não são intercambiáveis") + _check_aa_provenance(payload) records = payload.get("records") if not isinstance(records, list): raise ValueError("records deve ser lista") + return records + + +def _aa_record_key(record: Any) -> tuple[str, str, str | None]: + """Valida a estrutura e o mapeamento exato de um registro AA; devolve (provider, model_id, variant).""" + if not isinstance(record, dict) or set(record) - _AA_RECORD_FIELDS: + raise ValueError("registro AA inválido ou contém campo não permitido") + if record.get("metric_id", AA_METRIC) != AA_METRIC or str(record.get("version", AA_VERSION)) != AA_VERSION: + raise ValueError("índice/versão diferente não pode ser misturado") + provider, model_id = record.get("provider"), record.get("model_id") + if not _is_exact_model_ref(provider, model_id): + raise ValueError("mapeamento provider/model_id exato obrigatório") + variant = record.get("variant") + if variant is not None and (not isinstance(variant, str) or not variant): + raise ValueError("variante AA inválida") + return provider, model_id, variant + + +def _aa_record_score(record: dict[str, Any]) -> Decimal: + """Valida nota, grau de correspondência e proveniência de um registro AA; devolve a nota.""" + score = decimal_value(record.get("score")) + if score is None or score < 0 or score > 100: + raise ValueError("score AA precisa estar no intervalo 0..100") + if record.get("match_grade") not in ("exact", "proxy"): + raise ValueError("match_grade deve ser exact ou proxy, sem fuzzy match") + for field in _AA_PROVENANCE: + if field not in record or record[field] in (None, ""): + raise ValueError(f"campo de proveniência AA obrigatório: {field}") + if not _secure_url(record["evidence_url"], "artificialanalysis.ai"): + raise ValueError("evidence_url AA inválida") + return score + + +def validate_aa_snapshot(payload: Any) -> list[dict[str, Any]]: + """Aceita exclusivamente export manual auditável do Coding Agent Index v1.5.""" + records = _aa_records(payload) validated = [] seen_keys: set[tuple[str, str, str | None]] = set() - allowed_record = {"metric_id", "version", "provider", "model_id", "variant", "score", "agent", "harness", - "settings", "match_grade", "evidence_url", "components", "fallback", "effort", "reasoning", "configuration"} for record in records: - if not isinstance(record, dict) or set(record) - allowed_record: - raise ValueError("registro AA inválido ou contém campo não permitido") - if record.get("metric_id", AA_METRIC) != AA_METRIC or str(record.get("version", AA_VERSION)) != AA_VERSION: - raise ValueError("índice/versão diferente não pode ser misturado") - provider, model_id = record.get("provider"), record.get("model_id") - if not isinstance(provider, str) or not isinstance(model_id, str) or not model_id or "/" in model_id or "#" in model_id: - raise ValueError("mapeamento provider/model_id exato obrigatório") - variant = record.get("variant") - if variant is not None and (not isinstance(variant, str) or not variant): - raise ValueError("variante AA inválida") - key = (provider, model_id, variant) + key = _aa_record_key(record) if key in seen_keys: raise ValueError("benchmark inconsistente: modelo/variante duplicado no snapshot") seen_keys.add(key) - score = decimal_value(record.get("score")) - if score is None or score < 0 or score > 100: - raise ValueError("score AA precisa estar no intervalo 0..100") - grade = record.get("match_grade") - if grade not in ("exact", "proxy"): - raise ValueError("match_grade deve ser exact ou proxy, sem fuzzy match") - for key in ("agent", "harness", "settings", "evidence_url"): - if key not in record or record[key] in (None, ""): - raise ValueError(f"campo de proveniência AA obrigatório: {key}") - if not _secure_url(record["evidence_url"], "artificialanalysis.ai"): - raise ValueError("evidence_url AA inválida") + score = _aa_record_score(record) validated.append({ **record, "metric_id": AA_METRIC, "version": AA_VERSION, "score": decimal_text(score), - "source_url": source_url, "attribution": payload["attribution"], "observed_at": captured_at, + "source_url": payload["source_url"], "attribution": payload["attribution"], "observed_at": payload["captured_at"], "snapshot_hash": sha256(canonical_json(payload)), }) return validated -def validate_availability_snapshot(payload: Any) -> dict[tuple[str, str, str | None], dict[str, Any]]: - """Validates explicit account evidence without accepting or retaining secrets.""" +_AVAILABILITY_FIELDS = {"provider", "model_id", "variant", "local_listed", "authenticated", "entitled", + "verified_at", "source", "balance_confirmed", "balance_verified_at"} +_AVAILABILITY_REQUIRED = ("local_listed", "authenticated", "entitled", "verified_at", "source") + + +def _availability_records(payload: Any) -> list[Any]: + """Valida o cabeçalho do snapshot de disponibilidade e devolve a lista de registros.""" if not isinstance(payload, dict) or payload.get("tool") not in ("opencode", "omp") or payload.get("ingestion") != "manual_audited": raise ValueError("availability precisa ser snapshot manual auditado de opencode ou omp") - tool = payload["tool"] if payload.get("user_confirmed") is not True: raise ValueError("snapshot de disponibilidade exige user_confirmed=true") captured_at = payload.get("captured_at") @@ -418,100 +444,148 @@ def validate_availability_snapshot(payload: Any) -> dict[tuple[str, str, str | N records = payload.get("records") if not isinstance(records, list): raise ValueError("records de disponibilidade deve ser lista") + return records + + +def _is_exact_model_ref(provider: Any, model_id: Any) -> bool: + return isinstance(provider, str) and isinstance(model_id, str) and bool(model_id) and "/" not in model_id and "#" not in model_id + + +def _availability_key(record: Any) -> tuple[str, str, str | None]: + """Valida a estrutura de um registro e devolve a identidade (provider, model_id, variant).""" + if not isinstance(record, dict) or set(record) - _AVAILABILITY_FIELDS: + raise ValueError("snapshot de disponibilidade contém campos não permitidos") + provider, model_id, variant = record.get("provider"), record.get("model_id"), record.get("variant") + if not _is_exact_model_ref(provider, model_id) or (variant is not None and not isinstance(variant, str)): + raise ValueError("provider/model_id/variant exatos são obrigatórios") + if any(field not in record for field in _AVAILABILITY_REQUIRED): + raise ValueError("snapshot deve declarar todos os campos de disponibilidade") + return provider, model_id, variant + + +def _check_availability_balance(record: dict[str, Any], now: dt.datetime) -> None: + balance_confirmed = record.get("balance_confirmed") + if balance_confirmed is not None and not isinstance(balance_confirmed, bool): + raise ValueError("balance_confirmed deve ser bool ou null") + balance_verified_at = record.get("balance_verified_at") + balance_time = _parse_utc(balance_verified_at) if balance_verified_at else None + if balance_time is not None and (balance_time > now or balance_confirmed is not True): + raise ValueError("balance_verified_at requer confirmação explícita e não pode ser futuro") + if balance_confirmed is True and balance_time is None: + raise ValueError("saldo confirmado exige balance_verified_at UTC válido") + + +def _check_availability_evidence(record: dict[str, Any]) -> None: + """Valida flags booleanas, saldo confirmado e a evidência (verified_at/source) de um registro.""" + now = dt.datetime.now(dt.timezone.utc) + for field in ("local_listed", "authenticated", "entitled"): + if record[field] is not None and not isinstance(record[field], bool): + raise ValueError(f"{field} deve ser bool ou null") + _check_availability_balance(record, now) + verified_at = _parse_utc(record["verified_at"]) + if verified_at is None or verified_at > now or not isinstance(record["source"], str) or not record["source"]: + raise ValueError("source e verified_at UTC válido são necessários para auditar a disponibilidade") + + +def validate_availability_snapshot(payload: Any) -> dict[tuple[str, str, str | None], dict[str, Any]]: + """Validates explicit account evidence without accepting or retaining secrets.""" + records = _availability_records(payload) result = {} source_hash = sha256(canonical_json(payload)) - allowed_fields = {"provider", "model_id", "variant", "local_listed", "authenticated", "entitled", - "verified_at", "source", "balance_confirmed", "balance_verified_at"} for record in records: - if not isinstance(record, dict) or set(record) - allowed_fields: - raise ValueError("snapshot de disponibilidade contém campos não permitidos") - provider, model_id, variant = record.get("provider"), record.get("model_id"), record.get("variant") - if not isinstance(provider, str) or not isinstance(model_id, str) or not model_id or "/" in model_id or "#" in model_id or (variant is not None and not isinstance(variant, str)): - raise ValueError("provider/model_id/variant exatos são obrigatórios") - if any(field not in record for field in ("local_listed", "authenticated", "entitled", "verified_at", "source")): - raise ValueError("snapshot deve declarar todos os campos de disponibilidade") - key = (provider, model_id, variant) + key = _availability_key(record) if key in result: raise ValueError("snapshot de disponibilidade contém ID/variante duplicado") - for field in ("local_listed", "authenticated", "entitled"): - if record[field] is not None and not isinstance(record[field], bool): - raise ValueError(f"{field} deve ser bool ou null") - balance_confirmed = record.get("balance_confirmed") - if balance_confirmed is not None and not isinstance(balance_confirmed, bool): - raise ValueError("balance_confirmed deve ser bool ou null") - balance_verified_at = record.get("balance_verified_at") - balance_time = _parse_utc(balance_verified_at) if balance_verified_at else None - if balance_time is not None and (balance_time > dt.datetime.now(dt.timezone.utc) or balance_confirmed is not True): - raise ValueError("balance_verified_at requer confirmação explícita e não pode ser futuro") - if balance_confirmed is True and balance_time is None: - raise ValueError("saldo confirmado exige balance_verified_at UTC válido") - verified_at = _parse_utc(record["verified_at"]) - if verified_at is None or verified_at > dt.datetime.now(dt.timezone.utc) or not isinstance(record["source"], str) or not record["source"]: - raise ValueError("source e verified_at UTC válido são necessários para auditar a disponibilidade") + _check_availability_evidence(record) result[key] = { "local_listed": record["local_listed"], "authenticated": record["authenticated"], "entitled": record["entitled"], "verified_at": record["verified_at"], "source": "manual_audited", "source_hash": source_hash, - "balance_confirmed": balance_confirmed, "balance_verified_at": balance_verified_at, + "balance_confirmed": record.get("balance_confirmed"), "balance_verified_at": record.get("balance_verified_at"), } return result +_NATIVE_ALLOWED = {"billing_mode", "price_per_token_usd", "quota_window_remaining", "quota_window_debit", + "use_balance_fallback", "source_url", "captured_at", "account_confirmed", "available_balance_usd"} +_NATIVE_BILLING_MODES = ("zen_balance", "go_quota", "omp_payg") +_NATIVE_RATE_KEYS = ("input", "output", "cache_read", "cache_write", "cache_write_5m", "cache_write_1h", "cache_write_24h") +_QUOTA_WINDOWS = ("5h", "weekly", "monthly") + + +def _check_native_header(model_key: Any, value: Any) -> str: + """Valida estrutura, confirmação, modo, URL e data comuns a todo override; devolve o modo.""" + if not isinstance(model_key, str) or not isinstance(value, dict) or set(value) - _NATIVE_ALLOWED: + raise ValueError("override native contém estrutura/campo inválido") + if value.get("account_confirmed") is not True: + raise ValueError("tarifa/cota native exige confirmação explícita da conta") + mode = value.get("billing_mode") + if mode not in _NATIVE_BILLING_MODES: + raise ValueError("billing_mode native inválido") + if not _secure_url(value.get("source_url")): + raise ValueError("source_url HTTPS de tarifa/cota native obrigatório") + try: + captured = dt.datetime.fromisoformat(value.get("captured_at").replace("Z", "+00:00")) + except (AttributeError, ValueError) as exc: + raise ValueError("captured_at de tarifa/cota native inválido") from exc + if captured.tzinfo is None or captured > dt.datetime.now(dt.timezone.utc): + raise ValueError("captured_at de tarifa/cota native precisa ser aware e não futuro") + return mode + + +def _non_negative_text(text: str | None) -> bool: + return text is not None and Decimal(text) >= 0 + + +def _native_balance_fields(value: dict[str, Any]) -> dict[str, Any]: + """Campos de tarifa por token e saldo (zen_balance e omp_payg).""" + rates = value.get("price_per_token_usd") + if not isinstance(rates, dict) or not rates: + raise ValueError("tarifa nativa precisa de price_per_token_usd") + prices = {k: decimal_text(decimal_value(v)) for k, v in rates.items() if k in _NATIVE_RATE_KEYS} + if len(prices) != len(rates) or not all(_non_negative_text(v) for v in prices.values()): + raise ValueError("componentes de tarifa nativa inválidos") + balance = decimal_value(value.get("available_balance_usd")) + if balance is None or balance < 0: + raise ValueError("Zen/OMP payg exige saldo disponível confirmado") + return {"price_per_token_usd": prices, "available_balance_usd": decimal_text(balance)} + + +def _quota_window_texts(source: dict[str, Any]) -> dict[str, str | None]: + return {k: decimal_text(decimal_value(source[k])) for k in _QUOTA_WINDOWS} + + +def _native_quota_fields(value: dict[str, Any]) -> dict[str, Any]: + """Campos de cota por janela e fallback de saldo (go_quota).""" + remaining, debit = value.get("quota_window_remaining"), value.get("quota_window_debit") + if not isinstance(remaining, dict) or not isinstance(debit, dict) or not set(_QUOTA_WINDOWS) <= set(remaining) & set(debit): + raise ValueError("cota Go exige valores por modelo em 5h/weekly/monthly") + fields = { + "quota_window_remaining": _quota_window_texts(remaining), + "quota_window_debit": _quota_window_texts(debit), + } + if not all(_non_negative_text(v) for window_values in fields.values() for v in window_values.values()): + raise ValueError("quota Go precisa ter todos os valores decimais não negativos") + fallback = value.get("use_balance_fallback", False) + if not isinstance(fallback, bool): + raise ValueError("use_balance_fallback precisa ser bool") + fields["use_balance_fallback"] = fallback + return fields + def validate_native_tariff_overrides(payload: Any) -> dict[str, dict[str, Any]]: """Valida cobrança nativa/quotas manuais; preços OpenRouter nunca são tarifa.""" if not isinstance(payload, dict): raise ValueError("native_tariffs deve ser objeto JSON") result = {} - allowed = {"billing_mode", "price_per_token_usd", "quota_window_remaining", "quota_window_debit", - "use_balance_fallback", "source_url", "captured_at", "account_confirmed", "available_balance_usd"} for model_key, value in payload.items(): - if not isinstance(model_key, str) or not isinstance(value, dict) or set(value) - allowed: - raise ValueError("override native contém estrutura/campo inválido") - if value.get("account_confirmed") is not True: - raise ValueError("tarifa/cota native exige confirmação explícita da conta") - mode = value.get("billing_mode") - if mode not in ("zen_balance", "go_quota", "omp_payg"): - raise ValueError("billing_mode native inválido") - if not _secure_url(value.get("source_url")): - raise ValueError("source_url HTTPS de tarifa/cota native obrigatório") - captured_at = value.get("captured_at") - try: - captured = dt.datetime.fromisoformat(captured_at.replace("Z", "+00:00")) - except (AttributeError, ValueError) as exc: - raise ValueError("captured_at de tarifa/cota native inválido") from exc - if captured.tzinfo is None or captured > dt.datetime.now(dt.timezone.utc): - raise ValueError("captured_at de tarifa/cota native precisa ser aware e não futuro") + mode = _check_native_header(model_key, value) record = {k: value[k] for k in ("billing_mode", "source_url", "captured_at", "account_confirmed")} - if mode in ("zen_balance", "omp_payg"): - rates = value.get("price_per_token_usd") - if not isinstance(rates, dict) or not rates: - raise ValueError("tarifa nativa precisa de price_per_token_usd") - record["price_per_token_usd"] = { - k: decimal_text(decimal_value(v)) for k, v in rates.items() - if k in ("input", "output", "cache_read", "cache_write", "cache_write_5m", "cache_write_1h", "cache_write_24h") - } - if len(record["price_per_token_usd"]) != len(rates) or any(v is None or Decimal(v) < 0 for v in record["price_per_token_usd"].values()): - raise ValueError("componentes de tarifa nativa inválidos") - balance = decimal_value(value.get("available_balance_usd")) - if balance is None or balance < 0: - raise ValueError("Zen/OMP payg exige saldo disponível confirmado") - record["available_balance_usd"] = decimal_text(balance) - else: - remaining, debit = value.get("quota_window_remaining"), value.get("quota_window_debit") - if not isinstance(remaining, dict) or not isinstance(debit, dict) or any(k not in remaining or k not in debit for k in ("5h", "weekly", "monthly")): - raise ValueError("cota Go exige valores por modelo em 5h/weekly/monthly") - record["quota_window_remaining"] = {k: decimal_text(decimal_value(remaining[k])) for k in ("5h", "weekly", "monthly")} - record["quota_window_debit"] = {k: decimal_text(decimal_value(debit[k])) for k in ("5h", "weekly", "monthly")} - if any(record[key][window] is None or Decimal(record[key][window]) < 0 for key in ("quota_window_remaining", "quota_window_debit") for window in ("5h", "weekly", "monthly")): - raise ValueError("quota Go precisa ter todos os valores decimais não negativos") - fallback = value.get("use_balance_fallback", False) - if not isinstance(fallback, bool): - raise ValueError("use_balance_fallback precisa ser bool") - record["use_balance_fallback"] = fallback + record.update(_native_quota_fields(value) if mode == "go_quota" else _native_balance_fields(value)) result[model_key] = record return result + def discover_local_models(provider: str, executable: str = "opencode", runner: Callable[..., Any] = subprocess.run) -> dict[str, Any]: """Lista IDs pela CLI sem persistir stdout bruto, stderr, environment ou credenciais.""" try: diff --git a/scripts/model_router.py b/scripts/model_router.py index c848c15..91e5d15 100755 --- a/scripts/model_router.py +++ b/scripts/model_router.py @@ -79,103 +79,160 @@ def _has_token_profile(tokens: dict[str, Any]) -> bool: return any((_decimal(tokens.get(key)) or Decimal(0)) > 0 for key in ("input", "output", "cache_read", "cache_write")) -def _staleness(row: dict[str, Any], now: dt.datetime, max_price_age_hours: Decimal, max_score_age_days: Decimal, - tokens: dict[str, Any]) -> list[str]: +_PAID_MODES = ("zen_balance", "go_quota", "omp_payg", "claude_subscription") +_BALANCE_MODES = ("zen_balance", "omp_payg") +_SIXTY_SECONDS = dt.timedelta(seconds=60) +_DAY = dt.timedelta(hours=24) + + +def _expired(moment: dt.datetime | None, now: dt.datetime, limit: dt.timedelta) -> bool: + """Ausente, no futuro ou mais antigo que o limite.""" + return moment is None or moment > now or now - moment > limit + + +def _is_https(value: Any) -> bool: + return str(value).startswith("https://") + + +def _native_tariff_stale(native: dict[str, Any], now: dt.datetime, limit: dt.timedelta) -> bool: + return (native.get("account_confirmed") is not True or not _is_https(native.get("source_url", "")) + or _expired(_time(native.get("captured_at")), now, limit)) + + +def _openrouter_record(row: dict[str, Any]) -> dict[str, Any]: + return row.get("openrouter") or {"price_per_token_usd": row.get("or_price_per_token") or {}} + + +def _go_quota_stale(row: dict[str, Any], now: dt.datetime) -> bool: + return (_expired(_time(row.get("quota_verified_at")), now, _SIXTY_SECONDS) + or not _is_https(row.get("quota_source_url", ""))) + + +def _reference_price_stale(price: dict[str, Any], now: dt.datetime, limit: dt.timedelta) -> bool: + """Só há o que revalidar quando existe alguma tarifa referencial.""" + rates = price.get("price_per_token_usd", {}) + has_reference_price = any(rates.get(k) is not None for k in ("input", "output", "cache_read", "cache_write")) + return has_reference_price and _expired(_time(price.get("captured_at")), now, limit) + + +def _availability_reasons(row: dict[str, Any], mode: Any, now: dt.datetime) -> list[str]: reasons: list[str] = [] - mode = row.get("billing_mode") - paid = mode in ("zen_balance", "go_quota", "omp_payg", "claude_subscription") - price_record = row.get("openrouter") or {"price_per_token_usd": row.get("or_price_per_token") or {}} - price = openrouter_price_for_profile(price_record, tokens) - native = row.get("native_tariff") or {} - has_reference_price = any(price.get("price_per_token_usd", {}).get(k) is not None - for k in ("input", "output", "cache_read", "cache_write")) - if paid: - if has_reference_price: - price_time = _time(price.get("captured_at")) - if price_time is None or price_time > now or (now - price_time).total_seconds() > float(max_price_age_hours * Decimal(3600)): - reasons.append("preco_referencial_obsoleto_ou_ausente") - score = row.get("score") or {} - score_time = _time(score.get("observed_at")) - if score_time is None or score_time > now or (now - score_time).total_seconds() > float(max_score_age_days * Decimal(86400)): - reasons.append("benchmark_obsoleto_ou_ausente") - if mode in ("zen_balance", "omp_payg"): - native_time = _time(native.get("captured_at")) - if native.get("account_confirmed") is not True or not str(native.get("source_url", "")).startswith("https://") or native_time is None or native_time > now or now - native_time > dt.timedelta(hours=float(max_price_age_hours)): - reasons.append("tarifa_nativa_obsoleta_ou_nao_confirmada") - if mode == "go_quota": - quota_time = _time(row.get("quota_verified_at")) - if quota_time is None or quota_time > now or now - quota_time > dt.timedelta(seconds=60) or not str(row.get("quota_source_url", "")).startswith("https://"): - reasons.append("quota_go_obsoleta_ou_nao_confirmada") - availability = row.get("availability") or {} - local_time = _time(availability.get("local_verified_at") or availability.get("verified_at")) - if local_time is None or local_time > now or now - local_time > dt.timedelta(hours=24): - reasons.append("disponibilidade_local_precisa_revalidacao") - account_time = _time(availability.get("account_verified_at")) - if availability.get("authenticated") is True and availability.get("entitled") is True: - if account_time is None or account_time > now or now - account_time > dt.timedelta(hours=24): - reasons.append("entitlement_precisa_revalidacao") - if mode in ("zen_balance", "omp_payg"): - balance_time = _time(availability.get("balance_verified_at")) - if availability.get("balance_confirmed") is not True or balance_time is None or balance_time > now or now - balance_time > dt.timedelta(seconds=60): - reasons.append("saldo_nao_confirmado_ou_obsoleto") + availability = row.get("availability") or {} + if _expired(_time(availability.get("local_verified_at") or availability.get("verified_at")), now, _DAY): + reasons.append("disponibilidade_local_precisa_revalidacao") + if (availability.get("authenticated") is True and availability.get("entitled") is True + and _expired(_time(availability.get("account_verified_at")), now, _DAY)): + reasons.append("entitlement_precisa_revalidacao") + if mode in _BALANCE_MODES and (availability.get("balance_confirmed") is not True + or _expired(_time(availability.get("balance_verified_at")), now, _SIXTY_SECONDS)): + reasons.append("saldo_nao_confirmado_ou_obsoleto") return reasons +def _staleness(row: dict[str, Any], now: dt.datetime, max_price_age_hours: Decimal, max_score_age_days: Decimal, + tokens: dict[str, Any]) -> list[str]: + mode = row.get("billing_mode") + price = openrouter_price_for_profile(_openrouter_record(row), tokens) + if mode not in _PAID_MODES: + return [] + reasons: list[str] = [] + price_limit = dt.timedelta(seconds=float(max_price_age_hours * Decimal(3600))) + if _reference_price_stale(price, now, price_limit): + reasons.append("preco_referencial_obsoleto_ou_ausente") + score_limit = dt.timedelta(seconds=float(max_score_age_days * Decimal(86400))) + if _expired(_time((row.get("score") or {}).get("observed_at")), now, score_limit): + reasons.append("benchmark_obsoleto_ou_ausente") + if mode in _BALANCE_MODES and _native_tariff_stale(row.get("native_tariff") or {}, now, price_limit): + reasons.append("tarifa_nativa_obsoleta_ou_nao_confirmada") + if mode == "go_quota" and _go_quota_stale(row, now): + reasons.append("quota_go_obsoleta_ou_nao_confirmada") + return reasons + _availability_reasons(row, mode, now) + + def _profile_cost(row: dict[str, Any], tokens: dict[str, Any]) -> Decimal | None: - record = row.get("openrouter") or {"price_per_token_usd": row.get("or_price_per_token") or {}} - profile = openrouter_price_for_profile(record, tokens) + profile = openrouter_price_for_profile(_openrouter_record(row), tokens) return reference_cost(profile.get("price_per_token_usd") or {}, tokens) -def _native_cost_and_headroom(row: dict[str, Any], tokens: dict[str, Any]) -> tuple[Decimal | None, Decimal | None, str | None]: - mode = row.get("billing_mode") +_CostResult = tuple[Decimal | None, Decimal | None, str | None] + + +def _has_confirmed_rates(native: Any) -> bool: + return (isinstance(native, dict) and isinstance(native.get("price_per_token_usd"), dict) + and native.get("account_confirmed") is True) + + +def _balance_cost(row: dict[str, Any], tokens: dict[str, Any]) -> _CostResult: + """zen_balance e omp_payg: custo pela tarifa nativa confirmada e folga de saldo.""" native = row.get("native_tariff") - if mode in ("zen_balance", "omp_payg"): - if not _has_token_profile(tokens): - return None, None, "perfil_tokens_native_nao_configurado" - if not isinstance(native, dict) or not isinstance(native.get("price_per_token_usd"), dict) or native.get("account_confirmed") is not True: - return None, None, "tarifa_native_nao_confirmada" - amount = reference_cost(native["price_per_token_usd"], tokens) - if amount is None: - return None, None, "tarifa_native_incompleta_para_perfil" - balance = _decimal(native.get("available_balance_usd")) - availability = row.get("availability") or {} - if availability.get("balance_confirmed") is not True: - return None, None, "saldo_nao_confirmado" - if balance is None or balance < 0: - return None, None, "saldo_desconhecido" - if balance < amount: - return None, balance, "saldo_insuficiente" - return amount, balance - amount, None - if mode == "go_quota": - if not _has_token_profile(tokens): - return None, None, "perfil_tokens_go_nao_configurado" - remaining = row.get("quota_window_remaining") - debit = row.get("quota_window_debit") - if row.get("use_balance_fallback"): - return None, None, "fallback_go_para_saldo_zen_nao_autorizado" - if not isinstance(remaining, dict) or debit is None: - return None, None, "quota_go_nao_confirmada" - debit_map = debit if isinstance(debit, dict) else {window: debit for window in WINDOWS} - headrooms: list[Decimal] = [] - for window in WINDOWS: - available = _decimal(remaining.get(window)) - used = _decimal(debit_map.get(window)) - if available is None or used is None or available < 0 or used < 0: - return None, None, f"janela_go_{window}_desconhecida" - if used > available: - return None, None, f"quota_go_{window}_insuficiente" - headrooms.append(available - used) - representative = _decimal(debit_map.get("monthly")) - if representative is None: - return None, min(headrooms), "debito_go_nao_confirmado" - return representative, min(headrooms), None - if mode == "claude_subscription": - amount = _decimal(row.get("subscription_consumption_usd")) - if amount is None: - return None, None, "consumo_assinatura_nao_mensuravel" - return amount, None, None - return None, None, "modo_de_cobranca_desconhecido" + if not _has_token_profile(tokens): + return None, None, "perfil_tokens_native_nao_configurado" + if not _has_confirmed_rates(native): + return None, None, "tarifa_native_nao_confirmada" + amount = reference_cost(native["price_per_token_usd"], tokens) + if amount is None: + return None, None, "tarifa_native_incompleta_para_perfil" + balance = _decimal(native.get("available_balance_usd")) + if (row.get("availability") or {}).get("balance_confirmed") is not True: + return None, None, "saldo_nao_confirmado" + if balance is None or balance < 0: + return None, None, "saldo_desconhecido" + if balance < amount: + return None, balance, "saldo_insuficiente" + return amount, balance - amount, None + + +def _go_window_headrooms(remaining: dict[str, Any], debit_map: dict[str, Any]) -> tuple[list[Decimal], str | None]: + """Folga por janela; no primeiro problema devolve a razão de recusa.""" + headrooms: list[Decimal] = [] + for window in WINDOWS: + available = _decimal(remaining.get(window)) + used = _decimal(debit_map.get(window)) + if available is None or used is None or available < 0 or used < 0: + return headrooms, f"janela_go_{window}_desconhecida" + if used > available: + return headrooms, f"quota_go_{window}_insuficiente" + headrooms.append(available - used) + return headrooms, None + + +def _go_quota_cost(row: dict[str, Any], tokens: dict[str, Any]) -> _CostResult: + """go_quota: débito mensal como custo representativo e a menor folga entre as janelas.""" + if not _has_token_profile(tokens): + return None, None, "perfil_tokens_go_nao_configurado" + remaining = row.get("quota_window_remaining") + debit = row.get("quota_window_debit") + if row.get("use_balance_fallback"): + return None, None, "fallback_go_para_saldo_zen_nao_autorizado" + if not isinstance(remaining, dict) or debit is None: + return None, None, "quota_go_nao_confirmada" + debit_map = debit if isinstance(debit, dict) else {window: debit for window in WINDOWS} + headrooms, reason = _go_window_headrooms(remaining, debit_map) + if reason: + return None, None, reason + representative = _decimal(debit_map.get("monthly")) + if representative is None: + return None, min(headrooms), "debito_go_nao_confirmado" + return representative, min(headrooms), None + + +def _subscription_cost(row: dict[str, Any], tokens: dict[str, Any]) -> _CostResult: + amount = _decimal(row.get("subscription_consumption_usd")) + if amount is None: + return None, None, "consumo_assinatura_nao_mensuravel" + return amount, None, None + + +_COST_BY_MODE = {"zen_balance": _balance_cost, "omp_payg": _balance_cost, + "go_quota": _go_quota_cost, "claude_subscription": _subscription_cost} + + +def _native_cost_and_headroom(row: dict[str, Any], tokens: dict[str, Any]) -> _CostResult: + mode = row.get("billing_mode") + handler = _COST_BY_MODE.get(mode) if isinstance(mode, str) else None + if handler is None: + return None, None, "modo_de_cobranca_desconhecido" + return handler(row, tokens) def route(candidates: Iterable[dict[str, Any]], role: str, *, allowed_providers: Iterable[str] = (),