| """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)) |
| |
| |
| 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) |
| |
| 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())) |
| |
| |
| 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() |
|
|
