"""Frozen, single-attempt evaluation of both roles on a context-bearing pack.""" import argparse import hashlib import json from pathlib import Path import sys import time ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from tancho.mission_context import canonical_sha256 from training.observation_program_loader import ObservationProgramDataset, verify_pack from training.score_observation_program import score_predictions def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--pack', type=Path, required=True) parser.add_argument('--base', type=Path, required=True) parser.add_argument('--export', type=Path, required=True) parser.add_argument('--thresholds', type=Path, required=True) parser.add_argument('--output', type=Path, required=True) parser.add_argument('--max-new-tokens', type=int, default=512) args = parser.parse_args() if args.output.exists(): raise ValueError('Use a fresh evaluation output directory') manifest = verify_pack(args.pack, purpose='evaluation') thresholds = json.loads(args.thresholds.read_text()) if thresholds.get('schema_version') != 'tancho-observation-program-eval-thresholds-1.0' \ or thresholds.get('pack_manifest_sha256') != manifest['manifest_sha256']: raise ValueError('Evaluation thresholds are not bound to this frozen pack') args.output.mkdir(parents=True) import torch from transformers import AutoModelForImageTextToText import cosmos_framework.model.generator.reasoner.cosmos3_edge from cosmos_framework.data.generator.processors import build_processor from tancho.edge_chat import install from tancho.observation_intent import parse_intent_output, validate_and_bind_intents from tancho.observation_program import parse_program_output, validate_observation_program if not torch.cuda.is_available() or torch.cuda.device_count() != 1: raise ValueError('Frozen evaluation requires exactly one CUDA GPU') install(); torch.manual_seed(42) model, loading = AutoModelForImageTextToText.from_pretrained( str(args.export), torch_dtype=torch.bfloat16, device_map='cuda:0', attn_implementation='sdpa', local_files_only=True, output_loading_info=True) if any(loading.get(key) for key in ('missing_keys', 'unexpected_keys', 'mismatched_keys', 'error_msgs')): raise ValueError('Export reload changed tensors') model.eval(); processor = build_processor(tokenizer_type=str(args.base), config_variant='hf') dataset = ObservationProgramDataset(args.pack, 'test', purpose='evaluation') predictions, records, receipts = {}, [], [] started = time.monotonic() for index, row in enumerate(dataset.rows): if 'context' not in row: raise ValueError('Evaluation row lacks validator context') item = dataset[index] inputs = processor.apply_chat_template([item['texts'][0]], tokenize=True, add_generation_prompt=True, return_tensors='pt') tensor_inputs = {key: (value.unsqueeze(0) if key in ('input_ids', 'attention_mask') and value.ndim == 1 else value).to('cuda:0') for key, value in inputs.items() if torch.is_tensor(value)} case_started = time.monotonic() with torch.inference_mode(): generated = model.generate(**tensor_inputs, max_new_tokens=args.max_new_tokens, do_sample=False, use_cache=False, eos_token_id=11, pad_token_id=0) tokens = generated[0, tensor_inputs['input_ids'].shape[-1]:].detach().cpu().tolist() raw = processor.processor.tokenizer.decode(tokens, skip_special_tokens=True) predictions[row['sample_id']] = raw valid, error = False, None try: if row['role'] == 'reasoner': parsed = parse_intent_output(raw) validate_and_bind_intents(parsed, row['context'], now_ms=row['context']['captured_at_ms'], raw_text=raw) else: parsed = parse_program_output(raw) validate_observation_program(parsed, row['context'], row['verified_intents'], now_ms=row['context']['captured_at_ms'], raw_text=raw) valid = True except Exception as exc: error = type(exc).__name__ + ': ' + str(exc) raw_path = args.output / (row['sample_id'] + '.txt'); raw_path.write_text(raw) receipts.append({'sample_id': row['sample_id'], 'role': row['role'], 'single_attempt': True, 'valid': valid, 'error': error, 'tokens': len(tokens), 'seconds': time.monotonic() - case_started, 'raw_sha256': hashlib.sha256(raw.encode()).hexdigest()}) records.append({'sample_id': row['sample_id'], 'role': row['role'], 'mission_profile': row['mission_profile'], 'visual_pair_id': row['visual_pair_id'], 'context': row['context'], 'verified_intents': row['verified_intents'], 'target': json.loads(row['target'])}) del tensor_inputs, generated score = score_predictions(records, predictions) gates = { 'schema_valid_rate': score['overall']['schema_valid_rate'] >= thresholds['minimum_schema_valid_rate'], 'exact_accuracy': score['overall']['exact_accuracy'] >= thresholds['minimum_exact_accuracy'], 'paired_visual_sensitivity': score['paired_visual_sensitivity']['rate'] >= thresholds['minimum_paired_visual_sensitivity'], 'no_invalid_outputs': all(row['valid'] for row in receipts), } raw_hash = canonical_sha256({row['sample_id']: row['raw_sha256'] for row in receipts}) report = {'schema_version': 'tancho-observation-program-frozen-evaluation-1.0', 'passed': all(gates.values()), 'pack_manifest_sha256': manifest['manifest_sha256'], 'thresholds_sha256': hashlib.sha256(args.thresholds.read_bytes()).hexdigest(), 'frozen_before_inference': True, 'single_attempt_no_repair': True, 'loading_info': loading, 'score': score, 'gates': gates, 'receipts': receipts, 'model_raw_outputs_sha256': raw_hash, 'elapsed_seconds': time.monotonic() - started} (args.output/'report.json').write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n') print('TANCHO_FROZEN_EVAL '+json.dumps({'passed': report['passed'], 'overall': score['overall'], 'paired': score['paired_visual_sensitivity']})) return 0 if report['passed'] else 2 if __name__ == '__main__': raise SystemExit(main())