"""Versioned full common-core likelihood suite. No model-card metrics are imported.

Dataset identity, item counts, prompt schema, model revisions and runtime backend
are closed over source-pinned profiles. This is not the historical 32-item pilot.
"""
from __future__ import annotations
import hashlib, importlib.metadata, json, platform, re, sys, time
from pathlib import Path
ROOT = Path(__file__).resolve().parent
DEPENDENCIES = {'evaluate_sub1b_standard.py': 'ea8141a818424f6e010d35f8e237d2f8682640b6d123ed87438ac01655492c8a', 'evaluate_qwen35_challenge.py': 'bd025b3fd12a0b84102a8713ca2af2d08bf6f16178dcaac6fce4a1e720bf1f4c', 'requirements-qwen35.txt': 'd41a79a7c63d4a6deb12a1125c806b4729bcdb68d77f240928e72714da40cbfa', 'sub1b_native.py': 'ea56adc25566f7f6745ab882684dfc67e7f1cc48f29e53422fb3cab0cce769e7', 'sub1b_slayer139.py': 'f23a691bb330db4e6311b4b7e83936d5a4c584cc3f61ecf7f7025435ad87d326', 'sub1b_gollem.py': 'b9bff95fe326165b5685572727d5c50b1f86a50676932ad5b7da17018965e1d7', 'sub1b_balanced.py': 'f2949b3c2a6f3585ba9eb4b3bc7ef8306afea74492670d7c82d8c3f7e2c2a64b', 'sub1b-models.json': 'e6b253dc7b72d796866c2d6d9fb2ff63f0dd86203b40f2d1088059ef6a4995bd', 'common-core-profiles.json': 'a54c99201b8e0784bc5d7efab18b39f3dc50168ff6f0f7bee1315e77bb9bb4f1'}

def verify_sources():
    for name, digest in DEPENDENCIES.items():
        if hashlib.sha256((ROOT / name).read_bytes()).hexdigest() != digest:
            raise ValueError('Pinned common-core dependency differs: ' + name)

def validate_job(job, backend='native'):
    verify_sources()
    profiles = json.loads((ROOT / 'common-core-profiles.json').read_text())
    supplied = job['task']
    profile = next((t for t in profiles if t['id'] == supplied.get('id')), None)
    if profile is None or profile['runtime_backend'] != backend:
        raise ValueError('Unknown common-core profile or mismatched runtime backend')
    # Permit server-added operational metadata only; every protocol field must match.
    for key, value in profile.items():
        if supplied.get(key) != value:
            raise ValueError('Common-core protocol mismatch: ' + key)
    if not any(m['id'] == job['model'] and m['revision'] == job['revision'] for m in profile['models']):
        raise ValueError('Model revision is not approved for this exact cohort')
    return profile

def map_item(row, schema):
    if schema == 'arc-challenge-v1':
        choices = row['choices']['text']
        target = [str(x) for x in row['choices']['label']].index(str(row['answerKey']))
        prompt = 'Question: ' + row['question'] + '\nAnswer:'
        source_id = row['id']
    elif schema == 'hellaswag-lm-eval-v1':
        def preprocess(text):
            text = text.strip().replace(' [title]', '. ')
            return re.sub(r'\[.*?\]', '', text).replace('  ', ' ')
        prompt = preprocess(row['activity_label'] + ': ' + row['ctx_a'] + ' ' + row['ctx_b'].capitalize())
        choices = [preprocess(c) for c in row['endings']]; target = int(row['label'])
        source_id = row.get('ind')
        if len(choices) != 4: raise ValueError('Expected four HellaSwag endings')
    elif schema == 'boolq-lm-eval-v1':
        if type(row['label']) is not int or row['label'] not in (0, 1): raise ValueError('BoolQ gold must be integer 0/1')
        choices = ['no', 'yes']; target = row['label']
        prompt = row['passage'] + '\nQuestion: ' + row['question'] + '?\nAnswer:'
        source_id = row['idx']
    else: raise ValueError('Unsupported dataset schema')
    if not isinstance(prompt, str) or not prompt.strip() or not isinstance(choices, list) or not 0 <= target < len(choices):
        raise ValueError('Invalid question/answer mapping')
    if any(not isinstance(c, str) or not c.strip() for c in choices): raise ValueError('Empty candidate text')
    return prompt, choices, target, source_id

def load_rows(task):
    from huggingface_hub import hf_hub_download
    import pyarrow.parquet as pq
    filename = hf_hub_download(task['dataset'], task['recipe']['data_file'], repo_type='dataset', revision=task['dataset_revision'], token=False)
    table = pq.ParquetFile(filename)
    if table.metadata.num_rows != task['limit']: raise ValueError('Pinned split row count differs')
    rows = [r for batch in table.iter_batches(batch_size=128) for r in batch.to_pylist()]
    for row in rows: map_item(row, task['recipe']['schema'])
    return rows

