"""Locked, cost-accounted evaluation of decision-sufficient representations.""" from __future__ import annotations import argparse import csv import gzip import json from pathlib import Path import time import torch from .core import calibration, count_samples, dispersion, encode, estimate, plan, project, push, realized from .worlds import generate, sha, write ROOT = Path('artifacts/decision_sufficiency_01') PROTOCOL = Path('configs/decision_sufficiency_protocol_01.json') class Timer: def __init__(self, device): self.device = device def __enter__(self): torch.cuda.synchronize(self.device) self.started = time.perf_counter() return self def __exit__(self, *exc): torch.cuda.synchronize(self.device) self.elapsed = time.perf_counter()-self.started def verify_lock(): lock = json.loads((ROOT/'preconfirmation_lock.json').read_text()) changed = [p for p, digest in lock['files'].items() if sha(p) != digest] assert not changed, changed return lock def register(): assert not (ROOT/'preconfirmation_lock.json').exists() assert not (ROOT/'confirmation').exists() files = sorted(Path('unionmind/decision_sufficiency').glob('*.py'))+[ PROTOCOL, Path('tests/test_decision_sufficiency.py'), ROOT/'interface_tests_final.log', ROOT/'development/worlds.pt', ROOT/'development/manifest.json', ROOT/'development_evaluation/summary.json'] write(ROOT/'preconfirmation_lock.json', dict( files={str(p): sha(p) for p in files}, confirmation_generated=False, protocol_sha256=sha(PROTOCOL), new_controller_training=False, independent_external_confirmation=False, thresholds_and_all_comparisons_fixed=True, test_log=str(ROOT/'interface_tests_final.log'))) def evaluate(role): protocol = json.loads(PROTOCOL.read_text()) if role == 'confirmation': verify_lock() destination = ROOT/f'{role}_evaluation' assert not destination.exists(), destination destination.mkdir() device = protocol['execution']['device'] torch.set_num_threads(4) torch.backends.cuda.matmul.allow_tf32 = False torch.cuda.init() torch.cuda.synchronize(device) torch.cuda.reset_peak_memory_stats(device) started = time.perf_counter() source = ROOT/role manifest = json.loads((source/'manifest.json').read_text()) assert sha(source/'worlds.pt') == manifest['file_sha256'] with Timer(device) as imported: archive = torch.load(source/'worlds.pt', map_location=device, weights_only=True) truth, samples = archive['joint'], archive['samples'] metadata = manifest['worlds'] b = len(metadata) timing, saved, rows, inference_controls = [], {}, 0, [] max_decomposition_error, minimum_loss, max_bound_violation = 0., 0., 0. costs = tuple(protocol['task']['query_costs']) methods = protocol['representations']['methods'] models = protocol['estimation']['models'] with Timer(device) as scored: full_oracle = plan(truth, costs)['expected'] timing.append(dict(stage='privileged_full_oracle', seconds=scored.elapsed)) output = gzip.open(destination/'predictions.csv.gz', 'wt', newline='') writer = None for availability, available in [('full', 6), ('three_available', 3)]: raw_truth = project(truth, available) with Timer(device) as scored: raw_oracle = plan(raw_truth, costs)['expected'] timing.append(dict(stage='privileged_available_oracle', availability=availability, seconds=scored.elapsed)) # Faithful probabilities, raw representation, deliberately inadequate # action computation: force a cause choice even when abstention wins. with Timer(device) as scored: hm = raw_truth.sum(-1) decision = hm.argmax(-1) action_values = torch.stack((hm[..., 0]-2*hm[..., 1], hm[..., 1]-2*hm[..., 0]), -1) forced = action_values.gather(-1, decision[..., None]).squeeze(-1).sum(1) timing.append(dict(stage='inference_negative_control', availability=availability, seconds=scored.elapsed, evaluator_only=True)) for w, meta in enumerate(metadata): inference_controls.append(dict(world=meta['id'], family=meta['family'], availability=availability, query_budget=0, faithful_model=True, raw_representation=True, policy='force_argmax_hypothesis_without_abstention', realized_utility=float(forced[w]), available_oracle=float(raw_oracle[w, 0, 0]), inference_loss=float(raw_oracle[w, 0, 0]-forced[w]))) for stream in [-1, 0, 1, 2]: budgets = [0] if stream == -1 else protocol['estimation']['label_budgets'] previous, accumulated = 0, None for n in budgets: with Timer(device) as counted: if stream >= 0: delta = count_samples(samples[:, stream, previous:n], available) accumulated = delta if accumulated is None else accumulated+delta previous = n timing.append(dict(stage='incremental_sample_count', availability=availability, sample_stream=stream, label_budget=n, seconds=counted.elapsed, worlds=b)) for model_name in (['faithful_explicit'] if stream == -1 else models): with Timer(device) as fitted: model = raw_truth if stream == -1 else estimate(accumulated, model_name) timing.append(dict(stage='estimation', availability=availability, sample_stream=stream, label_budget=n, model=model_name, seconds=fitted.elapsed, worlds=b, fitted_probability_cells_per_world=model[0].numel())) for method in methods: with Timer(device) as encoded: code, k = encode(model, method) with Timer(device) as materialized: compact = model if method == 'identity' else push(model, code, k) with Timer(device) as planned: policy = plan(compact, costs) # The policy is fully fixed before the evaluator touches this law. with Timer(device) as scored: true_compact = raw_truth if method == 'identity' else push(raw_truth, code, k) outcome = realized(true_compact, policy) code_oracle = plan(true_compact, costs)['expected'] scores = calibration(compact, true_compact) information = full_oracle-raw_oracle representation = raw_oracle-code_oracle scores['estimated_conditional_dispersion'] = dispersion(model, code, compact) scores['true_conditional_dispersion'] = dispersion(raw_truth, code, true_compact) span = truth.new_tensor([[3+budget*cost for cost in costs] for budget in (0, 1, 2)]) bound = span[None]*scores['true_conditional_dispersion'][:, None, None] max_bound_violation = max(max_bound_violation, float((representation-bound).max())) model_error = code_oracle-outcome['utility'] total = full_oracle-outcome['utility'] error = (information+representation+model_error-total).abs().max() max_decomposition_error = max(max_decomposition_error, float(error)) minimum_loss = min(minimum_loss, float(information.min()), float(representation.min()), float(model_error.min())) # A proper conditional predictive score includes hypotheses and # all recorded query outcomes, rather than marginal fit alone. conditional = compact/compact.sum((2, 3))[..., None, None].clamp_min(1e-300) scores['joint_forecast_log_loss'] = -(true_compact* conditional.clamp_min(1e-300).log()).sum((1, 2, 3)) timing.append(dict(stage='representation_and_decision', availability=availability, sample_stream=stream, label_budget=n, model=model_name, method=method, encoder_seconds=encoded.elapsed, materialization_seconds=materialized.elapsed, planner_seconds=planned.elapsed, evaluator_seconds=scored.elapsed, worlds=b, raw_states=model.shape[1], code_states=k, encoder_bytes_per_world=0 if method == 'identity' else model.shape[1], temporary_encoder_index_bytes_per_world=code.shape[1]*code.element_size(), compressed_joint_bytes_per_world=k*8*8, planner_choice_entries_per_world=3*k*len(costs)*9, neural_parameters=0, exact_planning=True)) if stream == 0 and n == protocol['primary']['label_budget'] and model_name == protocol['primary']['model']: saved[f'{availability}:{method}'] = dict(code=None if method == 'identity' else code.to(torch.uint8).cpu(), joint=compact.cpu()) data = {**outcome, **scores, 'full_oracle': full_oracle, 'available_oracle': raw_oracle, 'encoded_oracle': code_oracle, 'information_loss': information, 'representation_loss': representation, 'model_loss': model_error, 'total_regret': total} data = {key: value.detach().cpu().tolist() for key, value in data.items()} for w, meta in enumerate(metadata): for budget in protocol['task']['query_budgets']: for c, cost in enumerate(costs): row = dict(world=meta['id'], family=meta['family'], held_mechanism=meta['held_mechanism'], availability=availability, sample_stream=stream, label_budget=n, model=model_name, method=method, query_budget=budget, query_cost=cost) for key, value in data.items(): row[key] = value[w][budget][c] if isinstance(value[w], list) else value[w] row['inference_loss'] = 0. if writer is None: writer = csv.DictWriter(output, fieldnames=list(row)); writer.writeheader() writer.writerow(row); rows += 1 output.close() torch.save(saved, destination/'primary_representations.pt') write(destination/'inference_control.json', inference_controls) write(destination/'timings.json', dict(import_seconds=imported.elapsed, blocks=timing, scope='CUDA-synchronized batched CPU+GPU wall time; never per-incident latency')) assert max_decomposition_error < 1e-10 and minimum_loss > -1e-10 and max_bound_violation < 1e-10 expected = b*2*(1+3*len(protocol['estimation']['label_budgets'])*len(models))*len(methods)*9 assert rows == expected, (rows, expected) write(destination/'summary.json', dict(role=role, worlds=b, prediction_rows=rows, expected_prediction_rows=expected, code_sha256={str(p): sha(p) for p in Path('unionmind/decision_sufficiency').glob('*.py')}, protocol_sha256=sha(PROTOCOL), source_sha256=manifest['file_sha256'], maximum_decomposition_error=max_decomposition_error, maximum_conditional_dispersion_bound_violation=max_bound_violation, minimum_component_regret=minimum_loss, models_supplied_to_estimator=False, true_law_used_only_for_sampling_and_separate_evaluator=True, total_research_seconds=time.perf_counter()-started, peak_gpu_allocated_bytes=torch.cuda.max_memory_allocated(device), peak_gpu_reserved_bytes=torch.cuda.max_memory_reserved(device), new_neural_modules=0, controller_updates=0, backbone_executions=0)) print(json.dumps(dict(role=role, worlds=b, rows=rows, status='EVALUATED'))) def audit_coordinates(): source = Path('artifacts/external_representation_transfer_01/coordinate_disagreements.csv') all_changed = list(csv.DictReader(source.open())) rows = [r for r in all_changed if r['both_optimal'] == 'False'] assert len(rows) == 10 improved = sum(float(r['root_correct_native']) < float(r['root_correct_coordinate']) for r in rows) worsened = sum(float(r['root_correct_native']) > float(r['root_correct_coordinate']) for r in rows) path = ROOT/'coordinate_ten_cases.csv' assert not path.exists() with path.open('w') as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])); writer.writeheader(); writer.writerows(rows) write(ROOT/'coordinate_ten_cases.json', dict(status='POST_HOC_EXPLANATORY_ONLY', source_sha256=sha(source), changed_decisions=len(all_changed), residual_cases=len(rows), improved_under_permutation=improved, worsened_under_permutation=worsened, both_nonoptimal=sum(not float(r['root_correct_native']) and not float(r['root_correct_coordinate']) for r in rows), distinct_motifs=len({r['motif'] for r in rows}), distinct_publications=len({r['publication'] for r in rows}), max_absolute_normalized_policy_regret_change=max(abs(float(r['regret_delta'])) for r in rows), registered_action_identity_gate_changed=False, original_results_modified=False, interpretation='Six root choices improve and four worsen; residual disagreement does not mean all ten permuted choices are wrong. Policy regret is reported separately from root optimal-set membership.')) def main(): p = argparse.ArgumentParser() p.add_argument('action', choices=['generate', 'evaluate', 'register', 'audit_coordinates']) p.add_argument('--role', choices=['development', 'confirmation'], default='development') args = p.parse_args() if args.action == 'generate': generate(json.loads(PROTOCOL.read_text()), args.role, ROOT/args.role) elif args.action == 'evaluate': evaluate(args.role) else: globals()[args.action]() if __name__ == '__main__': main()