#!/usr/bin/env python3
"""Isolated Otari cost-core replay; synthetic prices, no SDK carrier or gateway."""
import importlib.util
import json
import logging
from pathlib import Path
import hashlib
import sys
import types
from decimal import Decimal as D
from types import SimpleNamespace as NS

root = Path(sys.argv[1]).resolve() / "src"
hashes = {
    "metered_pricing.py": "910d93f1a755b8ca2c9b89ec2e9bc46664cd34c39b67166870637f9ec6765128",
    "usage.py": "71a0fef861df23e875136f9e3767433976a2941b4603ee4ecba25f14cd27d9d0",
}
for name, expected in hashes.items():
    assert hashlib.sha256((root / "gateway/core" / name).read_bytes()).hexdigest() == expected, name

# Only import scaffolding is replaced. The SDK type cannot be instantiated,
# and the actual usage helper and pricing module execute unchanged.
for name in ["any_llm", "any_llm.types", "any_llm.types.completion", "gateway", "gateway.core", "gateway.log_config"]:
    module = types.ModuleType(name)
    module.__path__ = []
    sys.modules[name] = module

class AnnotationOnlyUsage:
    def __init__(self, *args, **kwargs):
        raise RuntimeError("The real SDK carrier is outside this isolated probe")

sys.modules["any_llm.types.completion"].CompletionUsage = AnnotationOnlyUsage
logger = logging.getLogger("otari-isolated-pricing")
sys.modules["gateway.log_config"].logger = logger
for name in ["usage", "metered_pricing"]:
    full_name = "gateway.core." + name
    spec = importlib.util.spec_from_file_location(full_name, root / "gateway/core" / f"{name}.py")
    module = importlib.util.module_from_spec(spec)
    sys.modules[full_name] = module
    spec.loader.exec_module(module)
core = sys.modules["gateway.core.metered_pricing"]

def rates(**overrides):
    values = dict(input_price_per_million=D("30"), output_price_per_million=D("60"),
                  cache_read_price_per_million=D("5"), cache_write_price_per_million=None,
                  pricing_tiers=[])
    values.update(overrides)
    return NS(**values)

def price(pricing, included, input_tokens=1000, cached=400):
    usage = core.billable_usage(input_tokens=input_tokens, output_tokens=0,
                               cache_read_tokens=cached, cache_tokens_included=included)
    return core.price_billable_usage(pricing, usage)

flat = rates()
tier = rates(pricing_tiers=[{"min_input_tokens": 1400, "input_price_per_million": "60",
                            "output_price_per_million": "90", "cache_read_price_per_million": "10"}])
rows = []
for name, pricing, included, expected in [
    ("flat-inclusive", flat, True, "0.020000"),
    ("flat-additive", flat, False, "0.032000"),
    ("tier-inclusive", tier, True, "0.020000"),
    ("tier-additive", tier, False, "0.064000"),
]:
    cost, meters, lines = price(pricing, included)
    assert cost == D(expected), name
    rows.append(dict(case=name, total_input=meters["total_input_tokens"],
                     fresh_input=meters["fresh_input_tokens"], cost=str(cost)))
    if name == "tier-additive":
        assert lines == [{"meter": "input", "units": 1000, "rate_per_million": 60.0, "cost": 0.06},
                         {"meter": "cache_read", "units": 400, "rate_per_million": 10.0, "cost": 0.004}]

assert price(flat, True)[0] == price(flat, False, input_tokens=600)[0]
assert price(rates(cache_read_price_per_million=None), True)[0] == D("0.030000")
assert price(rates(cache_read_price_per_million=D(0)), True)[0] == D("0.018000")
warnings = []
class Capture(logging.Handler):
    def emit(self, record):
        warnings.append(record.getMessage())
handler = Capture()
logger.addHandler(handler)
try:
    contradictory = price(flat, True, input_tokens=100, cached=400)
finally:
    logger.removeHandler(handler)
assert contradictory[0] == D("0.003000")
assert contradictory[1]["cache_read_tokens"] == 0
assert any("Cache token count exceeds input tokens" in text for text in warnings)

try:
    core.billable_usage(input_tokens=1, output_tokens=0)
except TypeError:
    pass
else:
    raise AssertionError("Missing convention was accepted")
try:
    price(rates(input_price_per_million=D("NaN")), True)
except ValueError:
    pass
else:
    raise AssertionError("Invalid base rate was accepted")

estimate = core.estimate_metered_cost(tier, estimated_input_tokens=1400, estimated_output_tokens=0)
assert estimate == D("0.084000")
assert estimate >= price(tier, False)[0]

# Two semantically equivalent encodings must select the same tier and price.
matrix_rates = rates(pricing_tiers=[{"min_input_tokens": 7, "input_price_per_million": "60",
                                    "cache_read_price_per_million": "10"}])
pairs = 0
for total in range(14):
    for cached in range(total + 1):
        inclusive = price(matrix_rates, True, input_tokens=total, cached=cached)
        additive = price(matrix_rates, False, input_tokens=total-cached, cached=cached)
        assert inclusive == additive, (total, cached)
        pairs += 1
assert pairs == 105
print("Python " + sys.version.split()[0])
for row in rows:
    print(json.dumps(row, separators=(",", ":")))
print("equivalent payload: inclusive input=1000 and additive input=600 both cost 0.020000")
print("missing cache rate: 0.030000; explicit zero rate: 0.018000")
print("contradictory inclusive counts: cache attribution dropped; cost=0.003000; warning captured")
print("missing convention: TypeError; invalid base rate: ValueError")
print("normalized-input estimate: 0.084000; tier-additive settlement: 0.064000")
print(f"PASS: 11 named scenarios and {pairs} equivalent-encoding pairs")