def load_native(ident, revision, device):
    import torch
    from transformers import AutoTokenizer, AutoModelForCausalLM
    approved = next(m for m in json.loads((ROOT / 'sub1b-models.json').read_text()) if m['id'] == ident and m['revision'] == revision)
    if approved['adapter'] != 'hf':
        if device != 'cuda:0': raise ValueError('Reviewed native Slayer loader requires CUDA')
        from sub1b_native import load
        tokenizer, model = load(approved)
    else:
        tokenizer = AutoTokenizer.from_pretrained(ident, revision=revision, trust_remote_code=False)
        model, info = AutoModelForCausalLM.from_pretrained(ident, revision=revision, trust_remote_code=False, torch_dtype=torch.float32, output_loading_info=True)
        if any(info.get(k) for k in ('missing_keys','unexpected_keys','mismatched_keys','error_msgs')):
            raise ValueError('Stock checkpoint did not load cleanly: ' + str(info)[:1000])
        model = model.to(device).eval()
    if tokenizer.eos_token_id is None: raise ValueError('Missing EOS padding ID')
    parameters = sum(p.numel() for p in model.parameters())
    if not 0 < parameters < 1_000_000_000: raise ValueError('Not a sub-1B system')
    return tokenizer, model, {'parameters': parameters}

def evaluate(job, output, backend='native', runner_path=None):
    task = validate_job(job, backend)
    import numpy as np
    import torch
    import transformers
    from evaluate_sub1b_standard import batched_choice, choice_decision
    started = time.monotonic(); torch.manual_seed(42); np.random.seed(42); torch.set_num_threads(4)
    torch.backends.cuda.matmul.allow_tf32 = False; torch.backends.cudnn.allow_tf32 = False
    device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
    rows = load_rows(task); data_seconds = time.monotonic() - started; began = time.monotonic()
    if backend == 'qwen35':
        from evaluate_qwen35_challenge import load_model
        tokenizer, model, details = load_model(job['model'], job['revision'], device)
    else: tokenizer, model, details = load_native(job['model'], job['revision'], device)
    loading_seconds = time.monotonic() - began
    problems = [map_item(r, task['recipe']['schema']) for r in rows]
    truncated = 0
    for prompt, choices, _, _ in problems:
        prefix = tokenizer.encode(prompt, add_special_tokens=False)
        for c in choices:
            suffix = tokenizer.encode(' ' + c, add_special_tokens=False)
            if not suffix or len(suffix) > 512: raise ValueError('Candidate exceeds fixed answer budget')
            truncated += int(len(prefix) + len(suffix) > 1024)
    began = time.monotonic()
    values, timings = batched_choice(model, tokenizer, [(p,c) for p,c,_,_ in problems], device, batch_size=16)
    scoring_seconds = time.monotonic() - began; samples = []
    for index, ((prompt, choices, target, source_id), scores) in enumerate(zip(problems, values, strict=True)):
        raw_prediction, norm = choice_decision(scores, choices)
        prediction = norm if task['metric'] == 'accuracy_char_normalized' else raw_prediction
        samples.append(dict(id=str(index), source_id=source_id, prompt=prompt, choices=choices, reference=target, prediction=prediction, score=float(prediction==target), prediction_raw=raw_prediction, score_raw=float(raw_prediction==target), log_likelihood_sum=scores, answer_token_lengths=[len(tokenizer.encode(' '+c,add_special_tokens=False)) for c in choices], prediction_char_norm=norm, score_char_norm=float(norm==target)))
    runner = Path(runner_path or __file__)
    dependencies = dict(DEPENDENCIES)
    if backend == 'qwen35': dependencies['evaluate_common_core.py'] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    result = dict(score=sum(s['score'] for s in samples)/len(samples), n=len(samples), acc_raw=sum(s['score_raw'] for s in samples)/len(samples), acc_norm=sum(s['score_char_norm'] for s in samples)/len(samples), acc_norm_definition='argmax summed conditional likelihood / Python character length of answer, excluding leading delimiter', model_revision=job['revision'], dataset_revision=task['dataset_revision'], runner_sha256=hashlib.sha256(runner.read_bytes()).hexdigest(), dependency_sources=dependencies, seconds=time.monotonic()-started, data_loading_seconds=data_seconds, model_load_seconds=loading_seconds, scoring_seconds=scoring_seconds, peak_gpu_bytes=torch.cuda.max_memory_allocated() if device.startswith('cuda') else 0, seed=42, device=device, protocol=task, selection=task['selection'], execution_profile=task['protocol_version'], precision='float32', tf32=False, chat_template=False, batch_policy=dict(choice_candidates=16,padded_token_budget=8192,context=1024,answer_limit=512), batch_timings=timings, truncated_candidates=truncated, measured_cost_usd=None, notes=('Full pinned split; primary character-normalized accuracy: argmax summed conditional answer-token log likelihood divided by Python len(choice), excluding the leading delimiter; raw accuracy is a separate diagnostic. ' if task['metric'] == 'accuracy_char_normalized' else 'Full pinned split; primary raw summed conditional answer-token log likelihood accuracy; character-normalized accuracy is a separate diagnostic. ') + 'Explicit pinned prompt schema, no chat/BOS. Left context truncated only if prompt plus answer exceeds 1024 tokens. Generic text diagnostic does not validate specialist declared tasks.', **details)
    environment = dict(python=sys.version,platform=platform.platform(),torch=torch.__version__,transformers=transformers.__version__,cuda=torch.version.cuda,gpu=torch.cuda.get_device_name() if device.startswith('cuda') else None,packages={d.metadata['Name']:d.version for d in importlib.metadata.distributions()})
    output.parent.mkdir(parents=True,exist_ok=True);output.write_text(json.dumps(dict(result=result,samples=samples,environment=environment),allow_nan=False))

if __name__ == '__main__': evaluate(json.loads(Path(sys.argv[1]).read_text()), Path(sys.argv[2]))