캐시·바이트·개발 사용량을 섞지 않는 합성 원장

"""Strict synthetic usage ledger. Counts are caller declarations, not billing evidence."""
from copy import deepcopy

TOKEN_FIELDS = ('input', 'cached', 'output', 'reasoning', 'total')
BYTE_FIELDS = ('request_bytes', 'response_bytes')
FIELDS = {'id', 'bucket', 'status', *TOKEN_FIELDS, *BYTE_FIELDS}


def count(value):
    if value is not None and (type(value) is not int or value < 0 or value > 9007199254740991):
        raise ValueError('NONNEGATIVE_INTEGER_OR_UNKNOWN_REQUIRED')
    return value


def normalize(row):
    if type(row) is not dict or set(row) != FIELDS:
        raise ValueError('EXACT_RECORD_FIELDS_REQUIRED')
    if type(row['id']) is not str or not row['id'].strip() or len(row['id']) > 120:
        raise ValueError('CALL_ID_REQUIRED')
    if row['bucket'] not in ('development', 'normal_run') or row['status'] not in ('completed', 'failed', 'cancelled'):
        raise ValueError('RECORD_CATEGORY_INVALID')
    for key in TOKEN_FIELDS + BYTE_FIELDS:
        count(row[key])
    i, c, o, r, t = (row[k] for k in TOKEN_FIELDS)
    if i is not None and c is not None and c > i:
        raise ValueError('CACHE_IS_A_SUBSET_OF_INPUT')
    if o is not None and r is not None and r > o:
        raise ValueError('REASONING_IS_A_SUBSET_OF_OUTPUT')
    if t is not None:
        if (i is not None and o is not None and t != i + o) or t < (i or 0) + (o or 0):
            raise ValueError('TOTAL_INCONSISTENT_WITH_INPUT_OUTPUT')
    return deepcopy(row)


def derived(row):
    i, c, o, r, t = (row[k] for k in TOKEN_FIELDS)
    uncached = None if i is None or c is None else i - c
    total = t if t is not None else None if i is None or o is None else i + o
    return {**{k: row[k] for k in TOKEN_FIELDS + BYTE_FIELDS},
            'effective_total': total,
            'uncached_input': uncached,
            'uncached_input_plus_output': None if uncached is None or o is None else uncached + o,
            'total_source': 'reported' if t is not None else 'derived' if total is not None else 'unknown'}


def aggregate(rows):
    values = [derived(row) for row in rows]
    keys = TOKEN_FIELDS + BYTE_FIELDS + ('effective_total', 'uncached_input', 'uncached_input_plus_output')
    result = {'calls': len(rows), 'status_counts': {s: sum(r['status'] == s for r in rows) for s in ('completed', 'failed', 'cancelled')},
              'metrics': {}, 'total_sources': {s: sum(v['total_source'] == s for v in values) for s in ('reported', 'derived', 'unknown')},
              'provider_authenticated': False, 'currency_cost_calculated': False}
    for key in keys:
        known = [v[key] for v in values if v[key] is not None]
        result['metrics'][key] = {'known_sum': sum(known), 'unknown_calls': len(rows) - len(known), 'complete': len(rows) == len(known)}
    return result


class UsageLedger:
    def __init__(self):
        self._rows = {}

    def add(self, row):
        clean = normalize(row)
        old = self._rows.get(clean['id'])
        if old is not None:
            if old != clean:
                raise ValueError('CALL_ID_CONFLICT')
            return False
        self._rows[clean['id']] = clean
        return True

    def rows(self):
        return deepcopy(list(self._rows.values()))

    def summary(self, bucket=None):
        if bucket is not None and bucket not in ('development', 'normal_run'):
            raise ValueError('BUCKET_INVALID')
        return aggregate([r for r in self._rows.values() if bucket is None or r['bucket'] == bucket])

    def compare_samples(self, before_ids, after_ids):
        for ids in (before_ids, after_ids):
            if type(ids) is not list or not ids or any(type(x) is not str for x in ids) or len(ids) != len(set(ids)):
                raise ValueError('DISTINCT_NONEMPTY_CALL_LIST_REQUIRED')
            if any(x not in self._rows for x in ids):
                raise ValueError('CALL_NOT_FOUND')
        if set(before_ids) & set(after_ids):
            raise ValueError('OVERLAPPING_SAMPLES')
        rows = [self._rows[x] for x in before_ids + after_ids]
        if any(r['bucket'] != 'normal_run' for r in rows):
            raise ValueError('COMPARE_NORMAL_RUNS_ONLY')
        before, after = (aggregate([self._rows[x] for x in ids]) for ids in (before_ids, after_ids))
        changes = {}
        for key in ('effective_total', 'uncached_input_plus_output'):
            a, b = before['metrics'][key], after['metrics'][key]
            complete = a['complete'] and b['complete']
            changes[key] = {'before': a, 'after': b,
                            'difference': b['known_sum'] - a['known_sum'] if complete else None,
                            'reduction_percent': round(100 * (a['known_sum'] - b['known_sum']) / a['known_sum'], 4) if complete and a['known_sum'] else None}
        return {'sample_changes': changes, 'controlled_experiment_verified': False,
                'general_savings_guaranteed': False, 'currency_cost_calculated': False}