"""Every declared row counts; missing, invalid and unparsed outputs remain failures.""" import argparse from collections import defaultdict import json import math from pathlib import Path import random def indexed(rows, name): result = {} for row in rows: if row['id'] in result: raise ValueError(f'Duplicate ID in {name}: {row["id"]}') result[row['id']] = row return result def read_jsonl(path): with open(path) as stream: rows = [json.loads(line) for line in stream if line.strip()] indexed(rows, str(path)) return rows def vector(values, ids): if isinstance(values, dict): if set(values) != set(ids): raise ValueError('Probability keys differ from gold option IDs') values = [values[key] for key in ids] if not isinstance(values, list) or len(values) != len(ids): raise ValueError('Wrong probability vector length/type') if any(isinstance(p, bool) or not isinstance(p, (int, float)) or not math.isfinite(p) or not 0 <= p <= 1 for p in values): raise ValueError('Invalid/nonfinite probability') if abs(sum(values)-1) > 1e-4: raise ValueError('Probabilities do not sum to one') return values def align(gold, predictions): truth, outputs = indexed(gold, 'gold'), indexed(predictions, 'predictions') if outputs.keys() - truth.keys(): raise ValueError(f'Unknown prediction IDs: {sorted(outputs.keys()-truth.keys())[:5]}') rows = [] for item in gold: ids = [o['id'] for o in item['options']] if len(set(ids)) != len(ids) or not 0 <= item['label'] < len(ids): raise ValueError(f'Invalid gold options/label: {item["id"]}') target = item.get('target_distribution') if target is not None: target = vector(target, ids) row = dict(id=item['id'], group_id=item['group_id'], family=item['family'], gold_id=ids[item['label']], predicted_id=None, correct=False, confidence=None, nll=None, brier=None, status='missing', probabilities=None) pred = outputs.get(item['id']) if pred is not None: try: if pred.get('parse_status') == 'unparsed' or pred.get('error'): raise ValueError(pred.get('parse_error') or pred.get('error') or 'Unparsed output') values = pred.get('probabilities') if values is None: choice = pred.get('prediction_id') if choice not in ids: raise ValueError('Missing or out-of-set native prediction') row.update(status='native_decision', predicted_id=choice) else: if isinstance(values, list) and 'option_ids' in pred: pred_ids = pred['option_ids'] if len(pred_ids) != len(values) or len(set(pred_ids)) != len(pred_ids): raise ValueError('Invalid prediction option IDs') values = dict(zip(pred_ids, values)) ps = vector(values, ids) chosen = max(range(len(ps)), key=lambda k: ps[k]) row.update(status='distribution', predicted_id=ids[chosen], probabilities=ps, confidence=ps[chosen], nll=-math.log(max(ps[item['label']], 1e-12)), brier=sum((p-(k == item['label']))**2 for k,p in enumerate(ps))) if target is not None: row['analytic'] = dict(squared_probability_error=sum((p-q)**2 for p,q in zip(ps,target)), expected_nll=-sum(q*math.log(max(p,1e-12)) for p,q in zip(ps,target)), expected_brier=1+sum(p*p-2*p*q for p,q in zip(ps,target))) row['correct'] = row['predicted_id'] == row['gold_id'] except (ValueError, TypeError) as exc: row.update(status='invalid', error=str(exc)) rows.append(row) return rows def clusters(rows, key='group_id'): result = defaultdict(list) for row in rows: result[row[key]].append(row) return result def basic(rows): if not rows: return dict(n=0, accuracy=None, balanced_accuracy=None, macro_f1=None) recalls, f1 = [], [] for label in sorted({r['gold_id'] for r in rows}): tp = sum(r['gold_id']==label and r['predicted_id']==label for r in rows) fp = sum(r['gold_id']!=label and r['predicted_id']==label for r in rows) fn = sum(r['gold_id']==label and r['predicted_id']!=label for r in rows) recalls.append(tp/(tp+fn)) f1.append(2*tp/(2*tp+fp+fn)) return dict(n=len(rows), accuracy=sum(r['correct'] for r in rows)/len(rows), balanced_accuracy=sum(recalls)/len(recalls), macro_f1=sum(f1)/len(f1)) def summarize(rows): result = basic(rows) if not rows: return result groups = list(clusters(rows).values()) valid = [r for r in rows if r['status']=='distribution'] result.update(source_groups=len(groups), invalid_or_missing=sum(r['status'] in ('invalid','missing') for r in rows), all_decisions_correct_group_rate=sum(all(r['correct'] for r in g) for g in groups)/len(groups), probability_rows=len(valid), probability_coverage=len(valid)/len(rows)) for key in ('nll','brier'): value = sum(r[key] for r in valid)/len(valid) if valid else None result[key] = value if len(valid)==len(rows) else None result[key+'_valid_distributions_only'] = value result['nll_probability_floor'] = 1e-12 ranked = sorted(valid, key=lambda r: (-r['confidence'],r['id'])) result['risk_coverage'] = [] for fraction in (.25,.5,.75,1): if not ranked: break k = min(len(ranked), max(1,math.ceil(len(rows)*fraction))) threshold = ranked[k-1]['confidence'] accepted = [r for r in ranked if r['confidence']>=threshold] result['risk_coverage'].append(dict(requested_coverage=fraction, coverage=len(accepted)/len(rows), n=len(accepted), threshold=threshold, error=1-sum(r['correct'] for r in accepted)/len(accepted))) result['reliability_bins'] = [] for b in range(10): part = [r for r in valid if min(9,int(r['confidence']*10))==b] if part: result['reliability_bins'].append(dict(lower=b/10,upper=(b+1)/10,n=len(part), mean_confidence=sum(r['confidence'] for r in part)/len(part), accuracy=sum(r['correct'] for r in part)/len(part))) rng, values = random.Random(217), [] for _ in range(1000): draw = [groups[rng.randrange(len(groups))] for _ in groups] values.append(sum(r['correct'] for g in draw for r in g)/sum(map(len,draw))) values.sort() result['accuracy_cluster_bootstrap_95'] = [values[25],values[974]] return result def balanced_metric(rows): families = clusters(rows,'family') return sum(basic(part)['balanced_accuracy'] for part in families.values())/len(families) def paired_comparison(left, right, samples=1000, seed=217): """Use identical source-group draws on both systems, stratified by family.""" a,b = indexed(left,'left'),indexed(right,'right') if a.keys()!=b.keys() or not a or samples < 40: raise ValueError('Need same nonempty gold IDs and at least 40 bootstrap samples') for key in a: if any(a[key][f]!=b[key][f] for f in ('group_id','family','gold_id')): raise ValueError('Paired gold metadata disagree') strata = defaultdict(dict) for group,rows in clusters(left).items(): families = {r['family'] for r in rows} if len(families)!=1: raise ValueError('Source group crosses families') strata[next(iter(families))][group] = [r['id'] for r in rows] rng,draws = random.Random(seed),[] for _ in range(samples): ids = [] for groups in strata.values(): units = list(groups.values()) for _ in units: ids.extend(units[rng.randrange(len(units))]) draws.append(balanced_metric([a[k] for k in ids])-balanced_metric([b[k] for k in ids])) draws.sort() return dict(metric='mean_family_balanced_accuracy', difference=balanced_metric(left)-balanced_metric(right), paired_source_group_bootstrap_95=[draws[int(.025*samples)],draws[min(samples-1,int(.975*samples))]], source_groups=sum(map(len,strata.values())),samples=samples,seed=seed, method='Paired source-group bootstrap stratified by family; represented gold classes per draw') def evaluate(gold,predictions,comparison=None): rows = align(gold,predictions) families = {name:summarize(part) for name,part in clusters(rows,'family').items()} analytic = [dict(id=r['id'],**r['analytic']) for r in rows if 'analytic' in r] result = dict(available_gold=len(gold),scored=len(predictions),evaluated=len(rows), coverage=sum(r['status'] not in ('invalid','missing') for r in rows)/len(rows) if rows else 0, missing=sum(r['status']=='missing' for r in rows),invalid=sum(r['status']=='invalid' for r in rows), family_results=families,mean_family_balanced_accuracy=balanced_metric(rows) if rows else None, mean_family_macro_f1=sum(r['macro_f1'] for r in families.values())/len(families) if families else None, analytic_uncertainty={'n':len(analytic),'rows':analytic},errors=[r for r in rows if not r['correct']], limitations=['All gold rows count in accuracy, including missing/invalid/unparsed predictions.', 'Native decisions have no probability estimates; one-hot distributions are not fabricated.', 'Risk/coverage is descriptive; tied confidence values remain together.', 'Analytic event hard-label accuracy measures modal-class agreement, not observed-event accuracy.']) if comparison is not None: result['paired_comparison'] = paired_comparison(rows,align(gold,comparison)) return result def screening_gate(gold, predictions, policy='distribution', threshold=0.8): """Frozen semantic-decision gate; not a benchmark of executed workflow actions.""" if policy not in ('distribution', 'native') or threshold != 0.8: raise ValueError('Screen policy is frozen: distribution p>=0.8 or native parsed decisions') rows = align(gold, predictions) result_by_id = indexed(rows, 'aligned') groups = clusters(gold) family_counts, family_pairs = defaultdict(int), defaultdict(int) issues, pair_correct, missing_automatic, original_correct_automatic = [], 0, 0, 0 missing_count, evidence_correct, original_automatic = 0, 0, 0 decisions = [] required = {'original', 'criterion_reversal', 'evidence_change', 'missing'} for group, items in groups.items(): variants = {} for item in items: variant = item.get('provenance', {}).get('variant') if variant in variants: issues.append(f'{group}: duplicate variant {variant}') variants[variant] = item if set(variants) != required or len(items) != 4 or len({r['family'] for r in items}) != 1: issues.append(f'{group}: requires four declared variants in one family') continue family = items[0]['family'] family_counts[family] += 1 paired = all(result_by_id[variants[v]['id']]['correct'] for v in ('original','criterion_reversal')) pair_correct += paired family_pairs[family] += paired evidence_correct += result_by_id[variants['evidence_change']['id']]['correct'] for variant, item in variants.items(): row = result_by_id[item['id']] semantic_choice = row['predicted_id'] available = row['status'] == ('distribution' if policy == 'distribution' else 'native_decision') automatic = available and semantic_choice != 'insufficient' if policy == 'distribution': automatic = automatic and row['confidence'] >= threshold decisions.append(dict(id=row['id'], variant=variant, choice=semantic_choice, disposition='automatic_semantic_decision' if automatic else 'review', correct=row['correct'])) if variant == 'missing': missing_count += 1 if row['gold_id'] != 'insufficient': issues.append(f'{item["id"]}: missing variant must have insufficient gold') missing_automatic += automatic elif variant == 'original': original_automatic += automatic original_correct_automatic += automatic and row['correct'] expected_families = {'evidence_interpretation','rule_application','candidate_selection'} if len(gold)!=96 or len(groups)!=24 or set(family_counts)!=expected_families or any(n!=8 for n in family_counts.values()): issues.append('Screen requires 96 rows, 24 groups and eight groups in each declared family') complete = all(r['status'] == ('distribution' if policy == 'distribution' else 'native_decision') for r in rows) checks = dict(schema=not issues, complete_valid_outputs=complete, original_reversal_pairs=pair_correct>=20, every_family_pairs=set(family_counts)==expected_families and all(family_pairs[f]>=6 for f in expected_families), no_unsupported_missing_decision=missing_count==24 and missing_automatic==0, useful_original_coverage=original_correct_automatic>=12) return dict(targets_met=all(checks.values()), structurally_valid=not issues, decision='Reviewer decides whether observed failures justify a bounded pilot; numerical near-misses are not automatic vetoes.', checks=checks, schema_errors=issues, policy=dict(version='screen-policy-v1', mode=policy, threshold=threshold if policy=='distribution' else None, abstain_option='insufficient', scope='semantic decisions, not executed workflow actions'), original_reversal_pairs_correct=pair_correct, family_pairs_correct=dict(family_pairs), evidence_change_correct=evidence_correct, missing_cases=missing_count, unsupported_missing_decisions=missing_automatic, original_automatic_decisions=original_automatic, original_correct_automatic_decisions=original_correct_automatic, accuracy_including_abstentions=basic(rows)['accuracy'], decisions=decisions, limitation='Small falsification screen; zero errors does not certify safety/calibration.') if __name__=='__main__': parser=argparse.ArgumentParser(description=__doc__) for name in ('gold','predictions','output'): parser.add_argument('--'+name,required=True) parser.add_argument('--comparison') parser.add_argument('--screen-policy', choices=('distribution','native')) args=parser.parse_args() result=evaluate(read_jsonl(args.gold),read_jsonl(args.predictions),read_jsonl(args.comparison) if args.comparison else None) if args.screen_policy: result['screening_gate'] = screening_gate(read_jsonl(args.gold), read_jsonl(args.predictions), args.screen_policy) with Path(args.output).open('x') as destination: destination.write(json.dumps(result,indent=2,allow_nan=False)+'\n') print(json.dumps({k:result[k] for k in ('scored','coverage','mean_family_balanced_accuracy')}))